diff --git a/meshterm/context.py b/meshterm/context.py index 2697b15..85a42ee 100644 --- a/meshterm/context.py +++ b/meshterm/context.py @@ -15,6 +15,7 @@ from rich.console import Console from .core.admin_store import AdminStore from .core.advert_store import AdvertStore from .core.config import DeviceProfile, Settings +from .core.remote_store import RemoteStore from .core.connection import Device, make_device from .core.device_store import DeviceStore from .core.discovery import DiscoveredDevice, discover_devices @@ -43,6 +44,8 @@ class AppContext: admin_store: Store for remembered remote-node admin passwords. advert_store: Store for per-device background-advert schedules (defaults to ``/adverts.json`` when not injected). + remote_store: Store for remote-node admin state — cached settings and CLI + history (defaults to ``/remote.json`` when not injected). mock: Whether the simulator device is in use. port_override: Explicit serial port (from ``--port`` or the interactive picker), overriding the profile. @@ -62,6 +65,7 @@ class AppContext: device_store: DeviceStore admin_store: AdminStore advert_store: Optional[AdvertStore] = None + remote_store: Optional[RemoteStore] = None profile: Optional[DeviceProfile] = None mock: bool = False port_override: Optional[str] = None @@ -90,9 +94,11 @@ class AppContext: _ui: "Optional[Ui]" = field(default=None, init=False, repr=False) def __post_init__(self) -> None: - """Derive the advert store's default location when one was not injected.""" + """Derive the JSON stores' default locations when they were not injected.""" if self.advert_store is None: self.advert_store = AdvertStore(self.settings.config_dir / "adverts.json") + if self.remote_store is None: + self.remote_store = RemoteStore(self.settings.config_dir / "remote.json") @property def profile_name(self) -> Optional[str]: diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 3664997..4f1b93f 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -349,6 +349,30 @@ class Device(ABC): password or no response). """ + @abstractmethod + async def send_remote_command( + self, node: Contact, command: str, *, timeout: float = 8.0 + ) -> Optional[str]: + """Send one CLI command to a logged-in remote node and await its text reply. + + The generic remote-administration primitive: repeaters and room servers are + configured through their text CLI (``get``/``set``/``advert``/…) carried as + admin messages, and every higher-level remote operation is a spelling of this. + Requires an authenticated session (:meth:`admin_login` first) — firmware + silently ignores commands from strangers, which surfaces as a ``None`` reply. + + Args: + node: The remote contact (must already be logged in). + command: The CLI command text, e.g. ``"set txdelay 5"``. + timeout: Seconds to wait for the node's reply. + + Returns: + The reply text, or ``None`` if the node did not answer in time. + + Raises: + DeviceCommandError: If the companion rejected the send outright. + """ + @abstractmethod async def get_remote_tx_power(self, node: Contact) -> Optional[int]: """Read a remote (admin) node's current transmit power. @@ -1391,6 +1415,11 @@ class MeshCoreDevice(Device): payload = getattr(reply, "payload", {}) or {} return str(payload.get("text", payload.get("msg", ""))) + async def send_remote_command( # noqa: D102 - inherited docstring + self, node: Contact, command: str, *, timeout: float = 8.0 + ) -> Optional[str]: + return await self._send_admin_cmd(node, command, timeout=timeout) + async def get_remote_tx_power(self, node: Contact) -> Optional[int]: # noqa: D102 reply = await self._send_admin_cmd(node, "get tx") return _parse_tx_reply(reply) @@ -1969,6 +1998,9 @@ class MockDevice(Device): self._admin_sessions: set[str] = set() self._remote_tx: dict[str, int] = {} self._default_remote_tx = 20 + # Each simulated repeater's CLI-visible configuration, populated with the + # defaults below on first touch (keyed by full public key, like the TX map). + self._remote_cfg: dict[str, dict[str, str]] = {} # Simulated neighbour tables, keyed by the repeater's key prefix: what each # repeater "hears directly" as ``(neighbour_prefix, snr_db, secs_ago)``. The # ``e5f6a7b8`` entry is deliberately absent from the contact list, so the @@ -2070,6 +2102,58 @@ class MockDevice(Device): self._admin_sessions.add(self._mock_key(node)) return True + #: The simulated repeater CLI's configuration defaults (see send_remote_command). + _REMOTE_CFG_DEFAULTS = { + "freq": "910.525", "bw": "62.5", "sf": "7", "cr": "5", + "lat": "0", "lon": "0", + "repeat": "on", "txdelay": "0", "direct.txdelay": "0", "rxdelay": "0", + "af": "1", "allow.read.only": "off", + "advert.interval": "240", "flood.advert.interval": "12", "flood.max": "64", + } + + async def send_remote_command( # noqa: D102 - inherited docstring + self, node: Contact, command: str, *, timeout: float = 8.0 + ) -> Optional[str]: + await asyncio.sleep(0) + key = self._mock_key(node) + if key not in self._admin_sessions: + return None # firmware ignores strangers — reads as a timeout, like hardware + cfg = self._remote_cfg.setdefault( + key, {"name": node.name, **self._REMOTE_CFG_DEFAULTS} + ) + parts = command.strip().split() + verb = parts[0].lower() if parts else "" + if verb == "ver": + return "MeshCore v1.15.0 (simulator)" + if verb == "clock": + return "OK - clock synced" if parts[1:] == ["sync"] else "12:00 - 1/1/2026 UTC" + if verb == "advert": + return "OK - Advert sent" + if verb in ("reboot", "password", "time", "start"): + return "OK" + if verb == "neighbors": + table = self._neighbour_tables.get(node.key_prefix or "", []) + return "\n".join(f"{p} {snr:+.1f}dB {ago}s" for p, snr, ago in table) or "none" + if verb == "get" and len(parts) == 2: + param = parts[1].lower() + if param == "tx": + return str(self._remote_tx.get(key, self._default_remote_tx)) + if param == "guest.password": + return f"ERR: unknown config: {param}" # write-only, like hardware + value = cfg.get(param) + return f"> {value}" if value is not None else f"ERR: unknown config: {param}" + if verb == "set" and len(parts) >= 3: + param = parts[1].lower() + value = " ".join(parts[2:]) + if param == "tx": + self._remote_tx[key] = int(float(value)) + return "OK" + if param in cfg or param == "guest.password": + cfg[param] = value + return "OK" + return f"ERR: unknown config: {param}" + return f"ERR: unknown command: {command.strip()}" + async def get_remote_tx_power(self, node: Contact) -> Optional[int]: # noqa: D102 key = self._mock_key(node) if key not in self._admin_sessions: diff --git a/meshterm/core/remote_config.py b/meshterm/core/remote_config.py new file mode 100644 index 0000000..074c188 --- /dev/null +++ b/meshterm/core/remote_config.py @@ -0,0 +1,260 @@ +"""The remote-node settings catalog: what a repeater can be asked and told over the mesh. + +Repeaters and room servers are administered through their text CLI — ``get``/``set`` +commands sent as admin messages (see +:meth:`~meshterm.core.connection.Device.send_remote_command`) — not through the binary +companion protocol the local device-configuration editor drives. This module is the +data-driven bridge: one :class:`RemoteSetting` per knob the standard MeshCore repeater +firmware exposes, each carrying its CLI spelling, a one-line explanation, and enough +typing to prompt and validate sensibly. The repeater-admin screen renders the catalog in +the local editor's lanes, stages values, and applies them as ``set`` commands; anything +the catalog doesn't cover is one keystroke away in the raw command line. + +Replies are parsed *loosely* on purpose: repeater firmware answers tersely and +inconsistently across versions (``"tx: 20"``, ``"TX power = 20 dBm"``, ``"> ok"``), so +values are extracted rather than pattern-matched, and the raw reply is always kept for +display. Firmware without a given key simply answers with an error string, which shows +in the value lane instead of breaking the screen. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Optional + +#: Reply text (lowercased) that reads as the firmware refusing/unknowing a command. +_ERRORISH = ("unknown", "error", "err:", "invalid", "denied", "bad ") + + +@dataclass(frozen=True, slots=True) +class RemoteSetting: + """One remote-CLI setting: its spelling, presentation, and typing. + + Attributes: + key: The CLI parameter name as the firmware spells it (e.g. ``"txdelay"``). + label: The display name shown in the editor lane. + help: One-line explanation (no trailing period), shown beside the value. + category: Editor section heading the setting sorts under. + kind: ``"int"``, ``"float"``, ``"str"``, or ``"bool"`` (on/off). + writable: Whether the firmware accepts a ``set`` for this key. + readable: Whether the firmware answers a ``get`` (passwords do not). + minimum: Lower bound for numeric prompts, when known. + maximum: Upper bound for numeric prompts, when known. + unit: Display unit hint (e.g. ``"dBm"``, ``"min"``), purely cosmetic. + """ + + key: str + label: str + help: str + category: str + kind: str = "str" + writable: bool = True + readable: bool = True + minimum: Optional[float] = None + maximum: Optional[float] = None + unit: str = "" + + @property + def get_command(self) -> str: + """The CLI command that reads this setting.""" + return f"get {self.key}" + + def set_command(self, value: str) -> str: + """The CLI command that writes ``value`` to this setting.""" + return f"set {self.key} {value}" + + +#: Every knob the standard MeshCore repeater/room firmware exposes over its CLI, in the +#: editor's display order. TX delay and Direct TX delay — absent from the local +#: companion editor because companions don't repeat — are first-class here. +REPEATER_SETTINGS: tuple[RemoteSetting, ...] = ( + # -- Identity --------------------------------------------------------------- + RemoteSetting( + key="name", label="Name", category="Identity", kind="str", + help="The node's advertised name", + ), + RemoteSetting( + key="lat", label="Latitude", category="Identity", kind="float", + minimum=-90, maximum=90, help="Advertised position, decimal degrees", + ), + RemoteSetting( + key="lon", label="Longitude", category="Identity", kind="float", + minimum=-180, maximum=180, help="Advertised position, decimal degrees", + ), + # -- Radio ------------------------------------------------------------------ + RemoteSetting( + key="freq", label="Frequency", category="Radio", kind="float", unit="MHz", + help="Carrier frequency — every node on the mesh must match", + ), + RemoteSetting( + key="bw", label="Bandwidth", category="Radio", kind="float", unit="kHz", + help="Channel bandwidth — must match the mesh", + ), + RemoteSetting( + key="sf", label="Spreading factor", category="Radio", kind="int", + minimum=7, maximum=12, help="LoRa spreading factor — must match the mesh", + ), + RemoteSetting( + key="cr", label="Coding rate", category="Radio", kind="int", + minimum=5, maximum=8, help="LoRa coding rate denominator — must match the mesh", + ), + RemoteSetting( + key="tx", label="TX power", category="Radio", kind="int", unit="dBm", + minimum=1, maximum=30, + help="Transmit power — the TX optimize tool tunes this by measurement", + ), + # -- Repeating ---------------------------------------------------------------- + RemoteSetting( + key="repeat", label="Repeat", category="Repeating", kind="bool", + help="Whether the node rebroadcasts mesh traffic at all", + ), + RemoteSetting( + key="txdelay", label="TX delay", category="Repeating", kind="int", + minimum=0, help="Delay factor before rebroadcasting a flood packet", + ), + RemoteSetting( + key="direct.txdelay", label="Direct TX delay", category="Repeating", kind="int", + minimum=0, help="Delay factor before forwarding a directed packet", + ), + RemoteSetting( + key="rxdelay", label="RX delay", category="Repeating", kind="int", + minimum=0, help="Base delay before acting on a received packet", + ), + RemoteSetting( + key="af", label="Airtime factor", category="Repeating", kind="float", + minimum=0, help="Duty-cycle multiplier applied to every transmission", + ), + RemoteSetting( + key="allow.read.only", label="Allow read-only", category="Repeating", kind="bool", + help="Whether guests may log in with the guest password", + ), + # -- Adverts ----------------------------------------------------------------- + RemoteSetting( + key="advert.interval", label="Advert interval", category="Adverts", kind="int", + unit="min", minimum=0, + help="Minutes between the node's own zero-hop adverts (0 = off)", + ), + RemoteSetting( + key="flood.advert.interval", label="Flood advert interval", category="Adverts", + kind="int", unit="h", minimum=0, + help="Hours between the node's own flood adverts (0 = off)", + ), + RemoteSetting( + key="flood.max", label="Flood max", category="Adverts", kind="int", minimum=0, + help="Hop cap the node applies when rebroadcasting floods (0 = no cap)", + ), + # -- Access ------------------------------------------------------------------ + RemoteSetting( + key="guest.password", label="Guest password", category="Access", kind="str", + readable=False, help="Read-only login password (write-only over the CLI)", + ), +) + + +#: Commands the remote CLI screen offers as completions, beyond the catalog's +#: ``get``/``set`` spellings: the fixed verbs a MeshCore repeater understands. +KNOWN_COMMANDS: tuple[str, ...] = ( + "advert", + "clock", + "clock sync", + "get ", + "neighbors", + "password ", + "reboot", + "set ", + "start ota", + "time ", + "ver", +) + + +def known_commands() -> list[str]: + """Every completion the remote CLI offers: verbs plus catalog get/set spellings.""" + commands = set(KNOWN_COMMANDS) + for spec in REPEATER_SETTINGS: + if spec.readable: + commands.add(spec.get_command) + if spec.writable: + commands.add(f"set {spec.key} ") + return sorted(commands) + + +def get_setting(key: str) -> Optional[RemoteSetting]: + """Look up a catalog setting by its CLI key.""" + return next((s for s in REPEATER_SETTINGS if s.key == key), None) + + +def settings_by_category() -> list[tuple[str, list[RemoteSetting]]]: + """The catalog grouped by category, in declaration order (the editor's layout).""" + grouped: dict[str, list[RemoteSetting]] = {} + for spec in REPEATER_SETTINGS: + grouped.setdefault(spec.category, []).append(spec) + return list(grouped.items()) + + +def reply_is_error(reply: str) -> bool: + """Whether a reply text reads as the firmware refusing the command.""" + lowered = reply.lower() + return any(marker in lowered for marker in _ERRORISH) + + +def parse_reply_value(spec: RemoteSetting, reply: Optional[str]) -> Optional[str]: + """Extract a display value from a ``get`` reply, or ``None`` when unusable. + + Numeric settings pull the first number out of the terse, version-varying reply + text (the ``get tx`` convention); strings strip a leading ``key:``/``>`` echo. + Error-ish replies parse as ``None`` so a firmware without the key shows unknown + rather than adopting the error string as a value. + + Args: + spec: The setting the reply answers. + reply: The raw reply text, or ``None`` if the node never answered. + + Returns: + The value as display text, or ``None``. + """ + if not reply or reply_is_error(reply): + return None + text = reply.strip() + if spec.kind in ("int", "float"): + match = re.search(r"-?\d+(?:\.\d+)?", text) + if match is None: + return None + raw = match.group() + if spec.kind == "int": + return str(int(float(raw))) + return f"{float(raw):g}" + if spec.kind == "bool": + lowered = text.lower() + if any(t in lowered for t in ("on", "true", "yes", "1")): + return "on" + if any(t in lowered for t in ("off", "false", "no", "0")): + return "off" + return None + # Strings: drop a "key:" or "-> " echo prefix if the firmware included one. + text = re.sub(rf"^\s*(?:>|->|{re.escape(spec.key)}\s*[:=])\s*", "", text, flags=re.I) + return text or None + + +def validate_value(spec: RemoteSetting, raw: str) -> "bool | str": + """Validate a prompted value for ``spec``: ``True``, or the error message to show.""" + text = raw.strip() + if not text: + return "Enter a value." + if spec.kind == "bool": + if text.lower() in ("on", "off"): + return True + return "Enter on or off." + if spec.kind in ("int", "float"): + try: + value = float(text) + except ValueError: + return "Enter a number." + if spec.kind == "int" and not float(text).is_integer(): + return "Enter a whole number." + if spec.minimum is not None and value < spec.minimum: + return f"Must be ≥ {spec.minimum:g}." + if spec.maximum is not None and value > spec.maximum: + return f"Must be ≤ {spec.maximum:g}." + return True diff --git a/meshterm/core/remote_store.py b/meshterm/core/remote_store.py new file mode 100644 index 0000000..337925a --- /dev/null +++ b/meshterm/core/remote_store.py @@ -0,0 +1,116 @@ +"""Persistence for remote-node admin state: last-read settings and CLI history. + +Reading a repeater's configuration costs one paced mesh round trip per value, so the +repeater-admin screen never bulk-reads on open — it shows what the *last* read (or the +last applied ``set``) said, stamped with when, and refreshes on demand. That cache lives +here, per node (keyed like the admin passwords, by public key), in a small JSON file in +the config directory (``/remote.json``) alongside each node's remote +command-line history — global machine state, like the other JSON stores, deliberately +outside the per-invocation SQLite database. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Optional + +from .admin_store import admin_key +from .models import Contact, utcnow + +#: How many command-line entries are kept per node (newest last). +HISTORY_CAP = 100 + + +@dataclass(frozen=True, slots=True) +class CachedValue: + """One remembered remote setting: what the node last said, and when. + + Attributes: + value: The value as display text. + read_at: When it was read from (or written to) the node. + """ + + value: str + read_at: Optional[datetime] + + +class RemoteStore: + """Reads and writes per-node remote-admin state (setting cache + CLI history).""" + + def __init__(self, path: Path) -> None: + """Open the store against a JSON file location. + + Args: + path: Path to the JSON state file (created lazily on first write). + """ + self._path = path + + def _load_all(self) -> dict[str, dict]: + """Return the raw key -> record mapping, or empty on missing/corrupt file.""" + try: + data = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + # -- the settings cache ------------------------------------------------------- + + def settings(self, node: Contact) -> dict[str, CachedValue]: + """Every remembered setting for ``node``, keyed by CLI parameter name.""" + record = self._load_all().get(admin_key(node)) + raw = record.get("settings") if isinstance(record, dict) else None + out: dict[str, CachedValue] = {} + if not isinstance(raw, dict): + return out + for key, entry in raw.items(): + if not isinstance(entry, dict) or "value" not in entry: + continue + read_at: Optional[datetime] = None + stamp = entry.get("read_at") + if isinstance(stamp, str): + try: + read_at = datetime.fromisoformat(stamp) + except ValueError: + read_at = None + out[key] = CachedValue(value=str(entry["value"]), read_at=read_at) + return out + + def remember_setting(self, node: Contact, key: str, value: str) -> None: + """Cache one setting's value for ``node``, stamped now.""" + records = self._load_all() + record = records.setdefault(admin_key(node), {}) + settings = record.setdefault("settings", {}) + settings[key] = {"value": value, "read_at": utcnow().isoformat()} + self._write(records) + + # -- the command-line history --------------------------------------------------- + + def history(self, node: Contact) -> list[str]: + """The node's remote CLI history, oldest first.""" + record = self._load_all().get(admin_key(node)) + raw = record.get("history") if isinstance(record, dict) else None + return [str(c) for c in raw] if isinstance(raw, list) else [] + + def append_history(self, node: Contact, command: str) -> None: + """Append one sent command to the node's history (dropping an adjacent dupe).""" + command = command.strip() + if not command: + return + records = self._load_all() + record = records.setdefault(admin_key(node), {}) + history = record.setdefault("history", []) + if history and history[-1] == command: + return # re-running the last command shouldn't stutter the recall + history.append(command) + del history[:-HISTORY_CAP] + self._write(records) + + def _write(self, records: dict[str, dict]) -> None: + """Persist ``records`` atomically (crash mid-write keeps the previous file).""" + self._path.parent.mkdir(parents=True, exist_ok=True) + tmp = self._path.with_suffix(self._path.suffix + ".tmp") + tmp.write_text(json.dumps(records, indent=2), encoding="utf-8") + tmp.replace(self._path) diff --git a/meshterm/tools/repeater_admin.py b/meshterm/tools/repeater_admin.py new file mode 100644 index 0000000..bebbf50 --- /dev/null +++ b/meshterm/tools/repeater_admin.py @@ -0,0 +1,116 @@ +"""The ``repeater-admin`` tool: set up remote repeaters and room servers over the mesh. + +Interactively it opens the repeater-admin flow (see :mod:`meshterm.ui.repeater_admin`): +pick a node (credentialed ones lead the list), log in with a remembered or prompted +password, and land in a device-configuration-style editor speaking the node's text CLI — +including the repeater-only knobs the local editor never had (TX delay, Direct TX delay, +airtime factor, advert intervals) — plus one-shot actions (advert, clock sync, password, +reboot) and the readline-style remote command line. + +On the CLI it stays a scriptable one-shot: ``repeater-admin `` logs in +(remembered password, or ``--password``) and prints the node's reply — one transmission +per invocation, the trace tool's rule. +""" + +from __future__ import annotations + +from typing import Any, Optional + +import typer + +from ..context import AppContext +from ..core.connection import DeviceCommandError +from .base import Tool, ToolResult, register + + +@register +class RepeaterAdminTool(Tool): + """Configure a remote repeater/room server: settings, actions, and its raw CLI.""" + + name = "repeater-admin" + title = "Repeater admin" + icon = "🗼" + help = "Set up a remote repeater — settings, actions, command line" + category = "Device management" + order = 12 # after Device info: the remote sibling of the local config tools + + async def prompt_params(self, ctx: AppContext) -> Optional[dict[str, Any]]: + """Run the interactive flow; there are never parameters to collect. + + The flow presents everything itself (the same pattern as ``device-actions``), + so returning ``None`` tells the menu the invocation is complete. + + Args: + ctx: Shared application context. + + Returns: + Always ``None``. + """ + from ..ui.repeater_admin import open_repeater_admin + + await open_repeater_admin(ctx) + return None + + async def run(self, ctx: AppContext, params: dict[str, Any]) -> ToolResult: + """Send one CLI command to a remote node and print its reply (scripted path). + + Args: + ctx: Shared application context. + params: ``node`` (contact name), ``command`` (the CLI text), and an + optional ``password`` overriding the remembered one. + + Returns: + A :class:`ToolResult` carrying the node's reply (or its absence). + """ + device = await ctx.device() + contacts = await device.get_contacts() + name = str(params["node"]) + node = next((c for c in contacts if c.name == name), None) + if node is None: + raise DeviceCommandError(f"unknown contact: {name!r}") + + password = params.get("password") or ctx.admin_store.get(node) + if not password: + raise DeviceCommandError( + f"no admin password for {name!r}; pass --password or run the " + "interactive flow once to store it." + ) + if not await device.admin_login(node, str(password)): + ctx.admin_store.forget(node) + raise DeviceCommandError( + f"admin login to {name!r} failed (wrong password?). " + "The saved password was cleared." + ) + ctx.admin_store.remember(node, str(password)) + + command = str(params["command"]) + ctx.remote_store.append_history(node, command) + reply = await device.send_remote_command(node, command, timeout=10.0) + if reply is None: + ctx.ui.note("[warn]no reply — the command may still have landed[/warn]") + else: + ctx.ui.note(reply) + return ToolResult( + summary={"node": name, "command": command, "replied": reply is not None} + ) + + def register_cli(self, app: typer.Typer) -> None: + """Register the ``repeater-admin`` subcommand. + + Args: + app: The Typer application. + """ + from ..cli import run_tool_command + + @app.command(name=self.name, help=self.help) + def _repeater_admin( + node: str = typer.Argument(..., help="The remote contact's name"), + command: list[str] = typer.Argument(..., help="The CLI command to send"), + password: Optional[str] = typer.Option( + None, "--password", help="Admin password (else remembered)" + ), + ) -> None: + tool_params: dict[str, Any] = {"node": node, "command": " ".join(command)} + if password is not None: + tool_params["password"] = password + run_tool_command(self, tool_params) diff --git a/meshterm/tools/tx_optimize.py b/meshterm/tools/tx_optimize.py index 40f8058..b9820a7 100644 --- a/meshterm/tools/tx_optimize.py +++ b/meshterm/tools/tx_optimize.py @@ -23,7 +23,7 @@ import typer from ..context import AppContext from ..core.connection import DeviceCommandError -from ..core.models import NODE_TYPE_REPEATER, NODE_TYPE_ROOM, Contact +from ..core.models import Contact from ..services import trace_runner, tx_optimizer from ..ui.widgets import tx_opt_summary, tx_opt_table from ..viz.tx_plot import render_tx_optimization @@ -57,9 +57,16 @@ class TxOptimizeTool(Tool): Returns: ``{"live": True, "admin": name, "target": name}``, or ``None`` on cancel. """ + from ..ui.admin_picker import pick_admin_node + device = await ctx.device() contacts = await device.get_contacts() - admin = await _pick_admin(ctx, contacts) + admin = await pick_admin_node( + ctx, + contacts, + title="TX optimize — node to tune", + prompt="Whose transmit power gets tuned (you need its admin password):", + ) if admin is None: return None target = await _pick_target(ctx, contacts, admin) @@ -395,72 +402,6 @@ def _contact_for_hash(hash_hex: str, contacts: list[Contact]) -> Optional[Contac return None -async def _pick_admin(ctx: AppContext, contacts: list[Contact]) -> Optional[Contact]: - """Pick the node to tune: credentialed repeaters lead, the rest follow by recency. - - Repeaters (and room servers) with a remembered admin password sit in their own - section wearing a key glyph — those are the nodes previously tuned or administered, - the likeliest picks. Any contact with a public key can be chosen, since holding a - password is a fact about the *user*, not the node. - - Args: - ctx: Shared application context (for the UI surface and the admin store). - contacts: The device's known contacts. - - Returns: - The chosen contact, or ``None`` if cancelled (or there is nothing to pick). - """ - from ..ui.tui import Choice, Separator - from ..ui.widgets import _DEFAULT_GLYPH, _NODE_GLYPHS - - candidates = [c for c in contacts if (c.public_key or c.key_prefix).strip()] - if not candidates: - ctx.ui.note("[err]no contacts with a key — receive an advert first[/err]") - await ctx.ui.present(title="TX optimize") - return None - - def row(contact: Contact) -> Any: - glyph, style = _NODE_GLYPHS.get(contact.node_type, _DEFAULT_GLYPH) - from rich.text import Text - - label = Text(glyph, style=style) - label.append(f" {contact.name}") - return Choice(title=label, value=contact.name) - - def recency(contact: Contact) -> float: - return -(contact.last_seen.timestamp() if contact.last_seen else 0.0) - - remembered = [c for c in candidates if ctx.admin_store.get(c) is not None] - infrastructure = [ - c for c in candidates - if c not in remembered and c.node_type in (NODE_TYPE_REPEATER, NODE_TYPE_ROOM) - ] - others = [c for c in candidates if c not in remembered and c not in infrastructure] - - items: list = [] - if remembered: - items.append(Separator("── 🔑 Remembered admins ──", style="accent")) - items.extend(row(c) for c in sorted(remembered, key=recency)) - if infrastructure: - items.append(Separator("── Repeaters & rooms ──", style="accent")) - items.extend(row(c) for c in sorted(infrastructure, key=recency)) - if others: - items.append(Separator("── Other contacts ──", style="accent")) - items.extend(row(c) for c in sorted(others, key=recency)) - items.append(Separator(" ")) - items.append(Choice(title="Back", value=None)) - - choice = await ctx.ui.select( - "TX optimize — node to tune", - items, - prompt="Whose transmit power gets tuned (you need its admin password):", - wrap=False, - ) - if choice is None: - return None - return next((c for c in candidates if c.name == choice), None) - - async def _pick_target( ctx: AppContext, contacts: list[Contact], admin: Contact ) -> Optional[str]: diff --git a/meshterm/ui/admin_picker.py b/meshterm/ui/admin_picker.py new file mode 100644 index 0000000..064e19e --- /dev/null +++ b/meshterm/ui/admin_picker.py @@ -0,0 +1,82 @@ +"""The admin-node picker: choose a remote node you hold (or will enter) credentials for. + +Shared by the tools that administer remote nodes over the mesh — TX optimize and the +repeater admin screen — so "pick the node to manage" reads identically everywhere. +Credentialed nodes lead the list in their own 🔑 section (those are the nodes previously +administered, the likeliest picks), repeaters and room servers follow, then everything +else; each section orders by how recently the node was heard. Any contact with a public +key is offerable, since holding a password is a fact about the *user*, not the node. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Optional + +from rich.text import Text + +from ..core.models import NODE_TYPE_REPEATER, NODE_TYPE_ROOM, Contact + +if TYPE_CHECKING: + from ..context import AppContext + + +async def pick_admin_node( + ctx: "AppContext", + contacts: list[Contact], + *, + title: str, + prompt: str, +) -> Optional[Contact]: + """Pick a remote node to administer, credentialed and infrastructure nodes first. + + Args: + ctx: Shared application context (for the UI surface and the admin store). + contacts: The device's known contacts. + title: The select screen's heading (names the calling feature). + prompt: One line above the list saying what the pick is for. + + Returns: + The chosen contact, or ``None`` if cancelled (or there is nothing to pick). + """ + from .tui import Choice, Separator + from .widgets import _DEFAULT_GLYPH, _NODE_GLYPHS + + candidates = [c for c in contacts if (c.public_key or c.key_prefix).strip()] + if not candidates: + ctx.ui.note("[err]no contacts with a key — receive an advert first[/err]") + await ctx.ui.present(title=title) + return None + + def row(contact: Contact) -> Choice: + glyph, style = _NODE_GLYPHS.get(contact.node_type, _DEFAULT_GLYPH) + label = Text(glyph, style=style) + label.append(f" {contact.name}") + return Choice(title=label, value=contact.name) + + def recency(contact: Contact) -> float: + return -(contact.last_seen.timestamp() if contact.last_seen else 0.0) + + remembered = [c for c in candidates if ctx.admin_store.get(c) is not None] + infrastructure = [ + c for c in candidates + if c not in remembered and c.node_type in (NODE_TYPE_REPEATER, NODE_TYPE_ROOM) + ] + others = [c for c in candidates if c not in remembered and c not in infrastructure] + + items: list = [] + if remembered: + items.append(Separator("── 🔑 Remembered admins ──", style="accent")) + items.extend(row(c) for c in sorted(remembered, key=recency)) + if infrastructure: + items.append(Separator("── Repeaters & rooms ──", style="accent")) + items.extend(row(c) for c in sorted(infrastructure, key=recency)) + if others: + items.append(Separator("── Other contacts ──", style="accent")) + items.extend(row(c) for c in sorted(others, key=recency)) + items.append(Separator(" ")) + items.append(Choice(title="Back", value=None)) + + choice = await ctx.ui.select(title, items, prompt=prompt, wrap=False) + if choice is None: + return None + return next((c for c in candidates if c.name == choice), None) diff --git a/meshterm/ui/remote_cli.py b/meshterm/ui/remote_cli.py new file mode 100644 index 0000000..99163d6 --- /dev/null +++ b/meshterm/ui/remote_cli.py @@ -0,0 +1,203 @@ +"""The remote command line: talk to a repeater's CLI over the mesh, readline-style. + +A full-screen terminal onto a logged-in remote node (part of the repeater-admin +feature): type a command, watch it fly, read the reply in a growing transcript. Every +command is one mesh transmission, so the screen sends exactly what you commit and one +at a time — a command in flight parks the prompt until the reply lands (or times out; +repeaters answer tersely and sometimes not at all). + +The input line behaves like a shell: + +* **↑/↓** recall the node's command history, which persists per node (see + :class:`~meshterm.core.remote_store.RemoteStore`) so last week's incantation is one + keystroke away next session. +* **Tab** completes the current text against the known repeater commands — the fixed + verbs plus every ``get``/``set`` spelling in the settings catalog — and the best + match previews muted, ghost-text style, after the cursor. +* **Enter** sends. The transcript keeps every exchange of the session, newest at the + bottom, and the view follows the prompt as it grows. + +The screen renders state and routes keys; the owning flow injects ``send`` (which owns +the radio, the history store, and the reply plumbing) and feeds replies back through +:meth:`reply` / :meth:`failed`. +""" + +from __future__ import annotations + +from typing import Callable, Optional + +from rich.console import Group +from rich.text import Text + +from ..core.remote_config import known_commands +from .tui.prompt import _LineEditor +from .tui.render import render_lines +from .tui.screen import Screen +from .tui.spinner import Spinner + +#: The transcript's history depth in exchanges (the body scrolls; this caps memory). +_LOG_CAP = 400 + + +class RemoteCliScreen(Screen): + """A shell-like screen onto one remote node's CLI. The owner does the radio work.""" + + floating = False + + def __init__( + self, + *, + node_label: str, + history: list[str], + send: Callable[[str], None], + session: object, + ) -> None: + """Create the command line over a node's persisted history. + + Args: + node_label: The remote node's display name (title and prompt chrome). + history: The node's prior commands, oldest first (recalled with ↑). + send: Commits one command (owner-guarded; no-op while one is flying). + session: The running TUI session (for repaints). + """ + super().__init__() + self.title = f"command line · {node_label}" + self._node_label = node_label + self._send = send + self._session = session + self._editor = _LineEditor() + self._log: list[Text] = [] + self._history = list(history) + self._recall: Optional[int] = None # index into history while ↑/↓ browse it + self._draft = "" # what was typed before recall began, restored on ↓ past the end + self._completions = known_commands() + self.busy = False + self._spinner = Spinner() + self._prompt_line = 0 # body index of the input line (pinned into view) + + # --- owner feed ------------------------------------------------------------------ + + def sent(self, command: str) -> None: + """Echo a just-sent command into the transcript and park the prompt.""" + row = Text("❯ ", style="brand") + row.append(command) + self._append(row) + self._history.append(command) + self._recall = None + self.busy = True + self._spinner.reset() + self._session.invalidate() + + def reply(self, text: str) -> None: + """Land the node's reply (possibly multi-line) and free the prompt.""" + for line in text.splitlines() or [""]: + self._append(Text(f" {line}")) + self.busy = False + self._session.invalidate() + + def failed(self, note: str, *, error: bool = False) -> None: + """Land a timeout/error note and free the prompt.""" + self._append(Text(f" {note}", style="err" if error else "muted")) + self.busy = False + self._session.invalidate() + + def tick(self) -> None: + """Advance the in-flight spinner (driven by the owner's animation timer).""" + self._spinner.tick() + + def _append(self, row: Text) -> None: + row.no_wrap = False + self._log.append(row) + del self._log[:-_LOG_CAP] + + # --- input ----------------------------------------------------------------------- + + @property + def footer_hint(self) -> str: # type: ignore[override] + """The footer keys, tracking whether a command is in flight.""" + if self.busy: + return "waiting for the reply… · PgUp/PgDn scroll · Esc back" + return "Enter send · ↑↓ history · Tab complete · PgUp/PgDn scroll · Esc back" + + def handle(self, action: str, data: str = "") -> None: + """Edit the buffer, recall history, complete, send, scroll, or dismiss.""" + if action == "enter": + command = self._editor.text.strip() + if command and not self.busy: + self._editor = _LineEditor() + self._send(command) + elif action == "up": + self._recall_step(-1) + elif action == "down": + self._recall_step(1) + elif action == "tab": + completion = self._completion() + if completion is not None: + self._editor = _LineEditor(completion) + elif action == "pageup": + self.scroll_pages(-1) + elif action in ("pagedown",): + self.scroll_pages(1) + elif action == "escape": + self.resolve(None) + elif self._editor.edit(action, data): + self._recall = None # typing leaves the history walk; the buffer is a draft + self._session.invalidate() + + def _recall_step(self, direction: int) -> None: + """Walk the history with ↑/↓, keeping the un-sent draft safe at the new end.""" + if not self._history: + return + if self._recall is None: + if direction > 0: + return # nothing newer than the draft + self._draft = self._editor.text + self._recall = len(self._history) - 1 + else: + self._recall += direction + if self._recall >= len(self._history): + self._recall = None + self._editor = _LineEditor(self._draft) + return + self._recall = max(0, self._recall) + self._editor = _LineEditor(self._history[self._recall]) + + def _completion(self) -> Optional[str]: + """The first known command extending the current text, or ``None``.""" + prefix = self._editor.text.lstrip() + if not prefix: + return None + return next((c for c in self._completions if c.startswith(prefix) and c != prefix), + None) + + # --- rendering --------------------------------------------------------------------- + + def render_body(self, width: int) -> list[str]: + """Render the intro, the transcript, and the prompt with its ghost completion.""" + intro = Text( + f"Talking to {self._node_label} over the mesh — every command is one " + "transmission.", + style="muted", + ) + lines = render_lines(Group(intro, Text(), *self._log), width) + self._prompt_line = len(lines) + lines.extend(render_lines(self._prompt(), width)) + self._scroll_total = max(1, len(lines)) + return lines + + def cursor_line(self) -> Optional[int]: + """Pin the prompt into view, so the transcript follows itself as it grows.""" + return self._prompt_line + + def _prompt(self) -> Text: + """The input line: spinner while waiting, else the editor and its ghost text.""" + if self.busy: + line = self._spinner.text() + line.append(" waiting for the reply…", style="muted") + return line + line = Text("❯ ", style="brand") + line.append_text(self._editor.render()) + completion = self._completion() + if completion is not None: + line.append(completion[len(self._editor.text.lstrip()):], style="faint") + return line diff --git a/meshterm/ui/repeater_admin.py b/meshterm/ui/repeater_admin.py new file mode 100644 index 0000000..e231361 --- /dev/null +++ b/meshterm/ui/repeater_admin.py @@ -0,0 +1,650 @@ +"""The repeater-admin screens: configure a remote node over the mesh, editor-style. + +The interactive face of the ``repeater-admin`` tool. Picking a node (the shared +credentialed-first picker) and logging in (remembered passwords, the one-time prompt +otherwise) opens an editor deliberately shaped like the local device-configuration +screen — the same setting / value / description lanes, staged ``current → new`` values, +Apply — but speaking the repeater's text CLI (see :mod:`meshterm.core.remote_config`) +instead of the companion's binary protocol, which brings the repeater-only knobs the +local editor never had: TX delay, Direct TX delay, airtime factor, advert intervals. + +Because every read is one paced mesh round trip, the editor never bulk-reads on open: +values show what the last read (or the last applied set) said, stamped with age, from +the per-node cache (:class:`~meshterm.core.remote_store.RemoteStore`); the *Read +settings* action refreshes them all under an abortable progress dialog, one paced +``get`` at a time. Apply sends the staged ``set`` commands the same way and folds each +confirmed value straight back into the cache. + +Beyond the settings, the action rows cover the box itself — advert, clock sync, change +admin password, reboot, each behind its own floating confirmation — and the **Command +line** opens the readline-style remote CLI (:mod:`meshterm.ui.remote_cli`) for anything +the catalog doesn't spell. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING, Any, Optional + +from rich.cells import cell_len +from rich.text import Text + +from ..core.models import Contact +from ..core.remote_config import ( + RemoteSetting, + parse_reply_value, + reply_is_error, + settings_by_category, +) +from .trace_screen import TracingDialog +from .tui import Choice, Separator +from .tui.spinner import Spinner +from .widgets import _age_seconds, _format_age + +if TYPE_CHECKING: + from ..context import AppContext + from ..core.connection import Device + +#: Seconds between spinner frames while a remote command is in flight. +_SPINNER_INTERVAL = 0.12 + +#: Seconds to wait for one remote reply. Repeaters answer over the mesh — multi-hop +#: routes take seconds — so this is deliberately more patient than a local command. +_REPLY_TIMEOUT_S = 10.0 + +# Menu action sentinels (distinct from setting keys, which are CLI parameter names). +_CLI = "__cli__" +_READ = "__read__" +_ADVERT = "__advert__" +_CLOCK = "__clock__" +_PASSWORD = "__password__" +_REBOOT = "__reboot__" +_APPLY = "__apply__" +_CANCEL = "__cancel__" + + +async def open_repeater_admin(ctx: "AppContext") -> Optional[dict[str, Any]]: + """Run the repeater-admin flow: pick a node, log in, and administer it. + + Args: + ctx: The shared application context (must be running the interactive TUI surface). + + Returns: + A summary of what happened (for the tool's log), or ``None`` on cancel. + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + from .admin_picker import pick_admin_node + from .surface import TuiUi + + if not isinstance(ctx.ui, TuiUi): # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("repeater admin is only available in the menu") + session = ctx.ui.session + + device = await ctx.device() + contacts = await device.get_contacts() + node = await pick_admin_node( + ctx, + contacts, + title="Repeater admin — node to manage", + prompt="The remote node to set up (you need its admin password):", + ) + if node is None: + return None + if not await _login(ctx, device, node): + return None + return await _admin_session(ctx, device, node) + + +async def _login(ctx: "AppContext", device: "Device", node: Contact) -> bool: + """Log in to ``node``: remembered password silently, else one floating prompt. + + A working password is remembered; a rejected one is forgotten so the next attempt + asks fresh (the app-wide remote-admin convention). + """ + session = ctx.ui.session + password = ctx.admin_store.get(node) + prompted = password is None + if password is None: + password = await session.text( + f"Admin password for {node.name}", + prompt="The node ignores admin commands without a login.", + password=True, + ) + if not password: + return False + async with ctx.ui.busy_overlay(): + accepted = await device.admin_login(node, password) + if not accepted: + ctx.admin_store.forget(node) + await session.message_dialog( + Text( + f"{node.name!r} rejected the admin login (wrong password?). " + + ("" if prompted else "The saved password was cleared; ") + + "try again to enter a new one.", + style="err", + ), + title="admin login", + ) + return False + ctx.admin_store.remember(node, password) + return True + + +async def _admin_session( + ctx: "AppContext", device: "Device", node: Contact +) -> dict[str, Any]: + """Run the editor loop for one logged-in node (the persistent-backdrop pattern).""" + from .tui import CANCEL, SelectScreen + + session = ctx.ui.session + loop = asyncio.get_running_loop() + + pending: dict[str, str] = {} # setting key -> staged new value + applied = 0 + cursor: Any = None # the row to re-highlight, so the menu reopens where you left it + + while True: + cache = ctx.remote_store.settings(node) + title, items = _menu_items(node, cache, pending) + menu = SelectScreen( + title, items, default=cursor, wrap=False, + footer_hint="↑↓ move · type to filter · Enter select · Esc back", + ) + menu.future = loop.create_future() + session.push(menu) + try: + choice = await menu.future + if choice is CANCEL: + choice = _CANCEL + if choice not in (None, _CANCEL): + cursor = choice + + if choice in (None, _CANCEL): + if pending and not await _confirm_discard(ctx, len(pending)): + continue + return {"node": node.name, "applied": applied} + if choice == _APPLY: + applied += await _apply(ctx, device, node, pending) + elif choice == _READ: + await _read_all(ctx, device, node) + elif choice == _CLI: + await _command_line(ctx, device, node) + elif choice == _ADVERT: + await _simple_action( + ctx, device, node, "advert", + title="📡 Send advert", + prompt=f"Ask {node.name} to announce itself (flood) now?", + commit="Send", + ) + elif choice == _CLOCK: + await _simple_action( + ctx, device, node, "clock sync", + title="🕒 Sync clock", + prompt=f"Set {node.name}'s clock from this companion's time?", + commit="Sync", + ) + elif choice == _PASSWORD: + await _change_password(ctx, device, node) + elif choice == _REBOOT: + await _simple_action( + ctx, device, node, "reboot", + title="🔄 Reboot node", + prompt=f"Reboot {node.name} now? It drops off the mesh while booting.", + commit="Reboot", + danger=True, + ) + else: # a setting key + await _stage_setting(ctx, str(choice), cache, pending) + finally: + session.pop(menu) + + +# --- the menu ------------------------------------------------------------------------ + + +def _menu_items( + node: Contact, cache: dict, pending: dict[str, str] +) -> tuple[str, list]: + """Build the editor menu's title and rows for the cache + staged state. + + The same lane layout as the local device-configuration editor — setting, value + (with any staged ``→ new``), description under one header line — so administering + a remote node reads exactly like configuring the local one. + """ + sections: list[tuple[str, list[tuple[str, Text, str, Any]]]] = [] + for category, specs in settings_by_category(): + rows: list[tuple[str, Text, str, Any]] = [] + for spec in specs: + rows.append((spec.label, _value_text(spec, cache, pending), spec.help, spec.key)) + sections.append((category, rows)) + + label_w = max(cell_len(label) for _, rows in sections for label, _, _, _ in rows) + value_w = max(cell_len(value.plain) for _, rows in sections for _, value, _, _ in rows) + + items: list = [ + Separator( + " " + "SETTING".ljust(label_w + 2) + "VALUE".ljust(value_w + 2) + "DESCRIPTION" + ) + ] + for category, rows in sections: + items.append(Separator(f"── {category} ──", style="accent")) + for label, value, help_text, key in rows: + items.append( + Choice(title=_lane_row(label, value, help_text, label_w, value_w), value=key) + ) + + items.append(Separator("── Actions ──", style="accent")) + actions: list[tuple[str, str, str]] = [ + ("↻ Read settings", "Fetch every value from the node, one paced get", _READ), + ("⌨ Command line…", "Talk to the node's CLI directly", _CLI), + ("📡 Send advert…", "Have the node announce itself now", _ADVERT), + ("🕒 Sync clock…", "Set the node's clock over the mesh", _CLOCK), + ("🔐 Admin password…", "Change the node's admin password", _PASSWORD), + ("🔄 Reboot node…", "Restart it remotely", _REBOOT), + ] + action_w = max(cell_len(label) for label, _, _ in actions) + for label, help_text, value in actions: + row = Text(label) + row.append(" " * (action_w - cell_len(label) + 2)) + row.append(help_text, style="muted") + items.append(Choice(title=row, value=value)) + + items.append(Separator(" ")) + staged = len(pending) + if staged: + items.append( + Choice( + title=Text.assemble(("✓ ", "ok"), f"Apply {_changes(staged)}"), + value=_APPLY, + ) + ) + items.append( + Choice( + title=Text.assemble(("✗ ", "err"), "Back — discard staged changes"), + value=_CANCEL, + ) + ) + else: + items.append(Choice(title="Back", value=_CANCEL)) + + title = f"Repeater admin — {node.name}" + (f" · {staged} staged" if staged else "") + return title, items + + +def _value_text(spec: RemoteSetting, cache: dict, pending: dict[str, str]) -> Text: + """One setting's VALUE lane: cached value (with its age), and any staged arrow.""" + cached = cache.get(spec.key) + if not spec.readable: + value = Text("write-only", style="faint") + elif cached is None: + value = Text("?", style="muted") + else: + value = Text(cached.value) + age = _format_age(_age_seconds(cached.read_at)) + if age != "now": + value.append(f" ·{age}", style="faint") + if spec.key in pending: + value.append(f" → {pending[spec.key]}", style="warn") + return value + + +def _lane_row(label: str, value: Text, help_text: str, label_w: int, value_w: int) -> Text: + """Lay one row out in the SETTING / VALUE / DESCRIPTION lanes (cell-padded).""" + row = Text(label) + row.append(" " * (label_w - cell_len(label) + 2)) + row.append_text(value) + row.append(" " * (value_w - cell_len(value.plain) + 2)) + row.append(help_text, style="muted") + return row + + +def _changes(count: int) -> str: + """``"1 staged change"`` / ``"3 staged changes"`` for dialogs and menu rows.""" + return f"{count} staged change{'' if count == 1 else 's'}" + + +async def _confirm_discard(ctx: "AppContext", staged: int) -> bool: + """Ask before dropping staged changes on the way out; ``True`` means discard.""" + choice = await ctx.ui.dialog( + f"Discard {_changes(staged)} without sending them?", + [("Keep editing", "keep"), ("Discard", "discard")], + title="Unsaved changes", + default=1, + danger=True, + ) + return choice == "discard" + + +# --- staging and applying --------------------------------------------------------- + + +async def _stage_setting( + ctx: "AppContext", key: str, cache: dict, pending: dict[str, str] +) -> None: + """Prompt for one setting's new value and stage it (nothing is sent yet).""" + from ..core.remote_config import get_setting, validate_value + + spec = get_setting(key) + if spec is None or not spec.writable: # pragma: no cover - menu offers only real keys + return + cached = cache.get(key) + current = pending.get(key, cached.value if cached is not None else "") + + if spec.kind == "bool": + picked = await ctx.ui.dialog( + spec.help, + [("Off", "off"), ("On", "on")], + title=spec.label, + default=1 if current == "on" else 0, + keys={"0": "off", "1": "on", "n": "off", "y": "on"}, + ) + if picked is None: + return + value = str(picked) + else: + hint = "" + if spec.minimum is not None and spec.maximum is not None: + hint = f"Allowed: {spec.minimum:g} – {spec.maximum:g}" + elif spec.minimum is not None: + hint = f"Allowed: ≥ {spec.minimum:g}" + if spec.unit: + hint = f"{hint} ({spec.unit})" if hint else f"In {spec.unit}." + raw = await ctx.ui.text( + spec.label, + prompt=spec.help, + default=str(current), + validate=lambda t: validate_value(spec, t), + help_text=hint, + ) + if raw is None: + return + value = raw.strip() + + if cached is not None and value == cached.value: + pending.pop(key, None) # back to what the node last said — nothing to send + else: + pending[key] = value + + +async def _apply( + ctx: "AppContext", device: "Device", node: Contact, pending: dict[str, str] +) -> int: + """Send every staged ``set`` command, paced, under an abortable progress dialog. + + Each confirmed value folds straight into the per-node cache; a rejected or + unanswered set stays staged so it can be retried (or unstaged) rather than being + silently dropped. Records one ``runs`` row for the batch. + + Returns: + How many settings the node accepted. + """ + from ..core.remote_config import get_setting + + session = ctx.ui.session + run_id = ctx.repo.start_run( + "repeater-admin", + {"node": node.name, "mode": "apply", "settings": sorted(pending)}, + ctx.profile_name, + ) + outcomes: list[Text] = [] + applied = 0 + + async def work(dialog: TracingDialog) -> None: + nonlocal applied + total = len(pending) + for i, (key, value) in enumerate(sorted(pending.items()), start=1): + spec = get_setting(key) + assert spec is not None + dialog.status = f"set {key} · {i}/{total}" + session.invalidate() + reply = await device.send_remote_command( + node, spec.set_command(value), timeout=_REPLY_TIMEOUT_S + ) + if reply is not None and not reply_is_error(reply): + ctx.remote_store.remember_setting(node, key, value) + pending.pop(key, None) + applied += 1 + outcomes.append(Text.assemble(("✓ ", "ok"), f"{spec.label} = {value}")) + elif reply is None: + outcomes.append(Text.assemble( + ("? ", "warn"), f"{spec.label} — no reply (still staged)" + )) + else: + outcomes.append(Text.assemble( + ("✗ ", "err"), f"{spec.label} — {reply.strip()} (still staged)" + )) + if i < total: + await asyncio.sleep(ctx.settings.trace_cooldown_s) + + aborted = await _run_under_dialog(ctx, f"applying · {node.name}", work) + ctx.repo.finish_run( + run_id, "error" if aborted else "ok", + {"applied": applied, "staged_left": len(pending)}, + ) + summary = Text.assemble( + (f"{applied}", "brand"), f" of {applied + len(pending)} settings applied" + ) + if aborted: + summary.append(" (aborted — the rest stay staged)", style="warn") + await ctx.ui.session.message_dialog( + Text("\n").join([summary, Text(), *outcomes]) if outcomes else summary, + title=f"apply · {node.name}", + ) + return applied + + +async def _read_all(ctx: "AppContext", device: "Device", node: Contact) -> None: + """Refresh every readable setting from the node, one paced ``get`` at a time. + + Values that parse land in the cache (and on screen); firmware that doesn't know a + key just leaves that row unknown. Abort keeps everything already read. + """ + session = ctx.ui.session + specs = [s for cat, ss in settings_by_category() for s in ss if s.readable] + run_id = ctx.repo.start_run( + "repeater-admin", {"node": node.name, "mode": "read"}, ctx.profile_name + ) + read = 0 + + async def work(dialog: TracingDialog) -> None: + nonlocal read + for i, spec in enumerate(specs, start=1): + dialog.status = f"get {spec.key} · {i}/{len(specs)}" + session.invalidate() + reply = await device.send_remote_command( + node, spec.get_command, timeout=_REPLY_TIMEOUT_S + ) + value = parse_reply_value(spec, reply) + if value is not None: + ctx.remote_store.remember_setting(node, spec.key, value) + read += 1 + if i < len(specs): + await asyncio.sleep(ctx.settings.trace_cooldown_s) + + aborted = await _run_under_dialog(ctx, f"reading · {node.name}", work) + ctx.repo.finish_run( + run_id, "error" if aborted else "ok", {"read": read, "asked": len(specs)} + ) + + +async def _run_under_dialog(ctx: "AppContext", title: str, work) -> bool: + """Run an async remote batch under a floating spinner dialog with Abort. + + Args: + ctx: The shared application context. + title: The dialog's heading. + work: ``async work(dialog)`` performing the batch, updating ``dialog.status``. + + Returns: + ``True`` if the user aborted, ``False`` if the batch ran to completion. + """ + session = ctx.ui.session + spinner = Spinner() + dialog = TracingDialog(title, spinner=spinner, on_abort=lambda: None) + dialog.show_last = False + + task = asyncio.ensure_future(work(dialog)) + dialog.on_abort = task.cancel + + async def animate() -> None: + while True: + await asyncio.sleep(_SPINNER_INTERVAL) + spinner.tick() + session.invalidate() + + ticker = asyncio.ensure_future(animate()) + session.push(dialog) + aborted = False + try: + await task + except asyncio.CancelledError: + aborted = True + except Exception as exc: # noqa: BLE001 - surface in a dialog, keep the screen + await session.message_dialog(Text(str(exc), style="err"), title=title) + finally: + ticker.cancel() + try: + await ticker + except asyncio.CancelledError: + pass + except Exception: # noqa: BLE001 - a spinner hiccup must never break the batch + pass + session.pop(dialog) + return aborted + + +# --- one-shot actions ------------------------------------------------------------ + + +async def _simple_action( + ctx: "AppContext", + device: "Device", + node: Contact, + command: str, + *, + title: str, + prompt: str, + commit: str, + danger: bool = False, +) -> None: + """Confirm and send one fixed CLI command, showing the node's reply.""" + choice = await ctx.ui.dialog( + prompt, + [("Cancel", None), (commit, "go")], + title=title, + default=1, + danger=danger, + ) + if choice != "go": + return + async with ctx.ui.busy_overlay(): + reply = await device.send_remote_command(node, command, timeout=_REPLY_TIMEOUT_S) + if reply is None: + body = Text("no reply — the command may still have landed", style="warn") + else: + body = Text(reply.strip(), style="err" if reply_is_error(reply) else "ok") + await ctx.ui.session.message_dialog(body, title=title) + + +async def _change_password(ctx: "AppContext", device: "Device", node: Contact) -> None: + """Change the node's admin password (and re-remember it on success).""" + new = await ctx.ui.text( + f"New admin password for {node.name}", + prompt="Sent over the mesh; the node applies it immediately.", + password=True, + ) + if not new: + return + confirmed = await ctx.ui.dialog( + f"Change {node.name}'s admin password now? You will need the new one everywhere.", + [("Cancel", None), ("Change", "go")], + title="🔐 Admin password", + default=1, + danger=True, + ) + if confirmed != "go": + return + async with ctx.ui.busy_overlay(): + reply = await device.send_remote_command( + node, f"password {new}", timeout=_REPLY_TIMEOUT_S + ) + if reply is not None and not reply_is_error(reply): + ctx.admin_store.remember(node, new) # the working password just changed + body = Text("✓ password changed and remembered", style="ok") + elif reply is None: + body = Text( + "no reply — the change may or may not have landed; the old password " + "stays remembered until a login proves otherwise", + style="warn", + ) + else: + body = Text(reply.strip(), style="err") + await ctx.ui.session.message_dialog(body, title="admin password") + + +# --- the command line --------------------------------------------------------------- + + +async def _command_line(ctx: "AppContext", device: "Device", node: Contact) -> None: + """Open the readline-style remote CLI for ``node`` (history persists per node).""" + from .remote_cli import RemoteCliScreen + + session = ctx.ui.session + worker: Optional[asyncio.Task] = None + ticker: Optional[asyncio.Task] = None + + def send(command: str) -> None: + nonlocal worker + if worker is not None and not worker.done(): + return # one command in flight at a time — every send is a transmission + screen.sent(command) + ctx.remote_store.append_history(node, command) + worker = asyncio.ensure_future(roundtrip(command)) + + async def roundtrip(command: str) -> None: + try: + reply = await device.send_remote_command( + node, command, timeout=_REPLY_TIMEOUT_S + ) + except asyncio.CancelledError: + screen.failed("cancelled", error=False) + raise + except Exception as exc: # noqa: BLE001 - shown inline, the screen stays up + screen.failed(str(exc), error=True) + return + if reply is None: + screen.failed(f"no reply within {_REPLY_TIMEOUT_S:.0f} s") + else: + screen.reply(reply) + + screen = RemoteCliScreen( + node_label=node.name, + history=ctx.remote_store.history(node), + send=send, + session=session, + ) + + async def animate() -> None: + while True: + await asyncio.sleep(_SPINNER_INTERVAL) + if screen.busy: + screen.tick() + session.invalidate() + + ticker = asyncio.ensure_future(animate()) + try: + await session.run_screen(screen) + finally: + for task in (worker, ticker): + if task is not None and not task.done(): + task.cancel() + for task in (worker, ticker): + if task is not None: + try: + await task + except asyncio.CancelledError: + pass + except Exception: # noqa: BLE001 - the screen is closed; nothing to surface + pass diff --git a/tests/test_repeater_admin.py b/tests/test_repeater_admin.py new file mode 100644 index 0000000..2dd2a1b --- /dev/null +++ b/tests/test_repeater_admin.py @@ -0,0 +1,206 @@ +"""Repeater-admin tests: the settings catalog, the per-node store, the simulated +remote CLI, and the command-line screen's readline behavior.""" + +from __future__ import annotations + +from pathlib import Path + +from meshterm.core.connection import MockDevice +from meshterm.core.models import Contact +from meshterm.core.remote_config import ( + REPEATER_SETTINGS, + get_setting, + known_commands, + parse_reply_value, + validate_value, +) +from meshterm.core.remote_store import HISTORY_CAP, RemoteStore +from meshterm.ui.remote_cli import RemoteCliScreen + +NODE = Contact(name="Yagi-Repeater", public_key="a1" * 32, key_prefix="a1b2c3d4") + + +class _FakeSession: + def __init__(self) -> None: + self.repaints = 0 + + def invalidate(self) -> None: + self.repaints += 1 + + +# --- the catalog ---------------------------------------------------------------------- + + +def test_catalog_carries_the_repeater_only_knobs() -> None: + """TX delay and Direct TX delay — the reason this feature exists — are first-class.""" + txdelay = get_setting("txdelay") + direct = get_setting("direct.txdelay") + assert txdelay is not None and txdelay.set_command("5") == "set txdelay 5" + assert direct is not None and direct.get_command == "get direct.txdelay" + assert any(s.key == "af" for s in REPEATER_SETTINGS) + assert any(s.key == "advert.interval" for s in REPEATER_SETTINGS) + + +def test_parse_reply_value_survives_terse_and_verbose_firmware() -> None: + """Numbers are extracted from any phrasing; errors parse as unknown, not values.""" + tx = get_setting("tx") + assert parse_reply_value(tx, "20") == "20" + assert parse_reply_value(tx, "tx: 20") == "20" + assert parse_reply_value(tx, "TX power = 20 dBm") == "20" + assert parse_reply_value(tx, "ERR: unknown config: tx") is None + assert parse_reply_value(tx, None) is None + + name = get_setting("name") + assert parse_reply_value(name, "> Yagi") == "Yagi" + assert parse_reply_value(name, "name: Yagi") == "Yagi" + + repeat = get_setting("repeat") + assert parse_reply_value(repeat, "on") == "on" + assert parse_reply_value(repeat, "repeat is off") == "off" + + +def test_validate_value_enforces_kind_and_bounds() -> None: + """Prompt validation speaks in the setting's own terms.""" + sf = get_setting("sf") + assert validate_value(sf, "9") is True + assert "≥ 7" in validate_value(sf, "3") + assert "number" in validate_value(sf, "fast") + assert validate_value(get_setting("repeat"), "maybe") == "Enter on or off." + assert validate_value(get_setting("name"), " ") == "Enter a value." + + +def test_known_commands_cover_catalog_and_verbs() -> None: + """The CLI completions include every catalog spelling plus the fixed verbs.""" + commands = known_commands() + assert "get txdelay" in commands + assert "set direct.txdelay " in commands + assert "reboot" in commands and "ver" in commands + assert "get guest.password" not in commands # write-only stays uncompletable + + +# --- the store ------------------------------------------------------------------------ + + +def test_remote_store_caches_settings_per_node(tmp_path: Path) -> None: + """Values round-trip with their read stamps, keyed per node.""" + store = RemoteStore(tmp_path / "remote.json") + store.remember_setting(NODE, "txdelay", "5") + cached = store.settings(NODE) + assert cached["txdelay"].value == "5" + assert cached["txdelay"].read_at is not None + other = Contact(name="Other", public_key="b2" * 32) + assert store.settings(other) == {} + + +def test_remote_store_history_dedupes_and_caps(tmp_path: Path) -> None: + """History appends in order, skips adjacent repeats, and stays capped.""" + store = RemoteStore(tmp_path / "remote.json") + store.append_history(NODE, "get tx") + store.append_history(NODE, "get tx") # adjacent repeat — no stutter on recall + store.append_history(NODE, "set tx 20") + assert store.history(NODE) == ["get tx", "set tx 20"] + for i in range(HISTORY_CAP + 20): + store.append_history(NODE, f"cmd {i}") + history = store.history(NODE) + assert len(history) == HISTORY_CAP and history[-1] == f"cmd {HISTORY_CAP + 19}" + + +# --- the simulated remote CLI ---------------------------------------------------------- + + +async def test_mock_remote_cli_requires_login() -> None: + """A stranger's command reads as a timeout (None), exactly like hardware.""" + device = MockDevice() + await device.connect() + assert await device.send_remote_command(NODE, "get tx") is None + await device.disconnect() + + +async def test_mock_remote_cli_round_trips_settings() -> None: + """get/set work for catalog keys; unknown keys answer with an error string.""" + device = MockDevice() + await device.connect() + assert await device.admin_login(NODE, "admin") + assert await device.send_remote_command(NODE, "get txdelay") == "> 0" + assert await device.send_remote_command(NODE, "set txdelay 5") == "OK" + assert await device.send_remote_command(NODE, "get txdelay") == "> 5" + assert await device.send_remote_command(NODE, "set tx 22") == "OK" + assert await device.get_remote_tx_power(NODE) == 22 # one shared TX state + reply = await device.send_remote_command(NODE, "get nonsense") + assert reply is not None and "unknown" in reply.lower() + assert "simulator" in (await device.send_remote_command(NODE, "ver")) + await device.disconnect() + + +# --- the command-line screen ----------------------------------------------------------- + + +def _cli(history=None, sent=None) -> RemoteCliScreen: + return RemoteCliScreen( + node_label="Yagi-Repeater", + history=list(history or []), + send=(sent.append if sent is not None else lambda c: None), + session=_FakeSession(), + ) + + +def _type(screen: RemoteCliScreen, text: str) -> None: + for ch in text: + screen.handle("text", ch) + + +def test_cli_screen_sends_on_enter_and_parks_while_waiting() -> None: + """Enter commits the buffer once; a command in flight blocks the next send.""" + sent: list[str] = [] + screen = _cli(sent=sent) + _type(screen, "get tx") + screen.handle("enter") + assert sent == ["get tx"] + screen.sent("get tx") # the owner echoes and parks the prompt + _type(screen, "ver") + screen.handle("enter") + assert sent == ["get tx"] # parked: nothing new goes out until the reply lands + screen.reply("20") + screen.handle("enter") # the typed-while-parked buffer is still there to send + assert sent == ["get tx", "ver"] + + +def test_cli_screen_history_recall_keeps_the_draft() -> None: + """↑ walks back through history; ↓ past the newest restores the unsent draft.""" + screen = _cli(history=["get tx", "set tx 20"]) + _type(screen, "dra") + screen.handle("up") + assert screen._editor.text == "set tx 20" + screen.handle("up") + assert screen._editor.text == "get tx" + screen.handle("down") + screen.handle("down") + assert screen._editor.text == "dra" # the draft came back + + +def test_cli_screen_tab_completes_known_commands() -> None: + """Tab adopts the first known command extending the typed prefix.""" + screen = _cli() + _type(screen, "get txd") + screen.handle("tab") + assert screen._editor.text == "get txdelay" + + +def test_cli_screen_transcript_shows_exchange() -> None: + """The transcript keeps the echoed command and its reply, prompt at the bottom.""" + screen = _cli() + screen.sent("get txdelay") + screen.reply("> 5") + body = "\n".join(screen.render_body(80)) + assert "get txdelay" in body and "> 5" in body + assert screen.cursor_line() == screen._prompt_line # the view follows the prompt + + +def test_cli_screen_timeout_note_frees_the_prompt() -> None: + """A timeout lands as a muted note and the prompt accepts input again.""" + screen = _cli() + screen.sent("advert") + assert screen.busy + screen.failed("no reply within 10 s") + assert not screen.busy + assert "no reply within 10 s" in "\n".join(screen.render_body(80))