Jean-Pierre MartineauandClaude Opus 4.8
b692cce463
Reserve the "MeshCore device" tag for hardware we have actually talked to, rather than guessing from the USB vendor ID: - DeviceStore now keeps a registry of every confirmed device (keyed by stable serial id), remembered forever, migrating the old single-record file transparently. - Selecting a device on the splash runs an immediate smoke test (smoke_test_meshcore): a genuine companion is remembered and confirmed; anything else re-prompts with a message to choose another. The probe uses a short handshake window so a non-MeshCore port is rejected cleanly in a few seconds instead of blocking on meshcore's full default timeout. - The probe renders in place on the chromeless splash via a new BusyScreen (animated Braille spinner) driven by session.busy_startup, so the wordmark and box stay put instead of a separate progress dialog. - Silence the meshcore library's console logging (its import calls logging.basicConfig, which printed "INFO:meshcore:..." over the TUI): route it to the JSON-lines file log with a NullHandler guard on root. - The devices tool now marks confirmed devices and stops asserting "LoRa" before a device is proven. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
403 lines
14 KiB
Python
403 lines
14 KiB
Python
"""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:`~meshterm.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 select_startup(
|
|
self,
|
|
title: str,
|
|
items: list,
|
|
*,
|
|
default: Any = None,
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Any:
|
|
"""Choose one item on a chromeless startup splash; ``None`` if skipped."""
|
|
raise NotImplementedError
|
|
|
|
async def notify_startup(
|
|
self,
|
|
renderable: RenderableType,
|
|
*,
|
|
title: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> None:
|
|
"""Show a message on a chromeless startup splash until the user dismisses it."""
|
|
raise NotImplementedError
|
|
|
|
async def busy_startup(
|
|
self,
|
|
message: str,
|
|
coro: Any,
|
|
*,
|
|
title: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Any:
|
|
"""Await ``coro`` while showing a spinner on the startup splash; return its result."""
|
|
raise NotImplementedError
|
|
|
|
async def reorder(self, title: str, labels: list[str]) -> list[int]:
|
|
"""Let the user rearrange rows with the arrows; return the new order of row indices."""
|
|
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 select_startup(
|
|
self,
|
|
title: str,
|
|
items: list,
|
|
*,
|
|
default: Any = None,
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Any:
|
|
"""Unsupported in scripted CLI mode."""
|
|
raise self._no_prompt()
|
|
|
|
async def notify_startup(
|
|
self,
|
|
renderable: RenderableType,
|
|
*,
|
|
title: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> None:
|
|
"""Unsupported in scripted CLI mode."""
|
|
raise self._no_prompt()
|
|
|
|
async def busy_startup(
|
|
self,
|
|
message: str,
|
|
coro: Any,
|
|
*,
|
|
title: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Any:
|
|
"""No splash in scripted CLI mode; just await the task and return its result."""
|
|
return await coro
|
|
|
|
async def reorder(self, title: str, labels: list[str]) -> list[int]:
|
|
"""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 select_startup(
|
|
self,
|
|
title: str,
|
|
items: list,
|
|
*,
|
|
default: Any = None,
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Any:
|
|
"""Delegate to the session's chromeless startup select splash."""
|
|
return await self.session.select_startup(
|
|
title, items, default=default, banner=banner, footnote=footnote
|
|
)
|
|
|
|
async def notify_startup(
|
|
self,
|
|
renderable: RenderableType,
|
|
*,
|
|
title: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> None:
|
|
"""Delegate to the session's chromeless startup message splash."""
|
|
await self.session.notify_startup(
|
|
renderable, title=title, banner=banner, footnote=footnote
|
|
)
|
|
|
|
async def busy_startup(
|
|
self,
|
|
message: str,
|
|
coro: Any,
|
|
*,
|
|
title: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Any:
|
|
"""Delegate to the session's animated-spinner startup splash."""
|
|
return await self.session.busy_startup(
|
|
message, coro, title=title, banner=banner, footnote=footnote
|
|
)
|
|
|
|
async def reorder(self, title: str, labels: list[str]) -> list[int]:
|
|
"""Delegate to the session's reorder screen."""
|
|
return await self.session.reorder(title, labels)
|
|
|
|
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")
|