Functional updates

benchmark/runner/.gitignore

@@ -1,3 +1,4 @@
 .idea
+datasets
 venv
 

benchmark/runner/README.md

@@ -5,7 +5,7 @@
 ### Power Board (LPM01A)
 ![LPM01A Wiring](img/LPM01A.jpg)
 ### Interface Board (STM32H573I-DK)
-![STM32H573I-DK Top Wiring](img/STM32H573I-DK-Top.jpg)
+![STM32H573I-DK Top Wiring](img/STM32H573I-DK-Top.png)
 ![STM32H573I-DK Bottom Wiring](img/STM32H573I-DK-Bottom.png)
 ### Device Under Test (L4R5ZI)
 ![DUT Wiring](img/L4R5ZI.png)

benchmark/runner/datasets.py

@@ -0,0 +1,27 @@
+import csv
+import os
+from random import randrange
+
+
+class DataSet:
+  def __init__(self, dataset_path, truth_file):
+    self._dataset_path = dataset_path
+    self._truth_file = truth_file
+    self._truth_data = None
+
+  def _read_truth_file(self):
+    if not self._truth_data:
+      with open(os.path.join(self._dataset_path, self._truth_file)) as file:
+        reader = csv.DictReader(file, fieldnames=["file", "classes", "class"])
+        self._truth_data = [f for f in reader]
+
+  def get_file_by_index(self, index):
+    self._read_truth_file()
+    data = []
+    index = index if index is not None else randrange(len(self._truth_data))
+    truth = None
+    if index < len(self._truth_data):
+      truth = self._truth_data[index]
+      with open(os.path.join(self._dataset_path, truth.get("file")), "rb") as file:
+        data = file.read()
+    return truth, data

benchmark/runner/device_under_test.py

@@ -1,6 +1,5 @@
 import re
 import sys
-import time
 
 from interface_device import InterfaceDevice
 from serial_device import SerialDevice
@@ -12,21 +11,22 @@ def __init__(self, port_device, baud_rate=115200, power_manager=None):
     if not isinstance(port_device, InterfaceDevice):
       interface = SerialDevice(port_device, baud_rate, "m-ready", '%')
     self._port = interface
-    self._power_manager = power_manager
+    self.power_manager = power_manager
     self._profile = None
     self._model = None
     self._name = 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__()
+    if self.power_manager:
+      self.power_manager.__exit__(*args)
 
   def _get_name(self):
     for l in self._port.send_command("name"):
@@ -66,20 +66,21 @@ def load(self, data):
     self._port.send_command(f"db load {len(data)}")
     i = 0
     while i < len(data):
-      time.sleep(0.5)
-      # print(".", end='', file=sys.stderr)
-      cmd = f"db {''.join(f'{d:02x}' for d in data[i:i+32])}"
+      cmd = f"db {''.join(f'{d:02x}' for d in data[i:i+self._max_bytes])}"
       result = self._port.send_command(cmd)
-      print(f"{result} ({i})")
-      i += 32
-    # print("", file=sys.stderr)
+      i += self._max_bytes
+    return result
 
   def infer(self, number, warmups):
     command = f"infer {number}"
     if warmups:
       command += f" {warmups}"
-    self._port.send_command(command)
-    return self._port.send_command("results")
+    if self.power_manager:
+      print(self.power_manager.start())
+    result = self._port.send_command(command)
+    if self.power_manager:
+      print(self.power_manager.stop())
+    return result
 
   def get_help(self):
     return self._port.send_command("help")

benchmark/runner/devices.yaml

@@ -1,11 +1,9 @@
 - name: stm32h573i-dk
-  usb_description: STLINK
   type: interface
   preference: 2
   usb:
     0x0483: 0x374E
 - name: arduino
-  port: auto
   type: interface
   preference: 1
   usb:
@@ -18,3 +16,7 @@
 - name: lpm01a
   usb_description: PowerShield
   type: power
+- name: l4r5zi
+  type: dut
+  usb:
+    0x0483: 0x374B

benchmark/runner/io_manager.py

@@ -32,15 +32,19 @@ def send_data(self, data):
     size = len(data)
     pass
 
+  def read_line(self):
+    resp = self.port.read_line()
+    resp = resp.replace("[dut]: ", "")
+    return resp;
+
   def send_command(self, command, end=None, echo=False):
     resp = self.port.send_command(f"dut {command}")
     if len(resp) != 2 or resp[1] != "m-ready":
       return None
     resp = None
     lines = []
     while resp != 'm-ready':
-      resp = self.port.read_line()
-      resp = resp.replace("[dut]: ", "")
+      resp = self.read_line()
       lines.append(resp)
     return lines if len(lines) != 1 else lines[0]
 

