The binary format is time since the beginning of the measurement : double unknown and irrelevant field : double momentary consumption calculated for the current time segment : double
| 7 | |
| 8 | |
| 9 | class Analyzer: |
| 10 | """ |
| 11 | The binary format is |
| 12 | time since the beginning of the measurement : double |
| 13 | unknown and irrelevant field : double |
| 14 | momentary consumption calculated for the current time segment : double |
| 15 | """ |
| 16 | def __init__(self): |
| 17 | self.duration = 0.0 |
| 18 | self.consumption = [] |
| 19 | self.mean = 0.0 |
| 20 | self.std = 0.0 |
| 21 | self.avg = 0.0 |
| 22 | self.averages = [] |
| 23 | |
| 24 | |
| 25 | def read_file(self, file_path): |
| 26 | binary = bytearray() |
| 27 | with open(file_path, "r") as f: |
| 28 | binary = bytearray(f.read()) |
| 29 | |
| 30 | for i in range(0, len(binary) - 24, 24): |
| 31 | res = struct.unpack(">ddd", binary[i:i+24]) |
| 32 | |
| 33 | current_duration = res[0] |
| 34 | if not current_duration > self.duration: |
| 35 | print("Unexpected elapsed time value, lower than the previous one.") |
| 36 | exit(2) # this should never happen because the file is written sequentially |
| 37 | |
| 38 | current_consumption = res[2] |
| 39 | self.averages.append(current_consumption / (current_duration - self.duration)) |
| 40 | self.duration = current_duration |
| 41 | |
| 42 | self.consumption.append(current_consumption) |
| 43 | |
| 44 | self.calculate_stats() |
| 45 | |
| 46 | |
| 47 | def calculate_stats(self): |
| 48 | self.mean = numpy.mean(self.averages) |
| 49 | self.std = numpy.std(self.averages) |
| 50 | self.avg = sum(self.consumption) / self.duration |
| 51 | |
| 52 | |
| 53 | if __name__ == "__main__": |
no outgoing calls
no test coverage detected