CLI: Show Port in Status Output

Either read the busid from sysfs, when the USB device is attached or
calculate the possible busids, when the device is not attached.

not_my_board/_agent.py

@@ -80,6 +80,14 @@ def usbip_refresh_status():
     def usbip_is_attached(vhci_port):
         return usbip.is_attached(vhci_port)
 
+    @staticmethod
+    def usbip_port_num_to_busid(port_num):
+        return usbip.port_num_to_busid(port_num)
+
+    @staticmethod
+    def usbip_vhci_port_to_busid(vhci_port):
+        return usbip.vhci_port_to_busid(vhci_port)
+
     async def usbip_attach(self, proxy, target, port_num, usbid):
         tunnel = self._http.open_tunnel(*proxy, *target)
         async with tunnel as (reader, writer, trailing_data):
@@ -267,6 +275,7 @@ async def status(self):
                 "interface": tunnel.iface_name,
                 "type": tunnel.type_name,
                 "attached": tunnel.is_attached(),
+                "port": tunnel.port,
             }
             for name, reservation in self._reservations.items()
             for tunnel in reservation.tunnels.values()
@@ -487,13 +496,25 @@ def is_attached(self):
             return False
         return self._io.usbip_is_attached(self._vhci_port)
 
+    @property
+    def port(self):
+        if self.is_attached():
+            return self._io.usbip_vhci_port_to_busid(self._vhci_port)
+        else:
+            usbids = self._io.usbip_port_num_to_busid(self.port_num)
+            return "/".join(usbids)
+
 
 class _TcpTunnel(_Tunnel):
     async def _task_func(self):
         await self._io.port_forward(
             self._ready_event, self._proxy, self.remote, self.local_port
         )
 
+    @property
+    def port(self):
+        return str(self.local_port)
+
 
 @dataclass(frozen=True)
 class _TunnelDesc:

not_my_board/_usbip.py

@@ -20,7 +20,7 @@
 
 
 logger = logging.getLogger(__name__)
-_vhci_status_attached = {}
+_vhci_status = {}
 
 
 class UsbIpServer(util.ContextStack):
@@ -85,13 +85,42 @@ def __init__(self, default=None):
         super().__init__(16, default)
 
 
-class UsbIpDevice(util.ContextStack):
+class _UsbDevice:
+    def __init__(self, sysfs_path):
+        self._sysfs_path = sysfs_path
+
+    @property
+    def speed(self):
+        string_to_code = {
+            "1.5": 1,
+            "12": 2,
+            "480": 3,
+            "53.3-480": 4,
+            "5000": 5,
+        }
+        string = (self._sysfs_path / "speed").read_text()[:-1]  # strip newline
+        return string_to_code.get(string, 0)
+
+    busnum = _SysfsFileInt()
+    devnum = _SysfsFileInt()
+    idVendor = _SysfsFileHex()  # noqa: N815
+    idProduct = _SysfsFileHex()  # noqa: N815
+    bcdDevice = _SysfsFileHex()  # noqa: N815
+    bDeviceClass = _SysfsFileHex()  # noqa: N815
+    bDeviceSubClass = _SysfsFileHex()  # noqa: N815
+    bDeviceProtocol = _SysfsFileHex()  # noqa: N815
+    bConfigurationValue = _SysfsFileHex(default=0)  # noqa: N815
+    bNumConfigurations = _SysfsFileHex()  # noqa: N815
+    bNumInterfaces = _SysfsFileHex(default=0)  # noqa: N815
+
+
+class UsbIpDevice(_UsbDevice, util.ContextStack):
     def __init__(self, busid):
         self._busid = busid
-        self._sysfs_path = pathlib.Path("/sys/bus/usb/devices/") / busid
         self._lock = asyncio.Lock()
         self._refresh_event = asyncio.Event()
         self._is_exported = False
+        super().__init__(pathlib.Path("/sys/bus/usb/devices/") / busid)
 
     def refresh(self):
         self._refresh_event.set()
@@ -187,30 +216,7 @@ def busid(self):
     def path(self):
         return self._sysfs_path.as_posix().encode("utf-8")
 
