Run the image_classification test

benchmark/runner/device_manager.py

@@ -0,0 +1,73 @@
+import yaml
+
+from serial.tools import list_ports
+
+from device_under_test import DUT
+from io_manager import IOManager
+from io_manager_enhanced import IOManagerEnhanced
+from power_manager import PowerManager
+
+
+class DeviceManager:
+  def __init__(self, devices):
+    self._device_defs = devices
+
+  def __getitem__(self, item):
+    return self.__dict__[item]
+
+  def __setitem__(self, key, value):
+    self.__dict__[key] = value
+
+  def get(self, item, default=None):
+    return self.__dict__.get(item, default)
+
+  def values(self):
+    return (a for a in self.__dict__.values() if isinstance(a, dict))
+
+  def _add_device(self, usb, definition):
+    type = definition.get("type")
+    if type and (not self.__dict__.get(type)
+                 or definition.get("preference", 0) > self.__dict__[type].get("preference", 0)):
+      self.__dict__[type] = {k: v for k, v in definition.items()}
+      self.__dict__[type]["port"] = usb.device
+
+  def _instantiate(self, definition):
+    args = {
+      "port_device": definition.get("port")
+    }
+    if definition.get("baud"):
+      args["baud_rate"] = definition.get("baud")
+    if definition.get("type") == "interface":
+      definition["instance"] = IOManagerEnhanced(**args) if definition.get("name") == "stm32h573i-dk" \
+        else IOManager(**args)
+    elif definition.get("type") == "power":
+      definition["instance"] = PowerManager(**args)
+    elif definition.get("type") == "dut":
+      definition["instance"] = DUT(**args)
+
+  def scan(self):
+    """Scan fpr USB serial devices
+    This scans the connected usb devices.  It compares the device definitions
+    based on the USB vid and pid primarily and then by the text of the description.
+    """
+    pending = [p for p in list_ports.comports(True) if p.vid]
+    matched = []
+    for p in pending:
+      for d in self._device_defs:
+        found = False
+        for vid, pids in d.get("usb", {}).items():
+          for pid in (pids if isinstance(pids, list) else [pids]):
+            if pid == p.pid and vid == p.vid:
+              self._add_device(p, d)
+              matched.append(p)
+              found = True
+              break
+        if found: break
+    for p in (a for a in pending if a not in matched):
+      for d in (d1 for d1 in self._device_defs if d1.get("usb_description", "zZzZzZzZ") in p.description):
+        self._add_device(p, d)
+        matched.append(p)
+        break
+
+    for d in self.values():
+      self._instantiate(d)

benchmark/runner/device_under_test.py

@@ -1,25 +1,32 @@
 import re
+import sys
+import time
 
 from interface_device import InterfaceDevice
 from serial_device import SerialDevice
 
 
 class DUT:
-  def __init__(self, port_device, baud_rate=115200):
+  def __init__(self, port_device, baud_rate=115200, power_manager=None):
     interface = port_device
     if not isinstance(port_device, InterfaceDevice):
       interface = SerialDevice(port_device, baud_rate, "m-ready", '%')
     self._port = interface
+    self._power_manager = power_manager
     self._profile = None
     self._model = None
     self._name = None
 
   def __enter__(self):
+    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__()
 
   def _get_name(self):
     for l in self._port.send_command("name"):
@@ -55,12 +62,24 @@ def send_data(self, data):
     size = len(data)
     pass
 
+  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])}"
+      result = self._port.send_command(cmd)
+      print(f"{result} ({i})")
+      i += 32
+    # print("", file=sys.stderr)
+
   def infer(self, number, warmups):
     command = f"infer {number}"
     if warmups:
       command += f" {warmups}"
     self._port.send_command(command)
-    self._port.send_command("results")
+    return self._port.send_command("results")
 
   def get_help(self):
     return self._port.send_command("help")

benchmark/runner/devices.yaml

@@ -1,20 +1,20 @@
-  - name: stm32h573i-dk
-    usb_description: STLINK
-    type: interface
-    preference: 2
-    usb:
-        0x0483: 0x374E
-  - name: arduino
-    port: auto
-    type: interface
-    preference: 1
-    usb:
-        0x2341:
-          - 0x0043
-          - 0x0001
-        0x2a03:
-          - 0x0043
-          - 0x0243
-  - name: lpm01a
-    usb_description: PowerShield
-    type: power
+- name: stm32h573i-dk
+  usb_description: STLINK
+  type: interface
+  preference: 2
+  usb:
+    0x0483: 0x374E
+- name: arduino
+  port: auto
+  type: interface
+  preference: 1
+  usb:
+    0x2341:
+      - 0x0043
+      - 0x0001
+    0x2a03:
+      - 0x0043
+      - 0x0243
+- name: lpm01a
+  usb_description: PowerShield
+  type: power

benchmark/runner/dut.yaml

@@ -0,0 +1 @@
+voltage: 3000m