From b692cce4633f6871ff5087145681565bf8c2b038 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Wed, 8 Jul 2026 17:52:30 -0400 Subject: [PATCH] Confirm devices by smoke test on the startup splash Reserve the "MeshCore device" tag for hardware we have actually talked to, rather than guessing from the USB vendor ID: - DeviceStore now keeps a registry of every confirmed device (keyed by stable serial id), remembered forever, migrating the old single-record file transparently. - Selecting a device on the splash runs an immediate smoke test (smoke_test_meshcore): a genuine companion is remembered and confirmed; anything else re-prompts with a message to choose another. The probe uses a short handshake window so a non-MeshCore port is rejected cleanly in a few seconds instead of blocking on meshcore's full default timeout. - The probe renders in place on the chromeless splash via a new BusyScreen (animated Braille spinner) driven by session.busy_startup, so the wordmark and box stay put instead of a separate progress dialog. - Silence the meshcore library's console logging (its import calls logging.basicConfig, which printed "INFO:meshcore:..." over the TUI): route it to the JSON-lines file log with a NullHandler guard on root. - The devices tool now marks confirmed devices and stops asserting "LoRa" before a device is proven. Co-Authored-By: Claude Opus 4.8 --- meshterm/core/connection.py | 65 +++++++++++++++- meshterm/core/device_store.py | 131 +++++++++++++++++++++++-------- meshterm/persistence/logging.py | 38 +++++++++ meshterm/tools/devices.py | 38 +++++---- meshterm/ui/device_picker.py | 134 ++++++++++++++++++++++++-------- meshterm/ui/menu.py | 10 ++- meshterm/ui/surface.py | 38 +++++++++ meshterm/ui/tui/screen.py | 41 ++++++++++ meshterm/ui/tui/session.py | 49 +++++++++++- tests/test_discovery.py | 38 +++++++++ tests/test_tui.py | 101 +++++++++++++++++++++++- 11 files changed, 594 insertions(+), 89 deletions(-) diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index ff7e562..bf3bee6 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -398,15 +398,23 @@ class MeshCoreDevice(Device): firmware version, as event payload field names vary between releases. """ - def __init__(self, port: str, baudrate: int = 115200) -> None: + def __init__( + self, port: str, baudrate: int = 115200, connect_timeout: Optional[float] = None + ) -> None: """Initialize the device wrapper. Args: port: Serial port path (e.g. ``COM5`` or ``/dev/ttyUSB0``). baudrate: Serial baud rate. + connect_timeout: Handshake timeout (seconds) for the initial connection, passed + to the client as its default command timeout. ``None`` uses the ``meshcore`` + library default (~15s). The startup smoke test sets a short value so a + non-responsive port is rejected quickly instead of blocking on the full + default handshake window. """ self._port = port self._baudrate = baudrate + self._connect_timeout = connect_timeout self._mc = None # type: ignore[var-annotated] # meshcore.MeshCore # Serializes channel reads. The meshcore library's get_channel waits for "the next # CHANNEL_INFO event" with no correlation to the index it asked for, and the dispatcher @@ -427,7 +435,18 @@ class MeshCoreDevice(Device): "--mock for the simulator." ) from exc - self._mc = await MeshCore.create_serial(self._port, self._baudrate) + self._mc = await MeshCore.create_serial( + self._port, self._baudrate, default_timeout=self._connect_timeout + ) + # ``create_serial`` returns ``None`` (after cleaning up its own connection) when the + # node never answers the identity handshake — i.e. the port isn't a MeshCore serial + # companion. Surface that as a clean, recoverable error rather than leaving a + # half-open device whose next command fails with a confusing "not connected". + if self._mc is None: + raise DeviceCommandError( + f"no response from a MeshCore companion on {self._port}; it may not be a " + "MeshCore device, or it may be powered off or in use by another program." + ) async def disconnect(self) -> None: # noqa: D102 - inherited docstring if self._mc is None: @@ -1650,3 +1669,45 @@ def make_device( "No serial port configured. Pass --port, set a profile, or use --mock." ) return MeshCoreDevice(port=port, baudrate=baudrate) + + +async def smoke_test_meshcore( + port: str, baudrate: int = 115200, *, timeout: float = 6.0 +) -> Optional[dict]: + """Briefly open ``port`` and confirm a MeshCore companion answers there. + + Opens a real serial connection and issues a single identity query (the APPSTART that + backs :meth:`MeshCoreDevice.get_self_info`). A genuine companion replies with a + self-info payload; anything else — a non-MeshCore serial gadget, the wrong baud rate, + an unresponsive port — never answers and is cut off by ``timeout``. The connection is + always closed before returning, so this is a non-destructive probe used by the startup + picker to verify a chosen device before committing to it. + + Args: + port: Serial port to probe (e.g. ``COM5`` or ``/dev/ttyUSB0``). + baudrate: Serial baud rate. + timeout: Seconds to wait for the connection and for the identity reply. + + Returns: + The device's self-info dict on success (so the caller can also learn its mesh node + name from the same handshake), or ``None`` if it is not a reachable MeshCore + companion. + """ + # ``timeout`` is the handshake window handed to the client, so a non-MeshCore port is + # rejected in ~``timeout`` seconds and the client cleans up its own connection. The outer + # ``wait_for`` is only a safety net a few seconds beyond that, so we never cancel the + # client mid-handshake (which would leak the open port). A real companion answers in well + # under a second, so this never delays a good device. + device = MeshCoreDevice(port=port, baudrate=baudrate, connect_timeout=timeout) + info: Optional[dict] = None + try: + await asyncio.wait_for(device.connect(), timeout + 4.0) + info = await asyncio.wait_for(device.get_self_info(), timeout) + except Exception: # noqa: BLE001 - any failure just means "not confirmed" + info = None + finally: + try: + await device.disconnect() + except Exception: # noqa: BLE001 - best-effort cleanup of the probe connection + pass + return dict(info) if info else None diff --git a/meshterm/core/device_store.py b/meshterm/core/device_store.py index 7a41184..ab21c0e 100644 --- a/meshterm/core/device_store.py +++ b/meshterm/core/device_store.py @@ -1,10 +1,16 @@ -"""Persistence for the last successfully connected companion device. +"""Persistence for confirmed companion devices. -The "last known good" device is remembered in a small JSON file in the config directory -(``/devices.json``), deliberately *not* in the SQLite database: the database -can be swapped per-invocation with ``--db``, whereas the remembered device is global -machine state that should survive that. The record is matched back to currently attached -hardware by :attr:`~meshterm.core.discovery.DiscoveredDevice.stable_id`. +Every device that has ever spoken the MeshCore protocol to us successfully is remembered +forever in a small JSON registry in the config directory (``/devices.json``), +keyed by :attr:`~meshterm.core.discovery.DiscoveredDevice.stable_id` (a USB serial number +when available, so the record survives a changed COM number). This is deliberately *not* in +the SQLite database: the database can be swapped per-invocation with ``--db``, whereas the +remembered devices are global machine state that should survive that. + +The registry also tracks which device was most recently connected (``last``), used to +preselect and star a default on the startup splash. Membership in the registry is what marks +a device as a confirmed MeshCore companion — the splash reserves its "MeshCore device" tag +for these, rather than guessing from the USB vendor ID. """ from __future__ import annotations @@ -20,7 +26,7 @@ from .models import utcnow @dataclass(slots=True) class RememberedDevice: - """A previously connected device recorded for next-time defaulting. + """A previously confirmed device recorded for next-time defaulting. Attributes: stable_id: The device's :attr:`DiscoveredDevice.stable_id` at connect time. @@ -42,7 +48,7 @@ class RememberedDevice: class DeviceStore: - """Reads and writes the remembered "last known good" device.""" + """Reads and writes the registry of confirmed companion devices.""" def __init__(self, path: Path) -> None: """Open the store against a JSON file location. @@ -52,59 +58,118 @@ class DeviceStore: """ self._path = path - def load(self) -> Optional[RememberedDevice]: - """Return the remembered device, or ``None`` if absent or unreadable. + def _read(self) -> tuple[dict[str, RememberedDevice], Optional[str]]: + """Return the parsed ``(registry, last_id)``; empty on a missing/corrupt file. - A missing or corrupt file is treated as "nothing remembered" rather than an - error, so a stray edit never blocks startup. - - Returns: - The :class:`RememberedDevice`, or ``None``. + A missing or corrupt file is treated as "nothing remembered" rather than an error, + so a stray edit never blocks startup. An old flat-format file (a single record at + the top level, from before the registry) is migrated in-memory to a one-entry + registry so upgrades are seamless. """ try: data = json.loads(self._path.read_text(encoding="utf-8")) except (OSError, ValueError): + return {}, None + if not isinstance(data, dict): + return {}, None + + # Old flat shape: a single record with ``stable_id`` at the top level. + if "devices" not in data and "stable_id" in data: + record = self._record_from(data) + if record is None: + return {}, None + return {record.stable_id: record}, record.stable_id + + registry: dict[str, RememberedDevice] = {} + for entry in (data.get("devices") or {}).values(): + record = self._record_from(entry) + if record is not None: + registry[record.stable_id] = record + last = data.get("last") + if last not in registry: + last = None + return registry, last + + @staticmethod + def _record_from(entry: object) -> Optional[RememberedDevice]: + """Build a :class:`RememberedDevice` from one raw JSON entry, or ``None`` if invalid.""" + if not isinstance(entry, dict): return None try: return RememberedDevice( - stable_id=data["stable_id"], - port=data["port"], - label=data.get("label", data["port"]), - last_connected=data.get("last_connected", ""), - node_name=data.get("node_name", ""), + stable_id=entry["stable_id"], + port=entry["port"], + label=entry.get("label", entry["port"]), + last_connected=entry.get("last_connected", ""), + node_name=entry.get("node_name", ""), ) except (KeyError, TypeError): return None + def load(self) -> Optional[RememberedDevice]: + """Return the most recently connected device, or ``None`` if none is remembered. + + This is the "last known good" default used to preselect and star a row on the + startup splash and to resolve a port non-interactively. + + Returns: + The last-connected :class:`RememberedDevice`, or ``None``. + """ + registry, last = self._read() + return registry.get(last) if last is not None else None + + def load_all(self) -> dict[str, RememberedDevice]: + """Return the full registry of confirmed devices, keyed by ``stable_id``.""" + registry, _ = self._read() + return registry + + def is_known(self, device: DiscoveredDevice) -> bool: + """Return whether ``device`` has ever been confirmed as a MeshCore companion.""" + registry, _ = self._read() + return device.stable_id in registry + def remember(self, device: DiscoveredDevice, *, node_name: str = "") -> None: - """Record ``device`` as the last known good connection. + """Record ``device`` as a confirmed connection and the new default. + + Upserts the device into the registry (so it is remembered forever) and marks it as + the most recently connected one. Args: device: The device that just connected successfully. node_name: The device's own mesh node name, if known; preserved across reconnects and shown in the picker. A blank value keeps any name already - on file rather than erasing it. + on file for this device rather than erasing it. """ - if not node_name: - existing = self.load() - if existing is not None and existing.matches(device): - node_name = existing.node_name - record = RememberedDevice( + registry, _ = self._read() + existing = registry.get(device.stable_id) + if not node_name and existing is not None: + node_name = existing.node_name + registry[device.stable_id] = RememberedDevice( stable_id=device.stable_id, port=device.port, label=device.label, last_connected=utcnow().isoformat(), node_name=node_name, ) + self._write(registry, device.stable_id) + + def _write(self, registry: dict[str, RememberedDevice], last: str) -> None: + """Persist the registry, marking ``last`` as the most recently connected device.""" self._path.parent.mkdir(parents=True, exist_ok=True) payload = { - "stable_id": record.stable_id, - "port": record.port, - "label": record.label, - "last_connected": record.last_connected, - "node_name": record.node_name, + "devices": { + stable_id: { + "stable_id": record.stable_id, + "port": record.port, + "label": record.label, + "last_connected": record.last_connected, + "node_name": record.node_name, + } + for stable_id, record in registry.items() + }, + "last": last, } - # Atomic-ish replace so a crash mid-write can't truncate the existing record. + # Atomic-ish replace so a crash mid-write can't truncate the existing registry. tmp = self._path.with_suffix(self._path.suffix + ".tmp") tmp.write_text(json.dumps(payload, indent=2), encoding="utf-8") tmp.replace(self._path) diff --git a/meshterm/persistence/logging.py b/meshterm/persistence/logging.py index 77a3bae..8ab2002 100644 --- a/meshterm/persistence/logging.py +++ b/meshterm/persistence/logging.py @@ -76,9 +76,47 @@ def configure_logging( file_handler.setFormatter(JsonlFormatter()) logger.addHandler(file_handler) + _quiet_library_console(file_handler) + return logger +#: Third-party loggers that must never write to the terminal — their records would print +#: straight over the full-screen TUI. Routed to the file handler instead so they stay +#: captured for debugging. ``meshcore`` is the notable offender: it calls +#: ``logging.basicConfig(level=INFO)`` on import (e.g. "INFO:meshcore:Serial Connection +#: started"). +_LIBRARY_LOGGERS: tuple[str, ...] = ("meshcore",) + + +def _quiet_library_console(file_handler: logging.Handler) -> None: + """Keep noisy third-party libraries off the terminal (they corrupt the TUI). + + The ``meshcore`` client calls ``logging.basicConfig(level=INFO)`` when it is imported, + which installs a stream handler on the *root* logger that prints its INFO lines straight + over the screen. Two guards prevent that: + + 1. Give the root logger a :class:`~logging.NullHandler` so ``basicConfig`` — which only + acts when the root has no handlers — becomes a no-op and never adds its stream handler. + 2. Take each known library logger off propagation and attach only ``file_handler``, so + its records are still captured to the JSON-lines log but never reach the console, even + if some other code path installs a root stream handler anyway. + + Args: + file_handler: The application's file handler to also capture library records to. + """ + root = logging.getLogger() + if not any(isinstance(h, logging.NullHandler) for h in root.handlers): + root.addHandler(logging.NullHandler()) + + for name in _LIBRARY_LOGGERS: + lib = logging.getLogger(name) + lib.handlers.clear() + lib.setLevel(logging.INFO) + lib.propagate = False + lib.addHandler(file_handler) + + def get_logger(name: Optional[str] = None) -> logging.Logger: """Return a child of the application logger. diff --git a/meshterm/tools/devices.py b/meshterm/tools/devices.py index 1c829fd..bab3576 100644 --- a/meshterm/tools/devices.py +++ b/meshterm/tools/devices.py @@ -1,9 +1,10 @@ """The ``devices`` tool: enumerate attached serial companion devices. Discovery never opens the radio — it only lists what is attached. This is a read-only -inventory in both the menu and the CLI; it flags likely LoRa hardware and marks the active -and remembered devices. Selecting a companion is a startup-only concern: pass ``--port`` on -the CLI (remembered after it connects), or pick from the prompt shown when the menu launches. +inventory in both the menu and the CLI; it marks devices already confirmed as MeshCore +companions (and the active one). Selecting a companion is a startup-only concern: pass +``--port`` on the CLI (remembered after it connects), or pick from the prompt shown when the +menu launches (which smoke-tests the choice before confirming it). """ from __future__ import annotations @@ -17,14 +18,19 @@ from ..context import AppContext from ..core.discovery import discover_devices from .base import Tool, ToolResult, register -#: "LoRa?" table cell per discovery confidence tier: a bare serial bridge is only a "maybe", -#: since the same chip shows up on plenty of non-LoRa hardware. -_LORA_CELL: dict[str, str] = { - "board": "[ok]yes[/ok]", +#: "MeshCore?" table cell per discovery confidence tier, for devices we have *not* yet +#: confirmed. The USB vendor ID is only a hint — a native-USB board or a bare bridge chip +#: is a "maybe", never a "yes" — so nothing is billed as MeshCore until a connection proves +#: it (see ``_confirmed_cell``). +_MAYBE_CELL: dict[str, str] = { + "board": "[warn]maybe[/warn]", "bridge": "[warn]maybe[/warn]", "unknown": "[muted]—[/muted]", } +#: "MeshCore?" cell for a device already confirmed to speak the protocol. +_CONFIRMED_CELL = "[ok]yes[/ok]" + @register class DevicesTool(Tool): @@ -46,6 +52,7 @@ class DevicesTool(Tool): A :class:`ToolResult` summarizing how many devices were found. """ devices = discover_devices() + known = ctx.device_store.load_all() remembered = ctx.device_store.load() active = ctx.selected_device active_port = ctx.port_override or (active.port if active else None) @@ -60,6 +67,7 @@ class DevicesTool(Tool): "vendor": d.vendor_label, "likely_lora": d.is_likely_lora, "confidence": d.confidence, + "confirmed": d.stable_id in known, "serial_number": d.serial_number, "stable_id": d.stable_id, "remembered": remembered is not None and remembered.matches(d), @@ -78,30 +86,30 @@ class DevicesTool(Tool): return ToolResult(summary={"count": 0}) table = Table(title="Serial devices", border_style="muted", expand=False) - table.add_column("", style="ok", no_wrap=True) # active/remembered markers + table.add_column("", style="ok", no_wrap=True) # active/confirmed markers table.add_column("Port", style="brand") table.add_column("Device") table.add_column("Vendor", style="muted") - table.add_column("LoRa?", justify="center") + table.add_column("MeshCore?", justify="center") table.add_column("Serial", style="muted") for d in devices: - is_remembered = remembered is not None and remembered.matches(d) + confirmed = known.get(d.stable_id) # the remembered record, if ever confirmed is_active = d.port == active_port - marker = ("●" if is_active else "") + ("★" if is_remembered else "") + marker = ("●" if is_active else "") + ("★" if confirmed else "") device_name = d.product or d.description or "[muted]?[/muted]" - if is_remembered and remembered.node_name: # the mesh name learned on connect - device_name = f"{remembered.node_name} [muted]({device_name})[/muted]" + if confirmed and confirmed.node_name: # the mesh name learned on connect + device_name = f"{confirmed.node_name} [muted]({device_name})[/muted]" table.add_row( marker, d.port, device_name, d.vendor_label or "[muted]?[/muted]", - _LORA_CELL[d.confidence], + _CONFIRMED_CELL if confirmed else _MAYBE_CELL[d.confidence], d.serial_number or "[muted]—[/muted]", ) ctx.ui.show(table) ctx.ui.note( - "[muted]● active ★ remembered default. " + "[muted]● active ★ confirmed MeshCore device. " "Pass --port to select on the CLI.[/muted]" ) return ToolResult(summary={"count": len(devices)}) diff --git a/meshterm/ui/device_picker.py b/meshterm/ui/device_picker.py index 5639c1c..9815d76 100644 --- a/meshterm/ui/device_picker.py +++ b/meshterm/ui/device_picker.py @@ -3,20 +3,23 @@ Shown once at the start of the interactive menu when no port was given explicitly. Unlike the in-menu prompts, it is drawn as a chromeless splash — the MeshTerm wordmark centered above a content-sized box, with no header/footer status bars. It lists the discovered devices in -aligned columns, marks the remembered "last known good" one, and preselects it as the default. -The chosen device becomes the session's active device; it is persisted as the new default once -it actually connects. +aligned columns, tags the ones already confirmed as MeshCore companions, marks the remembered +"last known good" one, and preselects it as the default. + +Selecting a device runs an immediate smoke test (via the ``verify`` callback): a genuine +MeshCore companion is remembered — forever — as confirmed and becomes the session's active +device; anything else sends the user back to the list to choose another. """ from __future__ import annotations from datetime import datetime -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Awaitable, Callable, Optional from rich.cells import cell_len from rich.text import Text -from ..core.device_store import RememberedDevice +from ..core.device_store import DeviceStore, RememberedDevice from ..core.discovery import DiscoveredDevice from .logo import load_logo from .tui import Choice, Separator @@ -24,19 +27,16 @@ from .tui import Choice, Separator if TYPE_CHECKING: from .surface import Ui +#: A smoke test: probe a chosen device and return its self-info dict if it is a MeshCore +#: companion, else ``None``. Supplied by the caller so this UI module stays free of the +#: connection layer. +Verify = Callable[[DiscoveredDevice], Awaitable[Optional[dict]]] + def _copyright() -> str: """The splash's muted copyright line, dated to the current year.""" return f"© {datetime.now().year} Johnputer" -#: Trailing tag per discovery confidence tier (see :attr:`DiscoveredDevice.confidence`). -#: A bare serial bridge is only a weak hint, so it is not billed as a LoRa device. -_TAG: dict[str, str] = { - "board": "· LoRa device", - "bridge": "· serial adapter", - "unknown": "", -} - def _pad(text: str, width: int) -> str: """Right-pad ``text`` with spaces to ``width`` display cells (wide-char aware).""" @@ -52,21 +52,34 @@ def _hardware_name(device: DiscoveredDevice) -> str: return name +def _node_name_from(info: dict) -> str: + """Return a device's own mesh node name from its self-info payload, or ``""``.""" + return str(info.get("adv_name") or info.get("name") or "") + + async def prompt_device( ui: "Ui", devices: list[DiscoveredDevice], - remembered: Optional[RememberedDevice], + store: DeviceStore, + verify: Verify, ) -> Optional[DiscoveredDevice]: """Prompt the user to choose a companion device on the startup splash. + The chosen device is smoke-tested before it is accepted: only a device that answers the + MeshCore identity query is returned (and recorded as confirmed). A device that fails the + test re-opens the picker with a message asking the user to choose another. + Args: ui: The interactive UI surface used to render the picker. devices: Discovered devices (likely-LoRa first). - remembered: The remembered default, if any, used to mark and preselect a row. + store: The confirmed-device registry, used to tag/preselect known devices and to + record a device once its smoke test passes. + verify: Async smoke test returning a device's self-info dict, or ``None`` if it is + not a reachable MeshCore companion. Returns: - The chosen :class:`DiscoveredDevice`, or ``None`` if the user skipped the picker - (e.g. pressed Esc to run against ``--mock`` / configure later). + The chosen, confirmed :class:`DiscoveredDevice`, or ``None`` if the user skipped the + picker (e.g. pressed Esc to run against ``--mock`` / configure later). """ if not devices: await ui.notify_startup( @@ -80,18 +93,76 @@ async def prompt_device( ) return None - # The row for each device leads with its display name (the remembered node's name when - # known, else the hardware name); columns are padded to a shared width so they align. + remembered = store.load() + known: set[str] = set(store.load_all()) + # Preselect the remembered "last known good" device when it is currently attached. + default = next((d for d in devices if remembered and remembered.matches(d)), None) + + while True: + chosen = await ui.select_startup( + "Select a companion device", + _build_items(devices, remembered, known), + default=default, + banner=load_logo(), + footnote=_copyright(), + ) + if chosen is None: + return None + + name = remembered.node_name if ( + remembered and remembered.matches(chosen) and remembered.node_name + ) else _hardware_name(chosen) + # Smoke-test the choice in place: the splash keeps its wordmark and box, only the box + # contents swap for an animated spinner while we talk to the device. + info = await ui.busy_startup( + f"Talking to {name} on {chosen.port}…", + verify(chosen), + title="Checking companion", + banner=load_logo(), + footnote=_copyright(), + ) + + if info is None: + await ui.notify_startup( + Text.from_markup( + f"[warn]{name} on {chosen.port} didn't answer as a MeshCore device.[/warn]\n" + "It may be a different kind of serial device, powered off, or busy.\n" + "Choose another device." + ), + title="Not a MeshCore device", + banner=load_logo(), + footnote=_copyright(), + ) + continue + + # Confirmed: remember it forever, and reflect that on any re-entry of the loop. + store.remember(chosen, node_name=_node_name_from(info)) + known.add(chosen.stable_id) + return chosen + + +def _build_items( + devices: list[DiscoveredDevice], + remembered: Optional[RememberedDevice], + known: set[str], +) -> list: + """Build the aligned splash rows (a muted header + one :class:`Choice` per device). + + The row for each device leads with its display name (the remembered node's name when + known, else the hardware name); columns are padded to a shared width so they align. + Devices already confirmed as MeshCore companions carry a bright tag; the remembered + default is starred. + """ def _name(device: DiscoveredDevice) -> str: if remembered is not None and remembered.matches(device) and remembered.node_name: return remembered.node_name return _hardware_name(device) - name_w = max(cell_len(_name(d)) for d in devices) if devices else 0 + name_w = max(cell_len(_name(d)) for d in devices) name_w = max(name_w, len("DEVICE")) - port_w = max((cell_len(d.port) for d in devices), default=0) + port_w = max(cell_len(d.port) for d in devices) port_w = max(port_w, len("PORT")) - vendor_w = max((cell_len(d.vendor_label) for d in devices), default=0) + vendor_w = max(cell_len(d.vendor_label) for d in devices) vendor_w = max(vendor_w, len("VENDOR")) # A muted, aligned header. The leading spaces mirror the row pointer (2) and the star @@ -103,7 +174,6 @@ async def prompt_device( + " " + "VENDOR" ) - default: Optional[DiscoveredDevice] = None items: list = [header] for device in devices: is_remembered = remembered is not None and remembered.matches(device) @@ -115,15 +185,13 @@ async def prompt_device( row.append(_pad(device.port, port_w), style="muted") row.append(" ") row.append(_pad(device.vendor_label, vendor_w), style="muted") - tag = _TAG[device.confidence] - if tag: + # Only devices we've actually confirmed are billed as MeshCore companions; the USB + # vendor ID is a sort hint, not a claim. A bare serial bridge earns an honest label. + if device.stable_id in known: row.append(" ") - row.append(tag, style="muted") + row.append("· MeshCore device", style="ok") + elif device.confidence == "bridge": + row.append(" ") + row.append("· serial adapter", style="muted") items.append(Choice(title=row, value=device)) - if is_remembered: - default = device - - return await ui.select_startup( - "Select a companion device", items, default=default, banner=load_logo(), - footnote=_copyright(), - ) + return items diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index 45bf950..7f39a2c 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -153,11 +153,17 @@ async def _startup(ctx: AppContext) -> None: ctx: The shared application context to update with the selection. """ if not (ctx.mock or ctx.explicit_selection): - from ..core.discovery import discover_devices + from ..core.connection import smoke_test_meshcore + from ..core.discovery import DiscoveredDevice, discover_devices from .device_picker import prompt_device devices = discover_devices() - chosen = await prompt_device(ctx.ui, devices, ctx.device_store.load()) + baudrate = ctx.profile.baudrate if ctx.profile else 115200 + + async def verify(device: DiscoveredDevice): + return await smoke_test_meshcore(device.port, baudrate) + + chosen = await prompt_device(ctx.ui, devices, ctx.device_store, verify) if chosen is not None: ctx.selected_device = chosen ctx.port_override = chosen.port diff --git a/meshterm/ui/surface.py b/meshterm/ui/surface.py index ea2bbdd..1ba32f5 100644 --- a/meshterm/ui/surface.py +++ b/meshterm/ui/surface.py @@ -80,6 +80,18 @@ class Ui: """Show a message on a chromeless startup splash until the user dismisses it.""" raise NotImplementedError + async def busy_startup( + self, + message: str, + coro: Any, + *, + title: str = "", + banner: Any = None, + footnote: Optional[str] = None, + ) -> Any: + """Await ``coro`` while showing a spinner on the startup splash; return its result.""" + raise NotImplementedError + async def reorder(self, title: str, labels: list[str]) -> list[int]: """Let the user rearrange rows with the arrows; return the new order of row indices.""" raise NotImplementedError @@ -179,6 +191,18 @@ class PlainUi(Ui): """Unsupported in scripted CLI mode.""" raise self._no_prompt() + async def busy_startup( + self, + message: str, + coro: Any, + *, + title: str = "", + banner: Any = None, + footnote: Optional[str] = None, + ) -> Any: + """No splash in scripted CLI mode; just await the task and return its result.""" + return await coro + async def reorder(self, title: str, labels: list[str]) -> list[int]: """Unsupported in scripted CLI mode.""" raise self._no_prompt() @@ -320,6 +344,20 @@ class TuiUi(Ui): renderable, title=title, banner=banner, footnote=footnote ) + async def busy_startup( + self, + message: str, + coro: Any, + *, + title: str = "", + banner: Any = None, + footnote: Optional[str] = None, + ) -> Any: + """Delegate to the session's animated-spinner startup splash.""" + return await self.session.busy_startup( + message, coro, title=title, banner=banner, footnote=footnote + ) + async def reorder(self, title: str, labels: list[str]) -> list[int]: """Delegate to the session's reorder screen.""" return await self.session.reorder(title, labels) diff --git a/meshterm/ui/tui/screen.py b/meshterm/ui/tui/screen.py index c8e8734..266df4f 100644 --- a/meshterm/ui/tui/screen.py +++ b/meshterm/ui/tui/screen.py @@ -16,6 +16,7 @@ import asyncio from typing import Any, Optional, Sequence from rich.console import RenderableType +from rich.text import Text from .render import render_lines @@ -191,3 +192,43 @@ class ScrollScreen(Screen): self.scroll = max(0, self._total - self._viewport) elif action in ("escape", "enter"): self.resolve(None) + + +class BusyScreen(Screen): + """A non-interactive splash body: an animated Braille spinner beside a message. + + Used to keep the startup splash on screen — its wordmark and box unchanged — while a + short async task runs (e.g. smoke-testing a chosen companion device). It set as a + chromeless base so it redraws the same centered-under-the-wordmark splash, only swapping + the box's contents. It resolves nothing and ignores every key; the session pops it when + the awaited task completes. + """ + + #: Braille-dot spinner frames, cycled on each :meth:`tick`. + _FRAMES = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏" + + footer_hint = "working…" + floating = False + + def __init__(self, message: str, *, title: str = "") -> None: + """Start a busy splash showing ``message`` under an optional ``title``.""" + super().__init__() + self.title = title + self._message = message + self._frame = 0 + + def tick(self) -> None: + """Advance the spinner to its next frame (driven by the session's animation timer).""" + self._frame = (self._frame + 1) % len(self._FRAMES) + + def render_body(self, width: int) -> list[str]: + """Render the current spinner frame followed by the message.""" + line = Text() + line.append(self._FRAMES[self._frame], style="accent") + line.append(" ") + line.append(self._message) + return render_lines(line, width) + + def handle(self, action: str, data: str = "") -> None: + """Swallow all keys: the splash dismisses itself when the task finishes.""" + return diff --git a/meshterm/ui/tui/session.py b/meshterm/ui/tui/session.py index d976812..1ce8e53 100644 --- a/meshterm/ui/tui/session.py +++ b/meshterm/ui/tui/session.py @@ -27,7 +27,7 @@ from rich.text import Text from . import frame from .progress import TuiProgress from .prompt import AutocompleteScreen, ConfirmScreen, TextScreen, Validator -from .screen import CANCEL, Screen, ScrollScreen +from .screen import CANCEL, BusyScreen, Screen, ScrollScreen from .select import Choice, ReorderScreen, SelectScreen, Separator #: Maps prompt_toolkit keys to the normalized action names screens understand. @@ -172,6 +172,53 @@ class TuiSession: screen.footnote = footnote await self.run_screen(screen) + async def busy_startup( + self, + message: str, + coro: Any, + *, + title: str = "", + banner: Optional[Any] = None, + footnote: Optional[str] = None, + interval: float = 0.12, + ) -> Any: + """Await ``coro`` while showing an animated spinner on the chromeless splash. + + Keeps the startup splash on screen (same wordmark and box) and swaps its contents for + an ASCII spinner beside ``message`` while the awaited task runs, then returns the + task's result. A background timer advances the spinner and repaints every + ``interval`` seconds; it is always cancelled and the splash popped before returning. + + Args: + message: The line shown beside the spinner (e.g. "Talking to Wio on COM5…"). + coro: The awaitable to run (e.g. a device smoke test). + title: Optional panel title for the splash box. + banner: Wordmark rows drawn above the box (as on the other startup splashes). + footnote: Muted line drawn below the box. + interval: Seconds between spinner frames. + + Returns: + Whatever ``coro`` resolves to. + """ + screen = BusyScreen(message, title=title) + screen.chrome = False + screen.banner = banner + screen.footnote = footnote + self.push(screen) + + async def animate() -> None: + while True: + await asyncio.sleep(interval) + screen.tick() + self.invalidate() + + ticker = asyncio.ensure_future(animate()) + try: + return await coro + finally: + ticker.cancel() + self.pop(screen) + async def reorder(self, title: str, labels: list[str]) -> list[int]: """Show a drag-with-arrows reorder screen; return the final order of row indices.""" result = await self.run_screen(ReorderScreen(title, labels)) diff --git a/tests/test_discovery.py b/tests/test_discovery.py index e1f4bc2..82e07ac 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -156,6 +156,44 @@ def test_device_store_tolerates_corrupt_file(tmp_path: Path) -> None: assert DeviceStore(path).load() is None +def test_device_store_remembers_every_confirmed_device(tmp_path: Path) -> None: + """Confirmed devices are all kept; ``load`` returns the most recently connected one.""" + store = DeviceStore(tmp_path / "devices.json") + first = DiscoveredDevice("COM5", serial_number="SN1", product="Wio") + second = DiscoveredDevice("COM6", serial_number="SN2", product="Heltec") + + store.remember(first, node_name="Base") + store.remember(second, node_name="Roamer") + + registry = store.load_all() + assert set(registry) == {"sn:SN1", "sn:SN2"} # both remembered forever + assert store.is_known(first) and store.is_known(second) + assert not store.is_known(DiscoveredDevice("COM7", serial_number="SN3")) + + last = store.load() + assert last is not None and last.stable_id == "sn:SN2" # most recent is the default + + # Re-confirming the first makes it the default again without dropping the second. + store.remember(first) + assert store.load().stable_id == "sn:SN1" + assert set(store.load_all()) == {"sn:SN1", "sn:SN2"} + + +def test_device_store_migrates_old_flat_format(tmp_path: Path) -> None: + """A pre-registry flat record still reads back as a one-entry registry.""" + path = tmp_path / "devices.json" + path.write_text( + '{"stable_id": "sn:SN1", "port": "COM5", "label": "Wio (COM5)", ' + '"last_connected": "2025-01-01T00:00:00", "node_name": "Base"}', + encoding="utf-8", + ) + store = DeviceStore(path) + loaded = store.load() + assert loaded is not None and loaded.stable_id == "sn:SN1" + assert loaded.node_name == "Base" + assert set(store.load_all()) == {"sn:SN1"} + + # -- selection ----------------------------------------------------------------- diff --git a/tests/test_tui.py b/tests/test_tui.py index 3bd5a0c..c755d3c 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -284,8 +284,9 @@ def test_autocomplete_accepts_free_text() -> None: # --- device picker ----------------------------------------------------------- -def test_device_picker_builds_aligned_columns() -> None: +def test_device_picker_builds_aligned_columns(tmp_path) -> None: """The picker lays devices out in columns that line up across rows of differing widths.""" + from meshterm.core.device_store import DeviceStore from meshterm.core.discovery import DiscoveredDevice from meshterm.ui.device_picker import prompt_device @@ -302,9 +303,13 @@ def test_device_picker_builds_aligned_columns() -> None: captured["items"] = items captured["banner"] = banner captured["footnote"] = footnote - return None + return None # skip: never reaches the smoke test - asyncio.run(prompt_device(_Ui(), devices, None)) + async def _never(_device): # verify is unused when the user skips + raise AssertionError("verify should not run when selection is skipped") + + store = DeviceStore(tmp_path / "devices.json") + asyncio.run(prompt_device(_Ui(), devices, store, _never)) # The banner (wordmark) is passed through so the splash can draw it. assert captured["banner"] and any("█" in row for row in captured["banner"]) # A copyright footnote rides along for the splash to render below the box. @@ -316,6 +321,73 @@ def test_device_picker_builds_aligned_columns() -> None: assert rows[0].index("COM5") == rows[1].index("/dev/ttyUSB0") +class _PickerUi: + """A fake splash UI that always selects the first device, then dismisses messages.""" + + def __init__(self) -> None: + self.notes: list = [] + + async def select_startup(self, title, items, *, default=None, banner=None, footnote=None): + return next(it.value for it in items if isinstance(it, Choice)) + + async def notify_startup(self, renderable, *, title="", banner=None, footnote=None): + self.notes.append(renderable) + + async def busy_startup(self, message, coro, *, title="", banner=None, footnote=None): + return await coro + + +def test_device_picker_smoke_tests_and_reprompts(tmp_path) -> None: + """A failed smoke test re-prompts; a passing one is remembered as confirmed.""" + from meshterm.core.device_store import DeviceStore + from meshterm.core.discovery import DiscoveredDevice + from meshterm.ui.device_picker import prompt_device + + devices = [DiscoveredDevice(port="COM5", serial_number="SN1", product="Wio SX1262")] + store = DeviceStore(tmp_path / "devices.json") + ui = _PickerUi() + + # First probe fails (not MeshCore), second answers with self-info. + results = [None, {"adv_name": "BaseStation"}] + + async def verify(_device): + return results.pop(0) + + chosen = asyncio.run(prompt_device(ui, devices, store, verify)) + assert chosen is devices[0] + assert len(ui.notes) == 1 # the "not a MeshCore device" message was shown once + remembered = store.load() + assert remembered is not None and remembered.node_name == "BaseStation" + assert store.is_known(devices[0]) + + +# --- busy splash -------------------------------------------------------------- + + +def test_busy_screen_spins_over_its_message() -> None: + """The busy splash shows an ASCII spinner beside its message and advances on tick.""" + from rich.text import Text as RichText + + from meshterm.ui.tui.screen import BusyScreen + + screen = BusyScreen("Talking to Wio on COM5…") + + def glyph() -> str: + """The first (spinner) character of the rendered body, ANSI codes stripped.""" + return RichText.from_ansi("\n".join(screen.render_body(60))).plain.lstrip()[0] + + plain = RichText.from_ansi("\n".join(screen.render_body(60))).plain + assert "Talking to Wio on COM5" in plain + assert glyph() in BusyScreen._FRAMES # a spinner glyph leads the line + + # Ticking cycles through every frame and returns to the first. + seen = {glyph()} + for _ in range(len(BusyScreen._FRAMES) - 1): + screen.tick() + seen.add(glyph()) + assert seen == set(BusyScreen._FRAMES) + + # --- progress ---------------------------------------------------------------- @@ -365,6 +437,29 @@ async def test_session_select_dispatches_piped_keys() -> None: assert captured["value"] == 3 +async def test_session_busy_startup_animates_and_returns() -> None: + """busy_startup awaits the task behind a chromeless spinner splash, then pops it.""" + from meshterm.ui.tui.screen import BusyScreen + + with create_pipe_input() as inp: + session = TuiSession(input=inp, output=DummyOutput()) + captured = {} + + async def work() -> str: + # While the task runs the busy splash is the chromeless base screen. + assert isinstance(session.top, BusyScreen) + assert session.top.chrome is False + await asyncio.sleep(0.3) # long enough for the spinner to tick at least once + return "ok" + + async def main() -> None: + captured["value"] = await session.busy_startup("checking…", work()) + + await asyncio.wait_for(session.run(main()), timeout=5) + assert captured["value"] == "ok" + assert session.top is None # the splash was popped when the task finished + + async def test_session_text_dispatches_typed_keys() -> None: """A text prompt captures typed characters and commits on Enter, end to end.""" with create_pipe_input() as inp: