Add power profiler class

Author: NejcKle

power_profiler.py

@@ -0,0 +1,182 @@
+import time
+import csv
+import datetime
+from threading import Thread
+# import numpy as np
+# import matplotlib.pyplot as plt
+# import matplotlib
+from src.ppk2_api import PPK2_API
+
+class PowerProfiler():
+    def __init__(self, serial_port=None, source_voltage_mV=3300, filename=None):
+        """Initialize PPK2 power profiler with serial"""
+        self.measuring = None
+        self.measurement_thread = None
+        self.ppk2 = None
+
+        try:
+            if serial_port:
+                self.ppk2 = PPK2_API(serial_port)
+            else:
+                serial_port = self.discover_port()
+                if serial_port:
+                    self.ppk2 = PPK2_API(serial_port)
+            ret = self.ppk2.get_modifiers()  # try to read modifiers, if it fails serial port is probably not correct
+        except Exception as e:
+            ret = None
+            raise e
+
+        if not ret:
+            self.ppk2 = None
+            #raise Exception(f"Error when initing PowerProfiler with serial port {serial_port}")
+        else:
+            self.ppk2.use_source_meter()
+
+            self.source_voltage_mV = source_voltage_mV
+
+            self.ppk2.set_source_voltage(self.source_voltage_mV)  # set to 3.3V
+
+            self.measuring = False
+            self.current_measurements = []
+
+            # local variables used to calculate power consumption
+            self.measurement_start_time = None
+            self.measurement_stop_time = None
+
+            time.sleep(1)
+
+            self.stop = False
+
+            self.measurement_thread = Thread(target=self.measurement_loop, daemon=True)
+            self.measurement_thread.start()
+
+            # write to csv
+            self.filename = filename
+            if self.filename is not None:
+                with open(self.filename, 'w', newline='') as file:
+                    writer = csv.writer(file)
+                    row = []
+                    for key in ["ts", "avg1000"]:
+                        row.append(key)
+                    writer.writerow(row)
+
+    def write_csv_rows(self, samples):
+        """Write csv row"""
+        with open(self.filename, 'a', newline='') as file:
+            writer = csv.writer(file)
+            for sample in samples:
+                row = [datetime.datetime.now().strftime('%d-%m-%Y %H:%M:%S.%f'), sample]
+                writer.writerow(row)
+
+    def delete_power_profiler(self):
+        """Join thread"""
+        self.measuring = False
+        self.stop = True
+
+        if self.measurement_thread:
+            self.measurement_thread.join()
+            self.measurement_thread = None
+
+        if self.ppk2:
+            self.disable_power()
+
+    def discover_port(self):
+        """Discovers ppk2 serial port"""
+        ppk2s_connected = PPK2_API.list_devices()
+        if(len(ppk2s_connected) == 1):
+            ppk2_port = ppk2s_connected[0]
+            print(f'Found PPK2 at {ppk2_port}')
+            return ppk2_port
+        else:
+            print(f'Too many connected PPK2\'s: {ppk2s_connected}')
+            return None
+
+    def enable_power(self):
+        """Enable ppk2 power"""
+        if self.ppk2:
+            self.ppk2.toggle_DUT_power("ON")
+            return True
+        return False
+
+    def disable_power(self):
+        """Disable ppk2 power"""
+        if self.ppk2:
+            self.ppk2.toggle_DUT_power("OFF")
+            return True
+        return False
+
+    def measurement_loop(self):
+        """Endless measurement loop will run in a thread"""
+        while True and not self.stop:
+            if self.measuring:  # read data if currently measuring
+                read_data = self.ppk2.get_data()
+                if read_data != b'':
+                    #samples = self.ppk2.get_samples(read_data)
+                    samples = self._average_samples(self.ppk2.get_samples(read_data), 1024)  # optionally average samples
+                    self.current_measurements += samples  # can easily sum lists, will append individual data
+            time.sleep(0.001)  # TODO figure out correct sleep duration
+
+    def _average_samples(self, list, window_size):
+        """Average samples based on window size"""
+        chunks = [list[val:val + window_size] for val in range(0, len(list), window_size)]
+        avgs = []
+        for chunk in chunks:
+            avgs.append(sum(chunk) / len(chunk))
+
+        return avgs
+
+    def start_measuring(self):
+        """Start measuring"""
+        if not self.measuring:  # toggle measuring flag only if currently not measuring
+            self.current_measurements = []  # reset current measurements
+            self.ppk2.start_measuring()  # send command to ppk2
+            self.measuring = True  # set internal flag
+            self.measurement_start_time = time.time()
+
+    def stop_measuring(self):
+        """Stop measuring and return average of period"""
+        self.measurement_stop_time = time.time()
+        self.measuring = False
+        self.ppk2.stop_measuring()  # send command to ppk2
+
+        #samples_average = self._average_samples(self.current_measurements, 1000)
+        if self.filename is not None:
+            self.write_csv_rows(self.current_measurements)
+
+    def get_min_current_mA(self):
+        return min(self.current_measurements) / 1000
+
+    def get_max_current_mA(self):
+        return max(self.current_measurements) / 1000
+
+    def get_average_current_mA(self):
+        """Returns average current of last measurement in mA"""
+        if len(self.current_measurements) == 0:
+            return 0
+
+        average_current_mA = (sum(self.current_measurements) / len(self.current_measurements)) / 1000 # measurements are in microamperes, divide by 1000
+        return average_current_mA
+
+    def get_average_power_consumption_mWh(self):
+        """Return average power consumption of last measurement in mWh"""
+        average_current_mA = self.get_average_current_mA()  # convert microamperes to milliamperes
+        average_power_mW = (self.source_voltage_mV / 1000) * average_current_mA  # divide by 1000 as source voltage is in millivolts - this gives us milliwatts
+        measurement_duration_h = self.get_measurement_duration_s() / 3600  # duration in seconds, divide by 3600 to get hours
+        average_consumption_mWh = average_power_mW * measurement_duration_h
+        return average_consumption_mWh
+
+    def get_average_charge_mC(self):
+        """Returns average charge in milli coulomb"""
+        average_current_mA = self.get_average_current_mA()
+        measurement_duration_s = self.get_measurement_duration_s()  # in seconds
+        return average_current_mA * measurement_duration_s
+
+    def get_measurement_duration_s(self):
+        """Returns duration of measurement"""
+        measurement_duration_s = (self.measurement_stop_time - self.measurement_start_time)  # measurement duration in seconds
+        return measurement_duration_s
+
+# pp = PowerProfiler("/dev/ttyACM1")
+# pp.start_measuring()
+# time.sleep(10)
+# pp.stop_measuring()