Lift the chat compose bar's byte budget into a reusable gauge and reuse it wherever a field feeds a size-capped LoRa packet. - prompt.py gains byte_counter/byte_style and the DM/CHANNEL byte limits, plus a byte_limit option on TextScreen: a field so limited rides the shared used/limit gauge (right-pinned, green→yellow→orange→red) and blocks Enter once the text tops the cap, with a "Too long by n bytes" note. Ui.text and session.text thread byte_limit through. - chat.py drops its private copies and calls the shared byte_counter; the limits now come from prompt.py. - The courier's "Message for …" entry passes byte_limit=DM_BYTE_LIMIT so a queued message is gauged and capped the same way a live chat message is. Also fixes a courier scheduling bug uncovered here: the "When it's next heard" row carried the value None, indistinguishable from session.select's own cancel sentinel — backing out of the picker silently queued the message. It now carries its own WHEN_HEARD sentinel, so Esc cancels and the row means no schedule. Full suite green (806). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
828 lines
29 KiB
Python
828 lines
29 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 — or, when the whole result is just a line or two of text
|
|
("✓ flood advertisement sent"), as a small centered popup with an OK button instead of
|
|
a full window (see :func:`_collapse_to_message`) — 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 contextlib import asynccontextmanager
|
|
from typing import Any, AsyncIterator, Callable, Optional
|
|
|
|
from rich import box
|
|
from rich.console import Console, Group, RenderableType
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
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"]
|
|
|
|
#: Buffered output no taller than this many lines qualifies for the message dialog —
|
|
#: the OK popup :meth:`TuiUi.present` shows instead of a full result window. Kept small:
|
|
#: a dialog is an acknowledgement, not a reading surface.
|
|
_DIALOG_MAX_LINES = 3
|
|
|
|
#: ... and no line may be wider than this many cells, so the popup stays a tidy box on a
|
|
#: typical terminal and never has to wrap. Anything wider (a long path, a verbose error)
|
|
#: falls back to the scrollable result window, which wraps properly.
|
|
_DIALOG_MAX_CELLS = 76
|
|
|
|
|
|
def _collapse_to_message(buffered: list[RenderableType]) -> Optional[Text]:
|
|
"""Collapse small, text-only buffered output into one dialog message.
|
|
|
|
This is the gate for :meth:`TuiUi.present`'s popup upgrade: output qualifies only
|
|
when everything buffered is plain note text (:class:`Text` — a table or panel from
|
|
``show()`` disqualifies the lot) totalling at most :data:`_DIALOG_MAX_LINES` lines,
|
|
none wider than :data:`_DIALOG_MAX_CELLS` cells. Styling (the ``[ok]``/``[err]``
|
|
markup on outcome notes) is preserved in the joined message.
|
|
|
|
Args:
|
|
buffered: The renderables collected since the last present.
|
|
|
|
Returns:
|
|
The lines joined into one :class:`Text`, or ``None`` when the output belongs in
|
|
the scrollable result window instead.
|
|
"""
|
|
if not buffered or not all(isinstance(item, Text) for item in buffered):
|
|
return None
|
|
lines: list[Text] = []
|
|
for item in buffered:
|
|
lines.extend(item.split("\n") or [item])
|
|
if len(lines) > _DIALOG_MAX_LINES:
|
|
return None
|
|
if any(line.cell_len > _DIALOG_MAX_CELLS for line in lines):
|
|
return None
|
|
return Text("\n").join(lines)
|
|
|
|
|
|
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
|
|
|
|
def busy_overlay(self, message: str = "", *, title: str = ""): # noqa: ANN201 - async CM, varies by backend
|
|
"""Return an async context manager that floats a "working" skeleton while its block runs.
|
|
|
|
In the interactive menu this shows the top-most skeleton card (see
|
|
:meth:`~meshterm.ui.tui.session.TuiSession.busy_overlay`), used to cover the lag of a
|
|
Bluetooth operation that would otherwise leave the screen blank. In scripted CLI mode
|
|
it does nothing (there is no full-screen surface to float over)."""
|
|
raise NotImplementedError
|
|
|
|
async def select(
|
|
self,
|
|
title: str,
|
|
items: list,
|
|
*,
|
|
prompt: str = "",
|
|
default: Any = None,
|
|
wrap: bool = True,
|
|
filterable: bool = True,
|
|
) -> Any:
|
|
"""Prompt the user to choose one item; return its value or ``None`` if cancelled.
|
|
|
|
``prompt`` draws an instruction inside the popup, above the list. Pass
|
|
``filterable=False`` for a short, fixed list so a stray key can't narrow (and
|
|
resize) it.
|
|
"""
|
|
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 confirm_startup(
|
|
self,
|
|
prompt: "str | Text",
|
|
*,
|
|
title: str = "",
|
|
confirm_label: str = "Remove",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
backdrop_items: Optional[list] = None,
|
|
backdrop_default: Any = None,
|
|
) -> bool:
|
|
"""Confirm a destructive action on the startup splash; ``True`` only if committed.
|
|
|
|
``backdrop_items`` (the picker's rows) floats the confirm over a redrawn device list;
|
|
``backdrop_default`` pre-highlights the row it acts on.
|
|
"""
|
|
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 prompt_pin_startup(
|
|
self,
|
|
device_name: str,
|
|
*,
|
|
error: str = "",
|
|
help_text: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Ask for a Bluetooth companion's pairing PIN on the startup splash; ``None`` if cancelled."""
|
|
raise NotImplementedError
|
|
|
|
async def prompt_text_startup(
|
|
self,
|
|
title: str,
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
help_text: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Ask for a line of text on a chromeless startup splash; ``None`` if cancelled."""
|
|
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,
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
help_text: str = "",
|
|
password: bool = False,
|
|
floating: bool = False,
|
|
) -> Optional[str]:
|
|
"""Prompt for a line of text; return it or ``None`` if cancelled.
|
|
|
|
Keep ``title`` short (it heads the popup's border) and put the question/instruction
|
|
in ``prompt`` (drawn above the field), so a text popup reads like the button dialogs.
|
|
``floating`` forces the prompt to draw as a centered popup even with nothing beneath
|
|
it (a mid-flow modal such as a remote-admin password) instead of filling the frame;
|
|
surfaces without a screen stack ignore it.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
async def confirm(self, title: str, *, default: bool = True) -> Optional[bool]:
|
|
"""Prompt yes/no; return the answer or ``None`` if cancelled."""
|
|
raise NotImplementedError
|
|
|
|
async def dialog(
|
|
self,
|
|
prompt: str,
|
|
buttons: list[tuple[str, Any]],
|
|
*,
|
|
title: str = "",
|
|
default: int = 0,
|
|
keys: Optional[dict[str, Any]] = None,
|
|
danger: bool = False,
|
|
destructive: bool = False,
|
|
) -> Any:
|
|
"""Show a centered button dialog; return the chosen value or ``None`` on Esc.
|
|
|
|
The general choose-one popup (a prompt above a row of buttons). Two escalating
|
|
cautionary tiers theme the prompt and border: ``danger`` (amber) for a disruptive
|
|
choice — discarding edits, a reboot — and ``destructive`` (the reserved error red)
|
|
for irreversible data loss, so a delete confirm reads red like its typed-confirm
|
|
sibling. Buttons follow the platform-dialog convention: the safe way out sits on
|
|
the left and the committing action on the right, which is also the sensible
|
|
``default`` so Enter commits it while Esc always backs out.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
async def typed_confirm(
|
|
self, warning: str, word: str, *, title: str = "Are you sure?"
|
|
) -> bool:
|
|
"""Gate a destructive action behind typing ``word``; return whether it was typed."""
|
|
raise NotImplementedError
|
|
|
|
async def autocomplete(
|
|
self,
|
|
title: str,
|
|
choices: list[str],
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
) -> Optional[str]:
|
|
"""Prompt for free text with suggestions; return it or ``None`` if cancelled."""
|
|
raise NotImplementedError
|
|
|
|
async def path(self, title: str, *, prompt: str = "", default: str = "") -> Optional[str]:
|
|
"""Prompt for a filesystem path; return it or ``None`` if cancelled."""
|
|
raise NotImplementedError
|
|
|
|
|
|
def _heading(title: "str | Text") -> Text:
|
|
"""Render a Panel's or Table's hoisted title as an accent heading line.
|
|
|
|
The ``accent`` theme style (bold indigo) is the one bit of emphasis that earns its
|
|
keep here: it marks where each block starts so a scripted run stays scannable, without
|
|
reaching for the interactive menu's framing. It must stay a bare theme name — Rich
|
|
can't combine the ``bold`` keyword with a named style in one string — but ``accent``
|
|
already carries bold, so nothing is lost.
|
|
"""
|
|
text = title if isinstance(title, Text) else Text.from_markup(str(title))
|
|
text.style = "accent"
|
|
return text
|
|
|
|
|
|
def _deframe(renderable: RenderableType) -> list[RenderableType]:
|
|
"""Strip a renderable's chrome for direct terminal output.
|
|
|
|
The scripted CLI prints straight to the terminal, with no result window to hold the
|
|
output, so a tool's heavy Rich framing — a :class:`Panel`'s border, a :class:`Table`'s
|
|
full box and centered title — is chrome between the reader and the data. This lightens
|
|
it to the middle ground: enough styling to stay legible, none of the interactive menu's
|
|
weight. A Panel becomes its title as a plain heading above its (also de-framed) body; a
|
|
Table keeps just a single rule under its header (:data:`~rich.box.SIMPLE_HEAD`) so the
|
|
columns still read, and hoists its title to the same plain heading; a :class:`Group` is
|
|
de-framed member by member. Everything else passes through untouched.
|
|
|
|
The tables and panels are built fresh for each scripted run and discarded after
|
|
printing, so mutating them in place (clearing ``box``/``title``) is safe here.
|
|
|
|
Args:
|
|
renderable: A renderable a tool handed to :meth:`PlainUi.show` or :meth:`view`.
|
|
|
|
Returns:
|
|
The frame-light renderables to print, in order.
|
|
"""
|
|
if isinstance(renderable, Panel):
|
|
out: list[RenderableType] = []
|
|
if renderable.title:
|
|
out.append(_heading(renderable.title))
|
|
out.extend(_deframe(renderable.renderable))
|
|
return out
|
|
if isinstance(renderable, Table):
|
|
out = []
|
|
if renderable.title:
|
|
out.append(_heading(renderable.title))
|
|
renderable.title = None
|
|
renderable.box = box.SIMPLE_HEAD
|
|
out.append(renderable)
|
|
return out
|
|
if isinstance(renderable, Group):
|
|
out = []
|
|
for child in renderable.renderables:
|
|
out.extend(_deframe(child))
|
|
return out
|
|
return [renderable]
|
|
|
|
|
|
class PlainUi(Ui):
|
|
"""CLI surface: print directly to the console; interactive prompts are unsupported.
|
|
|
|
Output is de-framed on the way out (see :func:`_deframe`): a scripted run prints
|
|
to the bare terminal, so a tool's Panels and boxed Tables are lightened to plain
|
|
headings and header-ruled columns — legible and easy to scan, without the framing.
|
|
"""
|
|
|
|
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, de-framed (see :func:`_deframe`)."""
|
|
for renderable in renderables:
|
|
for item in _deframe(renderable):
|
|
self.console.print(item)
|
|
|
|
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, de-framed (there is no windowing in CLI mode)."""
|
|
for item in _deframe(renderable):
|
|
self.console.print(item)
|
|
|
|
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)
|
|
|
|
@asynccontextmanager
|
|
async def busy_overlay(
|
|
self, message: str = "", *, title: str = ""
|
|
) -> AsyncIterator[None]:
|
|
"""Do nothing: the scripted CLI has no full-screen surface to float a skeleton over."""
|
|
yield
|
|
|
|
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,
|
|
*,
|
|
prompt: str = "",
|
|
default: Any = None,
|
|
wrap: bool = True,
|
|
filterable: bool = True,
|
|
) -> 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 confirm_startup(
|
|
self,
|
|
prompt: "str | Text",
|
|
*,
|
|
title: str = "",
|
|
confirm_label: str = "Remove",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
backdrop_items: Optional[list] = None,
|
|
backdrop_default: Any = None,
|
|
) -> bool:
|
|
"""Unsupported in scripted CLI mode — the picker splash is interactive-only."""
|
|
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 prompt_pin_startup(
|
|
self,
|
|
device_name: str,
|
|
*,
|
|
error: str = "",
|
|
help_text: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Unsupported in scripted CLI mode — a PIN must be supplied non-interactively.
|
|
|
|
The scripted path can't pop a dialog, so a PIN-protected device is handled by the
|
|
clean ``DeviceAuthenticationError`` message (pass ``--ble-pin``) rather than a prompt.
|
|
"""
|
|
raise self._no_prompt()
|
|
|
|
async def prompt_text_startup(
|
|
self,
|
|
title: str,
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
help_text: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Unsupported in scripted CLI mode — a network endpoint comes from ``--tcp`` instead."""
|
|
raise self._no_prompt()
|
|
|
|
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,
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
help_text: str = "",
|
|
password: bool = False,
|
|
floating: 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. An in-body
|
|
``prompt`` (used by the interactive popups) is printed once as a lead-in line here.
|
|
``floating`` is a full-screen-popup nicety with no meaning on the plain terminal, so
|
|
it is accepted and ignored.
|
|
|
|
Returns:
|
|
The entered string, or ``None`` on EOF / interrupt.
|
|
"""
|
|
import getpass
|
|
|
|
if prompt:
|
|
self.console.print(prompt)
|
|
head = f"{title} " if not default else f"{title} [{default}] "
|
|
prompt = head
|
|
while True:
|
|
try:
|
|
raw = getpass.getpass(prompt) if password else input(prompt)
|
|
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 dialog(
|
|
self,
|
|
prompt: str,
|
|
buttons: list[tuple[str, Any]],
|
|
*,
|
|
title: str = "",
|
|
default: int = 0,
|
|
keys: Optional[dict[str, Any]] = None,
|
|
danger: bool = False,
|
|
destructive: bool = False,
|
|
) -> Any:
|
|
"""Unsupported in scripted CLI mode."""
|
|
raise self._no_prompt()
|
|
|
|
async def typed_confirm(
|
|
self, warning: str, word: str, *, title: str = "Are you sure?"
|
|
) -> bool:
|
|
"""Unsupported in scripted CLI mode (destructive CLI commands gate on ``--yes``)."""
|
|
raise self._no_prompt()
|
|
|
|
async def autocomplete(
|
|
self,
|
|
title: str,
|
|
choices: list[str],
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
) -> Optional[str]:
|
|
"""Unsupported in scripted CLI mode."""
|
|
raise self._no_prompt()
|
|
|
|
async def path(self, title: str, *, prompt: 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; window or popup to fit.
|
|
|
|
A short, text-only outcome (a line or two of notes — "✓ flood advertisement
|
|
sent") floats as a centered OK dialog over the current screen, so a one-line
|
|
result never commandeers the whole frame; anything bigger, or anything holding
|
|
a table/panel, opens the bounded scrollable result window as before (see
|
|
:func:`_collapse_to_message` for the exact gate). 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 or popup.
|
|
"""
|
|
if not self._buffer:
|
|
return
|
|
buffered = self._buffer
|
|
self._buffer = []
|
|
message = _collapse_to_message(buffered)
|
|
if message is not None:
|
|
await self.session.message_dialog(message, title=title)
|
|
return
|
|
body = buffered[0] if len(buffered) == 1 else Group(*buffered)
|
|
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)
|
|
|
|
def busy_overlay(self, message: str = "", *, title: str = ""): # noqa: ANN201
|
|
"""Float the session's top-most skeleton card while the wrapped block runs."""
|
|
return self.session.busy_overlay(message, title=title)
|
|
|
|
# --- input ---------------------------------------------------------------
|
|
|
|
async def select(
|
|
self,
|
|
title: str,
|
|
items: list,
|
|
*,
|
|
prompt: str = "",
|
|
default: Any = None,
|
|
wrap: bool = True,
|
|
filterable: bool = True,
|
|
) -> Any:
|
|
"""Delegate to the session's select screen."""
|
|
return await self.session.select(
|
|
title, items, prompt=prompt, default=default, wrap=wrap, filterable=filterable
|
|
)
|
|
|
|
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 confirm_startup(
|
|
self,
|
|
prompt: "str | Text",
|
|
*,
|
|
title: str = "",
|
|
confirm_label: str = "Remove",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
backdrop_items: Optional[list] = None,
|
|
backdrop_default: Any = None,
|
|
) -> bool:
|
|
"""Delegate to the session's startup confirm dialog (floated over the picker)."""
|
|
return await self.session.confirm_startup(
|
|
prompt,
|
|
title=title,
|
|
confirm_label=confirm_label,
|
|
banner=banner,
|
|
footnote=footnote,
|
|
backdrop_items=backdrop_items,
|
|
backdrop_default=backdrop_default,
|
|
)
|
|
|
|
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 prompt_pin_startup(
|
|
self,
|
|
device_name: str,
|
|
*,
|
|
error: str = "",
|
|
help_text: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Delegate to the session's startup PIN dialog."""
|
|
return await self.session.prompt_pin_startup(
|
|
device_name, error=error, help_text=help_text, banner=banner, footnote=footnote
|
|
)
|
|
|
|
async def prompt_text_startup(
|
|
self,
|
|
title: str,
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
help_text: str = "",
|
|
banner: Any = None,
|
|
footnote: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
"""Delegate to the session's chromeless startup text dialog."""
|
|
return await self.session.prompt_text_startup(
|
|
title,
|
|
prompt=prompt,
|
|
default=default,
|
|
validate=validate,
|
|
help_text=help_text,
|
|
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,
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
help_text: str = "",
|
|
password: bool = False,
|
|
byte_limit: Optional[int] = None,
|
|
floating: bool = False,
|
|
) -> Optional[str]:
|
|
"""Delegate to the session's text screen."""
|
|
return await self.session.text(
|
|
title,
|
|
prompt=prompt,
|
|
default=default,
|
|
validate=validate,
|
|
help_text=help_text,
|
|
password=password,
|
|
byte_limit=byte_limit,
|
|
floating=floating,
|
|
)
|
|
|
|
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 dialog(
|
|
self,
|
|
prompt: str,
|
|
buttons: list[tuple[str, Any]],
|
|
*,
|
|
title: str = "",
|
|
default: int = 0,
|
|
keys: Optional[dict[str, Any]] = None,
|
|
danger: bool = False,
|
|
destructive: bool = False,
|
|
) -> Any:
|
|
"""Delegate to the session's button dialog, themed cautionary when ``danger``,
|
|
or in the reserved error red when ``destructive`` (irreversible data loss)."""
|
|
tier = "err" if destructive else "warn" if danger else ""
|
|
return await self.session.button_dialog(
|
|
prompt,
|
|
buttons,
|
|
title=title,
|
|
default=default,
|
|
keys=keys,
|
|
prompt_style=tier,
|
|
border_style=tier or "accent",
|
|
)
|
|
|
|
async def typed_confirm(
|
|
self, warning: str, word: str, *, title: str = "Are you sure?"
|
|
) -> bool:
|
|
"""Delegate to the session's typed-confirmation dialog."""
|
|
return await self.session.typed_confirm(warning, word, title=title)
|
|
|
|
async def autocomplete(
|
|
self,
|
|
title: str,
|
|
choices: list[str],
|
|
*,
|
|
prompt: str = "",
|
|
default: str = "",
|
|
validate: Optional[Validator] = None,
|
|
) -> Optional[str]:
|
|
"""Delegate to the session's autocomplete screen."""
|
|
return await self.session.autocomplete(
|
|
title, choices, prompt=prompt, default=default, validate=validate
|
|
)
|
|
|
|
async def path(self, title: str, *, prompt: str = "", default: str = "") -> Optional[str]:
|
|
"""Prompt for a path as free text (with the current value prefilled)."""
|
|
return await self.session.text(
|
|
title, prompt=prompt, default=default, help_text="filesystem path"
|
|
)
|