Fixed issue where metadata would come with garbage

examples/example_measure_mw.py

@@ -0,0 +1,107 @@
+import time
+import threading
+
+from src.ppk2_api.ppk2_api import PPK2_API
+
+
+class PowerProfiler:
+    conversion_factor = 1_000_000  # uA to mW
+
+    def __init__(self, voltage_mv=3700):
+        self.voltage_mv = voltage_mv
+        self.total_power_mW = 0
+        self.total_samples = 0
+
+        self.lock = threading.Lock()
+        self.ppk2 = None
+        self.sampling_thread = None
+        self.sampling_enabled = False
+
+    def _connect_ppk2(self):
+        ppk2s_connected = []
+        #After reset, the PPK2 can take a moment to reappear on the system. We wait for it to show up for a few seconds before giving up.
+        for _ in range(10):
+            ppk2s_connected = PPK2_API.list_devices()
+            print(f"PPK2 devices found: {ppk2s_connected}")
+            if ppk2s_connected:
+                break
+            time.sleep(0.5)
+
+        if not ppk2s_connected:
+            raise ConnectionError("No PPK2 devices found!")
+
+        print(ppk2s_connected)
+        # Just select the first available PPK2 device
+        for ppk2_port_tuple in ppk2s_connected:
+            ppk2_port = ppk2_port_tuple[0] #Just get the port part of the tuple
+            print(f"Connecting to {ppk2_port}")
+            try:
+                self.ppk2 = PPK2_API(ppk2_port, timeout=1, write_timeout=1, exclusive=True)
+                self.ppk2.get_modifiers()
+                return
+            except Exception:
+                print(f"Failed to connect to {ppk2_port}")
+                self.ppk2 = None
+
+        raise ConnectionError("Could not initialize any detected PPK2 device.")
+
+
+    def _setup_ppk2(self):
+        self._connect_ppk2()
+
+        self.ppk2.set_source_voltage(self.voltage_mv)
+        self.ppk2.use_source_meter()
+        self.ppk2.toggle_DUT_power("ON")
+
+        self.ppk2.start_measuring()
+        print("Initialized Power Profiler")
+
+
+    def _run_sampling(self):
+        try:
+            self._setup_ppk2()
+            while self.sampling_enabled:
+                time.sleep(0.01)
+                read_data = self.ppk2.get_data()
+
+                if read_data == b"":
+                    continue
+
+                samples, raw_digital = self.ppk2.get_samples(read_data)
+                if not samples:
+                    continue
+
+                average_current_uA = sum(samples) / len(samples)
+                average_power_mW = (average_current_uA * self.voltage_mv) / self.conversion_factor
+                formatted_power = round(average_power_mW, 2)
+
+                with self.lock:
+                    self.total_power_mW += formatted_power
+                    self.total_samples += 1
+                    average_of_averages_mW = self.total_power_mW / self.total_samples
+
+                print(f"{formatted_power} mW, Avg: {average_of_averages_mW:.2f} mW")
+
+        except Exception as e:
+            self.sampling_enabled = False
+            print(f"An error occurred: {e}")
+
+    def start_sampling(self):
+        self.sampling_enabled = True
+        self.sampling_thread = threading.Thread(target=self._run_sampling, daemon=True)
+        self.sampling_thread.start()
+
+    def stop_sampling(self):
+        self.sampling_enabled = False
+        self.sampling_thread.join()
+
+
+def main():
+    sampler = PowerProfiler(voltage_mv=3800)
+    sampler.start_sampling()
+    input("Press Enter to exit...\n")
+    sampler.stop_sampling()
+
+
+if __name__ == "__main__":
+    main()

src/power_profiler.py

@@ -26,48 +26,43 @@ def __init__(self, serial_port=None, source_voltage_mV=3300, filename=None):
                 self.ppk2 = PPK2_API(serial_port)
 
         try:
-            ret = self.ppk2.get_modifiers()  # try to read modifiers, if it fails serial port is probably not correct
-            print(f"Initialized ppk2 api: {ret}")
+            self.ppk2.get_modifiers()  # try to read modifiers, if it fails serial port is probably not correct
+            print("Initialized ppk2 api")
         except Exception as e:
             print(f"Error initializing power profiler: {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.ppk2.use_source_meter()
 
-            self.source_voltage_mV = source_voltage_mV
+        self.source_voltage_mV = source_voltage_mV
 
-            self.ppk2.set_source_voltage(self.source_voltage_mV)  # set to 3.3V
+        self.ppk2.set_source_voltage(self.source_voltage_mV)  # set to 3.3V
 
-            print(f"Set power profiler source voltage: {self.source_voltage_mV}")
+        print(f"Set power profiler source voltage: {self.source_voltage_mV}")
 
-            self.measuring = False
-            self.current_measurements = []
+        self.measuring = False
+        self.current_measurements = []
 
-            # local variables used to calculate power consumption
-            self.measurement_start_time = None
-            self.measurement_stop_time = None
+        # local variables used to calculate power consumption
+        self.measurement_start_time = None
+        self.measurement_stop_time = None
 
-            time.sleep(1)
+        time.sleep(1)
 
-            self.stop = False
+        self.stop = False
 
-            self.measurement_thread = Thread(target=self.measurement_loop, daemon=True)
-            self.measurement_thread.start()
+        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)
+        # 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"""
@@ -127,7 +122,7 @@ def measurement_loop(self):
             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, raw_digital = self.ppk2.get_samples(read_data)
                     self.current_measurements += samples  # can easily sum lists, will append individual data
             time.sleep(0.001)  # TODO figure out correct sleep duration
 
@@ -192,4 +187,4 @@ def get_average_charge_mC(self):
     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
+        return measurement_duration_s