From 2a9facc56ee56c99ec70eabc06346917f757964c Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Sun, 5 Jul 2026 14:09:18 -0400 Subject: [PATCH] Replace questionary with a full-screen prompt_toolkit TUI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a UI surface abstraction (ctx.ui) that tools use for all input and output instead of touching questionary or the console directly. Two backends implement it: PlainUi for scripted CLI runs (prints immediately, byte-for-byte unchanged) and TuiUi for the interactive menu (bounded, scrollable result windows over a full-screen TuiSession). The new meshtools/ui/tui package reimplements the slice of questionary the app used — validated text input, confirm, autocomplete, and selectable lists — plus framing, progress, and screen management. Swap the dependency from questionary to prompt_toolkit. Co-Authored-By: Claude Opus 4.8 --- meshtools/context.py | 21 +++ meshtools/tools/config.py | 34 ++-- meshtools/tools/devices.py | 16 +- meshtools/tools/history.py | 2 +- meshtools/tools/info.py | 4 +- meshtools/tools/link_budget.py | 32 ++-- meshtools/tools/link_health.py | 16 +- meshtools/tools/monitor.py | 18 +- meshtools/tools/nodes.py | 2 +- meshtools/tools/route_map.py | 17 +- meshtools/tools/trace.py | 37 ++-- meshtools/tools/tx_optimize.py | 55 +++--- meshtools/ui/config_editor.py | 240 +++++++++++++------------ meshtools/ui/device_picker.py | 61 +++---- meshtools/ui/menu.py | 209 +++++++++++----------- meshtools/ui/surface.py | 279 +++++++++++++++++++++++++++++ meshtools/ui/tui/__init__.py | 35 ++++ meshtools/ui/tui/frame.py | 149 ++++++++++++++++ meshtools/ui/tui/progress.py | 139 +++++++++++++++ meshtools/ui/tui/prompt.py | 269 ++++++++++++++++++++++++++++ meshtools/ui/tui/render.py | 85 +++++++++ meshtools/ui/tui/screen.py | 182 +++++++++++++++++++ meshtools/ui/tui/select.py | 180 +++++++++++++++++++ meshtools/ui/tui/session.py | 316 +++++++++++++++++++++++++++++++++ pyproject.toml | 2 +- tests/test_tui.py | 303 +++++++++++++++++++++++++++++++ 26 files changed, 2311 insertions(+), 392 deletions(-) create mode 100644 meshtools/ui/surface.py create mode 100644 meshtools/ui/tui/__init__.py create mode 100644 meshtools/ui/tui/frame.py create mode 100644 meshtools/ui/tui/progress.py create mode 100644 meshtools/ui/tui/prompt.py create mode 100644 meshtools/ui/tui/render.py create mode 100644 meshtools/ui/tui/screen.py create mode 100644 meshtools/ui/tui/select.py create mode 100644 meshtools/ui/tui/session.py create mode 100644 tests/test_tui.py diff --git a/meshtools/context.py b/meshtools/context.py index 166c5da..1939fad 100644 --- a/meshtools/context.py +++ b/meshtools/context.py @@ -23,6 +23,7 @@ from .persistence.repository import Repository if TYPE_CHECKING: from .services.monitor_service import MonitorService + from .ui.surface import Ui @dataclass(slots=True) @@ -59,12 +60,32 @@ class AppContext: explicit_selection: bool = False _device: Optional[Device] = field(default=None, init=False, repr=False) _monitor: "Optional[MonitorService]" = field(default=None, init=False, repr=False) + _ui: "Optional[Ui]" = field(default=None, init=False, repr=False) @property def profile_name(self) -> Optional[str]: """Name of the active profile, if any.""" return self.profile.name if self.profile else None + @property + def ui(self) -> "Ui": + """The active UI surface, defaulting to the plain console surface for the CLI. + + The interactive menu replaces this with a full-screen TUI surface for the session; + scripted CLI runs use the lazily-created plain surface, which prints directly. The + surface module (and prompt_toolkit) is imported lazily here to keep startup fast. + """ + if self._ui is None: + from .ui.surface import PlainUi + + self._ui = PlainUi(self.console) + return self._ui + + @ui.setter + def ui(self, value: "Ui") -> None: + """Install a UI surface (used by the menu to switch to the full-screen TUI).""" + self._ui = value + @property def monitor(self) -> "MonitorService": """Return the session's passive-monitor service, creating it on first use. diff --git a/meshtools/tools/config.py b/meshtools/tools/config.py index 8eb7354..3732cc5 100644 --- a/meshtools/tools/config.py +++ b/meshtools/tools/config.py @@ -203,18 +203,18 @@ async def apply_ops( await _show(ctx, device, snapshot) elif kind == "get": spec = get_spec(op[1]) - ctx.console.print( + ctx.ui.note( f"[muted]{spec.key}[/muted] = {format_value(spec, spec.getter(snapshot))}" ) elif kind == "set": changes += await _apply_setting(ctx, device, op[1], op[2], snapshot) elif kind == "set_custom": await device.set_custom_var(op[1], op[2]) - ctx.console.print(f"[ok]✓[/ok] custom [brand]{op[1]}[/brand] = {op[2]}") + ctx.ui.note(f"[ok]✓[/ok] custom [brand]{op[1]}[/brand] = {op[2]}") changes += 1 elif kind == "set_channel": await device.set_channel(op[1], op[2], op[3]) - ctx.console.print(f"[ok]✓[/ok] channel {op[1]} = [brand]{op[2]}[/brand]") + ctx.ui.note(f"[ok]✓[/ok] channel {op[1]} = [brand]{op[2]}[/brand]") changes += 1 elif kind == "backup": artifacts.append(str(await _backup(device, snapshot, op[1]))) @@ -222,21 +222,21 @@ async def apply_ops( changes += await _restore(ctx, device, snapshot, op[1], op[2]) elif kind == "advert": await device.send_advert(len(op) > 1 and bool(op[1])) - ctx.console.print("[ok]✓[/ok] advertisement sent") + ctx.ui.note("[ok]✓[/ok] advertisement sent") elif kind == "reboot": await device.reboot() - ctx.console.print("[warn]device rebooting[/warn]") + ctx.ui.note("[warn]device rebooting[/warn]") elif kind == "export_key": await _export_key(ctx, device, op[1] if len(op) > 1 else None, artifacts) elif kind == "import_key": await device.import_private_key(op[1]) - ctx.console.print("[ok]✓[/ok] private key imported") + ctx.ui.note("[ok]✓[/ok] private key imported") changes += 1 elif kind == "factory_reset": await device.factory_reset() - ctx.console.print("[err]device factory-reset[/err]") + ctx.ui.note("[err]device factory-reset[/err]") else: # pragma: no cover - guarded by the call sites that build ops - ctx.console.print(f"[err]unknown config op: {kind}[/err]") + ctx.ui.note(f"[err]unknown config op: {kind}[/err]") return changes, artifacts @@ -255,16 +255,16 @@ async def _apply_setting( value = parse_value(spec, raw) await spec.apply(device, value, snapshot) snapshot[key] = value - ctx.console.print(f"[ok]✓[/ok] [brand]{key}[/brand] = {format_value(spec, value)}") + ctx.ui.note(f"[ok]✓[/ok] [brand]{key}[/brand] = {format_value(spec, value)}") return 1 async def _show(ctx: AppContext, device: Device, snapshot: dict) -> None: """Render the full configuration table plus custom variables.""" - from ..ui.config_editor import render_config + from ..ui.config_editor import config_table custom = await device.get_custom_vars() - render_config(ctx.console, snapshot, custom) + ctx.ui.show(config_table(snapshot, custom)) async def _backup(device: Device, snapshot: dict, path: Path) -> Path: @@ -286,7 +286,7 @@ async def _restore( custom = await device.get_custom_vars() ops = plan_restore(backup, snapshot, custom) if not ops: - ctx.console.print("[muted]restore: device already matches the backup.[/muted]") + ctx.ui.note("[muted]restore: device already matches the backup.[/muted]") return 0 if dry_run: @@ -296,8 +296,8 @@ async def _restore( table.add_column("Value") for op in ops: table.add_row(op[0], str(op[1]), str(op[2] if len(op) > 2 else "")) - ctx.console.print(table) - ctx.console.print("[muted]dry run — nothing was changed.[/muted]") + ctx.ui.show(table) + ctx.ui.note("[muted]dry run — nothing was changed.[/muted]") return 0 changes = 0 @@ -323,11 +323,9 @@ async def _export_key( path.parent.mkdir(parents=True, exist_ok=True) path.write_text(key_hex, encoding="utf-8") artifacts.append(str(path)) - ctx.console.print("[warn]private key written — keep this file secret.[/warn]") + ctx.ui.note("[warn]private key written — keep this file secret.[/warn]") else: - ctx.console.print( - "[warn]private key (keep secret):[/warn] [muted]" + key_hex + "[/muted]" - ) + ctx.ui.note("[warn]private key (keep secret):[/warn] [muted]" + key_hex + "[/muted]") async def _read_channels(device: Device) -> list[dict]: diff --git a/meshtools/tools/devices.py b/meshtools/tools/devices.py index 1ba7beb..3d05164 100644 --- a/meshtools/tools/devices.py +++ b/meshtools/tools/devices.py @@ -36,24 +36,20 @@ class DevicesTool(Tool): Returns: ``{"select": True}`` when the user opted to re-pick a device, else ``{}``. """ - import questionary - if ctx.mock: return {} - change = await questionary.confirm( - "Change the active companion device?", default=False - ).ask_async() + change = await ctx.ui.confirm("Change the active companion device?", default=False) if not change: return {} from ..ui.device_picker import prompt_device devices = discover_devices() - chosen = await prompt_device(ctx.console, devices, ctx.device_store.load()) + chosen = await prompt_device(ctx.ui, devices, ctx.device_store.load()) if chosen is not None: ctx.selected_device = chosen ctx.port_override = chosen.port - ctx.console.print( + ctx.ui.note( f"[ok]●[/ok] active device set to [accent]{chosen.label}[/accent] " "[muted](remembered once it connects)[/muted]" ) @@ -94,7 +90,7 @@ class DevicesTool(Tool): return ToolResult(summary={"count": len(devices)}) if not devices: - ctx.console.print( + ctx.ui.note( "[warn]No serial devices detected.[/warn] " "Connect a companion device, or use [accent]--mock[/accent] for the simulator." ) @@ -119,8 +115,8 @@ class DevicesTool(Tool): "[ok]yes[/ok]" if d.is_likely_lora else "[muted]—[/muted]", d.serial_number or "[muted]—[/muted]", ) - ctx.console.print(table) - ctx.console.print( + ctx.ui.show(table) + ctx.ui.note( "[muted]● active ★ remembered default. " "Pass --port to select on the CLI.[/muted]" ) diff --git a/meshtools/tools/history.py b/meshtools/tools/history.py index 0c0c5f0..8e3c7ad 100644 --- a/meshtools/tools/history.py +++ b/meshtools/tools/history.py @@ -50,7 +50,7 @@ class HistoryTool(Tool): f"[{style}]{r.status}[/{style}]", r.started_at.replace("T", " ")[:19], ) - ctx.console.print(table) + ctx.ui.show(table) return ToolResult(summary={"shown": len(runs)}) def register_cli(self, app: typer.Typer) -> None: diff --git a/meshtools/tools/info.py b/meshtools/tools/info.py index 260686b..b5412a1 100644 --- a/meshtools/tools/info.py +++ b/meshtools/tools/info.py @@ -33,14 +33,14 @@ class InfoTool(Tool): A :class:`ToolResult` summarizing the device name. """ from ..core.device_config import build_snapshot - from ..ui.config_editor import render_config + from ..ui.config_editor import config_table device = await ctx.device() snapshot = await build_snapshot(device) custom = await device.get_custom_vars() # Render every current setting with a short explanation of each. - render_config(ctx.console, snapshot, custom) + ctx.ui.show(config_table(snapshot, custom)) return ToolResult(summary={"name": snapshot.get("name")}) diff --git a/meshtools/tools/link_budget.py b/meshtools/tools/link_budget.py index 87ed83f..a980f5a 100644 --- a/meshtools/tools/link_budget.py +++ b/meshtools/tools/link_budget.py @@ -11,7 +11,6 @@ from __future__ import annotations from typing import Any, Optional -import questionary import typer from rich.panel import Panel from rich.table import Table @@ -20,6 +19,7 @@ from rich.text import Text from ..context import AppContext from ..services import link_budget from ..services.link_budget import PATH_LOSS_EXPONENTS, RadioConfig +from ..ui.tui import Choice from .base import Tool, ToolResult, register @@ -43,38 +43,32 @@ class LinkBudgetTool(Tool): """ defaults = await self._device_radio(ctx) - payload = await questionary.text( + payload = await ctx.ui.text( "Payload size (bytes):", default="32", validate=_is_pos_int - ).ask_async() + ) if payload is None: return None - tx = await questionary.text( + tx = await ctx.ui.text( "TX power (dBm):", default=str(int(defaults.get("tx_power") or 22)), validate=_is_number, - ).ask_async() + ) if tx is None: return None - tx_gain = await questionary.text( - "TX antenna gain (dBi):", default="2", validate=_is_number - ).ask_async() + tx_gain = await ctx.ui.text("TX antenna gain (dBi):", default="2", validate=_is_number) if tx_gain is None: return None - rx_gain = await questionary.text( - "RX antenna gain (dBi):", default="2", validate=_is_number - ).ask_async() + rx_gain = await ctx.ui.text("RX antenna gain (dBi):", default="2", validate=_is_number) if rx_gain is None: return None - env = await questionary.select( + env = await ctx.ui.select( "Environment (path-loss model):", - choices=list(PATH_LOSS_EXPONENTS), + [Choice(name, name) for name in PATH_LOSS_EXPONENTS], default="suburban", - ).ask_async() + ) if env is None: return None - duty = await questionary.text( - "Duty-cycle limit (%):", default="1.0", validate=_is_number - ).ask_async() + duty = await ctx.ui.text("Duty-cycle limit (%):", default="1.0", validate=_is_number) if duty is None: return None @@ -120,8 +114,8 @@ class LinkBudgetTool(Tool): max_dwell_ms=params.get("dwell"), ) - ctx.console.print(_radio_panel(config)) - ctx.console.print(_budget_panel(budget)) + ctx.ui.show(_radio_panel(config)) + ctx.ui.show(_budget_panel(budget)) return ToolResult( summary={ diff --git a/meshtools/tools/link_health.py b/meshtools/tools/link_health.py index 812520c..fbf2113 100644 --- a/meshtools/tools/link_health.py +++ b/meshtools/tools/link_health.py @@ -11,7 +11,6 @@ from __future__ import annotations from typing import Any, Optional -import questionary import typer from rich.panel import Panel from rich.table import Table @@ -20,7 +19,6 @@ from rich.text import Text from ..context import AppContext from ..services import link_quality, trace_runner from ..services.link_quality import HealthReport, MetricChange -from ..ui.widgets import make_progress from .base import Tool, ToolResult, register _STATUS_STYLE = { @@ -59,19 +57,17 @@ class LinkHealthTool(Tool): choices = list(dict.fromkeys(history + contacts)) prompt = "Target to check:" if choices: - target = await questionary.autocomplete( - prompt, choices=choices, ignore_case=True - ).ask_async() + target = await ctx.ui.autocomplete(prompt, choices) else: - target = await questionary.text(prompt).ask_async() + target = await ctx.ui.text(prompt) if not target: return None - samples = await questionary.text( + samples = await ctx.ui.text( "Take how many fresh traces first? (0 = use stored history only)", default="0", validate=_is_nonneg_int, - ).ask_async() + ) if samples is None: return None return {"target": target.strip(), "samples": int(samples)} @@ -99,7 +95,7 @@ class LinkHealthTool(Tool): path = None if params.get("path"): path = trace_runner.parse_trace_path(params["path"], contacts) - with make_progress(ctx.console) as progress: + with ctx.ui.progress("link-health") as progress: task = progress.add_task(f"tracing {target}", total=samples) await trace_runner.run_traces( device, @@ -130,7 +126,7 @@ class LinkHealthTool(Tool): resolve=resolve, ) - ctx.console.print(_report_panel(report)) + ctx.ui.show(_report_panel(report)) if report.status == "insufficient-data": message = ( f"[muted]not enough history for[/muted] [brand]{target}[/brand] " diff --git a/meshtools/tools/monitor.py b/meshtools/tools/monitor.py index 58d9968..3562a37 100644 --- a/meshtools/tools/monitor.py +++ b/meshtools/tools/monitor.py @@ -20,7 +20,6 @@ from __future__ import annotations from typing import Any, Optional -import questionary import typer from rich.table import Table from rich.text import Text @@ -28,6 +27,7 @@ from rich.text import Text from ..context import AppContext from ..core.models import HeardNode, Observation from ..services import trace_runner +from ..ui.tui import Choice from ..ui.widgets import make_progress from .base import Tool, ToolResult, register @@ -56,14 +56,14 @@ class MonitorTool(Tool): """ monitor = ctx.monitor toggle_label = "Turn monitoring OFF" if monitor.enabled else "Turn monitoring ON" - choice = await questionary.select( + choice = await ctx.ui.select( f"Passive monitor — {monitor.status_text()}", - choices=[ - questionary.Choice(toggle_label, value="toggle"), - questionary.Choice("View heard nodes (all time)", value="view"), - questionary.Choice("Back", value="__back__"), + [ + Choice(toggle_label, value="toggle"), + Choice("View heard nodes (all time)", value="view"), + Choice("Back", value="__back__"), ], - ).ask_async() + ) if choice in (None, "__back__"): return None return {"action": choice} @@ -138,9 +138,9 @@ class MonitorTool(Tool): heard = ctx.repo.heard_nodes() if heard: # Names come from the stored observations, so no device lookup is needed. - ctx.console.print(_heard_table(heard, lambda _node: None)) + ctx.ui.show(_heard_table(heard, lambda _node: None)) else: - ctx.console.print("[muted]no packets heard yet — turn monitoring on[/muted]") + ctx.ui.note("[muted]no packets heard yet — turn monitoring on[/muted]") packets = sum(n.count for n in heard) located = sum(1 for n in heard if n.has_location) diff --git a/meshtools/tools/nodes.py b/meshtools/tools/nodes.py index 250a047..5b3a4d5 100644 --- a/meshtools/tools/nodes.py +++ b/meshtools/tools/nodes.py @@ -46,7 +46,7 @@ class NodesTool(Tool): self_name = str(info.get("name") or "this node") self_key = str(info.get("public_key") or "") - ctx.console.print(nodes_table(self_name, self_key, contacts, prefix_bytes)) + ctx.ui.show(nodes_table(self_name, self_key, contacts, prefix_bytes)) return ToolResult(summary={"contacts": len(contacts)}) diff --git a/meshtools/tools/route_map.py b/meshtools/tools/route_map.py index 0262866..309fd42 100644 --- a/meshtools/tools/route_map.py +++ b/meshtools/tools/route_map.py @@ -10,7 +10,6 @@ from __future__ import annotations from typing import Any, Optional -import questionary import typer from rich.panel import Panel from rich.table import Table @@ -20,7 +19,7 @@ from ..context import AppContext from ..services import route_stability, trace_runner from ..services.route_stability import RouteGraph from ..ui.theme import snr_style -from ..ui.widgets import make_progress +from ..ui.tui import Choice from ..viz.route_graph import render_route_graph from .base import Tool, ToolResult, register @@ -46,17 +45,17 @@ class RouteMapTool(Tool): A parameter dict, or ``None`` if the user cancelled. """ targets = ctx.repo.traced_targets() - choice = await questionary.select( + choice = await ctx.ui.select( "Map routing for:", - choices=[_WHOLE_MESH, *targets] if targets else [_WHOLE_MESH], - ).ask_async() + [Choice(_WHOLE_MESH, _WHOLE_MESH), *(Choice(t, t) for t in targets)], + ) if choice is None: return None - samples = await questionary.text( + samples = await ctx.ui.text( "Take how many fresh traces first? (0 = use stored history only)", default="0", validate=_is_nonneg_int, - ).ask_async() + ) if samples is None: return None params: dict[str, Any] = {"samples": int(samples)} @@ -92,7 +91,7 @@ class RouteMapTool(Tool): if params.get("path") else None ) - with make_progress(ctx.console) as progress: + with ctx.ui.progress("route-map") as progress: task = progress.add_task(f"tracing {target}", total=samples) await trace_runner.run_traces( device, target, samples=samples, path=path, @@ -111,7 +110,7 @@ class RouteMapTool(Tool): graph = route_stability.build_route_graph(traces, target=target, resolve=resolve) - ctx.console.print(_graph_panel(graph)) + ctx.ui.show(_graph_panel(graph)) artifacts: list[str] = [] if graph.total_routes and params.get("viz", True) and ctx.settings.output_dir: diff --git a/meshtools/tools/trace.py b/meshtools/tools/trace.py index 1b266c1..cdcac56 100644 --- a/meshtools/tools/trace.py +++ b/meshtools/tools/trace.py @@ -9,13 +9,12 @@ from __future__ import annotations from typing import Any, Optional -import questionary import typer from ..context import AppContext from ..core.models import LOCAL_DEVICE_LABEL from ..services import trace_runner -from ..ui.widgets import make_progress, stats_panel, traces_table +from ..ui.widgets import stats_panel, traces_table from .base import Tool, ToolResult, register #: Upper bound on traces per run — the per-trace table renders one column each, so @@ -45,20 +44,18 @@ class TraceTool(Tool): contacts = await device.get_contacts() choices = [c.name for c in contacts] prompt = "Target node (name or key prefix):" - # ``autocomplete`` requires a non-empty choice list; with no known contacts fall - # back to a free-text entry so the user can still type a name or key prefix. + # With no known contacts, fall back to a free-text entry so the user can still + # type a name or key prefix; otherwise suggest the known contacts. if choices: - target = await questionary.autocomplete( - prompt, choices=choices, ignore_case=True - ).ask_async() + target = await ctx.ui.autocomplete(prompt, choices) else: - target = await questionary.text(prompt).ask_async() + target = await ctx.ui.text(prompt) if not target: return None - samples_raw = await questionary.text( + samples_raw = await ctx.ui.text( f"How many traces? (1-{MAX_TRACES})", default="3", validate=_is_valid_sample_count - ).ask_async() + ) if samples_raw is None: return None @@ -67,13 +64,9 @@ class TraceTool(Tool): path_prompt = "Force a path? (comma-separated contacts/hex prefixes, blank = auto):" validate_path = lambda v: _validate_path(v, contacts) # noqa: E731 if choices: - path_spec = await questionary.autocomplete( - path_prompt, choices=choices, ignore_case=True, validate=validate_path - ).ask_async() + path_spec = await ctx.ui.autocomplete(path_prompt, choices, validate=validate_path) else: - path_spec = await questionary.text( - path_prompt, validate=validate_path - ).ask_async() + path_spec = await ctx.ui.text(path_prompt, validate=validate_path) if path_spec is None: return None @@ -95,7 +88,7 @@ class TraceTool(Tool): target = params["target"] samples = max(1, min(MAX_TRACES, int(params.get("samples", 3)))) if int(params.get("samples", 3)) > MAX_TRACES: - ctx.console.print(f"[warn]capping at {MAX_TRACES} traces[/warn]") + ctx.ui.note(f"[warn]capping at {MAX_TRACES} traces[/warn]") run_id = params["_run_id"] device = await ctx.device() @@ -119,12 +112,12 @@ class TraceTool(Tool): path_spec = params.get("path") if path_spec: path = trace_runner.parse_trace_path(path_spec, contacts) - ctx.console.print(f"[dim]forcing path:[/dim] [brand]{path}[/brand]") + ctx.ui.note(f"[muted]forcing path:[/muted] [brand]{path}[/brand]") else: - ctx.console.print("[dim]path:[/dim] [muted]auto (device-routed)[/muted]") + ctx.ui.note("[muted]path: auto (device-routed)[/muted]") traces: list = [] - with make_progress(ctx.console) as progress: + with ctx.ui.progress("trace") as progress: task = progress.add_task(f"tracing {target}", total=samples) def on_result(done: int, total: int, result) -> None: # noqa: ANN001 @@ -145,8 +138,8 @@ class TraceTool(Tool): # (forced path, or the route the device resolved when auto-routing). current = next((t for t in traces if t.success), None) if traces: - ctx.console.print(traces_table(traces, device_label, resolve, device_hash)) - ctx.console.print( + ctx.ui.show(traces_table(traces, device_label, resolve, device_hash)) + ctx.ui.show( stats_panel(stats, device_label, resolve, route=current, device_hash=device_hash) ) diff --git a/meshtools/tools/tx_optimize.py b/meshtools/tools/tx_optimize.py index aaf4d48..6ad3918 100644 --- a/meshtools/tools/tx_optimize.py +++ b/meshtools/tools/tx_optimize.py @@ -11,14 +11,13 @@ from __future__ import annotations from typing import Any, Optional -import questionary import typer from ..context import AppContext from ..core.connection import DeviceCommandError from ..core.models import Contact from ..services import trace_runner, tx_optimizer -from ..ui.widgets import make_progress, tx_opt_summary, tx_opt_table +from ..ui.widgets import tx_opt_summary, tx_opt_table from ..viz.tx_plot import render_tx_optimization from .base import Tool, ToolResult, register @@ -55,37 +54,33 @@ class TxOptimizeTool(Tool): ) validate_path = lambda v: _validate_link_path(v, contacts) # noqa: E731 if choices: - path_spec = await questionary.autocomplete( - path_prompt, choices=choices, ignore_case=True, validate=validate_path - ).ask_async() + path_spec = await ctx.ui.autocomplete(path_prompt, choices, validate=validate_path) else: - path_spec = await questionary.text(path_prompt, validate=validate_path).ask_async() + path_spec = await ctx.ui.text(path_prompt, validate=validate_path) if not path_spec: return None - samples = await questionary.text( + samples = await ctx.ui.text( f"Traces per TX level? (1-{MAX_SAMPLES})", default="3", validate=_is_valid_sample_count - ).ask_async() + ) if samples is None: return None - tx_min = await questionary.text( + tx_min = await ctx.ui.text( "Lowest TX power to try:", default=str(ctx.settings.tx_opt_min), validate=_is_int - ).ask_async() + ) if tx_min is None: return None - tx_max = await questionary.text( + tx_max = await ctx.ui.text( "Highest TX power to try:", default=str(ctx.settings.tx_opt_max), validate=_is_int - ).ask_async() + ) if tx_max is None: return None - step = await questionary.text( - "Coarse step:", default="3", validate=_is_valid_sample_count - ).ask_async() + step = await ctx.ui.text("Coarse step:", default="3", validate=_is_valid_sample_count) if step is None: return None - apply = await questionary.confirm( + apply = await ctx.ui.confirm( "Set the winning TX power on the node afterward?", default=True - ).ask_async() + ) if apply is None: return None @@ -121,12 +116,10 @@ class TxOptimizeTool(Tool): samples = max(1, min(MAX_SAMPLES, int(params.get("samples", 3)))) if int(params.get("samples", 3)) > MAX_SAMPLES: - ctx.console.print(f"[warn]capping at {MAX_SAMPLES} traces per level[/warn]") + ctx.ui.note(f"[warn]capping at {MAX_SAMPLES} traces per level[/warn]") password = await self._resolve_password(ctx, admin_node, params) - ctx.console.print( - f"[dim]logging in to[/dim] [brand]{admin_node.name}[/brand][dim]…[/dim]" - ) + ctx.ui.note(f"[muted]logging in to[/muted] [brand]{admin_node.name}[/brand] [muted]…[/muted]") if not await device.admin_login(admin_node, password): ctx.admin_store.forget(admin_node) # bad password: don't keep reusing it raise DeviceCommandError( @@ -135,13 +128,13 @@ class TxOptimizeTool(Tool): ) ctx.admin_store.remember(admin_node, password) - ctx.console.print( - f"[dim]tuning[/dim] [brand]{admin_node.name}[/brand] " - f"[dim]→ target[/dim] [brand]{target_label}[/brand] " - f"[dim]via[/dim] [muted]{path}[/muted]" + ctx.ui.note( + f"[muted]tuning[/muted] [brand]{admin_node.name}[/brand] " + f"[muted]→ target[/muted] [brand]{target_label}[/brand] " + f"[muted]via {path}[/muted]" ) - with make_progress(ctx.console) as progress: + with ctx.ui.progress("tx-optimize") as progress: task = progress.add_task(f"optimizing TX -> {target_label}", total=None) def on_level(done: int, total: int, level) -> None: # noqa: ANN001 @@ -173,8 +166,8 @@ class TxOptimizeTool(Tool): if result.applied: ctx.log.info("set TX power %s on %s", result.best_tx, admin_node.name) - ctx.console.print(tx_opt_table(result)) - ctx.console.print(tx_opt_summary(result)) + ctx.ui.show(tx_opt_table(result)) + ctx.ui.show(tx_opt_summary(result)) artifacts: list[str] = [] if params.get("viz", True): @@ -238,9 +231,9 @@ class TxOptimizeTool(Tool): f"no admin password for {admin_node.name!r}; pass --password or run once " "interactively to store it." ) - entered = await questionary.password( - f"Admin password for {admin_node.name}:" - ).ask_async() + entered = await ctx.ui.text( + f"Admin password for {admin_node.name}:", password=True + ) if not entered: raise DeviceCommandError("an admin password is required to tune a remote node.") return entered diff --git a/meshtools/ui/config_editor.py b/meshtools/ui/config_editor.py index bbb9586..5d07501 100644 --- a/meshtools/ui/config_editor.py +++ b/meshtools/ui/config_editor.py @@ -1,10 +1,11 @@ -"""Interactive device-configuration editor (questionary + Rich). +"""Interactive device-configuration editor, rendered in the full-screen session. -Drives the menu flow for the ``config`` tool: it shows the device's current configuration, -lets the user stage changes setting-by-setting (with type-aware prompts and validation), -and handles custom variables, channels, backup/restore and the destructive "danger zone". -It returns an operation list for :class:`~meshtools.tools.config.ConfigTool` to execute and -log; it performs no device writes itself. +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, channels, +backup/restore and the destructive "danger zone". It returns an operation list for +:class:`~meshtools.tools.config.ConfigTool` to execute and log; it performs no device writes +itself (except danger-zone actions, which run immediately and show their result at once). """ from __future__ import annotations @@ -12,9 +13,8 @@ from __future__ import annotations from pathlib import Path from typing import Any, Optional -import questionary -from rich.console import Console from rich.table import Table +from rich.text import Text from ..context import AppContext from ..core.device_config import ( @@ -25,7 +25,7 @@ from ..core.device_config import ( parse_value, settings_by_category, ) -from .menu import _MENU_STYLE +from .tui import Choice, Separator # Menu action sentinels (distinct from setting keys, which are plain strings). _PRESETS = "__presets__" @@ -34,6 +34,7 @@ _CHANNELS = "__channels__" _BACKUP = "__backup__" _RESTORE = "__restore__" _DANGER = "__danger__" +_VIEW = "__view__" _APPLY = "__apply__" _CANCEL = "__cancel__" # Sentinel for the "enter a value myself" option on non-strict enum prompts. @@ -44,13 +45,12 @@ async def edit_config(ctx: AppContext) -> Optional[list[tuple]]: """Run the interactive editor and return the operations to perform. Args: - ctx: Shared application context (provides the connected device and console). + 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. """ - console = ctx.console device = await ctx.device() snapshot = await build_snapshot(device) custom = await device.get_custom_vars() @@ -59,8 +59,7 @@ async def edit_config(ctx: AppContext) -> Optional[list[tuple]]: extra_ops: list[tuple] = [] # custom/channel/backup/restore/danger ops, in order while True: - render_config(console, snapshot, custom, pending) - choice = await _main_menu(snapshot, pending, extra_ops) + choice = await _main_menu(ctx, snapshot, pending, extra_ops) if choice in (None, _CANCEL): return None @@ -68,41 +67,47 @@ async def edit_config(ctx: AppContext) -> Optional[list[tuple]]: ops: list[tuple] = [("set", k, v) for k, v in pending.items()] ops.extend(extra_ops) return ops or None - if choice == _PRESETS: - await _stage_preset(console, pending) + 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(console, custom, extra_ops) + await _stage_custom_var(ctx, custom, extra_ops) elif choice == _CHANNELS: - await _stage_channel(console, extra_ops) + await _stage_channel(ctx, extra_ops) elif choice == _BACKUP: - await _stage_backup(extra_ops) + await _stage_backup(ctx, extra_ops) elif choice == _RESTORE: - await _stage_restore(extra_ops) + 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(console, choice, snapshot, pending) + await _stage_setting(ctx, choice, snapshot, pending) # --- rendering --------------------------------------------------------------- -def render_config( - console: Console, +def config_table( snapshot: dict, custom: dict[str, str], pending: Optional[dict[str, Any]] = None, -) -> None: - """Print the current configuration, overlaying any staged changes. +) -> Table: + """Build the full configuration table, overlaying any staged changes. Args: - console: Console to render into. snapshot: Device snapshot from ``build_snapshot``. custom: Current custom variables. pending: Optional staged changes (setting key -> new value). + + 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) @@ -124,23 +129,31 @@ def render_config( if show_staged: row.append(format_value(spec, pending[spec.key]) if spec.key in pending else "") table.add_row(*row, spec.help) - console.print(table) if custom: - console.print( - "[muted]custom vars:[/muted] " - + ", ".join(f"{k}={v}" for k, v in custom.items()) + 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"custom [muted]({key})[/muted]", value] + if show_staged: + row.append("") + table.add_row(*row, "") + return table -async def _main_menu(snapshot: dict, pending: dict, extra_ops: list) -> Optional[str]: +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. Each setting row shows its current value (and any staged new value) plus a one-line explanation, so the user can see and understand what they're changing in place. """ - choices: list[questionary.Choice | questionary.Separator] = [] + items: list = [] for category, specs in settings_by_category(): - choices.append(questionary.Separator(f"── {category} ──")) + items.append(Separator(f"── {category} ──")) for spec in specs: current = format_value(spec, spec.getter(snapshot)) shown = ( @@ -148,66 +161,62 @@ async def _main_menu(snapshot: dict, pending: dict, extra_ops: list) -> Optional if spec.key in pending else current ) - choices.append( - questionary.Choice( - title=f"{spec.label}: {shown} — {spec.help}", value=spec.key - ) + items.append( + Choice(title=f"{spec.label}: {shown} — {spec.help}", value=spec.key) ) - choices.append(questionary.Separator("── More ──")) - choices.append(questionary.Choice(title="Radio presets (standard configs)", value=_PRESETS)) - choices.append(questionary.Choice(title="Custom / experimental vars", value=_CUSTOM)) - choices.append(questionary.Choice(title="Channels", value=_CHANNELS)) - choices.append(questionary.Choice(title="Backup to file", value=_BACKUP)) - choices.append(questionary.Choice(title="Restore from file", value=_RESTORE)) - choices.append(questionary.Choice(title="⚠ Danger zone", value=_DANGER)) - choices.append(questionary.Separator(" ")) + 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="Channels", value=_CHANNELS)) + 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) - choices.append(questionary.Choice(title=f"✓ Apply ({staged} staged)", value=_APPLY)) - choices.append(questionary.Choice(title="Cancel (discard)", value=_CANCEL)) + items.append(Choice(title=f"✓ Apply ({staged} staged)", value=_APPLY)) + items.append(Choice(title="Cancel (discard)", value=_CANCEL)) - return await questionary.select( - "Edit which setting?", choices=choices, style=_MENU_STYLE, qmark="◆" - ).ask_async() + choice = await ctx.ui.select("Edit which setting?", items) + return _CANCEL if choice is None else choice # --- staging individual changes ---------------------------------------------- async def _stage_setting( - console: Console, 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(spec, current) + value = await _prompt_value(ctx, spec, current) if value is not None: pending[key] = value -async def _prompt_value(spec: SettingSpec, current: Any) -> Any: +async def _prompt_value(ctx: AppContext, spec: SettingSpec, current: Any) -> Any: """Prompt for a typed value for ``spec``, returning ``None`` on cancel.""" if spec.value_type == "bool": - return await questionary.confirm( - f"{spec.label}?", default=bool(current) - ).ask_async() + return await ctx.ui.confirm(f"{spec.label}?", default=bool(current)) if spec.value_type == "enum" and spec.choices is not None: - items: list[questionary.Choice] = [ - questionary.Choice(title=f"{k} — {label}", value=k) - for k, label in spec.choices.items() + items: list = [ + Choice(title=f"{k} — {label}", value=k) for k, label in spec.choices.items() ] # Non-strict enums list the common values for convenience but still accept any # in-range integer, so offer an escape hatch to type one in. if not spec.strict_choices: - items.append(questionary.Choice(title="Other (enter a value)…", value=_OTHER)) - selected = await questionary.select( + items.append(Choice(title="Other (enter a value)…", value=_OTHER)) + selected = await ctx.ui.select( f"{spec.label} — {spec.help}", - choices=items, + items, default=current if current in spec.choices else None, - style=_MENU_STYLE, - ).ask_async() + ) + if selected is None: + return None if selected != _OTHER: return selected # else: fall through to the free-text prompt below. @@ -219,89 +228,78 @@ async def _prompt_value(spec: SettingSpec, current: Any) -> Any: except DeviceConfigError as exc: return str(exc) - raw = await questionary.text( + raw = await ctx.ui.text( f"{spec.label} — {spec.help}", default="" if current is None else str(current), validate=validate, - ).ask_async() + ) return None if raw is None else parse_value(spec, raw) -async def _stage_preset(console: Console, pending: dict[str, Any]) -> None: +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 - choices = [ - questionary.Choice( + items = [ + Choice( title=f"{p.name}: {p.freq} MHz, BW {p.bw}, SF{p.sf}, CR{p.cr} — {p.help}", value=i, ) for i, p in enumerate(RADIO_PRESETS) ] - choices.append(questionary.Choice(title="Cancel", value=None)) - idx = await questionary.select( - "Apply which radio preset?", choices=choices, style=_MENU_STYLE - ).ask_async() + idx = await ctx.ui.select("Apply which radio preset?", items) if idx is None: return preset = RADIO_PRESETS[idx] pending.update(preset.as_settings()) - console.print( - f"[muted]staged preset[/muted] [brand]{preset.name}[/brand] " - "[muted](review the radio rows, then Apply)[/muted]" - ) async def _stage_custom_var( - console: Console, custom: dict[str, str], extra_ops: list[tuple] + ctx: AppContext, custom: dict[str, str], extra_ops: list[tuple] ) -> None: """Prompt for a custom/experimental variable and stage a set operation.""" - key = await questionary.text("Custom variable name:").ask_async() + key = await ctx.ui.text("Custom variable name:") if not key: return - value = await questionary.text( - f"Value for {key}:", default=custom.get(key, "") - ).ask_async() + value = await ctx.ui.text(f"Value for {key}:", default=custom.get(key, "")) if value is None: return extra_ops.append(("set_custom", key.strip(), value)) -async def _stage_channel(console: Console, extra_ops: list[tuple]) -> None: +async def _stage_channel(ctx: AppContext, extra_ops: list[tuple]) -> None: """Prompt for a channel slot's name/secret and stage a set operation.""" - idx_raw = await questionary.text( - "Channel index:", default="0", validate=_is_int - ).ask_async() + idx_raw = await ctx.ui.text("Channel index:", default="0", validate=_is_int) if idx_raw is None: return - name = await questionary.text( - "Channel name (leading # derives the secret from the name):" - ).ask_async() + name = await ctx.ui.text("Channel name (leading # derives the secret from the name):") if not name: return - secret_hex = await questionary.text( + secret_hex = await ctx.ui.text( "Secret (32 hex chars / 16 bytes; blank to derive from name):", validate=_is_optional_secret, - ).ask_async() + ) + if secret_hex is None: + return secret = bytes.fromhex(secret_hex) if secret_hex else None extra_ops.append(("set_channel", int(idx_raw), name, secret)) -async def _stage_backup(extra_ops: list[tuple]) -> None: +async def _stage_backup(ctx: AppContext, extra_ops: list[tuple]) -> None: """Prompt for a backup destination path and stage the operation.""" - path = await questionary.path("Write backup to:", default="meshtools-config.toml").ask_async() + path = await ctx.ui.path("Write backup to:", default="meshtools-config.toml") if path: extra_ops.append(("backup", Path(path))) -async def _stage_restore(extra_ops: list[tuple]) -> None: +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 questionary.path("Restore from:").ask_async() + path = await ctx.ui.path("Restore from:") if not path: return - dry_run = await questionary.confirm( - "Preview changes only (dry run)?", default=True - ).ask_async() + 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))) @@ -309,7 +307,8 @@ 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. + have no meaningful "preview" and are executed the moment they're confirmed; their + output is shown at once in a result window. Returns: ``True`` if the action may have changed device state the editor should re-read @@ -317,19 +316,17 @@ async def _danger_zone(ctx: AppContext, device: Any, snapshot: dict) -> bool: """ from ..tools.config import apply_ops - console = ctx.console - action = await questionary.select( + action = await ctx.ui.select( "⚠ Danger zone (runs immediately on confirmation):", - choices=[ - questionary.Choice("Send advert", value="advert"), - questionary.Choice("Reboot device", value="reboot"), - questionary.Choice("Export private key", value="export_key"), - questionary.Choice("Import private key", value="import_key"), - questionary.Choice("Factory reset (erase all)", value="factory_reset"), - questionary.Choice("Back", value=None), + [ + 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("Back", value=None), ], - style=_MENU_STYLE, - ).ask_async() + ) op: Optional[tuple] = None if action in (None, "Back"): @@ -337,29 +334,38 @@ async def _danger_zone(ctx: AppContext, device: Any, snapshot: dict) -> bool: if action in ("advert", "export_key"): op = (action,) elif action == "import_key": - key_hex = await questionary.text("Private key (hex):", validate=_is_hex).ask_async() - if key_hex and await _confirm_typed(console, "IMPORT"): + 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 questionary.confirm("Reboot the device now?", default=False).ask_async(): + if await ctx.ui.confirm("Reboot the device now?", default=False): op = ("reboot",) elif action == "factory_reset": - console.print("[err]This erases ALL data on the device and cannot be undone.[/err]") - if await _confirm_typed(console, "RESET"): + 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: 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") -async def _confirm_typed(console: Console, word: str) -> bool: +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 questionary.text(f"Type {word!r} to confirm:").ask_async() + typed = await ctx.ui.text( + f"Type {word!r} to confirm:", help_text=warning, validate=None + ) if typed == word: return True - console.print("[muted]confirmation did not match; skipped.[/muted]") + await ctx.ui.view( + Text.from_markup("[muted]confirmation did not match; skipped.[/muted]"), + title="cancelled", + ) return False diff --git a/meshtools/ui/device_picker.py b/meshtools/ui/device_picker.py index abf8190..8554380 100644 --- a/meshtools/ui/device_picker.py +++ b/meshtools/ui/device_picker.py @@ -1,68 +1,63 @@ -"""Interactive companion-device picker (questionary). +"""Interactive companion-device picker, rendered in the full-screen session. -Shown at the start of the interactive menu (and from the ``devices`` tool) when no port -was given explicitly. It lists the discovered devices, marks the remembered "last known -good" one, and preselects it as the default. The chosen device becomes the session's -active device; it is persisted as the new default once it actually connects. +Shown at the start of the interactive menu (and from the ``devices`` tool) when no port was +given explicitly. It lists the discovered devices, marks the remembered "last known good" +one, and preselects it as the default. The chosen device becomes the session's active +device; it is persisted as the new default once it actually connects. """ from __future__ import annotations -from typing import Optional +from typing import TYPE_CHECKING, Optional -import questionary -from rich.console import Console +from rich.text import Text from ..core.device_store import RememberedDevice from ..core.discovery import DiscoveredDevice -from .menu import _MENU_STYLE +from .tui import Choice, Separator + +if TYPE_CHECKING: + from .surface import Ui async def prompt_device( - console: Console, + ui: "Ui", devices: list[DiscoveredDevice], remembered: Optional[RememberedDevice], ) -> Optional[DiscoveredDevice]: """Prompt the user to choose a companion device. Args: - console: Console used for any surrounding messaging. + ui: The interactive UI surface used to render the picker. devices: Discovered devices (likely-LoRa first). remembered: The remembered default, if any, used to mark and preselect a row. Returns: - The chosen :class:`DiscoveredDevice`, or ``None`` if the user cancelled or chose - to skip device selection (e.g. to run against ``--mock`` / configure later). + The chosen :class:`DiscoveredDevice`, or ``None`` if the user cancelled or chose to + skip device selection (e.g. to run against ``--mock`` / configure later). """ if not devices: - console.print( - "[warn]No serial devices detected.[/warn] " - "Plug one in, pass [accent]--port[/accent], or run with [accent]--mock[/accent]." + await ui.view( + Text.from_markup( + "[warn]No serial devices detected.[/warn]\n" + "Plug one in, pass [accent]--port[/accent], or run with [accent]--mock[/accent]." + ), + title="Select a companion device", ) return None - default_choice: Optional[questionary.Choice] = None - choices: list[questionary.Choice | questionary.Separator] = [] + default: Optional[DiscoveredDevice] = None + items: list = [] for device in devices: is_remembered = remembered is not None and remembered.matches(device) star = "★ " if is_remembered else " " vendor = f" [{device.vendor_label}]" if device.vendor_label else "" lora = " · likely LoRa" if device.is_likely_lora else "" - choice = questionary.Choice(title=f"{star}{device.label}{vendor}{lora}", value=device) - choices.append(choice) - # Preselect the remembered device; else default to the first (top-ranked) row. + items.append(Choice(title=f"{star}{device.label}{vendor}{lora}", value=device)) if is_remembered: - default_choice = choice - if default_choice is None: - default_choice = choices[0] # type: ignore[assignment] # always a Choice here + default = device - choices.append(questionary.Separator(" ")) - choices.append(questionary.Choice(title="skip (use --mock / configure later)", value=None)) + items.append(Separator(" ")) + items.append(Choice(title="skip (use --mock / configure later)", value=None)) - return await questionary.select( - "Select a companion device:", - choices=choices, - default=default_choice, - style=_MENU_STYLE, - qmark="◆", - ).ask_async() + return await ui.select("Select a companion device:", items, default=default) diff --git a/meshtools/ui/menu.py b/meshtools/ui/menu.py index 39b7b55..1991209 100644 --- a/meshtools/ui/menu.py +++ b/meshtools/ui/menu.py @@ -1,110 +1,121 @@ """The interactive main menu, built dynamically from the tool registry. -The menu groups registered tools by category and runs the selected tool's interactive -flow. Because it is generated from the registry, adding a tool automatically adds a menu -entry with no changes here. +The menu groups registered tools by category and runs the selected tool's interactive flow +inside the full-screen :class:`~meshtools.ui.tui.session.TuiSession`. Because it is generated +from the registry, adding a tool automatically adds a menu entry with no changes here. The +persistent header shows the connection target and live passive-monitor counters; each tool's +prompts layer as dialogs, and its output appears in a bounded, scrollable result window. """ from __future__ import annotations -import questionary +from rich.text import Text +from .. import __version__ from ..context import AppContext from ..tools import all_tools +from .surface import TuiUi from .theme import make_console -from .widgets import banner - -_MENU_STYLE = questionary.Style( - [ - ("qmark", "fg:#5eead4 bold"), - ("pointer", "fg:#818cf8 bold"), - ("highlighted", "fg:#5eead4 bold"), - ("selected", "fg:#4ade80"), - ("separator", "fg:#475569"), - ] -) - -#: How often (seconds) prompt_toolkit repaints the menu, so the live monitor counters in -#: the header tick up while the user sits at the prompt without pressing a key. -_MENU_REFRESH_S = 1.0 +from .tui import Choice, Separator, TuiSession -class _MenuPrompt: - """The menu's question line, re-rendered with live monitor counts on every repaint. +def _header(ctx: AppContext) -> Text: + """Build the persistent one-line header (target + live monitor status). - questionary formats the ``message`` into the prompt on each render, so an object - whose ``__str__`` reads the current counters gives a genuinely live header (paired - with the select's ``refresh_interval``) at no cost when nothing is happening. + Args: + ctx: The shared application context, read for the target and monitor counters. + + Returns: + A Rich :class:`Text` shown at the top of every screen; re-read on each repaint so + the passive-monitor counters tick live. """ - - def __init__(self, ctx: AppContext) -> None: - """Bind the prompt to the application context. - - Args: - ctx: The shared application context, read for live monitor state. - """ - self._ctx = ctx - - def __str__(self) -> str: - """Return the prompt text plus the current passive-monitor status line.""" - return f"What would you like to do? [{self._ctx.monitor.status_text()}]" + if ctx.mock: + target = "[warn]simulator[/warn]" + elif ctx.profile_name: + target = ctx.profile_name + elif ctx.selected_device is not None: + target = ctx.selected_device.label + else: + target = "[muted]no device[/muted]" + return Text.from_markup( + f"[brand]MeshTools[/brand] [muted]v{__version__}[/muted] · " + f"[muted]device:[/muted] {target} · [muted]{ctx.monitor.status_text()}[/muted]" + ) async def run_menu(ctx: AppContext) -> None: - """Run the interactive menu loop until the user quits. + """Run the interactive menu loop inside the full-screen session until the user quits. - Before showing the menu, prompts for a companion device (unless the simulator is in - use or a port/profile was already chosen explicitly), so the banner reflects the - target and tools connect to the right radio. If passive monitoring was left enabled - in a previous session, it resumes in the background here. + Installs the TUI surface, starts the session, and drives device selection, monitor + resume, and the menu loop on the session's event loop. The full-screen frame is torn + down before the closing message prints. Args: ctx: The shared application context. """ - console = ctx.console - await _select_device_at_startup(ctx) - console.print(banner(ctx.profile_name, ctx.mock, ctx.selected_device)) + session = TuiSession(header=lambda: _header(ctx)) + ctx.ui = TuiUi(session) + + async def main() -> None: + try: + await _startup(ctx) + await _menu_loop(ctx, session) + finally: + # Stop background capture and close its run record even on an unexpected exit. + await ctx.monitor.aclose() + + await session.run(main()) + ctx.console.print("[muted]bye 73![/muted]") + + +async def _menu_loop(ctx: AppContext, session: TuiSession) -> None: + """Show the tool menu and run selections until the user quits. + + Args: + ctx: The shared application context. + session: The running TUI session. + """ + while True: + items: list = [] + current_category: str | None = None + for tool in all_tools(): + if tool.category != current_category: + current_category = tool.category + items.append(Separator(f"── {current_category} ──")) + items.append(Choice(title=f"{tool.name} — {tool.help}", value=tool.name)) + items.append(Separator(" ")) + items.append(Choice(title="quit", value="__quit__")) + + selection = await session.select("What would you like to do?", items) + if selection in (None, "__quit__"): + return + await _run_selection(ctx, selection) + + +async def _startup(ctx: AppContext) -> None: + """Pick a companion device (if needed) and resume passive monitoring. + + Args: + ctx: The shared application context to update with the selection. + """ + if not (ctx.mock or ctx.explicit_selection): + from ..core.discovery import discover_devices + from .device_picker import prompt_device + + devices = discover_devices() + chosen = await prompt_device(ctx.ui, devices, ctx.device_store.load()) + if chosen is not None: + ctx.selected_device = chosen + ctx.port_override = chosen.port await _resume_monitor(ctx) - prompt = _MenuPrompt(ctx) - try: - while True: - choices: list[questionary.Choice | questionary.Separator] = [] - current_category: str | None = None - for tool in all_tools(): - if tool.category != current_category: - current_category = tool.category - choices.append(questionary.Separator(f"── {current_category} ──")) - choices.append( - questionary.Choice(title=f"{tool.name} — {tool.help}", value=tool.name) - ) - choices.append(questionary.Separator(" ")) - choices.append(questionary.Choice(title="quit", value="__quit__")) - - selection = await questionary.select( - prompt, - choices=choices, - style=_MENU_STYLE, - qmark="◆", - refresh_interval=_MENU_REFRESH_S, - ).ask_async() - - if selection in (None, "__quit__"): - console.print("[muted]bye 73![/muted]") - return - - await _run_selection(ctx, selection) - console.print() - finally: - # Stop background capture and close its run record even on an unexpected exit. - await ctx.monitor.aclose() - async def _resume_monitor(ctx: AppContext) -> None: - """Resume passive monitoring if it was left enabled, warning if it can't start. + """Resume passive monitoring if it was left enabled, ignoring a benign start failure. A failure to start (typically no companion device selected) is non-fatal: the preference stays on, so monitoring resumes automatically once a device is available. + The header reflects the resulting state, so nothing needs to be printed here. Args: ctx: The shared application context. @@ -113,36 +124,12 @@ async def _resume_monitor(ctx: AppContext) -> None: return try: await ctx.monitor.start() - ctx.console.print("[muted]passive monitor resumed in the background[/muted]") - except Exception as exc: # noqa: BLE001 - surface, don't crash the menu - ctx.console.print(f"[warn]passive monitor could not start:[/warn] {exc}") - - -async def _select_device_at_startup(ctx: AppContext) -> None: - """Discover devices and let the user pick one, recording it on the context. - - No-ops when the simulator is in use or a ``--port``/``--profile`` was given - explicitly. A cancelled prompt leaves the context unresolved; tools needing the radio - will then surface a clear error (or the user can pick again via the ``devices`` tool). - - Args: - ctx: The shared application context to update with the selection. - """ - if ctx.mock or ctx.explicit_selection: - return - - from ..core.discovery import discover_devices - from .device_picker import prompt_device - - devices = discover_devices() - chosen = await prompt_device(ctx.console, devices, ctx.device_store.load()) - if chosen is not None: - ctx.selected_device = chosen - ctx.port_override = chosen.port + except Exception: # noqa: BLE001 - surface via the header, don't crash the menu + pass async def _run_selection(ctx: AppContext, name: str) -> None: - """Gather parameters for and execute a single tool from the menu. + """Gather parameters for, execute, and present a single tool from the menu. Args: ctx: The shared application context. @@ -152,22 +139,26 @@ async def _run_selection(ctx: AppContext, name: str) -> None: tool = get_tool(name) if tool is None: # pragma: no cover - registry and menu are always in sync - ctx.console.print(f"[err]Unknown tool: {name}[/err]") + ctx.ui.note(f"[err]Unknown tool: {name}[/err]") + await ctx.ui.present(title=name) return try: params = await tool.prompt_params(ctx) if params is None: # user cancelled a prompt + ctx.ui.discard() return result = await tool.execute(ctx, params) except Exception as exc: # noqa: BLE001 - surface errors without crashing the menu - ctx.console.print(f"[err]✗ {tool.name} failed:[/err] {exc}") + ctx.ui.note(f"[err]✗ {tool.name} failed:[/err] {exc}") + await ctx.ui.present(title=tool.name) return if result.message: - ctx.console.print(result.message) + ctx.ui.note(result.message) for artifact in result.artifacts: - ctx.console.print(f"[ok]●[/ok] wrote [accent]{artifact}[/accent]") + ctx.ui.note(f"[ok]●[/ok] wrote [accent]{artifact}[/accent]") + await ctx.ui.present(title=tool.name) # Exposed for tools that need a standalone console outside a context (rare). diff --git a/meshtools/ui/surface.py b/meshtools/ui/surface.py new file mode 100644 index 0000000..21d219e --- /dev/null +++ b/meshtools/ui/surface.py @@ -0,0 +1,279 @@ +"""The UI surface: one small API tools use for input and output, in either front-end. + +Tools never touch ``questionary`` or a raw console anymore — they talk to ``ctx.ui``. Two +implementations back it: + +* :class:`PlainUi` (scripted CLI): prints immediately and uses a Rich progress bar, so CLI + behavior is byte-for-byte what it was before the TUI existed. +* :class:`TuiUi` (interactive menu): collects a tool's output and presents it in a bounded, + scrollable result window, and routes every prompt/progress through the full-screen + :class:`~meshtools.ui.tui.session.TuiSession`. + +The interactive prompt methods are only ever reached from a tool's ``prompt_params`` (menu +only), so :class:`PlainUi` leaves them unsupported. +""" + +from __future__ import annotations + +from typing import Any, Callable, Optional + +from rich.console import Console, Group, RenderableType +from rich.text import Text + +from .tui.session import TuiSession + +#: A validator returns ``True`` when input is acceptable, or an error message to show. +Validator = Callable[[str], "bool | str"] + + +class Ui: + """Abstract UI surface. See :class:`PlainUi` and :class:`TuiUi` for the two backends.""" + + def show(self, *renderables: RenderableType) -> None: + """Display one or more Rich renderables (tables, panels, text).""" + raise NotImplementedError + + def note(self, markup: str) -> None: + """Display a short line of Rich-markup text.""" + raise NotImplementedError + + async def view( + self, renderable: RenderableType, *, title: str = "", footer_hint: str = "" + ) -> None: + """Show a renderable immediately in a dismissable window (prints in CLI mode).""" + raise NotImplementedError + + async def present(self, *, title: str = "") -> None: + """Flush any buffered output to the user (a no-op when output is immediate).""" + + def discard(self) -> None: + """Drop any buffered-but-unshown output (a no-op when output is immediate).""" + + def progress(self, title: str = "Working"): # noqa: ANN201 - context manager, varies by backend + """Return a progress context manager exposing ``add_task``/``advance``/``update``.""" + raise NotImplementedError + + async def select(self, title: str, items: list, *, default: Any = None) -> Any: + """Prompt the user to choose one item; return its value or ``None`` if cancelled.""" + raise NotImplementedError + + async def text( + self, + title: 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.""" + 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 autocomplete( + self, + title: str, + choices: list[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]: + """Prompt for a filesystem path; return it or ``None`` if cancelled.""" + raise NotImplementedError + + +class PlainUi(Ui): + """CLI surface: print directly to the console; interactive prompts are unsupported.""" + + def __init__(self, console: Console) -> None: + """Bind the surface to a Rich console. + + Args: + console: The console tool output is printed to. + """ + self.console = console + + def show(self, *renderables: RenderableType) -> None: + """Print each renderable to the console immediately.""" + for renderable in renderables: + self.console.print(renderable) + + def note(self, markup: str) -> None: + """Print a markup line to the console immediately.""" + self.console.print(markup) + + async def view( + self, renderable: RenderableType, *, title: str = "", footer_hint: str = "" + ) -> None: + """Print the renderable immediately (there is no windowing in CLI mode).""" + self.console.print(renderable) + + def progress(self, title: str = "Working"): # noqa: ANN201 + """Return the Rich progress bar used for scripted runs.""" + from .widgets import make_progress + + return make_progress(self.console) + + def _no_prompt(self) -> RuntimeError: + """Build the error raised if a rich prompt is reached on the non-interactive path.""" + return RuntimeError("interactive prompts are only available in the menu") + + async def select(self, title: str, items: list, *, default: Any = None) -> Any: + """Unsupported in scripted CLI mode.""" + raise self._no_prompt() + + async def text( + self, + title: str, + *, + default: str = "", + validate: Optional[Validator] = None, + help_text: str = "", + password: bool = False, + ) -> Optional[str]: + """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. + + Returns: + The entered string, or ``None`` on EOF / interrupt. + """ + import getpass + + prompt = f"{title} " if not default else f"{title} [{default}] " + while True: + try: + raw = getpass.getpass(prompt) if password else input(prompt) + except (EOFError, KeyboardInterrupt): + return None + value = raw if raw != "" else default + if validate is not None: + result = validate(value) + if result is not True: + self.console.print(f"[err]{result}[/err]") + continue + return value + + async def confirm(self, title: str, *, default: bool = True) -> Optional[bool]: + """Unsupported in scripted CLI mode.""" + raise self._no_prompt() + + async def autocomplete( + self, + title: str, + choices: list[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]: + """Unsupported in scripted CLI mode.""" + raise self._no_prompt() + + +class TuiUi(Ui): + """Interactive surface: buffer output for a result window; route prompts to the session.""" + + def __init__(self, session: TuiSession) -> None: + """Bind the surface to a running session. + + Args: + session: The full-screen session that renders prompts and windows. + """ + self.session = session + self._buffer: list[RenderableType] = [] + + # --- output -------------------------------------------------------------- + + def show(self, *renderables: RenderableType) -> None: + """Collect renderables for the next result window.""" + self._buffer.extend(renderables) + + def note(self, markup: str) -> None: + """Collect a markup line for the next result window.""" + self._buffer.append(Text.from_markup(markup)) + + async def present(self, *, title: str = "") -> None: + """Show everything buffered since the last present in a scrollable window. + + Clears the buffer afterward. Does nothing if nothing was buffered (e.g. a tool + that only produced a file artifact and an empty message). + + Args: + title: Heading for the result window. + """ + if not self._buffer: + return + body = self._buffer[0] if len(self._buffer) == 1 else Group(*self._buffer) + self._buffer = [] + await self.session.scroll(body, title=title) + + def discard(self) -> None: + """Drop any buffered output without showing it.""" + self._buffer = [] + + async def view( + self, renderable: RenderableType, *, title: str = "", footer_hint: str = "" + ) -> None: + """Show a renderable immediately in a dismissable scroll window.""" + await self.session.scroll(renderable, title=title, footer_hint=footer_hint) + + def progress(self, title: str = "Working"): # noqa: ANN201 + """Return a progress dialog context manager for the session.""" + return self.session.progress(title) + + # --- input --------------------------------------------------------------- + + async def select(self, title: str, items: list, *, default: Any = None) -> Any: + """Delegate to the session's select screen.""" + return await self.session.select(title, items, default=default) + + async def text( + self, + title: str, + *, + default: str = "", + validate: Optional[Validator] = None, + help_text: str = "", + password: bool = False, + ) -> Optional[str]: + """Delegate to the session's text screen.""" + return await self.session.text( + title, + default=default, + validate=validate, + help_text=help_text, + password=password, + ) + + async def confirm(self, title: str, *, default: bool = True) -> Optional[bool]: + """Delegate to the session's confirm screen.""" + return await self.session.confirm(title, default=default) + + async def autocomplete( + self, + title: str, + choices: list[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 + ) + + async def path(self, title: 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") diff --git a/meshtools/ui/tui/__init__.py b/meshtools/ui/tui/__init__.py new file mode 100644 index 0000000..f9420fe --- /dev/null +++ b/meshtools/ui/tui/__init__.py @@ -0,0 +1,35 @@ +"""A small, centralized, reusable full-screen text-UI library. + +Built on prompt_toolkit (for the terminal, input, and resize handling) and Rich (for all +content rendering), this package gives the interactive app a consistent, bounded, layered +UI: screens never overflow the terminal, long content scrolls, going deeper stacks dialogs +over a dimmed parent, and Esc always dismisses the current layer. + +Public API: + - :class:`~meshtools.ui.tui.session.TuiSession` — the running session and prompt helpers. + - :class:`~meshtools.ui.tui.select.Choice` / :class:`~meshtools.ui.tui.select.Separator` + — items for ``session.select``. + - Screen classes for advanced/custom layers. +""" + +from __future__ import annotations + +from .progress import ProgressScreen +from .prompt import AutocompleteScreen, ConfirmScreen, TextScreen +from .screen import CANCEL, Screen, ScrollScreen +from .select import Choice, SelectScreen, Separator +from .session import TuiSession + +__all__ = [ + "TuiSession", + "Screen", + "ScrollScreen", + "SelectScreen", + "Choice", + "Separator", + "TextScreen", + "ConfirmScreen", + "AutocompleteScreen", + "ProgressScreen", + "CANCEL", +] diff --git a/meshtools/ui/tui/frame.py b/meshtools/ui/tui/frame.py new file mode 100644 index 0000000..60ce37d --- /dev/null +++ b/meshtools/ui/tui/frame.py @@ -0,0 +1,149 @@ +"""Composition helpers: turn a screen + terminal size into a bounded, framed ANSI view. + +The session owns input and the screen stack; this module owns *layout* — slicing a screen's +body to a scroll viewport, wrapping it in a titled panel, and assembling the persistent +header and footer so the whole view fits the terminal exactly (never overflowing it). +""" + +from __future__ import annotations + +from rich.console import Group, RenderableType +from rich.panel import Panel +from rich.text import Text + +from .render import render_lines +from .screen import Screen, ScrollScreen + + +def _visible_slice(screen: Screen, lines: list[str], viewport: int) -> tuple[list[str], bool, bool]: + """Clamp the screen's scroll and return the visible lines plus clip flags. + + Keeps the screen's cursor line in view (for select/text screens) and clamps the scroll + offset to the content, then pads the slice to exactly ``viewport`` rows so the panel + always fills its allotted height. + + Args: + screen: The screen being rendered (its ``scroll`` is adjusted in place). + lines: The screen's full, unclipped body lines. + viewport: The visible body height in rows. + + Returns: + A tuple of (padded visible lines, more-above, more-below). + """ + total = len(lines) + if isinstance(screen, ScrollScreen): + screen.note_viewport(viewport) + cursor = screen.cursor_line() + if cursor is not None: + if cursor < screen.scroll: + screen.scroll = cursor + elif cursor >= screen.scroll + viewport: + screen.scroll = cursor - viewport + 1 + screen.scroll = max(0, min(screen.scroll, max(0, total - viewport))) + + visible = lines[screen.scroll : screen.scroll + viewport] + more_above = screen.scroll > 0 + more_below = screen.scroll + viewport < total + visible = visible + [""] * (viewport - len(visible)) + return visible, more_above, more_below + + +def _panel(screen: Screen, inner_w: int, viewport: int, active: bool) -> Panel: + """Render a screen's body into a titled, scroll-aware panel. + + Args: + screen: The screen to frame. + inner_w: Inner content width in columns. + viewport: Visible body height in rows. + active: Whether this is the focused (top) screen, brightening its border. + + Returns: + A Rich :class:`Panel` of exactly ``viewport + 2`` rows. + """ + body_lines = screen.render_body(inner_w) + visible, more_above, more_below = _visible_slice(screen, body_lines, viewport) + body = Text.from_ansi("\n".join(visible)) + subtitle = None + if more_above or more_below: + arrow = ("↑" if more_above else " ") + ("↓" if more_below else " ") + subtitle = f"[muted]{arrow} more[/muted]" + return Panel( + body, + title=f"[accent]{screen.title}[/accent]" if screen.title else None, + subtitle=subtitle, + border_style="accent" if active else "muted", + padding=(0, 1), + ) + + +def compose_base( + header: RenderableType, + base: Screen, + footer_hint: str, + cols: int, + rows: int, +) -> str: + """Compose the full-screen ANSI view: header, the base screen's panel, and a footer. + + Args: + header: The persistent header renderable (banner + live status). + base: The screen filling the background (the deepest non-floating layer). + footer_hint: The active screen's key hint, shown at the very bottom. + cols: Terminal width. + rows: Terminal height. + + Returns: + An ANSI string of exactly ``rows`` lines, each within ``cols`` columns. + """ + header_lines = render_lines(header, cols) + header_h = len(header_lines) + viewport = max(1, rows - header_h - 1 - 2) # minus footer(1) and panel border(2) + panel = _panel(base, cols - 4, viewport, active=True) + footer = Text.from_markup(f"[muted]{footer_hint}[/muted]") + group = Group(header, panel, footer) + lines = render_lines(group, cols) + # Guarantee we never exceed the terminal height (pt would otherwise clip unpredictably). + if len(lines) > rows: + lines = lines[:rows] + else: + lines += [""] * (rows - len(lines)) + return "\n".join(lines) + + +def compose_dialog(screen: Screen, cols: int, rows: int) -> str: + """Compose a centered dialog panel for a floating screen, bounded to the terminal. + + Args: + screen: The floating (top) screen. + cols: Terminal width. + rows: Terminal height. + + Returns: + An ANSI string sized to the dialog's content (never larger than the terminal). + """ + max_w = min(cols - 6, 100) + max_h = max(3, rows - 6) + body_lines = screen.render_body(max_w - 4) + viewport = min(max_h, max(1, len(body_lines))) + visible, more_above, more_below = _visible_slice(screen, body_lines, viewport) + body = Text.from_ansi("\n".join(visible)) + subtitle = f"[muted]{screen.footer_hint}[/muted]" + if more_above or more_below: + arrow = ("↑" if more_above else " ") + ("↓" if more_below else " ") + subtitle = f"[muted]{arrow} · {screen.footer_hint}[/muted]" + panel = Panel( + body, + title=f"[accent]{screen.title}[/accent]" if screen.title else None, + subtitle=subtitle, + border_style="accent", + padding=(0, 1), + width=max_w, + ) + return render_to_ansi_dialog(panel, max_w) + + +def render_to_ansi_dialog(panel: Panel, width: int) -> str: + """Render a dialog panel to ANSI at a fixed width (thin wrapper for clarity).""" + from .render import render_to_ansi + + return render_to_ansi(panel, width) diff --git a/meshtools/ui/tui/progress.py b/meshtools/ui/tui/progress.py new file mode 100644 index 0000000..1c0f52c --- /dev/null +++ b/meshtools/ui/tui/progress.py @@ -0,0 +1,139 @@ +"""A progress screen whose handle mimics the slice of ``rich.progress`` the tools use. + +Trace and TX-optimize sweeps drive progress with ``add_task`` / ``advance`` / ``update`` on a +Rich ``Progress`` in CLI mode. In the TUI the same calls update a bounded, centered progress +dialog instead, so tool code stays identical across both front-ends (see +:meth:`meshtools.ui.surface.PlainUi.progress` vs :meth:`~meshtools.ui.surface.TuiUi.progress`). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Optional + +from rich.console import Group +from rich.progress_bar import ProgressBar +from rich.table import Table +from rich.text import Text + +from .render import render_lines +from .screen import Screen + +if TYPE_CHECKING: + from .session import TuiSession + + +@dataclass +class _Task: + """One tracked progress task.""" + + description: str + total: Optional[float] + completed: float = 0.0 + + +class ProgressScreen(Screen): + """A non-interactive dialog showing one or more running progress bars.""" + + footer_hint = "working…" + floating = True + + def __init__(self, title: str = "Working") -> None: + """Start an empty progress dialog with the given heading.""" + super().__init__() + self.title = title + self._tasks: dict[int, _Task] = {} + self._next = 0 + + def add_task(self, description: str, total: Optional[float] = None) -> int: + """Add a task and return its id (mirrors ``rich.progress.Progress.add_task``).""" + task_id = self._next + self._next += 1 + self._tasks[task_id] = _Task(description=description, total=total) + return task_id + + def advance(self, task_id: int, amount: float = 1.0) -> None: + """Advance a task's completed count (mirrors ``Progress.advance``).""" + self._tasks[task_id].completed += amount + + def update( + self, + task_id: int, + *, + description: Optional[str] = None, + completed: Optional[float] = None, + total: Optional[float] = None, + ) -> None: + """Update a task's fields (mirrors the ``Progress.update`` kwargs used).""" + task = self._tasks[task_id] + if description is not None: + task.description = description + if completed is not None: + task.completed = completed + if total is not None: + task.total = total + + def render_body(self, width: int) -> list[str]: + """Render each task as ``description [bar] m/n``.""" + bar_width = max(10, min(40, width - 24)) + table = Table.grid(padding=(0, 1)) + table.add_column() # description + table.add_column() # bar + table.add_column(justify="right") # counts + for task in self._tasks.values(): + bar = ProgressBar( + total=task.total or None, + completed=task.completed, + width=bar_width, + complete_style="brand", + finished_style="ok", + ) + counts = ( + f"{int(task.completed)}/{int(task.total)}" + if task.total + else str(int(task.completed)) + ) + table.add_row(Text(task.description, style="accent"), bar, Text(counts, style="muted")) + body = table if self._tasks else Text("starting…", style="muted") + return render_lines(Group(body), width) + + def handle(self, action: str, data: str = "") -> None: + """Swallow all keys: progress advances with the work, not the keyboard.""" + return + + +@dataclass +class TuiProgress: + """Context manager yielding a :class:`ProgressScreen` handle, pushed for its lifetime. + + Presents the same ``with ... as progress:`` shape as ``make_progress`` so tools need + only swap the factory. Entering pushes the dialog; the ``add_task`` / ``advance`` / + ``update`` methods mutate it and repaint; exiting pops it. + """ + + session: "TuiSession" + title: str = "Working" + _screen: ProgressScreen = field(init=False, default=None) # type: ignore[assignment] + + def __enter__(self) -> ProgressScreen: + """Push the progress dialog and return it as the handle.""" + self._screen = ProgressScreen(self.title) + # Wrap add_task/advance/update so each repaints the session automatically. + for name in ("add_task", "advance", "update"): + setattr(self._screen, name, self._wrap(getattr(self._screen, name))) + self.session.push(self._screen) + return self._screen + + def __exit__(self, *exc: object) -> None: + """Pop the progress dialog.""" + self.session.pop(self._screen) + + def _wrap(self, method): # type: ignore[no-untyped-def] + """Wrap a mutating method so it repaints the session after running.""" + + def wrapped(*args, **kwargs): # type: ignore[no-untyped-def] + result = method(*args, **kwargs) + self.session.invalidate() + return result + + return wrapped diff --git a/meshtools/ui/tui/prompt.py b/meshtools/ui/tui/prompt.py new file mode 100644 index 0000000..b9c5e6c --- /dev/null +++ b/meshtools/ui/tui/prompt.py @@ -0,0 +1,269 @@ +"""Input screens: text, confirm, and autocomplete — the ``questionary`` replacements. + +These reimplement the small slice of ``questionary`` the app uses (a validated line editor, +a yes/no toggle, and a free-text field with suggestions) as uniform TUI screens, so every +prompt shares the framework's look, layering, resize, and Esc-to-cancel behavior. +""" + +from __future__ import annotations + +from typing import Callable, Optional + +from rich.console import Group, RenderableType +from rich.text import Text + +from .render import render_lines +from .screen import Screen + +#: A validator returns ``True`` when the input is acceptable, or an error message to show. +Validator = Callable[[str], "bool | str"] + + +class _LineEditor: + """A minimal single-line text editor (insert, delete, and cursor movement). + + Shared by :class:`TextScreen` and :class:`AutocompleteScreen`. Tracks the text and the + cursor position; rendering shows the cursor as a reverse-video cell. + """ + + def __init__(self, initial: str = "") -> None: + """Start the editor with ``initial`` text and the cursor at its end.""" + self.text = initial + self.cursor = len(initial) + + def edit(self, action: str, data: str = "") -> bool: + """Apply an editing action, returning ``True`` if it changed the buffer/cursor. + + Args: + action: The normalized key action. + data: The character to insert when ``action`` is ``text``. + + Returns: + ``True`` if the action was an editing action handled here. + """ + if action == "text" and data.isprintable(): + self.text = self.text[: self.cursor] + data + self.text[self.cursor :] + self.cursor += len(data) + elif action == "backspace" and self.cursor > 0: + self.text = self.text[: self.cursor - 1] + self.text[self.cursor :] + self.cursor -= 1 + elif action == "delete" and self.cursor < len(self.text): + self.text = self.text[: self.cursor] + self.text[self.cursor + 1 :] + elif action == "left": + self.cursor = max(0, self.cursor - 1) + elif action == "right": + self.cursor = min(len(self.text), self.cursor + 1) + elif action == "home": + self.cursor = 0 + elif action == "end": + self.cursor = len(self.text) + else: + return False + return True + + def render(self, mask: bool = False) -> Text: + """Render the current line with a reverse-video cursor cell. + + Args: + mask: When ``True``, replace each character with a bullet (password entry). + """ + shown = "•" * len(self.text) if mask else self.text + text = Text("› ", style="accent") + before, at, after = ( + shown[: self.cursor], + shown[self.cursor : self.cursor + 1], + shown[self.cursor + 1 :], + ) + text.append(before) + text.append(at or " ", style="reverse") + text.append(after) + return text + + +class TextScreen(Screen): + """A validated single-line text prompt. Resolves with the string, or CANCEL on Esc.""" + + def __init__( + self, + title: str, + *, + default: str = "", + validate: Optional[Validator] = None, + help_text: str = "", + password: bool = False, + footer_hint: str = "Enter accept · Esc cancel", + ) -> None: + """Build a text prompt. + + Args: + title: The question shown above the field. + default: Prefilled text. + validate: Optional validator run on Enter; a returned string is shown as an + error and blocks submission. + help_text: Optional muted hint shown under the field. + password: When ``True``, mask the entered text. + footer_hint: Footer key hint. + """ + super().__init__() + self.title = title + self.footer_hint = footer_hint + self._editor = _LineEditor(default) + self._validate = validate + self._help = help_text + self._password = password + self._error = "" + + 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)] + if self._help: + parts.append(Text(self._help, style="muted")) + 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 buffer, submit on Enter (if valid), or cancel on Esc.""" + if action == "enter": + value = self._editor.text + if self._validate is not None: + result = self._validate(value) + if result is not True: + self._error = str(result) + return + self.resolve(value) + elif action == "escape": + super().handle("escape") + else: + if self._editor.edit(action, data): + self._error = "" + + +class ConfirmScreen(Screen): + """A yes/no prompt. Resolves with a bool, or CANCEL on Esc.""" + + def __init__( + self, + title: str, + *, + default: bool = True, + footer_hint: str = "←→/Y/N choose · Enter accept · Esc cancel", + ) -> None: + """Build a confirm prompt. + + Args: + title: The yes/no question. + default: The initially highlighted answer. + footer_hint: Footer key hint. + """ + super().__init__() + self.title = title + self.footer_hint = footer_hint + self._value = default + + def render_body(self, width: int) -> list[str]: + """Render the Yes / No options with the current choice highlighted.""" + text = Text() + text.append(" Yes ", style="reverse brand" if self._value else "muted") + text.append(" ") + text.append(" No ", style="muted" if self._value else "reverse brand") + return render_lines(text, width) + + def handle(self, action: str, data: str = "") -> None: + """Toggle the choice, submit on Enter, or cancel on Esc.""" + if action in ("left", "right", "tab"): + self._value = not self._value + elif action == "text" and data.lower() in ("y", "n"): + self._value = data.lower() == "y" + elif action == "enter": + self.resolve(self._value) + elif action == "escape": + super().handle("escape") + + +class AutocompleteScreen(Screen): + """A free-text field with a live suggestion list (``questionary.autocomplete``). + + The user may type any value; matching suggestions appear below and can be highlighted + with the arrows and accepted into the field with Tab. Enter commits the typed text + (after validation). Resolves with the string, or CANCEL on Esc. + """ + + #: Cap the suggestion list so it never dominates the dialog. + MAX_SUGGESTIONS = 8 + + def __init__( + self, + title: str, + choices: list[str], + *, + default: str = "", + validate: Optional[Validator] = None, + footer_hint: str = "type · ↑↓ Tab complete · Enter accept · Esc cancel", + ) -> None: + """Build an autocomplete prompt. + + Args: + title: The question shown above the field. + choices: Suggestion strings to match against. + default: Prefilled text. + validate: Optional validator run on Enter. + footer_hint: Footer key hint. + """ + super().__init__() + self.title = title + self.footer_hint = footer_hint + self._editor = _LineEditor(default) + self._choices = choices + self._validate = validate + self._error = "" + self._sugg = 0 + + def _suggestions(self) -> list[str]: + """Return suggestions matching the current text, capped for display.""" + needle = self._editor.text.lower() + if not needle: + matches = list(self._choices) + else: + matches = [c for c in self._choices if needle in c.lower()] + 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()] + suggestions = self._suggestions() + self._sugg = max(0, min(self._sugg, len(suggestions) - 1)) if suggestions else 0 + for i, sug in enumerate(suggestions): + is_sel = i == self._sugg + row = Text(("❯ " if is_sel else " ") + sug, style="brand" if is_sel else "muted") + row.truncate(width) + parts.append(row) + 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, move/accept suggestions, submit on Enter, or cancel on Esc.""" + suggestions = self._suggestions() + if action == "up": + self._sugg = (self._sugg - 1) % len(suggestions) if suggestions else 0 + elif action == "down": + self._sugg = (self._sugg + 1) % len(suggestions) if suggestions else 0 + elif action == "tab": + if suggestions: + self._editor = _LineEditor(suggestions[self._sugg]) + self._error = "" + elif action == "enter": + value = self._editor.text + if self._validate is not None: + result = self._validate(value) + if result is not True: + self._error = str(result) + return + self.resolve(value) + elif action == "escape": + super().handle("escape") + else: + if self._editor.edit(action, data): + self._error = "" + self._sugg = 0 diff --git a/meshtools/ui/tui/render.py b/meshtools/ui/tui/render.py new file mode 100644 index 0000000..c74cc50 --- /dev/null +++ b/meshtools/ui/tui/render.py @@ -0,0 +1,85 @@ +"""The single Rich-to-prompt_toolkit bridge. + +The TUI keeps rendering in Rich (all existing tables, panels, and text widgets are reused +verbatim) and lets prompt_toolkit own the terminal, input, and resize events. This module +is the seam between the two: it renders any Rich renderable to an ANSI string at a chosen +width, which prompt_toolkit then draws via :class:`prompt_toolkit.formatted_text.ANSI`. + +Because rendering is redone at the *current* width on every repaint, content reflows +automatically when the terminal is resized. +""" + +from __future__ import annotations + +from io import StringIO + +from rich.console import Console, RenderableType + +from ..theme import MESH_THEME + +#: Cache one headless render console per width. Consoles are cheap but repaint happens on +#: every keystroke and on the live-monitor timer, so caching avoids needless churn. Keyed +#: by width because a console's width is fixed at construction. +_CONSOLES: dict[int, Console] = {} + + +def _console(width: int) -> Console: + """Return a themed, headless console that emits ANSI at the given width. + + Args: + width: Target render width in columns. + + Returns: + A cached :class:`rich.console.Console` writing to an internal buffer. + """ + console = _CONSOLES.get(width) + if console is None: + console = Console( + theme=MESH_THEME, + width=width, + file=StringIO(), + force_terminal=True, + color_system="truecolor", + highlight=False, + soft_wrap=False, + ) + _CONSOLES[width] = console + return console + + +def render_to_ansi(renderable: RenderableType, width: int) -> str: + """Render a Rich renderable to an ANSI string at ``width`` columns. + + Args: + renderable: Any Rich renderable (table, panel, text, group, markup string). + width: Target width in columns; content wraps/pads to it. + + Returns: + The rendered output as an ANSI-escaped string, without a trailing newline. + """ + width = max(1, width) + console = _console(width) + with console.capture() as capture: + console.print(renderable, end="") + return capture.get() + + +def render_lines(renderable: RenderableType, width: int) -> list[str]: + """Render a Rich renderable to a list of ANSI lines at ``width`` columns. + + Each returned line is an independently styled ANSI string, so the caller can slice a + vertical viewport (for scrolling) without splitting escape sequences. + + Args: + renderable: Any Rich renderable. + width: Target width in columns. + + Returns: + The rendered lines, newline-free. A trailing empty line (from the final newline) + is dropped so line counts match visible rows. + """ + ansi = render_to_ansi(renderable, width) + lines = ansi.split("\n") + if lines and lines[-1] == "": + lines.pop() + return lines diff --git a/meshtools/ui/tui/screen.py b/meshtools/ui/tui/screen.py new file mode 100644 index 0000000..d26dc1c --- /dev/null +++ b/meshtools/ui/tui/screen.py @@ -0,0 +1,182 @@ +"""The screen abstraction: one layer in the TUI's window stack. + +A :class:`Screen` knows how to render its body (as Rich, converted to ANSI lines by +:mod:`~meshtools.ui.tui.render`) and how to react to normalized key *actions* dispatched by +the :class:`~meshtools.ui.tui.session.TuiSession`. Screens never touch prompt_toolkit or the +terminal directly, so they stay small, uniform, and testable. + +Interactive screens (select, text, confirm) resolve an :class:`asyncio.Future` when the user +commits or cancels; the session awaits that future and then pops the screen, giving the +push/await/pop model behind ``await session.select(...)`` and friends. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Optional + +from rich.console import RenderableType + +from .render import render_lines + +#: Sentinel a screen resolves with when the user presses Esc to cancel, distinct from a +#: legitimately selected ``None`` value. +CANCEL = object() + + +class Screen: + """Base class for every TUI layer. + + Attributes: + title: Heading shown at the top of the screen/dialog. + footer_hint: One-line key hint shown in the footer. + scroll: Current vertical scroll offset into the body's rendered lines. + future: Resolved with the screen's result (or :data:`CANCEL`) when it commits. + floating: Whether the session should draw this screen as a centered dialog over + the dimmed screen beneath it (deeper layers float; the base does not). + """ + + title: str = "" + footer_hint: str = "Esc back" + floating: bool = True + + def __init__(self) -> None: + """Initialize scroll state and the (later-assigned) result future.""" + self.scroll = 0 + self.future: Optional[asyncio.Future] = None + + # --- rendering ----------------------------------------------------------- + + def render_body(self, width: int) -> list[str]: + """Return the body's ANSI lines at ``width`` columns (unclipped). + + Args: + width: Inner content width in columns. + + Returns: + One ANSI string per body line; the session slices these to the viewport. + """ + raise NotImplementedError + + def cursor_line(self) -> Optional[int]: + """Return a body line that must stay visible, or ``None`` for free scrolling. + + Selection/text screens return the active row so the session can keep it in view; + plain scroll screens return ``None``. + """ + return None + + # --- input --------------------------------------------------------------- + + def handle(self, action: str, data: str = "") -> None: + """React to a normalized key action from the session. + + Args: + action: One of ``up``, ``down``, ``pageup``, ``pagedown``, ``home``, ``end``, + ``left``, ``right``, ``enter``, ``escape``, ``backspace``, ``delete``, + ``space``, ``tab``, or ``text`` (with ``data`` set to the character). + data: The typed character when ``action`` is ``text``. + """ + if action == "escape": + self.resolve(CANCEL) + + def resolve(self, value: Any) -> None: + """Resolve this screen's future, committing ``value`` (or :data:`CANCEL`). + + Args: + value: The result to hand back to the awaiting caller. + """ + if self.future is not None and not self.future.done(): + self.future.set_result(value) + + # --- scrolling helpers (shared) ------------------------------------------ + + def scroll_by(self, delta: int, total: int, viewport: int) -> None: + """Adjust :attr:`scroll` by ``delta`` lines, clamped to the content. + + Args: + delta: Lines to move (negative scrolls up). + total: Total body line count. + viewport: Visible body height in lines. + """ + self.scroll = _clamp_scroll(self.scroll + delta, total, viewport) + + +def _clamp_scroll(scroll: int, total: int, viewport: int) -> int: + """Clamp a scroll offset to the valid range for the content and viewport. + + Args: + scroll: Proposed offset. + total: Total line count. + viewport: Visible height in lines. + + Returns: + The offset clamped to ``0 .. max(0, total - viewport)``. + """ + return max(0, min(scroll, max(0, total - viewport))) + + +class ScrollScreen(Screen): + """A read-only screen that shows a Rich renderable in a scrollable viewport. + + Used for tool result windows and any long list. Content taller than the viewport + scrolls with the arrow keys / PageUp / PageDown / Home / End; Esc dismisses it. + """ + + def __init__( + self, + renderable: RenderableType, + *, + title: str = "", + footer_hint: str = "↑↓ PgUp/PgDn scroll · Esc back", + floating: bool = True, + ) -> None: + """Wrap a renderable in a dismissable, scrollable screen. + + Args: + renderable: The Rich content to display. + title: Heading for the screen/dialog. + footer_hint: Footer key hint. + floating: Whether to draw as a centered dialog over the parent. + """ + super().__init__() + self._renderable = renderable + self.title = title + self.footer_hint = footer_hint + self.floating = floating + # The last viewport height the session rendered with, so PageUp/PageDown and the + # End key can move by a full page without the session having to pass it in. + self._viewport = 1 + self._total = 1 + + def render_body(self, width: int) -> list[str]: + """Render the wrapped content to ANSI lines and remember the total count.""" + lines = render_lines(self._renderable, width) + self._total = max(1, len(lines)) + return lines + + def note_viewport(self, viewport: int) -> None: + """Record the viewport height the session is about to render with. + + Args: + viewport: Visible body height in lines. + """ + self._viewport = max(1, viewport) + + def handle(self, action: str, data: str = "") -> None: + """Scroll the viewport or dismiss the screen.""" + page = max(1, self._viewport - 1) + if action == "up": + self.scroll_by(-1, self._total, self._viewport) + elif action == "down": + self.scroll_by(1, self._total, self._viewport) + elif action == "pageup": + self.scroll_by(-page, self._total, self._viewport) + elif action in ("pagedown", "space"): + self.scroll_by(page, self._total, self._viewport) + elif action == "home": + self.scroll = 0 + elif action == "end": + self.scroll = max(0, self._total - self._viewport) + elif action in ("escape", "enter"): + self.resolve(None) diff --git a/meshtools/ui/tui/select.py b/meshtools/ui/tui/select.py new file mode 100644 index 0000000..e1acd52 --- /dev/null +++ b/meshtools/ui/tui/select.py @@ -0,0 +1,180 @@ +"""The selectable-list screen: the reusable replacement for ``questionary.select``. + +A :class:`SelectScreen` shows grouped, arrow-navigable choices with type-to-filter, bounded +to the viewport and scrolled to keep the highlighted row visible. Choice values are +arbitrary objects, so the same screen drives the main menu (tool names), the device picker +(device objects), and the config editor (setting keys). +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Optional + +from rich.text import Text + +from .render import render_to_ansi +from .screen import Screen + + +@dataclass +class Choice: + """One selectable row. + + Attributes: + title: Text shown for the row. + value: Value returned when the row is chosen. + """ + + title: str + value: Any + + +@dataclass +class Separator: + """A non-selectable group heading. + + Attributes: + title: The heading text (rendered muted). + """ + + title: str + + +Item = "Choice | Separator" + + +class SelectScreen(Screen): + """A grouped, filterable, single-choice list. + + Resolves with the chosen :class:`Choice` value, or :data:`~meshtools.ui.tui.screen.CANCEL` + if the user presses Esc. + """ + + def __init__( + self, + title: str, + items: list, + *, + default: Any = None, + footer_hint: str = "↑↓ move · type to filter · Enter select · Esc back", + ) -> None: + """Build a select screen. + + Args: + title: Heading shown above the list. + items: A list of :class:`Choice` and :class:`Separator` in display order. + default: A choice value to pre-highlight, if present. + footer_hint: Footer key hint. + """ + super().__init__() + self.title = title + self.footer_hint = footer_hint + self._items = items + self._filter = "" + # Index into the currently-selectable (filtered) choices. + self._index = 0 + if default is not None: + selectable = [it for it in items if isinstance(it, Choice)] + for i, choice in enumerate(selectable): + if choice.value == default: + self._index = i + break + + # --- filtering ----------------------------------------------------------- + + def _rows(self) -> list: + """Return the items to display given the active filter. + + Without a filter, groups and headings show as authored. With a filter, headings + are dropped and only matching choices are shown, flattened. + """ + if not self._filter: + return self._items + needle = self._filter.lower() + return [ + it for it in self._items if isinstance(it, Choice) and needle in it.title.lower() + ] + + def _choices(self, rows: Optional[list] = None) -> list: + """Return just the selectable choices among ``rows`` (or the current rows).""" + rows = self._rows() if rows is None else rows + return [it for it in rows if isinstance(it, Choice)] + + # --- rendering ----------------------------------------------------------- + + def render_body(self, width: int) -> list[str]: + """Render each row to a single ANSI line, the highlighted 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] = [] + if self._filter: + lines.append(render_to_ansi(Text(f"/{self._filter}", style="warn"), width)) + for item in rows: + if isinstance(item, Separator): + lines.append(render_to_ansi(Text(item.title, style="muted"), width)) + continue + is_sel = item is selected + pointer = "❯ " if is_sel else " " + style = "brand" if is_sel else "" + text = Text(pointer + item.title, style=style, no_wrap=True, overflow="ellipsis") + text.truncate(width) + 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)) + return lines + + def cursor_line(self) -> Optional[int]: + """Return the body line index of the highlighted row.""" + return getattr(self, "_cursor", None) + + # --- input --------------------------------------------------------------- + + def handle(self, action: str, data: str = "") -> None: + """Move the highlight, edit the filter, or commit/cancel the selection.""" + choices = self._choices() + if action == "up": + self._index = (self._index - 1) % len(choices) if choices else 0 + elif action == "down": + self._index = (self._index + 1) % len(choices) if choices else 0 + elif action == "home": + self._index = 0 + elif action == "end": + self._index = max(0, len(choices) - 1) + elif action == "enter": + if choices: + self.resolve(choices[self._index].value) + elif action == "escape": + super().handle("escape") + elif action == "backspace": + self._filter = self._filter[:-1] + self._index = 0 + elif action == "text" and data.isprintable(): + self._filter += data + self._index = 0 + + +def _cursor_line(rows: list, selected: Optional[Choice], filtered: bool) -> Optional[int]: + """Compute the rendered body line index of the selected row. + + Args: + rows: The displayed rows (choices and separators). + selected: The currently highlighted choice, if any. + filtered: Whether a filter line precedes the rows (offsetting every row by one). + + Returns: + The zero-based line index of the highlighted row, or ``None`` if nothing is + selected. + """ + if selected is None: + return None + offset = 1 if filtered else 0 + for i, item in enumerate(rows): + if item is selected: + return i + offset + return None diff --git a/meshtools/ui/tui/session.py b/meshtools/ui/tui/session.py new file mode 100644 index 0000000..489e81b --- /dev/null +++ b/meshtools/ui/tui/session.py @@ -0,0 +1,316 @@ +"""The persistent full-screen session: the heart of the TUI library. + +A :class:`TuiSession` owns one prompt_toolkit :class:`Application`, a stack of +:class:`~meshtools.ui.tui.screen.Screen` layers, and the persistent header/footer frame. It +translates prompt_toolkit key events into normalized *actions* dispatched to the top screen, +composes the current view (bounded to the terminal, reflowing on resize) via +:mod:`~meshtools.ui.tui.frame`, and exposes ``async`` helpers (``select``/``text``/ +``confirm``/``autocomplete``/``scroll``/``progress``) that push a screen, await its result, +and pop it — the push/await/pop model behind every prompt. +""" + +from __future__ import annotations + +import asyncio +from typing import Any, Callable, Optional + +from prompt_toolkit.application import Application +from prompt_toolkit.filters import Condition +from prompt_toolkit.formatted_text import ANSI +from prompt_toolkit.key_binding import KeyBindings +from prompt_toolkit.keys import Keys +from prompt_toolkit.layout import ConditionalContainer, Float, FloatContainer, Layout, Window +from prompt_toolkit.layout.controls import FormattedTextControl +from rich.console import RenderableType +from rich.text import Text + +from . import frame +from .progress import TuiProgress +from .prompt import AutocompleteScreen, ConfirmScreen, TextScreen, Validator +from .screen import CANCEL, Screen, ScrollScreen +from .select import Choice, SelectScreen, Separator + +#: Maps prompt_toolkit keys to the normalized action names screens understand. +_KEY_ACTIONS: dict[Any, str] = { + Keys.Up: "up", + Keys.Down: "down", + Keys.Left: "left", + Keys.Right: "right", + Keys.PageUp: "pageup", + Keys.PageDown: "pagedown", + Keys.Home: "home", + Keys.End: "end", + Keys.Enter: "enter", + Keys.Escape: "escape", + Keys.Backspace: "backspace", + Keys.Delete: "delete", + Keys.Tab: "tab", +} + + +class TuiSession: + """A running full-screen TUI: screen stack, frame, input loop, and async prompts.""" + + def __init__( + self, + header: Optional[Callable[[], RenderableType]] = None, + *, + input: Any = None, # noqa: A002 - matches prompt_toolkit's Application(input=) name + output: Any = None, + ) -> None: + """Create a session. + + Args: + header: Callable returning the persistent header renderable (banner + live + status), re-invoked on every repaint. Defaults to a plain title. + input: Optional prompt_toolkit input to drive the app from (tests use a pipe); + defaults to the real terminal. + output: Optional prompt_toolkit output to render to (tests use a dummy); + defaults to the real terminal. + """ + self._stack: list[Screen] = [] + self._header = header or (lambda: Text("MeshTools", style="brand")) + self._app: Optional[Application] = None + self._input = input + self._output = output + + # --- stack --------------------------------------------------------------- + + @property + def top(self) -> Optional[Screen]: + """The active (top-most) screen, or ``None`` when the stack is empty.""" + return self._stack[-1] if self._stack else None + + def push(self, screen: Screen) -> None: + """Push a screen onto the stack and repaint.""" + self._stack.append(screen) + self.invalidate() + + def pop(self, screen: Optional[Screen] = None) -> None: + """Pop ``screen`` (or the top) off the stack and repaint.""" + if not self._stack: + return + if screen is None or self._stack[-1] is screen: + self._stack.pop() + elif screen in self._stack: + self._stack.remove(screen) + self.invalidate() + + def invalidate(self) -> None: + """Request a repaint if the application is running.""" + if self._app is not None: + self._app.invalidate() + + # --- async prompt helpers ------------------------------------------------ + + async def run_screen(self, screen: Screen) -> Any: + """Push a screen, await its result, then pop it. + + Args: + screen: The screen to run. + + Returns: + The screen's resolved value, or :data:`~meshtools.ui.tui.screen.CANCEL`. + """ + loop = asyncio.get_running_loop() + screen.future = loop.create_future() + self.push(screen) + try: + return await screen.future + finally: + self.pop(screen) + + async def select(self, title: str, items: list, *, default: Any = None) -> Any: + """Show a select screen; return the chosen value or ``None`` if cancelled.""" + result = await self.run_screen(SelectScreen(title, items, default=default)) + return None if result is CANCEL else result + + async def text( + self, + title: str, + *, + default: str = "", + validate: Optional[Validator] = None, + help_text: str = "", + password: bool = False, + ) -> Optional[str]: + """Show a text prompt; return the string or ``None`` if cancelled.""" + result = await self.run_screen( + TextScreen( + title, + default=default, + validate=validate, + help_text=help_text, + password=password, + ) + ) + return None if result is CANCEL else result + + async def confirm(self, title: str, *, default: bool = True) -> Optional[bool]: + """Show a yes/no prompt; return the bool or ``None`` if cancelled.""" + result = await self.run_screen(ConfirmScreen(title, default=default)) + return None if result is CANCEL else result + + async def autocomplete( + self, + title: str, + choices: list[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) + ) + return None if result is CANCEL else result + + async def scroll( + self, renderable: RenderableType, *, title: str = "", footer_hint: str = "" + ) -> None: + """Show a dismissable, scrollable view of a renderable (a result window).""" + screen = ScrollScreen( + renderable, + title=title, + footer_hint=footer_hint or "↑↓ PgUp/PgDn scroll · Esc back", + ) + await self.run_screen(screen) + + def progress(self, title: str = "Working") -> TuiProgress: + """Return a progress context manager backed by a pushed :class:`ProgressScreen`.""" + return TuiProgress(self, title) + + # --- application lifecycle ------------------------------------------------ + + async def run(self, main: Any) -> None: + """Run ``main`` (the menu loop) inside the full-screen application. + + Starts the prompt_toolkit event loop, drives ``main`` as a background task on that + same loop, and exits the application when ``main`` returns. Any exception from + ``main`` is re-raised after the loop unwinds. + + Args: + main: The coroutine driving the session (typically the menu loop). + """ + self._app = self._build_app() + box: dict[str, BaseException] = {} + + async def driver() -> None: + try: + await main + except BaseException as exc: # noqa: BLE001 - re-raised after the app unwinds + box["exc"] = exc + finally: + if self._app is not None and not self._app.is_done: + self._app.exit() + + def pre_run() -> None: + asyncio.ensure_future(driver()) + + await self._app.run_async(pre_run=pre_run) + if "exc" in box: + raise box["exc"] + + def _build_app(self) -> Application: + """Construct the prompt_toolkit application, layout, and key bindings.""" + base_control = FormattedTextControl(self._render_base, focusable=True) + base_window = Window(base_control, always_hide_cursor=True) + float_window = Window( + FormattedTextControl(self._render_float), always_hide_cursor=True + ) + root = FloatContainer( + content=base_window, + floats=[ + Float( + ConditionalContainer( + float_window, filter=Condition(self._has_float) + ) + ) + ], + ) + return Application( + layout=Layout(root, focused_element=base_window), + key_bindings=self._key_bindings(), + full_screen=True, + mouse_support=False, + refresh_interval=1.0, # keep the live monitor counter in the header ticking + input=self._input, + output=self._output, + ) + + # --- rendering ----------------------------------------------------------- + + def _size(self) -> tuple[int, int]: + """Return the current (cols, rows) of the terminal.""" + size = self._app.output.get_size() # type: ignore[union-attr] + return max(20, size.columns), max(6, size.rows) + + def _base_screen(self) -> Optional[Screen]: + """The screen drawn as the background (parent of a floating dialog, else the top).""" + if not self._stack: + return None + if self._has_float() and len(self._stack) >= 2: + return self._stack[-2] + return self._stack[-1] + + def _has_float(self) -> bool: + """Whether the top screen should be drawn as a centered dialog over the base.""" + return len(self._stack) >= 2 and bool(getattr(self.top, "floating", False)) + + def _render_base(self) -> ANSI: + """Render the persistent frame around the background screen.""" + cols, rows = self._size() + base = self._base_screen() + if base is None: + return ANSI("") + footer = self.top.footer_hint if self.top else base.footer_hint + return ANSI(frame.compose_base(self._header(), base, footer, cols, rows)) + + def _render_float(self) -> ANSI: + """Render the top screen as a centered dialog (only when floating).""" + if not self._has_float() or self.top is None: + return ANSI("") + cols, rows = self._size() + return ANSI(frame.compose_dialog(self.top, cols, rows)) + + # --- input --------------------------------------------------------------- + + def _key_bindings(self) -> KeyBindings: + """Build the global key bindings that dispatch normalized actions to the top.""" + kb = KeyBindings() + + def bind(key: Any, action: str) -> None: + @kb.add(key) + def _(event: Any, action: str = action) -> None: # noqa: ANN401 + self._dispatch(action) + + for key, action in _KEY_ACTIONS.items(): + bind(key, action) + + @kb.add(Keys.ControlC) + def _quit(event: Any) -> None: # noqa: ANN401 + if self._app is not None: + self._app.exit() + + @kb.add(Keys.Any) + def _typed(event: Any) -> None: # noqa: ANN401 + data = event.data + if data and len(data) == 1 and data.isprintable(): + self._dispatch("text", data) + + return kb + + def _dispatch(self, action: str, data: str = "") -> None: + """Forward an action to the top screen and repaint.""" + if self.top is not None: + self.top.handle(action, data) + self.invalidate() + + +__all__ = [ + "TuiSession", + "Choice", + "Separator", + "SelectScreen", + "ScrollScreen", +] diff --git a/pyproject.toml b/pyproject.toml index 8fd0aa4..39f8e8e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,7 @@ dependencies = [ "pyserial>=3.5", # serial-port enumeration for device discovery "typer>=0.12", # scripted CLI; subcommands mirror the menu "rich>=13.7", # modern console rendering (progress, tables, panels) - "questionary>=2.0", # interactive arrow-key menu navigation + "prompt_toolkit>=3.0", # full-screen interactive TUI (terminal, input, resize) "plotly>=5.20", # interactive self-contained HTML charts "pyvis>=0.3.2", # interactive HTML mesh graphs "networkx>=3.2", # graph model + path scoring diff --git a/tests/test_tui.py b/tests/test_tui.py new file mode 100644 index 0000000..28b6825 --- /dev/null +++ b/tests/test_tui.py @@ -0,0 +1,303 @@ +"""Unit tests for the reusable text-UI library (``meshtools.ui.tui``). + +These exercise the pure logic — ANSI rendering/slicing, selection filtering and navigation, +scroll math, the frame composition's terminal-fit guarantee, prompt editing/validation, and +the progress handle's Rich-``Progress`` parity — without standing up a real prompt_toolkit +application, so they run fast and headless. +""" + +from __future__ import annotations + +import asyncio + +from prompt_toolkit.input.defaults import create_pipe_input +from prompt_toolkit.output import DummyOutput +from rich.table import Table +from rich.text import Text + +from meshtools.ui.tui import frame +from meshtools.ui.tui.progress import ProgressScreen +from meshtools.ui.tui.prompt import AutocompleteScreen, ConfirmScreen, TextScreen +from meshtools.ui.tui.render import render_lines, render_to_ansi +from meshtools.ui.tui.screen import CANCEL, ScrollScreen +from meshtools.ui.tui.select import Choice, SelectScreen, Separator +from meshtools.ui.tui.session import TuiSession + +_UNSET = object() + + +class _Fut: + """A minimal stand-in for an asyncio.Future used to capture a screen's result.""" + + def __init__(self) -> None: + self.result = _UNSET + self._done = False + + def done(self) -> bool: + return self._done + + def set_result(self, value: object) -> None: + self.result = value + self._done = True + + +def _run(screen, action: str, data: str = ""): + """Attach a fake future, dispatch one action, and return the captured result.""" + screen.future = _Fut() + screen.handle(action, data) + return screen.future.result + + +# --- render ------------------------------------------------------------------ + + +def test_render_lines_counts_visible_rows() -> None: + """A three-row table renders to a matching count of ANSI lines (no phantom blank).""" + table = Table(show_header=False, box=None) + table.add_column("a") + for value in ("one", "two", "three"): + table.add_row(value) + lines = render_lines(table, 40) + assert len(lines) == 3 + assert not lines[-1].strip().endswith("\n") + + +def test_render_to_ansi_wraps_to_width() -> None: + """Rendering honors the requested width, wrapping long text onto multiple lines.""" + long = Text("word " * 40) # 200 chars, must wrap under width 20 + lines = render_to_ansi(long, 20).split("\n") + assert len(lines) > 1 + + +# --- select ------------------------------------------------------------------ + + +def _menu() -> SelectScreen: + items = [ + Separator("── group ──"), + Choice("alpha", 1), + Choice("beta", 2), + Choice("gamma", 3), + ] + return SelectScreen("pick", items, default=2) + + +def test_select_default_and_arrow_wrap() -> None: + """The default choice is preselected and Down wraps around the selectable choices.""" + screen = _menu() + assert _run(screen, "enter") == 2 # default is beta + screen = _menu() + screen.handle("down") # beta -> gamma + screen.handle("down") # gamma -> alpha (wrap) + assert _run(screen, "enter") == 1 + + +def test_select_filter_narrows_and_hides_separators() -> None: + """Typing filters to matching choices only; separators drop out while filtering.""" + screen = _menu() + screen.handle("text", "a") # matches alpha, beta(?), gamma -> those containing 'a' + rows = screen._rows() + assert all(isinstance(r, Choice) for r in rows) + assert {r.title for r in rows} == {"alpha", "beta", "gamma"} + screen.handle("text", "l") # now 'al' -> only alpha + assert [r.title for r in screen._rows()] == ["alpha"] + assert _run(screen, "enter") == 1 + + +def test_select_escape_cancels() -> None: + """Esc resolves the sentinel rather than a value.""" + assert _run(_menu(), "escape") is CANCEL + + +def test_select_cursor_line_tracks_selection() -> None: + """The reported cursor line accounts for separators (and the filter line).""" + screen = _menu() # default beta -> row index 2 (sep, alpha, beta) + screen.render_body(40) + assert screen.cursor_line() == 2 + + +# --- scroll ------------------------------------------------------------------ + + +def test_scroll_screen_paging_and_clamp() -> None: + """PageDown advances by a page, End jumps to the bottom, both clamped to content.""" + body = Text("\n".join(f"line {i}" for i in range(100))) + screen = ScrollScreen(body, title="log") + screen.render_body(40) # sets total = 100 + screen.note_viewport(10) + screen.handle("pagedown") + assert screen.scroll == 9 # page = viewport - 1 + screen.handle("end") + assert screen.scroll == 90 # total - viewport + screen.handle("home") + assert screen.scroll == 0 + screen.handle("up") + assert screen.scroll == 0 # clamped, never negative + + +def test_scroll_screen_escape_resolves_none() -> None: + """A result window dismisses to ``None`` on Esc or Enter.""" + assert _run(ScrollScreen(Text("x")), "escape") is None + assert _run(ScrollScreen(Text("x")), "enter") is None + + +# --- frame ------------------------------------------------------------------- + + +def test_compose_base_fills_exactly_terminal_height() -> None: + """The composed base view is exactly ``rows`` lines regardless of content size.""" + tall = Text("\n".join(f"row {i}" for i in range(200))) + screen = ScrollScreen(tall, title="big") + for rows in (10, 24, 50): + out = frame.compose_base(Text("header"), screen, "Esc back", 80, rows) + assert out.count("\n") + 1 == rows + + +def test_compose_dialog_is_bounded() -> None: + """A dialog for a huge renderable never exceeds the terminal height.""" + tall = Text("\n".join(f"row {i}" for i in range(200))) + out = frame.compose_dialog(ScrollScreen(tall, title="d"), 80, 20) + assert out.count("\n") + 1 <= 20 + + +# --- prompts ----------------------------------------------------------------- + + +def test_text_screen_edits_and_validates() -> None: + """Typing edits the buffer; a failing validator blocks submit and shows the error.""" + screen = TextScreen("name?", validate=lambda v: True if v == "ok" else "nope") + for ch in "xy": + screen.handle("text", ch) + screen.future = _Fut() + screen.handle("enter") # 'xy' fails validation + assert not screen.future.done() + assert screen._error == "nope" + screen.handle("backspace") + screen.handle("backspace") + for ch in "ok": + screen.handle("text", ch) + assert _run(screen, "enter") == "ok" + + +def test_text_screen_password_masks() -> None: + """A password field renders bullets, not the typed characters.""" + screen = TextScreen("pw?", password=True) + for ch in "secret": + screen.handle("text", ch) + rendered = "\n".join(screen.render_body(40)) + assert "secret" not in rendered + assert "•" in rendered + + +def test_confirm_toggle_and_default() -> None: + """The confirm toggles with arrows/letters and returns the chosen bool.""" + screen = ConfirmScreen("sure?", default=True) + assert _run(screen, "enter") is True + screen = ConfirmScreen("sure?", default=True) + screen.handle("left") # toggle to No + assert _run(screen, "enter") is False + screen = ConfirmScreen("sure?", default=True) + screen.handle("text", "n") + assert _run(screen, "enter") is False + + +def test_autocomplete_suggests_and_tab_completes() -> None: + """Suggestions match case-insensitively; Tab fills the highlighted one; Enter commits.""" + screen = AutocompleteScreen("target?", ["Alice", "Bob", "alfred"]) + for ch in "al": + screen.handle("text", ch) + assert screen._suggestions() == ["Alice", "alfred"] + screen.handle("down") # highlight 'alfred' + screen.handle("tab") # fill it + assert screen._editor.text == "alfred" + assert _run(screen, "enter") == "alfred" + + +def test_autocomplete_accepts_free_text() -> None: + """Text with no matching suggestion still commits verbatim (e.g. a hex prefix).""" + screen = AutocompleteScreen("target?", ["Alice"]) + for ch in "3d": + screen.handle("text", ch) + assert _run(screen, "enter") == "3d" + + +# --- progress ---------------------------------------------------------------- + + +def test_progress_handle_matches_rich_api() -> None: + """add_task/advance/update mirror the Rich Progress subset the tools rely on.""" + screen = ProgressScreen("work") + task = screen.add_task("tracing", total=5) + screen.advance(task) + screen.advance(task, 2) + assert screen._tasks[task].completed == 3 + screen.update(task, description="tracing more", completed=5, total=5) + assert screen._tasks[task].description == "tracing more" + assert screen._tasks[task].completed == 5 + # Renders without error at a realistic width. + assert screen.render_body(60) + + +# --- session stack ----------------------------------------------------------- + + +async def test_session_runs_and_exits_when_main_returns() -> None: + """The app starts, drives the main coroutine, and exits cleanly when it returns.""" + with create_pipe_input() as inp: + session = TuiSession(input=inp, output=DummyOutput()) + ran = {} + + async def main() -> None: + ran["done"] = True + + await asyncio.wait_for(session.run(main()), timeout=5) + assert ran["done"] is True + + +async def test_session_select_dispatches_piped_keys() -> None: + """A select resolves the value chosen via piped Down+Enter keystrokes, end to end.""" + with create_pipe_input() as inp: + session = TuiSession(input=inp, output=DummyOutput()) + captured = {} + + async def main() -> None: + captured["value"] = await session.select( + "pick", [Choice("a", 1), Choice("b", 2), Choice("c", 3)] + ) + + inp.send_text("\x1b[B\x1b[B\r") # Down, Down, Enter -> third choice + await asyncio.wait_for(session.run(main()), timeout=5) + assert captured["value"] == 3 + + +async def test_session_text_dispatches_typed_keys() -> None: + """A text prompt captures typed characters and commits on Enter, end to end.""" + with create_pipe_input() as inp: + session = TuiSession(input=inp, output=DummyOutput()) + captured = {} + + async def main() -> None: + captured["value"] = await session.text("name?") + + inp.send_text("hi\r") + await asyncio.wait_for(session.run(main()), timeout=5) + assert captured["value"] == "hi" + + +def test_session_stack_and_float_selection() -> None: + """The stack tracks top/base and only floats a deeper screen over its parent.""" + session = TuiSession() + assert session.top is None + base = ScrollScreen(Text("base"), title="base") + session.push(base) + assert session.top is base + assert session._base_screen() is base + assert not session._has_float() # single screen: no float + + dialog = SelectScreen("pick", [Choice("a", 1)]) + session.push(dialog) + assert session._has_float() # deeper screen floats + assert session._base_screen() is base # base is the parent beneath the dialog + session.pop(dialog) + assert session.top is base + assert not session._has_float()