Merge branch 'master' into sreckamp/benchmark-interface

benchmark/reference_submissions/.gitignore

@@ -1,2 +1,3 @@
 *.zip*
 main.cpp
+

benchmark/runner/.gitignore

@@ -0,0 +1,3 @@
+.idea
+venv
+

benchmark/runner/README.md

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

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

@@ -0,0 +1,100 @@
+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, 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"):
+      match = re.match(r'^m-(name)-dut-\[([^]]+)]$', l)
+      if match:
+        self.__setattr__(f"_{match.group(1)}", match.group(2))
+
+  def get_name(self):
+    if self._name is None:
+      self._get_name()
+    return self._name
+
+  def _get_profile(self):
+    for l in self._port.send_command("profile"):
+      match = re.match(r'^m-(model|profile)-\[([^]]+)]$', l)
+      if match:
+        self.__setattr__(f"_{match.group(1)}", match.group(2))
+
+  def get_model(self):
+    if self._model is None:
+      self._get_profile()
+    return self._model
+
+  def get_profile(self):
+    if self._profile is None:
+      self._get_profile()
+    return self._profile
+
+  def timestamp(self):
+    return self._port.send_command("timestamp")
+
+  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)
+    return self._port.send_command("results")
+
+  def get_help(self):
+    return self._port.send_command("help")
+
+  """
+  ULPMark for tinyML Firmware V0.0.1
+
+  help         : Print this information
+  name         : Print the name of the device
+  timestsamp   : Generate a timetsamp
+  db SUBCMD    : Manipulate a generic byte buffer
+    load N     : Allocate N bytes and set load counter
+    db HH[HH]* : Load 8-bit hex byte(s) until N bytes
+    print [N=16] [offset=0]
+               : Print N bytes at offset as hex
+  infer N [W=0]: Load input, execute N inferences after W warmup loops
+  results      : Return the result fp32 vector
+  """

benchmark/runner/devices.yaml

@@ -0,0 +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

benchmark/runner/dut.yaml

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

benchmark/runner/interface_device.py

@@ -0,0 +1,3 @@
+class InterfaceDevice:
+  def send_command(self, command, end=None, echo=False):
+    pass

benchmark/runner/io_manager.py

@@ -0,0 +1,62 @@
+from interface_device import InterfaceDevice
+from serial_device import SerialDevice
+
+
+class IOManager(InterfaceDevice):
+  def __init__(self, port_device, baud_rate=115200):
+    self.port = SerialDevice(port_device, baud_rate, "m-ready", '%')
+    self.entry_count = 0
+
+  def __enter__(self):
+    if not self.entry_count:
+      self.port.__enter__()
+      self.get_name()
+    self.entry_count += 1
+    return self
+
+  def __exit__(self, *args):
+    self.entry_count -= 1
+    if not self.entry_count:
+      self.port.__exit__(*args)
+
+  def get_name(self):
+    return self.port.send_command("name")
+
+  def timestamp(self):
+    return self.port.send_command("timestamp")
+
+  def get_help(self):
+    return self.port.send_command("help")
+
+  def send_data(self, data):
+    size = len(data)
+    pass
+
+  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]: ", "")
+      lines.append(resp)
+    return lines if len(lines) != 1 else lines[0]
+
+  """
+   help
+   name              device name
+   res               Reset polarity 0*,1
+   enable-timer      Enable timer ISR
+   disable-timer     Disable ISR
+   dut               DUT passthrough
+   i2c-enable        N : Enable I2C slave-mode to send N bytes (mod16)
+   i2c-disable       Disable above
+   load-vdata        Load 16 random bytes into vdata
+   show-vdata        Show vdata
+   et                Start Emon
+   tm                0=fall/1*=change
+   tp                [res]
+   version           firmware version
+  """

benchmark/runner/io_manager_enhanced.py

@@ -0,0 +1,16 @@
+from io_manager import IOManager
+
+
+class IOManagerEnhanced(IOManager):
+  def __init__(self, port_device, baud_rate=921600):
+    IOManager.__init__(self, port_device, baud_rate)
+
+  def get_files(self):
+    return self.port.send_command('ls')
+
+  def get_waves(self):
+    return [x for x in self.get_files() if x.lower().endswith("wav") or x == "m-ready"]
+
+  def play_wave(self, filename=None):
+    command = "play" + (f" {filename}" if filename else "")
+    return self.port.send_command(command)