From 99643f52d9d785eaed556e39f89c44c1b26eed6b Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Thu, 9 Jul 2026 22:28:42 -0400 Subject: [PATCH] Overhaul the config editor into a modal-dialog experience MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework the device-configuration screen from a flat list with an advert-laden "danger zone" into one grouped menu whose every prompt is a real floating modal popup over a persistent backdrop, matching the quit dialog. The main menu now stays pushed on the session stack for the whole editor session, so sub-prompts float over it rather than replacing the screen. Settings are staged and applied together (with a dirty-cancel confirmation); device actions — adverts, reboot, backup/restore, identity key, factory reset — run immediately behind their own dialogs. Adverts get a dedicated menu (zero-hop, flood, or share this node as a QR/URI contact card). Destructive actions gate behind a new TypedConfirmDialog that requires typing RESET / IMPORT. Reboot hands off to the reconnect dialog and waits for the companion to return. Location can be set by pointing at the full-screen map (new LocationPickScreen) or by typing coordinates. Make every dialog consistent: a short title in the border, the instruction as an in-body prompt, and a width sized to its content instead of a fixed wide cap. TextScreen/SelectScreen/AutocompleteScreen gain an optional in-body prompt, and those plus ConfirmScreen gain a dialog_width; the prompt is threaded through the surface and session. Co-Authored-By: Claude Opus 4.8 --- meshterm/context.py | 5 + meshterm/tools/config.py | 40 +- meshterm/tools/map.py | 160 +++---- meshterm/ui/config_editor.py | 794 ++++++++++++++++++++++++++++------- meshterm/ui/map_screen.py | 159 +++++++ meshterm/ui/menu.py | 12 +- meshterm/ui/surface.py | 138 +++++- meshterm/ui/tui/__init__.py | 9 +- meshterm/ui/tui/prompt.py | 142 ++++++- meshterm/ui/tui/select.py | 40 +- meshterm/ui/tui/session.py | 36 +- tests/test_config_editor.py | 559 ++++++++++++++++++++++++ tests/test_map.py | 2 +- 13 files changed, 1832 insertions(+), 264 deletions(-) create mode 100644 tests/test_config_editor.py diff --git a/meshterm/context.py b/meshterm/context.py index 1fcb157..958e2f6 100644 --- a/meshterm/context.py +++ b/meshterm/context.py @@ -70,6 +70,11 @@ class AppContext: _active_transport: Optional[str] = field(default=None, init=False, repr=False) _active_address: Optional[str] = field(default=None, init=False, repr=False) unpair_on_exit: bool = field(default=False, init=False, repr=False) + #: Set by the config editor just before it sends a reboot command, so the session's + #: disconnect watcher can label the ensuing (expected) link drop as a reboot in + #: progress rather than a surprise unplug. Cleared by the reconnect dialog that + #: consumes it (see :func:`meshterm.ui.menu._handle_disconnect`). + reboot_in_progress: bool = field(default=False, init=False, repr=False) _resume_intent: Optional[tuple[bool, bool, bool]] = field( default=None, init=False, repr=False ) diff --git a/meshterm/tools/config.py b/meshterm/tools/config.py index df0f806..5fc9ec2 100644 --- a/meshterm/tools/config.py +++ b/meshterm/tools/config.py @@ -2,8 +2,10 @@ Interactively it launches a full editor (see :mod:`meshterm.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. +and a ``--yes``-gated set of destructive operations. Both paths funnel through +:func:`apply_ops`: the editor stages value changes for :meth:`ConfigTool.run` to execute +and log, and runs its device actions (adverts, reboot, key management, factory reset) +through the same executor immediately. """ from __future__ import annotations @@ -132,12 +134,16 @@ class ConfigTool(Tool): 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.") + @config_app.command("advert", help="Broadcast an advertisement (zero-hop by default).") 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("share", help="Show this node's contact card as a QR code / URI.") + def _share_cmd() -> None: + run_tool_command(self, {"ops": [("share",)]}) + @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."), @@ -180,7 +186,8 @@ async def apply_ops( 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). + device actions (adverts, reboot, backup/restore, key management, factory reset — + which run immediately rather than staging). Args: ctx: Shared application context (for console output). @@ -219,8 +226,12 @@ async def apply_ops( 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.ui.note("[ok]✓[/ok] advertisement sent") + flood = len(op) > 1 and bool(op[1]) + await device.send_advert(flood) + kind_label = "flood" if flood else "zero-hop" + ctx.ui.note(f"[ok]✓[/ok] {kind_label} advertisement sent") + elif kind == "share": + _share_contact(ctx, snapshot) elif kind == "reboot": await device.reboot() ctx.ui.note("[warn]device rebooting[/warn]") @@ -265,6 +276,23 @@ async def _show(ctx: AppContext, device: Device, snapshot: dict) -> None: ctx.ui.show(config_table(snapshot, custom)) +def _share_contact(ctx: AppContext, snapshot: dict) -> None: + """Render this node's contact card — a scannable QR code plus the raw URI.""" + from rich.console import Group + from rich.text import Text + + from ..ui.config_editor import contact_share_url + from ..ui.qr import qr_text + + public_key = str(snapshot.get("public_key") or "") + if not public_key: + ctx.ui.note("[err]the device did not report a public key — nothing to share[/err]") + return + name = str(snapshot.get("name") or "this node") + url = contact_share_url(name, public_key, int(snapshot.get("adv_type") or 1)) + ctx.ui.show(Group(qr_text(url), Text(""), Text(url, style="accent"))) + + async def _backup(device: Device, snapshot: dict, path: Path) -> Path: """Write a TOML backup of the current configuration.""" custom = await device.get_custom_vars() diff --git a/meshterm/tools/map.py b/meshterm/tools/map.py index 4965e95..f50ef39 100644 --- a/meshterm/tools/map.py +++ b/meshterm/tools/map.py @@ -87,82 +87,8 @@ class MapTool(Tool): # -- marker gathering ------------------------------------------------------- async def _gather(self, ctx: AppContext) -> list["MapMarker"]: - """Collect every located node to plot: the device's contacts, plus our own node. - - The companion's **contact list** is the authoritative source for a node's name, - type (repeater vs. leaf), and advertised position — the passive-monitor - observations only carry a location for the rare node that broadcasts one in an - advert. So contacts drive the markers, and each contact is enriched with signal - detail from the observations when we've overheard it directly. - """ - from ..ui.map_render import MapMarker - - observed = {n.node: n for n in ctx.repo.heard_nodes() if n.node} - markers: list[MapMarker] = [] - seen: set[str] = set() - - for contact in await self._contacts(ctx): - if not contact.has_location: - continue - key = contact.key_prefix or (contact.public_key or "")[:12] - if key: - seen.add(key) - markers.append( - MapMarker( - label=contact.name or key or "?", - lat=float(contact.lat), - lon=float(contact.lon), - is_repeater=contact.is_repeater, - detail=_signal_detail(observed.get(key)), - ) - ) - - # A node we overheard advertising a location but that isn't in our contacts. - for node in observed.values(): - if not node.has_location or node.node in seen: - continue - markers.append( - MapMarker( - label=node.name or node.node or "?", - lat=float(node.lat), - lon=float(node.lon), - is_repeater=node.is_repeater, - detail=_signal_detail(node), - ) - ) - - self_marker = await self._self_marker(ctx) - if self_marker is not None: - markers.append(self_marker) - return markers - - @staticmethod - async def _contacts(ctx: AppContext) -> list: - """Fetch the device's contacts, best-effort (an unreachable radio yields none).""" - try: - return await (await ctx.device()).get_contacts() - except Exception: # noqa: BLE001 - the map still works from observations alone - return [] - - @staticmethod - async def _self_marker(ctx: AppContext) -> Optional["MapMarker"]: - """Build a marker for our own node from the device, if its location is known.""" - from ..ui.map_render import MapMarker - - try: - info = await (await ctx.device()).get_self_info() - except Exception: # noqa: BLE001 - the map is useful without our own position - return None - lat, lon = _as_float(info.get("adv_lat")), _as_float(info.get("adv_lon")) - if lat is None or lon is None or (abs(lat) < 1e-6 and abs(lon) < 1e-6): - return None # a device with no fix reports 0/0 - return MapMarker( - label=str(info.get("name") or "this node"), - lat=lat, - lon=lon, - is_repeater=info.get("adv_type") == NODE_TYPE_REPEATER, - is_self=True, - ) + """Collect every located node to plot (see :func:`gather_markers`).""" + return await gather_markers(ctx) # -- static (CLI) render ---------------------------------------------------- @@ -248,6 +174,88 @@ class MapTool(Tool): run_tool_command(self, params) +async def gather_markers(ctx: AppContext) -> list["MapMarker"]: + """Collect every located node to plot: the device's contacts, plus our own node. + + The companion's **contact list** is the authoritative source for a node's name, + type (repeater vs. leaf), and advertised position — the passive-monitor + observations only carry a location for the rare node that broadcasts one in an + advert. So contacts drive the markers, and each contact is enriched with signal + detail from the observations when we've overheard it directly. + + Shared by the ``map`` tool and the config editor's pick-a-location map, so both + show the same mesh. + """ + from ..ui.map_render import MapMarker + + observed = {n.node: n for n in ctx.repo.heard_nodes() if n.node} + markers: list[MapMarker] = [] + seen: set[str] = set() + + for contact in await _contacts(ctx): + if not contact.has_location: + continue + key = contact.key_prefix or (contact.public_key or "")[:12] + if key: + seen.add(key) + markers.append( + MapMarker( + label=contact.name or key or "?", + lat=float(contact.lat), + lon=float(contact.lon), + is_repeater=contact.is_repeater, + detail=_signal_detail(observed.get(key)), + ) + ) + + # A node we overheard advertising a location but that isn't in our contacts. + for node in observed.values(): + if not node.has_location or node.node in seen: + continue + markers.append( + MapMarker( + label=node.name or node.node or "?", + lat=float(node.lat), + lon=float(node.lon), + is_repeater=node.is_repeater, + detail=_signal_detail(node), + ) + ) + + self_marker = await _self_marker(ctx) + if self_marker is not None: + markers.append(self_marker) + return markers + + +async def _contacts(ctx: AppContext) -> list: + """Fetch the device's contacts, best-effort (an unreachable radio yields none).""" + try: + return await (await ctx.device()).get_contacts() + except Exception: # noqa: BLE001 - the map still works from observations alone + return [] + + +async def _self_marker(ctx: AppContext) -> Optional["MapMarker"]: + """Build a marker for our own node from the device, if its location is known.""" + from ..ui.map_render import MapMarker + + try: + info = await (await ctx.device()).get_self_info() + except Exception: # noqa: BLE001 - the map is useful without our own position + return None + lat, lon = _as_float(info.get("adv_lat")), _as_float(info.get("adv_lon")) + if lat is None or lon is None or (abs(lat) < 1e-6 and abs(lon) < 1e-6): + return None # a device with no fix reports 0/0 + return MapMarker( + label=str(info.get("name") or "this node"), + lat=lat, + lon=lon, + is_repeater=info.get("adv_type") == NODE_TYPE_REPEATER, + is_self=True, + ) + + def _legend(markers: list["MapMarker"]) -> Table: """A compact legend: self → repeaters → leaf nodes, with coordinates and detail.""" from ..ui.map_render import _NODE, _REPEATER, _SELF diff --git a/meshterm/ui/config_editor.py b/meshterm/ui/config_editor.py index ad6706e..2ba0751 100644 --- a/meshterm/ui/config_editor.py +++ b/meshterm/ui/config_editor.py @@ -1,92 +1,175 @@ """Interactive device-configuration editor, rendered in the full-screen session. -Drives the menu flow for the ``config`` tool: it presents the device's settings (each row -showing its current value and any staged change), lets the user stage changes setting-by- -setting with type-aware prompts and validation, and handles custom variables, -backup/restore and the destructive "danger zone". (Channels have their own first-class -manager — see the ``channels`` tool.) It returns an operation list for -:class:`~meshterm.tools.config.ConfigTool` to execute and log; it performs no device writes -itself (except danger-zone actions, which run immediately and show their result at once). +Drives the menu flow for the ``config`` tool. The screen is one grouped list: every +setting under its category heading (each row showing its current value, any staged change, +and a one-line explanation), followed by a *Device actions* section for the operations that +act on the box itself rather than a value — adverts, reboot, backup/restore, the identity +key, and factory reset. + +Two kinds of interaction live here, deliberately kept distinct: + +* **Settings are staged.** Editing a row stages the new value (shown as ``current → new`` + in the row) and nothing touches the radio until *Apply*; backing out with staged changes + asks before discarding them. The editor returns the staged operations for + :class:`~meshterm.tools.config.ConfigTool` to execute and log. +* **Device actions run immediately** (after their own confirmation dialog — destructive + ones gate behind typing a confirmation word). They have no meaningful "preview", so their + result is shown at once. + +Multiple-choice values are picked in dialogs (booleans as an On/Off button pair, enums as +a floating select), the node's location can be set by pointing at the full-screen map (see +:class:`~meshterm.ui.map_screen.LocationPickScreen`), and a reboot hands off to the same +reconnect dialog the app shows when a device is unplugged. (Channels have their own +first-class manager — see the ``channels`` tool.) """ from __future__ import annotations +import asyncio from pathlib import Path -from typing import Any, Optional +from typing import TYPE_CHECKING, Any, Optional +from urllib.parse import quote from rich import box +from rich.console import Group from rich.table import Table from rich.text import Text -from ..context import AppContext from ..core.device_config import ( DeviceConfigError, SettingSpec, build_snapshot, format_value, + get_spec, parse_value, settings_by_category, ) from .tui import Choice, Separator +if TYPE_CHECKING: + from ..context import AppContext + from ..core.connection import Device + # Menu action sentinels (distinct from setting keys, which are plain strings). +_LOCATION = "__location__" _PRESETS = "__presets__" _CUSTOM = "__custom__" +_ADVERT = "__advert__" +_REBOOT = "__reboot__" _BACKUP = "__backup__" _RESTORE = "__restore__" -_DANGER = "__danger__" +_IDENTITY_KEY = "__identity_key__" +_RESET = "__reset__" _VIEW = "__view__" _APPLY = "__apply__" _CANCEL = "__cancel__" # Sentinel for the "enter a value myself" option on non-strict enum prompts. _OTHER = "__other__" +#: The setting keys folded into the single "Location" row (they stay individually +#: addressable from the CLI; only the editor presents them as one place-on-earth value). +_COORD_KEYS = ("adv_lat", "adv_lon") -async def edit_config(ctx: AppContext) -> Optional[list[tuple]]: - """Run the interactive editor and return the operations to perform. +#: How long the reboot flow waits to *observe* the link actually dropping before handing +#: off to the session's reconnect dialog (seconds). A companion normally vanishes from the +#: bus well within this; on timeout we hand off anyway. +_REBOOT_DROP_TIMEOUT_S = 10.0 + +#: Poll cadence while waiting for the rebooting companion's link to drop (seconds). +_REBOOT_DROP_POLL_S = 0.25 + + +async def edit_config(ctx: "AppContext") -> Optional[list[tuple]]: + """Run the interactive editor and return the staged operations to perform. Args: ctx: Shared application context (provides the connected device and UI surface). Returns: A list of operation tuples for the tool to execute, or ``None`` if the user - cancelled without choosing to apply anything. + cancelled (or a device action — e.g. a reboot — ended the session) without + anything staged to apply. """ + from .tui import CANCEL, SelectScreen + device = await ctx.device() snapshot = await build_snapshot(device) custom = await device.get_custom_vars() + # The editor is menu-only, so a full-screen session is always present. Keep the main + # menu *pushed on the stack* for the whole session (rather than popping it between + # prompts): every sub-prompt then floats over it as a modal popup with its own border + # — the quit-dialog pattern — instead of replacing the screen. See _menu_loop. + session = getattr(ctx.ui, "session", None) + if session is None: # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("the config editor is only available in the menu") + loop = asyncio.get_running_loop() + pending: dict[str, Any] = {} # setting key -> staged new value - extra_ops: list[tuple] = [] # custom/channel/backup/restore/danger ops, in order + extra_ops: list[tuple] = [] # staged custom-variable ops, in order + cursor: Any = None # the row to re-highlight, so the menu reopens where you left it while True: - choice = await _main_menu(ctx, snapshot, pending, extra_ops) + staged = len(pending) + len(extra_ops) + title, items = _menu_items(snapshot, pending, staged) + menu = SelectScreen( + title, items, default=cursor, wrap=False, + footer_hint="↑↓ move · type to filter · Enter select · Esc close", + ) + menu.future = loop.create_future() + session.push(menu) + # Dispatch the choice while the menu is still pushed, so a sub-prompt floats over + # it; the menu is always popped in the finally, even on an early return. + try: + choice = await menu.future + if choice is CANCEL: # Esc at the menu + choice = _CANCEL + if choice not in (None, _CANCEL): + cursor = choice - 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 == _VIEW: - await ctx.ui.view( - config_table(snapshot, custom, pending), title="Device configuration" - ) - elif choice == _PRESETS: - await _stage_preset(ctx, pending) - elif choice == _CUSTOM: - await _stage_custom_var(ctx, custom, extra_ops) - elif choice == _BACKUP: - await _stage_backup(ctx, extra_ops) - elif choice == _RESTORE: - await _stage_restore(ctx, 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(ctx, choice, snapshot, pending) + if choice in (None, _CANCEL): + if staged and not await _confirm_discard(ctx, staged): + continue # keep editing — the same menu is rebuilt next loop + 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 == _VIEW: + await ctx.ui.view( + config_table(snapshot, custom, pending), title="Device configuration" + ) + elif choice == _LOCATION: + await _stage_location(ctx, snapshot, pending) + elif choice == _PRESETS: + await _stage_preset(ctx, pending) + elif choice == _CUSTOM: + await _stage_custom_var(ctx, custom, extra_ops) + elif choice == _ADVERT: + await _advert_menu(ctx, device, snapshot) + elif choice == _REBOOT: + if await _reboot(ctx, device, snapshot, staged): + return None # the link is dropping; the reconnect dialog takes over + elif choice == _BACKUP: + await _backup_now(ctx, device, snapshot) + elif choice == _RESTORE: + if await _restore_now(ctx, device, snapshot): + snapshot = await build_snapshot(device) + custom = await device.get_custom_vars() + elif choice == _IDENTITY_KEY: + if await _identity_key_menu(ctx, device, snapshot): + snapshot = await build_snapshot(device) + elif choice == _RESET: + if await _factory_reset(ctx, device, snapshot): + # Everything the editor knew about the device is gone; start clean. + pending.clear() + extra_ops.clear() + snapshot = await build_snapshot(device) + custom = await device.get_custom_vars() + else: # a setting key + await _stage_setting(ctx, choice, snapshot, pending) + finally: + session.pop(menu) # --- rendering --------------------------------------------------------------- @@ -156,75 +239,188 @@ def config_table( return table -async def _main_menu( - ctx: AppContext, snapshot: dict, pending: dict, extra_ops: list -) -> Optional[str]: - """Show the top-level editor menu and return the chosen action or setting key. +def _setting_row(spec: SettingSpec, snapshot: dict, pending: dict) -> Text: + """Build one setting's menu row: ``label: current [→ staged] — help``. - 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. + The staged arrow is drawn in the warn style so a dirty row stands out at a glance, + and the help trails muted; both spans survive the select highlight (which only tints + the row's base style). + """ + row = Text(f"{spec.label}: ") + row.append(format_value(spec, spec.getter(snapshot))) + if spec.key in pending: + row.append(f" → {format_value(spec, pending[spec.key])}", style="warn") + row.append(f" — {spec.help}", style="muted") + return row + + +def _format_coords(lat: Any, lon: Any) -> str: + """Render a coordinate pair for display (``"not set"`` for the 0,0 no-fix value).""" + try: + lat_f, lon_f = float(lat or 0.0), float(lon or 0.0) + except (TypeError, ValueError): + return "?" + if abs(lat_f) < 1e-6 and abs(lon_f) < 1e-6: + return "not set" + return f"{lat_f:.5f}, {lon_f:.5f}" + + +def _location_row(snapshot: dict, pending: dict) -> Text: + """The single Location row standing in for the ``adv_lat``/``adv_lon`` pair.""" + row = Text("Location: ") + row.append(_format_coords(snapshot.get("adv_lat"), snapshot.get("adv_lon"))) + if any(k in pending for k in _COORD_KEYS): + lat = pending.get("adv_lat", snapshot.get("adv_lat")) + lon = pending.get("adv_lon", snapshot.get("adv_lon")) + row.append(f" → {_format_coords(lat, lon)}", style="warn") + row.append(" — Advertised position; pick it on the map", style="muted") + return row + + +def _menu_items( + snapshot: dict, pending: dict, staged: int +) -> tuple[str, list]: + """Build the editor menu's title and rows for the current snapshot + staged state. + + Returns the ``(title, items)`` the caller pushes as a persistent backdrop screen (so + sub-prompts float over it). 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. The device actions that run immediately live in their own + section below the settings. """ items: list = [] for category, specs in settings_by_category(): items.append(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 - ) + if spec.key in _COORD_KEYS: + # Latitude/longitude collapse into one Location row (inserted in + # adv_lat's slot so it sits where the coordinates used to). + if spec.key == "adv_lat": + items.append(Choice(title=_location_row(snapshot, pending), value=_LOCATION)) + continue + items.append(Choice(title=_setting_row(spec, snapshot, pending), value=spec.key)) + if category == "Radio": items.append( - Choice(title=f"{spec.label}: {shown} — {spec.help}", value=spec.key) + Choice( + title=Text.assemble( + "Radio presets…", + (" — Apply a standard regional or trade-off config", "muted"), + ), + value=_PRESETS, + ) + ) + elif category == "Experimental": + items.append( + Choice( + title=Text.assemble( + "Custom variables…", + (" — Set a raw firmware variable by name", "muted"), + ), + value=_CUSTOM, + ) ) - items.append(Separator("── More ──")) - items.append(Choice(title="View current config (full table)", value=_VIEW)) - items.append(Choice(title="Radio presets (standard configs)", value=_PRESETS)) - items.append(Choice(title="Custom / experimental vars", value=_CUSTOM)) - items.append(Choice(title="Backup to file", value=_BACKUP)) - items.append(Choice(title="Restore from file", value=_RESTORE)) - items.append(Choice(title="⚠ Danger zone", value=_DANGER)) - items.append(Separator(" ")) - staged = len(pending) + len(extra_ops) - items.append(Choice(title=f"✓ Apply ({staged} staged)", value=_APPLY)) - items.append(Choice(title="Cancel (discard)", value=_CANCEL)) - choice = await ctx.ui.select("Edit which setting?", items) - return _CANCEL if choice is None else choice + items.append(Separator("── Device actions (run immediately) ──")) + items.append(_action("📡 Send advert…", "Zero-hop, flood, or share this node as a QR code", _ADVERT)) + items.append(_action("💾 Back up config to a file…", "Write every setting to TOML", _BACKUP)) + items.append(_action("📂 Restore config from a backup…", "Preview or apply a saved TOML", _RESTORE)) + items.append(_action("🔐 Identity key…", "Export or import the node's private key", _IDENTITY_KEY)) + items.append(_action("🔄 Reboot device…", "Restart the companion and reconnect", _REBOOT)) + items.append( + Choice( + title=Text.assemble( + ("⚠ Factory reset…", "err"), + (" — Erase everything (typed confirmation)", "muted"), + ), + value=_RESET, + ) + ) + + items.append(Separator("── Review ──")) + items.append(_action("🧾 View full configuration", "Every value in one table", _VIEW)) + if staged: + items.append( + Choice( + title=Text.assemble(("✓ ", "ok"), f"Apply {staged} staged change(s)"), + value=_APPLY, + ) + ) + items.append( + Choice( + title=Text.assemble(("✗ ", "err"), "Discard staged changes & close"), + value=_CANCEL, + ) + ) + else: + items.append(Choice(title="Close", value=_CANCEL)) + + title = "Device configuration" + (f" — {staged} staged" if staged else "") + return title, items + + +def _action(label: str, help_text: str, value: str) -> Choice: + """Build a device-action menu row: a label with a muted explanation.""" + return Choice(title=Text.assemble(label, (f" — {help_text}", "muted")), value=value) # --- staging individual changes ---------------------------------------------- async def _stage_setting( - ctx: AppContext, key: str, snapshot: dict, pending: dict[str, Any] + ctx: "AppContext", 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(ctx, spec, current) - if value is not None: + if value is None: + return + if value == spec.getter(snapshot): + pending.pop(key, None) # set back to the device's value — nothing to change + else: pending[key] = value -async def _prompt_value(ctx: AppContext, spec: SettingSpec, current: Any) -> Any: - """Prompt for a typed value for ``spec``, returning ``None`` on cancel.""" +def _range_hint(spec: SettingSpec) -> str: + """A muted "allowed values" hint for a numeric prompt, from the spec's bounds.""" + if spec.minimum is not None and spec.maximum is not None: + return f"Allowed: {spec.minimum:g} – {spec.maximum:g}" + if spec.minimum is not None: + return f"Allowed: ≥ {spec.minimum:g}" + if spec.maximum is not None: + return f"Allowed: ≤ {spec.maximum:g}" + return "" + + +async def _prompt_value(ctx: "AppContext", spec: SettingSpec, current: Any) -> Any: + """Prompt for a typed value for ``spec`` (in the fitting dialog), ``None`` on cancel.""" if spec.value_type == "bool": - return await ctx.ui.confirm(f"{spec.label}?", default=bool(current)) + # A straight two-state choice reads best as a button pair; the current state is + # the highlighted default so Enter changes nothing by accident. + return await ctx.ui.dialog( + spec.help, + [("Off", False), ("On", True)], + title=spec.label, + default=1 if current else 0, + keys={"0": False, "1": True, "n": False, "y": True}, + ) if spec.value_type == "enum" and spec.choices is not None: items: list = [ - Choice(title=f"{k} — {label}", value=k) for k, label in spec.choices.items() + Choice( + title=f"{k} — {label}" + (" (current)" if k == current else ""), + 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(Choice(title="Other (enter a value)…", value=_OTHER)) selected = await ctx.ui.select( - f"{spec.label} — {spec.help}", + spec.label, items, + prompt=spec.help, default=current if current in spec.choices else None, ) if selected is None: @@ -241,14 +437,93 @@ async def _prompt_value(ctx: AppContext, spec: SettingSpec, current: Any) -> Any return str(exc) raw = await ctx.ui.text( - f"{spec.label} — {spec.help}", + spec.label, + prompt=spec.help, default="" if current is None else str(current), validate=validate, + help_text=_range_hint(spec), ) return None if raw is None else parse_value(spec, raw) -async def _stage_preset(ctx: AppContext, pending: dict[str, Any]) -> None: +async def _stage_location( + ctx: "AppContext", snapshot: dict, pending: dict[str, Any] +) -> None: + """Set the advertised location: on the map, typed as a pair, or cleared. + + Staged like any other setting — the coordinates only reach the device on Apply. + """ + lat = pending.get("adv_lat", snapshot.get("adv_lat")) + lon = pending.get("adv_lon", snapshot.get("adv_lon")) + choice = await ctx.ui.dialog( + f"Advertised location: {_format_coords(lat, lon)}", + [("Pick on map", "map"), ("Type coordinates", "type"), ("Clear", "clear")], + title="📍 Location", + ) + if choice is None: + return + + if choice == "map": + from .map_screen import pick_location + + initial = None + try: + if abs(float(lat or 0.0)) >= 1e-6 or abs(float(lon or 0.0)) >= 1e-6: + initial = (float(lat), float(lon)) + except (TypeError, ValueError): + initial = None + picked = await pick_location(ctx, initial=initial) + if picked is None: + return + # Six decimals ≈ 0.1 m — beyond the map's own precision, plenty for an advert. + pending["adv_lat"] = round(picked[0], 6) + pending["adv_lon"] = round(picked[1], 6) + elif choice == "type": + raw = await ctx.ui.text( + "Set location", + prompt="Enter latitude, longitude in decimal degrees.", + default=f"{lat}, {lon}" if _format_coords(lat, lon) != "not set" else "", + validate=_valid_coords, + help_text="e.g. 45.50000, -73.60000", + ) + if not raw: + return + parsed = _parse_coords(raw) + pending["adv_lat"], pending["adv_lon"] = parsed + elif choice == "clear": + # 0, 0 is MeshCore's "no fix" value: the node stops advertising a position. + pending["adv_lat"], pending["adv_lon"] = 0.0, 0.0 + + # Staging the device's own values back is a no-op; drop them so the row reads clean. + for key in _COORD_KEYS: + if key in pending and pending[key] == snapshot.get(key): + del pending[key] + + +def _parse_coords(text: str) -> tuple[float, float]: + """Parse a ``lat, lon`` pair (comma or space separated), range-checked via the specs. + + Raises: + DeviceConfigError: If the text is not two in-range decimal degrees. + """ + parts = [p for p in text.replace(",", " ").split() if p] + if len(parts) != 2: + raise DeviceConfigError("enter two numbers: latitude, longitude") + lat = parse_value(get_spec("adv_lat"), parts[0]) + lon = parse_value(get_spec("adv_lon"), parts[1]) + return float(lat), float(lon) + + +def _valid_coords(text: str) -> bool | str: + """Validate a typed coordinate pair, returning the parse error as the message.""" + try: + _parse_coords(text) + return True + except DeviceConfigError as exc: + return str(exc) + + +async def _stage_preset(ctx: "AppContext", 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 @@ -261,7 +536,11 @@ async def _stage_preset(ctx: AppContext, pending: dict[str, Any]) -> None: ] items.append(Separator(" ")) items.append(Choice(title="Back", value=None)) - idx = await ctx.ui.select("Apply which radio preset?", items) + idx = await ctx.ui.select( + "Radio presets", + items, + prompt="Stage a standard set of radio parameters:", + ) if idx is None: return preset = RADIO_PRESETS[idx] @@ -269,101 +548,322 @@ async def _stage_preset(ctx: AppContext, pending: dict[str, Any]) -> None: async def _stage_custom_var( - ctx: AppContext, custom: dict[str, str], extra_ops: list[tuple] + ctx: "AppContext", custom: dict[str, str], extra_ops: list[tuple] ) -> None: - """Prompt for a custom/experimental variable and stage a set operation.""" - key = await ctx.ui.text("Custom variable name:") - if not key: + """Prompt for a custom/experimental variable and stage a set operation. + + Known variable names are offered as suggestions so an existing one can be recalled + without retyping it; any new name is accepted as free text. + """ + if custom: + key = await ctx.ui.autocomplete( + "Custom variable", + sorted(custom), + prompt="Name of the firmware variable to set:", + ) + else: + key = await ctx.ui.text( + "Custom variable", prompt="Name of the firmware variable to set:" + ) + if not key or not key.strip(): return - value = await ctx.ui.text(f"Value for {key}:", default=custom.get(key, "")) + key = key.strip() + value = await ctx.ui.text( + "Custom variable", prompt=f"Value for {key}:", default=custom.get(key, "") + ) if value is None: return - extra_ops.append(("set_custom", key.strip(), value)) + extra_ops.append(("set_custom", key, value)) -async def _stage_backup(ctx: AppContext, extra_ops: list[tuple]) -> None: - """Prompt for a backup destination path and stage the operation.""" - path = await ctx.ui.path("Write backup to:", default="meshterm-config.toml") - if path: - extra_ops.append(("backup", Path(path))) +# --- device actions (run immediately) ----------------------------------------- -async def _stage_restore(ctx: AppContext, extra_ops: list[tuple]) -> None: - """Prompt for a backup file and whether to preview, then stage the operation.""" - path = await ctx.ui.path("Restore from:") - if not path: - return - dry_run = await ctx.ui.confirm("Preview changes only (dry run)?", default=True) - if dry_run is None: - return - extra_ops.append(("restore", Path(path), bool(dry_run))) +async def _run_now( + ctx: "AppContext", device: "Device", snapshot: dict, ops: list[tuple], title: str +) -> int: + """Execute ``ops`` on the device right away and show the result window. - -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; their - output is shown at once in a result window. + The immediate-action counterpart of the staged Apply path: same executor + (:func:`~meshterm.tools.config.apply_ops`), so the notes and behavior match, but the + output is presented at once instead of waiting for the tool to finish. 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. + The number of changes applied. """ from ..tools.config import apply_ops - action = await ctx.ui.select( - "⚠ Danger zone (runs immediately on confirmation):", + changes, artifacts = await apply_ops(ctx, device, snapshot, ops) + for artifact in artifacts: + ctx.ui.note(f"[ok]●[/ok] wrote [accent]{artifact}[/accent]") + await ctx.ui.present(title=title) + return changes + + +async def _advert_menu(ctx: "AppContext", device: "Device", snapshot: dict) -> None: + """Send an advert (zero-hop or flood) or show this node's shareable contact card.""" + choice = await ctx.ui.select( + "📡 Send advert", [ - Choice("Send advert", value="advert"), - Choice("Reboot device", value="reboot"), - Choice("Export private key", value="export_key"), - Choice("Import private key", value="import_key"), - Choice("Factory reset (erase all)", value="factory_reset"), + Choice(title="Zero-hop — Announce directly to neighbours in range", value="zero"), + Choice(title="Flood — Repeaters rebroadcast it across the mesh", value="flood"), + Choice(title="Share QR / URI — Show this node's contact card", value="share"), Separator(" "), - Choice("Back", value=None), + Choice(title="Back", value=None), ], + prompt="Announce this node to the mesh:", + ) + if choice is None: + return + if choice == "share": + await _show_contact_card(ctx, snapshot) + return + await _run_now(ctx, device, snapshot, [("advert", choice == "flood")], "advert") + + +def contact_share_url(name: str, public_key: str, node_type: int = 1) -> str: + """Build the MeshCore ``meshcore://contact/add`` share URL for this node. + + The companion-app format (see the MeshCore ``qr_codes`` doc): the advertised name, + the full 32-byte public key as hex, and the node type (1 = companion, 2 = repeater, + 3 = room server, 4 = sensor). + + Args: + name: The node's advertised name. + public_key: The node's public key as a hex string. + node_type: The MeshCore advert type byte. + + Returns: + A ``meshcore://contact/add?name=…&public_key=…&type=…`` URL. + """ + return ( + f"meshcore://contact/add?name={quote(name, safe='')}" + f"&public_key={public_key.lower()}&type={int(node_type)}" ) - op: Optional[tuple] = None - if action in (None, "Back"): + +async def _show_contact_card(ctx: "AppContext", snapshot: dict) -> None: + """Show this node's contact card as a scannable QR code plus the raw URI.""" + from .qr import qr_text + + public_key = str(snapshot.get("public_key") or "") + if not public_key: + ctx.ui.note("[err]the device did not report a public key — nothing to share[/err]") + await ctx.ui.present(title="share contact") + return + name = str(snapshot.get("name") or "this node") + url = contact_share_url(name, public_key, int(snapshot.get("adv_type") or 1)) + body = Group( + Text("Scan to add this node as a contact:", style="muted"), + Text(""), + qr_text(url), + Text(""), + Text(url, style="accent"), + ) + await ctx.ui.view(body, title=f"Share {name}", footer_hint="Esc back") + + +async def _reboot( + ctx: "AppContext", device: "Device", snapshot: dict, staged: int +) -> bool: + """Confirm and reboot the device, handing off to the session's reconnect dialog. + + The confirmation dialog warns when staged changes would be lost (a reboot ends the + editor, discarding them). After the command is sent, we wait to actually observe the + link dropping — flagging :attr:`~meshterm.context.AppContext.reboot_in_progress` so + the session-wide disconnect watcher labels the ensuing dialog as a reboot, waits for + the companion to come back, and reconnects — exactly the unplugged-device flow. + + Returns: + ``True`` if the reboot was sent and the editor should close; ``False`` if the + user backed out (or the simulator, which has no link to drop, absorbed it). + """ + warning = "Reboot the device now?" + if staged: + warning = ( + f"Reboot the device now? Your {staged} staged change(s) have not been " + "applied and will be discarded." + ) + choice = await ctx.ui.dialog( + warning, + [("Cancel", None), ("Reboot", "reboot")], + title="🔄 Reboot device", + default=1, + danger=True, + ) + if choice != "reboot": return False - if action in ("advert", "export_key"): - op = (action,) - elif action == "import_key": - key_hex = await ctx.ui.text("Private key (hex):", validate=_is_hex) - if key_hex and await _confirm_typed(ctx, "IMPORT"): - op = ("import_key", key_hex.strip()) - elif action == "reboot": - if await ctx.ui.confirm("Reboot the device now?", default=False): - op = ("reboot",) - elif action == "factory_reset": - if await _confirm_typed( - ctx, - "RESET", - warning="This erases ALL data on the device and cannot be undone.", - ): - op = ("factory_reset",) - if op is None: + if ctx.active_transport is None: + # The simulator has no link to drop and comes back instantly; just send it. + await device.reboot() + ctx.ui.note("[warn]device rebooting[/warn]") + await ctx.ui.present(title="reboot") return False - await apply_ops(ctx, device, snapshot, [op]) - await ctx.ui.present(title="danger zone") # show the immediate result now - return action in ("factory_reset", "import_key") + + # Flag the drop as expected *before* sending, so however quickly the watcher fires, + # the reconnect dialog already knows to present it as a reboot. + ctx.reboot_in_progress = True + try: + await device.reboot() + except Exception: + ctx.reboot_in_progress = False + raise + # Hold here until the link is actually observed down (or a generous timeout), so the + # editor doesn't flash back to the menu for the second or two before the watcher + # notices. The watcher may cancel us mid-wait when it fires — that's the handoff. + deadline = asyncio.get_running_loop().time() + _REBOOT_DROP_TIMEOUT_S + while asyncio.get_running_loop().time() < deadline: + if not await ctx.link_alive(): + break + await asyncio.sleep(_REBOOT_DROP_POLL_S) + return True -async def _confirm_typed(ctx: AppContext, word: str, *, warning: str = "") -> bool: - """Require the user to type ``word`` exactly to confirm a destructive action.""" - typed = await ctx.ui.text( - f"Type {word!r} to confirm:", help_text=warning, validate=None +async def _backup_now(ctx: "AppContext", device: "Device", snapshot: dict) -> None: + """Prompt for a destination and write the TOML backup immediately.""" + path = await ctx.ui.path( + "Back up config", + prompt="Write every setting to this TOML file:", + default="meshterm-config.toml", ) - if typed == word: - return True - await ctx.ui.view( - Text.from_markup("[muted]confirmation did not match; skipped.[/muted]"), - title="cancelled", + if path: + await _run_now(ctx, device, snapshot, [("backup", Path(path))], "backup") + + +async def _restore_now(ctx: "AppContext", device: "Device", snapshot: dict) -> bool: + """Restore from a TOML backup: pick the file, preview if wanted, then apply. + + Returns: + ``True`` if the device was changed (so the caller refreshes its snapshot). + """ + raw = await ctx.ui.path( + "Restore config", prompt="Read settings from this TOML backup file:" ) - return False + if not raw: + return False + path = Path(raw) + if not path.exists(): + ctx.ui.note(f"[err]no such file:[/err] {path}") + await ctx.ui.present(title="restore") + return False + + choice = await ctx.ui.dialog( + "Apply the backup now, or preview the changes first?", + [("Cancel", None), ("Preview", "preview"), ("Apply", "apply")], + title="📂 Restore from backup", + default=1, + ) + if choice == "preview": + await _run_now(ctx, device, snapshot, [("restore", path, True)], "restore preview") + choice = await ctx.ui.dialog( + "Apply these changes to the device?", + [("Cancel", None), ("Apply", "apply")], + title="📂 Restore from backup", + default=1, + ) + if choice != "apply": + return False + changed = await _run_now(ctx, device, snapshot, [("restore", path, False)], "restore") + return changed > 0 + + +async def _identity_key_menu(ctx: "AppContext", device: "Device", snapshot: dict) -> bool: + """Export or import the device's private identity key. + + Returns: + ``True`` if the identity changed (a key was imported), so the caller re-reads + its snapshot. + """ + choice = await ctx.ui.select( + "🔐 Identity key", + [ + Choice(title="Show private key — Display it on screen (sensitive)", value="show"), + Choice(title="Export to a file… — Write it to disk (keep it secret)", value="file"), + Choice(title="Import a key… — Replace this device's identity", value="import"), + Separator(" "), + Choice(title="Back", value=None), + ], + prompt="Manage this node's private identity key:", + ) + if choice is None: + return False + + if choice == "show": + ok = await ctx.ui.dialog( + "The private key IS the node's identity — anyone who sees it can impersonate " + "this node. Show it on screen?", + [("Cancel", None), ("Show key", "show")], + title="🔐 Show private key", + default=1, + danger=True, + ) + if ok == "show": + await _run_now(ctx, device, snapshot, [("export_key",)], "private key") + return False + + if choice == "file": + path = await ctx.ui.path( + "Export identity key", + prompt="Write the private key to this file (keep it secret):", + default="meshterm-identity.key", + ) + if path: + await _run_now(ctx, device, snapshot, [("export_key", Path(path))], "private key") + return False + + # Import: collect the key, then gate behind the typed confirmation. + key_hex = await ctx.ui.text( + "Import identity key", + prompt="Paste the private key as hex:", + validate=_is_hex, + ) + if not key_hex: + return False + confirmed = await ctx.ui.typed_confirm( + "Importing a key permanently overwrites this device's identity. Contacts and " + "messages keyed to the old identity will no longer match it.", + "IMPORT", + title="🔐 Import private key", + ) + if not confirmed: + return False + await _run_now(ctx, device, snapshot, [("import_key", key_hex.strip())], "import key") + return True + + +async def _factory_reset(ctx: "AppContext", device: "Device", snapshot: dict) -> bool: + """Factory-reset the device behind a typed confirmation. + + Returns: + ``True`` if the reset ran (so the caller drops everything it staged and re-reads + the device). + """ + confirmed = await ctx.ui.typed_confirm( + "This erases EVERYTHING on the device — identity, contacts, channels, and every " + "setting — and cannot be undone.", + "RESET", + title="⚠ Factory reset", + ) + if not confirmed: + return False + await _run_now(ctx, device, snapshot, [("factory_reset",)], "factory reset") + return True + + +# --- confirmations ------------------------------------------------------------- + + +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 {staged} staged change(s) without applying them?", + [("Keep editing", "keep"), ("Discard", "discard")], + title="Unsaved changes", + default=1, + danger=True, + ) + return choice == "discard" # --- validators -------------------------------------------------------------- diff --git a/meshterm/ui/map_screen.py b/meshterm/ui/map_screen.py index aeadc93..7b7e462 100644 --- a/meshterm/ui/map_screen.py +++ b/meshterm/ui/map_screen.py @@ -281,6 +281,165 @@ class MapScreen(Screen): self.resolve(None) +class LocationPickScreen(MapScreen): + """The map, repurposed as a coordinate picker: pan the crosshair, Enter to choose. + + Used by the config editor to set the node's advertised location by *pointing at the + map* instead of typing degrees. It is a :class:`MapScreen` with three changes: a + crosshair marker rides the view centre (labelled with the live coordinates, so the + user always sees exactly what they're about to pick), Enter resolves with the centre's + ``(lat, lon)`` instead of doing nothing, and ``r`` recentres on the *initial* location + rather than refitting the node cloud. The surrounding mesh nodes are still drawn, so + placing yourself relative to a known repeater is easy. Esc cancels (resolves CANCEL, + surfaced as ``None`` by the caller). + """ + + def __init__( + self, + session, # noqa: ANN001 - TuiSession, imported lazily to avoid a cycle + markers: list[MapMarker], + source: BasemapSource, + max_tile_zoom: int, + *, + initial: Optional[tuple[float, float]] = None, + zoom: int = 13, + ) -> None: + """Create the picker. + + Args: + session: The running TUI session (for size + repaint scheduling). + markers: Located mesh nodes to draw for context (may be empty). + source: The vector-tile source (already resolved/warmed). + max_tile_zoom: The source's max zoom, captured off the event loop at open time. + initial: The location to open centred on (the node's current position), or + ``None`` to frame the mesh instead (falling back to a world view when no + nodes are located either). + zoom: The zoom to open at when ``initial`` is given. + """ + saved = (initial[0], initial[1], zoom) if initial is not None else None + super().__init__(session, markers, source, max_tile_zoom, saved_view=saved) + self._initial = initial + self._pick_zoom = zoom + + @property + def footer_hint(self) -> str: # type: ignore[override] + """Key hints for picking, plus the live tile-loading indicator.""" + base = "wasd/↑↓←→ pan (⇧ fine) · +/- zoom · Enter set location · Esc cancel" + if self._pending: + return f"{base} · [muted]loading {len(self._pending)} tiles…[/muted]" + if not self._source.available: + return f"{base} · [warn]offline — no basemap[/warn]" + return base + + def _initial_viewport(self, dot_w: int, dot_h: int) -> Viewport: + """Open on the initial location, else frame the nodes, else a world view.""" + if self._saved_view is None and not self._markers: + return Viewport(20.0, 0.0, 2, dot_w, dot_h) # nothing to frame — the world + return super()._initial_viewport(dot_w, dot_h) + + def render_body(self, width: int) -> list[str]: + """Render the map with the crosshair marker pinned to the view centre. + + The crosshair is a transient marker appended for just this frame (never stored in + :attr:`_markers`), drawn in the "self" style so it reads as *your* position-to-be + and labelled with the live coordinates it would commit. + """ + if self._viewport is None: + # First paint: let the base class establish the viewport so the crosshair can + # ride the centre from the very first frame (the extra render is one-off). + super().render_body(width) + real = self._markers + vp = self._viewport + if vp is not None: + cross = MapMarker( + label=f"⌖ {vp.center_lat:.5f}, {vp.center_lon:.5f}", + lat=vp.center_lat, + lon=vp.center_lon, + is_self=True, + ) + self._markers = real + [cross] + try: + return super().render_body(width) + finally: + self._markers = real + + def _title(self, vp: Viewport) -> str: + """A live status title: the coordinates under the crosshair and the scale.""" + base = super()._title(vp) + scale = base.rsplit("·", 1)[-1].strip() + return f"set location · {vp.center_lat:.5f}, {vp.center_lon:.5f} · z{vp.zoom} · {scale}" + + def handle(self, action: str, data: str = "") -> None: + """Commit the centre on Enter; ``r`` returns to the initial spot; else the map keys.""" + if action == "enter": + vp = self._viewport + if vp is not None: + self.resolve((vp.center_lat, vp.center_lon)) + return + if action == "text" and data.lower() == "r" and self._viewport is not None: + # Reset returns to the *starting* view — the initial location when one was + # given, else the node frame — rather than refitting a cloud that now includes + # nowhere in particular. With neither, fall back to the world view. + vp = self._viewport + if self._initial is not None: + lat, lon = self._initial + self._viewport = Viewport( + clamp_lat(lat), lon, self._pick_zoom, vp.dot_w, vp.dot_h + ) + elif not self._markers: + self._viewport = Viewport(20.0, 0.0, 2, vp.dot_w, vp.dot_h) + else: + super().handle(action, data) + return + self._needs_scrub = True + return + super().handle(action, data) + + +async def pick_location( + ctx: "AppContext", *, initial: Optional[tuple[float, float]] = None +) -> Optional[tuple[float, float]]: + """Open the full-screen map as a coordinate picker; return ``(lat, lon)`` or ``None``. + + Gathers the mesh's located nodes for context (best-effort — an unreachable radio just + means a barer map), warms the tile source off the event loop, then runs a + :class:`LocationPickScreen` until the user commits a spot with Enter or backs out + with Esc. + + Args: + ctx: Shared application context (must be in the interactive menu). + initial: The location to open centred on (e.g. the node's current coordinates), + or ``None`` to frame the mesh. + + Raises: + RuntimeError: If called outside the interactive menu (no full-screen session). + """ + import asyncio + + from ..tools.map import gather_markers + from .surface import TuiUi + from .tui.screen import CANCEL + + if not isinstance(ctx.ui, TuiUi): # pragma: no cover - guarded by the menu-only caller + raise RuntimeError("the interactive map is only available in the menu") + session = ctx.ui.session + try: + markers = await gather_markers(ctx) + except Exception: # noqa: BLE001 - context markers are a nicety, never a requirement + markers = [] + source = basemap_source(ctx) + max_zoom = await asyncio.to_thread(lambda: source.max_zoom) + screen = LocationPickScreen( + session, markers, source, max_zoom, initial=initial + ) + try: + result = await session.run_screen(screen) + finally: + # Same clean-slate repaint as open_map: the braille may have smeared the terminal. + session.request_full_repaint() + return None if result is CANCEL or result is None else result + + def basemap_source(ctx: "AppContext") -> BasemapSource: """Build the vector-tile source backed by the app's on-disk tile cache.""" return BasemapSource(ctx.settings.config_dir / "tilecache") diff --git a/meshterm/ui/menu.py b/meshterm/ui/menu.py index 08d0ae5..f969a7c 100644 --- a/meshterm/ui/menu.py +++ b/meshterm/ui/menu.py @@ -584,7 +584,17 @@ async def _handle_disconnect(ctx: AppContext, session: TuiSession) -> bool: Returns: ``True`` if the user chose to quit; ``False`` once the device reconnected. """ - dialog = ReconnectDialog("Waiting for your device — reconnect it to resume.") + # A drop the config editor announced (it just sent a reboot command) is expected, so + # the dialog says what is actually happening instead of implying an unplug. The flag + # is consumed here so a later, genuine disconnect goes back to the generic wording. + if ctx.reboot_in_progress: + ctx.reboot_in_progress = False + dialog = ReconnectDialog( + "Rebooting — waiting for the device to come back…", + title="Device rebooting", + ) + else: + dialog = ReconnectDialog("Waiting for your device — reconnect it to resume.") dialog.future = asyncio.get_running_loop().create_future() # The session stack was cleared before we were called (see _session_loop), so push a # clean, empty base frame for the popup to float over. A lone floating screen with nothing diff --git a/meshterm/ui/surface.py b/meshterm/ui/surface.py index 4c5fa0a..ccc642a 100644 --- a/meshterm/ui/surface.py +++ b/meshterm/ui/surface.py @@ -64,9 +64,18 @@ class Ui: raise NotImplementedError async def select( - self, title: str, items: list, *, default: Any = None, wrap: bool = True + self, + title: str, + items: list, + *, + prompt: str = "", + default: Any = None, + wrap: bool = True, ) -> Any: - """Prompt the user to choose one item; return its value or ``None`` if cancelled.""" + """Prompt the user to choose one item; return its value or ``None`` if cancelled. + + ``prompt`` draws an instruction inside the popup, above the list. + """ raise NotImplementedError async def select_startup( @@ -124,30 +133,62 @@ class Ui: self, title: str, *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, help_text: str = "", password: bool = False, ) -> Optional[str]: - """Prompt for a line of text; return it or ``None`` if cancelled.""" + """Prompt for a line of text; return it or ``None`` if cancelled. + + Keep ``title`` short (it heads the popup's border) and put the question/instruction + in ``prompt`` (drawn above the field), so a text popup reads like the button dialogs. + """ raise NotImplementedError async def confirm(self, title: str, *, default: bool = True) -> Optional[bool]: """Prompt yes/no; return the answer or ``None`` if cancelled.""" raise NotImplementedError + async def dialog( + self, + prompt: str, + buttons: list[tuple[str, Any]], + *, + title: str = "", + default: int = 0, + keys: Optional[dict[str, Any]] = None, + danger: bool = False, + ) -> Any: + """Show a centered button dialog; return the chosen value or ``None`` on Esc. + + The general choose-one popup (a prompt above a row of buttons). ``danger`` themes + the prompt and border in the cautionary style for destructive or disruptive + choices. Buttons follow the platform-dialog convention: the safe way out sits on + the left and the committing action on the right, which is also the sensible + ``default`` so Enter commits it while Esc always backs out. + """ + raise NotImplementedError + + async def typed_confirm( + self, warning: str, word: str, *, title: str = "Are you sure?" + ) -> bool: + """Gate a destructive action behind typing ``word``; return whether it was typed.""" + raise NotImplementedError + async def autocomplete( self, title: str, choices: list[str], *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, ) -> Optional[str]: """Prompt for free text with suggestions; return it or ``None`` if cancelled.""" raise NotImplementedError - async def path(self, title: str, *, default: str = "") -> Optional[str]: + async def path(self, title: str, *, prompt: str = "", default: str = "") -> Optional[str]: """Prompt for a filesystem path; return it or ``None`` if cancelled.""" raise NotImplementedError @@ -194,7 +235,13 @@ class PlainUi(Ui): return RuntimeError("interactive prompts are only available in the menu") async def select( - self, title: str, items: list, *, default: Any = None, wrap: bool = True + self, + title: str, + items: list, + *, + prompt: str = "", + default: Any = None, + wrap: bool = True, ) -> Any: """Unsupported in scripted CLI mode.""" raise self._no_prompt() @@ -258,6 +305,7 @@ class PlainUi(Ui): self, title: str, *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, help_text: str = "", @@ -266,14 +314,18 @@ class PlainUi(Ui): """Prompt on the terminal (line editor / getpass), re-asking until valid. A few tools (e.g. a remote-admin password) can legitimately prompt from a scripted - run when no flag was supplied, so this stays functional on the CLI. + run when no flag was supplied, so this stays functional on the CLI. An in-body + ``prompt`` (used by the interactive popups) is printed once as a lead-in line here. Returns: The entered string, or ``None`` on EOF / interrupt. """ import getpass - prompt = f"{title} " if not default else f"{title} [{default}] " + if prompt: + self.console.print(prompt) + head = f"{title} " if not default else f"{title} [{default}] " + prompt = head while True: try: raw = getpass.getpass(prompt) if password else input(prompt) @@ -291,18 +343,38 @@ class PlainUi(Ui): """Unsupported in scripted CLI mode.""" raise self._no_prompt() + async def dialog( + self, + prompt: str, + buttons: list[tuple[str, Any]], + *, + title: str = "", + default: int = 0, + keys: Optional[dict[str, Any]] = None, + danger: bool = False, + ) -> Any: + """Unsupported in scripted CLI mode.""" + raise self._no_prompt() + + async def typed_confirm( + self, warning: str, word: str, *, title: str = "Are you sure?" + ) -> bool: + """Unsupported in scripted CLI mode (destructive CLI commands gate on ``--yes``).""" + raise self._no_prompt() + async def autocomplete( self, title: str, choices: list[str], *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, ) -> Optional[str]: """Unsupported in scripted CLI mode.""" raise self._no_prompt() - async def path(self, title: str, *, default: str = "") -> Optional[str]: + async def path(self, title: str, *, prompt: str = "", default: str = "") -> Optional[str]: """Unsupported in scripted CLI mode.""" raise self._no_prompt() @@ -365,10 +437,18 @@ class TuiUi(Ui): # --- input --------------------------------------------------------------- async def select( - self, title: str, items: list, *, default: Any = None, wrap: bool = True + self, + title: str, + items: list, + *, + prompt: str = "", + default: Any = None, + wrap: bool = True, ) -> Any: """Delegate to the session's select screen.""" - return await self.session.select(title, items, default=default, wrap=wrap) + return await self.session.select( + title, items, prompt=prompt, default=default, wrap=wrap + ) async def select_startup( self, @@ -433,6 +513,7 @@ class TuiUi(Ui): self, title: str, *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, help_text: str = "", @@ -441,6 +522,7 @@ class TuiUi(Ui): """Delegate to the session's text screen.""" return await self.session.text( title, + prompt=prompt, default=default, validate=validate, help_text=help_text, @@ -451,19 +533,49 @@ class TuiUi(Ui): """Delegate to the session's confirm screen.""" return await self.session.confirm(title, default=default) + async def dialog( + self, + prompt: str, + buttons: list[tuple[str, Any]], + *, + title: str = "", + default: int = 0, + keys: Optional[dict[str, Any]] = None, + danger: bool = False, + ) -> Any: + """Delegate to the session's button dialog, themed cautionary when ``danger``.""" + return await self.session.button_dialog( + prompt, + buttons, + title=title, + default=default, + keys=keys, + prompt_style="warn" if danger else "", + border_style="warn" if danger else "accent", + ) + + async def typed_confirm( + self, warning: str, word: str, *, title: str = "Are you sure?" + ) -> bool: + """Delegate to the session's typed-confirmation dialog.""" + return await self.session.typed_confirm(warning, word, title=title) + async def autocomplete( self, title: str, choices: list[str], *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, ) -> Optional[str]: """Delegate to the session's autocomplete screen.""" return await self.session.autocomplete( - title, choices, default=default, validate=validate + title, choices, prompt=prompt, default=default, validate=validate ) - async def path(self, title: str, *, default: str = "") -> Optional[str]: + async def path(self, title: str, *, prompt: str = "", default: str = "") -> Optional[str]: """Prompt for a path as free text (with the current value prefilled).""" - return await self.session.text(title, default=default, help_text="filesystem path") + return await self.session.text( + title, prompt=prompt, default=default, help_text="filesystem path" + ) diff --git a/meshterm/ui/tui/__init__.py b/meshterm/ui/tui/__init__.py index 7241c54..5301f61 100644 --- a/meshterm/ui/tui/__init__.py +++ b/meshterm/ui/tui/__init__.py @@ -16,7 +16,13 @@ from __future__ import annotations from .overlay import BusyOverlay from .progress import ProgressScreen -from .prompt import AutocompleteScreen, ConfirmScreen, ReconnectDialog, TextScreen +from .prompt import ( + AutocompleteScreen, + ConfirmScreen, + ReconnectDialog, + TextScreen, + TypedConfirmDialog, +) from .ring import RingSpinner from .screen import CANCEL, BusyScreen, Screen, ScrollScreen from .select import Choice, SelectScreen, Separator @@ -35,6 +41,7 @@ __all__ = [ "ConfirmScreen", "AutocompleteScreen", "ReconnectDialog", + "TypedConfirmDialog", "ProgressScreen", "Spinner", "RingSpinner", diff --git a/meshterm/ui/tui/prompt.py b/meshterm/ui/tui/prompt.py index 1406219..9ab63c0 100644 --- a/meshterm/ui/tui/prompt.py +++ b/meshterm/ui/tui/prompt.py @@ -181,6 +181,7 @@ class TextScreen(Screen): self, title: str, *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, help_text: str = "", @@ -190,7 +191,10 @@ class TextScreen(Screen): """Build a text prompt. Args: - title: The question shown above the field. + title: Short heading shown in the dialog's border. + prompt: The question/instruction shown inside the box, above the field. Keep + the ``title`` short and put the detail here, so the popup reads like the + button dialogs (a prompt above its controls) rather than a lone field. default: Prefilled text. validate: Optional validator run on Enter; a returned string is shown as an error and blocks submission. @@ -201,15 +205,38 @@ class TextScreen(Screen): super().__init__() self.title = title self.footer_hint = footer_hint + self._prompt = prompt self._editor = _LineEditor(default) self._validate = validate self._help = help_text self._password = password self._error = "" + @property + def dialog_width(self) -> int: + """Natural outer width so the frame sizes the box to its content (see ButtonDialog). + + The widest of the prompt, title, hint, footer, and the field's current text, plus a + comfortable field minimum — so a short prompt is a tidy box, not a banner stretched + across the terminal. The compositor still caps this to the width available. + """ + inner = max( + cell_len(self._prompt), + cell_len(self.title), + cell_len(self._help), + cell_len(self.footer_hint), + cell_len(self._editor.text) + 4, # the field plus its cursor and a little slack + 36, # a comfortable minimum so a short field isn't a cramped sliver + ) + return inner + 8 # panel padding + border, plus horizontal breathing room + def render_body(self, width: int) -> list[str]: - """Render the field, any help text, and the current validation error.""" - parts: list[RenderableType] = [self._editor.render(mask=self._password)] + """Render the optional prompt, the field, any help text, and a validation error.""" + parts: list[RenderableType] = [] + if self._prompt: + parts.append(Text(self._prompt)) + parts.append(Text("")) + parts.append(self._editor.render(mask=self._password)) if self._help: parts.append(Text(self._help, style="muted")) if self._error: @@ -319,6 +346,16 @@ class ConfirmScreen(Screen): self.footer_hint = footer_hint self._value = default + @property + def dialog_width(self) -> int: + """Natural outer width so the box hugs the question rather than stretching wide.""" + inner = max( + cell_len(self.title), + cell_len(self.footer_hint), + len(" Yes No "), + ) + return inner + 8 + def render_body(self, width: int) -> list[str]: """Render the Yes / No options with the current choice highlighted.""" text = Text() @@ -438,6 +475,79 @@ class ButtonDialog(Screen): super().handle("escape") +class TypedConfirmDialog(Screen): + """A destructive-action gate: the user must type a confirmation word to proceed. + + A centered, error-themed dialog for the actions a stray Enter must never be able to + trigger (factory reset, identity overwrite). It shows the warning, then a field the + user has to type the exact confirmation word into; Enter commits only when the field + matches (anything else shows a nudge and keeps the dialog up), and Esc always backs + out. Matching is case-insensitive so a forgotten CapsLock doesn't read as hesitation — + the friction is having to *type the word*, not its case. Resolves ``True`` on a match, + or :data:`~meshterm.ui.tui.screen.CANCEL` on Esc (never ``False``). + """ + + border_style = "err" + footer_hint = "Enter confirm · Esc cancel" + + def __init__(self, warning: str, word: str, *, title: str = "Are you sure?") -> None: + """Build the typed-confirmation dialog. + + Args: + warning: The consequence, spelled out (e.g. "This erases ALL data …"). + word: The word the user must type to confirm (e.g. ``"RESET"``). + title: Dialog heading. + """ + super().__init__() + self.title = title + self._warning = warning + self._word = word + self._editor = _LineEditor("") + self._error = "" + + @property + def dialog_width(self) -> int: + """Natural outer width so the frame sizes the box to its content (see ButtonDialog).""" + inner = max( + cell_len(self._warning), + cell_len(self.title), + cell_len(self.footer_hint), + cell_len(self._ask), + ) + return min(inner, 72) + 12 # panel padding + border, plus breathing room + + @property + def _ask(self) -> str: + """The instruction line naming the word to type.""" + return f"Type {self._word} to confirm:" + + def render_body(self, width: int) -> list[str]: + """Render the warning, the instruction, the typed field, and any mismatch nudge.""" + parts: list[RenderableType] = [ + Text(self._warning, style="err"), + Text(""), + Text(self._ask, style="muted"), + self._editor.render(), + ] + if self._error: + parts.append(Text(self._error, style="err")) + return render_lines(Group(*parts), width) + + def handle(self, action: str, data: str = "") -> None: + """Edit the field, confirm on an exact match, or cancel on Esc.""" + if action == "enter": + if self._editor.text.strip().lower() == self._word.lower(): + self.resolve(True) + else: + self._error = f"That doesn't match — type {self._word}, or press Esc to cancel." + return + if action == "escape": + super().handle("escape") + return + if self._editor.edit(action, data): + self._error = "" + + class ReconnectDialog(Screen): """A centered dialog shown when the companion link drops mid-session. @@ -528,6 +638,7 @@ class AutocompleteScreen(Screen): title: str, choices: list[str], *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, footer_hint: str = "type · ↑↓ Tab complete · Enter accept · Esc cancel", @@ -535,8 +646,9 @@ class AutocompleteScreen(Screen): """Build an autocomplete prompt. Args: - title: The question shown above the field. + title: Short heading shown in the dialog's border. choices: Suggestion strings to match against. + prompt: The instruction shown inside the box, above the field. default: Prefilled text. validate: Optional validator run on Enter. footer_hint: Footer key hint. @@ -544,12 +656,26 @@ class AutocompleteScreen(Screen): super().__init__() self.title = title self.footer_hint = footer_hint + self._prompt = prompt self._editor = _LineEditor(default) self._choices = choices self._validate = validate self._error = "" self._sugg = 0 + @property + def dialog_width(self) -> int: + """Natural outer width so the box fits its prompt/suggestions, not the whole screen.""" + widths = [ + cell_len(self._prompt), + cell_len(self.title), + cell_len(self.footer_hint), + cell_len(self._editor.text) + 4, + 36, + ] + widths.extend(cell_len(c) + 2 for c in self._choices[: self.MAX_SUGGESTIONS]) + return max(widths) + 8 + def _suggestions(self) -> list[str]: """Return suggestions matching the current text, capped for display.""" needle = self._editor.text.lower() @@ -560,8 +686,12 @@ class AutocompleteScreen(Screen): return matches[: self.MAX_SUGGESTIONS] def render_body(self, width: int) -> list[str]: - """Render the field, the matching suggestions, and any error.""" - parts: list[RenderableType] = [self._editor.render()] + """Render the optional prompt, the field, the matching suggestions, and any error.""" + parts: list[RenderableType] = [] + if self._prompt: + parts.append(Text(self._prompt)) + parts.append(Text("")) + parts.append(self._editor.render()) suggestions = self._suggestions() self._sugg = max(0, min(self._sugg, len(suggestions) - 1)) if suggestions else 0 for i, sug in enumerate(suggestions): diff --git a/meshterm/ui/tui/select.py b/meshterm/ui/tui/select.py index e30ef01..223d271 100644 --- a/meshterm/ui/tui/select.py +++ b/meshterm/ui/tui/select.py @@ -11,9 +11,10 @@ from __future__ import annotations from dataclasses import dataclass from typing import Any, Callable, Optional, Union +from rich.cells import cell_len from rich.text import Text -from .render import render_to_ansi +from .render import render_lines, render_to_ansi from .screen import Screen @@ -69,6 +70,7 @@ class SelectScreen(Screen): title: str, items: list, *, + prompt: str = "", default: Any = None, footer_hint: str = "↑↓ move · type to filter · Enter select · Esc back", filterable: bool = True, @@ -77,8 +79,10 @@ class SelectScreen(Screen): """Build a select screen. Args: - title: Heading shown above the list. + title: Short heading shown in the border. items: A list of :class:`Choice` and :class:`Separator` in display order. + prompt: An optional instruction shown inside the box, above the list — so a + floating select reads like the other dialogs (a prompt above its controls). default: A choice value to pre-highlight, if present. footer_hint: Footer key hint. filterable: Whether typing narrows the list. Off for short, fixed lists (e.g. @@ -90,6 +94,7 @@ class SelectScreen(Screen): super().__init__() self.title = title self.footer_hint = footer_hint + self._prompt = prompt self._filterable = filterable self._wrap = wrap self._items = items @@ -171,14 +176,37 @@ class SelectScreen(Screen): # --- rendering ----------------------------------------------------------- + @property + def dialog_width(self) -> int: + """Natural outer width so a floating select hugs its widest row, not the terminal. + + The widest of the prompt, title, footer, and every row (with room for the pointer), + so a short menu is a tidy popup rather than a full-width banner. The compositor still + caps this to the space available, and this is only read for a *floating* select — the + full-screen base menu is laid out by ``compose_base`` and ignores it. + """ + widths = [cell_len(self.title), cell_len(self.footer_hint), cell_len(self._prompt)] + for item in self._items: + label = item.title if isinstance(item, Separator) else _plain(item.label) + widths.append(cell_len(label) + 2) # + the "❯ " / " " pointer column + return max(widths, default=20) + 8 + def render_body(self, width: int) -> list[str]: - """Render each row to a single ANSI line, the highlighted choice marked.""" + """Render the optional prompt then each row as one ANSI line, the choice marked.""" rows = self._rows() choices = self._choices(rows) self._index = max(0, min(self._index, len(choices) - 1)) if choices else 0 selected = choices[self._index] if choices else None lines: list[str] = [] + # A prompt (when set) sits above the list, offsetting every row below it; the cursor + # line and sticky-header indices below are shifted by exactly this many lines. + prefix = 0 + if self._prompt: + plines = render_lines(Text(self._prompt), width) + lines.extend(plines) + lines.append("") + prefix = len(plines) + 1 # Record each section heading as a sticky-header candidate, so one that scrolls off is # re-pinned to the top row by the base Screen.sticky_header. Filtering flattens the list # to bare choices (see _rows), so this stays empty then and nothing is pinned. @@ -207,8 +235,10 @@ class SelectScreen(Screen): lines.append(render_to_ansi(text, width)) if not choices: lines.append(render_to_ansi(Text("(no matches)", style="muted"), width)) - # Remember where the highlighted row landed so the session can keep it in view. - self._cursor = _cursor_line(rows, selected, bool(self._filter)) + # Remember where the highlighted row landed so the session can keep it in view, + # shifted past any prompt lines drawn above the list. + base_cursor = _cursor_line(rows, selected, bool(self._filter)) + self._cursor = None if base_cursor is None else base_cursor + prefix return lines def cursor_line(self) -> Optional[int]: diff --git a/meshterm/ui/tui/session.py b/meshterm/ui/tui/session.py index 8147139..adb7fda 100644 --- a/meshterm/ui/tui/session.py +++ b/meshterm/ui/tui/session.py @@ -34,6 +34,7 @@ from .prompt import ( ConfirmScreen, PinDialog, TextScreen, + TypedConfirmDialog, Validator, ) from .screen import CANCEL, BusyScreen, Screen, ScrollScreen @@ -219,6 +220,7 @@ class TuiSession: title: str, items: list, *, + prompt: str = "", default: Any = None, wrap: bool = True, filterable: bool = True, @@ -226,15 +228,16 @@ class TuiSession: ) -> Any: """Show a select screen; return the chosen value or ``None`` if cancelled. - ``filterable`` and ``footer_hint`` are forwarded to the screen for short, fixed - lists (a yes-or-no style choice) that want no type-to-filter and a tailored hint. + ``prompt`` draws an instruction inside the box above the list; ``filterable`` and + ``footer_hint`` are forwarded for short, fixed lists (a yes-or-no style choice) that + want no type-to-filter and a tailored hint. """ - screen = ( - SelectScreen(title, items, default=default, wrap=wrap, filterable=filterable, footer_hint=footer_hint) - if footer_hint is not None - else SelectScreen(title, items, default=default, wrap=wrap, filterable=filterable) + kwargs: dict[str, Any] = dict( + prompt=prompt, default=default, wrap=wrap, filterable=filterable ) - result = await self.run_screen(screen) + if footer_hint is not None: + kwargs["footer_hint"] = footer_hint + result = await self.run_screen(SelectScreen(title, items, **kwargs)) return None if result is CANCEL else result async def select_startup( @@ -376,6 +379,7 @@ class TuiSession: self, title: str, *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, help_text: str = "", @@ -385,6 +389,7 @@ class TuiSession: result = await self.run_screen( TextScreen( title, + prompt=prompt, default=default, validate=validate, help_text=help_text, @@ -434,17 +439,32 @@ class TuiSession: result = await self.run_screen(screen) return None if result is CANCEL else result + async def typed_confirm( + self, warning: str, word: str, *, title: str = "Are you sure?" + ) -> bool: + """Gate a destructive action behind typing ``word``; return whether it was typed. + + Shows the error-themed :class:`~meshterm.ui.tui.prompt.TypedConfirmDialog` and + collapses its result to a plain bool: ``True`` only when the user typed the word, + ``False`` when they backed out with Esc. + """ + result = await self.run_screen(TypedConfirmDialog(warning, word, title=title)) + return result is True + async def autocomplete( self, title: str, choices: list[str], *, + prompt: str = "", default: str = "", validate: Optional[Validator] = None, ) -> Optional[str]: """Show a free-text prompt with suggestions; return text or ``None`` if cancelled.""" result = await self.run_screen( - AutocompleteScreen(title, choices, default=default, validate=validate) + AutocompleteScreen( + title, choices, prompt=prompt, default=default, validate=validate + ) ) return None if result is CANCEL else result diff --git a/tests/test_config_editor.py b/tests/test_config_editor.py new file mode 100644 index 0000000..a46d1b4 --- /dev/null +++ b/tests/test_config_editor.py @@ -0,0 +1,559 @@ +"""Tests for the overhauled config editor: dialogs, staging, and the location picker. + +The pure pieces (the contact share URL, coordinate parsing, the typed-confirmation +dialog, the map picker screen) are exercised directly; the editor's flows run end-to-end +against the :class:`MockDevice` simulator through a scripted fake UI surface that answers +each prompt from a queue. +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Optional + +import pytest +from rich.console import Console + +from meshterm.context import AppContext +from meshterm.core.admin_store import AdminStore +from meshterm.core.config import Settings +from meshterm.core.device_store import DeviceStore +from meshterm.persistence.repository import Repository +from meshterm.ui.config_editor import ( + _parse_coords, + _valid_coords, + contact_share_url, + edit_config, +) +from meshterm.ui.tui.prompt import TypedConfirmDialog +from meshterm.ui.tui.screen import CANCEL + +_ANSI = re.compile(r"\x1b\[[0-9;]*m") + + +def _plain(lines: list[str]) -> str: + """Strip ANSI escapes and join rendered lines for content assertions.""" + return _ANSI.sub("", "\n".join(lines)) + + +class _Fut: + """A minimal future stand-in so screens can resolve without an event loop.""" + + def __init__(self) -> None: + self.value: Any = None + self._done = False + + def done(self) -> bool: + return self._done + + def set_result(self, value: Any) -> None: + self.value = value + self._done = True + + +# -- contact share URL ---------------------------------------------------------- + + +def test_contact_share_url_matches_meshcore_format() -> None: + """The contact card URL carries the encoded name, hex key, and node type.""" + url = contact_share_url("Base Camp", "AB" * 32, node_type=2) + assert url.startswith("meshcore://contact/add?") + assert "name=Base%20Camp" in url + assert f"public_key={'ab' * 32}" in url # key is normalized to lowercase + assert url.endswith("&type=2") + + +def test_contact_share_url_defaults_to_companion_type() -> None: + """With no explicit type the card marks the node as a companion (type 1).""" + assert contact_share_url("n", "00" * 32).endswith("&type=1") + + +# -- coordinate parsing ---------------------------------------------------------- + + +def test_parse_coords_accepts_comma_and_space_pairs() -> None: + """A typed pair parses with a comma, a space, or both.""" + assert _parse_coords("45.5, -73.6") == (45.5, -73.6) + assert _parse_coords("45.5 -73.6") == (45.5, -73.6) + assert _parse_coords(" 45.5 , -73.6 ") == (45.5, -73.6) + + +def test_parse_coords_rejects_garbage_and_out_of_range() -> None: + """Wrong shape or out-of-range degrees surface as validation messages.""" + assert _valid_coords("45.5") != True # noqa: E712 - message, not False + assert _valid_coords("45.5, -73.6, 7") != True # noqa: E712 + assert _valid_coords("91.0, 0") != True # noqa: E712 - latitude out of range + assert _valid_coords("0, 181") != True # noqa: E712 - longitude out of range + assert _valid_coords("here, there") != True # noqa: E712 + assert _valid_coords("45.5, -73.6") is True + + +# -- the typed-confirmation dialog ------------------------------------------------ + + +def test_typed_confirm_requires_the_exact_word() -> None: + """Enter only commits once the confirmation word has been typed.""" + dialog = TypedConfirmDialog("This erases everything.", "RESET") + dialog.future = _Fut() + + dialog.handle("enter") # empty field: nudge, no resolve + assert not dialog.future.done() + assert "doesn't match" in _plain(dialog.render_body(60)) + + for ch in "nope": + dialog.handle("text", ch) + dialog.handle("enter") + assert not dialog.future.done() + + for _ in "nope": + dialog.handle("backspace") + for ch in "RESET": + dialog.handle("text", ch) + dialog.handle("enter") + assert dialog.future.done() and dialog.future.value is True + + +def test_typed_confirm_is_case_insensitive_and_cancellable() -> None: + """A lowercase match still confirms (the friction is the word); Esc cancels.""" + dialog = TypedConfirmDialog("Danger.", "IMPORT") + dialog.future = _Fut() + for ch in "import": + dialog.handle("text", ch) + dialog.handle("enter") + assert dialog.future.value is True + + cancelled = TypedConfirmDialog("Danger.", "IMPORT") + cancelled.future = _Fut() + cancelled.handle("escape") + assert cancelled.future.value is CANCEL + + +def test_typed_confirm_renders_warning_and_instruction() -> None: + """The dialog shows the consequence and names the word to type.""" + dialog = TypedConfirmDialog("This erases ALL data.", "RESET", title="⚠ Factory reset") + body = _plain(dialog.render_body(60)) + assert "This erases ALL data." in body + assert "Type RESET to confirm" in body + assert dialog.border_style == "err" + + +# -- the map location picker ------------------------------------------------------- + + +class _StubSession: + """Minimal stand-in for :class:`TuiSession` for driving the picker screen.""" + + def __init__(self, cols: int = 80, rows: int = 24) -> None: + self._cols, self._rows = cols, rows + + def base_body_size(self) -> tuple[int, int]: + return self._cols, self._rows + + def invalidate(self) -> None: + pass + + +class _StubSource: + """An always-offline tile source, so the picker renders nodes only (no network).""" + + available = False + max_zoom = 14 + + def load_tile(self, z: int, x: int, y: int): + return None + + +def _picker(markers=None, initial=None): + from meshterm.ui.map_screen import LocationPickScreen + + screen = LocationPickScreen( + _StubSession(), markers or [], _StubSource(), 14, initial=initial + ) + screen.future = _Fut() + return screen + + +def test_location_picker_opens_on_the_initial_spot_and_commits_the_centre() -> None: + """The picker centres on the node's current location and Enter resolves it.""" + screen = _picker(initial=(45.5, -73.6)) + screen.render_body(80) + assert screen._viewport.center_lat == pytest.approx(45.5) + assert screen._viewport.center_lon == pytest.approx(-73.6) + assert screen._viewport.zoom == 13 + + screen.handle("enter") + lat, lon = screen.future.value + assert lat == pytest.approx(45.5) and lon == pytest.approx(-73.6) + + +def test_location_picker_moves_the_crosshair_and_resets_to_initial() -> None: + """Panning moves the committed point; ``r`` returns to the starting location.""" + screen = _picker(initial=(45.5, -73.6)) + screen.render_body(80) + screen.handle("text", "d") # pan east + screen.handle("text", "s") # pan south + screen.handle("enter") + lat, lon = screen.future.value + assert lon > -73.6 and lat < 45.5 + + screen.handle("text", "r") # back to the initial spot + assert screen._viewport.center_lat == pytest.approx(45.5) + assert screen._viewport.center_lon == pytest.approx(-73.6) + + +def test_location_picker_draws_a_live_crosshair_without_keeping_it() -> None: + """The centre crosshair (with live coordinates) renders each frame but is transient.""" + from meshterm.ui.map_render import MapMarker + + markers = [MapMarker("Repeater", 45.51, -73.61, is_repeater=True)] + screen = _picker(markers=markers, initial=(45.5, -73.6)) + body = _plain(screen.render_body(80)) + assert "⌖" in body and "45.5" in body # crosshair + its live coordinates + assert screen._markers == markers # not accumulated across frames + assert "set location" in screen._title(screen._viewport) + + +def test_location_picker_without_nodes_or_initial_shows_the_world() -> None: + """With nothing to frame the picker opens on a world view rather than crashing.""" + screen = _picker() + screen.render_body(80) + assert screen._viewport.zoom == 2 + screen.handle("text", "r") # reset with nothing to fit stays on the world view + assert screen._viewport.zoom == 2 + screen.handle("escape") + assert screen.future.value is None + + +# -- the editor's flows against the simulator --------------------------------------- + + +class _FakeSession: + """A minimal stand-in for the TUI session behind the editor's persistent menu. + + The editor keeps its main menu *pushed on the session stack* (so sub-prompts float + over it as modal popups). Here, pushing the menu resolves its future straight from the + script's next ``select`` answer — the same FIFO the sub-prompts draw from — so the + modal-backdrop control flow is exercised without a real full-screen session. + """ + + def __init__(self, ui: "_ScriptedUi") -> None: + self.ui = ui + self.pushed: list = [] + + def push(self, screen: Any) -> None: + from meshterm.ui.tui.screen import CANCEL + + self.pushed.append(screen) + value = self.ui._answer("select") + screen.resolve(CANCEL if value is None else value) + + def pop(self, screen: Any = None) -> None: + pass + + +class _ScriptedUi: + """A fake UI surface answering every prompt from a FIFO script. + + Each entry is ``(method, answer)``; the method name is asserted so a flow that asks + an unexpected question fails loudly instead of silently mis-answering. It exposes a + :class:`_FakeSession` so the editor's push-a-persistent-menu path works. + """ + + def __init__(self, script: list[tuple[str, Any]]) -> None: + self.script = list(script) + self.notes: list[str] = [] + self.views: list[str] = [] + self.presented: list[str] = [] + self.session = _FakeSession(self) + + def _answer(self, method: str) -> Any: + assert self.script, f"unexpected prompt: {method}" + expected, value = self.script.pop(0) + assert expected == method, f"expected {expected} prompt, got {method}" + return value + + async def select(self, title: str, items: list, **kwargs: Any) -> Any: + return self._answer("select") + + async def dialog(self, prompt: str, buttons: list, **kwargs: Any) -> Any: + return self._answer("dialog") + + async def typed_confirm(self, warning: str, word: str, **kwargs: Any) -> bool: + return self._answer("typed_confirm") + + async def text(self, title: str, **kwargs: Any) -> Optional[str]: + return self._answer("text") + + async def path(self, title: str, **kwargs: Any) -> Optional[str]: + return self._answer("path") + + async def autocomplete(self, title: str, choices: list, **kwargs: Any) -> Optional[str]: + return self._answer("autocomplete") + + async def confirm(self, title: str, **kwargs: Any) -> Optional[bool]: + return self._answer("confirm") + + async def view(self, renderable: Any, **kwargs: Any) -> None: + self.views.append(str(kwargs.get("title", ""))) + + def note(self, markup: str) -> None: + self.notes.append(markup) + + def show(self, *renderables: Any) -> None: + pass + + async def present(self, *, title: str = "") -> None: + self.presented.append(title) + + def discard(self) -> None: + pass + + +@pytest.fixture() +def ctx(tmp_path: Path) -> AppContext: + """A mock-backed application context whose UI is swapped per test.""" + settings = Settings(config_dir=tmp_path, db_path=tmp_path / "cfg.db") + context = AppContext( + console=Console(file=__import__("io").StringIO()), + settings=settings, + repo=Repository(settings.db_path), + device_store=DeviceStore(tmp_path / "devices.json"), + admin_store=AdminStore(tmp_path / "admin.json"), + mock=True, + ) + yield context + context.repo.close() + + +def _install(ctx: AppContext, script: list[tuple[str, Any]]) -> _ScriptedUi: + """Install a scripted UI on the context and return it.""" + ui = _ScriptedUi(script) + ctx.ui = ui # type: ignore[assignment] + return ui + + +async def test_editor_stages_a_setting_and_returns_ops_on_apply(ctx: AppContext) -> None: + """Editing a value stages it; Apply returns the set operations for the tool to run.""" + _install(ctx, [ + ("select", "name"), + ("text", "NewName"), + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ops == [("set", "name", "NewName")] + + +async def test_editor_restaging_the_current_value_clears_the_stage(ctx: AppContext) -> None: + """Typing the device's existing value back un-stages the row (nothing to apply).""" + device = await ctx.device() + current = (await device.get_self_info())["name"] + _install(ctx, [ + ("select", "name"), + ("text", "Changed"), + ("select", "name"), + ("text", str(current)), # back to what the device already has + ("select", "__cancel__"), # nothing staged now — closes without a discard dialog + ]) + assert await edit_config(ctx) is None + + +async def test_editor_asks_before_discarding_staged_changes(ctx: AppContext) -> None: + """Cancelling with staged changes confirms; Keep editing returns to the menu.""" + _install(ctx, [ + ("select", "name"), + ("text", "NewName"), + ("select", "__cancel__"), + ("dialog", "keep"), # changed my mind — keep editing + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ops == [("set", "name", "NewName")] + + +async def test_editor_discards_staged_changes_when_confirmed(ctx: AppContext) -> None: + """Cancelling with staged changes and confirming the discard applies nothing.""" + _install(ctx, [ + ("select", "name"), + ("text", "NewName"), + ("select", None), # Esc from the main menu + ("dialog", "discard"), + ]) + assert await edit_config(ctx) is None + + +async def test_editor_clean_close_needs_no_confirmation(ctx: AppContext) -> None: + """With nothing staged, Close leaves immediately — no dialog in the script.""" + _install(ctx, [("select", "__cancel__")]) + assert await edit_config(ctx) is None + + +async def test_editor_location_typed_coordinates_stage_both_axes(ctx: AppContext) -> None: + """Typing a coordinate pair stages latitude and longitude together.""" + _install(ctx, [ + ("select", "__location__"), + ("dialog", "type"), + ("text", "45.5, -73.6"), + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ("set", "adv_lat", 45.5) in ops + assert ("set", "adv_lon", -73.6) in ops + + +async def test_editor_location_map_pick_stages_rounded_coordinates( + ctx: AppContext, monkeypatch: pytest.MonkeyPatch +) -> None: + """Picking on the map stages the chosen point, rounded to advert precision.""" + async def _fake_pick(_ctx: AppContext, *, initial=None): + assert initial is None # the mock device has no fix (0, 0) + return (45.51234567, -73.65432109) + + monkeypatch.setattr("meshterm.ui.map_screen.pick_location", _fake_pick) + _install(ctx, [ + ("select", "__location__"), + ("dialog", "map"), + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ("set", "adv_lat", 45.512346) in ops + assert ("set", "adv_lon", -73.654321) in ops + + +async def test_editor_location_clear_stages_the_no_fix_pair(ctx: AppContext) -> None: + """Clear stages 0, 0 — MeshCore's "no fix" value — for both axes.""" + device = await ctx.device() + await device.set_coords(45.5, -73.6) # give the device a position to clear + _install(ctx, [ + ("select", "__location__"), + ("dialog", "clear"), + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ("set", "adv_lat", 0.0) in ops + assert ("set", "adv_lon", 0.0) in ops + + +async def test_editor_advert_zero_hop_and_flood_run_immediately(ctx: AppContext) -> None: + """The advert dialog sends immediately: zero-hop by default, flood when chosen.""" + device = await ctx.device() + sent: list[bool] = [] + + async def _record(flood: bool = False) -> None: + sent.append(flood) + + device.send_advert = _record # type: ignore[method-assign] + ui = _install(ctx, [ + ("select", "__advert__"), + ("select", "zero"), + ("select", "__advert__"), + ("select", "flood"), + ("select", "__cancel__"), + ]) + assert await edit_config(ctx) is None + assert sent == [False, True] + assert any("zero-hop" in n for n in ui.notes) + assert any("flood" in n for n in ui.notes) + + +async def test_editor_advert_share_shows_the_contact_card(ctx: AppContext) -> None: + """The share option opens the QR/URI view built from the node's own identity.""" + ui = _install(ctx, [ + ("select", "__advert__"), + ("select", "share"), + ("select", "__cancel__"), + ]) + assert await edit_config(ctx) is None + assert any(v.startswith("Share ") for v in ui.views) + + +async def test_editor_factory_reset_gates_on_typed_confirmation(ctx: AppContext) -> None: + """Factory reset only runs behind the typed dialog, and drops staged changes.""" + device = await ctx.device() + await device.set_custom_var("mode", "test") + + # Declined: the device is untouched. + _install(ctx, [ + ("select", "__reset__"), + ("typed_confirm", False), + ("select", "__cancel__"), + ]) + assert await edit_config(ctx) is None + assert await device.get_custom_vars() == {"mode": "test"} + + # Confirmed: the reset runs and previously staged changes are cleared with it. + _install(ctx, [ + ("select", "name"), + ("text", "Doomed"), + ("typed_confirm", True), + ("select", "__cancel__"), # clean close: the staged change died with the reset + ]) + ui = ctx.ui + ui.script.insert(2, ("select", "__reset__")) # reset after staging the name + assert await edit_config(ctx) is None + assert await device.get_custom_vars() == {} + + +async def test_editor_import_key_gates_on_typed_confirmation(ctx: AppContext) -> None: + """Importing an identity key requires the typed IMPORT confirmation.""" + device = await ctx.device() + new_key = "ab" * 32 + _install(ctx, [ + ("select", "__identity_key__"), + ("select", "import"), + ("text", new_key), + ("typed_confirm", True), + ("select", "__cancel__"), + ]) + assert await edit_config(ctx) is None + assert await device.export_private_key() == new_key + + +async def test_editor_reboot_on_simulator_stays_in_the_editor(ctx: AppContext) -> None: + """The simulator has no link to drop, so a confirmed reboot just notes and returns.""" + ui = _install(ctx, [ + ("select", "__reboot__"), + ("dialog", "reboot"), + ("select", "__cancel__"), + ]) + assert await edit_config(ctx) is None + assert ctx.reboot_in_progress is False + assert any("rebooting" in n for n in ui.notes) + + +async def test_editor_backup_writes_immediately(ctx: AppContext, tmp_path: Path) -> None: + """Backup is a device action: the TOML lands on disk before the editor closes.""" + target = tmp_path / "out" / "backup.toml" + ui = _install(ctx, [ + ("select", "__backup__"), + ("path", str(target)), + ("select", "__cancel__"), + ]) + assert await edit_config(ctx) is None + assert target.exists() + assert any("wrote" in n for n in ui.notes) + + +async def test_editor_bool_prompt_uses_the_button_dialog(ctx: AppContext) -> None: + """A boolean setting is toggled through the On/Off dialog and staged.""" + _install(ctx, [ + ("select", "manual_add_contacts"), + ("dialog", True), + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ops == [("set", "manual_add_contacts", True)] + + +async def test_editor_custom_var_suggests_known_names(ctx: AppContext) -> None: + """With existing custom vars the name prompt offers them as suggestions.""" + device = await ctx.device() + await device.set_custom_var("existing", "1") + _install(ctx, [ + ("select", "__custom__"), + ("autocomplete", "existing"), + ("text", "2"), + ("select", "__apply__"), + ]) + ops = await edit_config(ctx) + assert ops == [("set_custom", "existing", "2")] diff --git a/tests/test_map.py b/tests/test_map.py index 5829893..450ac77 100644 --- a/tests/test_map.py +++ b/tests/test_map.py @@ -398,7 +398,7 @@ async def test_map_tool_reports_nothing_to_plot(ctx, monkeypatch) -> None: async def _no_contacts(_ctx): return [] - monkeypatch.setattr(MapTool, "_contacts", staticmethod(_no_contacts)) + monkeypatch.setattr("meshterm.tools.map._contacts", _no_contacts) result = await MapTool().run(ctx, {"static": True, "basemap": False}) assert result.summary == {"located": 0}