From d01bef5172042a3c8c086760ca6861381e5344ea Mon Sep 17 00:00:00 2001 From: malice Date: Sun, 5 Jul 2026 23:01:08 -0600 Subject: [PATCH] fix(serial): detect passed-through /dev USB-serial nodes (container GUI detect) (#53) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pyserial comports() reads /sys USB metadata, absent inside a container for a bind-mounted device node — so /api/serial-ports returned [] and the GUI "Detect" showed nothing despite /dev/meshcore-rak (major 166) being present + openable. Supplement comports() with a direct /dev scan: include char devices whose major is a USB-serial major (166 ttyACM, 188 ttyUSB), catching /dev/ttyACM*/ttyUSB* AND custom udev names like /dev/meshcore-rak that a tty* glob misses; exclude legacy ttyS* (major 4). Merge deduped by realpath (pyserial metadata wins on overlap). stable_path: by-id > stable custom name > by-path > raw. likely_radio heuristic on the name (mesh|rak|lora|tbeam|heltec|nrf|companion) for bare nodes. Resilient (unreadable /dev / stat error skipped, never raises). 8 new tests; suite at 10-failure baseline. Co-authored-by: Matt Johnson Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/serial_ports.py | 153 ++++++++++++++++++++++++++- work/tests/test_serial_ports.py | 176 ++++++++++++++++++++++++++++++++ 2 files changed, 325 insertions(+), 4 deletions(-) diff --git a/work/meshai/serial_ports.py b/work/meshai/serial_ports.py index ae741ab..60f0640 100644 --- a/work/meshai/serial_ports.py +++ b/work/meshai/serial_ports.py @@ -2,16 +2,24 @@ Enumerates available serial ports, resolves stable by-id/by-path symlinks, and flags likely radio devices by USB VID. + +In addition to pyserial's ``comports()`` (which reads ``/sys`` USB metadata and +therefore misses bind-mounted device nodes inside containers), a supplementary +direct ``/dev`` scan finds USB-serial character devices by their device major. +This catches passed-through nodes (e.g. ``/dev/meshcore-rak``, major 166) and +custom udev symlinks that ``comports()`` returns nothing for. """ import os import re +import stat as _stat from serial.tools.list_ports import comports # Module-level dir constants — tests monkeypatch these to tmp_path locations. BY_ID_DIR = "/dev/serial/by-id" BY_PATH_DIR = "/dev/serial/by-path" +DEV_DIR = "/dev" # USB VIDs for known mesh-radio hardware: # 0x239A Adafruit/RAK nRF52840 @@ -21,10 +29,25 @@ BY_PATH_DIR = "/dev/serial/by-path" # 0x55D4 WCH CH9102 (alternate VID) RADIO_VIDS: frozenset[int] = frozenset({0x239A, 0x1915, 0x10C4, 0x1A86, 0x55D4}) +# USB-serial character-device majors (Linux): +# 166 ttyACM* / USB CDC-ACM (RAK nRF52840, native-USB radios) +# 188 ttyUSB* / USB serial (CP210x, CH340, FTDI bridges) +# Legacy ttyS* (major 4, 16550 UART) is deliberately NOT here. +USB_SERIAL_MAJORS: frozenset[int] = frozenset({166, 188}) + # Pattern for ACM/USB tty devices (not legacy ttyS*) _ACMUSB_RE = re.compile(r"tty(ACM|USB)\d") +# Anchored pattern for a RAW ACM/USB name (ttyACM0, ttyUSB1). A device name that +# does NOT match this (e.g. "meshcore-rak") is treated as a stable custom name. +_RAW_ACMUSB_RE = re.compile(r"tty(ACM|USB)\d+$") # Pattern for legacy ttyS ports to exclude _TTYS_RE = re.compile(r"ttyS\d") +# Heuristic: basename words that suggest a mesh radio (used for bare /dev nodes +# that carry no USB VID/PID metadata). +_RADIO_NAME_RE = re.compile( + r"mesh|meshcore|rak|lora|tbeam|t-?beam|heltec|nrf|companion", + re.IGNORECASE, +) def serial_by_id_available() -> bool: @@ -52,12 +75,33 @@ def _build_symlink_map(dirpath: str) -> dict[str, str]: return result +def _char_major(path: str) -> int | None: + """Return the device major of ``path`` if it is a character device. + + Follows symlinks. Returns None when the path can't be stat'd or is not a + character device. Isolated in its own helper so tests can monkeypatch the + stat/major lookup without needing real device nodes. + """ + try: + st = os.stat(path) # follows symlinks + except OSError: + return None + if not _stat.S_ISCHR(st.st_mode): + return None + return os.major(st.st_rdev) + + def list_serial_ports() -> list[dict]: """Enumerate and return usable serial ports. - Includes a port if it has a USB VID/PID OR its device name matches - /dev/tty(ACM|USB). Excludes /dev/ttyS* legacy ports (unless they carry - a vid/pid — edge case handled by checking vid presence first). + Combines two sources: + * pyserial ``comports()`` — rich vid/pid/serial metadata on real hosts. + * a direct ``/dev`` scan — finds USB-serial character devices by major + (166 ttyACM / 188 ttyUSB), catching container-passed-through nodes and + custom udev names that ``comports()`` misses. + + Results are merged and deduped by the real device path; a device found by + both keeps the pyserial metadata. Each entry dict: device, by_id, by_path, description, hwid, vid, pid, @@ -75,7 +119,9 @@ def _scan_ports() -> list[dict]: by_id_map = _build_symlink_map(BY_ID_DIR) by_path_map = _build_symlink_map(BY_PATH_DIR) - ports = [] + ports: list[dict] = [] + seen: set[str] = set() + for p in comports(): device: str = p.device or "" vid: int | None = p.vid @@ -92,6 +138,7 @@ def _scan_ports() -> list[dict]: # Resolve stable path via realpath comparison. real_device = os.path.realpath(device) if device else device + seen.add(real_device) by_id_link: str | None = by_id_map.get(real_device) by_path_link: str | None = by_path_map.get(real_device) @@ -119,4 +166,102 @@ def _scan_ports() -> list[dict]: "stable_path": stable_path, }) + # Supplement with a direct /dev scan for nodes comports() couldn't see. + ports.extend(_scan_dev_ports(seen, by_id_map, by_path_map)) + + return ports + + +def _gather_dev_entries() -> list[str]: + """Return candidate paths to examine: /dev entries (non-recursive) plus the + two /dev/serial/ subdirs when present. Never raises.""" + paths: list[str] = [] + for dirpath in (DEV_DIR, BY_ID_DIR, BY_PATH_DIR): + try: + names = os.listdir(dirpath) + except OSError: + continue + for name in names: + paths.append(os.path.join(dirpath, name)) + return paths + + +def _scan_dev_ports( + seen: set[str], + by_id_map: dict[str, str], + by_path_map: dict[str, str], +) -> list[dict]: + """Scan /dev for USB-serial character devices missed by comports(). + + ``seen`` holds realpaths already emitted by the comports() pass; devices + resolving to one of those are skipped (deduped, pyserial metadata wins). + """ + dev_dir_norm = os.path.normpath(DEV_DIR) + + # Group candidate paths by the real device node they resolve to. + groups: dict[str, set[str]] = {} + for path in _gather_dev_entries(): + try: + real = os.path.realpath(path) + except OSError: + continue + groups.setdefault(real, set()).add(path) + + ports: list[dict] = [] + for real, sources in groups.items(): + if real in seen: + continue # already found via comports() — keep its rich metadata + + major = _char_major(real) + if major not in USB_SERIAL_MAJORS: + continue + + by_id_link = by_id_map.get(real) + by_path_link = by_path_map.get(real) + + # A direct /dev entry whose basename is NOT a raw ttyACM/ttyUSB + # (and not a legacy ttyS) is a stable custom udev name in its own right. + custom_name: str | None = None + for src in sorted(sources): + if os.path.dirname(src) != dev_dir_norm: + continue + base = os.path.basename(src) + if _TTYS_RE.match(base): + continue + if not _RAW_ACMUSB_RE.match(base): + custom_name = src + break + + # stable_path precedence: by-id > stable custom name > by-path > raw. + if by_id_link: + stable_path = by_id_link + elif custom_name: + stable_path = custom_name + elif by_path_link: + stable_path = by_path_link + else: + stable_path = real + + # likely_radio heuristic: any candidate basename hits the radio words. + candidate_names = {os.path.basename(s) for s in sources} + candidate_names.add(os.path.basename(stable_path)) + candidate_names.add(os.path.basename(real)) + likely_radio = any(_RADIO_NAME_RE.search(n) for n in candidate_names) + + seen.add(real) + ports.append({ + "device": real, + "by_id": by_id_link, + "by_path": by_path_link, + "description": os.path.basename(stable_path), + "hwid": "", + "vid": None, + "pid": None, + "serial_number": None, + "manufacturer": None, + "product": None, + "likely_radio": likely_radio, + "stable_path": stable_path, + }) + return ports diff --git a/work/tests/test_serial_ports.py b/work/tests/test_serial_ports.py index eae6c87..2b02db1 100644 --- a/work/tests/test_serial_ports.py +++ b/work/tests/test_serial_ports.py @@ -251,3 +251,179 @@ def test_serial_by_id_available_false(monkeypatch): import meshai.serial_ports as sp monkeypatch.setattr(sp, "BY_ID_DIR", "/does/not/exist/by-id") assert sp.serial_by_id_available() is False + + +# --------------------------------------------------------------------------- +# /dev supplementary scan — detects nodes comports() misses +# --------------------------------------------------------------------------- + +def _major_by_basename(mapping): + """Return a _char_major replacement that looks up majors by basename. + + ``mapping`` maps a basename -> major (int). Unlisted paths return None + (treated as "not a USB-serial char device"). + """ + def _fake(path): + return mapping.get(os.path.basename(path)) + return _fake + + +def _setup_dev_scan(sp, tmp_path, monkeypatch, filenames, majors, comports=()): + """Create a fake /dev dir with ``filenames`` and wire up the module. + + Points DEV_DIR at the tmp dir, disables the by-id/by-path dirs, fakes + _char_major from ``majors`` (basename -> major), and sets comports(). + Returns the tmp dev dir path. + """ + dev = tmp_path / "dev" + dev.mkdir() + for name in filenames: + (dev / name).write_text("") # stand-in for a device node + + monkeypatch.setattr(sp, "DEV_DIR", str(dev)) + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "by-id-none")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "by-path-none")) + monkeypatch.setattr(sp, "_char_major", _major_by_basename(majors)) + monkeypatch.setattr(sp, "comports", lambda: list(comports)) + return dev + + +def test_dev_scan_custom_node_meshcore_rak(tmp_path, monkeypatch): + """A passed-through custom node /dev/meshcore-rak (major 166) with no /sys + backing (comports() returns []) is detected with likely_radio=True and its + own name as the stable path.""" + import meshai.serial_ports as sp + + dev = _setup_dev_scan( + sp, tmp_path, monkeypatch, + filenames=["meshcore-rak"], + majors={"meshcore-rak": 166}, + comports=[], + ) + + result = sp.list_serial_ports() + assert len(result) == 1 + entry = result[0] + assert entry["device"] == str(dev / "meshcore-rak") + assert entry["stable_path"] == str(dev / "meshcore-rak") + assert entry["likely_radio"] is True + assert entry["vid"] is None and entry["pid"] is None + assert entry["serial_number"] is None + + +def test_dev_scan_raw_ttyacm(tmp_path, monkeypatch): + """A raw ttyACM0 (major 166) with no by-id link is included; its basename + doesn't match the radio-name heuristic so likely_radio is False and the + stable path is the raw device.""" + import meshai.serial_ports as sp + + dev = _setup_dev_scan( + sp, tmp_path, monkeypatch, + filenames=["ttyACM0"], + majors={"ttyACM0": 166}, + comports=[], + ) + + result = sp.list_serial_ports() + assert len(result) == 1 + entry = result[0] + assert entry["device"] == str(dev / "ttyACM0") + assert entry["stable_path"] == str(dev / "ttyACM0") + assert entry["likely_radio"] is False + + +def test_dev_scan_ttyusb_included(tmp_path, monkeypatch): + """ttyUSB0 (major 188) is a USB-serial major and is included.""" + import meshai.serial_ports as sp + + dev = _setup_dev_scan( + sp, tmp_path, monkeypatch, + filenames=["ttyUSB0"], + majors={"ttyUSB0": 188}, + comports=[], + ) + + result = sp.list_serial_ports() + assert len(result) == 1 + assert result[0]["stable_path"] == str(dev / "ttyUSB0") + + +def test_dev_scan_ttys_excluded(tmp_path, monkeypatch): + """ttyS0 (major 4, legacy UART) is NOT a USB-serial major → excluded.""" + import meshai.serial_ports as sp + + _setup_dev_scan( + sp, tmp_path, monkeypatch, + filenames=["ttyS0"], + majors={"ttyS0": 4}, + comports=[], + ) + + assert sp.list_serial_ports() == [] + + +def test_dev_scan_dedup_keeps_pyserial_metadata(tmp_path, monkeypatch): + """A device found by BOTH comports() and the /dev scan yields one entry + that keeps the pyserial vid/pid/manufacturer metadata.""" + import meshai.serial_ports as sp + + dev = _setup_dev_scan( + sp, tmp_path, monkeypatch, + filenames=["ttyACM0"], + majors={"ttyACM0": 166}, + comports=[], + ) + device_path = str(dev / "ttyACM0") + + # comports() reports the same node with rich metadata. + port = _fake_port(device_path, vid=0x239A, pid=0x0001) + monkeypatch.setattr(sp, "comports", lambda: [port]) + + result = sp.list_serial_ports() + assert len(result) == 1 + entry = result[0] + assert entry["vid"] == 0x239A + assert entry["pid"] == 0x0001 + assert entry["manufacturer"] == "Acme" + assert entry["serial_number"] == "SN001" + assert entry["likely_radio"] is True + + +def test_dev_scan_custom_symlink_stable_path(tmp_path, monkeypatch): + """A custom udev symlink /dev/meshcore-rak -> ttyACM0 (both in /dev) groups + to one node; the custom name is preferred as the stable path over the raw + ttyACM0, and the radio heuristic fires on the custom name.""" + import meshai.serial_ports as sp + + dev = tmp_path / "dev" + dev.mkdir() + raw = dev / "ttyACM0" + raw.write_text("") + link = dev / "meshcore-rak" + link.symlink_to(raw) + + monkeypatch.setattr(sp, "DEV_DIR", str(dev)) + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "by-id-none")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "by-path-none")) + monkeypatch.setattr(sp, "_char_major", _major_by_basename({"ttyACM0": 166})) + monkeypatch.setattr(sp, "comports", lambda: []) + + result = sp.list_serial_ports() + assert len(result) == 1 + entry = result[0] + # realpath collapses to the raw node; stable_path prefers the custom name. + assert entry["device"] == str(raw) + assert entry["stable_path"] == str(link) + assert entry["likely_radio"] is True + + +def test_dev_scan_unreadable_dir_no_raise(tmp_path, monkeypatch): + """A nonexistent DEV_DIR must not raise — just yields nothing extra.""" + import meshai.serial_ports as sp + + monkeypatch.setattr(sp, "DEV_DIR", "/does/not/exist/dev") + monkeypatch.setattr(sp, "BY_ID_DIR", str(tmp_path / "noid")) + monkeypatch.setattr(sp, "BY_PATH_DIR", str(tmp_path / "nopath")) + monkeypatch.setattr(sp, "comports", lambda: []) + + assert sp.list_serial_ports() == []