benchmark/runner/main.py

@@ -1,65 +1,12 @@
 import argparse
-import csv
+import os
 import time
+import yaml
 
+from datasets import DataSet
 from device_manager import DeviceManager
 from device_under_test import DUT
-
-from io_manager_enhanced import IOManagerEnhanced
-from power_manager import PowerManager
-from io_manager import IOManager
-
-from contextlib import nullcontext, ExitStack
-from serial.tools import list_ports
-import yaml
-
-
-def test_power(device):
-  with device as power:
-    pass
-
-
-def run_dut_test():
-  """
-  power on
-  m-ready
-  -mode energy
-  imit-done
-  m-ready
-  :return:
-  dut passthrough: profile
-  """
-
-
-def test_dut(device):
-  if device:
-    with device as dut:
-      print(dut.get_name())
-      print(dut.get_model())
-      print(dut.get_profile())
-      print(dut.timestamp())
-      for l in dut.get_help():
-        print(l)
-
-
-def test_io_manager(device):
-  with device as io:
-    print(io.get_name())
-    for l in io.get_help():
-      print(l)
-
-
-def test_io_manager_enhanced(device):
-  with device as io:
-    print(io.get_name())
-    print(io.get_waves())
-    print(io.play_wave())
-    print(io.play_wave("spaceship.wav"))
-    # waves = io.list_waves()
-    # io.play_wave();
-    # for w in waves:
-    #     print(w)
-    #     # io.play_wave(w)
+from script import Script
 
 
 def init_dut(device):
@@ -69,6 +16,7 @@ def init_dut(device):
     dut.get_model()
     dut.get_profile()
 
+
 def identify_dut(manager):
   interface = manager.get("interface", {}).get("instance")
   power = manager.get("power", {}).get("instance")
@@ -82,31 +30,21 @@ def identify_dut(manager):
   init_dut(dut)
 
 
-def run_test(devices_config, dut_config, test_script):
+def run_test(devices_config, dut_config, test_script, dataset_path):
   manager = DeviceManager(devices_config)
   manager.scan()
-  if manager.get("power", {}).get("instance") and dut_config and dut_config.get("voltage"):
-    manager["power"]["instance"].configure_voltage(dut_config["voltage"])
+  power = manager.get("power", {}).get("instance")
+  if power and dut_config and dut_config.get("voltage"):
+    power.configure_voltage(dut_config["voltage"])
   identify_dut(manager)
+
   dut = manager.get("dut", {}).get("instance")
-  test = test_script.get(dut.get_model())
-  with open("/Users/sreckamp/eembc/runner/benchmarks/ulp-mlperf/datasets/ic01/y_labels.csv") as file:
-    reader = csv.DictReader(file, fieldnames=["file", "classes", "class"])
-    file_name = reader.__next__()["file"]
-  with open(f"/Users/sreckamp/eembc/runner/benchmarks/ulp-mlperf/datasets/ic01/{file_name}", mode="rb") as file:
-    data = file.read()
-  with dut:
-    dut.load(data)
-    print(dut.infer(10, 1))
-  # for _def, i in ((t, t.get("instance")) for t in manager.values() if t and t.get("instance")):
-  #   if isinstance(i, PowerManager):
-  #     test_power(i)
-  #   elif isinstance(i, IOManagerEnhanced):
-  #     test_io_manager_enhanced(i)
-  #   elif isinstance(i, IOManager):
-  #     test_io_manager(i)
-  #   elif isinstance(i, DUT):
-  #     test_dut(i)
+  io = manager.get("interface", {}).get("instance")
+
+  script = Script(test_script.get(dut.get_model()))
+  set = DataSet(os.path.join(dataset_path, script.model), script.truth)
+
+  return script.run(io, dut, set)
 
 
 def parse_device_config(device_list, device_yaml):
@@ -143,10 +81,12 @@ def parse_test_script(test_script):
   parser.add_argument("-v", "--dut_voltage", required=False)
   parser.add_argument("-b", "--dut_baud", required=False)
   parser.add_argument("-t", "--test_script", default="tests.yaml")
+  parser.add_argument("-s", "--dataset_path", default="datasets")
   args = parser.parse_args()
   config = {
     "devices_config": parse_device_config(args.device_list, args.device_yaml),
     "dut_config": parse_dut_config(args.dut, args.dut_voltage, args.dut_baud),
-    "test_script": parse_test_script(args.test_script)
+    "test_script": parse_test_script(args.test_script),
+    "dataset_path": args.dataset_path
   }
-  run_test(**config)
+  print(run_test(**config))