mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
Compare commits
1 commit
main
...
fix/serial
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
33adc721d9 |
2 changed files with 325 additions and 4 deletions
|
|
@ -2,16 +2,24 @@
|
||||||
|
|
||||||
Enumerates available serial ports, resolves stable by-id/by-path symlinks,
|
Enumerates available serial ports, resolves stable by-id/by-path symlinks,
|
||||||
and flags likely radio devices by USB VID.
|
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 os
|
||||||
import re
|
import re
|
||||||
|
import stat as _stat
|
||||||
|
|
||||||
from serial.tools.list_ports import comports
|
from serial.tools.list_ports import comports
|
||||||
|
|
||||||
# Module-level dir constants — tests monkeypatch these to tmp_path locations.
|
# Module-level dir constants — tests monkeypatch these to tmp_path locations.
|
||||||
BY_ID_DIR = "/dev/serial/by-id"
|
BY_ID_DIR = "/dev/serial/by-id"
|
||||||
BY_PATH_DIR = "/dev/serial/by-path"
|
BY_PATH_DIR = "/dev/serial/by-path"
|
||||||
|
DEV_DIR = "/dev"
|
||||||
|
|
||||||
# USB VIDs for known mesh-radio hardware:
|
# USB VIDs for known mesh-radio hardware:
|
||||||
# 0x239A Adafruit/RAK nRF52840
|
# 0x239A Adafruit/RAK nRF52840
|
||||||
|
|
@ -21,10 +29,25 @@ BY_PATH_DIR = "/dev/serial/by-path"
|
||||||
# 0x55D4 WCH CH9102 (alternate VID)
|
# 0x55D4 WCH CH9102 (alternate VID)
|
||||||
RADIO_VIDS: frozenset[int] = frozenset({0x239A, 0x1915, 0x10C4, 0x1A86, 0x55D4})
|
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*)
|
# Pattern for ACM/USB tty devices (not legacy ttyS*)
|
||||||
_ACMUSB_RE = re.compile(r"tty(ACM|USB)\d")
|
_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
|
# Pattern for legacy ttyS ports to exclude
|
||||||
_TTYS_RE = re.compile(r"ttyS\d")
|
_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:
|
def serial_by_id_available() -> bool:
|
||||||
|
|
@ -52,12 +75,33 @@ def _build_symlink_map(dirpath: str) -> dict[str, str]:
|
||||||
return result
|
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]:
|
def list_serial_ports() -> list[dict]:
|
||||||
"""Enumerate and return usable serial ports.
|
"""Enumerate and return usable serial ports.
|
||||||
|
|
||||||
Includes a port if it has a USB VID/PID OR its device name matches
|
Combines two sources:
|
||||||
/dev/tty(ACM|USB). Excludes /dev/ttyS* legacy ports (unless they carry
|
* pyserial ``comports()`` — rich vid/pid/serial metadata on real hosts.
|
||||||
a vid/pid — edge case handled by checking vid presence first).
|
* 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:
|
Each entry dict:
|
||||||
device, by_id, by_path, description, hwid, vid, pid,
|
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_id_map = _build_symlink_map(BY_ID_DIR)
|
||||||
by_path_map = _build_symlink_map(BY_PATH_DIR)
|
by_path_map = _build_symlink_map(BY_PATH_DIR)
|
||||||
|
|
||||||
ports = []
|
ports: list[dict] = []
|
||||||
|
seen: set[str] = set()
|
||||||
|
|
||||||
for p in comports():
|
for p in comports():
|
||||||
device: str = p.device or ""
|
device: str = p.device or ""
|
||||||
vid: int | None = p.vid
|
vid: int | None = p.vid
|
||||||
|
|
@ -92,6 +138,7 @@ def _scan_ports() -> list[dict]:
|
||||||
|
|
||||||
# Resolve stable path via realpath comparison.
|
# Resolve stable path via realpath comparison.
|
||||||
real_device = os.path.realpath(device) if device else device
|
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_id_link: str | None = by_id_map.get(real_device)
|
||||||
by_path_link: str | None = by_path_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,
|
"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<N>/ttyUSB<N>
|
||||||
|
# (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
|
return ports
|
||||||
|
|
|
||||||
|
|
@ -251,3 +251,179 @@ def test_serial_by_id_available_false(monkeypatch):
|
||||||
import meshai.serial_ports as sp
|
import meshai.serial_ports as sp
|
||||||
monkeypatch.setattr(sp, "BY_ID_DIR", "/does/not/exist/by-id")
|
monkeypatch.setattr(sp, "BY_ID_DIR", "/does/not/exist/by-id")
|
||||||
assert sp.serial_by_id_available() is False
|
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() == []
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue