Overhaul the config editor into a modal-dialog experience

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 <noreply@anthropic.com>
This commit is contained in:
Jean-Pierre MartineauandClaude Opus 4.8 committed 2026-07-09 22:28:42 -04:00
1 parent 62f19da405
commit 99643f52d9
13 files changed
+1832 -264

No files matched your search

+5
View File
@@ -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
)
+34 -6
View File
@@ -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()
+84 -76
View File
@@ -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
+647 -147
View File
@@ -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 --------------------------------------------------------------
+159
View File
@@ -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")
+11 -1
View File
@@ -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
+125 -13
View File
@@ -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"
)
+8 -1
View File
@@ -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",
+136 -6
View File
@@ -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):
+35 -5
View File
@@ -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]:
+28 -8
View File
@@ -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
+559
View File
@@ -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")]
+1 -1
View File
@@ -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}