Jean-Pierre MartineauandClaude Opus 5
ea9b7c1cb1
Two rows renamed, and the rename carried into the screens they open — the config editor's table title and its staged-count header, the support page's own title (it comes from the tool, so it followed on its own). "Support this project" never said which project; the menu is MeshTerm's and the section above it is already About MeshTerm, so the name is the useful half of that phrase. "Device configuration" was the longest label in the list and set the width of the whole name column: at 19 cells instead of 23, every description now gets 51 and the widest row is 68. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1104 lines
43 KiB
Python
1104 lines
43 KiB
Python
"""Interactive device-configuration screens, rendered in the full-screen session.
|
||
|
||
Two sibling screens live here, deliberately kept distinct:
|
||
|
||
* **Device config** (:func:`edit_config`, behind the ``config`` tool) — one grouped,
|
||
column-aligned list of every setting under its category heading, each row showing its
|
||
current value, any staged change, and a one-line explanation. Editing a row *stages* the
|
||
new value (shown as ``current → new``) 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** (:func:`device_actions`, behind the ``device-actions`` tool) — the
|
||
operations that act on the box itself rather than a value: backup/restore, the
|
||
identity key, clock sync, reboot, and factory reset. These *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. They share
|
||
the settings snapshot and :func:`~meshterm.tools.config.apply_ops` executor with the
|
||
editor, which is why both screens live in this module. The everyday advert action
|
||
lives in its own main-menu entry instead (the ``advert`` tool, via
|
||
:func:`send_advert`), one keystroke away as a popup over the menu.
|
||
|
||
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 TYPE_CHECKING, Any, Optional
|
||
from urllib.parse import quote
|
||
|
||
from rich import box
|
||
from rich.cells import cell_len
|
||
from rich.table import Table
|
||
from rich.text import Text
|
||
|
||
from ..core.advert_store import (
|
||
DIRECT_CADENCE_HOURS,
|
||
FLOOD_CADENCE_HOURS,
|
||
OFF,
|
||
AdvertPolicy,
|
||
cadence_label,
|
||
)
|
||
from ..core.device_config import (
|
||
DeviceConfigError,
|
||
SettingSpec,
|
||
build_snapshot,
|
||
effective_maximum,
|
||
format_value,
|
||
get_spec,
|
||
parse_value,
|
||
settings_by_category,
|
||
)
|
||
from .menus import (
|
||
back_rows,
|
||
confirm_discard,
|
||
exit_rows,
|
||
lane_header,
|
||
lane_row,
|
||
menu_rows,
|
||
section_heading,
|
||
)
|
||
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_DIRECT = "__advert_direct__"
|
||
_ADVERT_FLOOD = "__advert_flood__"
|
||
_SYNC_CLOCK = "__sync_clock__"
|
||
_REBOOT = "__reboot__"
|
||
_BACKUP = "__backup__"
|
||
_RESTORE = "__restore__"
|
||
_IDENTITY_KEY = "__identity_key__"
|
||
_RESET = "__reset__"
|
||
_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")
|
||
|
||
#: 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 cached_snapshot(ctx: "AppContext", device: "Device") -> dict:
|
||
"""A config snapshot for a *screen open*, reusing the devstate session cache.
|
||
|
||
The two facts devstate already holds — ``SELF_INFO`` and the path-hash mode — are
|
||
the slowest part of :func:`~meshterm.core.device_config.build_snapshot` to re-ask
|
||
the radio for, and every config apply invalidates them (see
|
||
``tools/config.py``), so an open can trust the cache. The refresh after a restore,
|
||
key change, or factory reset keeps calling ``build_snapshot(device)`` raw: the
|
||
device state genuinely changed under us there.
|
||
"""
|
||
self_info: Optional[dict] = None
|
||
path_hash_mode: Optional[int] = None
|
||
try:
|
||
self_info = await ctx.devstate.self_info()
|
||
path_hash_mode = await ctx.devstate.path_hash_mode()
|
||
except Exception: # noqa: BLE001 - fall back to the raw reads below
|
||
pass
|
||
return await build_snapshot(
|
||
device, self_info=self_info, path_hash_mode=path_hash_mode
|
||
)
|
||
|
||
|
||
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 anything staged to apply.
|
||
"""
|
||
from .tui import CANCEL, SelectScreen
|
||
|
||
device = await ctx.device()
|
||
snapshot = await cached_snapshot(ctx, device)
|
||
custom = await device.get_custom_vars()
|
||
# The background-advert cadences are app-side settings (MeshTerm sends the adverts,
|
||
# not the firmware), but they're staged and applied exactly like device values so
|
||
# the editor stays one coherent surface.
|
||
policy = ctx.advert_store.load(str(snapshot.get("public_key") or ""))
|
||
|
||
# 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] = [] # staged custom-variable ops, in order
|
||
cursor: Any = None # the row to re-highlight, so the menu reopens where you left it
|
||
|
||
while True:
|
||
staged = len(pending) + len(extra_ops)
|
||
title, items = _menu_items(snapshot, pending, staged, policy)
|
||
menu = SelectScreen(
|
||
title, items, default=cursor, wrap=False,
|
||
footer_hint="↑↓ move · type to filter · Enter select · Esc back",
|
||
)
|
||
menu.future = loop.create_future()
|
||
session.push(menu)
|
||
# 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):
|
||
if staged and not await confirm_discard(ctx, staged, verb="applying"):
|
||
continue # keep editing — the same menu is rebuilt next loop
|
||
return None
|
||
if choice == _APPLY:
|
||
ops: list[tuple] = []
|
||
for k, v in pending.items():
|
||
if k in (_ADVERT_DIRECT, _ADVERT_FLOOD):
|
||
ops.append(("advert_cadence", k == _ADVERT_FLOOD, v))
|
||
else:
|
||
ops.append(("set", k, v))
|
||
ops.extend(extra_ops)
|
||
return ops or None
|
||
if choice in (_ADVERT_DIRECT, _ADVERT_FLOOD):
|
||
await _stage_advert_cadence(ctx, choice == _ADVERT_FLOOD, policy, pending)
|
||
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)
|
||
else: # a setting key
|
||
await _stage_setting(ctx, choice, snapshot, pending)
|
||
finally:
|
||
session.pop(menu)
|
||
|
||
|
||
# --- rendering ---------------------------------------------------------------
|
||
|
||
|
||
def config_table(
|
||
snapshot: dict,
|
||
custom: dict[str, str],
|
||
pending: Optional[dict[str, Any]] = None,
|
||
) -> Table:
|
||
"""Build the full configuration table, overlaying any staged changes.
|
||
|
||
Args:
|
||
snapshot: Device snapshot from ``build_snapshot``.
|
||
custom: Current custom variables.
|
||
pending: Optional staged changes (setting key -> new value).
|
||
|
||
Returns:
|
||
A Rich :class:`Table` of every setting's current (and staged) value plus custom
|
||
variables, ready to hand to ``ctx.ui.show`` / ``ctx.ui.view``.
|
||
"""
|
||
pending = pending or {}
|
||
show_staged = bool(pending)
|
||
# Match the contacts list: a frameless SIMPLE_HEAD table with a left-justified accent title
|
||
# and muted headers, so the two screens read as one family (see widgets.contacts_table).
|
||
table = Table(
|
||
title="[accent]Device config[/accent]",
|
||
title_justify="left",
|
||
box=box.SIMPLE_HEAD,
|
||
show_edge=False,
|
||
pad_edge=False,
|
||
header_style="muted",
|
||
expand=False,
|
||
padding=(0, 2, 0, 0),
|
||
)
|
||
table.add_column("SETTING", style="muted", no_wrap=True)
|
||
table.add_column("CURRENT", no_wrap=True)
|
||
if show_staged:
|
||
table.add_column("STAGED", style="warn", no_wrap=True)
|
||
table.add_column("DESCRIPTION", style="muted")
|
||
# Each setting's name is indented two spaces so the rows read as sitting *under* their
|
||
# accent section heading, which stays flush-left.
|
||
indent = " "
|
||
for category, specs in settings_by_category():
|
||
table.add_section()
|
||
header = [f"[accent]── {category} ──[/accent]", ""]
|
||
if show_staged:
|
||
header.append("")
|
||
table.add_row(*header, "")
|
||
for spec in specs:
|
||
current = format_value(spec, spec.getter(snapshot))
|
||
row = [f"{indent}{spec.label}", current]
|
||
if show_staged:
|
||
row.append(format_value(spec, pending[spec.key]) if spec.key in pending else "")
|
||
table.add_row(*row, spec.help)
|
||
if custom:
|
||
table.add_section()
|
||
cols = 4 if show_staged else 3
|
||
table.add_row(
|
||
"[accent]── Custom ──[/accent]", *([""] * (cols - 1))
|
||
)
|
||
for key, value in custom.items():
|
||
row = [f"{indent}{key}", value]
|
||
if show_staged:
|
||
row.append("")
|
||
table.add_row(*row, "")
|
||
return table
|
||
|
||
|
||
def _setting_value(spec: SettingSpec, snapshot: dict, pending: dict) -> Text:
|
||
"""One setting's VALUE lane: ``current [→ staged]``.
|
||
|
||
The staged arrow is drawn in the warn style so a dirty row stands out at a glance;
|
||
the span survives the select highlight (which only tints the row's base style).
|
||
"""
|
||
value = Text(format_value(spec, spec.getter(snapshot)))
|
||
if spec.key in pending:
|
||
value.append(f" → {format_value(spec, pending[spec.key])}", style="warn")
|
||
return value
|
||
|
||
|
||
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_value(snapshot: dict, pending: dict) -> Text:
|
||
"""The Location row's VALUE lane, standing in for the ``adv_lat``/``adv_lon`` pair."""
|
||
value = Text(_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"))
|
||
value.append(f" → {_format_coords(lat, lon)}", style="warn")
|
||
return value
|
||
|
||
|
||
def _menu_items(
|
||
snapshot: dict, pending: dict, staged: int, policy: AdvertPolicy
|
||
) -> 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). The rows sit in three aligned columns — setting, current
|
||
value (and any staged new value), description — under one header line, so the list
|
||
reads like the full-configuration table it stages changes for.
|
||
"""
|
||
# First pass: collect every row's lanes per category, so the columns can be sized to
|
||
# their content (including any staged ``→ new`` arrows) before a single row is built.
|
||
sections: list[tuple[str, list[tuple[str, Text, str, Any]]]] = []
|
||
for category, specs in settings_by_category():
|
||
rows: list[tuple[str, Text, str, Any]] = []
|
||
for spec in specs:
|
||
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":
|
||
rows.append((
|
||
"Location",
|
||
_location_value(snapshot, pending),
|
||
"Advertised position; pick it on the map",
|
||
_LOCATION,
|
||
))
|
||
continue
|
||
rows.append((spec.label, _setting_value(spec, snapshot, pending), spec.help, spec.key))
|
||
if category == "Radio":
|
||
rows.append((
|
||
"Radio presets…", Text(),
|
||
"Apply a standard regional or trade-off config", _PRESETS,
|
||
))
|
||
elif category == "Experimental":
|
||
rows.append((
|
||
"Custom variables…", Text(),
|
||
"Set a raw firmware variable by name", _CUSTOM,
|
||
))
|
||
sections.append((category, rows))
|
||
|
||
# App-side rows: the background-advert cadences MeshTerm itself runs (see the
|
||
# advert scheduler). They stage and apply like device settings, so they sit in the
|
||
# same lanes under their own heading.
|
||
sections.append((
|
||
"Background adverts",
|
||
[
|
||
(
|
||
"Direct advert",
|
||
_cadence_value(policy, False, pending),
|
||
"Scheduled zero-hop announce; any manual send resets it",
|
||
_ADVERT_DIRECT,
|
||
),
|
||
(
|
||
"Flood advert",
|
||
_cadence_value(policy, True, pending),
|
||
"Scheduled mesh-wide announce via repeaters",
|
||
_ADVERT_FLOOD,
|
||
),
|
||
],
|
||
))
|
||
|
||
label_w = max(cell_len(label) for _, rows in sections for label, _, _, _ in rows)
|
||
value_w = max(cell_len(value.plain) for _, rows in sections for _, value, _, _ in rows)
|
||
|
||
# The shared editor header (menus.lane_header): each label over its own lane, clear of
|
||
# the pointer column, abbreviating rather than wrapping where the terminal is too narrow
|
||
# for the full words. It is pinned for the whole list — the lanes mean the same in every
|
||
# category — so a scrolled row keeps both its column header and its section heading
|
||
# overhead (see Screen.sticky_rows).
|
||
items: list = [
|
||
Separator(lambda w: lane_header(label_w, value_w, w), pinned=True)
|
||
]
|
||
for category, rows in sections:
|
||
items.append(section_heading(category))
|
||
for label, value, help_text, key in rows:
|
||
items.append(
|
||
Choice(title=lane_row(label, value, help_text, label_w, value_w), value=key)
|
||
)
|
||
|
||
# The backtracking rows sit together below one blank line, like every other screen's
|
||
# Back (the main menu's Quit included); with changes staged, Apply joins the group and
|
||
# Back spells out the consequence of leaving (see menus.exit_rows).
|
||
items.extend(exit_rows(staged, apply_value=_APPLY, back_value=_CANCEL))
|
||
|
||
title = "Device config" + (f" — {staged} staged" if staged else "")
|
||
return title, items
|
||
|
||
|
||
def _cadence_value(policy: AdvertPolicy, flood: bool, pending: dict) -> Text:
|
||
"""One background-advert row's VALUE lane: ``current [→ staged]``."""
|
||
key = _ADVERT_FLOOD if flood else _ADVERT_DIRECT
|
||
value = Text(cadence_label(policy.cadence(flood)))
|
||
if key in pending:
|
||
value.append(f" → {cadence_label(pending[key])}", style="warn")
|
||
return value
|
||
|
||
|
||
# --- staging individual changes ----------------------------------------------
|
||
|
||
|
||
async def _stage_setting(
|
||
ctx: "AppContext", key: str, snapshot: dict, pending: dict[str, Any]
|
||
) -> None:
|
||
"""Prompt for one setting's new value and stage it."""
|
||
spec = get_spec(key)
|
||
current = pending.get(key, spec.getter(snapshot))
|
||
value = await _prompt_value(ctx, spec, current, snapshot)
|
||
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
|
||
|
||
|
||
def _range_hint(spec: SettingSpec, snapshot: dict) -> str:
|
||
"""A muted "allowed values" hint for a numeric prompt.
|
||
|
||
Uses the *effective* maximum — the device-reported bound (e.g. this board's max TX
|
||
power) when the spec names one, else the static bound — so the hint promises exactly
|
||
what validation will accept.
|
||
"""
|
||
maximum = effective_maximum(spec, snapshot)
|
||
if spec.minimum is not None and maximum is not None:
|
||
return f"Allowed: {spec.minimum:g} – {maximum:g}"
|
||
if spec.minimum is not None:
|
||
return f"Allowed: ≥ {spec.minimum:g}"
|
||
if maximum is not None:
|
||
return f"Allowed: ≤ {maximum:g}"
|
||
return ""
|
||
|
||
|
||
async def _prompt_value(
|
||
ctx: "AppContext", spec: SettingSpec, current: Any, snapshot: dict
|
||
) -> Any:
|
||
"""Prompt for a typed value for ``spec`` (in the fitting dialog), ``None`` on cancel."""
|
||
if spec.value_type == "bool":
|
||
# 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}" + (" (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(
|
||
spec.label,
|
||
items,
|
||
prompt=spec.help,
|
||
default=current if current in spec.choices else None,
|
||
)
|
||
if selected is None:
|
||
return None
|
||
if selected != _OTHER:
|
||
return selected
|
||
# else: fall through to the free-text prompt below.
|
||
|
||
def validate(text: str) -> bool | str:
|
||
try:
|
||
parse_value(spec, text, snapshot)
|
||
return True
|
||
except DeviceConfigError as exc:
|
||
return str(exc)
|
||
|
||
raw = await ctx.ui.text(
|
||
spec.label,
|
||
prompt=spec.help,
|
||
default="" if current is None else str(current),
|
||
validate=validate,
|
||
help_text=_range_hint(spec, snapshot),
|
||
)
|
||
return None if raw is None else parse_value(spec, raw, snapshot)
|
||
|
||
|
||
async def _stage_advert_cadence(
|
||
ctx: "AppContext", flood: bool, policy: AdvertPolicy, pending: dict[str, Any]
|
||
) -> None:
|
||
"""Pick one background-advert type's cadence and stage it.
|
||
|
||
Staged under the type's sentinel key; Apply turns it into an ``advert_cadence`` op
|
||
(see :func:`~meshterm.tools.config.apply_ops`). Picking the value already in force
|
||
un-stages the row, matching :func:`_stage_setting`.
|
||
"""
|
||
key = _ADVERT_FLOOD if flood else _ADVERT_DIRECT
|
||
in_force = policy.cadence(flood)
|
||
current = pending.get(key, in_force)
|
||
hours_choices = FLOOD_CADENCE_HOURS if flood else DIRECT_CADENCE_HOURS
|
||
items: list = [
|
||
Choice(
|
||
title=cadence_label(hours).capitalize()
|
||
+ (" (current)" if hours == current else ""),
|
||
value=hours,
|
||
)
|
||
for hours in (*hours_choices, OFF)
|
||
]
|
||
if flood:
|
||
prompt = "How often repeaters rebroadcast this node across the mesh:"
|
||
else:
|
||
prompt = "How often this node announces itself to neighbours in range:"
|
||
selected = await ctx.ui.select(
|
||
"Flood advert" if flood else "Direct advert",
|
||
items,
|
||
prompt=prompt,
|
||
default=current,
|
||
)
|
||
if selected is None:
|
||
return
|
||
if selected == in_force:
|
||
pending.pop(key, None) # set back to the value in force — nothing to change
|
||
else:
|
||
pending[key] = selected
|
||
|
||
|
||
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
|
||
|
||
items: list = [
|
||
Choice(
|
||
title=f"{p.name}: {p.freq} MHz, BW {p.bw}, SF{p.sf}, CR{p.cr} — {p.help}",
|
||
value=i,
|
||
)
|
||
for i, p in enumerate(RADIO_PRESETS)
|
||
]
|
||
items.extend(back_rows())
|
||
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]
|
||
pending.update(preset.as_settings())
|
||
|
||
|
||
async def _stage_custom_var(
|
||
ctx: "AppContext", custom: dict[str, str], extra_ops: list[tuple]
|
||
) -> None:
|
||
"""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
|
||
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, value))
|
||
|
||
|
||
# --- device actions (run immediately) -----------------------------------------
|
||
|
||
|
||
async def device_actions(ctx: "AppContext") -> None:
|
||
"""Run the Device actions screen: immediate operations on the companion itself.
|
||
|
||
The action counterpart of :func:`edit_config`, behind the ``device-actions`` tool.
|
||
Nothing here is staged — each action runs as soon as its own confirmation is given
|
||
(destructive ones gate behind typing a confirmation word) and presents its result at
|
||
once. The menu stays open between actions so several can be run in a row; Esc (or
|
||
Back) returns to the main menu, and a reboot closes the screen and hands off to the
|
||
session's reconnect dialog.
|
||
|
||
Args:
|
||
ctx: Shared application context (provides the connected device and UI surface).
|
||
"""
|
||
from .tui import CANCEL, SelectScreen
|
||
|
||
device = await ctx.device()
|
||
snapshot = await cached_snapshot(ctx, device)
|
||
|
||
# Same persistent-backdrop pattern as the editor: the menu stays pushed while each
|
||
# action's prompts float over it as modal popups (see edit_config).
|
||
session = getattr(ctx.ui, "session", None)
|
||
if session is None: # pragma: no cover - guarded by the menu-only caller
|
||
raise RuntimeError("device actions are only available in the menu")
|
||
loop = asyncio.get_running_loop()
|
||
cursor: Any = None # the row to re-highlight, so the menu reopens where you left it
|
||
|
||
while True:
|
||
menu = SelectScreen("Device actions", _action_items(), default=cursor, wrap=False)
|
||
menu.future = loop.create_future()
|
||
session.push(menu)
|
||
try:
|
||
choice = await menu.future
|
||
if choice is CANCEL or choice in (None, _CANCEL):
|
||
return
|
||
cursor = choice
|
||
if choice == _SYNC_CLOCK:
|
||
await _sync_clock(ctx, device, snapshot)
|
||
elif choice == _BACKUP:
|
||
await _backup_now(ctx, device, snapshot)
|
||
elif choice == _RESTORE:
|
||
if await _restore_now(ctx, device, snapshot):
|
||
snapshot = await build_snapshot(device)
|
||
elif choice == _IDENTITY_KEY:
|
||
if await _identity_key_menu(ctx, device, snapshot):
|
||
snapshot = await build_snapshot(device)
|
||
elif choice == _REBOOT:
|
||
if await _reboot(ctx, device, snapshot):
|
||
return # the link is dropping; the reconnect dialog takes over
|
||
elif choice == _RESET:
|
||
if await _factory_reset(ctx, device, snapshot):
|
||
snapshot = await build_snapshot(device)
|
||
finally:
|
||
session.pop(menu)
|
||
|
||
|
||
def _action_items() -> list:
|
||
"""Build the Device actions rows: label and description in two aligned columns.
|
||
|
||
The shared menu-row presentation (see :func:`~meshterm.ui.menus.menu_rows`).
|
||
Factory reset keeps its err-tinted label so the one irreversible row reads as such.
|
||
"""
|
||
items = menu_rows(
|
||
[
|
||
("🕒 Sync clock…", "Set the device clock from this computer", _SYNC_CLOCK),
|
||
("💾 Back up config to a file…", "Write every setting to TOML", _BACKUP),
|
||
("📂 Restore config from a backup…", "Preview or apply a saved TOML", _RESTORE),
|
||
("🔐 Identity key…", "Export or import the node's private key", _IDENTITY_KEY),
|
||
("🔄 Reboot device…", "Restart the companion and reconnect", _REBOOT),
|
||
(
|
||
Text("⚠ Factory reset…", style="err"),
|
||
"Erase everything (typed confirmation)",
|
||
_RESET,
|
||
),
|
||
]
|
||
)
|
||
items.extend(back_rows(_CANCEL))
|
||
return items
|
||
|
||
|
||
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.
|
||
|
||
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:
|
||
The number of changes applied.
|
||
"""
|
||
from ..tools.config import apply_ops
|
||
|
||
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 send_advert(ctx: "AppContext") -> None:
|
||
"""Run the Send advert flow, behind the main menu's ``advert`` popup tool.
|
||
|
||
Send an advert (zero-hop or flood) or show this node's shareable contact card. Only
|
||
``SELF_INFO`` is read here: the advert command itself never consults the snapshot,
|
||
and the contact card needs just the name, public key, and advert type — so opening
|
||
this skips the tuning/path-hash reads of a full
|
||
:func:`~meshterm.core.device_config.build_snapshot` and stays snappy over Bluetooth.
|
||
|
||
Args:
|
||
ctx: Shared application context (provides the connected device and UI surface).
|
||
"""
|
||
device = await ctx.device()
|
||
snapshot = dict(await ctx.devstate.self_info()) # the session cache; no re-read
|
||
# No emoji in this floating dialog's title: a terminal that paints an emoji a cell
|
||
# narrower than Rich measures it leaves the content-sized popup's title border short,
|
||
# bleeding the backdrop through the frame. The menu row keeps its 📡 icon (drawn in the
|
||
# full-width base, where the miscount has nowhere to show). filterable=False too: a
|
||
# stray key must not narrow (and so resize) this fixed four-item list.
|
||
choice = await ctx.ui.select(
|
||
"Send advert",
|
||
[
|
||
*menu_rows(
|
||
[
|
||
("Zero-hop", "Announce to direct neighbours", "zero"),
|
||
("Flood", "Repeaters rebroadcast it across the mesh", "flood"),
|
||
("Share QR / URI", "Show this node's contact card", "share"),
|
||
]
|
||
),
|
||
*back_rows(),
|
||
],
|
||
prompt="Announce this node to the mesh:",
|
||
filterable=False,
|
||
)
|
||
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)}"
|
||
)
|
||
|
||
|
||
async def show_contact_card(
|
||
ctx: "AppContext", name: str, public_key: str, node_type: int = 1
|
||
) -> None:
|
||
"""Pop up a node's contact card: a scannable QR code over the raw share link.
|
||
|
||
THE share-a-contact popup, used both for our own node (the advert menu's
|
||
``Share QR / URI``) and for any full-keyed contact (the node detail page's
|
||
``Share contact``): a phone scans the code — or the link is passed along as text —
|
||
and the companion app adds the node as a contact.
|
||
|
||
Args:
|
||
ctx: Shared application context (provides the UI surface).
|
||
name: The node's advertised name (the card's title and the link's name field).
|
||
public_key: The node's full public key as hex (the link is useless without it).
|
||
node_type: The MeshCore advert type byte (1 companion, 2 repeater, 3 room,
|
||
4 sensor).
|
||
"""
|
||
from .qr import share_popup
|
||
|
||
await share_popup(
|
||
ctx,
|
||
name=name,
|
||
url=contact_share_url(name, public_key, node_type),
|
||
intro="Scan to add this node as a contact:",
|
||
)
|
||
|
||
|
||
async def _show_contact_card(ctx: "AppContext", snapshot: dict) -> None:
|
||
"""Show our own node's contact card from its ``SELF_INFO`` snapshot."""
|
||
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")
|
||
await show_contact_card(ctx, name, public_key, int(snapshot.get("adv_type") or 1))
|
||
|
||
|
||
async def _reboot(ctx: "AppContext", device: "Device", snapshot: dict) -> bool:
|
||
"""Confirm and reboot the device, handing off to the session's reconnect dialog.
|
||
|
||
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 actions screen should close; ``False``
|
||
if the user backed out (or the simulator, which has no link to drop, absorbed it).
|
||
"""
|
||
choice = await ctx.ui.dialog(
|
||
"Reboot the device now?",
|
||
[("Cancel", None), ("Reboot", "reboot")],
|
||
title="Reboot device",
|
||
default=1,
|
||
danger=True,
|
||
)
|
||
if choice != "reboot":
|
||
return False
|
||
|
||
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
|
||
|
||
# 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
|
||
# actions screen 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 _sync_clock(ctx: "AppContext", device: "Device", snapshot: dict) -> None:
|
||
"""Show the device clock's drift against this computer and offer to correct it.
|
||
|
||
A companion that boots with a bad RTC stamps every message wrongly, so the dialog
|
||
leads with the measured drift (or admits the clock is unreadable) before the Sync
|
||
button writes the host's time.
|
||
"""
|
||
import time
|
||
|
||
device_time: Optional[int] = None
|
||
try:
|
||
device_time = await device.get_time()
|
||
except Exception: # noqa: BLE001 - old firmware; offer the blind sync instead
|
||
pass
|
||
if device_time:
|
||
drift = device_time - int(time.time())
|
||
stamp = _clock_text(device_time)
|
||
prompt = (
|
||
f"The device clock reads {stamp} — "
|
||
f"{_drift_text(drift)}. Set it from this computer?"
|
||
)
|
||
else:
|
||
prompt = (
|
||
"The device did not report its clock. Set it from this computer anyway?"
|
||
)
|
||
choice = await ctx.ui.dialog(
|
||
prompt,
|
||
[("Cancel", None), ("Sync", "sync")],
|
||
title="Sync clock",
|
||
default=1,
|
||
)
|
||
if choice == "sync":
|
||
await _run_now(ctx, device, snapshot, [("sync_clock",)], "Sync clock")
|
||
|
||
|
||
def _clock_text(epoch: int) -> str:
|
||
"""A device timestamp rendered in this computer's local time."""
|
||
from datetime import datetime
|
||
|
||
return datetime.fromtimestamp(epoch).astimezone().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
|
||
def _drift_text(drift: int) -> str:
|
||
"""Describe a clock drift in seconds: ``"12 s behind"``, ``"3 s ahead"``, ``"in sync"``."""
|
||
if abs(drift) < 2:
|
||
return "in sync with this computer"
|
||
direction = "ahead of" if drift > 0 else "behind"
|
||
return f"{abs(drift)} s {direction} this computer"
|
||
|
||
|
||
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 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:"
|
||
)
|
||
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",
|
||
[
|
||
*menu_rows(
|
||
[
|
||
("Show private key", "Display it on screen (sensitive)", "show"),
|
||
("Export to a file…", "Write it to disk (keep it secret)", "file"),
|
||
("Import a key…", "Replace this device's identity", "import"),
|
||
]
|
||
),
|
||
*back_rows(),
|
||
],
|
||
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
|
||
|
||
|
||
# --- validators --------------------------------------------------------------
|
||
|
||
|
||
def _is_hex(text: str) -> bool | str:
|
||
"""Validate that ``text`` is a hex string."""
|
||
try:
|
||
bytes.fromhex(text)
|
||
return True
|
||
except ValueError:
|
||
return "Enter hex characters only."
|