factored out device manifest to be accessed via debug interface; upda…

…tes #1303

.gitignore

@@ -59,4 +59,4 @@ fp-info-cache
 
 #performance test data
 test_data/
-device_list.json
+device_list.*

software/controller/controller.sh

@@ -107,7 +107,7 @@ clean_all() {
 }
 
 install_linux() {
-  pip3 install pyserial matplotlib pandas gitpython numpy pytest gdown
+  pip3 install pyserial matplotlib pandas gitpython numpy pytest
   pip3 install platformio==${PIO_VERSION}
   source ${HOME}/.profile
 }
@@ -233,9 +233,8 @@ launch_browser() {
 # get_serial_sn <alias>
 # prints ST-Link serial number by defined alias
 get_serial_sn() {
-  device_id="$1"
-  serial_sn=$(awk '$1 == LOOKUPVAL { print $2 }' "LOOKUPVAL=$device_id" platformio/device_lookup_table.txt)
-  echo "$serial_sn"
+  device_alias="$1"
+  ../utils/debug/debug_cli.py -c "devices find $device_alias h"
 }
 
 # build <target_name>
@@ -496,17 +495,13 @@ elif [ "$1" == "integrate" ]; then
   else
     deploy_integration_test "${@:2}"
   fi
-
   exit $EXIT_SUCCESS
 
 ###########
 # DEVICES #
 ###########
 elif [ "$1" == "devices" ]; then
-  pushd ../utils/debug/util
-  ./serial_detect.py
-  popd
-#  udevadm info /dev/ttyACM* | grep SERIAL_SHORT | awk -F  "=" '{print $2}'
+  ../utils/debug/debug_cli.py -c "devices list"
   exit $EXIT_SUCCESS
 
 ################

software/utils/debug/debug_cli.py

@@ -176,12 +176,12 @@ def do_debug(self, line):
             print("Unknown command; pass 'on' or 'off'.")
 
     def maybe_connect(self, alias_or_port):
-        devices = DeviceScanner("device_list.json").filter(connected=True)
+        devices = DeviceScanner(load_manifest=True).filter(connected=True)
         selected = None
-        if not alias_or_port:
+        if alias_or_port == "auto":
             print("Auto-selecting device")
             selected = devices.auto_select()
-        else:
+        elif alias_or_port and len(alias_or_port):
             print(f"Looking for device with alias={alias_or_port}")
             selected = devices.get(alias_or_port)
             if not selected:
@@ -191,27 +191,31 @@ def maybe_connect(self, alias_or_port):
             print(f"Attempting to connect to: {selected.print()}")
             self.interface.connect(selected.port)
 
-    def do_connect(self, line):
+    def do_devices(self, line):
         """This command manages the connection of the debug interface to the ventilator controller
 
         If ventilator USB cable is unplugged, the CLI prompt should update to reflect this on the next loop
         (blank line or failed command). You may also start the debugger without a plugged in device.
         Use one of the following commands to reconnect.
 
-        connect list
+        devices list
           searches and lists available STM32 devices on the serial bus
 
-        connect off
+        devices off
           disconnect if connected
 
-        connect auto
-          attempt to connect to a plugged in controller automatically
-
-        connect <alias>
-          connect to controller pointed to by alias in device manifest
+        devices find <alias> [h/p/c]
+          print information for specific device
+          alias - string must match alias as defined in manifest
+          h - print HLA serial number only
+          p - print port only
+          c - print configuration only
 
-        connect <port>
-          connect to controller on a specific port
+        devices connect <device>
+          connect to device, where <device> is one of:
+          - `auto` find port of and connect to plugged in device, must be only one
+          - alias, as definded in device manifest
+          - port name, e.g. /dev/ttyACM0
 
         """
 
@@ -222,25 +226,39 @@ def do_connect(self, line):
         subcommand = params[0]
 
         if subcommand == "list":
-            print(DeviceScanner("device_list.json").list_devices())
+            print(DeviceScanner(load_manifest=True).list_devices())
 
         elif subcommand == "off":
             self.interface.disconnect()
             self.interface.resynchronize()
             self.interface.variables_update_info()
             self.update_prompt()
 
-        elif subcommand == "auto":
-            self.maybe_connect(None)
+        elif subcommand == "connect":
+            self.maybe_connect(params[1])
             self.interface.resynchronize()
             self.interface.variables_update_info()
             self.update_prompt()
 
+        elif subcommand == "find":
+            alias = params[1]
+            devices = DeviceScanner(load_manifest=True)
+            device = devices.get(alias)
+            if not device:
+                print(f"No definition for device '{alias}'")
+                return
+            if len(params) > 2:
+                if params[2] == "h":
+                    print(device.hla_serial)
+                elif params[2] == "p":
+                    print(device.port)
+                elif params[2] == "c":
+                    print(device.configuration)
+            else:
+                print(device.print())
+
         else:
-            self.maybe_connect(subcommand)
-            self.interface.resynchronize()
-            self.interface.variables_update_info()
-            self.update_prompt()
+            print("Invalid device args: {}", params)
 
     def help_run(self):
         print(
@@ -840,14 +858,14 @@ def do_eeprom(self, line):
             print("Error: Unknown subcommand %s" % cl[0])
             return
 
+
 def main():
     terminal_parser = argparse.ArgumentParser()
     terminal_parser.add_argument(
         "--device",
         "-d",
         type=str,
-        help="Serial port device is connected to, e.g. /dev/ttyACM0, or alias from manifest."
-        " If unspecified, we try to auto-detect.",
+        help="Device to connect to: serial, e.g. /dev/ttyACM0, or alias from manifest, or 'auto'.",
     )
     terminal_parser.add_argument(
         "--command", "-c", type=str, help="Run the given command and exit."

software/utils/debug/util/devices.py

@@ -0,0 +1,8 @@
+#!/usr/bin/env python3
+
+import pandas
+import json
+
+url = "https://docs.google.com/spreadsheets/d/e/2PACX-1vRduOfterWmAy_xrc356rRhjz4QDLgOScgG1VPx2-KNeH8zYEe29SCw_DKOJG-5hqSO6BXmG1BumUul/pub?gid=0&single=true&output=tsv"
+data = json.loads(pandas.read_csv(url, sep="\t", header=0).to_json(orient="records"))
+print(json.dumps(data, indent=4))

software/utils/debug/util/serial_detect.py

@@ -23,10 +23,13 @@
 """
 
 import json
-import gdown
+import pandas
 import subprocess
 from typing import List
-from colors import green, red, gray
+from util.colors import green, red, gray
+
+MANIFEST_URL = "https://docs.google.com/spreadsheets/d/e/2PACX-1vRduOfterWmAy_xrc356rRhjz4QDLgOScgG1VPx2-KNeH8zYEe29SCw_DKOJG-5hqSO6BXmG1BumUul/pub?gid=0&single=true&output=tsv"
+
 
 class DeviceInfo:
     alias: str
@@ -54,13 +57,14 @@ def print(self):
         else:
             return green(output_string)
 
+
 class DeviceScanner:
     manifest: List
 
-    def __init__(self, file_name = ""):
+    def __init__(self, load_manifest=True):
         self.manifest = []
-        if file_name:
-            self.get_devices(file_name)
+        if load_manifest:
+            self.get_devices()
 
     @staticmethod
     def detect_stm32_ports():
@@ -75,21 +79,23 @@ def detect_stm32_ports():
                 ports[port] = hla_serial
         return ports
 
-
     @staticmethod
     def check_value(data, val):
         return any(entry["hla_serial"] == val for entry in data)
 
-    def get_devices(self, file_name):
+    def get_devices(self):
         ports = DeviceScanner.detect_stm32_ports()
         ports_inverted = {v: k for k, v in ports.items()}
-        json_file = open(file_name, "r")
-        loaded_manifest = json.load(json_file)
+        loaded_manifest = json.loads(
+            pandas.read_csv(MANIFEST_URL, sep="\t", header=0).to_json(orient="records")
+        )
         for entry in loaded_manifest:
-            dev = DeviceInfo(alias=entry.get("alias"),
-                             hla_serial=entry.get("hla_serial"),
-                             configuration=entry.get("configuration"),
-                             description=entry.get("description"))
+            dev = DeviceInfo(
+                alias=entry.get("Alias"),
+                hla_serial=entry.get("HLA serial"),
+                configuration=entry.get("Configuration"),
+                description=entry.get("Description"),
+            )
             if dev.hla_serial in ports_inverted:
                 dev.port = ports_inverted[dev.hla_serial]
             self.manifest.append(dev)
@@ -100,7 +106,7 @@ def get_devices(self, file_name):
                 self.manifest.append(dev)
 
     def filter(self, known=False, connected=False):
-        filtered = DeviceScanner()
+        filtered = DeviceScanner(load_manifest=False)
         for m in self.manifest:
             if not len(m.alias) and known:
                 continue
@@ -129,27 +135,27 @@ def find(self, port):
 
     def auto_select(self):
         if not self.manifest:
-            print(red(
-                "Could not auto-detect serial port; "
-                "platformio device list did not yield any STM32 devices."
-            ))
+            print(
+                red(
+                    "Could not auto-detect serial port; "
+                    "platformio device list did not yield any STM32 devices."
+                )
+            )
             return None
         if len(self.manifest) > 1:
-            print(red(
-                "Could not auto-detect serial port; "
-                "platformio device list yielded multiple STM32 devices:\n"
-                f"{self.list_devices()}"
-            ))
+            print(
+                red(
+                    "Could not auto-detect serial port; "
+                    "platformio device list yielded multiple STM32 devices:\n"
+                    f"{self.list_devices()}"
+                )
+            )
             return None
         return self.manifest[0]
-        # "Choose port explicitly with --port."
 
-if __name__ == "__main__":
-    url = "https://drive.google.com/file/d/11Hsspy9_VFy6HVnV8iIkUo0Y2vPf67Lm/view?usp=sharing"
-    output = "device_list.json"
-    gdown.download(url=url, output=output, quiet=True, fuzzy=True)
 
-    devices = DeviceScanner("device_list.json")
+if __name__ == "__main__":
+    devices = DeviceScanner(load_manifest=True)
     print("ALL: ")
     print(devices.list_devices())
     print("ONLY KNOWN: ")