Streaming ww dev

benchmark/runner/device_under_test.py

@@ -18,15 +18,15 @@ def __init__(self, port_device, baud_rate=76800, power_manager=None):
     self._max_bytes = 26 if power_manager else 31
 
   def __enter__(self):
-    if self.power_manager:
-      self.power_manager.__enter__()
+    #if self.power_manager:
+      #self.power_manager.__enter__()
     self._port.__enter__()
     return self
 
   def __exit__(self, *args):
     self._port.__exit__(*args)
-    if self.power_manager:
-      self.power_manager.__exit__(*args)
+    #if self.power_manager:
+      #self.power_manager.__exit__(*args)
 
   def _retry(self, method, retries=3):
     for attempt in range(retries):
@@ -95,11 +95,11 @@ def infer(self, number, warmups):
     result = self._port.send_command("db print")
 
     command = f"infer {number} {warmups}"  # must include warmups, even if 0, because default warmups=10
-    if self.power_manager:
-      print(self.power_manager.start())
+    #if self.power_manager:
+      #print(self.power_manager.start())
     result = self._port.send_command(command)
-    if self.power_manager:
-      print(self.power_manager.stop())
+    #if self.power_manager:
+      #print(self.power_manager.stop())
     return result
 
   def get_help(self):

benchmark/runner/devices.yaml

@@ -14,9 +14,10 @@
       - 0x0043
       - 0x0243
 - name: lpm01a
-  usb_description: PowerShield
   type: power
+  usb:
+    0x0483: 0x5740  # Ensures detection by VID/PID (1155 / 22336)
 - name: l4r5zi
   type: dut
   usb:
-    0x0483: 0x374B
+    0x0483: 0x374B

benchmark/runner/dut.yaml

@@ -1 +1 @@
-voltage: 3000m
+voltage: 3300m

benchmark/runner/main.py

@@ -6,16 +6,15 @@
 from sklearn.metrics import roc_auc_score
 from collections import Counter
 
+from power_manager import PowerManager
 from datasets import DataSet
 from device_manager import DeviceManager
 from device_under_test import DUT
 from script import Script
-
 """
 Application to execute test scripts to measure power consumption, turn on and off power, send commands to a device
 under test.
 """
-
 def init_dut(device):
     if device:
         with device as dut:
@@ -25,8 +24,8 @@ def init_dut(device):
             dut.get_profile()
 
 def identify_dut(manager):
-    interface = manager.get("interface", {}).get("instance")
     power = manager.get("power", {}).get("instance")
+    interface = manager.get("interface", {}).get("instance")
     if not manager.get("dut") and interface: # removed and power:
         dut = DUT(interface, power_manager=power)
         manager["dut"] = {
@@ -44,7 +43,6 @@ def run_test(devices_config, dut_config, test_script, dataset_path, mode):
     :param dut_config:
     :param test_script:
     :param dataset_path:
-    :param mode: Test mode (energy, performance, accuracy)
     """
     manager = DeviceManager(devices_config)
     manager.scan()
@@ -53,18 +51,20 @@ def run_test(devices_config, dut_config, test_script, dataset_path, mode):
 
     if power and dut_config and dut_config.get("voltage"):
         power.configure_voltage(dut_config["voltage"])
-    identify_dut(manager)  # hangs in identify_dut()=>init_dut()=>time.sleep()
+    identify_dut(manager)
 
     dut = manager.get("dut", {}).get("instance")
     io = manager.get("interface", {}).get("instance")
 
-    # Create a Script object from the dict that was read from the tests yaml file.
-    script = Script(test_script.get(dut.get_model()))
-
-    # Run the test script with the mode passed in
-    data_set = DataSet(os.path.join(dataset_path, script.model), script.truth)
-    return script.run(io, dut, data_set, mode)  # Pass mode to the run method
+    # with io:
+    #   start_time = time.time()
+    #   io.play_wave("cd16m.wav")
+    #   elapsed = time.time() - start_time
 
+    script = Script(test_script.get(dut.get_model()))
+    set = DataSet(os.path.join(dataset_path, script.model), script.truth)
+    result = script.run(io, dut, set, mode)
+    return result, power
 
 def parse_device_config(device_list_file, device_yaml):
     """Parsee the device discovery configuration
@@ -167,11 +167,11 @@ def calculate_auc(y_pred, labels, n_classes):
 
 
 # Summarize results
-def summarize_result(result):
+def summarize_result(result, mode, power):
     num_correct_files = 0
     total_files = 0
-    true_labels = []
-    predicted_probabilities = []
+    y_pred = []
+    y_true = []
 
     file_infer_results = {}
 
@@ -193,34 +193,34 @@ def summarize_result(result):
             infer_results = normalize_probabilities(infer_results)
             file_infer_results[file_name]['results'].append(infer_results)
 
-    y_pred = []
-    y_true = []
-
-    for file_name, data in file_infer_results.items():
-        true_class = data['true_class']
-        results = data['results']
+    if mode != 'e':
+        for file_name, data in file_infer_results.items():
+            true_class = data['true_class']
+            results = data['results']
 
-        class_counts = Counter([np.argmax(res) for res in results])
-        majority_class = class_counts.most_common(1)[0][0]
+            class_counts = Counter([np.argmax(res) for res in results])
+            majority_class = class_counts.most_common(1)[0][0]
 
-        if majority_class == true_class:
-            num_correct_files += 1
+            if majority_class == true_class:
+                num_correct_files += 1
 
-        averaged_result = np.mean(results, axis=0)
-        y_pred.append(averaged_result)
-        y_true.append(true_class)
+            averaged_result = np.mean(results, axis=0)
+            y_pred.append(averaged_result)
+            y_true.append(true_class)
 
-        total_files += 1
+            total_files += 1
 
-    accuracy = calculate_accuracy(np.array(y_pred), np.array(y_true))
-    auc = calculate_auc(np.array(y_pred), np.array(y_true), n_classes)
+        calculate_accuracy(np.array(y_pred), np.array(y_true))
+        calculate_auc(np.array(y_pred), np.array(y_true), n_classes)
+    else:
+        #power manager output
+        print("Power Edition Output")
+        power.stop()  # Stop power capture
+        power.send_command_wait_for_response("pwr off")
 
-    print(f"File-based accuracy = {accuracy:.2f}")
-    print(f"AUC = {auc}")
 
-    return accuracy, auc
+    return
 
-
 if __name__ == '__main__':
     parser = argparse.ArgumentParser(prog="TestRunner", description=__doc__)
     parser.add_argument("-d", "--device_list", default="devices.yaml", help="Device definition YAML file")
@@ -239,5 +239,6 @@ def summarize_result(result):
         "dataset_path": args.dataset_path,
         "mode": args.mode
     }
-    result = run_test(**config)
-    summarize_result(result)
+    result, power = run_test(**config)  # Unpack power from run_test
+    summarize_result(result, args.mode, power)  # Pass power to summarize_result
+