From 421781f13d9f05ab8ca6ec277be479434564cf2d Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Sun, 21 Jun 2026 11:19:21 -0400 Subject: [PATCH] Fix auto-path traces, harden contacts retrieval, add device config tooling Trace: - Auto-path now ends every route at the target's own hash (verified on hardware: a trace only replies when its destination is the final hop), so direct neighbors and learned multi-hop routes both resolve correctly. - Retry the transient "no event received during contacts retrieval" blip and surface a clean DeviceCommandError instead of a raw traceback. - Route summary annotates each node with its hash at the command's path-hash width, wrapping after an arrow when the line is too long. - Default trace count changed from 5 to 3. Also includes the in-progress device-config feature (config_io, device_config, config tool, config editor UI, tests) and related supporting edits. Co-Authored-By: Claude Opus 4.8 --- .claude/settings.json | 16 +- .gitignore | 20 +- README.md | 13 + meshtools/cli.py | 7 +- meshtools/core/config_io.py | 152 ++++++++ meshtools/core/connection.py | 537 ++++++++++++++++++++++++++-- meshtools/core/device_config.py | 419 ++++++++++++++++++++++ meshtools/core/models.py | 113 ++++++ meshtools/persistence/repository.py | 51 ++- meshtools/services/trace_runner.py | 134 ++++++- meshtools/tools/base.py | 9 +- meshtools/tools/config.py | 358 +++++++++++++++++++ meshtools/tools/info.py | 19 +- meshtools/tools/trace.py | 144 ++++++-- meshtools/ui/config_editor.py | 394 ++++++++++++++++++++ meshtools/ui/widgets.py | 265 ++++++++++++-- pyproject.toml | 1 + tests/test_device_config.py | 193 ++++++++++ tests/test_skeleton.py | 344 +++++++++++++++++- 19 files changed, 3081 insertions(+), 108 deletions(-) create mode 100644 meshtools/core/config_io.py create mode 100644 meshtools/core/device_config.py create mode 100644 meshtools/tools/config.py create mode 100644 meshtools/ui/config_editor.py create mode 100644 tests/test_device_config.py diff --git a/.claude/settings.json b/.claude/settings.json index a566692..98b1b73 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -6,7 +6,21 @@ "Bash(.venv/Scripts/python.exe -c \"import meshcore; print\\('meshcore', getattr\\(meshcore,'__version__','?'\\)\\)\")", "Bash(.venv/Scripts/python.exe -m pytest -q)", "Bash(git -C d:/vibe/MeshTools status)", - "Bash(git -C d:/vibe/MeshTools log --oneline -5)" + "Bash(git -C d:/vibe/MeshTools log --oneline -5)", + "Bash(.venv/Scripts/python.exe -m meshtools --mock config set radio_sf 99)", + "Bash(echo \"exit=$?\")", + "Bash(python -c \"import meshcore, os; print\\(os.path.dirname\\(meshcore.__file__\\)\\)\")", + "Bash(python -m meshtools trace --mock --target Alice --samples 4)", + "Bash(pip show *)", + "Bash(.venv/Scripts/python.exe -c \"import meshtools.ui.widgets, meshtools.tools.trace; print\\('ok'\\)\")", + "PowerShell(python -m pytest tests/test_skeleton.py -k \"trace or neighbor or flood\" -q)", + "PowerShell(.venv\\\\Scripts\\\\python.exe -m pytest -q)", + "PowerShell(.venv\\\\Scripts\\\\python.exe -m meshtools devices 2>&1)", + "PowerShell(.venv\\\\Scripts\\\\python.exe -m meshtools info 2>&1)", + "PowerShell(foreach \\($p in 'COM26','COM16','COM17'\\) { Write-Output \"=== $p ===\"; .venv\\\\Scripts\\\\python.exe -m meshtools --port $p --quiet info 2>&1 | Select-Object -First 4 })", + "PowerShell(Get-CimInstance Win32_SerialPort | Where-Object { $_.DeviceID -eq 'COM11' } | Select-Object DeviceID, Description, Name | Format-List; Write-Output \"---\"; Get-Process | Where-Object { $_.Name -match 'meshcore|arduino|putty|python|node|serial|cu' } | Select-Object Id, ProcessName, MainWindowTitle | Format-Table -AutoSize)", + "PowerShell(foreach \\($id in 25324,30408\\) { try { Stop-Process -Id $id -Force -ErrorAction Stop; Write-Output \"killed $id\" } catch { Write-Output \"could not kill ${id}: $\\($_.Exception.Message\\)\" } }; Start-Sleep -Milliseconds 800; Get-CimInstance Win32_Process -Filter \"Name='python.exe'\" | Select-Object ProcessId | Format-Table -AutoSize)", + "PowerShell(.venv\\\\Scripts\\\\python.exe -m meshtools --quiet info 2>&1)" ] } } diff --git a/.gitignore b/.gitignore index db15f9c..3e1b140 100644 --- a/.gitignore +++ b/.gitignore @@ -5,21 +5,15 @@ __pycache__/ build/ dist/ -# Virtual environments -.venv/ -venv/ - -# Tooling caches -.pytest_cache/ -.ruff_cache/ -.mypy_cache/ - -# Editor -.vscode/ - -# Local Claude settings (project settings.json is tracked) +# Ignore all dot-directories (.venv, .pytest_cache, .ruff_cache, .vscode, ...) +.*/ +# ...except the project's tracked Claude config (settings.local.json stays ignored). +!.claude/ .claude/settings.local.json +# Other virtual environments +venv/ + # MeshTools runtime data *.db output/ diff --git a/README.md b/README.md index 78bcb0b..fbbe263 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,8 @@ turned into interactive HTML visualizations. | --- | --- | --- | | **Trace** | ✅ skeleton | Run repeated path traces to a target and aggregate per-hop SNR. | | **Device info** | ✅ skeleton | Show the connected companion device's identity and radio config. | +| **Device config** | ✅ working | View and change every setting (name, radio, behavior, experimental), with TOML backup/restore, channels, and gated destructive ops. | +| **Device discovery** | ✅ working | Enumerate serial devices, pick one interactively, and remember the last good default. | | **History / DB** | ✅ skeleton | Every tool execution and measurement is persisted and queryable. | | **TX-power optimization** | ✅ working | Sweep transmit power (coarse + refine), converge on the best signal, optionally apply, render an interactive chart. | | **Path optimization** | 🚧 planned | Discover and score mesh paths from neighbors outward. | @@ -34,10 +36,21 @@ meshtools # Scripted: every menu option is also a subcommand meshtools trace --target Alice --samples 10 --profile yagi + +# Force a route through specific repeaters (MeshCore-app style): comma-separated +# contact names and/or hex key prefixes, mixed freely. Blank lets the device route. +meshtools trace --target Alice --path "3d,f2,3d" +meshtools trace --target Alice --path "3d,Bravo-Repeater,f2" meshtools tx-optimize --target Alice --samples 6 --step 3 --apply meshtools info meshtools history +# Device configuration: view, set, back up, restore +meshtools config # show all current settings +meshtools config set radio_sf 9 # change one setting +meshtools config backup node.toml # archive every setting to TOML +meshtools config restore node.toml --dry-run + # No radio attached? Use the built-in simulator for development. meshtools --mock trace --target Alice ``` diff --git a/meshtools/cli.py b/meshtools/cli.py index b4bab27..7bfe3ee 100644 --- a/meshtools/cli.py +++ b/meshtools/cli.py @@ -17,6 +17,8 @@ import typer from .context import AppContext from .core.config import Settings +from .core.connection import DeviceCommandError +from .core.device_config import DeviceConfigError from .core.device_store import DeviceStore from .core.selection import DeviceSelectionError from .persistence.logging import configure_logging @@ -101,8 +103,9 @@ def run_tool_command(tool: Tool, params: dict) -> None: assert _state is not None # set by the callback before any subcommand runs try: asyncio.run(_drive(_execute_and_render(tool, params, _state), _state)) - except DeviceSelectionError as exc: - # Ambiguous/absent device: show the guidance message, not a traceback. + except (DeviceSelectionError, DeviceConfigError, DeviceCommandError) as exc: + # Expected user-facing error (ambiguous/absent device, bad config value, or a + # transient command failure): show the message, not a traceback. _state.console.print(f"[err]✗[/err] {exc}") raise typer.Exit(1) from exc diff --git a/meshtools/core/config_io.py b/meshtools/core/config_io.py new file mode 100644 index 0000000..2ee5dda --- /dev/null +++ b/meshtools/core/config_io.py @@ -0,0 +1,152 @@ +"""TOML backup and restore for device configuration. + +A backup captures every registry setting's current value plus experimental custom +variables and configured channels, so a device's configuration can be archived, diffed, +or cloned onto another node. Restore turns a backup into the same operation list the +``config`` tool executes for live edits, and supports a dry-run diff against the device's +current state. +""" + +from __future__ import annotations + +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import tomli_w + +from .device_config import DEVICE_SETTINGS, get_spec, parse_value + +if sys.version_info >= (3, 11): + import tomllib +else: # pragma: no cover - exercised only on 3.10 + import tomli as tomllib + + +@dataclass(slots=True) +class Backup: + """A parsed configuration backup. + + Attributes: + settings: Registry setting key -> raw value. + custom: Experimental custom variable name -> value. + channels: One dict per channel with ``index``, ``name`` and hex ``secret``. + """ + + settings: dict[str, Any] = field(default_factory=dict) + custom: dict[str, str] = field(default_factory=dict) + channels: list[dict[str, Any]] = field(default_factory=list) + + +def backup_config( + path: Path, + snapshot: dict, + custom_vars: dict[str, str], + channels: list[dict], +) -> Path: + """Write the current configuration to a TOML file. + + Args: + path: Destination file path (parent directories are created). + snapshot: A device snapshot (see ``device_config.build_snapshot``). + custom_vars: Experimental custom variables. + channels: Channel dicts (``channel_idx``, ``channel_name``, ``channel_secret`` + as raw bytes). + + Returns: + The path written. + """ + settings: dict[str, Any] = {} + for spec in DEVICE_SETTINGS: + value = spec.getter(snapshot) + if value is not None: + settings[spec.key] = value + + doc: dict[str, Any] = {"settings": settings} + if custom_vars: + doc["custom"] = dict(custom_vars) + if channels: + doc["channels"] = [ + { + "index": ch.get("channel_idx", idx), + "name": ch.get("channel_name", ""), + "secret": _to_hex(ch.get("channel_secret")), + } + for idx, ch in enumerate(channels) + ] + + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("wb") as fh: + tomli_w.dump(doc, fh) + return path + + +def read_backup(path: Path) -> Backup: + """Read and parse a TOML configuration backup. + + Args: + path: The backup file to read. + + Returns: + A :class:`Backup`. + + Raises: + FileNotFoundError: If ``path`` does not exist. + """ + with path.open("rb") as fh: + doc = tomllib.load(fh) + return Backup( + settings=dict(doc.get("settings") or {}), + custom=dict(doc.get("custom") or {}), + channels=list(doc.get("channels") or []), + ) + + +def plan_restore( + backup: Backup, + snapshot: dict, + current_custom: dict[str, str], +) -> list[tuple]: + """Diff a backup against the current state and return the operations to apply. + + Only values that differ from the device's current configuration are emitted, so a + restore (or its dry-run preview) shows exactly what would change. + + Args: + backup: The parsed backup. + snapshot: The device's current snapshot. + current_custom: The device's current custom variables. + + Returns: + A list of operation tuples compatible with the ``config`` tool: + ``("set", key, value)``, ``("set_custom", key, value)``, and + ``("set_channel", index, name, secret_bytes)``. + + Raises: + DeviceConfigError: If a backed-up setting key or value is invalid. + """ + ops: list[tuple] = [] + for key, raw in backup.settings.items(): + spec = get_spec(key) # raises on unknown key + value = parse_value(spec, raw) + if value != spec.getter(snapshot): + ops.append(("set", key, value)) + + for key, value in backup.custom.items(): + if current_custom.get(key) != value: + ops.append(("set_custom", key, str(value))) + + for ch in backup.channels: + secret_hex = ch.get("secret") or "" + secret = bytes.fromhex(secret_hex) if secret_hex else None + ops.append(("set_channel", int(ch.get("index", 0)), str(ch.get("name", "")), secret)) + + return ops + + +def _to_hex(secret: Any) -> str: + """Render a channel secret as a hex string, accepting bytes or str.""" + if isinstance(secret, (bytes, bytearray)): + return secret.hex() + return str(secret or "") diff --git a/meshtools/core/connection.py b/meshtools/core/connection.py index f864b9f..ae00993 100644 --- a/meshtools/core/connection.py +++ b/meshtools/core/connection.py @@ -24,6 +24,14 @@ TX_POWER_MIN = 1 TX_POWER_MAX = 22 +class DeviceCommandError(RuntimeError): + """A device command failed in a recoverable, user-facing way. + + Raised for conditions worth reporting cleanly (no traceback) — chiefly the + companion's intermittent failure to answer a query in time. Callers may retry. + """ + + class Device(ABC): """Abstract companion device exposing the operations MeshTools needs. @@ -73,21 +81,141 @@ class Device(ABC): self, target: str, *, - path: Optional[list[str]] = None, + path: Optional[str] = None, timeout: float = 10.0, ) -> TraceResult: """Run a single path trace to ``target`` and return per-hop SNR. Args: target: Name or key prefix of the destination node. - path: Optional explicit path to force (list of node identifiers). When - ``None`` the device chooses the path. + path: Optional explicit path to force, as a comma-separated string of + single-byte hex key prefixes (e.g. ``"3d,f2,3d"``). When ``None`` + the device chooses the path. timeout: Seconds to wait for the trace reply. Returns: A :class:`TraceResult`; ``success`` is ``False`` on timeout. """ + # -- configuration: extra reads --------------------------------------------- + + @abstractmethod + async def get_tuning(self) -> dict: + """Return radio tuning parameters. + + Returns: + A dict with ``rx_delay`` and ``airtime_factor`` (both ints). + """ + + @abstractmethod + async def get_path_hash_mode(self) -> int: + """Return the current path-hash mode (experimental routing flag).""" + + @abstractmethod + async def get_custom_vars(self) -> dict[str, str]: + """Return the device's experimental custom key/value variables.""" + + @abstractmethod + async def get_channel(self, index: int) -> Optional[dict]: + """Return one channel's configuration, or ``None`` if unset. + + Args: + index: Zero-based channel slot. + + Returns: + A dict with ``channel_idx``, ``channel_name`` and ``channel_secret`` + (16 raw bytes), or ``None`` when the slot is empty. + """ + + # -- configuration: settable values ----------------------------------------- + + @abstractmethod + async def set_name(self, name: str) -> None: + """Set the node's advertised name.""" + + @abstractmethod + async def set_coords(self, lat: float, lon: float) -> None: + """Set the node's advertised latitude/longitude (decimal degrees).""" + + @abstractmethod + async def set_device_pin(self, pin: int) -> None: + """Set the device's BLE pairing PIN.""" + + @abstractmethod + async def set_radio(self, freq: float, bw: float, sf: int, cr: int) -> None: + """Set the core radio parameters. + + Args: + freq: Frequency in MHz. + bw: Bandwidth in kHz. + sf: Spreading factor. + cr: Coding rate denominator (``5``-``8`` for 4/5-4/8). + """ + + @abstractmethod + async def set_tuning(self, rx_delay: int, airtime_factor: int) -> None: + """Set radio tuning parameters (RX delay and airtime budgeting factor).""" + + @abstractmethod + async def set_manual_add_contacts(self, enabled: bool) -> None: + """Set whether contacts must be added manually rather than automatically.""" + + @abstractmethod + async def set_adv_loc_policy(self, policy: int) -> None: + """Set the advert location-sharing policy.""" + + @abstractmethod + async def set_multi_acks(self, value: int) -> None: + """Set the multi-ack behavior flag.""" + + @abstractmethod + async def set_telemetry_modes(self, base: int, loc: int, env: int) -> None: + """Set the three telemetry mode fields together (each ``0``-``3``).""" + + @abstractmethod + async def set_path_hash_mode(self, mode: int) -> None: + """Set the experimental path-hash routing mode.""" + + @abstractmethod + async def set_custom_var(self, key: str, value: str) -> None: + """Set an experimental custom variable.""" + + @abstractmethod + async def set_channel(self, index: int, name: str, secret: Optional[bytes]) -> None: + """Configure a channel slot. + + Args: + index: Zero-based channel slot. + name: Channel name (a leading ``#`` derives the secret from the name). + secret: 16-byte shared secret, or ``None`` to derive it from ``name``. + """ + + # -- configuration: actions ------------------------------------------------- + + @abstractmethod + async def set_time(self, epoch: int) -> None: + """Set the device clock to a UNIX epoch timestamp.""" + + @abstractmethod + async def send_advert(self, flood: bool = False) -> None: + """Broadcast an advertisement (``flood`` propagates it across the mesh).""" + + @abstractmethod + async def reboot(self) -> None: + """Reboot the device.""" + + @abstractmethod + async def export_private_key(self) -> str: + """Export the device's private key as a hex string (sensitive).""" + + @abstractmethod + async def import_private_key(self, key_hex: str) -> None: + """Import a private key from a hex string (overwrites the device identity).""" + + @abstractmethod + async def factory_reset(self) -> None: + """Erase all device data and restore factory defaults (destructive).""" + async def __aenter__(self) -> "Device": """Enter the async context manager, connecting the device.""" await self.connect() @@ -151,10 +279,45 @@ class MeshCoreDevice(Device): result = await mc.commands.send_appstart() return dict(getattr(result, "payload", {}) or {}) + async def _contacts_payload( + self, mc, *, retries: int = 3, delay: float = 0.5 + ) -> dict: # noqa: ANN001 + """Fetch the raw contacts map, retrying the transient "no event" blip. + + The companion intermittently fails to emit the contacts event in time and + returns an error event instead of data (``no event received during contacts + retrieval``). That's a recoverable timing hiccup, so retry a few times with a + short backoff before surfacing a clean, actionable error. + + Args: + mc: The connected ``MeshCore`` client. + retries: Number of extra attempts after the first. + delay: Seconds to wait between attempts. + + Returns: + The contacts payload mapping (possibly empty). + + Raises: + DeviceCommandError: If every attempt fails to retrieve contacts. + """ + reason = "" + for attempt in range(retries + 1): + event = await mc.commands.get_contacts() + if event is None or not getattr(event, "is_error", lambda: False)(): + return dict(getattr(event, "payload", {}) or {}) + payload = getattr(event, "payload", {}) or {} + reason = str(payload.get("reason", payload)) + if attempt < retries: + await asyncio.sleep(delay) + raise DeviceCommandError( + f"could not read contacts from the radio ({reason}). " + "The companion didn't respond in time — this is usually transient; " + "retry, or power-cycle/reconnect the radio if it persists." + ) + async def get_contacts(self) -> list[Contact]: # noqa: D102 - inherited docstring mc = self._require() - result = await mc.commands.get_contacts() - payload = getattr(result, "payload", {}) or {} + payload = await self._contacts_payload(mc) contacts: list[Contact] = [] for name, info in payload.items(): info = info or {} @@ -180,44 +343,231 @@ class MeshCoreDevice(Device): self, target: str, *, - path: Optional[list[str]] = None, + path: Optional[str] = None, timeout: float = 10.0, ) -> TraceResult: from meshcore import EventType # local import keeps mock path dependency-free + from ..services.trace_runner import path_hash_flags + mc = self._require() tag = random.randint(0, 0xFFFFFFFF) started = asyncio.get_event_loop().time() - await mc.commands.send_trace(auth_code=0, tag=tag, flags=0, path=path) + + # A trace packet has no destination field — it walks an explicit path of + # repeater hops. Send the path as raw bytes so any uniform hash width + # transmits; ``flags`` carries the path-hash mode (size - 1). + path_bytes: Optional[bytes] = None + flags = 0 + if path: + hops = [h.strip() for h in path.split(",") if h.strip()] + path_bytes = bytes.fromhex("".join(hops)) + flags = path_hash_flags(len(bytes.fromhex(hops[0]))) or 0 + else: + # No forced path: build a path that ends at the target's own hash + # (prepending any learned ``out_path`` repeaters), since a trace only + # replies when its destination is the final hop. ``None`` only when the + # contact is unknown, leaving the trace to run path-less. + resolved = await self._trace_path_to_contact(mc, target) + if resolved is not None: + path_bytes, flags = resolved + await mc.commands.send_trace(auth_code=0, tag=tag, flags=flags, path=path_bytes) event = await mc.wait_for_event( EventType.TRACE_DATA, attribute_filters={"tag": tag}, timeout=timeout, ) elapsed_ms = (asyncio.get_event_loop().time() - started) * 1000.0 + # The firmware addresses each hop by a hash of ``1 << flags`` bytes; record it + # so the summary can show node hashes at the width the command actually used. + hash_bytes = 1 << flags if event is None: - return TraceResult(target=target, success=False, round_trip_ms=None) + return TraceResult( + target=target, success=False, round_trip_ms=None, path_hash_bytes=hash_bytes + ) payload = getattr(event, "payload", {}) or {} - raw_snrs = payload.get("path_snrs") or payload.get("snrs") or [] - nodes = payload.get("path") or [] - hops = [ - Hop( - index=i, - node=str(nodes[i]) if i < len(nodes) else None, - snr=raw / 4.0, # protocol encodes SNR as a signed byte of SNR * 4 - ) - for i, raw in enumerate(raw_snrs) - ] + hops = parse_trace_hops(payload) return TraceResult( target=target, success=True, hops=hops, round_trip_ms=elapsed_ms, tx_power=await self.get_tx_power(), + path_hash_bytes=hash_bytes, raw=payload, ) + async def _trace_path_to_contact( + self, mc, target: str + ) -> Optional[tuple[bytes, int]]: # noqa: ANN001 + """Resolve a target contact into a trace ``(path_bytes, flags)``. + + A trace reply only comes back when the *destination's own hash* is the final + hop in the path — an empty/destination-less path is silently dropped (verified + on hardware: a direct neighbor answers a single-hop trace to its own hash but + not a path-less one). So we always end the path at the contact's key prefix, + prepending any learned repeater hops (``out_path``) ahead of it: + + * direct neighbor / no learned route → just ``[destination]``; + * learned multi-hop route → ``[repeater…, destination]``. + + Each hash is re-encoded at the trace's own width (``1 << flags``, only 1/2/4/8 + bytes), collapsing a region's routing width (e.g. 3) to the widest representable + value (2). The firmware matches by hash prefix, so a narrower prefix still + addresses the same node. + + Args: + mc: The connected ``MeshCore`` client. + target: Contact name (case-insensitive) or public-key prefix. + + Returns: + ``(path_bytes, flags)`` to walk, or ``None`` only when the contact is + unknown or carries no public key to address. + """ + from ..services.trace_runner import path_hash_flags + + payload = await self._contacts_payload(mc) + needle = target.casefold() + for name, info in payload.items(): + info = info or {} + adv = info.get("adv_name", name) + pub = (info.get("public_key", "") or "").lower().removeprefix("0x") + if adv.casefold() != needle and not pub.startswith(needle): + continue + if not pub: + return None # no key to address the trace's destination hop + + # Routing hash width (bytes): the contact's stored mode, or — when it has + # no learned route (mode == -1) — our region's mode. + mode = int(info.get("out_path_hash_mode", -1)) + if mode < 0: + try: + mode = int(await mc.commands.get_path_hash_mode()) + except Exception: + mode = 2 # region default: 3-byte hashes + size = max(mode + 1, 1) + trace_size = max(s for s in (1, 2, 4, 8) if s <= size) + + hops: list[bytes] = [] + out_path = (info.get("out_path") or "").strip().lower().removeprefix("0x") + out_path_len = int(info.get("out_path_len", -1)) + if 1 <= out_path_len <= 254 and out_path: + # Learned multi-hop route: walk each repeater, collapsed to trace width. + route = bytes.fromhex(out_path)[: out_path_len * size] + hops += [route[i * size : i * size + trace_size] for i in range(out_path_len)] + # Always finish at the destination's own hash so the target recognizes the + # trace and replies; for a direct neighbor this single hop is the whole path. + hops.append(bytes.fromhex(pub)[:trace_size]) + return b"".join(hops), path_hash_flags(trace_size) or 0 + return None + + @staticmethod + def _ok(event): # type: ignore[no-untyped-def] + """Return ``event`` if it succeeded, else raise its error payload. + + Args: + event: The :class:`meshcore.events.Event` returned by a command. + + Returns: + The same event, for convenient chaining. + + Raises: + RuntimeError: If the device reported an error. + """ + if event is not None and getattr(event, "is_error", lambda: False)(): + raise RuntimeError(f"device rejected command: {getattr(event, 'payload', {})}") + return event + + async def get_tuning(self) -> dict: # noqa: D102 - inherited docstring + event = self._ok(await self._require().commands.get_tuning()) + payload = getattr(event, "payload", {}) or {} + return { + "rx_delay": int(payload.get("rx_delay", 0)), + "airtime_factor": int(payload.get("airtime_factor", 0)), + } + + async def get_path_hash_mode(self) -> int: # noqa: D102 - inherited docstring + return int(await self._require().commands.get_path_hash_mode()) + + async def get_custom_vars(self) -> dict[str, str]: # noqa: D102 - inherited docstring + event = self._ok(await self._require().commands.get_custom_vars()) + return dict(getattr(event, "payload", {}) or {}) + + async def get_channel(self, index: int) -> Optional[dict]: # noqa: D102 + event = self._ok(await self._require().commands.get_channel(index)) + payload = getattr(event, "payload", {}) or {} + if not payload.get("channel_name"): + return None + return payload + + async def set_name(self, name: str) -> None: # noqa: D102 - inherited docstring + self._ok(await self._require().commands.set_name(name)) + + async def set_coords(self, lat: float, lon: float) -> None: # noqa: D102 + self._ok(await self._require().commands.set_coords(lat, lon)) + + async def set_device_pin(self, pin: int) -> None: # noqa: D102 - inherited docstring + self._ok(await self._require().commands.set_devicepin(pin)) + + async def set_radio(self, freq: float, bw: float, sf: int, cr: int) -> None: # noqa: D102 + self._ok(await self._require().commands.set_radio(freq, bw, sf, cr)) + + async def set_tuning(self, rx_delay: int, airtime_factor: int) -> None: # noqa: D102 + self._ok(await self._require().commands.set_tuning(rx_delay, airtime_factor)) + + async def set_manual_add_contacts(self, enabled: bool) -> None: # noqa: D102 + self._ok(await self._require().commands.set_manual_add_contacts(enabled)) + + async def set_adv_loc_policy(self, policy: int) -> None: # noqa: D102 + self._ok(await self._require().commands.set_advert_loc_policy(policy)) + + async def set_multi_acks(self, value: int) -> None: # noqa: D102 - inherited docstring + self._ok(await self._require().commands.set_multi_acks(value)) + + async def set_telemetry_modes(self, base: int, loc: int, env: int) -> None: # noqa: D102 + mc = self._require() + result = self._ok(await mc.commands.send_appstart()) + infos = dict(getattr(result, "payload", {}) or {}) + infos["telemetry_mode_base"] = base + infos["telemetry_mode_loc"] = loc + infos["telemetry_mode_env"] = env + self._ok(await mc.commands.set_other_params_from_infos(infos)) + + async def set_path_hash_mode(self, mode: int) -> None: # noqa: D102 + self._ok(await self._require().commands.set_path_hash_mode(mode)) + + async def set_custom_var(self, key: str, value: str) -> None: # noqa: D102 + self._ok(await self._require().commands.set_custom_var(key, value)) + + async def set_channel(self, index: int, name: str, secret: Optional[bytes]) -> None: # noqa: D102 + self._ok(await self._require().commands.set_channel(index, name, secret)) + + async def set_time(self, epoch: int) -> None: # noqa: D102 - inherited docstring + self._ok(await self._require().commands.set_time(epoch)) + + async def send_advert(self, flood: bool = False) -> None: # noqa: D102 + self._ok(await self._require().commands.send_advert(flood)) + + async def reboot(self) -> None: # noqa: D102 - inherited docstring + await self._require().commands.reboot() # device reboots; no OK reply expected + + async def export_private_key(self) -> str: # noqa: D102 - inherited docstring + event = self._ok(await self._require().commands.export_private_key()) + payload = getattr(event, "payload", {}) or {} + key = payload.get("private_key") + if key is None: + raise RuntimeError("device did not return a private key (export may be disabled)") + return key.hex() if isinstance(key, (bytes, bytearray)) else str(key) + + async def import_private_key(self, key_hex: str) -> None: # noqa: D102 + self._ok(await self._require().commands.import_private_key(bytes.fromhex(key_hex))) + + async def factory_reset(self) -> None: # noqa: D102 - inherited docstring + mc = self._require() + token = await mc.commands.request_factory_reset() + self._ok(await mc.commands.confirm_factory_reset(token)) + class MockDevice(Device): """A deterministic simulator implementing the full :class:`Device` interface. @@ -247,6 +597,33 @@ class MockDevice(Device): Contact(name="Observer-Bot", key_prefix="c3d4e5f6"), Contact(name="Alice", key_prefix="d4e5f6a7"), ] + # Mutable configuration state, keyed exactly like the real SELF_INFO payload so the + # settings registry behaves identically on the simulator and on hardware. + self._info: dict = { + "name": "MockCompanion", + "adv_type": 1, + "max_tx_power": 22, + "public_key": "00" * 32, + "adv_lat": 0.0, + "adv_lon": 0.0, + "multi_acks": 0, + "adv_loc_policy": 0, + "telemetry_mode_base": 0, + "telemetry_mode_loc": 0, + "telemetry_mode_env": 0, + "manual_add_contacts": False, + "radio_freq": 869.525, + "radio_bw": 250.0, + "radio_sf": 11, + "radio_cr": 5, + "simulated": True, + } + self._tuning: dict = {"rx_delay": 0, "airtime_factor": 0} + self._path_hash_mode = 0 + self._custom_vars: dict[str, str] = {} + self._channels: dict[int, dict] = {} + self._device_pin = 0 + self._private_key = "11" * 32 async def connect(self) -> None: # noqa: D102 - inherited docstring await asyncio.sleep(0) @@ -256,15 +633,7 @@ class MockDevice(Device): self._connected = False async def get_self_info(self) -> dict: # noqa: D102 - inherited docstring - return { - "name": "MockCompanion", - "tx_power": self._tx_power, - "freq": 869.525, - "bw": 250, - "sf": 11, - "cr": 5, - "simulated": True, - } + return {**self._info, "tx_power": self._tx_power} async def get_contacts(self) -> list[Contact]: # noqa: D102 - inherited docstring return list(self._contacts) @@ -275,6 +644,80 @@ class MockDevice(Device): async def set_tx_power(self, value: int) -> None: # noqa: D102 - inherited docstring self._tx_power = value + async def get_tuning(self) -> dict: # noqa: D102 - inherited docstring + return dict(self._tuning) + + async def get_path_hash_mode(self) -> int: # noqa: D102 - inherited docstring + return self._path_hash_mode + + async def get_custom_vars(self) -> dict[str, str]: # noqa: D102 - inherited docstring + return dict(self._custom_vars) + + async def get_channel(self, index: int) -> Optional[dict]: # noqa: D102 + return self._channels.get(index) + + async def set_name(self, name: str) -> None: # noqa: D102 - inherited docstring + self._info["name"] = name + + async def set_coords(self, lat: float, lon: float) -> None: # noqa: D102 + self._info["adv_lat"] = lat + self._info["adv_lon"] = lon + + async def set_device_pin(self, pin: int) -> None: # noqa: D102 - inherited docstring + self._device_pin = pin + + async def set_radio(self, freq: float, bw: float, sf: int, cr: int) -> None: # noqa: D102 + self._info.update(radio_freq=freq, radio_bw=bw, radio_sf=sf, radio_cr=cr) + + async def set_tuning(self, rx_delay: int, airtime_factor: int) -> None: # noqa: D102 + self._tuning = {"rx_delay": rx_delay, "airtime_factor": airtime_factor} + + async def set_manual_add_contacts(self, enabled: bool) -> None: # noqa: D102 + self._info["manual_add_contacts"] = enabled + + async def set_adv_loc_policy(self, policy: int) -> None: # noqa: D102 + self._info["adv_loc_policy"] = policy + + async def set_multi_acks(self, value: int) -> None: # noqa: D102 - inherited docstring + self._info["multi_acks"] = value + + async def set_telemetry_modes(self, base: int, loc: int, env: int) -> None: # noqa: D102 + self._info.update( + telemetry_mode_base=base, telemetry_mode_loc=loc, telemetry_mode_env=env + ) + + async def set_path_hash_mode(self, mode: int) -> None: # noqa: D102 + self._path_hash_mode = mode + + async def set_custom_var(self, key: str, value: str) -> None: # noqa: D102 + self._custom_vars[key] = value + + async def set_channel(self, index: int, name: str, secret: Optional[bytes]) -> None: # noqa: D102 + self._channels[index] = { + "channel_idx": index, + "channel_name": name, + "channel_secret": secret or (b"\x00" * 16), + } + + async def set_time(self, epoch: int) -> None: # noqa: D102 - inherited docstring + self._info["clock"] = epoch + + async def send_advert(self, flood: bool = False) -> None: # noqa: D102 + await asyncio.sleep(0) + + async def reboot(self) -> None: # noqa: D102 - inherited docstring + await asyncio.sleep(0) + + async def export_private_key(self) -> str: # noqa: D102 - inherited docstring + return self._private_key + + async def import_private_key(self, key_hex: str) -> None: # noqa: D102 + self._private_key = bytes.fromhex(key_hex).hex() # validates hex, normalizes + + async def factory_reset(self) -> None: # noqa: D102 - inherited docstring + self._custom_vars.clear() + self._channels.clear() + def _expected_snr(self, hop_index: int) -> float: """Model SNR for a hop as an inverted-U in TX power plus distance falloff. @@ -294,29 +737,61 @@ class MockDevice(Device): self, target: str, *, - path: Optional[list[str]] = None, + path: Optional[str] = None, timeout: float = 10.0, ) -> TraceResult: await asyncio.sleep(0.05) # mimic radio latency so progress bars are visible - depth = len(path) if path else self._rng.randint(1, 3) + forced = [h for h in path.split(",") if h.strip()] if path else None + # Mirror the real device: the path-hash width is the byte length of a forced hop. + hash_bytes = len(bytes.fromhex(forced[0])) if forced else None + depth = len(forced) if forced else self._rng.randint(1, 3) hops: list[Hop] = [] for i in range(depth): expected = self._expected_snr(i) snr = expected + self._rng.gauss(0, 1.2) # measurement noise - node = path[i] if path else f"hop{i}" + node = forced[i] if forced else f"hop{i}" hops.append(Hop(index=i, node=node, snr=round(snr, 1))) - # Very weak links occasionally drop entirely. + # Very weak links occasionally drop entirely (judged on the repeater hops). success = (hops[-1].snr if hops else -99) > -12 or self._rng.random() > 0.1 + + # The trace reply returns to us: firmware appends the local device as a final + # hash-less hop (``node=None``). Mirror that so the origin/destination framing + # naturally has our device at both ends of the path. The return link is modeled + # as symmetric to the first outbound hop, so it never alters the bottleneck SNR. + if hops: + hops.append(Hop(index=depth, node=None, snr=hops[0].snr)) return TraceResult( target=target, success=success, hops=hops if success else [], round_trip_ms=round(self._rng.uniform(120, 480), 1) if success else None, tx_power=self._tx_power, + path_hash_bytes=hash_bytes, ) +def parse_trace_hops(payload: dict) -> list[Hop]: + """Extract per-hop SNR from a ``TRACE_DATA`` event payload. + + meshcore parses a trace reply into ``payload["path"]`` — a list of nodes, each a + dict with a repeater ``"hash"`` and its ``"snr"`` (already in dB, signed-byte / 4). + The final node is the local device and carries an ``"snr"`` but no ``"hash"``. + + Args: + payload: The ``TRACE_DATA`` event payload. + + Returns: + The hops in path order; nodes without an SNR reading are skipped. + """ + hops: list[Hop] = [] + for node in payload.get("path") or []: + if not isinstance(node, dict) or node.get("snr") is None: + continue + hops.append(Hop(index=len(hops), node=node.get("hash"), snr=float(node["snr"]))) + return hops + + def clamp_tx_power(value: int) -> int: """Clamp a TX power value to the supported range. diff --git a/meshtools/core/device_config.py b/meshtools/core/device_config.py new file mode 100644 index 0000000..58b7ace --- /dev/null +++ b/meshtools/core/device_config.py @@ -0,0 +1,419 @@ +"""Declarative registry of every settable device configuration value. + +Each :class:`SettingSpec` describes one tunable setting: how to read its current value +from a device snapshot, how to parse/validate/format it, and how to apply it back to the +:class:`~meshtools.core.connection.Device`. The same registry drives the interactive +editor, the ``config`` CLI subcommands, and TOML backup/restore — add a setting once and +it appears everywhere, mirroring the tool registry pattern. + +Settings whose protocol command takes several fields at once (radio, coordinates, tuning, +telemetry modes) rebuild the full command from the current snapshot plus the one changed +field, so a single setting can be edited in isolation. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from typing import Any, Optional + +from .connection import Device + +# Display categories, in the order the editor and `config show` present them. +CATEGORIES = ("Identity", "Radio", "Tuning", "Behavior", "Experimental") + + +class DeviceConfigError(ValueError): + """Raised when a value cannot be parsed or fails validation. + + The message is user-facing (printed directly by the CLI and editor). + """ + + +@dataclass(slots=True) +class SettingSpec: + """Specification for one settable configuration value. + + Attributes: + key: Canonical key (matches the ``SELF_INFO`` field name where one exists). + label: Human-friendly name for display. + help: One-line description. + category: One of :data:`CATEGORIES`. + value_type: ``"str" | "int" | "float" | "bool" | "enum"``. + choices: For ``enum``, a mapping of allowed int value to label. + minimum: Inclusive lower bound for numeric types, if any. + maximum: Inclusive upper bound for numeric types, if any. + strict_choices: When ``True`` (the default) an ``enum`` value must be one of + :attr:`choices`. When ``False`` the choices are offered as a convenience menu + but any in-range integer is still accepted — used for firmware fields whose + full value domain we don't enumerate exhaustively. + getter: Extracts the current value from a snapshot dict. + apply: Coroutine applying a parsed value to a device, given the snapshot. + """ + + key: str + label: str + help: str + category: str + value_type: str + getter: Callable[[dict], Any] + apply: Callable[[Device, Any, dict], Awaitable[None]] + choices: Optional[dict[int, str]] = None + minimum: Optional[float] = None + maximum: Optional[float] = None + strict_choices: bool = True + + +# --- value parsing / formatting ---------------------------------------------- + +_TRUE = {"1", "true", "yes", "on", "y"} +_FALSE = {"0", "false", "no", "off", "n"} + + +def parse_value(spec: SettingSpec, raw: Any) -> Any: + """Parse and validate a raw value (typically a CLI/TOML string) for ``spec``. + + Args: + spec: The target setting. + raw: The raw value to coerce (string or already-typed scalar). + + Returns: + The typed, range-checked value ready for :meth:`SettingSpec.apply`. + + Raises: + DeviceConfigError: If the value is the wrong type or out of range. + """ + text = str(raw).strip() + if spec.value_type == "str": + return text + if spec.value_type == "bool": + low = text.lower() + if low in _TRUE: + return True + if low in _FALSE: + return False + raise DeviceConfigError(f"{spec.key}: expected a boolean, got {raw!r}") + + # Numeric (int / enum / float): wrap only the conversion, so a DeviceConfigError + # raised by the range/enum checks below is not caught and rewrapped here. + try: + if spec.value_type == "float": + value: Any = float(text) + else: # int or enum + value = int(text, 0) if isinstance(raw, str) else int(raw) + except (ValueError, TypeError) as exc: + raise DeviceConfigError(f"{spec.key}: invalid {spec.value_type} value {raw!r}") from exc + + if ( + spec.value_type == "enum" + and spec.strict_choices + and spec.choices is not None + and value not in spec.choices + ): + allowed = ", ".join(str(k) for k in spec.choices) + raise DeviceConfigError(f"{spec.key}: must be one of {allowed}, got {value}") + if spec.minimum is not None and value < spec.minimum: + raise DeviceConfigError(f"{spec.key}: must be >= {spec.minimum}, got {value}") + if spec.maximum is not None and value > spec.maximum: + raise DeviceConfigError(f"{spec.key}: must be <= {spec.maximum}, got {value}") + return value + + +def format_value(spec: SettingSpec, value: Any) -> str: + """Render a value for display. + + Args: + spec: The setting the value belongs to. + value: The current value (may be ``None`` if unknown). + + Returns: + A human-readable string. + """ + if value is None: + return "?" + if spec.value_type == "bool": + return "true" if value else "false" + if spec.value_type == "enum" and spec.choices is not None and value in spec.choices: + return f"{value} ({spec.choices[value]})" + return str(value) + + +# --- snapshot ---------------------------------------------------------------- + + +async def build_snapshot(device: Device) -> dict: + """Read a device's full current configuration into one dict. + + Merges ``SELF_INFO`` with the separately-read tuning parameters and path-hash mode. + Optional reads that a given firmware does not support are skipped silently so the rest + of the snapshot still renders. + + Args: + device: A connected device. + + Returns: + A dict keyed like ``SELF_INFO`` plus ``rx_delay``, ``airtime_factor`` and + ``path_hash_mode``. + """ + snapshot = dict(await device.get_self_info()) + try: + snapshot.update(await device.get_tuning()) + except Exception: # noqa: BLE001 - optional read; absence is acceptable + pass + try: + snapshot["path_hash_mode"] = await device.get_path_hash_mode() + except Exception: # noqa: BLE001 - optional read; absence is acceptable + pass + return snapshot + + +# --- coupled-command apply helpers ------------------------------------------- + + +def _radio_apply(field_name: str) -> Callable[[Device, Any, dict], Awaitable[None]]: + """Build an apply that updates one radio field, preserving the others.""" + + async def apply(device: Device, value: Any, snapshot: dict) -> None: + params = { + "freq": snapshot.get("radio_freq"), + "bw": snapshot.get("radio_bw"), + "sf": snapshot.get("radio_sf"), + "cr": snapshot.get("radio_cr"), + } + params[field_name] = value + await device.set_radio(params["freq"], params["bw"], params["sf"], params["cr"]) + + return apply + + +def _coords_apply(field_name: str) -> Callable[[Device, Any, dict], Awaitable[None]]: + """Build an apply that updates one coordinate, preserving the other.""" + + async def apply(device: Device, value: Any, snapshot: dict) -> None: + lat = value if field_name == "adv_lat" else snapshot.get("adv_lat", 0.0) + lon = value if field_name == "adv_lon" else snapshot.get("adv_lon", 0.0) + await device.set_coords(float(lat or 0.0), float(lon or 0.0)) + + return apply + + +def _tuning_apply(field_name: str) -> Callable[[Device, Any, dict], Awaitable[None]]: + """Build an apply that updates one tuning field, preserving the other.""" + + async def apply(device: Device, value: Any, snapshot: dict) -> None: + rx = value if field_name == "rx_delay" else snapshot.get("rx_delay", 0) + af = value if field_name == "airtime_factor" else snapshot.get("airtime_factor", 0) + await device.set_tuning(int(rx or 0), int(af or 0)) + + return apply + + +def _telemetry_apply(field_name: str) -> Callable[[Device, Any, dict], Awaitable[None]]: + """Build an apply that updates one telemetry mode, preserving the others.""" + + async def apply(device: Device, value: Any, snapshot: dict) -> None: + base = value if field_name == "telemetry_mode_base" else snapshot.get("telemetry_mode_base", 0) + loc = value if field_name == "telemetry_mode_loc" else snapshot.get("telemetry_mode_loc", 0) + env = value if field_name == "telemetry_mode_env" else snapshot.get("telemetry_mode_env", 0) + await device.set_telemetry_modes(int(base or 0), int(loc or 0), int(env or 0)) + + return apply + + +def _get(key: str) -> Callable[[dict], Any]: + """Build a getter that reads ``key`` from a snapshot.""" + return lambda snapshot: snapshot.get(key) + + +_TELEMETRY_CHOICES = {0: "off", 1: "always", 2: "device-only", 3: "all"} + +# adv_loc_policy / multi_acks are single firmware bytes; we list the values seen in the +# wild but keep them non-strict so an unfamiliar value is still accepted. +_ADV_LOC_CHOICES = {0: "don't share location", 1: "share location in adverts"} +_MULTI_ACKS_CHOICES = {0: "off", 1: "on"} +# path_hash_mode is a 2-bit field; the hash size carried per hop is mode + 1 bytes. +_PATH_HASH_CHOICES = { + 0: "1-byte hashes (default)", + 1: "2-byte hashes", + 2: "3-byte hashes", + 3: "4-byte hashes", +} + + +def _telemetry_spec(key: str, label: str) -> SettingSpec: + """Build a telemetry-mode setting spec (shared shape for base/loc/env).""" + return SettingSpec( + key=key, label=label, help="Telemetry reporting mode.", category="Behavior", + value_type="enum", choices=_TELEMETRY_CHOICES, minimum=0, maximum=3, + getter=_get(key), apply=_telemetry_apply(key), + ) + + +# --- radio presets ----------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class RadioPreset: + """A named, standard set of radio parameters applied as one unit. + + Attributes: + name: Human-friendly preset name (region or trade-off). + help: One-line description of when to use it. + freq: Carrier frequency in MHz. + bw: Channel bandwidth in kHz. + sf: LoRa spreading factor. + cr: LoRa coding-rate denominator (5-8 = 4/5-4/8). + """ + + name: str + help: str + freq: float + bw: float + sf: int + cr: int + + def as_settings(self) -> dict[str, Any]: + """Return this preset as a ``{setting_key: value}`` mapping for staging.""" + return { + "radio_freq": self.freq, + "radio_bw": self.bw, + "radio_sf": self.sf, + "radio_cr": self.cr, + } + + +# Standard MeshCore presets. The default modulation (250 kHz / SF11 / CR5) is shared +# across regions; the region presets differ only in the default channel frequency. The +# trade-off presets keep the EU frequency but trade airtime for range or speed. +RADIO_PRESETS: list[RadioPreset] = [ + RadioPreset("EU / UK 868", "MeshCore default for the 868 MHz band.", 869.525, 250.0, 11, 5), + RadioPreset("US 915", "MeshCore default for the US 915 MHz band.", 910.525, 250.0, 11, 5), + RadioPreset("AU / NZ 915", "MeshCore default for the AU/NZ 915 MHz band.", 915.525, 250.0, 11, 5), + RadioPreset("Long range (slow)", "Narrow bandwidth + high SF: max range, low data rate.", 869.525, 125.0, 12, 8), + RadioPreset("Fast (short range)", "Wide bandwidth + low SF: short range, low airtime.", 869.525, 500.0, 7, 5), +] + + +# --- the registry ------------------------------------------------------------ + +DEVICE_SETTINGS: list[SettingSpec] = [ + # Identity + SettingSpec( + "name", "Node name", "Advertised name of this node.", "Identity", "str", + getter=_get("name"), apply=lambda d, v, s: d.set_name(v), + ), + SettingSpec( + "adv_lat", "Latitude", "Advertised latitude (decimal degrees).", "Identity", + "float", minimum=-90.0, maximum=90.0, + getter=_get("adv_lat"), apply=_coords_apply("adv_lat"), + ), + SettingSpec( + "adv_lon", "Longitude", "Advertised longitude (decimal degrees).", "Identity", + "float", minimum=-180.0, maximum=180.0, + getter=_get("adv_lon"), apply=_coords_apply("adv_lon"), + ), + SettingSpec( + "device_pin", "Device PIN", "BLE pairing PIN.", "Identity", "int", + minimum=0, maximum=999999, + getter=_get("device_pin"), apply=lambda d, v, s: d.set_device_pin(v), + ), + # Radio + SettingSpec( + "radio_freq", "Frequency (MHz)", "Carrier frequency in MHz.", "Radio", "float", + minimum=100.0, maximum=1000.0, + getter=_get("radio_freq"), apply=_radio_apply("freq"), + ), + SettingSpec( + "radio_bw", "Bandwidth (kHz)", "Channel bandwidth in kHz.", "Radio", "float", + minimum=1.0, maximum=1000.0, + getter=_get("radio_bw"), apply=_radio_apply("bw"), + ), + SettingSpec( + "radio_sf", "Spreading factor", "LoRa spreading factor.", "Radio", "int", + minimum=5, maximum=12, + getter=_get("radio_sf"), apply=_radio_apply("sf"), + ), + SettingSpec( + "radio_cr", "Coding rate", "LoRa coding-rate denominator (5-8 = 4/5-4/8).", + "Radio", "int", minimum=5, maximum=8, + getter=_get("radio_cr"), apply=_radio_apply("cr"), + ), + SettingSpec( + "tx_power", "TX power (dBm)", "Transmit power.", "Radio", "int", + minimum=1, maximum=22, + getter=_get("tx_power"), apply=lambda d, v, s: d.set_tx_power(v), + ), + # Tuning + SettingSpec( + "rx_delay", "RX delay", "Receive delay tuning parameter.", "Tuning", "int", + minimum=0, + getter=_get("rx_delay"), apply=_tuning_apply("rx_delay"), + ), + SettingSpec( + "airtime_factor", "Airtime factor", "Airtime budgeting factor.", "Tuning", "int", + minimum=0, + getter=_get("airtime_factor"), apply=_tuning_apply("airtime_factor"), + ), + # Behavior + SettingSpec( + "manual_add_contacts", "Manual add contacts", + "Require contacts to be added manually.", "Behavior", "bool", + getter=_get("manual_add_contacts"), + apply=lambda d, v, s: d.set_manual_add_contacts(v), + ), + SettingSpec( + "adv_loc_policy", "Advert location policy", + "Whether this node shares its location in adverts.", "Behavior", "enum", + choices=_ADV_LOC_CHOICES, minimum=0, strict_choices=False, + getter=_get("adv_loc_policy"), apply=lambda d, v, s: d.set_adv_loc_policy(v), + ), + SettingSpec( + "multi_acks", "Multi-acks", + "Send multiple acknowledgements for added delivery reliability.", "Behavior", + "enum", choices=_MULTI_ACKS_CHOICES, minimum=0, strict_choices=False, + getter=_get("multi_acks"), apply=lambda d, v, s: d.set_multi_acks(v), + ), + _telemetry_spec("telemetry_mode_base", "Telemetry mode (base)"), + _telemetry_spec("telemetry_mode_loc", "Telemetry mode (location)"), + _telemetry_spec("telemetry_mode_env", "Telemetry mode (environment)"), + # Experimental + SettingSpec( + "path_hash_mode", "Path-hash mode", + "Per-hop path-hash size; larger resists hash collisions but adds packet overhead.", + "Experimental", "enum", choices=_PATH_HASH_CHOICES, minimum=0, maximum=3, + getter=_get("path_hash_mode"), apply=lambda d, v, s: d.set_path_hash_mode(v), + ), +] + +_BY_KEY: dict[str, SettingSpec] = {s.key: s for s in DEVICE_SETTINGS} + + +def get_spec(key: str) -> SettingSpec: + """Return the spec for ``key``. + + Args: + key: A setting key. + + Returns: + The matching :class:`SettingSpec`. + + Raises: + DeviceConfigError: If no such setting exists. + """ + spec = _BY_KEY.get(key) + if spec is None: + known = ", ".join(sorted(_BY_KEY)) + raise DeviceConfigError(f"unknown setting {key!r}. Known settings: {known}") + return spec + + +def settings_by_category() -> list[tuple[str, list[SettingSpec]]]: + """Return settings grouped by category in display order. + + Returns: + A list of ``(category, specs)`` pairs following :data:`CATEGORIES` order. + """ + return [ + (cat, [s for s in DEVICE_SETTINGS if s.category == cat]) + for cat in CATEGORIES + ] diff --git a/meshtools/core/models.py b/meshtools/core/models.py index 09812fb..4b9e3f4 100644 --- a/meshtools/core/models.py +++ b/meshtools/core/models.py @@ -7,10 +7,14 @@ real device, the simulator, or rehydrated from the database identically. from __future__ import annotations import statistics +from collections import Counter from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Optional +#: Label used for our own (local) device when framing a trace path's endpoints. +LOCAL_DEVICE_LABEL = "us" + def utcnow() -> datetime: """Return a timezone-aware UTC timestamp. @@ -63,6 +67,8 @@ class TraceResult: hops: Per-hop SNR readings, ordered from source to destination. round_trip_ms: Round-trip time of the trace in milliseconds, if measured. tx_power: TX power level in effect when the trace ran, if known. + path_hash_bytes: Per-hop path-hash width (bytes) used by the trace command, so + node hashes can be displayed at the same width the command addressed them. timestamp: When the trace completed. raw: Optional raw event payload for debugging/replay. """ @@ -72,6 +78,7 @@ class TraceResult: hops: list[Hop] = field(default_factory=list) round_trip_ms: Optional[float] = None tx_power: Optional[int] = None + path_hash_bytes: Optional[int] = None timestamp: datetime = field(default_factory=utcnow) raw: Optional[dict] = None @@ -95,6 +102,67 @@ class TraceResult: nodes = [h.node or f"hop{h.index}" for h in self.hops] return " -> ".join(nodes) + def edges(self, device_label: str = LOCAL_DEVICE_LABEL) -> list["HopEdge"]: + """Frame the per-hop SNR as directed ``origin -> destination`` edges. + + Each hop's SNR is the signal measured arriving at that node, so an edge runs + from the previous node to this one. The first edge therefore originates at our + own device, and (because firmware records the reply returning to us as a final + hash-less hop) the last edge's destination is our device too. + + Args: + device_label: Name to show for our own device at the path's endpoints. + + Returns: + One :class:`HopEdge` per hop, in path order. + """ + edges: list[HopEdge] = [] + origin = device_label + for hop in self.hops: + destination = hop.node or device_label + edges.append(HopEdge(index=hop.index, origin=origin, destination=destination, snr=hop.snr)) + origin = destination + return edges + + +@dataclass(slots=True) +class HopEdge: + """A directed link in a trace path: a hop framed as ``origin -> destination``. + + Attributes: + index: Zero-based position of the hop along the path. + origin: Identifier of the transmitting node (our device for the first edge). + destination: Identifier of the receiving node (our device for the last edge). + snr: Signal-to-noise ratio in dB measured at ``destination``. + """ + + index: int + origin: str + destination: str + snr: float + + +@dataclass(slots=True) +class HopAggregate: + """Median SNR for one hop position aggregated across several traces. + + ``origin`` and ``destination`` hold raw node identifiers; ``None`` means our own + device, so the display label can be applied at render time. + + Attributes: + index: Zero-based hop position along the path. + origin: Representative transmitting node at this position (``None`` = us). + destination: Representative receiving node at this position (``None`` = us). + median_snr: Median SNR in dB measured at ``destination`` across the samples. + samples: Number of traces that reported this hop. + """ + + index: int + origin: Optional[str] + destination: Optional[str] + median_snr: float + samples: int + @dataclass(slots=True) class TraceStats: @@ -110,6 +178,7 @@ class TraceStats: median_min_snr: Median of each trace's bottleneck SNR, the headline metric. median_rtt_ms: Median round-trip time, if measured. tx_power: TX power level in effect for these samples, if fixed. + hop_snrs: Per-hop median SNR across the successful traces, in path order. """ target: str @@ -118,6 +187,7 @@ class TraceStats: median_min_snr: Optional[float] median_rtt_ms: Optional[float] tx_power: Optional[int] = None + hop_snrs: list[HopAggregate] = field(default_factory=list) @property def success_rate(self) -> float: @@ -146,8 +216,51 @@ class TraceStats: median_min_snr=statistics.median(min_snrs) if min_snrs else None, median_rtt_ms=statistics.median(rtts) if rtts else None, tx_power=next(iter(tx_powers)) if len(tx_powers) == 1 else None, + hop_snrs=cls._aggregate_hops(successes), ) + @staticmethod + def _aggregate_hops(successes: list["TraceResult"]) -> list[HopAggregate]: + """Compute the median SNR per hop position across successful traces. + + Hops are grouped by their path position; for each position the SNR median is + taken and the most common origin/destination nodes are used so the aggregate + path reads like a single representative trace. Node identities are kept raw + (``None`` = our device) so a display label can be applied later. + + Args: + successes: The successful traces to aggregate. + + Returns: + One :class:`HopAggregate` per hop position, ordered along the path. + """ + snrs_by_index: dict[int, list[float]] = {} + origins_by_index: dict[int, list[Optional[str]]] = {} + dests_by_index: dict[int, list[Optional[str]]] = {} + for trace in successes: + origin: Optional[str] = None # the first hop originates at our device + for hop in trace.hops: + snrs_by_index.setdefault(hop.index, []).append(hop.snr) + origins_by_index.setdefault(hop.index, []).append(origin) + dests_by_index.setdefault(hop.index, []).append(hop.node) + origin = hop.node + + aggregates: list[HopAggregate] = [] + for index in sorted(snrs_by_index): + snrs = snrs_by_index[index] + origin = Counter(origins_by_index[index]).most_common(1)[0][0] + destination = Counter(dests_by_index[index]).most_common(1)[0][0] + aggregates.append( + HopAggregate( + index=index, + origin=origin, + destination=destination, + median_snr=statistics.median(snrs), + samples=len(snrs), + ) + ) + return aggregates + @dataclass(slots=True) class TxLevelResult: diff --git a/meshtools/persistence/repository.py b/meshtools/persistence/repository.py index 7dc0a91..5aa8027 100644 --- a/meshtools/persistence/repository.py +++ b/meshtools/persistence/repository.py @@ -12,7 +12,9 @@ from dataclasses import dataclass from pathlib import Path from typing import Any, Optional -from ..core.models import TraceResult, TraceStats, utcnow +from datetime import datetime + +from ..core.models import Hop, TraceResult, TraceStats, utcnow from . import db @@ -158,6 +160,53 @@ class Repository: self._conn.commit() return trace_id + def latest_trace( + self, + target: str, + *, + exclude_run_id: Optional[int] = None, + success_only: bool = True, + ) -> Optional[TraceResult]: + """Return the most recently recorded trace to ``target``, rehydrated with hops. + + Used to show the previous run's result before a new trace starts. + + Args: + target: The trace destination to look up. + exclude_run_id: A run to skip (typically the in-progress one) so we surface + a genuinely prior result. + success_only: When ``True``, ignore traces that timed out. + + Returns: + The latest matching :class:`TraceResult`, or ``None`` if none is stored. + """ + sql = "SELECT * FROM traces WHERE target = ?" + params: list[Any] = [target] + if success_only: + sql += " AND success = 1" + if exclude_run_id is not None: + sql += " AND run_id != ?" + params.append(exclude_run_id) + sql += " ORDER BY id DESC LIMIT 1" + row = self._conn.execute(sql, params).fetchone() + if row is None: + return None + + hop_rows = self._conn.execute( + "SELECT hop_index, node, snr FROM trace_hops WHERE trace_id = ? " + "ORDER BY hop_index", + (row["id"],), + ).fetchall() + hops = [Hop(index=h["hop_index"], node=h["node"], snr=h["snr"]) for h in hop_rows] + return TraceResult( + target=row["target"], + success=bool(row["success"]), + hops=hops, + round_trip_ms=row["round_trip_ms"], + tx_power=row["tx_power"], + timestamp=datetime.fromisoformat(row["created_at"]), + ) + def record_tx_sample(self, run_id: int, stats: TraceStats) -> None: """Persist one robust TX-power sample from an optimization sweep. diff --git a/meshtools/services/trace_runner.py b/meshtools/services/trace_runner.py index e159f54..3a856c0 100644 --- a/meshtools/services/trace_runner.py +++ b/meshtools/services/trace_runner.py @@ -11,17 +11,145 @@ import asyncio from typing import Awaitable, Callable, Optional from ..core.connection import Device -from ..core.models import TraceResult, TraceStats +from ..core.models import Contact, TraceResult, TraceStats ProgressCallback = Callable[[int, int, TraceResult], None] +#: Maps a trace hop's key-prefix hash to a display label (a contact name, or the +#: hash itself when unknown). ``None`` passes through (our own device). +NodeResolver = Callable[[Optional[str]], Optional[str]] + + +_HEX_DIGITS = frozenset("0123456789abcdef") + + +def make_node_resolver(contacts: Optional[list[Contact]]) -> NodeResolver: + """Build a resolver that names a trace hop from its key-prefix hash. + + Trace replies identify each repeater only by a short hash — a leading slice of + its public key. This maps that hash to the contact's friendly name when we know + the node, so results read as names instead of opaque hex, and falls back to the + raw hash for unknown nodes. + + Args: + contacts: Known contacts to resolve against. + + Returns: + A callable taking a hop label and returning a contact name when the hash + matches a known contact, otherwise the label unchanged (``None`` = our + device, passed through untouched). + """ + # (name, public_key, key_prefix), all lowercased for case-insensitive prefixing. + entries: list[tuple[str, str, str]] = [] + for c in contacts or []: + pub = (c.public_key or "").lower().removeprefix("0x") + prefix = (c.key_prefix or "").lower().removeprefix("0x") + if c.name and (pub or prefix): + entries.append((c.name, pub, prefix)) + + def resolve(label: Optional[str]) -> Optional[str]: + if not label: + return label + needle = label.lower().removeprefix("0x") + for name, pub, prefix in entries: + # The hash is a prefix of the node's key; match either against the full + # public key or the (possibly shorter) stored prefix, in either direction. + if pub and pub.startswith(needle): + return name + if prefix and (prefix.startswith(needle) or needle.startswith(prefix)): + return name + return label + + return resolve + + +def parse_trace_path(spec: str, contacts: Optional[list[Contact]] = None) -> str: + """Parse a user path spec into the hex path string ``send_trace`` expects. + + The MeshCore trace protocol forces a route through a list of repeaters, each + addressed by a leading slice of its public key (the "path hash"). This mirrors + the comma-separated path field in the MeshCore mobile apps, e.g. ``"3d,f2,3d"``. + + Each comma-separated token may be either a contact name (resolved to its key + prefix, case-insensitively) or a raw hex key prefix; the two may be mixed in a + single spec. The path-hash *width* is taken from the hex tokens you type (every + hop uses the same width); contact names are truncated to that width. With only + contact names and no hex token, each name's full key prefix is used. + + Args: + spec: Comma-separated path, e.g. ``"3d5f7a,Alice,f2a1b3"``. + contacts: Known contacts used to resolve names to key prefixes. + + Returns: + A comma-separated hex string of uniform-width prefixes, e.g. ``"3d5f7a,d4e5f6,f2a1b3"``. + + Raises: + ValueError: If a token is neither a known contact nor valid hex, if the hex + tokens disagree on width, or if the spec contains no usable hops. + """ + by_name = {c.name.casefold(): c.key_prefix for c in (contacts or [])} + # (hex_text, is_name) per hop; names keep their full prefix until width is known. + tokens: list[tuple[str, bool]] = [] + for raw in spec.split(","): + token = raw.strip() + if not token: + continue + key = token.casefold() + if key in by_name: + tokens.append((by_name[key].lower().removeprefix("0x"), True)) + else: + tokens.append((token.lower().removeprefix("0x"), False)) + if not tokens: + raise ValueError("path is empty") + + # The width is dictated by explicit hex tokens; names are truncated to match. + # With names only, fall back to the (uniform) length of the resolved prefixes. + hex_widths = {len(text) for text, is_name in tokens if not is_name} + if len(hex_widths) > 1: + raise ValueError("all hops must use the same number of hex digits") + width = hex_widths.pop() if hex_widths else max(len(text) for text, _ in tokens) + if width == 0 or width % 2: + raise ValueError("hex key prefixes must have an even number of digits") + + hops: list[str] = [] + for text, is_name in tokens: + hop = text[:width] if is_name else text + if len(hop) != width or any(ch not in _HEX_DIGITS for ch in hop): + raise ValueError( + f"{text!r} is not a known contact or a {width // 2}-byte hex prefix" + ) + hops.append(hop) + return ",".join(hops) + + +def path_hash_flags(width_bytes: int) -> Optional[int]: + """Return the trace ``flags`` value that encodes a per-hop path-hash width. + + The trace subsystem encodes the hash size as ``1 << (flags & 3)`` on both the + send and receive sides (``send_trace`` and the ``TRACE_DATA`` reader), so only + widths of 1, 2, 4, or 8 bytes are representable. This is independent of, and + differs from, the ``mode = size - 1`` encoding used for *contact routing* + (``out_path_hash_mode``). + + Args: + width_bytes: Path-hash width in bytes. + + Returns: + The flags value (the exponent ``s``), or ``None`` if the width is not a + representable power of two. + """ + for s in range(4): + if (1 << s) == width_bytes: + return s + return None + async def run_traces( device: Device, target: str, *, samples: int = 5, - path: Optional[list[str]] = None, + path: Optional[str] = None, cooldown_s: float = 1.0, timeout: float = 10.0, on_result: Optional[ProgressCallback] = None, @@ -65,7 +193,7 @@ async def measure( target: str, *, samples: int = 5, - path: Optional[list[str]] = None, + path: Optional[str] = None, cooldown_s: float = 1.0, on_result: Optional[ProgressCallback] = None, persist: Optional[Callable[[TraceResult], Awaitable[None] | None]] = None, diff --git a/meshtools/tools/base.py b/meshtools/tools/base.py index c45d395..b4cbf92 100644 --- a/meshtools/tools/base.py +++ b/meshtools/tools/base.py @@ -105,15 +105,18 @@ class Tool(ABC): Raises: Exception: Re-raises any error from :meth:`run` after recording it. """ + from ..core.connection import DeviceCommandError + from ..core.device_config import DeviceConfigError from ..core.selection import DeviceSelectionError run_id = ctx.repo.start_run(self.name, params, ctx.profile_name) ctx.log.debug("run %s start: tool=%s params=%s", run_id, self.name, params) try: result = await self.run(ctx, {**params, "_run_id": run_id}) - except DeviceSelectionError as exc: - # Expected user-facing condition (no/ambiguous device): record it but don't - # dump a traceback; callers print the guidance message cleanly. + except (DeviceSelectionError, DeviceConfigError, DeviceCommandError) as exc: + # Expected user-facing condition (no/ambiguous device, bad config value, or + # a transient command failure): record it but don't dump a traceback; + # callers print the message cleanly. ctx.repo.finish_run(run_id, "error", {"error": str(exc)}) ctx.log.debug("run %s aborted: %s", run_id, exc) raise diff --git a/meshtools/tools/config.py b/meshtools/tools/config.py new file mode 100644 index 0000000..8eb7354 --- /dev/null +++ b/meshtools/tools/config.py @@ -0,0 +1,358 @@ +"""The ``config`` tool: view and change every device configuration value. + +Interactively it launches a full editor (see :mod:`meshtools.ui.config_editor`); on the +CLI it exposes generic key/value subcommands plus backup/restore, channels, custom vars, +and a gated set of destructive operations. Both paths funnel into :meth:`ConfigTool.run`, +which executes a list of operation tuples so every change is logged through the run record. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Optional + +import typer +from rich.table import Table + +from ..context import AppContext +from ..core.config_io import backup_config, plan_restore, read_backup +from ..core.connection import Device +from ..core.device_config import ( + build_snapshot, + format_value, + get_spec, + parse_value, +) +from .base import Tool, ToolResult, register + +# How many channel slots to probe when backing up. +_MAX_CHANNELS = 8 + + +@register +class ConfigTool(Tool): + """View and edit the connected device's full configuration.""" + + name = "config" + help = "View and change device configuration (name, radio, experimental, ...)." + category = "Device" + order = 8 + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """Launch the interactive editor and collect the operations to perform. + + Args: + ctx: Shared application context. + + Returns: + ``{"ops": [...]}`` to execute, or ``None`` if the user cancelled. + """ + from ..ui.config_editor import edit_config + + ops = await edit_config(ctx) + if not ops: + return None + return {"ops": ops} + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Execute a list of configuration operations against the device. + + Args: + ctx: Shared application context. + params: ``ops`` — a list of operation tuples (see module docstring). + + Returns: + A :class:`ToolResult` summarizing how many values changed. + """ + device = await ctx.device() + ops: list[tuple] = list(params.get("ops") or []) + snapshot = await build_snapshot(device) + changes, artifacts = await apply_ops(ctx, device, snapshot, ops) + + message = f"[ok]✓[/ok] applied [brand]{changes}[/brand] change(s)" if changes else None + return ToolResult(summary={"changes": changes}, message=message, artifacts=artifacts) + + # -- CLI -------------------------------------------------------------------- + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``config`` subcommand group. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + config_app = typer.Typer( + help=self.help, no_args_is_help=False, rich_markup_mode="rich" + ) + + @config_app.callback(invoke_without_command=True) + def _root(ctx: typer.Context) -> None: + """Show the configuration when no subcommand is given.""" + if ctx.invoked_subcommand is None: + run_tool_command(self, {"ops": [("show",)]}) + + @config_app.command("show", help="Show all current settings.") + def _show_cmd() -> None: + run_tool_command(self, {"ops": [("show",)]}) + + @config_app.command("get", help="Print one setting's value.") + def _get_cmd(key: str = typer.Argument(..., help="Setting key.")) -> None: + run_tool_command(self, {"ops": [("get", key)]}) + + @config_app.command("set", help="Set one setting to a value.") + def _set_cmd( + key: str = typer.Argument(..., help="Setting key."), + value: str = typer.Argument(..., help="New value."), + ) -> None: + run_tool_command(self, {"ops": [("set", key, value)]}) + + @config_app.command("backup", help="Write all settings to a TOML file.") + def _backup_cmd(path: Path = typer.Argument(..., help="Destination file.")) -> None: + run_tool_command(self, {"ops": [("backup", path)]}) + + @config_app.command("restore", help="Apply settings from a TOML backup.") + def _restore_cmd( + path: Path = typer.Argument(..., help="Backup file."), + dry_run: bool = typer.Option(False, "--dry-run", help="Preview without applying."), + ) -> None: + run_tool_command(self, {"ops": [("restore", path, dry_run)]}) + + @config_app.command("custom", help="Set an experimental custom variable.") + def _custom_cmd( + key: str = typer.Argument(..., help="Variable name."), + value: str = typer.Argument(..., help="Variable value."), + ) -> None: + run_tool_command(self, {"ops": [("set_custom", key, value)]}) + + @config_app.command("channel", help="Configure a channel slot.") + def _channel_cmd( + index: int = typer.Argument(..., help="Channel slot index."), + name: str = typer.Argument(..., help="Channel name (# derives the secret)."), + secret: Optional[str] = typer.Option(None, "--secret", help="16-byte hex secret."), + ) -> None: + secret_bytes = bytes.fromhex(secret) if secret else None + run_tool_command(self, {"ops": [("set_channel", index, name, secret_bytes)]}) + + @config_app.command("advert", help="Broadcast an advertisement.") + def _advert_cmd( + flood: bool = typer.Option(False, "--flood", help="Flood across the mesh."), + ) -> None: + run_tool_command(self, {"ops": [("advert", flood)]}) + + @config_app.command("export-key", help="Export the private key (sensitive).") + def _export_key_cmd( + out: Optional[Path] = typer.Option(None, "--out", help="Write to file instead of stdout."), + ) -> None: + ops = [("export_key", out)] if out else [("export_key",)] + run_tool_command(self, {"ops": ops}) + + @config_app.command("import-key", help="Import a private key (overwrites identity).") + def _import_key_cmd( + key_hex: str = typer.Argument(..., help="Private key as hex."), + yes: bool = typer.Option(False, "--yes", help="Confirm this destructive action."), + ) -> None: + _require_yes(yes, "import-key overwrites the device identity") + run_tool_command(self, {"ops": [("import_key", key_hex)]}) + + @config_app.command("reboot", help="Reboot the device.") + def _reboot_cmd( + yes: bool = typer.Option(False, "--yes", help="Confirm reboot."), + ) -> None: + _require_yes(yes, "reboot restarts the device") + run_tool_command(self, {"ops": [("reboot",)]}) + + @config_app.command("factory-reset", help="Erase all data and reset to defaults.") + def _factory_reset_cmd( + yes: bool = typer.Option(False, "--yes", help="Confirm this destructive action."), + ) -> None: + _require_yes(yes, "factory-reset erases ALL data on the device") + run_tool_command(self, {"ops": [("factory_reset",)]}) + + app.add_typer(config_app, name=self.name) + + +# -- op execution (shared by the tool and the interactive editor) ------------- + + +async def apply_ops( + ctx: AppContext, device: Device, snapshot: dict, ops: list[tuple] +) -> tuple[int, list[str]]: + """Execute a list of configuration operation tuples against ``device``. + + This is the single executor for every config operation, used both by + :meth:`ConfigTool.run` (for staged, applied changes) and by the interactive editor's + danger zone (which runs its action immediately rather than staging it). + + Args: + ctx: Shared application context (for console output). + device: The connected device to act on. + snapshot: The current device snapshot; updated in place as settings change so + coupled commands rebuild from current values. + ops: Operation tuples (see the module docstring). + + Returns: + A ``(changes, artifacts)`` pair: the number of value changes applied and any + file paths produced. + """ + changes = 0 + artifacts: list[str] = [] + for op in ops: + kind = op[0] + if kind == "show": + await _show(ctx, device, snapshot) + elif kind == "get": + spec = get_spec(op[1]) + ctx.console.print( + f"[muted]{spec.key}[/muted] = {format_value(spec, spec.getter(snapshot))}" + ) + elif kind == "set": + changes += await _apply_setting(ctx, device, op[1], op[2], snapshot) + elif kind == "set_custom": + await device.set_custom_var(op[1], op[2]) + ctx.console.print(f"[ok]✓[/ok] custom [brand]{op[1]}[/brand] = {op[2]}") + changes += 1 + elif kind == "set_channel": + await device.set_channel(op[1], op[2], op[3]) + ctx.console.print(f"[ok]✓[/ok] channel {op[1]} = [brand]{op[2]}[/brand]") + changes += 1 + elif kind == "backup": + artifacts.append(str(await _backup(device, snapshot, op[1]))) + elif kind == "restore": + changes += await _restore(ctx, device, snapshot, op[1], op[2]) + elif kind == "advert": + await device.send_advert(len(op) > 1 and bool(op[1])) + ctx.console.print("[ok]✓[/ok] advertisement sent") + elif kind == "reboot": + await device.reboot() + ctx.console.print("[warn]device rebooting[/warn]") + elif kind == "export_key": + await _export_key(ctx, device, op[1] if len(op) > 1 else None, artifacts) + elif kind == "import_key": + await device.import_private_key(op[1]) + ctx.console.print("[ok]✓[/ok] private key imported") + changes += 1 + elif kind == "factory_reset": + await device.factory_reset() + ctx.console.print("[err]device factory-reset[/err]") + else: # pragma: no cover - guarded by the call sites that build ops + ctx.console.print(f"[err]unknown config op: {kind}[/err]") + return changes, artifacts + + +async def _apply_setting( + ctx: AppContext, device: Device, key: str, raw: Any, snapshot: dict +) -> int: + """Parse, apply and record one setting; keep ``snapshot`` consistent. + + The local snapshot is updated with the new value so a later coupled change in the + same batch (e.g. another radio field) is rebuilt from current values. + + Returns: + ``1`` (a change was applied). + """ + spec = get_spec(key) + value = parse_value(spec, raw) + await spec.apply(device, value, snapshot) + snapshot[key] = value + ctx.console.print(f"[ok]✓[/ok] [brand]{key}[/brand] = {format_value(spec, value)}") + return 1 + + +async def _show(ctx: AppContext, device: Device, snapshot: dict) -> None: + """Render the full configuration table plus custom variables.""" + from ..ui.config_editor import render_config + + custom = await device.get_custom_vars() + render_config(ctx.console, snapshot, custom) + + +async def _backup(device: Device, snapshot: dict, path: Path) -> Path: + """Write a TOML backup of the current configuration.""" + custom = await device.get_custom_vars() + channels = await _read_channels(device) + return backup_config(Path(path), snapshot, custom, channels) + + +async def _restore( + ctx: AppContext, device: Device, snapshot: dict, path: Path, dry_run: bool +) -> int: + """Apply (or preview) a backup file against the current configuration. + + Returns: + The number of changes applied (always ``0`` for a dry run). + """ + backup = read_backup(Path(path)) + custom = await device.get_custom_vars() + ops = plan_restore(backup, snapshot, custom) + if not ops: + ctx.console.print("[muted]restore: device already matches the backup.[/muted]") + return 0 + + if dry_run: + table = Table(title="Restore preview", border_style="warn", expand=False) + table.add_column("Operation") + table.add_column("Target") + table.add_column("Value") + for op in ops: + table.add_row(op[0], str(op[1]), str(op[2] if len(op) > 2 else "")) + ctx.console.print(table) + ctx.console.print("[muted]dry run — nothing was changed.[/muted]") + return 0 + + changes = 0 + for op in ops: + if op[0] == "set": + changes += await _apply_setting(ctx, device, op[1], op[2], snapshot) + elif op[0] == "set_custom": + await device.set_custom_var(op[1], op[2]) + changes += 1 + elif op[0] == "set_channel": + await device.set_channel(op[1], op[2], op[3]) + changes += 1 + return changes + + +async def _export_key( + ctx: AppContext, device: Device, out: Optional[Path], artifacts: list[str] +) -> None: + """Export the private key, to a file if ``out`` is given, else to the console.""" + key_hex = await device.export_private_key() + if out is not None: + path = Path(out) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(key_hex, encoding="utf-8") + artifacts.append(str(path)) + ctx.console.print("[warn]private key written — keep this file secret.[/warn]") + else: + ctx.console.print( + "[warn]private key (keep secret):[/warn] [muted]" + key_hex + "[/muted]" + ) + + +async def _read_channels(device: Device) -> list[dict]: + """Probe channel slots and return the configured ones.""" + channels: list[dict] = [] + for idx in range(_MAX_CHANNELS): + try: + ch = await device.get_channel(idx) + except Exception: # noqa: BLE001 - firmware may not support channel reads + break + if ch: + channels.append(ch) + return channels + + +def _require_yes(yes: bool, what: str) -> None: + """Abort a destructive CLI command unless ``--yes`` was passed. + + Args: + yes: Whether the user passed ``--yes``. + what: Human-readable description of the consequence. + + Raises: + typer.Exit: With code 1 if confirmation was not given. + """ + if not yes: + typer.secho(f"Refusing: {what}. Re-run with --yes to confirm.", fg="red") + raise typer.Exit(1) diff --git a/meshtools/tools/info.py b/meshtools/tools/info.py index 3bf37ab..9ce9e86 100644 --- a/meshtools/tools/info.py +++ b/meshtools/tools/info.py @@ -21,7 +21,7 @@ class InfoTool(Tool): order = 10 async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: - """Query the device and render its self-info and contacts. + """Query the device and render its full configuration and contacts. Args: ctx: Shared application context. @@ -30,17 +30,16 @@ class InfoTool(Tool): Returns: A :class:`ToolResult` summarizing device name and contact count. """ + from ..core.device_config import build_snapshot + from ..ui.config_editor import render_config + device = await ctx.device() - info = await device.get_self_info() + snapshot = await build_snapshot(device) + custom = await device.get_custom_vars() contacts = await device.get_contacts() - table = Table(title="Device", border_style="muted", expand=False) - table.add_column("Field", style="muted") - table.add_column("Value") - for key in ("name", "tx_power", "freq", "bw", "sf", "cr"): - if key in info: - table.add_row(key, str(info[key])) - ctx.console.print(table) + # Render every current setting with a short explanation of each. + render_config(ctx.console, snapshot, custom) if contacts: ctable = Table(title=f"Contacts ({len(contacts)})", border_style="muted") @@ -51,7 +50,7 @@ class InfoTool(Tool): ctx.console.print(ctable) return ToolResult( - summary={"name": info.get("name"), "contacts": len(contacts)}, + summary={"name": snapshot.get("name"), "contacts": len(contacts)}, ) def register_cli(self, app: typer.Typer) -> None: diff --git a/meshtools/tools/trace.py b/meshtools/tools/trace.py index fc3557f..db3ace0 100644 --- a/meshtools/tools/trace.py +++ b/meshtools/tools/trace.py @@ -13,10 +13,15 @@ import questionary import typer from ..context import AppContext +from ..core.models import LOCAL_DEVICE_LABEL from ..services import trace_runner -from ..ui.widgets import make_progress, stats_panel, trace_table +from ..ui.widgets import make_progress, stats_panel, traces_table from .base import Tool, ToolResult, register +#: Upper bound on traces per run — the per-trace table renders one column each, so +#: this keeps the output readable and the radio's duty cycle in check. +MAX_TRACES = 8 + @register class TraceTool(Tool): @@ -52,11 +57,30 @@ class TraceTool(Tool): return None samples_raw = await questionary.text( - "How many traces?", default="5", validate=_is_positive_int + f"How many traces? (1-{MAX_TRACES})", default="3", validate=_is_valid_sample_count ).ask_async() if samples_raw is None: return None - return {"target": target.strip(), "samples": int(samples_raw)} + + # Optional explicit path, MeshCore-app style: comma-separated contact names + # and/or hex key prefixes (e.g. ``3d,f2,3d``). Blank lets the device route. + path_prompt = "Force a path? (comma-separated contacts/hex prefixes, blank = auto):" + validate_path = lambda v: _validate_path(v, contacts) # noqa: E731 + if choices: + path_spec = await questionary.autocomplete( + path_prompt, choices=choices, ignore_case=True, validate=validate_path + ).ask_async() + else: + path_spec = await questionary.text( + path_prompt, validate=validate_path + ).ask_async() + if path_spec is None: + return None + + params: dict[str, Any] = {"target": target.strip(), "samples": int(samples_raw)} + if path_spec.strip(): + params["path"] = path_spec.strip() + return params async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: """Run the traces, persist each one, and render the summary. @@ -69,41 +93,80 @@ class TraceTool(Tool): A :class:`ToolResult` with the aggregated statistics. """ target = params["target"] - samples = int(params.get("samples", 5)) + samples = max(1, min(MAX_TRACES, int(params.get("samples", 3)))) + if int(params.get("samples", 3)) > MAX_TRACES: + ctx.console.print(f"[warn]capping at {MAX_TRACES} traces[/warn]") run_id = params["_run_id"] device = await ctx.device() - last_trace = None + # Our own device's name labels both endpoints of the path (first hop's origin, + # last hop's destination); fall back to a neutral label if it's unavailable. + self_info = await device.get_self_info() + device_label = str(self_info.get("name") or LOCAL_DEVICE_LABEL) + + # Resolve repeater hashes in the results to contact names where we know them, + # so the tables read as names instead of opaque hex prefixes. + contacts = await device.get_contacts() + resolve = trace_runner.make_node_resolver(contacts) + + # Resolve an optional forced path (names/hex) to the hex string the radio wants. + # Leaving it blank is fully supported: the device reuses the route it already + # learned for the target (or floods if it has none). + path: Optional[str] = None + path_spec = params.get("path") + if path_spec: + path = trace_runner.parse_trace_path(path_spec, contacts) + ctx.console.print(f"[dim]forcing path:[/dim] [brand]{path}[/brand]") + else: + ctx.console.print("[dim]path:[/dim] [muted]auto (device-routed)[/muted]") + + traces: list = [] with make_progress(ctx.console) as progress: task = progress.add_task(f"tracing {target}", total=samples) def on_result(done: int, total: int, result) -> None: # noqa: ANN001 - nonlocal last_trace - last_trace = result + traces.append(result) progress.advance(task) stats = await trace_runner.measure( device, target, samples=samples, + path=path, cooldown_s=ctx.settings.trace_cooldown_s, on_result=on_result, persist=lambda t: ctx.repo.record_trace(run_id, t), ) - if last_trace is not None and last_trace.success: - ctx.console.print(trace_table(last_trace)) - ctx.console.print(stats_panel(stats)) + # Use the first reply we got as the representative route shown in the summary + # (forced path, or the route the device resolved when auto-routing). + current = next((t for t in traces if t.success), None) + if traces: + ctx.console.print(traces_table(traces, device_label, resolve)) + ctx.console.print(stats_panel(stats, device_label, resolve, route=current)) + summary: dict[str, Any] = { + "target": stats.target, + "samples": stats.samples, + "success_rate": round(stats.success_rate, 3), + "median_min_snr": stats.median_min_snr, + "median_rtt_ms": stats.median_rtt_ms, + "median_hop_snrs": [ + { + "hop": agg.index, + "from": agg.origin or device_label, + "to": agg.destination or device_label, + "median_snr": agg.median_snr, + } + for agg in stats.hop_snrs + ], + } + if path: + summary["path"] = path + via = f" via [brand]{path}[/brand]" if path else "" return ToolResult( - summary={ - "target": stats.target, - "samples": stats.samples, - "success_rate": round(stats.success_rate, 3), - "median_min_snr": stats.median_min_snr, - "median_rtt_ms": stats.median_rtt_ms, - }, - message=f"[ok]✓[/ok] traced [brand]{target}[/brand] x{samples}", + summary=summary, + message=f"[ok]✓[/ok] traced [brand]{target}[/brand] x{samples}{via}", ) def register_cli(self, app: typer.Typer) -> None: @@ -117,13 +180,43 @@ class TraceTool(Tool): @app.command(name=self.name, help=self.help) def _trace( target: str = typer.Option(..., "--target", "-t", help="Target node name/prefix."), - samples: int = typer.Option(5, "--samples", "-n", help="Number of traces."), + samples: int = typer.Option( + 3, "--samples", "-n", help=f"Number of traces (max {MAX_TRACES})." + ), + path: Optional[str] = typer.Option( + None, + "--path", + "-p", + help="Force a route: comma-separated contact names/hex prefixes (e.g. 3d,f2,3d).", + ), ) -> None: - run_tool_command(self, {"target": target, "samples": samples}) + tool_params: dict[str, Any] = {"target": target, "samples": samples} + if path: + tool_params["path"] = path + run_tool_command(self, tool_params) -def _is_positive_int(value: str) -> bool | str: - """Validate that a questionary text answer is a positive integer. +def _validate_path(value: str, contacts: list[Any]) -> bool | str: + """Validate an optional forced-path spec for the interactive prompt. + + Args: + value: The raw comma-separated path entry; blank means "let the device route". + contacts: Known contacts used to resolve names in the spec. + + Returns: + ``True`` if valid (including blank), otherwise an error message string. + """ + if not value.strip(): + return True + try: + trace_runner.parse_trace_path(value, contacts) + return True + except ValueError as exc: + return str(exc) + + +def _is_valid_sample_count(value: str) -> bool | str: + """Validate that a questionary text answer is a trace count in ``1..MAX_TRACES``. Args: value: The raw user input. @@ -132,6 +225,11 @@ def _is_positive_int(value: str) -> bool | str: ``True`` if valid, otherwise an error message string. """ try: - return int(value) > 0 or "Enter a number greater than zero." + count = int(value) except ValueError: return "Enter a whole number." + if count < 1: + return "Enter a number greater than zero." + if count > MAX_TRACES: + return f"Maximum {MAX_TRACES} traces." + return True diff --git a/meshtools/ui/config_editor.py b/meshtools/ui/config_editor.py new file mode 100644 index 0000000..bbb9586 --- /dev/null +++ b/meshtools/ui/config_editor.py @@ -0,0 +1,394 @@ +"""Interactive device-configuration editor (questionary + Rich). + +Drives the menu flow for the ``config`` tool: it shows the device's current configuration, +lets the user stage changes setting-by-setting (with type-aware prompts and validation), +and handles custom variables, channels, backup/restore and the destructive "danger zone". +It returns an operation list for :class:`~meshtools.tools.config.ConfigTool` to execute and +log; it performs no device writes itself. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Optional + +import questionary +from rich.console import Console +from rich.table import Table + +from ..context import AppContext +from ..core.device_config import ( + DeviceConfigError, + SettingSpec, + build_snapshot, + format_value, + parse_value, + settings_by_category, +) +from .menu import _MENU_STYLE + +# Menu action sentinels (distinct from setting keys, which are plain strings). +_PRESETS = "__presets__" +_CUSTOM = "__custom__" +_CHANNELS = "__channels__" +_BACKUP = "__backup__" +_RESTORE = "__restore__" +_DANGER = "__danger__" +_APPLY = "__apply__" +_CANCEL = "__cancel__" +# Sentinel for the "enter a value myself" option on non-strict enum prompts. +_OTHER = "__other__" + + +async def edit_config(ctx: AppContext) -> Optional[list[tuple]]: + """Run the interactive editor and return the operations to perform. + + Args: + ctx: Shared application context (provides the connected device and console). + + Returns: + A list of operation tuples for the tool to execute, or ``None`` if the user + cancelled without choosing to apply anything. + """ + console = ctx.console + device = await ctx.device() + snapshot = await build_snapshot(device) + custom = await device.get_custom_vars() + + pending: dict[str, Any] = {} # setting key -> staged new value + extra_ops: list[tuple] = [] # custom/channel/backup/restore/danger ops, in order + + while True: + render_config(console, snapshot, custom, pending) + choice = await _main_menu(snapshot, pending, extra_ops) + + if choice in (None, _CANCEL): + return None + if choice == _APPLY: + ops: list[tuple] = [("set", k, v) for k, v in pending.items()] + ops.extend(extra_ops) + return ops or None + if choice == _PRESETS: + await _stage_preset(console, pending) + elif choice == _CUSTOM: + await _stage_custom_var(console, custom, extra_ops) + elif choice == _CHANNELS: + await _stage_channel(console, extra_ops) + elif choice == _BACKUP: + await _stage_backup(extra_ops) + elif choice == _RESTORE: + await _stage_restore(extra_ops) + elif choice == _DANGER: + # Danger-zone actions run immediately (after confirmation), not on Apply. + if await _danger_zone(ctx, device, snapshot): + snapshot = await build_snapshot(device) # state may have changed + custom = await device.get_custom_vars() + else: # a setting key + await _stage_setting(console, choice, snapshot, pending) + + +# --- rendering --------------------------------------------------------------- + + +def render_config( + console: Console, + snapshot: dict, + custom: dict[str, str], + pending: Optional[dict[str, Any]] = None, +) -> None: + """Print the current configuration, overlaying any staged changes. + + Args: + console: Console to render into. + snapshot: Device snapshot from ``build_snapshot``. + custom: Current custom variables. + pending: Optional staged changes (setting key -> new value). + """ + pending = pending or {} + show_staged = bool(pending) + table = Table(title="Device configuration", border_style="muted", expand=False) + table.add_column("Setting", style="muted") + table.add_column("Current") + if show_staged: + table.add_column("Staged", style="warn") + table.add_column("Description", style="muted") + for category, specs in settings_by_category(): + table.add_section() + header = [f"[accent]── {category} ──[/accent]", ""] + if show_staged: + header.append("") + table.add_row(*header, "") + for spec in specs: + current = format_value(spec, spec.getter(snapshot)) + row = [f"{spec.label} [muted]({spec.key})[/muted]", current] + if show_staged: + row.append(format_value(spec, pending[spec.key]) if spec.key in pending else "") + table.add_row(*row, spec.help) + console.print(table) + if custom: + console.print( + "[muted]custom vars:[/muted] " + + ", ".join(f"{k}={v}" for k, v in custom.items()) + ) + + +async def _main_menu(snapshot: dict, pending: dict, extra_ops: list) -> Optional[str]: + """Show the top-level editor menu and return the chosen action or setting key. + + Each setting row shows its current value (and any staged new value) plus a one-line + explanation, so the user can see and understand what they're changing in place. + """ + choices: list[questionary.Choice | questionary.Separator] = [] + for category, specs in settings_by_category(): + choices.append(questionary.Separator(f"── {category} ──")) + for spec in specs: + current = format_value(spec, spec.getter(snapshot)) + shown = ( + f"{current} → {format_value(spec, pending[spec.key])}" + if spec.key in pending + else current + ) + choices.append( + questionary.Choice( + title=f"{spec.label}: {shown} — {spec.help}", value=spec.key + ) + ) + choices.append(questionary.Separator("── More ──")) + choices.append(questionary.Choice(title="Radio presets (standard configs)", value=_PRESETS)) + choices.append(questionary.Choice(title="Custom / experimental vars", value=_CUSTOM)) + choices.append(questionary.Choice(title="Channels", value=_CHANNELS)) + choices.append(questionary.Choice(title="Backup to file", value=_BACKUP)) + choices.append(questionary.Choice(title="Restore from file", value=_RESTORE)) + choices.append(questionary.Choice(title="⚠ Danger zone", value=_DANGER)) + choices.append(questionary.Separator(" ")) + staged = len(pending) + len(extra_ops) + choices.append(questionary.Choice(title=f"✓ Apply ({staged} staged)", value=_APPLY)) + choices.append(questionary.Choice(title="Cancel (discard)", value=_CANCEL)) + + return await questionary.select( + "Edit which setting?", choices=choices, style=_MENU_STYLE, qmark="◆" + ).ask_async() + + +# --- staging individual changes ---------------------------------------------- + + +async def _stage_setting( + console: Console, key: str, snapshot: dict, pending: dict[str, Any] +) -> None: + """Prompt for one setting's new value and stage it.""" + from ..core.device_config import get_spec + + spec = get_spec(key) + current = pending.get(key, spec.getter(snapshot)) + value = await _prompt_value(spec, current) + if value is not None: + pending[key] = value + + +async def _prompt_value(spec: SettingSpec, current: Any) -> Any: + """Prompt for a typed value for ``spec``, returning ``None`` on cancel.""" + if spec.value_type == "bool": + return await questionary.confirm( + f"{spec.label}?", default=bool(current) + ).ask_async() + + if spec.value_type == "enum" and spec.choices is not None: + items: list[questionary.Choice] = [ + questionary.Choice(title=f"{k} — {label}", value=k) + for k, label in spec.choices.items() + ] + # Non-strict enums list the common values for convenience but still accept any + # in-range integer, so offer an escape hatch to type one in. + if not spec.strict_choices: + items.append(questionary.Choice(title="Other (enter a value)…", value=_OTHER)) + selected = await questionary.select( + f"{spec.label} — {spec.help}", + choices=items, + default=current if current in spec.choices else None, + style=_MENU_STYLE, + ).ask_async() + if selected != _OTHER: + return selected + # else: fall through to the free-text prompt below. + + def validate(text: str) -> bool | str: + try: + parse_value(spec, text) + return True + except DeviceConfigError as exc: + return str(exc) + + raw = await questionary.text( + f"{spec.label} — {spec.help}", + default="" if current is None else str(current), + validate=validate, + ).ask_async() + return None if raw is None else parse_value(spec, raw) + + +async def _stage_preset(console: Console, pending: dict[str, Any]) -> None: + """Pick a standard radio preset and stage all of its fields for review/apply.""" + from ..core.device_config import RADIO_PRESETS + + choices = [ + questionary.Choice( + title=f"{p.name}: {p.freq} MHz, BW {p.bw}, SF{p.sf}, CR{p.cr} — {p.help}", + value=i, + ) + for i, p in enumerate(RADIO_PRESETS) + ] + choices.append(questionary.Choice(title="Cancel", value=None)) + idx = await questionary.select( + "Apply which radio preset?", choices=choices, style=_MENU_STYLE + ).ask_async() + if idx is None: + return + preset = RADIO_PRESETS[idx] + pending.update(preset.as_settings()) + console.print( + f"[muted]staged preset[/muted] [brand]{preset.name}[/brand] " + "[muted](review the radio rows, then Apply)[/muted]" + ) + + +async def _stage_custom_var( + console: Console, custom: dict[str, str], extra_ops: list[tuple] +) -> None: + """Prompt for a custom/experimental variable and stage a set operation.""" + key = await questionary.text("Custom variable name:").ask_async() + if not key: + return + value = await questionary.text( + f"Value for {key}:", default=custom.get(key, "") + ).ask_async() + if value is None: + return + extra_ops.append(("set_custom", key.strip(), value)) + + +async def _stage_channel(console: Console, extra_ops: list[tuple]) -> None: + """Prompt for a channel slot's name/secret and stage a set operation.""" + idx_raw = await questionary.text( + "Channel index:", default="0", validate=_is_int + ).ask_async() + if idx_raw is None: + return + name = await questionary.text( + "Channel name (leading # derives the secret from the name):" + ).ask_async() + if not name: + return + secret_hex = await questionary.text( + "Secret (32 hex chars / 16 bytes; blank to derive from name):", + validate=_is_optional_secret, + ).ask_async() + secret = bytes.fromhex(secret_hex) if secret_hex else None + extra_ops.append(("set_channel", int(idx_raw), name, secret)) + + +async def _stage_backup(extra_ops: list[tuple]) -> None: + """Prompt for a backup destination path and stage the operation.""" + path = await questionary.path("Write backup to:", default="meshtools-config.toml").ask_async() + if path: + extra_ops.append(("backup", Path(path))) + + +async def _stage_restore(extra_ops: list[tuple]) -> None: + """Prompt for a backup file and whether to preview, then stage the operation.""" + path = await questionary.path("Restore from:").ask_async() + if not path: + return + dry_run = await questionary.confirm( + "Preview changes only (dry run)?", default=True + ).ask_async() + extra_ops.append(("restore", Path(path), bool(dry_run))) + + +async def _danger_zone(ctx: AppContext, device: Any, snapshot: dict) -> bool: + """Sub-menu for destructive operations, run immediately after confirmation. + + Unlike ordinary settings (which are staged and applied together), danger-zone actions + have no meaningful "preview" and are executed the moment they're confirmed. + + Returns: + ``True`` if the action may have changed device state the editor should re-read + (e.g. a factory reset or key import), so the caller can refresh its snapshot. + """ + from ..tools.config import apply_ops + + console = ctx.console + action = await questionary.select( + "⚠ Danger zone (runs immediately on confirmation):", + choices=[ + questionary.Choice("Send advert", value="advert"), + questionary.Choice("Reboot device", value="reboot"), + questionary.Choice("Export private key", value="export_key"), + questionary.Choice("Import private key", value="import_key"), + questionary.Choice("Factory reset (erase all)", value="factory_reset"), + questionary.Choice("Back", value=None), + ], + style=_MENU_STYLE, + ).ask_async() + + op: Optional[tuple] = None + if action in (None, "Back"): + return False + if action in ("advert", "export_key"): + op = (action,) + elif action == "import_key": + key_hex = await questionary.text("Private key (hex):", validate=_is_hex).ask_async() + if key_hex and await _confirm_typed(console, "IMPORT"): + op = ("import_key", key_hex.strip()) + elif action == "reboot": + if await questionary.confirm("Reboot the device now?", default=False).ask_async(): + op = ("reboot",) + elif action == "factory_reset": + console.print("[err]This erases ALL data on the device and cannot be undone.[/err]") + if await _confirm_typed(console, "RESET"): + op = ("factory_reset",) + + if op is None: + return False + await apply_ops(ctx, device, snapshot, [op]) + return action in ("factory_reset", "import_key") + + +async def _confirm_typed(console: Console, word: str) -> bool: + """Require the user to type ``word`` exactly to confirm a destructive action.""" + typed = await questionary.text(f"Type {word!r} to confirm:").ask_async() + if typed == word: + return True + console.print("[muted]confirmation did not match; skipped.[/muted]") + return False + + +# --- validators -------------------------------------------------------------- + + +def _is_int(text: str) -> bool | str: + """Validate that ``text`` is an integer.""" + try: + int(text) + return True + except ValueError: + return "Enter a whole number." + + +def _is_hex(text: str) -> bool | str: + """Validate that ``text`` is a hex string.""" + try: + bytes.fromhex(text) + return True + except ValueError: + return "Enter hex characters only." + + +def _is_optional_secret(text: str) -> bool | str: + """Validate an optional 16-byte hex channel secret.""" + if text == "": + return True + try: + return len(bytes.fromhex(text)) == 16 or "Secret must be exactly 16 bytes (32 hex chars)." + except ValueError: + return "Enter hex characters only." diff --git a/meshtools/ui/widgets.py b/meshtools/ui/widgets.py index 0f1b39e..963f1cf 100644 --- a/meshtools/ui/widgets.py +++ b/meshtools/ui/widgets.py @@ -2,9 +2,9 @@ from __future__ import annotations -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Callable, Optional -from rich.console import Console +from rich.console import Console, Group from rich.panel import Panel from rich.progress import ( BarColumn, @@ -18,12 +18,31 @@ from rich.table import Table from rich.text import Text from .. import __version__ -from ..core.models import TraceResult, TraceStats, TxOptResult +from ..core.models import ( + LOCAL_DEVICE_LABEL, + HopAggregate, + HopEdge, + TraceResult, + TraceStats, + TxOptResult, +) from .theme import snr_style if TYPE_CHECKING: from ..core.discovery import DiscoveredDevice +#: Maps a hop's raw key-prefix hash to a display label (a contact name when known). +NodeResolver = Callable[[Optional[str]], Optional[str]] + +#: Non-breaking space, used in the route line to keep ``name (hash)`` and a node's +#: trailing arrow on the same line so wraps only ever land *after* an arrow. +_NBSP = " " + + +def _identity(label: Optional[str]) -> Optional[str]: + """Default node resolver: leave labels untouched.""" + return label + def banner( profile: Optional[str], mock: bool, selected_device: Optional["DiscoveredDevice"] = None @@ -74,50 +93,256 @@ def make_progress(console: Console) -> Progress: ) -def trace_table(result: TraceResult) -> Table: - """Render a single trace's per-hop SNR as a table. +def _node_text( + label: Optional[str], device_label: str, resolve: NodeResolver = _identity +) -> Text: + """Style a node label, making our own device stand out from repeaters. Args: - result: The trace to display. + label: The node identifier to render; ``None`` means our own device. + device_label: The label used for our own device. + resolve: Maps a raw hop hash to a friendly name when the node is known. Returns: - A Rich :class:`Table` of hops and SNR values. + A styled :class:`Text` for the node. """ - table = Table(title=f"Trace -> {result.target}", border_style="muted", expand=False) + named = resolve(label) if label else label + resolved = named if named else device_label + style = "accent" if resolved == device_label else "brand" + return Text(resolved, style=style) + + +def _link_text( + origin: Optional[str], + destination: Optional[str], + device_label: str, + resolve: NodeResolver = _identity, +) -> Text: + """Render an ``origin -> destination`` link with a styled arrow. + + Args: + origin: The transmitting node (``None`` = our device). + destination: The receiving node (``None`` = our device). + device_label: The label used for our own device. + resolve: Maps a raw hop hash to a friendly name when the node is known. + + Returns: + A :class:`Text` like ``us → Alice`` with the arrow muted. + """ + text = _node_text(origin, device_label, resolve) + text.append(" → ", style="muted") + text.append_text(_node_text(destination, device_label, resolve)) + return text + + +def traces_table( + traces: list[TraceResult], + device_label: str = LOCAL_DEVICE_LABEL, + resolve: NodeResolver = _identity, +) -> Table: + """Render every trace's per-hop SNR side by side, one column per trace. + + Hops are framed as ``origin -> destination`` links (the first originates at our + device, the last returns to it) and aligned by position across traces, so each + column is one trace's SNR readings down the shared path. A trailing ``min`` row + shows each trace's bottleneck — the per-trace values the run's *median min SNR* + summarizes. Traces that never replied appear as a ``✗`` column. + + Args: + traces: The individual traces to display, in run order. + device_label: Name to show for our own device at the path's endpoints. + resolve: Maps a raw hop hash to a friendly contact name when known. + + Returns: + A Rich :class:`Table` with a column per trace. + """ + target = traces[0].target if traces else "" + table = Table(title=f"Trace → {target}", border_style="muted", expand=False) table.add_column("Hop", justify="right", style="muted") - table.add_column("Node") - table.add_column("SNR (dB)", justify="right") - if not result.success: - table.add_row("-", "[err]no reply[/err]", "-") - return table - for hop in result.hops: + table.add_column("From → To") + for i in range(1, len(traces) + 1): + table.add_column(f"#{i}", justify="right") + + # Index each trace's edges by hop position so columns line up even when traces + # take different-length paths (a missing hop shows as a muted dash). + edges_by_trace = [ + {e.index: e for e in t.edges(device_label)} if t.success else {} for t in traces + ] + indices = sorted({idx for edges in edges_by_trace for idx in edges}) + for idx in indices: + link = next( + ( + _link_text(edges[idx].origin, edges[idx].destination, device_label, resolve) + for edges in edges_by_trace + if idx in edges + ), + Text("—", style="muted"), + ) + row: list[Text] = [Text(str(idx)), link] + for edges in edges_by_trace: + edge = edges.get(idx) + if edge is None: + row.append(Text("—", style="muted")) + else: + row.append(Text(f"{edge.snr:+.1f}", style=snr_style(edge.snr))) + table.add_row(*row) + + # Bottleneck per trace: the weakest link the run's median min SNR is taken over. + min_row: list[Text] = [Text("min", style="muted"), Text("bottleneck", style="muted")] + for trace in traces: + if not trace.success: + min_row.append(Text("✗", style="err")) + elif trace.min_snr is not None: + min_row.append(Text(f"{trace.min_snr:+.1f}", style=snr_style(trace.min_snr))) + else: + min_row.append(Text("—", style="muted")) + table.add_section() + table.add_row(*min_row) + return table + + +def _route_node_text( + label: Optional[str], + device_label: str, + resolve: NodeResolver, + hash_bytes: Optional[int], +) -> Text: + """Render one route node as ``name (hash)``, the hash at the command's width. + + Our own device (``label`` is ``None`` or equal to ``device_label``) carries no + hash. A known node shows its friendly name followed by its key-prefix hash; an + unknown node shows just the hash. The hash is truncated to ``hash_bytes`` — the + per-hop width the trace command used — so it matches how the node was addressed. + + Args: + label: The node's raw hop hash, or ``None``/``device_label`` for our device. + device_label: The label used for our own device. + resolve: Maps a raw hop hash to a friendly contact name when known. + hash_bytes: Path-hash width (bytes) to truncate the shown hash to. + + Returns: + A styled :class:`Text` for the node. A non-breaking space joins name and hash + so they never split across a line wrap. + """ + if not label or label == device_label: + return Text(device_label, style="accent") + raw = label.lower().removeprefix("0x") + shown = raw[: hash_bytes * 2] if hash_bytes else raw + named = resolve(label) + if not named or named == label: + return Text(shown, style="brand") # unknown node: hash only + text = Text(named, style="brand") + text.append(f"{_NBSP}(", style="muted") # nbsp keeps "name (hash)" together + text.append(shown, style="muted") + text.append(")", style="muted") + return text + + +def _route_text( + result: TraceResult, + device_label: str = LOCAL_DEVICE_LABEL, + resolve: NodeResolver = _identity, +) -> Text: + """Render a trace's route as a sequence of ``name (hash)`` nodes. + + Shows the path the trace actually walked — the forced path, or the route the + device resolved when auto-routing — with each node annotated by its hash at the + command's path-hash width, joined by muted arrows, e.g. + ``Me → Alice (3d63) → Bob (f2a1) → Me``. The line wraps after an arrow when it is + too long for the panel, so each continuation line starts on a node. + + Args: + result: The trace whose route to display. + device_label: Name to show for our own device at the path's endpoints. + resolve: Maps a raw hop hash to a friendly contact name when known. + + Returns: + A :class:`Text` with the node sequence, or a muted note when no hops exist. + """ + edges = result.edges(device_label) + if not edges: + return Text("no hops recorded", style="muted") + hash_bytes = result.path_hash_bytes + nodes = [edges[0].origin] + [edge.destination for edge in edges] + text = Text() + for i, node in enumerate(nodes): + if i: + # A non-breaking space glues the arrow to the preceding node; the trailing + # regular space is the only wrap point, so wrapping lands *after* the arrow. + text.append(f"{_NBSP}→ ", style="muted") + text.append_text(_route_node_text(node, device_label, resolve, hash_bytes)) + return text + + +def _hop_medians_table( + hop_snrs: list[HopAggregate], device_label: str, resolve: NodeResolver = _identity +) -> Table: + """Render the per-hop median SNR aggregated across a run's traces. + + Args: + hop_snrs: The per-hop aggregates to display. + device_label: Name to show for our own device at the path's endpoints. + resolve: Maps a raw hop hash to a friendly contact name when known. + + Returns: + A compact Rich :class:`Table` of hop, link, and median SNR. + """ + table = Table(box=None, padding=(0, 1, 0, 0), expand=False) + table.add_column("hop", justify="right", style="muted") + table.add_column("from → to") + table.add_column("median SNR", justify="right") + for agg in hop_snrs: table.add_row( - str(hop.index), - hop.node or "[muted]?[/muted]", - Text(f"{hop.snr:+.1f}", style=snr_style(hop.snr)), + str(agg.index), + _link_text(agg.origin, agg.destination, device_label, resolve), + Text(f"{agg.median_snr:+.1f} dB", style=snr_style(agg.median_snr)), ) return table -def stats_panel(stats: TraceStats) -> Panel: +def stats_panel( + stats: TraceStats, + device_label: str = LOCAL_DEVICE_LABEL, + resolve: NodeResolver = _identity, + route: Optional[TraceResult] = None, +) -> Panel: """Summarize aggregated trace statistics in a panel. + Includes the median SNR for every hop along the path (not just the bottleneck), so + a weak link anywhere in the route is visible. When a representative ``route`` trace + is given, the path it actually walked is shown as a node sequence, e.g. + ``Me → Alice → Bob → Me``. + Args: stats: The aggregated statistics to display. + device_label: Name to show for our own device at the path's endpoints. + resolve: Maps a raw hop hash to a friendly contact name when known. + route: A representative trace whose walked route to display, if any. Returns: - A Rich :class:`Panel` with success rate and robust SNR/RTT. + A Rich :class:`Panel` with the route, success rate, robust SNR/RTT, and + per-hop medians. """ snr = stats.median_min_snr snr_text = Text(f"{snr:+.1f} dB", style=snr_style(snr)) if snr is not None else Text("n/a") rtt = f"{stats.median_rtt_ms:.0f} ms" if stats.median_rtt_ms is not None else "n/a" - body = Text.assemble( + summary = Text.assemble( ("target ", "muted"), (f"{stats.target}\n", ""), ("success rate ", "muted"), (f"{stats.success_rate:.0%} " f"({stats.successes}/{stats.samples})\n", ""), ("median min SNR ", "muted"), snr_text, ("\n", ""), ("median RTT ", "muted"), (rtt, ""), ) + sections: list[Text | Table] = [] + if route is not None: + sections.append(Text("route", style="muted")) + sections.append(_route_text(route, device_label, resolve)) + sections.append(Text()) # blank line before the stats block + sections.append(summary) + if stats.hop_snrs: + sections.append(Text("\nper-hop medians", style="muted")) + sections.append(_hop_medians_table(stats.hop_snrs, device_label, resolve)) + body: Text | Group = sections[0] if len(sections) == 1 else Group(*sections) return Panel(body, title="[accent]trace summary[/accent]", border_style="accent", expand=False) diff --git a/pyproject.toml b/pyproject.toml index 9638cb5..8fd0aa4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,6 +22,7 @@ dependencies = [ "pyvis>=0.3.2", # interactive HTML mesh graphs "networkx>=3.2", # graph model + path scoring "tomli>=2.0; python_version < '3.11'", # TOML config reader on 3.10 + "tomli-w>=1.0", # TOML writer for device-config backups ] [project.optional-dependencies] diff --git a/tests/test_device_config.py b/tests/test_device_config.py new file mode 100644 index 0000000..ad1a10a --- /dev/null +++ b/tests/test_device_config.py @@ -0,0 +1,193 @@ +"""Tests for the device-configuration registry, backup/restore, and safety gating. + +All run against :class:`MockDevice` — no hardware required. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import typer + +from meshtools.core.config_io import backup_config, plan_restore, read_backup +from meshtools.core.connection import MockDevice +from meshtools.core.device_config import ( + RADIO_PRESETS, + DeviceConfigError, + build_snapshot, + format_value, + get_spec, + parse_value, +) +from meshtools.tools.config import _require_yes + + +async def _connected_mock() -> MockDevice: + """Return a connected MockDevice.""" + device = MockDevice() + await device.connect() + return device + + +# -- parsing / formatting ------------------------------------------------------ + + +def test_parse_value_types() -> None: + """Each value type coerces from string correctly.""" + assert parse_value(get_spec("name"), "Node-7") == "Node-7" + assert parse_value(get_spec("radio_sf"), "9") == 9 + assert parse_value(get_spec("radio_freq"), "868.0") == 868.0 + assert parse_value(get_spec("manual_add_contacts"), "yes") is True + assert parse_value(get_spec("manual_add_contacts"), "off") is False + + +def test_parse_value_range_and_enum_errors() -> None: + """Out-of-range and invalid enum values raise DeviceConfigError.""" + with pytest.raises(DeviceConfigError, match="<= 12"): + parse_value(get_spec("radio_sf"), "99") + with pytest.raises(DeviceConfigError, match=">= 5"): + parse_value(get_spec("radio_cr"), "1") + with pytest.raises(DeviceConfigError, match="one of"): + parse_value(get_spec("telemetry_mode_base"), "7") + with pytest.raises(DeviceConfigError, match="boolean"): + parse_value(get_spec("manual_add_contacts"), "maybe") + + +def test_format_value() -> None: + """Formatting renders booleans, enums, and unknowns readably.""" + assert format_value(get_spec("manual_add_contacts"), True) == "true" + assert format_value(get_spec("telemetry_mode_base"), 1) == "1 (always)" + assert format_value(get_spec("name"), None) == "?" + + +def test_path_hash_mode_is_strict_enum() -> None: + """path_hash_mode accepts its four modes and rejects anything else.""" + assert parse_value(get_spec("path_hash_mode"), "3") == 3 + assert format_value(get_spec("path_hash_mode"), 0) == "0 (1-byte hashes (default))" + with pytest.raises(DeviceConfigError, match="one of"): + parse_value(get_spec("path_hash_mode"), "4") + + +def test_non_strict_enum_accepts_unlisted_value() -> None: + """multi_acks/adv_loc_policy list common values but still accept other ints.""" + spec = get_spec("multi_acks") + assert spec.value_type == "enum" and not spec.strict_choices + assert parse_value(spec, "5") == 5 # not in choices, but >= minimum: accepted + with pytest.raises(DeviceConfigError, match=">= 0"): + parse_value(spec, "-1") # still range-checked + + +def test_radio_presets_map_to_radio_settings() -> None: + """Every preset exposes the four radio fields by their registry keys.""" + assert RADIO_PRESETS, "expected at least one standard radio preset" + for preset in RADIO_PRESETS: + settings = preset.as_settings() + assert set(settings) == {"radio_freq", "radio_bw", "radio_sf", "radio_cr"} + # Each value parses cleanly through its setting spec. + for key, value in settings.items(): + assert parse_value(get_spec(key), value) == value + + +def test_get_spec_unknown_raises() -> None: + """An unknown key produces a helpful error listing valid keys.""" + with pytest.raises(DeviceConfigError, match="unknown setting"): + get_spec("not_a_setting") + + +# -- apply against the mock device --------------------------------------------- + + +async def test_apply_simple_setting() -> None: + """Setting the node name round-trips through the device.""" + device = await _connected_mock() + spec = get_spec("name") + await spec.apply(device, parse_value(spec, "Yagi-Hub"), await build_snapshot(device)) + assert (await device.get_self_info())["name"] == "Yagi-Hub" + + +async def test_apply_coupled_radio_field_preserves_others() -> None: + """Changing one radio field rebuilds set_radio from the current snapshot.""" + device = await _connected_mock() + spec = get_spec("radio_sf") + await spec.apply(device, 9, await build_snapshot(device)) + info = await device.get_self_info() + assert info["radio_sf"] == 9 + assert info["radio_freq"] == 869.525 # unchanged + assert info["radio_bw"] == 250.0 + assert info["radio_cr"] == 5 + + +async def test_apply_telemetry_mode_preserves_siblings() -> None: + """Editing one telemetry mode leaves the other two intact.""" + device = await _connected_mock() + spec = get_spec("telemetry_mode_loc") + await spec.apply(device, 2, await build_snapshot(device)) + info = await device.get_self_info() + assert info["telemetry_mode_loc"] == 2 + assert info["telemetry_mode_base"] == 0 + assert info["telemetry_mode_env"] == 0 + + +async def test_custom_var_round_trip() -> None: + """Custom variables persist on the device.""" + device = await _connected_mock() + await device.set_custom_var("exp_flag", "1") + assert (await device.get_custom_vars())["exp_flag"] == "1" + + +# -- backup / restore ---------------------------------------------------------- + + +async def test_backup_and_read_round_trip(tmp_path: Path) -> None: + """A backup writes all settings, custom vars, and channels and reads back.""" + device = await _connected_mock() + await device.set_name("Foo") + await device.set_channel(0, "Public", b"\x01" * 16) + snapshot = await build_snapshot(device) + path = backup_config(tmp_path / "cfg.toml", snapshot, {"exp": "1"}, await _channels(device)) + + backup = read_backup(path) + assert backup.settings["name"] == "Foo" + assert backup.settings["radio_sf"] == 11 + assert backup.custom["exp"] == "1" + assert backup.channels[0]["name"] == "Public" + assert backup.channels[0]["secret"] == ("01" * 16) + + +async def test_plan_restore_emits_only_differences(tmp_path: Path) -> None: + """The restore planner returns ops only for values that differ from the device.""" + source = await _connected_mock() + await source.set_name("Foo") + path = backup_config( + tmp_path / "cfg.toml", await build_snapshot(source), {"exp": "1"}, [] + ) + backup = read_backup(path) + + target = await _connected_mock() # fresh device: name still "MockCompanion" + ops = plan_restore(backup, await build_snapshot(target), await target.get_custom_vars()) + + assert ("set", "name", "Foo") in ops + assert ("set_custom", "exp", "1") in ops + # Unchanged values (e.g. radio_sf 11 == 11) are not re-applied. + assert "radio_sf" not in [op[1] for op in ops if op[0] == "set"] + + +# -- safety gating ------------------------------------------------------------- + + +def test_require_yes_blocks_without_confirmation() -> None: + """Destructive CLI commands abort unless --yes is passed.""" + with pytest.raises(typer.Exit): + _require_yes(False, "factory reset erases all data") + _require_yes(True, "factory reset erases all data") # does not raise + + +async def _channels(device: MockDevice) -> list[dict]: + """Collect the mock's configured channels for backup.""" + channels = [] + for idx in range(8): + ch = await device.get_channel(idx) + if ch: + channels.append(ch) + return channels diff --git a/tests/test_skeleton.py b/tests/test_skeleton.py index e782499..4aacdbc 100644 --- a/tests/test_skeleton.py +++ b/tests/test_skeleton.py @@ -10,7 +10,7 @@ from pathlib import Path import pytest from meshtools.core.connection import MockDevice, clamp_tx_power -from meshtools.core.models import TraceResult, TraceStats +from meshtools.core.models import Contact, TraceResult, TraceStats from meshtools.persistence.repository import Repository from meshtools.services import trace_runner @@ -38,6 +38,163 @@ def test_trace_stats_aggregation() -> None: assert stats.median_min_snr == pytest.approx(-1.0) # median of [-2.0, 0.0] +def test_parse_trace_path_mixes_names_and_hex() -> None: + """Names and hex mix; the hex width is preserved and names truncate to match.""" + contacts = [Contact(name="Alice", key_prefix="d4e5f6a7")] + # 1-byte hops: hex, a contact name (case-insensitively), hex again + assert trace_runner.parse_trace_path("3d, alice ,f2", contacts) == "3d,d4,f2" + # the width comes from the hex tokens you type — no truncation to 1 byte + assert trace_runner.parse_trace_path("a1b2c3", contacts) == "a1b2c3" + + +def test_parse_trace_path_preserves_three_byte_width() -> None: + """A 3-byte (6 hex) width is kept, and names truncate to that width.""" + contacts = [Contact(name="Alice", key_prefix="d4e5f6a7b8c9")] + assert ( + trace_runner.parse_trace_path("3d5f7a,Alice,f2a1b3", contacts) + == "3d5f7a,d4e5f6,f2a1b3" + ) + + +def test_parse_trace_path_rejects_mixed_widths() -> None: + """Hex hops must all use the same number of bytes.""" + with pytest.raises(ValueError): + trace_runner.parse_trace_path("3d,a1b2c3", contacts=[]) + + +def test_path_hash_flags_power_of_two_only() -> None: + """Trace flags encode the hash width as 1 << s, so only 1/2/4/8 bytes map.""" + assert trace_runner.path_hash_flags(1) == 0 + assert trace_runner.path_hash_flags(2) == 1 + assert trace_runner.path_hash_flags(4) == 2 + assert trace_runner.path_hash_flags(8) == 3 + assert trace_runner.path_hash_flags(3) is None + + +def test_parse_trace_hops_reads_path_snr() -> None: + """Per-hop SNR comes from payload['path'] dicts; the final hash-less node counts.""" + from meshtools.core.connection import parse_trace_hops + + payload = { + "path": [ + {"hash": "3d", "snr": 4.5}, + {"hash": "f2", "snr": -2.0}, + {"snr": 1.25}, # final node: our device, no hash + ] + } + hops = parse_trace_hops(payload) + assert [h.node for h in hops] == ["3d", "f2", None] + assert [h.snr for h in hops] == [4.5, -2.0, 1.25] + assert [h.index for h in hops] == [0, 1, 2] + + +def test_parse_trace_hops_empty_without_path() -> None: + """A reply with no parsed path yields no hops (SNR shows as n/a).""" + from meshtools.core.connection import parse_trace_hops + + assert parse_trace_hops({}) == [] + + +def test_route_text_annotates_nodes_with_command_width_hash() -> None: + """The route shows each node's hash at the command's path-hash width.""" + from meshtools.core.models import Hop + from meshtools.ui.widgets import _route_text + + resolve = trace_runner.make_node_resolver( + [Contact(name="Alice", public_key="3d63c6" + "00" * 26, key_prefix="3d63c6429436")] + ) + # Known node (resolves to Alice) then the reply returning to our device (node=None). + result = TraceResult( + target="Alice", + success=True, + hops=[Hop(0, "3d63c6", 12.0), Hop(1, None, 12.0)], + path_hash_bytes=2, # command used 2-byte hashes + ) + plain = _route_text(result, "Me", resolve).plain + assert "Alice" in plain + assert "(3d63)" in plain # truncated to 2 bytes, not the full 3d63c6 + assert "3d63c6" not in plain + assert plain.count("Me") == 2 # our device at both ends, carrying no hash + assert "Me (" not in plain.replace(" ", " ") # device never annotated + + +def test_route_text_unknown_node_shows_hash_only() -> None: + """An unresolved node shows just its hash, truncated to the command width.""" + from meshtools.core.models import Hop + from meshtools.ui.widgets import _route_text + + result = TraceResult( + target="x", + success=True, + hops=[Hop(0, "aa11bb", 12.0), Hop(1, None, 12.0)], + path_hash_bytes=1, # 1-byte width + ) + plain = _route_text(result, "Me").plain.replace(" ", " ") + assert "→ aa →" in plain # 1-byte hash, no name, no parens + assert "aa11bb" not in plain + + +def test_parse_trace_path_rejects_unknown_token() -> None: + """A token that is neither a known contact nor valid hex is an error.""" + with pytest.raises(ValueError): + trace_runner.parse_trace_path("nope", contacts=[]) + with pytest.raises(ValueError): + trace_runner.parse_trace_path(" ", contacts=[]) + + +async def test_mock_device_honors_forced_path() -> None: + """A forced path drives the repeater hops; a return-to-us hop is appended.""" + device = MockDevice() + await device.connect() + result = await device.run_trace("Alice", path="3d,f2,3d") + # Three forced repeater hops, plus the hash-less hop for the reply returning to us. + assert result.hop_count == 4 + assert [h.node for h in result.hops] == ["3d", "f2", "3d", None] + + +async def test_mock_device_blank_path_traces() -> None: + """A blank/auto path still produces a successful, non-empty trace.""" + device = MockDevice() + await device.connect() + # Try several times since the simulator drops very weak links occasionally. + results = [await device.run_trace("Alice") for _ in range(10)] + assert any(r.success and r.hops for r in results) + + +def test_trace_edges_endpoints_are_our_device() -> None: + """The first edge originates at us and the last edge returns to us (#3/#4).""" + from meshtools.core.models import Hop + + result = TraceResult( + target="Alice", + success=True, + hops=[Hop(0, "3d", 5.0), Hop(1, "f2", 1.0), Hop(2, None, -1.0)], + ) + edges = result.edges(device_label="us") + assert [(e.origin, e.destination) for e in edges] == [ + ("us", "3d"), + ("3d", "f2"), + ("f2", "us"), + ] + assert [e.snr for e in edges] == [5.0, 1.0, -1.0] + + +def test_trace_stats_reports_per_hop_medians() -> None: + """Aggregation yields a median SNR for every hop position (#5).""" + from meshtools.core.models import Hop + + traces = [ + TraceResult(target="x", success=True, hops=[Hop(0, "a", 4.0), Hop(1, None, 0.0)]), + TraceResult(target="x", success=True, hops=[Hop(0, "a", 8.0), Hop(1, None, 2.0)]), + ] + stats = TraceStats.from_traces("x", traces) + # Node identities are raw (None = our device) so a label is applied at render time. + assert [(h.origin, h.destination, h.median_snr) for h in stats.hop_snrs] == [ + (None, "a", 6.0), + ("a", None, 1.0), + ] + + async def test_mock_device_trace_is_unimodal_in_tx() -> None: """The simulator peaks near its optimal TX power (averaged over noise).""" device = MockDevice(optimal_tx=14) @@ -91,6 +248,191 @@ def test_tx_plot_writes_html(tmp_path: Path) -> None: assert path.stat().st_size > 1000 # plotly.js inlined → non-trivial size +class _Event: + def __init__(self, payload: dict) -> None: + self.payload = payload + + def is_error(self) -> bool: + return False + + +def _mc_with(contacts: dict, path_hash_mode: int = 2): + """Build a minimal mock ``MeshCore`` client exposing the trace commands used.""" + + class _Commands: + async def get_contacts(self) -> _Event: + return _Event(contacts) + + async def get_path_hash_mode(self) -> int: + return path_hash_mode + + class _MC: + commands = _Commands() + + return _MC() + + +async def test_three_byte_route_appends_destination_hash() -> None: + """A learned multi-hop route walks the repeaters then ends at the target (#1). + + Two things matter for a non-power-of-two region (3-byte hashes): each 3-byte + routing hash collapses to its leading 2 bytes (the widest representable trace + width), and the destination's own hash is appended as the final hop — a trace + only replies when its destination is the last hop. + """ + from meshtools.core.connection import MeshCoreDevice + + mc = _mc_with( + { + "Repeater": { + "adv_name": "Repeater", + "public_key": "aabbcc" + "00" * 29, + # Two hops, each a 3-byte routing hash (mode 2 => size 3). + "out_path": "112233445566", + "out_path_len": 2, + "out_path_hash_mode": 2, + } + } + ) + + device = MeshCoreDevice(port="COM-test") + resolved = await device._trace_path_to_contact(mc, "Repeater") + assert resolved is not None + path_bytes, flags = resolved + # Two repeater hops collapsed to 2 bytes, then the target's own 2-byte hash. + assert path_bytes == bytes.fromhex("1122" "4455" "aabb") + assert flags == trace_runner.path_hash_flags(2) # 2-byte width + + +async def test_direct_neighbor_resolves_to_destination_hash() -> None: + """A direct neighbor (no learned route) resolves to a single destination hop. + + Verified on hardware: these contacts report ``out_path_len == -1`` (no stored + route) yet answer a single-hop trace addressed to their own hash. With the + contact's mode unknown (-1) the region's path-hash mode (3-byte) sets the width, + collapsed to the representable 2 bytes. + """ + from meshtools.core.connection import MeshCoreDevice + + mc = _mc_with( + { + "Neighbor": { + "adv_name": "Neighbor", + "public_key": "3d63c6" + "00" * 29, + # Direct neighbor as the firmware reports it: no learned route. + "out_path": "", + "out_path_len": -1, + "out_path_hash_mode": -1, + } + }, + path_hash_mode=2, + ) + + device = MeshCoreDevice(port="COM-test") + resolved = await device._trace_path_to_contact(mc, "Neighbor") + assert resolved is not None + path_bytes, flags = resolved + # Just the target's own hash, region width 3 collapsed to a 2-byte trace hop. + assert path_bytes == bytes.fromhex("3d63") + assert flags == trace_runner.path_hash_flags(2) + + +async def test_unknown_contact_resolves_to_none() -> None: + """An unrecognized target yields ``None`` so the trace can fall back to path-less.""" + from meshtools.core.connection import MeshCoreDevice + + mc = _mc_with( + { + "Somebody": { + "adv_name": "Somebody", + "public_key": "abcdef" + "00" * 29, + "out_path": "", + "out_path_len": -1, + "out_path_hash_mode": -1, + } + } + ) + + device = MeshCoreDevice(port="COM-test") + resolved = await device._trace_path_to_contact(mc, "Nobody") + assert resolved is None + + +async def test_contacts_payload_retries_transient_error() -> None: + """A transient 'no event received' on contacts retrieval is retried, not fatal.""" + from meshtools.core.connection import MeshCoreDevice + + calls = {"n": 0} + + class _Err: + payload = {"reason": "no event received during contacts retrieval"} + + def is_error(self) -> bool: + return True + + class _Ok: + payload = {"Repeater": {"adv_name": "Repeater", "public_key": "aabbcc" + "00" * 29}} + + def is_error(self) -> bool: + return False + + class _Commands: + async def get_contacts(self): # noqa: ANN202 + calls["n"] += 1 + return _Err() if calls["n"] < 3 else _Ok() + + class _MC: + commands = _Commands() + + device = MeshCoreDevice(port="COM-test") + payload = await device._contacts_payload(_MC(), retries=3, delay=0) + assert "Repeater" in payload + assert calls["n"] == 3 # failed twice, succeeded on the third attempt + + +async def test_contacts_payload_raises_clean_error_after_retries() -> None: + """Persistent contacts-retrieval failure raises a clean, actionable error.""" + from meshtools.core.connection import DeviceCommandError, MeshCoreDevice + + class _Err: + payload = {"reason": "no event received during contacts retrieval"} + + def is_error(self) -> bool: + return True + + class _Commands: + async def get_contacts(self): # noqa: ANN202 + return _Err() + + class _MC: + commands = _Commands() + + device = MeshCoreDevice(port="COM-test") + with pytest.raises(DeviceCommandError) as exc: + await device._contacts_payload(_MC(), retries=2, delay=0) + assert "no event received" in str(exc.value) + + +async def test_latest_trace_returns_previous_run(tmp_path: Path) -> None: + """latest_trace rehydrates the most recent stored trace, excluding a given run.""" + from meshtools.core.models import Hop + + repo = Repository(tmp_path / "prev.db") + old_run = repo.start_run("trace", {"target": "Alice"}) + repo.record_trace( + old_run, + TraceResult(target="Alice", success=True, hops=[Hop(0, "3d", 4.0), Hop(1, None, 1.0)]), + ) + new_run = repo.start_run("trace", {"target": "Alice"}) + + prev = repo.latest_trace("Alice", exclude_run_id=new_run) + assert prev is not None + assert [h.node for h in prev.hops] == ["3d", None] + assert prev.edges("us")[-1].destination == "us" + assert repo.latest_trace("Nobody") is None + repo.close() + + async def test_repository_round_trip(tmp_path: Path) -> None: """A run and its traces persist and read back.""" repo = Repository(tmp_path / "test.db")