-    @property
-    def speed(self):
-        string_to_code = {
-            "1.5": 1,
-            "12": 2,
-            "480": 3,
-            "53.3-480": 4,
-            "5000": 5,
-        }
-        string = (self._sysfs_path / "speed").read_text()[:-1]  # strip newline
-        return string_to_code.get(string, 0)
-
     usbip_status = _SysfsFileInt()
-    busnum = _SysfsFileInt()
-    devnum = _SysfsFileInt()
-    idVendor = _SysfsFileHex()  # noqa: N815
-    idProduct = _SysfsFileHex()  # noqa: N815
-    bcdDevice = _SysfsFileHex()  # noqa: N815
-    bDeviceClass = _SysfsFileHex()  # noqa: N815
-    bDeviceSubClass = _SysfsFileHex()  # noqa: N815
-    bDeviceProtocol = _SysfsFileHex()  # noqa: N815
-    bConfigurationValue = _SysfsFileHex(default=0)  # noqa: N815
-    bNumConfigurations = _SysfsFileHex()  # noqa: N815
-    bNumInterfaces = _SysfsFileHex(default=0)  # noqa: N815
 
 
 async def _exec(*args, **kwargs):
@@ -256,7 +262,7 @@ async def attach(reader, writer, busid, port_num):
 def _port_num_to_vhci_port(port_num, speed):
     """Map port_num and speed to port, that is passed to Kernel
 
-    example with vhci_nr_hcs=2 and nports=8:
+    Example with vhci_nr_hcs=2 and nports=8:
 
         vhci_hcd    hub   speed  port_num  vhci_port
         vhci_hcd.0  usb1  hs     0         0
@@ -293,6 +299,40 @@ def _port_num_to_vhci_port(port_num, speed):
     return vhci_port
 
 
+def port_num_to_busid(port_num):
+    """Map port_num to busid
+
+    Example with vhci_nr_hcs=2 and nports=8:
+
+        vhci_hcd    hub   port_num  busid
+        vhci_hcd.0  usb5  0         5-1
+        vhci_hcd.0  usb5  1         5-2
+        vhci_hcd.0  usb6  0         6-1
+        vhci_hcd.0  usb6  1         6-2
+        vhci_hcd.1  usb7  2         7-1
+        vhci_hcd.1  usb7  3         7-2
+        vhci_hcd.1  usb8  2         8-1
+        vhci_hcd.1  usb8  3         8-2
+    """
+    platform_path = pathlib.Path("/sys/devices/platform")
+    vhci_nr_hcs = len(list(platform_path.glob("vhci_hcd.*")))
+    nports = int((platform_path / "vhci_hcd.0/nports").read_text())
+
+    # calculate number of ports each vhci_hcd.* has
+    vhci_ports = nports // vhci_nr_hcs
+    # calculate number of ports each hub has
+    vhci_hc_ports = vhci_ports // 2
+
+    vhci_hcd_nr = port_num // vhci_hc_ports
+
+    devnum = port_num - (vhci_hcd_nr * vhci_hc_ports) + 1
+
+    vhci_hcd = platform_path / f"vhci_hcd.{vhci_hcd_nr}"
+    for hub in vhci_hcd.glob("usb[0-9]*/"):
+        hub = _UsbDevice(hub)
+        yield f"{hub.busnum}-{devnum}"
+
+
 def detach(vhci_port):
     detach_path = pathlib.Path("/sys/devices/platform/vhci_hcd.0/detach")
 
@@ -301,6 +341,12 @@ def detach(vhci_port):
         detach_path.write_text(f"{vhci_port}")
 
 
+@dataclasses.dataclass
+class _VhciStatus:
+    attached: bool
+    busid: str
+
+
 def refresh_vhci_status():
     vhci_path = pathlib.Path("/sys/devices/platform/vhci_hcd.0")
     if not vhci_path.exists():
@@ -327,11 +373,16 @@ def status_paths():
                 entries = line.split()
                 port = int(entries[1])
                 status = int(entries[2])
-                _vhci_status_attached[port] = status == status_attached
+                busid = entries[6]
+                _vhci_status[port] = _VhciStatus(status == status_attached, busid)
+
+
+def is_attached(vhci_port):
+    return _vhci_status[vhci_port].attached
 
 
-def is_attached(port):
-    return _vhci_status_attached[port]
+def vhci_port_to_busid(vhci_port):
+    return _vhci_status[vhci_port].busid
 
 
 async def _ensure_vhci_hcd_driver_available():

not_my_board/cli/__init__.py

@@ -236,16 +236,16 @@ async def _status_command(args):
         if args.no_header:
             headers = []
         else:
-            headers = ["Place", "Part", "Type", "Interface", "Status"]
+            headers = ["Place", "Part", "Type", "Interface", "Status", "Port"]
             headers[0] = f"{Format.BOLD}{headers[0]}"
             headers[-1] = f"{headers[-1]}{Format.RESET}"
 
         def row(entry):
-            keys = ["place", "part", "type", "interface"]
+            keys = ["place", "part", "type", "interface", "port"]
             row = [entry[k] for k in keys]
             status = f"{Format.GREEN}Up" if entry["attached"] else f"{Format.RED}Down"
             status += Format.RESET
-            row.append(status)
+            row.insert(-1, status)
             return row
 
         table = [row(entry) for entry in status_list]

tests/test_agent.py

@@ -156,6 +156,14 @@ def usbip_refresh_status():
     def usbip_is_attached(self, vhci_port):
         return vhci_port in self.attached
 
+    @staticmethod
+    def usbip_port_num_to_busid(_):
+        return ["2-1", "3-1"]
+
+    @staticmethod
+    def usbip_vhci_port_to_busid(_):
+        return "2-1"
+
     async def usbip_attach(self, proxy, target, port_num, usbid):
         if port_num in self.detach_event:
             await self.detach_event[port_num].wait()
@@ -246,13 +254,15 @@ async def test_status_place_1(agent_io):
         "interface": "usb0",
         "type": "USB",
         "attached": False,
+        "port": "2-1/3-1",
     }
     ssh_status = {
         "place": "fake",
         "part": "fake-board",
         "interface": "ssh",
         "type": "TCP",
         "attached": False,
+        "port": "2222",
     }
     assert usb0_status in status
     assert ssh_status in status
@@ -266,6 +276,7 @@ async def test_status_place_1_attached(agent_io):
     assert len(status) == 2
     assert status[0]["attached"] is True
     assert status[1]["attached"] is True
+    assert "2-1" in [s["port"] for s in status]
 
 
 async def test_reserve_twice(agent_io):

tests/test_usbip.py

@@ -44,8 +44,22 @@ async def test_usb_forwarding(vms):
                 header = status_lines[0].split()
                 status_line = status_lines[1].split()
 
-                assert header == ["Place", "Part", "Type", "Interface", "Status"]
-                assert status_line == ["qemu-usb", "flash-drive", "USB", "usb0", "Up"]
+                assert header == [
+                    "Place",
+                    "Part",
+                    "Type",
+                    "Interface",
+                    "Status",
+                    "Port",
+                ]
+                assert status_line == [
+                    "qemu-usb",
+                    "flash-drive",
+                    "USB",
+                    "usb0",
+                    "Up",
+                    "2-1",
+                ]
 
                 try:
                     await vms.exporter.usb_detach()
@@ -54,9 +68,15 @@ async def test_usb_forwarding(vms):
                     await vms.exporter.usb_attach()
 
                 await vms.client.ssh_poll("test -e /sys/bus/usb/devices/2-1")
-                await vms.client.ssh("not-my-board detach qemu-usb")
+                await vms.client.ssh("not-my-board detach -k qemu-usb")
                 await vms.client.ssh("! test -e /sys/bus/usb/devices/2-1")
 
+                result = await vms.client.ssh("not-my-board status")
+                status_str = result.stdout.rstrip()
+                status_lines = status_str.split("\n")
+                status_line = status_lines[1].split()
+                assert status_line[5] == "1-1/2-1"
+
     # When the exporter is killed, then it should clean up and restore the
     # default USB driver.
     result = await vms.exporter.ssh("readlink /sys/bus/usb/devices/2-1/driver")