CI / lint (push) Canceled after 0s
CI / test macos-latest py3.14 (push) Canceled after 0s
CI / test ubuntu-24.04 py3.10 (push) Canceled after 0s
CI / test ubuntu-latest py3.11 (push) Canceled after 0s
CI / test ubuntu-latest py3.12 (push) Canceled after 0s
CI / test ubuntu-latest py3.13 (push) Canceled after 0s
CI / test ubuntu-latest py3.14 (push) Canceled after 0s
CI / test windows-latest py3.10 (push) Canceled after 0s
CI / test windows-latest py3.14 (push) Canceled after 0s
CI / build (push) Canceled after 0s
Issue trailers / check (push) Canceled after 0s
Bring in the 65 upstream commits since 0.11.1, with ^K lock and the self chip. Both sides kept their features. React gives way to Lock in two places. On the PicoCalc lane, Lock keeps the Shift half of F2 and React is ^E only. In the hint, ^E react is the first atom to go when ^K lock has no room. The long-message code now refills the compose line with MentionEditor, which replaced LineEditor upstream.
2443 lines
118 KiB
Python
2443 lines
118 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
||
"""The persistent full-screen session: the central part of the TUI library.
|
||
|
||
A :class:`TuiSession` owns one prompt_toolkit :class:`Application`, a stack of
|
||
:class:`~meshterm.ui.tui.screen.Screen` layers, and the persistent header and footer. It
|
||
does these tasks:
|
||
|
||
- It changes the prompt_toolkit key events into normalized *actions*, and it sends each
|
||
action to the top screen.
|
||
- It composes the current frame through :mod:`~meshterm.ui.tui.frame`. The frame stays in
|
||
the limits of the terminal, and its content flows again when the terminal changes size.
|
||
- It gives ``async`` helpers (``select``/``text``/``confirm``/``autocomplete``/``scroll``/
|
||
``progress``). Each helper pushes a screen, awaits its result, and pops it. This push,
|
||
await, and pop model is behind each prompt.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import os
|
||
import sys
|
||
from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
|
||
from contextlib import AbstractAsyncContextManager, asynccontextmanager, nullcontext, suppress
|
||
from typing import Any
|
||
|
||
from prompt_toolkit.application import Application
|
||
from prompt_toolkit.data_structures import Size
|
||
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.output.color_depth import ColorDepth
|
||
from prompt_toolkit.utils import get_cwidth
|
||
from rich.console import RenderableType
|
||
from rich.text import Text
|
||
|
||
from ...core import win32dll
|
||
from ...persistence.logging import get_logger
|
||
from ...platforms import get_platform
|
||
from ...services import hold_to_quit, modifier_watch
|
||
from . import colsnap, fastrender, fkeys, frame
|
||
from .emoji_width import ClusterTextControl
|
||
from .holdquit import EscHoldWatch, Quiet, WatchedEsc
|
||
from .overlay import BusyOverlay
|
||
from .progress import TuiProgress
|
||
from .prompt import (
|
||
AutocompleteScreen,
|
||
ButtonDialog,
|
||
ConfirmScreen,
|
||
PinDialog,
|
||
TextScreen,
|
||
TypedConfirmDialog,
|
||
Validator,
|
||
)
|
||
from .screen import CANCEL, POP_ALL, BusyDialog, BusyScreen, PopToMenu, Screen, ScrollScreen
|
||
from .select import Choice, ReorderScreen, SelectScreen, Separator
|
||
from .spinner import spinner_interval
|
||
|
||
#: Each Ctrl-letter chord that the app binds, indexed by the bare lowercase letter. This
|
||
#: table is the single source of truth for the two halves of the life of a chord:
|
||
#:
|
||
#: - the ``Keys.Control*`` bindings, which are folded into :data:`_KEY_ACTIONS` below.
|
||
#: - the right-Ctrl rescue in :meth:`TuiSession._dispatch` (refer to
|
||
#: :func:`_right_ctrl_down`). When a keyboard layout claims the right Ctrl key, that key
|
||
#: sends a bare letter as text. The rescue changes that letter back into the chord.
|
||
#:
|
||
#: Add a chord here, and the two Ctrl keys get it. A second edit is not necessary, and no
|
||
#: chord works on only one side of the keyboard. ``quit`` and ``paste_clipboard`` belong to
|
||
#: the session (``_dispatch`` answers them). The other entries are actions that the session
|
||
#: sends to the top screen. While the right Ctrl key is held, a bare letter gets to the text
|
||
#: only when the layout has no third-level glyph for that key (that is, when the key press is
|
||
#: the chord). A real third-level character arrives as a different glyph, and it never
|
||
#: matches this table.
|
||
#: ``m``/``i``/``h`` are not free: prompt_toolkit spells Enter, Tab, and Backspace as
|
||
#: ``Keys.ControlM``/``ControlI``/``ControlH``. If this table claims them, it binds those
|
||
#: keys again.
|
||
#: The letter of a chord is the mnemonic of the action, not of the word that one screen uses
|
||
#: for it. ``locate`` is ^L: on the map and in the mesh walk it goes to our node, and in the
|
||
#: chat and on Contacts it opens the map on the node of the selected row. It was ^U until
|
||
#: 2026-10-01, then ^Y until 2026-10-10, when ``login`` moved from ^L to ^O (JP). On
|
||
#: 2026-10-11, ``login`` moved again to ^G, and ^O became ``author_page``: the node page of
|
||
#: the author of the selected message, the same as ^L opens the map on that author (JP).
|
||
#: One action for each letter in all screens is the preference, not the rule (JP,
|
||
#: 2026-10-10). A screen that has no use for the action of a letter can give the letter a
|
||
#: different action, if no terminal or multiplexer takes it. ^A and ^B are not free for
|
||
#: this reason: they are the prefix keys of GNU screen and tmux.
|
||
#:
|
||
#: Two of these chords are for the full app, not for one screen:
|
||
#:
|
||
#: - ``to_menu`` (^W) unwinds the full navigation stack back to the main menu.
|
||
#: - ``quit`` (^Q, and also the older ^C) asks to quit the app from any screen. A second
|
||
#: press while the question is open quits (refer to :meth:`TuiSession.request_quit`).
|
||
#:
|
||
#: No footer hint and no F-key chip shows these two chords. The F-key lane has three free
|
||
#: slots for each screen, and a chip for a global verb uses one slot on each screen permanently
|
||
#: (JP, 2026-08-30). The only exception is ``^Q quit?`` in the hint of the main menu. There
|
||
#: the Esc key does nothing, and the user looks for the way out (issue #22). We chose ^W
|
||
#: because users press it by reflex to close everything. On the Canadian Multilingual layout,
|
||
#: we checked that the keyboard sends the two chords (^W ``U+0017``, ^Q ``U+0011``), and the
|
||
#: terminal does not consume either of them. The raw mode of prompt_toolkit clears
|
||
#: ``IXON``/``IXOFF``, so ^Q is not XON here.
|
||
_CTRL_LETTER_CHORDS: dict[str, str] = {
|
||
"c": "quit",
|
||
# ^E reacts to the selected message with an emoji. The raw mode of prompt_toolkit has no
|
||
# line editing, so ^E is not "end of line" here (the End key moves the caret).
|
||
"e": "react",
|
||
# ^G logs in to the room that a room screen shows. In the input, ^G is not a special key:
|
||
# it is the bell only in output, and no terminal or multiplexer takes it.
|
||
"g": "login",
|
||
# ^K locks or unlocks the author of the selected message in a conversation (JP,
|
||
# 2026-10-19). No terminal or multiplexer takes it, and the raw mode of prompt_toolkit
|
||
# has no binding for it.
|
||
"k": "lock",
|
||
"l": "locate",
|
||
# ^O opens the node page of the author of the selected message. The raw mode of
|
||
# prompt_toolkit clears IEXTEN, so a terminal on macOS or BSD does not take ^O to
|
||
# discard output.
|
||
"o": "author_page",
|
||
"p": "paths",
|
||
"q": "quit",
|
||
"r": "retry",
|
||
# ^S is the XOFF key. It is safe here for the same reason that ^Q is not XON (refer to
|
||
# the text above): the raw mode clears IXON/IXOFF. Thus the terminal never takes it to
|
||
# freeze the screen.
|
||
"s": "reveal",
|
||
"u": "url_code",
|
||
"v": "paste_clipboard",
|
||
"w": "to_menu",
|
||
}
|
||
|
||
#: Maps the prompt_toolkit keys to the normalized action names that the screens understand.
|
||
_KEY_ACTIONS: dict[Any, str] = {
|
||
Keys.Up: "up",
|
||
Keys.Down: "down",
|
||
Keys.Left: "left",
|
||
Keys.Right: "right",
|
||
Keys.ShiftUp: "shift_up",
|
||
Keys.ShiftDown: "shift_down",
|
||
Keys.ShiftLeft: "shift_left",
|
||
Keys.ShiftRight: "shift_right",
|
||
Keys.PageUp: "pageup",
|
||
Keys.PageDown: "pagedown",
|
||
Keys.Home: "home",
|
||
Keys.End: "end",
|
||
Keys.ControlHome: "ctrl_home",
|
||
Keys.ControlEnd: "ctrl_end",
|
||
Keys.ControlPageUp: "ctrl_pageup",
|
||
Keys.ControlPageDown: "ctrl_pagedown",
|
||
Keys.ControlLeft: "ctrl_left",
|
||
Keys.ControlRight: "ctrl_right",
|
||
Keys.ControlUp: "ctrl_up",
|
||
Keys.ControlDown: "ctrl_down",
|
||
Keys.Enter: "enter",
|
||
# Ctrl+Enter, as well as a terminal can spell it. No such keycode exists: Enter is ^M.
|
||
# Thus a console without modifyOtherKeys has nothing more to modify, and it sends a
|
||
# plain CR. Most consoles send LF (``ControlJ`` in prompt_toolkit). LF has no other
|
||
# binding here (it is an unprintable character, and ``_typed`` removes it). Thus this
|
||
# binding costs nothing, and it works on the terminals that send LF. The right-Ctrl
|
||
# rescue below covers the other cases through _CTRL_CHORDS. On a console that spells
|
||
# neither, the chip on the F-key lane is the affordance (that is already true on the
|
||
# platform of that console).
|
||
Keys.ControlJ: "ctrl_enter",
|
||
Keys.Escape: "escape",
|
||
Keys.Backspace: "backspace",
|
||
Keys.Delete: "delete",
|
||
Keys.ShiftDelete: "shift_delete",
|
||
Keys.Tab: "tab",
|
||
Keys.BackTab: "shift_tab",
|
||
# The function keys, for the F-key lane of a handheld. This table binds all of them,
|
||
# because the deck of the platform decides which keycodes drive the lane, not this
|
||
# table. The MCU of the PicoCalc sends its Shift bank as F6–F10. The emulator of the
|
||
# Cardputer sends Shift+F4..F8 the same as xterm, and prompt_toolkit reads them as
|
||
# F16–F20. In _dispatch, the session resolves them against the lane of the top screen.
|
||
# A key that the deck does not use, or a key on a platform without a lane, resolves to
|
||
# nothing and is ignored. On the PicoCalc, Alt+Fn is the VT switch of the kernel, and
|
||
# it must never have a binding.
|
||
**{getattr(Keys, f"F{n}"): f"f{n}" for n in range(1, 25)},
|
||
# The Ctrl-letter chords, made from the one table above. Thus a chord can never have a
|
||
# binding without its right-Ctrl rescue (or a rescue into an action that nothing binds).
|
||
**{
|
||
getattr(Keys, f"Control{letter.upper()}"): action
|
||
for letter, action in _CTRL_LETTER_CHORDS.items()
|
||
},
|
||
}
|
||
|
||
#: The plain actions that have a related Ctrl chord, for the right-Ctrl rescue in
|
||
#: :meth:`TuiSession._dispatch` (refer to :func:`_right_ctrl_down`). These are the
|
||
#: navigation keys, and also ``enter``. A terminal cannot reliably spell ``enter`` as a chord
|
||
#: (refer to ``Keys.ControlJ`` above). Thus, for ``enter``, the rescue is not a fallback: it
|
||
#: is the more reliable of the two ways.
|
||
_CTRL_CHORDS: dict[str, str] = {
|
||
"up": "ctrl_up",
|
||
"down": "ctrl_down",
|
||
"left": "ctrl_left",
|
||
"right": "ctrl_right",
|
||
"home": "ctrl_home",
|
||
"end": "ctrl_end",
|
||
"pageup": "ctrl_pageup",
|
||
"pagedown": "ctrl_pagedown",
|
||
"enter": "ctrl_enter",
|
||
}
|
||
|
||
|
||
def _right_ctrl_down() -> bool:
|
||
"""Whether the right Ctrl key is physically held now (Windows, ``False`` on other systems).
|
||
|
||
This function is the rescue behind the right-Ctrl chords. Some keyboard layouts claim the
|
||
right Ctrl key as a modifier for a character group. (The Canadian Multilingual Standard
|
||
uses it to get to a third character level.) These layouts can send an arrow with the
|
||
right Ctrl key to the console as a bare arrow. Then no Ctrl flag stays for prompt_toolkit
|
||
to map, so only the left Ctrl key could control a sortable list. This function probes the
|
||
physical state of the key (``GetAsyncKeyState``). It does not trust the event modifiers,
|
||
because the layout removed them. If an arrow got to our dispatch, our terminal had the
|
||
focus. Thus a right Ctrl key that is held at that instant is a chord that the user makes.
|
||
The platforms other than Windows report ``False``. A VT terminal itself encodes the Ctrl
|
||
modifier the same for the two sides, and there is no key state for each side to examine.
|
||
"""
|
||
if sys.platform != "win32":
|
||
return False
|
||
try:
|
||
return bool(win32dll.user32().GetAsyncKeyState(0xA3) & 0x8000) # VK_RCONTROL
|
||
except Exception: # noqa: BLE001 - if the probe fails, the plain action stays
|
||
return False
|
||
|
||
|
||
def _read_clipboard() -> str:
|
||
"""Try to read the Unicode text of the OS clipboard (Windows, ``""`` on other systems).
|
||
|
||
This function is the fallback behind Ctrl-V on a terminal that sends the key as it is,
|
||
instead of as a bracketed paste (refer to :meth:`TuiSession._dispatch`). It gets the text
|
||
of the clipboard directly from the Win32 API. Each failure gives ``""``: a platform that
|
||
is not Windows, an empty clipboard, a clipboard without text, or a clipboard that another
|
||
program uses, so that we could not open it. Thus the paste does nothing, and no exception
|
||
goes into the key handler. The calls that return a pointer declare a ``c_void_p`` result,
|
||
so that a 64-bit handle is not truncated to an int.
|
||
"""
|
||
if sys.platform != "win32":
|
||
return ""
|
||
try:
|
||
import ctypes
|
||
|
||
CF_UNICODETEXT = 13
|
||
# Private handles. Without them, the argtypes below are declared on objects that all
|
||
# the libraries in the process share. Refer to meshterm.core.win32dll.
|
||
user32 = win32dll.user32()
|
||
kernel32 = win32dll.kernel32()
|
||
user32.GetClipboardData.restype = ctypes.c_void_p
|
||
user32.GetClipboardData.argtypes = [ctypes.c_uint]
|
||
kernel32.GlobalLock.restype = ctypes.c_void_p
|
||
kernel32.GlobalLock.argtypes = [ctypes.c_void_p]
|
||
kernel32.GlobalUnlock.argtypes = [ctypes.c_void_p]
|
||
if not user32.OpenClipboard(None):
|
||
return ""
|
||
try:
|
||
handle = user32.GetClipboardData(CF_UNICODETEXT)
|
||
if not handle:
|
||
return ""
|
||
ptr = kernel32.GlobalLock(handle)
|
||
if not ptr:
|
||
return ""
|
||
try:
|
||
return ctypes.wstring_at(ptr)
|
||
finally:
|
||
kernel32.GlobalUnlock(handle)
|
||
finally:
|
||
user32.CloseClipboard()
|
||
except Exception: # noqa: BLE001 - a small clipboard problem must never break a key press
|
||
return ""
|
||
|
||
|
||
def _reclaim_last_column() -> bool | None:
|
||
"""Whether to use the last column of the terminal, or ``None`` to ask the terminal.
|
||
|
||
The Windows console output of prompt_toolkit reports the window as one column narrower
|
||
than it is. It does this on purpose (refer to :func:`_probe_hides_last_column`). Thus the
|
||
frame is drawn to ``columns - 1``, and the real last column is not used. The user can see
|
||
that column at the right of the border, and can select it. When this value is on,
|
||
:class:`_WidthExtendedOutput` tells the renderer and the frame compositor that the window
|
||
is one column wider. Then that last column is drawn.
|
||
|
||
This is correct only where the probe hid the column. In all other cases, the extra column
|
||
is outside the real screen. prompt_toolkit draws with autowrap off, so the phantom last
|
||
cell of each row is written over the real cell before it. Thus the text at the right edge
|
||
loses its last character. On Linux, this cut the battery gauge of the header: a full
|
||
battery showed ``10`` instead of ``100``, and 99% showed ``9%``. Thus the default is a
|
||
question, not an answer. ``None`` means "reclaim if, and only if, the output this session
|
||
draws on is one whose probe hides the column". The session decides this when that output
|
||
exists.
|
||
|
||
Three sources give the answer, in the order of confidence, because the user at the
|
||
keyboard can see the column and the code can only infer it:
|
||
|
||
1. ``MESHTERM_FULL_WIDTH=0``/``=1``, for one run.
|
||
2. The ``full_width`` preference (``yes``/``no``). ``auto``, the default, gives no
|
||
answer.
|
||
3. The platform. It answers ``False`` where its console is exact
|
||
(:data:`~meshterm.platforms.PICOCALC_LYRA`). In the other cases, it lets the terminal
|
||
decide.
|
||
|
||
The function reads the values again at each call (never at import time, into a cache).
|
||
Thus it agrees with the platform that :func:`~meshterm.platforms.set_platform` installed
|
||
for this process. Refer to the docstring of that module for the reason why a cached or
|
||
imported copy of the platform becomes old.
|
||
"""
|
||
from ...core.preferences import current as current_preferences
|
||
|
||
override = os.environ.get("MESHTERM_FULL_WIDTH")
|
||
if override is not None:
|
||
return override != "0"
|
||
preferred = current_preferences().full_width
|
||
if preferred != "auto":
|
||
return preferred == "yes"
|
||
return None if get_platform().width_reclaim else False
|
||
|
||
|
||
def _probe_hides_last_column(output: Any) -> bool:
|
||
"""Whether prompt_toolkit gives ``output`` a width one column narrower than its window.
|
||
|
||
True for its Windows console output, and for no other output. ``Win32Output.get_size``
|
||
takes the visible width as ``srWindow.Right - srWindow.Left``. This is an inclusive span,
|
||
so it is one column short. Then the function limits the width to less than the right
|
||
margin of the buffer ("windows will wrap otherwise"). ``Windows10_Output`` and
|
||
``ConEmuOutput`` both use the size of the ``Win32Output`` that they hold. The
|
||
``Vt100_Output`` of a POSIX terminal reads ``TIOCGWINSZ``, which is exact. Thus there is
|
||
no column to reclaim there.
|
||
"""
|
||
if sys.platform != "win32":
|
||
return False # the win32 module of prompt_toolkit asserts the platform on import
|
||
from prompt_toolkit.output.win32 import Win32Output
|
||
|
||
return isinstance(output, Win32Output) or isinstance(
|
||
getattr(output, "win32_output", None), Win32Output
|
||
)
|
||
|
||
|
||
#: The depth for each value of the ``color_depth`` preference. ``auto`` is not in the table
|
||
#: on purpose: it is the only value that is not a depth, and :func:`_color_depth` resolves it
|
||
#: separately.
|
||
_DEPTH_BY_NAME = {
|
||
"truecolor": ColorDepth.DEPTH_24_BIT,
|
||
"256": ColorDepth.DEPTH_8_BIT,
|
||
"16": ColorDepth.DEPTH_4_BIT,
|
||
}
|
||
|
||
#: The ``COLORTERM`` values of a terminal that tells the truth about 24-bit colour. There is
|
||
#: no in-band way to ask. A terminal without truecolor answers a truecolor SGR with an
|
||
#: approximation, and it does not tell us. From here, that looks exactly the same as a
|
||
#: terminal with truecolor. Thus this convention, which each truecolor terminal follows, is
|
||
#: all the evidence.
|
||
_TRUECOLOR_COLORTERM = frozenset({"truecolor", "24bit"})
|
||
|
||
|
||
def _color_depth() -> ColorDepth | None:
|
||
"""How many colours to send to this terminal, or ``None`` to keep the prompt_toolkit verdict.
|
||
|
||
prompt_toolkit chooses a depth for each *output class*, and its two classes do not agree.
|
||
``Windows10_Output`` returns ``TRUE_COLOR`` directly. ``Vt100_Output`` returns
|
||
``DEPTH_8_BIT`` for each ``TERM`` except ``linux`` and a dumb terminal. Thus the same
|
||
build of MeshTerm, with the same theme, has 24-bit colour on Windows and 256 colours on
|
||
macOS and Linux. The quantization is silent, and for this reason nobody saw it: the app
|
||
looks correct, only flatter. But the quantization damages exactly the thing that this
|
||
palette is made of. The seven ``heat.*`` steps and the hue wheel for the nodes
|
||
(:func:`~meshterm.ui.theme.node_style`) are pastels that are near each other, and we
|
||
chose them so that the user can tell them apart. In the 216-colour cube, neighbours become
|
||
the same colour, and the user cannot tell the identities apart by hue. prompt_toolkit
|
||
does not read ``COLORTERM``. It reads only ``PROMPT_TOOLKIT_COLOR_DEPTH``. Thus a
|
||
terminal that announces truecolor in the usual way still gets 256 colours.
|
||
|
||
Three sources give the answer, in the order of confidence, the same as in
|
||
:func:`_reclaim_last_column`:
|
||
|
||
1. ``MESHTERM_COLOR_DEPTH``, for one run.
|
||
2. The ``color_depth`` preference.
|
||
3. A reading of the environment.
|
||
|
||
The reading only increases the verdict. It never decreases it. If a terminal does not
|
||
give a positive claim, the function returns ``None``, and the terminal keeps exactly the
|
||
depth that it had. Thus this function cannot make a host worse when it does not know that
|
||
host. This is more important than a correct answer everywhere. If the function guesses
|
||
24-bit for a terminal without 24-bit colour, the result is not a duller palette: it is no
|
||
colour at all.
|
||
|
||
Returns:
|
||
The depth to force, or ``None`` to keep the default of prompt_toolkit.
|
||
"""
|
||
from ...core.preferences import current as current_preferences
|
||
|
||
override = os.environ.get("MESHTERM_COLOR_DEPTH")
|
||
preferred = override if override is not None else current_preferences().color_depth
|
||
if preferred in _DEPTH_BY_NAME:
|
||
return _DEPTH_BY_NAME[preferred]
|
||
if preferred != "auto":
|
||
return None
|
||
# A 16-slot console is already correct, and it has no truecolor to claim. The theme
|
||
# addresses its palette by index (refer to theme._VT_SLOTS). If the function increases
|
||
# the depth, it sends RGB to a screen that has only sixteen colours for it.
|
||
if not get_platform().truecolor:
|
||
return None
|
||
if os.environ.get("COLORTERM", "").strip().lower() in _TRUECOLOR_COLORTERM:
|
||
return ColorDepth.DEPTH_24_BIT
|
||
# The terminfo spelling for a direct-colour entry (``xterm-direct``, ``*-direct16m``).
|
||
if os.environ.get("TERM", "").endswith(("-direct", "-direct16m")):
|
||
return ColorDepth.DEPTH_24_BIT
|
||
return None
|
||
|
||
|
||
#: The maximum number of dialog layers that the layout can float over the background at one
|
||
#: time. It is a fixed pool of floats, each one a centred box (refer to
|
||
#: :meth:`TuiSession._build_app`). The pool is much larger than the deepest real nesting: the
|
||
#: list of a tool, a detail dialog for an item, and a confirm over it are only three.
|
||
_MAX_DIALOG_LAYERS = 8
|
||
|
||
#: How long :meth:`TuiSession.leave` waits for the quiesce of the app to start (that is, for
|
||
#: a transmission in progress to finish). After this time, the method continues all the same.
|
||
#: On the Cardputer Zero, the full exit must fit in the three seconds that the launcher gives
|
||
#: between SIGTERM and SIGKILL (:data:`~meshterm.services.hold_to_quit.GRACE_S`). The rest of
|
||
#: the exit took approximately one second there (measured on 2026-10-06).
|
||
QUIESCE_WAIT_S = 1.0
|
||
|
||
_log = get_logger()
|
||
|
||
|
||
def _changed_rows(before: str, after: str) -> list[int] | None:
|
||
"""Which lines of a full-screen frame are different, or ``None`` if a comparison fails.
|
||
|
||
The two frames are composed with one line for each terminal row (refer to
|
||
:func:`~meshterm.ui.tui.frame.compose_base`). Thus a line index is also a row index, and
|
||
:meth:`TuiSession._scrub_rows` uses this mapping. If the line counts are different, the
|
||
height of the frame changed (a resize, or a splash that gives its place to the framed
|
||
layout), and this mapping is not true. Then the caller does a full paint.
|
||
"""
|
||
old, new = before.split("\n"), after.split("\n")
|
||
if len(old) != len(new):
|
||
return None
|
||
return [i for i, (a, b) in enumerate(zip(old, new, strict=True)) if a != b]
|
||
|
||
|
||
def _has_wide_glyph(text: str) -> bool:
|
||
"""Whether ``text`` has a glyph for which prompt_toolkit reserves two cells.
|
||
|
||
The terminal and the renderer can disagree about a width-2 glyph. The usual case here is
|
||
an emoji (``👋``) that the terminal draws in one cell. pt reserves two cells, the
|
||
terminal moves forward one cell, and after that point the cursor model of the row is
|
||
incorrect. The node-type marks (``▲●■``) and the braille of a chart have a width of 1
|
||
everywhere, so they never cause a ``True`` result. The ``>= 0x1100`` guard skips the
|
||
ASCII and Latin characters, which are most of a frame, before the width lookup. This is
|
||
important, because this function examines the full composed frame at each paint.
|
||
|
||
On a platform that folds its frames to a fixed font, the answer is always ``False``.
|
||
That font is a set of 512 glyphs with no wide glyph in it. Thus nothing that a composed
|
||
frame can contain can return ``True``. The function answers from the platform instead of
|
||
from the text. Thus it skips a scan of each character of the full frame at each paint.
|
||
It also skips the scrub machinery that a ``True`` result arms, so the cheap differential
|
||
paint always stays in use.
|
||
|
||
The fold gives this guarantee, not the icon table. A terminal that cannot draw emoji (the
|
||
classic Windows console, refer to :func:`meshterm.ui.termfont.emoji_support`) still
|
||
renders the rest of the BMP. A contact with a CJK name is two cells wide there, the same
|
||
as everywhere else. If this function reads ``emoji`` instead, it skips the scan on a frame
|
||
that must have it, and leaves stray characters in the row.
|
||
"""
|
||
if get_platform().font:
|
||
return False
|
||
return any(ord(ch) >= 0x1100 and get_cwidth(ch) == 2 for ch in text)
|
||
|
||
|
||
class _WidthExtendedOutput:
|
||
"""A prompt_toolkit ``Output`` proxy that reports one more terminal column.
|
||
|
||
The proxy wraps the real output, and forwards all calls to it with no change, except
|
||
:meth:`get_size`, which adds a column. The full render pipeline uses
|
||
``output.get_size()``: the differential renderer of prompt_toolkit, and the frame
|
||
compositor of MeshTerm (through :meth:`TuiSession._size`). Thus this one override makes
|
||
the two use the reclaimed column together. Refer to :func:`_reclaim_last_column` for when
|
||
this is correct (and when it is not).
|
||
"""
|
||
|
||
def __init__(self, inner: Any) -> None:
|
||
"""Wrap ``inner`` (the concrete prompt_toolkit output for the real terminal)."""
|
||
self._inner = inner
|
||
|
||
def get_size(self) -> Size:
|
||
"""The wrapped size with one more column, so that the last column is used."""
|
||
size = self._inner.get_size()
|
||
return Size(rows=size.rows, columns=size.columns + 1)
|
||
|
||
def __getattr__(self, name: str) -> Any:
|
||
"""Forward each other attribute or method directly to the wrapped output."""
|
||
return getattr(self._inner, name)
|
||
|
||
|
||
def _message_border(message: Text | str) -> str:
|
||
"""Choose the border style of a message dialog from the strongest tone in the text.
|
||
|
||
A note about a result has its severity as theme spans (``[err]``, ``[warn]``, ``[ok]``).
|
||
Thus the border of the dialog can show the same severity. An error message gets the
|
||
``err`` border. A warning (for example "device rebooting") gets the caution border. All
|
||
the other messages, also the success messages, get the grey of the standard border. A
|
||
plain string has no spans, and it is always neutral.
|
||
|
||
Args:
|
||
message: The message of the dialog, with styles or plain.
|
||
|
||
Returns:
|
||
The Rich style name for the border of the dialog.
|
||
"""
|
||
if isinstance(message, Text):
|
||
styles = {str(span.style) for span in message.spans}
|
||
if "err" in styles:
|
||
return "err"
|
||
if "warn" in styles:
|
||
return "warn"
|
||
return "border"
|
||
|
||
|
||
class Visit:
|
||
"""The stay of one screen on the stack, which :meth:`TuiSession.stay` returns.
|
||
|
||
It holds only the session and the screen. :meth:`result` arms a new future on that screen
|
||
and awaits it, and the screen stays exactly where it is. A second call is the next round
|
||
of the same visit. This is the loop shape that a hub screen must have.
|
||
"""
|
||
|
||
__slots__ = ("_session", "_screen")
|
||
|
||
def __init__(self, session: TuiSession, screen: Screen) -> None:
|
||
"""Bind a visit to its session and screen, and arm the first round of the screen."""
|
||
self._session = session
|
||
self._screen = screen
|
||
self._arm()
|
||
|
||
@property
|
||
def screen(self) -> Screen:
|
||
"""The screen of this visit (the same object for the whole stay)."""
|
||
return self._screen
|
||
|
||
def _arm(self) -> None:
|
||
"""Give the screen a new future to resolve into.
|
||
|
||
A visited screen is always armed, from the moment when it is pushed until the end of
|
||
the visit. The reason is that it is on the stack all the time, and a key press can
|
||
get to it each time that it is on top. If only :meth:`result` arms it, there is a
|
||
gap between the rounds. In that gap,
|
||
:meth:`~meshterm.ui.tui.screen.Screen.resolve` has no place to put its value, and the
|
||
key press is lost with no message.
|
||
"""
|
||
self._screen.future = asyncio.get_running_loop().create_future()
|
||
|
||
async def result(self) -> Any:
|
||
"""Await one round of the visited screen: its resolved value, or ``CANCEL`` on Esc.
|
||
|
||
A round can be already resolved, because the screen was armed while the caller was
|
||
still busy with the last round. Such a round is returned immediately, and the method
|
||
does not wait for it again.
|
||
|
||
Raises:
|
||
PopToMenu: If the user pressed the pop-all key, here or while the caller was
|
||
busy.
|
||
"""
|
||
self._session._check_unwind()
|
||
future = self._screen.future
|
||
if future is None:
|
||
self._arm()
|
||
future = self._screen.future
|
||
try:
|
||
return self._session._unpack(await future)
|
||
finally:
|
||
self._arm() # the next round is live from the moment when this round is read
|
||
|
||
|
||
class TuiSession:
|
||
"""A running full-screen TUI: the screen stack, the frame, the input loop, and async prompts."""
|
||
|
||
def __init__(
|
||
self,
|
||
header: Callable[[int], RenderableType] | None = None,
|
||
*,
|
||
input: Any = None, # noqa: A002 - the same name as Application(input=) in prompt_toolkit
|
||
output: Any = None,
|
||
) -> None:
|
||
"""Create a session.
|
||
|
||
Args:
|
||
header: A callable that takes the width of the terminal in cells and returns
|
||
the persistent header renderable (the banner and the live status). The
|
||
session calls it again at each paint. With the width, the header can
|
||
stretch the content at its end (the activity sparkline) to fill the row
|
||
exactly. The default is a plain title.
|
||
input: An optional prompt_toolkit input that drives the app (the tests use a
|
||
pipe). The default is the real terminal.
|
||
output: An optional prompt_toolkit output to render to (the tests use a dummy).
|
||
The default is the real terminal.
|
||
"""
|
||
self._stack: list[Screen] = []
|
||
self._header = header or (lambda cols: Text("MeshTerm", style="brand"))
|
||
self._app: Application | None = None
|
||
self._input = input
|
||
self._output = output
|
||
# The top floating "working" overlay (a skeleton card), or None when the app is idle.
|
||
# It is not on the screen stack, on purpose: it floats above each layer, and
|
||
# busy_overlay shows and hides it, independently of the screens that are pushed.
|
||
self._overlay: BusyOverlay | None = None
|
||
# The busy card that :meth:`busy_dialog` pushed, or None. Different from the
|
||
# overlay, this card is a screen. It must be a screen, so that it takes the key
|
||
# presses. If it does not take them, the hub below it keeps them for later. This
|
||
# attribute holds the card, so that a nested wait uses the same card instead of a
|
||
# second card on the stack.
|
||
self._busy: BusyDialog | None = None
|
||
# The last composition of each drawn layer: ``layer -> (text, carries a wide
|
||
# glyph)``. With it, :meth:`_emit` can tell a frame that changed from a frame that
|
||
# the 1 Hz paint timer rendered again with no change. It can also find if anything
|
||
# on the screen now must have a full paint. At each paint, this dict is reconciled
|
||
# against the layers that were drawn (refer to :meth:`_reconcile_layers`).
|
||
self._layers: dict[str, tuple[str, bool]] = {}
|
||
# The pop-all key (^W) arms this flag, and the menu loop disarms it after it catches
|
||
# the unwind. While the flag is armed, each navigation boundary raises PopToMenu
|
||
# instead of showing another screen. Refer to :meth:`request_pop_all`.
|
||
self._unwinding = False
|
||
# The screen at the end of the unwind (the main menu). With it, ^W can do nothing
|
||
# when that screen is already the top, instead of a rebuild with no purpose. A
|
||
# dialog also floats over it when no other screen is pushed (refer to
|
||
# :meth:`_floated`). ``None`` until the menu declares it.
|
||
self._root: Screen | None = None
|
||
# The quit confirm that ^Q opens from any screen (refer to :meth:`request_quit`).
|
||
# The app declares it with :meth:`set_quit_confirm`. Then a flag that tells if the
|
||
# confirm is open now. When it is open, the next ^Q quits the app.
|
||
self._quit_confirm: Callable[[], Awaitable[bool]] | None = None
|
||
self._quit_asking = False
|
||
# What the app does to stop the start of new work before it possibly must quit
|
||
# (refer to :meth:`set_quiesce`). Then the quiesce while it is held. Then the
|
||
# one-way flag that :meth:`leave` raises, which tells if the app is on its way out.
|
||
self._quiesce: Callable[[], AbstractAsyncContextManager[Any]] | None = None
|
||
self._quiet: Quiet | None = None
|
||
self._leaving = False
|
||
# The Esc of a terminal, held back while the keyboard says that the key is still
|
||
# down (where something can tell this, refer to :meth:`set_esc_probe`).
|
||
self._watched_esc: WatchedEsc | None = None
|
||
|
||
# --- stack ---------------------------------------------------------------
|
||
|
||
@property
|
||
def top(self) -> Screen | None:
|
||
"""The active screen (the top 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 paint the frame again."""
|
||
self._stack.append(screen)
|
||
self.invalidate()
|
||
|
||
def pop(self, screen: Screen | None = None) -> None:
|
||
"""Pop ``screen`` (or the top screen) off the stack, and paint the frame again."""
|
||
if not self._stack:
|
||
return
|
||
popped: Screen | None = None
|
||
if screen is None or self._stack[-1] is screen:
|
||
popped = self._stack.pop()
|
||
elif screen in self._stack:
|
||
self._stack.remove(screen)
|
||
popped = screen
|
||
self._expose_overlay()
|
||
# A floating dialog is drawn as a box, sized to its content, over the screen below
|
||
# it. A cell of the dialog can hold a glyph that the terminal drew wider than
|
||
# prompt_toolkit tracks (an emoji or a box-drawing fallback that the differential
|
||
# renderer cannot see). Then the extra part stays as stray characters after the
|
||
# dialog closes. Thus, when the dialog closes, force the frame below it to write
|
||
# each cell again, so that nothing of the dialog stays.
|
||
if popped is not None and getattr(popped, "floating", False):
|
||
self._invalidate_last_frame()
|
||
self.invalidate()
|
||
|
||
def reset(self) -> None:
|
||
"""Clear the full screen stack, and paint the frame again.
|
||
|
||
MeshTerm uses this method to unwind to a blank frame after a cancelled activity. For
|
||
example, when the device disconnects during a session, MeshTerm abandons all the
|
||
screens that the interrupted work pushed. Then it shows the reconnect dialog over a
|
||
clean frame. Screens hold no external resources (their callers pop them in
|
||
``finally``). Thus it is safe to remove the remaining screens here.
|
||
"""
|
||
self._stack.clear()
|
||
# A reset is an unwind itself. The caller has already abandoned all that was running
|
||
# (refer to :func:`~meshterm.ui.menu._session_loop`). Thus an armed ^W has nothing
|
||
# more to unwind. If it stays armed, it goes off in the reconnect flow that
|
||
# replaces the abandoned work.
|
||
self._unwinding = False
|
||
self._expose_overlay()
|
||
self.invalidate()
|
||
|
||
def _expose_overlay(self) -> None:
|
||
"""Restart the fade of the busy overlay when a stack change shows it again.
|
||
|
||
A prompt that is pushed over an active overlay hides the card. A pop back to an empty
|
||
stack shows the card again. Then restart the intro, so that the black hold and the
|
||
fade-in start again from the beginning each time that the card is shown. Without the
|
||
restart, the card comes back immediately at full brightness (refer to
|
||
:meth:`~meshterm.ui.tui.overlay.BusyOverlay.restart`).
|
||
"""
|
||
if self._overlay is not None and not self._stack:
|
||
self._overlay.restart()
|
||
|
||
# --- navigation ----------------------------------------------------------
|
||
|
||
def set_root(self, screen: Screen | None) -> None:
|
||
"""Declare ``screen`` as the navigation root: the screen at the end of an unwind.
|
||
|
||
The main menu calls this method with its own list. Only two things depend on it:
|
||
|
||
- ^W does nothing while the root is already the top screen (you cannot go back to
|
||
where you are).
|
||
- A dialog with nothing below it floats over the root, not over an empty frame
|
||
(:meth:`_floated`).
|
||
|
||
No other part of the app must know which screen is the menu.
|
||
|
||
Args:
|
||
screen: The root screen, or ``None`` to forget it.
|
||
"""
|
||
self._root = screen
|
||
|
||
@property
|
||
def root(self) -> Screen | None:
|
||
"""The declared navigation root (the main menu), or ``None`` before it is declared."""
|
||
return self._root
|
||
|
||
def set_quit_confirm(self, ask: Callable[[], Awaitable[bool]] | None) -> None:
|
||
"""Declare the question that the quit chord asks before the app quits.
|
||
|
||
The app declares it one time, because only the app knows what the question offers.
|
||
Over a Bluetooth bond, the confirm of the main menu adds *Unpair & quit*, and that
|
||
choice must have the application context, which the session never holds. ``ask``
|
||
floats its dialog and returns whether to quit. It owns all side effects of the answer
|
||
(for example, it stores the record of an unpair).
|
||
|
||
Args:
|
||
ask: The confirm, or ``None`` if the quit chord must quit without a question.
|
||
"""
|
||
self._quit_confirm = ask
|
||
|
||
async def confirm_quit(self) -> bool:
|
||
"""Ask the declared quit confirm, as the Quit row of the main menu does.
|
||
|
||
It is the same question that ^Q asks, and the same dialog. While the dialog is open,
|
||
the next press of the chord quits immediately (refer to :meth:`request_quit`),
|
||
whichever way opened the dialog.
|
||
|
||
Returns:
|
||
Whether the user chose to quit. ``True`` if no confirm is declared.
|
||
"""
|
||
self._quit_asking = True
|
||
return await self._asking_quit()
|
||
|
||
async def _asking_quit(self) -> bool:
|
||
"""Run the declared confirm with :attr:`_quit_asking` already raised, then lower it."""
|
||
try:
|
||
return self._quit_confirm is None or bool(await self._quit_confirm())
|
||
finally:
|
||
self._quit_asking = False
|
||
|
||
def request_quit(self) -> None:
|
||
"""Answer the quit chord (^Q, and also ^C): ask first, and quit on the second press.
|
||
|
||
^Q is one key from ^W, the chord that goes back to the menu. Thus a wrong key press
|
||
on the way to the menu once closed the app, and with it a running capture or a
|
||
courier queue. Now the chord floats the quit confirm over the current screen, from
|
||
any screen (issue #22, option D). The confirm is a detached flow
|
||
(:meth:`run_detached`). Nothing below it is interrupted, a capture continues behind
|
||
the box, and Esc puts the user back exactly where they were.
|
||
|
||
While the confirm is open, the chord quits immediately. Thus the fastest way out is
|
||
still two key presses, and it still works when a screen is stuck or a device read
|
||
does not return. The second press never waits for the dialog to paint or to take a
|
||
key. The flag is raised here, synchronously. Thus two presses that arrive in one
|
||
input batch are still a quit, not two dialogs.
|
||
|
||
In the two cases, the app quits cleanly. :meth:`run` cancels the coroutine that
|
||
drives the app and waits for its unwind. Thus the history runs and the chat runs
|
||
close, the services stop, and the exit watchdog is armed, the same as a quit from
|
||
the menu.
|
||
"""
|
||
if self._quit_asking or self._quit_confirm is None:
|
||
self._exit_app()
|
||
return
|
||
self._quit_asking = True
|
||
|
||
async def ask() -> None:
|
||
if await self._asking_quit():
|
||
self._exit_app()
|
||
|
||
self.run_detached(ask())
|
||
|
||
def _exit_app(self) -> None:
|
||
"""Exit the running application, whatever flow drives it."""
|
||
if self._app is not None and self._app.is_running:
|
||
self._app.exit()
|
||
|
||
def set_quiesce(self, enter: Callable[[], AbstractAsyncContextManager[Any]] | None) -> None:
|
||
"""Declare how the app stops the start of new work, for when it possibly must quit soon.
|
||
|
||
The session enters the quiesce when a held Esc opens its box, and ends it again if
|
||
the user releases the key (refer to :mod:`~meshterm.services.hold_to_quit`).
|
||
:meth:`leave` also enters it on the way out, and then keeps it until the process
|
||
ends. The quiesce must not take anything apart, because a release of the key must
|
||
undo it. The quiesce of the app holds the transmit lock of the device. Thus nothing
|
||
new goes on the air, and a scoped channel send in progress closes its window first.
|
||
The app declares the quiesce, as it declares the quit confirm, because only the app
|
||
has a device.
|
||
|
||
Args:
|
||
enter: Makes the async context manager to hold, or ``None`` for nothing.
|
||
"""
|
||
self._quiesce = enter
|
||
|
||
def set_esc_probe(self, down: Callable[[], bool] | None) -> None:
|
||
"""Declare how to ask whether the Esc key is physically down, for the Esc of a terminal.
|
||
|
||
With a probe, an Esc that arrives while its key is still down is held back and
|
||
watched as a hold (:class:`~meshterm.ui.tui.holdquit.WatchedEsc`). This gives the
|
||
hold-to-quit gesture where a terminal sends the keys, and a terminal never reports
|
||
that a key comes up. The app declares the probe, because only the app knows that the
|
||
keys are the keys of this machine. They are not the keys of a front end that reads
|
||
them itself (the emulator), and not the keys of another machine over SSH.
|
||
|
||
Args:
|
||
down: Tells whether the Esc key is down now
|
||
(:func:`~meshterm.services.hold_to_quit.esc_probe`), or ``None``.
|
||
"""
|
||
self._watched_esc = None if down is None else WatchedEsc(self, down)
|
||
|
||
@property
|
||
def leaving(self) -> bool:
|
||
"""Whether the app is on its way out (:meth:`leave`). Nothing reverses this state."""
|
||
return self._leaving
|
||
|
||
def quiet(self) -> Quiet:
|
||
"""Engage the quiesce of the app (:meth:`set_quiesce`), or return the held quiesce."""
|
||
if self._quiet is None:
|
||
self._quiet = Quiet(self._quiesce)
|
||
return self._quiet
|
||
|
||
def unquiet(self) -> None:
|
||
"""Release the quiesce, except after :meth:`leave`: then the app keeps it to the end."""
|
||
if self._quiet is not None and not self._leaving:
|
||
self._quiet.release()
|
||
self._quiet = None
|
||
|
||
def leave(self) -> None:
|
||
"""Quit the app now, without a question, when no transmission is in progress.
|
||
|
||
The end of a held Esc and the SIGTERM of the launcher call this method. The decision
|
||
is already made, so there is no confirm. It is the same exit as Quit (:meth:`run`
|
||
unwinds the flow, and the app tears down), with one more step first. The quiesce of
|
||
the app is engaged, so that a scoped channel send in progress closes its window
|
||
before the device goes. :data:`QUIESCE_WAIT_S` limits that wait, because on the
|
||
Cardputer Zero, the SIGKILL of the launcher comes three seconds after its SIGTERM,
|
||
whatever the app does.
|
||
|
||
The method is idempotent and one-way. A SIGTERM that arrives when a hold has just run
|
||
out is the same exit, not a second exit.
|
||
"""
|
||
if self._leaving:
|
||
return
|
||
self._leaving = True
|
||
hold_to_quit.leaving()
|
||
quiet = self.quiet()
|
||
|
||
async def go() -> None:
|
||
if not await quiet.engaged(QUIESCE_WAIT_S):
|
||
_log.warning("leaving with a transmission still under way")
|
||
self._exit_app()
|
||
|
||
self.run_detached(go())
|
||
|
||
def request_pop_all(self) -> bool:
|
||
"""Arm the unwind to the navigation root (the ^W key). Return whether it is armed.
|
||
|
||
The method refuses in two cases, both on purpose:
|
||
|
||
* **A modal layer is on top.** A dialog is a question with no answer yet, and a
|
||
progress screen or a busy screen is work in progress
|
||
(:attr:`~meshterm.ui.tui.screen.Screen.modal`). An unwind past either one
|
||
discards something that the user is in the middle of. Thus ^W does nothing there,
|
||
and the user must first answer or wait.
|
||
* **The root is already the top.** There is no place to go.
|
||
|
||
In all other cases, the method sets the unwind flag, and resolves the future of
|
||
**each** screen on the stack with :data:`~meshterm.ui.tui.screen.POP_ALL`. It starts
|
||
at the top and goes down to the root, but it never includes the root. Each caller
|
||
that awaits one of these futures changes the value into
|
||
:class:`~meshterm.ui.tui.screen.PopToMenu`.
|
||
|
||
The method arms the full stack, not only the top. This makes ^W a verb of the stack,
|
||
not a verb of the call chain. The code that opens a screen from a key handler cannot
|
||
await that screen, because a handler is sync. Thus the live feed floats the packet
|
||
viewer with ``ensure_future(run_screen(...))``, and the chain that could carry an
|
||
unwind ends in a detached task. When the method sent the sentinel only to the top,
|
||
the unwind stopped there, while the hub below waited on a ``visit.result()`` that
|
||
nothing could resolve. ^W closed the viewer and stopped. Then the flag, which was
|
||
still armed, went off at the next key that the user pressed (JP, 2026-08-31). Each
|
||
screen is on the stack, whether something awaits it or not. Thus the stack is the
|
||
reliable way down.
|
||
|
||
The walk stops at a **modal** screen, for the same reason as the refusal above for
|
||
the top of the stack: the method must not discard work in progress from below that
|
||
work. The flag stays armed, so the unwind still goes off at the next navigation
|
||
boundary after that screen. If the method resolves a screen that nothing awaits,
|
||
nothing occurs: nothing reads the value, and
|
||
:meth:`~meshterm.ui.tui.screen.Screen.resolve` does nothing on a screen with no
|
||
future.
|
||
|
||
The flag is important independently of the futures. A key press can arrive while no
|
||
screen awaits anything (during a device read, below the busy overlay). Then the next
|
||
navigation boundary raises the unwind instead (refer to :meth:`_check_unwind`).
|
||
|
||
Returns:
|
||
``True`` if the unwind was armed, ``False`` if it was refused.
|
||
"""
|
||
top = self.top
|
||
if top is not None and (top.modal or top is self._root):
|
||
return False
|
||
self._unwinding = True
|
||
for screen in reversed(self._stack):
|
||
if screen is self._root:
|
||
break # the unwind ends at the root, and the root itself is not unwound
|
||
if screen is not top and screen.modal:
|
||
break
|
||
screen.resolve(POP_ALL)
|
||
return True
|
||
|
||
def unwound(self) -> None:
|
||
"""Disarm the unwind. The menu loop calls this after it catches :class:`PopToMenu`."""
|
||
self._unwinding = False
|
||
|
||
def _check_unwind(self) -> None:
|
||
"""Raise :class:`PopToMenu` if an unwind is armed, before another screen is shown.
|
||
|
||
Each navigation boundary calls this method on the way in, and also reads the result
|
||
on the way out. This makes ^W work during a device read. The key arms the flag while
|
||
no future is armed to carry :data:`POP_ALL`. Then the next screen that the flow tries
|
||
to open does not open, and it unwinds instead.
|
||
"""
|
||
if self._unwinding:
|
||
raise PopToMenu()
|
||
|
||
@staticmethod
|
||
def _unpack(result: Any) -> Any:
|
||
"""Return the result of a screen, and change :data:`POP_ALL` into :class:`PopToMenu`.
|
||
|
||
The sentinel exists only to go through an ``asyncio.Future``. No caller in the app
|
||
ever sees it. Thus it is changed at the only boundary that reads a future.
|
||
"""
|
||
if result is POP_ALL:
|
||
raise PopToMenu()
|
||
return result
|
||
|
||
@asynccontextmanager
|
||
async def stay(self, screen: Screen, *, dialog: bool = False) -> AsyncIterator[Visit]:
|
||
"""Keep ``screen`` pushed for a full visit, while the screens above it open and close.
|
||
|
||
This method is the counterpart of :meth:`run_screen`. Each screen that *owns a loop*
|
||
must use this shape::
|
||
|
||
screen = ContactsScreen(...)
|
||
async with session.stay(screen) as visit:
|
||
while True:
|
||
chosen = await visit.result()
|
||
if chosen is CANCEL:
|
||
return
|
||
await open_node_detail(ctx, chosen) # nests above the list
|
||
|
||
One push and one pop, and the object stays for the full visit. Thus the highlight,
|
||
the sort, the scroll offset, and any live filter are still there when a screen above
|
||
it closes. No ``default=`` restore is necessary, and such a restore can only recover
|
||
the highlight. The screen never leaves the stack. Thus a dialog that the loop opens
|
||
already has the screen as its backdrop, and a full-frame screen that is pushed above
|
||
it draws over it (refer to :meth:`_base_index`). Before, callers got these two
|
||
results by hand: they popped the same screen and pushed it again.
|
||
|
||
Args:
|
||
screen: The screen to keep pushed while the block runs.
|
||
dialog: The visit belongs to a dialog: a stepped dialog that turns its pages
|
||
(:func:`~meshterm.ui.menus.run_wizard`), not a hub. Thus it must draw as a
|
||
box, also when it is the only screen on the stack. A blank base goes below it
|
||
for the visit, exactly as :meth:`run_dialog` does for a one-time dialog.
|
||
|
||
Yields:
|
||
A :class:`Visit` whose :meth:`~Visit.result` awaits one round of the screen.
|
||
"""
|
||
self._check_unwind()
|
||
async with self._floated() if dialog else nullcontext():
|
||
self.push(screen)
|
||
try:
|
||
yield Visit(self, screen)
|
||
finally:
|
||
self.pop(screen)
|
||
|
||
def invalidate(self) -> None:
|
||
"""Request a paint if the application is running."""
|
||
if self._app is not None:
|
||
self._app.invalidate()
|
||
|
||
def request_full_repaint(self) -> None:
|
||
"""Force the next paint to write each cell again, then schedule that paint.
|
||
|
||
prompt_toolkit paints differentially: it does not touch the cells that are equal to
|
||
the cells of the last frame. Usually we want this. But then damage on the terminal
|
||
side (a double-width fallback glyph that the diff cannot see) stays until those exact
|
||
cells change. Callers use this method when the user leaves a screen that can leave
|
||
stray characters on the terminal (the map). Thus the screen that is drawn below it
|
||
starts from a clean frame.
|
||
"""
|
||
self._invalidate_last_frame()
|
||
self.invalidate()
|
||
|
||
def _invalidate_last_frame(self) -> None:
|
||
"""Remove the cached last frame of prompt_toolkit, so that the next paint writes all cells.
|
||
|
||
This method uses an internal of prompt_toolkit. Thus it fails softly if the attribute
|
||
moves.
|
||
"""
|
||
renderer = getattr(self._app, "renderer", None)
|
||
if renderer is not None and hasattr(renderer, "_last_screen"):
|
||
renderer._last_screen = None
|
||
|
||
def _scrub_columns(self, start: int, stop: int) -> None:
|
||
"""Force prompt_toolkit to paint the terminal columns ``[start, stop)`` at the next diff.
|
||
|
||
This method writes a sentinel over those cells in the last frame that pt remembers.
|
||
The sentinel can never be equal to real content. Thus the differential renderer
|
||
thinks that the cells changed, and draws them again. This removes a double-width
|
||
fallback glyph that left stray characters on a static edge (the panel border of the
|
||
map, or the border of a floating dialog). It does this without the flicker of the
|
||
full frame that occurs when the full cached frame is removed. This method uses an
|
||
internal of pt, so it fails softly if the structure moves.
|
||
"""
|
||
renderer = getattr(self._app, "renderer", None)
|
||
last = getattr(renderer, "_last_screen", None)
|
||
if last is None:
|
||
return
|
||
try:
|
||
from prompt_toolkit.layout.screen import Char
|
||
|
||
buffer = last.data_buffer
|
||
sentinel = Char("") # a non-character, never equal to real cell content
|
||
for x in range(max(0, start), stop):
|
||
for row in list(buffer.keys()):
|
||
buffer[row][x] = sentinel
|
||
except Exception: # noqa: BLE001 - a cosmetic scrub must never stop the render
|
||
pass
|
||
|
||
def _scrub_right_columns(self, count: int) -> None:
|
||
"""Scrub the last ``count`` columns of the terminal: the edge of a full-frame panel."""
|
||
cols, _ = self._size()
|
||
self._scrub_columns(cols - count, cols)
|
||
|
||
def _scrub_rows(self, rows: Sequence[int]) -> bool:
|
||
r"""Force prompt_toolkit to write these full terminal rows again at the next diff.
|
||
|
||
This method is the row version of :meth:`_scrub_columns`, and the cheap form of the
|
||
wide-glyph paint (:meth:`_emit`). It puts the sentinel in each column of each listed
|
||
row. Thus pt finds that the full row changed, and writes it from column 0 in one
|
||
continuous sequence. That is all that is necessary for a row with a glyph that the
|
||
terminal draws narrower than the cells that pt reserved. The method does not touch
|
||
the rows that did not change, and it erases nothing.
|
||
|
||
A scrub by row is correct, because the cursor of pt is relative. A step down one row
|
||
writes ``\\r\\n``, which puts the terminal back to a real column 0, whatever the
|
||
drift on the row above. Thus a row with an incorrect measurement can never move the
|
||
rows below it. The drift is important only in a row, and a row that is written again
|
||
in full never jumps inside itself.
|
||
|
||
Args:
|
||
rows: The indices of the terminal rows to mark as changed.
|
||
|
||
Returns:
|
||
Whether the scrub was done. ``False`` when pt has no remembered frame to scrub
|
||
(it will paint all the frame again), or when the internals moved.
|
||
"""
|
||
renderer = getattr(self._app, "renderer", None)
|
||
last = getattr(renderer, "_last_screen", None)
|
||
if last is None:
|
||
return False
|
||
cols, _ = self._size()
|
||
try:
|
||
from prompt_toolkit.layout.screen import Char
|
||
|
||
buffer = last.data_buffer
|
||
sentinel = Char("") # a non-character, never equal to real cell content
|
||
for y in rows:
|
||
row = buffer[y]
|
||
for x in range(cols):
|
||
row[x] = sentinel
|
||
except Exception: # noqa: BLE001 - a cosmetic scrub must never stop the render
|
||
return False
|
||
return True
|
||
|
||
# --- async prompt helpers ------------------------------------------------
|
||
|
||
def run_detached(self, work: Any) -> asyncio.Future:
|
||
"""Run ``work`` (a flow that opens a screen) from a key handler, without an await.
|
||
|
||
The ``handle`` method of a screen is synchronous. Thus a key that opens something
|
||
over the current screen can only start the flow, as a task. Examples are the packet
|
||
viewer of the live feed, the delivery paths of the chat, and the path flows of the
|
||
trace screen. That task is outside the navigation call chain: nothing awaits it.
|
||
Thus a :class:`~meshterm.ui.tui.screen.PopToMenu` that is raised in the task has no
|
||
place to propagate to. It goes to the exception handler of the event loop as an
|
||
unretrieved error.
|
||
|
||
It is safe to absorb the exception here, because the unwind never went this way.
|
||
^W arms each screen on the stack (refer to :meth:`request_pop_all`). Thus the screen
|
||
that opened this screen takes the unwind out, because its own caller awaits it. This
|
||
catch removes only a copy of an unwind that is already in progress. Also, the
|
||
``finally`` blocks of the flow still run when the exception goes through them.
|
||
|
||
Args:
|
||
work: The coroutine to run, usually a :meth:`run_screen` call.
|
||
|
||
Returns:
|
||
The task, for a caller that must cancel it or await it. The callers in the app
|
||
are key handlers, and they ignore it. The screen that they opened is on the
|
||
stack, and all other code finds the screen there.
|
||
"""
|
||
|
||
async def guarded() -> None:
|
||
try:
|
||
await work
|
||
except PopToMenu:
|
||
pass # the stack carries the unwind, and this task was never on its way
|
||
|
||
return asyncio.ensure_future(guarded())
|
||
|
||
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 resolved value of the screen, or :data:`~meshterm.ui.tui.screen.CANCEL`.
|
||
|
||
Raises:
|
||
PopToMenu: If the user pressed the pop-all key: on this screen, or while the
|
||
caller was busy and had nothing pushed.
|
||
"""
|
||
self._check_unwind()
|
||
loop = asyncio.get_running_loop()
|
||
screen.future = loop.create_future()
|
||
self.push(screen)
|
||
try:
|
||
result = await screen.future
|
||
finally:
|
||
self.pop(screen)
|
||
return self._unpack(result)
|
||
|
||
async def select(
|
||
self,
|
||
title: str,
|
||
items: list,
|
||
*,
|
||
prompt: str = "",
|
||
default: Any = None,
|
||
filterable: bool = True,
|
||
footer_hint: str | None = None,
|
||
delete_hint: str = "",
|
||
floating: bool = False,
|
||
) -> Any:
|
||
"""Show a select screen. Return the chosen value, or ``None`` if the user cancelled.
|
||
|
||
``prompt`` draws an instruction in the box, above the list. ``filterable`` and
|
||
``footer_hint`` go to the screen for short, fixed lists (a choice of the yes-or-no
|
||
type) that must have no type-to-filter and a special hint. ``delete_hint`` (with rows
|
||
marked :attr:`~meshterm.ui.tui.select.Choice.deletable`) shows the atom of the Delete
|
||
key while the highlight is on such a row. Then Delete resolves a
|
||
:class:`~meshterm.ui.tui.select.DeleteRequest`, which the caller unwraps.
|
||
``floating`` is the same as in :meth:`text`: the list is a question on the way into a
|
||
tool (which node to administer), not the page of the tool. Thus it must draw as a
|
||
box, also when it is the only screen on the stack. Refer to :meth:`_floated`.
|
||
"""
|
||
kwargs: dict[str, Any] = dict(
|
||
prompt=prompt,
|
||
default=default,
|
||
filterable=filterable,
|
||
delete_hint=delete_hint,
|
||
)
|
||
if footer_hint is not None:
|
||
kwargs["footer_hint"] = footer_hint
|
||
runner = self.run_dialog if floating else self.run_screen
|
||
result = await runner(SelectScreen(title, items, **kwargs))
|
||
return None if result is CANCEL else result
|
||
|
||
async def select_startup(
|
||
self,
|
||
title: str,
|
||
items: list,
|
||
*,
|
||
default: Any = None,
|
||
banner: Any | None = None,
|
||
footnote: str | None = None,
|
||
footer_hint: str = "↑↓ move · Enter select · Esc bye",
|
||
keys: Mapping[str, Any] | None = None,
|
||
key_hint: Callable[[Any], str] | None = None,
|
||
live: Callable[[Callable[[list], None]], Awaitable[None]] | None = None,
|
||
hscroll: bool | None = None,
|
||
) -> Any:
|
||
"""Show a chromeless select splash (a banner above a box sized to its content).
|
||
|
||
The splash is like :meth:`select`, but it has no header or footer status bars, and
|
||
it is centred below ``banner``. This is the presentation of the startup device
|
||
picker. Here the verb of Esc is ``bye``, the only farewell of the app. This splash is
|
||
the door, and a user who leaves it has started nothing to quit. Type-to-filter is
|
||
off: the device list is short and fixed, so stray key presses never make it shorter.
|
||
An optional ``footnote`` (for example a copyright notice) is muted, below the box.
|
||
When a row accepts removal (a :attr:`~meshterm.ui.tui.select.Choice.deletable` row),
|
||
a "Del remove" atom is added to the footer. But the atom shows only while the
|
||
highlight is on such a row, so the removal key shows itself exactly where it acts
|
||
(refer to :attr:`~meshterm.ui.tui.select.SelectScreen.footer_hint`).
|
||
|
||
``live`` is work to do while the list is visible: the rescan of the device picker,
|
||
which looks for a companion all the time that the splash is open. It gets one
|
||
function, ``redraw(items)``, which replaces the rows while the user looks at them,
|
||
and paints the frame again. The screen itself stays in this method, because a caller
|
||
that held it could not paint it again (``invalidate`` belongs to the session, not to
|
||
the screen). The work runs as a task for exactly as long as the screen runs, and it
|
||
is cancelled when the screen resolves. Thus nothing continues after the list that it
|
||
drew, and no paint goes onto a dialog that opened over the list later.
|
||
|
||
``hscroll`` overrules what the rows imply. A splash whose rows pin no head block
|
||
still must let ←→ read a long row to its end, and a row that pins nothing cannot ask
|
||
for that itself (refer to :class:`~meshterm.ui.tui.select.SelectScreen`).
|
||
|
||
``keys`` gives bare-key shortcuts to the list. A splash can have them because it does
|
||
not filter, so each letter is free (refer to :class:`SelectScreen`). ``key_hint``
|
||
gives their names on the row where the highlight is. Thus they show themselves
|
||
exactly where they act, as ``Del remove`` does.
|
||
"""
|
||
delete_hint = (
|
||
"Del remove" if any(isinstance(it, Choice) and it.deletable for it in items) else ""
|
||
)
|
||
screen = SelectScreen(
|
||
title,
|
||
items,
|
||
default=default,
|
||
footer_hint=footer_hint,
|
||
delete_hint=delete_hint,
|
||
filterable=False,
|
||
keys=keys,
|
||
key_hint=key_hint,
|
||
hscroll=hscroll,
|
||
)
|
||
screen.chrome = False
|
||
screen.banner = banner
|
||
screen.footnote = footnote
|
||
# The border of the splash is its only hint line, and it is narrow (47 cells on the
|
||
# PicoCalc). Thus, when the sentence becomes wider than the box, it first removes
|
||
# the scroll atom (the move atom in front of it already names ←→). Then it removes
|
||
# the hide key. Of the two shortcuts, the one that stays is the way back. ⇧H shows
|
||
# only after something is hidden, and that is exactly the state where the user must
|
||
# learn how to undo it. At that time, the user has already found h.
|
||
screen.spare_hint_atoms = ("←→ scroll", "h hide")
|
||
|
||
def redraw(new_items: list) -> None:
|
||
"""Replace the rows and paint again. The user keeps their place, by value."""
|
||
screen.replace_items(new_items)
|
||
self.invalidate()
|
||
|
||
worker = asyncio.ensure_future(live(redraw)) if live else None
|
||
try:
|
||
result = await self.run_screen(screen)
|
||
finally:
|
||
if worker is not None:
|
||
worker.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await worker
|
||
return None if result is CANCEL else result
|
||
|
||
async def confirm_startup(
|
||
self,
|
||
prompt: str | Text,
|
||
*,
|
||
title: str = "",
|
||
confirm_label: str = "Remove",
|
||
banner: Any | None = None,
|
||
footnote: str | None = None,
|
||
backdrop_items: list | None = None,
|
||
backdrop_default: Any = None,
|
||
backdrop_title: str = "Select a companion device",
|
||
footer_hint: str = "←→ choose · Enter select · Esc cancel",
|
||
) -> bool:
|
||
"""Confirm a destructive splash action with a Cancel/verb dialog (chromeless).
|
||
|
||
This is the startup-splash version of :meth:`button_dialog`, with the same
|
||
platform-dialog layout. The safe *Cancel* is on the left, and the committing verb is
|
||
on the right and is the default. Thus Enter commits, and Esc backs out. The theme is
|
||
the theme of data loss (the reserved red prompt and border), because this dialog only
|
||
asks before MeshTerm forgets a remembered device. It is the splash version of a
|
||
``destructive`` :meth:`button_dialog`.
|
||
|
||
When ``backdrop_items`` is given, the device list that they describe is drawn again
|
||
as the chromeless base. The red confirm floats over it as a centred box (a modal
|
||
dialog over the pushed backdrop, as each other dialog does). Thus the removal of a
|
||
device looks like a dialog on top of the picker, not a splash that replaces it. The
|
||
``backdrop_default`` row is highlighted first, so that the user sees that the confirm
|
||
is about it. Without ``backdrop_items``, the confirm draws as its own chromeless
|
||
splash below ``banner`` (the fallback for a caller with no list to float over).
|
||
|
||
Args:
|
||
prompt: The question above the buttons.
|
||
title: The short heading in the border of the dialog.
|
||
confirm_label: The label of the committing button (for example ``"Remove"``).
|
||
banner: The wordmark rows above the box (as on the other startup splashes).
|
||
footnote: The muted line below the box.
|
||
backdrop_items: The rows of the picker to draw again behind the confirm.
|
||
``None`` gives a standalone chromeless confirm splash instead.
|
||
backdrop_default: The row value to highlight in the backdrop list.
|
||
backdrop_title: The heading of the backdrop list (the title of the picker).
|
||
footer_hint: The key hint of the footer.
|
||
|
||
Returns:
|
||
``True`` only when the user chose the committing button. ``False`` on Cancel or
|
||
Esc.
|
||
"""
|
||
dialog = ButtonDialog(
|
||
prompt,
|
||
[("Cancel", False), (confirm_label, True)],
|
||
title=title,
|
||
default=1,
|
||
footer_hint=footer_hint,
|
||
prompt_style="err",
|
||
border_style="err",
|
||
)
|
||
if backdrop_items is None:
|
||
# No list to float over: draw the confirm as its own chromeless splash.
|
||
dialog.chrome = False
|
||
dialog.banner = banner
|
||
dialog.footnote = footnote
|
||
return await self.run_screen(dialog) is True
|
||
# Keep the picker on the screen as the chromeless base, and float the red confirm over
|
||
# it. Thus the removal confirm is on top of the device list that it acts on. The
|
||
# backdrop is a static copy of the same rows. It never takes a key, because the dialog
|
||
# above it owns the input.
|
||
backdrop = SelectScreen(
|
||
backdrop_title,
|
||
backdrop_items,
|
||
default=backdrop_default,
|
||
footer_hint="↑↓ move · Enter select · Esc bye",
|
||
delete_hint="Del remove",
|
||
filterable=False,
|
||
)
|
||
backdrop.chrome = False
|
||
backdrop.banner = banner
|
||
backdrop.footnote = footnote
|
||
self.push(backdrop)
|
||
try:
|
||
return await self.run_screen(dialog) is True
|
||
finally:
|
||
self.pop(backdrop)
|
||
|
||
async def notify_startup(
|
||
self,
|
||
renderable: RenderableType,
|
||
*,
|
||
title: str = "",
|
||
banner: Any | None = None,
|
||
footnote: str | None = None,
|
||
footer_hint: str = "Enter continue",
|
||
) -> None:
|
||
"""Show a chromeless message splash (a banner over a boxed renderable) until it closes."""
|
||
screen = ScrollScreen(renderable, title=title, footer_hint=footer_hint)
|
||
screen.chrome = False
|
||
screen.banner = banner
|
||
screen.footnote = footnote
|
||
await self.run_screen(screen)
|
||
|
||
async def busy_startup(
|
||
self,
|
||
message: str,
|
||
coro: Any,
|
||
*,
|
||
title: str = "",
|
||
banner: Any | None = None,
|
||
footnote: str | None = None,
|
||
interval: float | None = None,
|
||
) -> Any:
|
||
"""Await ``coro``, and show an animated spinner on the chromeless splash while it runs.
|
||
|
||
The method keeps the startup splash on the screen (the same wordmark and box). While
|
||
the awaited task runs, the method replaces the content of the splash with an ASCII
|
||
spinner next to ``message``. Then it returns the result of the task. A background
|
||
timer moves the spinner forward and paints the frame again every ``interval``
|
||
seconds. Before the method returns, it always cancels the timer and pops the splash.
|
||
|
||
Args:
|
||
message: The line next to the spinner (for example "Talking to Wio on COM5…").
|
||
coro: The awaitable to run (for example a device smoke test).
|
||
title: An optional panel title for the splash box.
|
||
banner: The wordmark rows above the box (as on the other startup splashes).
|
||
footnote: The muted line below the box.
|
||
interval: The seconds between two animation steps of the spinner. The default is
|
||
the cadence of the platform, and on a slow console that is the purpose.
|
||
There, a paint costs more time than this loop waited between two paints
|
||
before. Thus a hardcoded rate used all the event loop to draw the spinner,
|
||
and the work that the spinner reported on did not get time to run.
|
||
|
||
Returns:
|
||
The value that ``coro`` resolves to.
|
||
"""
|
||
tick = spinner_interval() if interval is None else interval
|
||
screen = BusyScreen(message, title=title)
|
||
screen.chrome = False
|
||
screen.banner = banner
|
||
screen.footnote = footnote
|
||
self.push(screen)
|
||
|
||
async def animate() -> None:
|
||
while True:
|
||
await asyncio.sleep(tick)
|
||
screen.tick()
|
||
self.invalidate()
|
||
|
||
ticker = asyncio.ensure_future(animate())
|
||
try:
|
||
return await coro
|
||
finally:
|
||
ticker.cancel()
|
||
# Await the cancelled ticker, so that it is never garbage-collected while it is
|
||
# still pending. If it is, the loop shows the error "Task was destroyed but it is
|
||
# pending!", which damages the screen. The spinner is cosmetic, so any problem
|
||
# is ignored.
|
||
try:
|
||
await ticker
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception: # noqa: BLE001 - a small spinner problem must never stop the startup
|
||
pass
|
||
self.pop(screen)
|
||
|
||
async def prompt_pin_startup(
|
||
self,
|
||
device_name: str,
|
||
*,
|
||
error: str = "",
|
||
help_text: str = "",
|
||
banner: Any | None = None,
|
||
footnote: str | None = None,
|
||
) -> str | None:
|
||
"""Ask for the Bluetooth PIN of a companion on the chromeless startup splash.
|
||
|
||
It is drawn like :meth:`notify_startup` / :meth:`select_startup`: a box with a
|
||
border, centred below ``banner``, with no status bars. Thus the PIN request looks like
|
||
a part of the same device-selection flow. When the method asks again after a rejected
|
||
code, ``error`` shows in the box.
|
||
|
||
Args:
|
||
device_name: The display name of the companion, shown in the prompt.
|
||
error: A message about a rejected PIN, to show (empty at the first request).
|
||
help_text: A muted hint below the field.
|
||
banner: The wordmark rows above the box (as on the other startup splashes).
|
||
footnote: The muted line below the box.
|
||
|
||
Returns:
|
||
The entered PIN, or ``None`` if the user pressed Esc to cancel.
|
||
"""
|
||
screen = PinDialog(device_name, error=error, help_text=help_text)
|
||
screen.chrome = False
|
||
screen.banner = banner
|
||
screen.footnote = footnote
|
||
result = await self.run_screen(screen)
|
||
return None if result is CANCEL else result
|
||
|
||
async def prompt_text_startup(
|
||
self,
|
||
title: str,
|
||
*,
|
||
prompt: str = "",
|
||
default: str = "",
|
||
validate: Validator | None = None,
|
||
help_text: str = "",
|
||
banner: Any | None = None,
|
||
footnote: str | None = None,
|
||
) -> str | None:
|
||
"""Ask for a line of text on the chromeless startup splash (for example a TCP host:port).
|
||
|
||
It is drawn like :meth:`prompt_pin_startup`: a :class:`TextScreen` with a border,
|
||
centred below ``banner``, with no status bars. Thus the entry of a network address
|
||
looks like a part of the same device-selection flow. ``validate`` blocks the
|
||
submission of a bad value, the same as the text prompt in the menu.
|
||
|
||
Args:
|
||
title: The short heading in the border of the dialog.
|
||
prompt: The instruction in the box, above the field.
|
||
default: The text that is in the field at the start.
|
||
validate: An optional validator that runs on Enter. If it returns a string, the
|
||
submission is blocked.
|
||
help_text: A muted hint below the field.
|
||
banner: The wordmark rows above the box (as on the other startup splashes).
|
||
footnote: The muted line below the box.
|
||
|
||
Returns:
|
||
The entered text, or ``None`` if the user pressed Esc to cancel.
|
||
"""
|
||
screen = TextScreen(
|
||
title, prompt=prompt, default=default, validate=validate, help_text=help_text
|
||
)
|
||
screen.chrome = False
|
||
screen.banner = banner
|
||
screen.footnote = footnote
|
||
result = await self.run_screen(screen)
|
||
return None if result is CANCEL else result
|
||
|
||
async def reorder(self, title: str, labels: list[str]) -> list[int]:
|
||
"""Show a reorder screen (arrows move rows). Return the final order of the row indices.
|
||
|
||
The Apply action row below the list commits the new order. Back (or Esc) cancels. A
|
||
cancel returns the original (identity) order, so that the caller sees it as "no
|
||
change".
|
||
"""
|
||
result = await self.run_screen(ReorderScreen(title, labels))
|
||
return list(range(len(labels))) if result is CANCEL else result
|
||
|
||
async def text(
|
||
self,
|
||
title: str,
|
||
*,
|
||
prompt: str = "",
|
||
default: str = "",
|
||
validate: Validator | None = None,
|
||
help_text: str = "",
|
||
password: bool = False,
|
||
byte_limit: int | None = None,
|
||
compose: bool = False,
|
||
floating: bool = False,
|
||
) -> str | None:
|
||
"""Show a text prompt. Return the string, or ``None`` if the user cancelled.
|
||
|
||
``floating`` makes sure that the prompt draws as a centred dialog, also on an empty
|
||
stack. Use it for a question on the way into a tool (a remote-admin password, a typed
|
||
trace target), not for the page of a tool. The prompt then floats over the menu, as
|
||
:meth:`button_dialog` and :meth:`typed_confirm` do (refer to :meth:`run_dialog`).
|
||
When a screen is already below it, there is no difference.
|
||
|
||
``byte_limit`` puts the shared UTF-8 byte gauge on the field, and blocks a submission
|
||
that is longer than the limit. Use it for a field that goes into a packet with a
|
||
maximum size (refer to :class:`TextScreen`).
|
||
|
||
``compose`` gives the field the colours of the compose line of a chat. Use it for a
|
||
field that holds a message.
|
||
"""
|
||
screen = TextScreen(
|
||
title,
|
||
prompt=prompt,
|
||
default=default,
|
||
validate=validate,
|
||
help_text=help_text,
|
||
password=password,
|
||
byte_limit=byte_limit,
|
||
compose=compose,
|
||
)
|
||
runner = self.run_dialog if floating else self.run_screen
|
||
result = await runner(screen)
|
||
return None if result is CANCEL else result
|
||
|
||
async def confirm(self, title: str, *, default: bool = True) -> bool | None:
|
||
"""Show a yes/no prompt. Return the bool, or ``None`` if the user cancelled."""
|
||
result = await self.run_screen(ConfirmScreen(title, default=default))
|
||
return None if result is CANCEL else result
|
||
|
||
async def button_dialog(
|
||
self,
|
||
prompt: str | Text,
|
||
buttons: list[tuple[str, Any]],
|
||
*,
|
||
title: str = "",
|
||
default: int = 0,
|
||
keys: dict[str, Any] | None = None,
|
||
footer_hint: str = "←→ choose · Enter select · Esc cancel",
|
||
prompt_style: str = "",
|
||
button_style: str = "selected",
|
||
button_idle_style: str = "muted",
|
||
border_style: str = "border",
|
||
lane: Sequence[fkeys.FPair | None] | None = None,
|
||
) -> Any:
|
||
"""Show a centred button dialog. Return the chosen value, or ``None`` if cancelled.
|
||
|
||
A reusable dialog with a prompt above buttons (refer to
|
||
:class:`~meshterm.ui.tui.prompt.ButtonDialog`). The colours, the prompt, the
|
||
buttons, and the single-key shortcuts are all parameters. Thus a caller can give it
|
||
a theme (for example, red for a destructive action), or connect immediate y/n keys.
|
||
``keys`` maps a shortcut character to the value that it commits. The prompt can be a
|
||
:class:`Text` that already has styles (refer to the ButtonDialog docs).
|
||
"""
|
||
screen = ButtonDialog(
|
||
prompt,
|
||
buttons,
|
||
title=title,
|
||
default=default,
|
||
keys=keys,
|
||
footer_hint=footer_hint,
|
||
prompt_style=prompt_style,
|
||
button_style=button_style,
|
||
button_idle_style=button_idle_style,
|
||
border_style=border_style,
|
||
lane=lane,
|
||
)
|
||
result = await self.run_dialog(screen)
|
||
return None if result is CANCEL else result
|
||
|
||
@asynccontextmanager
|
||
async def _floated(self) -> AsyncIterator[None]:
|
||
"""Make sure that all that is pushed in the block draws as a box, not as a full frame.
|
||
|
||
On an empty stack, a lone floating screen is drawn as the background: the full
|
||
chrome, which fills the terminal, and no dialog (refer to :meth:`_base_index`). Thus
|
||
a base is pushed first, and popped when the block ends. The base is the **root**:
|
||
the main menu. The menu is popped while a tool runs, but it is still the screen where
|
||
the user chose the tool. Thus a question on the way into a tool floats over the menu
|
||
that it came from. A dialog is smaller than the frame, and the area around it must
|
||
show the page behind it, never an empty frame. A blank base is the fallback for a
|
||
session with no declared root. When a background is already there, there is nothing
|
||
to do: the dialog floats over it.
|
||
"""
|
||
backdrop: Screen | None = None
|
||
if not self._stack:
|
||
backdrop = self._root or ScrollScreen("", floating=False, footer_hint="")
|
||
self.push(backdrop)
|
||
try:
|
||
yield
|
||
finally:
|
||
if backdrop is not None:
|
||
self.pop(backdrop)
|
||
|
||
async def run_dialog(self, screen: Screen) -> Any:
|
||
"""Run a floating dialog one time, drawn as a centred box (refer to :meth:`_floated`)."""
|
||
async with self._floated():
|
||
return await self.run_screen(screen)
|
||
|
||
async def typed_confirm(self, warning: str, word: str, *, title: str = "Are you sure?") -> bool:
|
||
"""Ask the user to type ``word`` before a destructive action. Return whether it was typed.
|
||
|
||
The method shows the :class:`~meshterm.ui.tui.prompt.TypedConfirmDialog` with the
|
||
error theme, and changes its result into a plain bool: ``True`` only when the user
|
||
typed the word, ``False`` when the user backed out with Esc.
|
||
"""
|
||
result = await self.run_dialog(TypedConfirmDialog(warning, word, title=title))
|
||
return result is True
|
||
|
||
async def autocomplete(
|
||
self,
|
||
title: str,
|
||
choices: list[str],
|
||
*,
|
||
prompt: str = "",
|
||
default: str = "",
|
||
validate: Validator | None = None,
|
||
) -> str | None:
|
||
"""Show a free-text prompt with suggestions. Return the text, or ``None`` if cancelled."""
|
||
result = await self.run_dialog(
|
||
AutocompleteScreen(title, choices, prompt=prompt, default=default, validate=validate)
|
||
)
|
||
return None if result is CANCEL else result
|
||
|
||
async def message_dialog(self, message: Text | str, *, title: str = "") -> None:
|
||
"""Show a short result in a centred dialog with one OK button.
|
||
|
||
This is the light acknowledgement counterpart of :meth:`scroll`. A one-line result
|
||
("✓ flood advertisement sent") is too small for a full result screen. Thus it floats
|
||
as a small dialog over the screen below it. Enter (OK) or Esc closes it. The border
|
||
takes the strongest tone of the message (refer to :func:`_message_border`). Thus an
|
||
error shows in red, and a success stays in the grey of the standard border. The method calls
|
||
:meth:`button_dialog`, which floats the dialog over a blank base when the stack is
|
||
empty (a tool that runs directly from the menu, which is popped while the tool
|
||
runs).
|
||
|
||
Args:
|
||
message: The result to show: a :class:`Text` that already has styles (the note
|
||
markup stays in the dialog), or a plain string.
|
||
title: An optional dialog heading (usually the name of the tool or of the
|
||
action).
|
||
"""
|
||
await self.button_dialog(
|
||
message,
|
||
[("OK", "ok")],
|
||
title=title,
|
||
footer_hint="Enter OK",
|
||
border_style=_message_border(message),
|
||
)
|
||
|
||
async def scroll(
|
||
self, renderable: RenderableType, *, title: str = "", footer_hint: str = ""
|
||
) -> None:
|
||
"""Show a renderable on a scrollable screen that the user can close (a result screen)."""
|
||
screen = ScrollScreen(
|
||
renderable,
|
||
title=title,
|
||
footer_hint=footer_hint or "↑↓ PgUp/PgDn scroll · Esc close",
|
||
)
|
||
await self.run_screen(screen)
|
||
|
||
def progress(self, title: str = "Working") -> TuiProgress:
|
||
"""Return a progress context manager that uses a pushed :class:`ProgressScreen`."""
|
||
return TuiProgress(self, title)
|
||
|
||
@asynccontextmanager
|
||
async def busy_overlay(
|
||
self,
|
||
message: str = "",
|
||
*,
|
||
title: str = "",
|
||
interval: float | None = None,
|
||
) -> AsyncIterator[BusyOverlay]:
|
||
"""Float a skeleton card on top of all other content while a block runs.
|
||
|
||
Put a slow operation that affects the screen in this block. The most useful case is
|
||
a device menu navigation, which can otherwise stay on a blank frame while the
|
||
companion answers::
|
||
|
||
async with session.busy_overlay("reading…", title="Nodes"):
|
||
await slow_work()
|
||
|
||
The card takes the place of the screen whose data MeshTerm reads: a title, and the
|
||
one-cell working chip next to the caption. While the block runs, a background timer
|
||
moves the chip forward, fades the card in, and paints the frame again. At the exit,
|
||
the card is always cleared and the timer cancelled, also after an error. The card
|
||
fades in from black (refer to :attr:`BusyOverlay.brightness`). Thus a quick
|
||
operation paints only an almost black card, and the card never appears suddenly. It
|
||
is drawn only in the gaps between screens (refer to :meth:`_overlay_visible`). Thus
|
||
it shows the wait, and it does not cover a prompt that the user works in.
|
||
|
||
Args:
|
||
message: An optional caption next to the working chip.
|
||
title: An optional heading that names the screen whose data MeshTerm reads.
|
||
interval: The seconds between two animation steps (also the cadence of the fade
|
||
paints). The default is the cadence of the platform. Refer to
|
||
:meth:`busy_startup`, which has the same hazard: the card animates over a
|
||
device read. Thus a rate that the console cannot sustain takes the loop from
|
||
the read that the card covers.
|
||
|
||
Yields:
|
||
The live :class:`BusyOverlay`, if the caller must change its caption.
|
||
"""
|
||
tick = spinner_interval() if interval is None else interval
|
||
# A nested busy_overlay keeps the outer overlay (the outermost wait owns the screen).
|
||
# Its own body still runs, but it does not install a second card.
|
||
if self._overlay is not None:
|
||
yield self._overlay
|
||
return
|
||
overlay = BusyOverlay(message, title=title)
|
||
self._overlay = overlay
|
||
if self._overlay_visible():
|
||
self.invalidate() # start the fade-in immediately, before the first tick
|
||
|
||
async def animate() -> None:
|
||
while True:
|
||
await asyncio.sleep(tick)
|
||
overlay.tick()
|
||
# Paint again only when the card is on the screen. Thus an overlay that waits
|
||
# behind a live prompt does not cause paints of that prompt with no purpose.
|
||
if self._overlay_visible():
|
||
self.invalidate()
|
||
|
||
ticker = asyncio.ensure_future(animate())
|
||
try:
|
||
yield overlay
|
||
finally:
|
||
ticker.cancel()
|
||
try:
|
||
await ticker
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception: # noqa: BLE001 - a cosmetic overlay must never break a flow
|
||
pass
|
||
self._overlay = None
|
||
self.invalidate()
|
||
|
||
@asynccontextmanager
|
||
async def busy_dialog(
|
||
self,
|
||
message: str = "",
|
||
*,
|
||
title: str = "",
|
||
interval: float | None = None,
|
||
) -> AsyncIterator[BusyDialog]:
|
||
"""Float a modal busy card over the current screen while a block runs.
|
||
|
||
This is the counterpart of :meth:`busy_overlay` on the stack. A screen must use it
|
||
when that screen itself starts the slow work::
|
||
|
||
async with session.busy_dialog("saving Lakeside…", title="Channels"):
|
||
await device.set_channel(...)
|
||
|
||
Two things make it different from the overlay, and for these two reasons the
|
||
overlay could not do this job:
|
||
|
||
- It is a real :class:`~meshterm.ui.tui.screen.BusyDialog` that is pushed on the
|
||
stack. Thus it draws over a hub, not only in the gaps between screens. The overlay
|
||
paints only on an empty stack (refer to :meth:`_overlay_visible`), and the stack
|
||
is never empty while the user visits a hub.
|
||
- It is modal, so it **owns the keyboard**. A hub that :meth:`stay` keeps up stays
|
||
armed while the work runs. Without a modal layer, each key that the user presses
|
||
during the wait still gets to the hub. Letters go into its live filter, and the
|
||
list comes back with nothing in it. Esc resolves the hub, and the hub is returned
|
||
at the instant when the work finishes: the screen seems to close by itself a
|
||
moment later. Now these key presses get to this card, and stop there.
|
||
|
||
Nesting keeps the outermost card. Thus a batch of writes shows as one wait, and a
|
||
box does not flash for each write. The inner block still runs. The caller can
|
||
change the caption of the card while the work continues (``busy.message = …``).
|
||
This is how a sequence tells which step it is on.
|
||
|
||
Different from :meth:`busy_overlay`, this card has no fade-in. The fade makes sure
|
||
that a fast operation shows nothing, and that is a good exchange for a cosmetic
|
||
card. But this card is also the key guard, and a guard that arrives late lets some
|
||
keys through.
|
||
|
||
Args:
|
||
message: The caption next to the spinner.
|
||
title: An optional heading that names the feature that does the work.
|
||
interval: The seconds between two animation steps of the spinner. The default is
|
||
the cadence of the platform (refer to :meth:`busy_startup`, which has the
|
||
same hazard: the card animates over a device read, so a rate that the console
|
||
cannot sustain takes the loop from the work that the card reports on).
|
||
|
||
Yields:
|
||
The live :class:`~meshterm.ui.tui.screen.BusyDialog`, so that its caption can
|
||
change.
|
||
"""
|
||
tick = spinner_interval() if interval is None else interval
|
||
if self._busy is not None: # a nested wait uses the card that its caller already shows
|
||
yield self._busy
|
||
return
|
||
screen = BusyDialog(message, title=title)
|
||
# A lone floating screen on an empty stack is drawn as the background, full-frame and
|
||
# with borders, not as a box. Thus it gets the same blank backdrop that each other
|
||
# dialog uses (refer to :meth:`run_dialog`).
|
||
backdrop = None
|
||
if not self._stack:
|
||
backdrop = ScrollScreen("", floating=False, footer_hint="")
|
||
self.push(backdrop)
|
||
self._busy = screen
|
||
self.push(screen)
|
||
|
||
async def animate() -> None:
|
||
while True:
|
||
await asyncio.sleep(tick)
|
||
screen.tick()
|
||
self.invalidate()
|
||
|
||
ticker = asyncio.ensure_future(animate())
|
||
try:
|
||
yield screen
|
||
finally:
|
||
ticker.cancel()
|
||
try:
|
||
await ticker
|
||
except asyncio.CancelledError:
|
||
pass
|
||
except Exception: # noqa: BLE001 - a small spinner problem must never break a flow
|
||
pass
|
||
self._busy = None
|
||
self.pop(screen)
|
||
if backdrop is not None:
|
||
self.pop(backdrop)
|
||
|
||
# --- application lifecycle ------------------------------------------------
|
||
|
||
async def run(self, main: Any) -> None:
|
||
"""Run ``main`` (the menu loop) in the full-screen application.
|
||
|
||
The method starts the prompt_toolkit event loop, and drives ``main`` as a background
|
||
task on that same loop. When ``main`` returns, the method exits the application. If
|
||
``main`` raises an exception, the method raises it again after the loop unwinds.
|
||
|
||
Args:
|
||
main: The coroutine that drives the session (usually the menu loop).
|
||
"""
|
||
self._app = self._build_app()
|
||
keyboard = get_platform().modifier_watch
|
||
if keyboard:
|
||
# The Shift watcher changes the labels of the F-key lane live. It reports from
|
||
# its own thread, so go to the app loop for the paint. If the watcher cannot
|
||
# start (no input device, no permission), the lane stays static. Refer to the
|
||
# module doc.
|
||
loop = asyncio.get_running_loop()
|
||
modifier_watch.start(lambda: loop.call_soon_threadsafe(self.invalidate), keyboard)
|
||
box: dict[str, BaseException] = {}
|
||
task: dict[str, asyncio.Future] = {}
|
||
|
||
async def driver() -> None:
|
||
try:
|
||
await main
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except BaseException as exc: # noqa: BLE001 - raised again after the app unwinds
|
||
box["exc"] = exc
|
||
finally:
|
||
# ``is_running`` instead of ``not is_done``: prompt_toolkit clears its future
|
||
# on the way out. Thus ``is_done`` reads False again after the app has
|
||
# finished, and ``exit()`` on a finished app raises. This ``finally`` now
|
||
# also runs after the app is gone (the quit chords exit the app from below
|
||
# us, and then ``run`` cancels this task). Thus it must tell "still up" from
|
||
# "already down", instead of "not yet finished".
|
||
if self._app is not None and self._app.is_running:
|
||
self._app.exit()
|
||
|
||
def pre_run() -> None:
|
||
task["driver"] = asyncio.ensure_future(driver())
|
||
|
||
# A held Esc opens the quit box, where a front end can detect one (the emulator).
|
||
# The SIGTERM of the launcher also quits through here. Refer to
|
||
# services.hold_to_quit. The listener costs nothing where nothing reports that a key
|
||
# goes down.
|
||
watch = EscHoldWatch(self, asyncio.get_running_loop())
|
||
hold_to_quit.listen(watch)
|
||
try:
|
||
# Do not let prompt_toolkit install its own loop exception handler. On a stray
|
||
# error in a background task, that handler prints a traceback and a "Press ENTER
|
||
# to continue..." prompt directly over the full-screen UI. When it is disabled,
|
||
# the default handler of asyncio logs such errors to the ``asyncio`` logger
|
||
# instead. That logger goes to the file log (refer to
|
||
# :func:`meshterm.persistence.logging.configure_logging`), and never touches the
|
||
# screen. Errors from ``main`` still propagate through ``driver``/``box``.
|
||
await self._app.run_async(pre_run=pre_run, set_exception_handler=False)
|
||
finally:
|
||
# Whichever way the app exited, it tears down from here. A SIGTERM that arrives
|
||
# now must join this teardown, not interrupt it.
|
||
self._leaving = True
|
||
hold_to_quit.leaving()
|
||
hold_to_quit.unlisten(watch)
|
||
# The app can also exit from below the driver. The quit chords (^Q/^C) call
|
||
# ``Application.exit`` from their detached confirm, or directly from the key handler
|
||
# on the second press (refer to request_quit). Thus ``run_async`` returns while
|
||
# ``main`` still waits on the screen that was open. If nothing is done, that
|
||
# coroutine is abandoned during its await, and its ``finally`` never runs. That
|
||
# ``finally`` closes the history runs and the chat runs, stops the background
|
||
# services, and arms the exit watchdog. Cancel the coroutine and wait for the
|
||
# unwind. Thus a quit from any screen tears down exactly as much as a quit from the
|
||
# menu.
|
||
driver_task = task.get("driver")
|
||
if driver_task is not None and not driver_task.done():
|
||
driver_task.cancel()
|
||
try:
|
||
await driver_task
|
||
except asyncio.CancelledError:
|
||
pass
|
||
if "exc" in box:
|
||
raise box["exc"]
|
||
|
||
def _build_app(self) -> Application:
|
||
"""Build the prompt_toolkit application, its layout, and its key bindings."""
|
||
base_control = ClusterTextControl(self._render_base, focusable=True)
|
||
base_window = Window(base_control, always_hide_cursor=True)
|
||
# One float with a centred box for each dialog layer on the stack, from bottom to
|
||
# top. Each float renders the k-th screen that floats above the background (refer to
|
||
# :meth:`_float_layers`). Thus a dialog that opens over a dialog draws over it, and
|
||
# the two keep their own sizes. The lower dialog is not stretched to fill the frame.
|
||
# A large fixed pool covers all realistic nesting. The ConditionalContainer hides
|
||
# the layers that this frame does not use.
|
||
dialog_floats = [
|
||
Float(
|
||
ConditionalContainer(
|
||
Window(
|
||
ClusterTextControl(lambda i=i: self._render_float_layer(i)),
|
||
always_hide_cursor=True,
|
||
),
|
||
filter=Condition(lambda i=i: len(self._float_layers()) > i),
|
||
)
|
||
)
|
||
for i in range(_MAX_DIALOG_LAYERS)
|
||
]
|
||
# The busy overlay is the last float, so it draws on top of each dialog float: the
|
||
# top of the z-order. It is a ``Window`` sized to its content (dont_extend_*), with
|
||
# no anchors. Thus the FloatContainer centres only its skeleton card over the screen,
|
||
# and does not blank the screen.
|
||
overlay_window = Window(
|
||
ClusterTextControl(self._render_overlay),
|
||
always_hide_cursor=True,
|
||
dont_extend_width=True,
|
||
dont_extend_height=True,
|
||
)
|
||
root = FloatContainer(
|
||
content=base_window,
|
||
floats=[
|
||
*dialog_floats,
|
||
Float(
|
||
ConditionalContainer(overlay_window, filter=Condition(self._overlay_visible))
|
||
),
|
||
],
|
||
)
|
||
app = Application(
|
||
layout=Layout(root, focused_element=base_window),
|
||
key_bindings=self._key_bindings(),
|
||
full_screen=True,
|
||
mouse_support=False,
|
||
# Keeps the live monitor counter in the header up to date. The value is set for
|
||
# each platform, so that a host where an idle paint is expensive can wait
|
||
# longer between frames.
|
||
refresh_interval=get_platform().tick_s,
|
||
# Paint as soon as the loop is free, instead of a spin of the event loop for up
|
||
# to 10 ms first. The default postpone of prompt_toolkit protects a UI whose own
|
||
# output floods it with invalidations (the case that caused it was a terminal
|
||
# multiplexer). Here an invalidation is a key press or the 2 s header tick. Thus
|
||
# the delay gives no advantage. We measured it as a constant 3 ms on the
|
||
# PicoCalc and 10 ms on the desktop, for each key press.
|
||
max_render_postpone_time=None,
|
||
input=self._input,
|
||
output=self._resolve_output(),
|
||
# None keeps the default of the output. Refer to _color_depth for the reason
|
||
# why the two prompt_toolkit outputs do not agree, and why the function only
|
||
# increases the depth.
|
||
color_depth=_color_depth(),
|
||
)
|
||
if fastrender.enabled():
|
||
# Use the row-diff renderer for plain full-screen frames. It is built with the
|
||
# same arguments that Application gave the standard renderer. Thus all is
|
||
# unchanged (CPR, alternate screen, mouse, cursor shape), except the paint
|
||
# itself.
|
||
app.renderer = fastrender.FastRenderer(
|
||
app._merged_style,
|
||
app.output,
|
||
full_screen=True,
|
||
mouse_support=False,
|
||
cpr_not_supported_callback=app.cpr_not_supported_callback,
|
||
frame_source=self._plain_frame,
|
||
)
|
||
return app
|
||
|
||
def _resolve_output(self) -> Any:
|
||
"""The output that the app renders to: the real terminal, with pins and possibly wider.
|
||
|
||
Only the real terminal is wrapped (``self._output is None``, so prompt_toolkit would
|
||
build its own output). A test that gives its own output keeps the exact size and the
|
||
exact bytes that it set. Thus a headless render stays deterministic. There are two
|
||
wraps. Each one applies where its own gate says so, and either one can be the only
|
||
one:
|
||
|
||
* :class:`~meshterm.ui.tui.colsnap.PinnedOutput`, where
|
||
:func:`~meshterm.ui.tui.colsnap.enabled` is true. Each glyph that the renderer of
|
||
prompt_toolkit writes goes to the column that the renderer measured for it.
|
||
* :class:`_WidthExtendedOutput`, where :func:`_reclaim_last_column` is true (or, if
|
||
the terminal decides, where :func:`_probe_hides_last_column` is true). This wrap is
|
||
the outermost, because the two renderers and the compositor lay out against the
|
||
size that it reports.
|
||
|
||
With neither wrap, this is ``None``, and prompt_toolkit builds its own output as
|
||
before.
|
||
"""
|
||
pin = colsnap.enabled()
|
||
reclaim = _reclaim_last_column()
|
||
if self._output is not None or not (pin or reclaim is not False):
|
||
return self._output
|
||
from prompt_toolkit.output.defaults import create_output
|
||
|
||
output: Any = create_output()
|
||
if reclaim is None:
|
||
reclaim = _probe_hides_last_column(output)
|
||
if not (pin or reclaim):
|
||
return None
|
||
if pin:
|
||
output = colsnap.PinnedOutput(output)
|
||
if reclaim:
|
||
output = _WidthExtendedOutput(output)
|
||
return 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_body_size(self) -> tuple[int, int]:
|
||
"""Return the ``(width, height)`` in cells that the body of the base screen can use.
|
||
|
||
This method copies the layout arithmetic of
|
||
:func:`~meshterm.ui.tui.frame.compose_base`. Thus a full-screen screen (for example
|
||
the map) can size its own content to fill the frame exactly, and it does not wait for
|
||
a paint to learn its height. It copies the two branches of that function. A
|
||
borderless platform replaces the two border rows and the four padding columns of the
|
||
panel with one title-bar row. If a screen uses the arithmetic for borders there, one
|
||
row of the frame that it gets stays blank permanently. The width is the width of
|
||
the current base screen. A :attr:`~meshterm.ui.tui.screen.Screen.flush` screen adds
|
||
the two padding columns that it does not use.
|
||
|
||
Returns:
|
||
The width of the inner content and the height of the body viewport, both in
|
||
cells.
|
||
"""
|
||
cols, rows = self._size()
|
||
header_h = len(frame.header_lines(self._header(cols), cols))
|
||
if get_platform().frame_border:
|
||
base = self._base_screen()
|
||
inset = frame.panel_inset(base is not None and base.flush)
|
||
# minus footer(1) and panel border(2)
|
||
return cols - inset, max(1, rows - header_h - 1 - 2)
|
||
return cols, max(1, rows - header_h - 1 - 1) # minus footer(1) and title bar(1)
|
||
|
||
def _base_index(self) -> int:
|
||
"""The stack index of the full-frame background screen.
|
||
|
||
The background is the top screen that does not float: a menu, the map, or a list
|
||
that fills the frame. Each floating dialog above it is drawn as a centred box over
|
||
it (refer to :meth:`_float_layers`). When all the screens on the stack float (a tool
|
||
whose own main screen is a floating select, for example Channels), the bottom screen
|
||
is the background: the dialogs above it must float over it.
|
||
"""
|
||
for i in range(len(self._stack) - 1, -1, -1):
|
||
if not getattr(self._stack[i], "floating", False):
|
||
return i
|
||
return 0
|
||
|
||
def _base_screen(self) -> Screen | None:
|
||
"""The screen drawn as the full-frame background, or ``None`` when the stack is empty."""
|
||
if not self._stack:
|
||
return None
|
||
return self._stack[self._base_index()]
|
||
|
||
def _float_layers(self) -> list[Screen]:
|
||
"""The floating dialogs on the stack above the background, from bottom to top.
|
||
|
||
Each one is drawn as its own centred box over the dialogs below it. Thus when a
|
||
dialog opens over a dialog, the lower dialog keeps its own size, and is not stretched
|
||
to fill the frame (this was the old fault of the compositor with one float).
|
||
"""
|
||
if not self._stack:
|
||
return []
|
||
return self._stack[self._base_index() + 1 :]
|
||
|
||
def _has_float(self) -> bool:
|
||
"""Whether a dialog floats over the background in this frame."""
|
||
return bool(self._float_layers())
|
||
|
||
def _emit(self, text: str, layer: str = "base") -> ANSI:
|
||
r"""Wrap a composed frame as a prompt_toolkit :class:`ANSI`.
|
||
|
||
A row with a glyph that the terminal may draw narrower than the cells that pt
|
||
reserves for it gets a full paint, not a differential paint.
|
||
|
||
prompt_toolkit paints differentially. It writes again only the cells that changed
|
||
since the last frame, and it moves the cursor relative to its own width model. That
|
||
is correct only while each glyph is one cell wide. The terminal can draw a width-2
|
||
glyph in one cell (an emoji in a chat line, a menu icon). Then all the text to its
|
||
right on that row is one column to the left of where pt thinks it is. A later paint
|
||
that jumps into the row (and skips the unchanged emoji) writes at the column of pt,
|
||
one column after the content that it must write over. Thus the old cell stays.
|
||
|
||
The repair has a narrow requirement: a row that has such a glyph must be written
|
||
again in full, from column 0, so that the cursor of the terminal lays it out again.
|
||
It is not necessary to erase the screen, or to touch the rows around it. pt goes down
|
||
a row with ``\\r\\n``, which puts the terminal back to a real column 0, whatever the
|
||
drift above it. Thus the incorrect alignment can never spread past its row.
|
||
|
||
Thus the changed rows are written again, and nothing else is (:meth:`_scrub_rows`).
|
||
Three rules limit the work to that:
|
||
|
||
* **Nothing changed → nothing to do.** The app paints on a 1 Hz timer to move the
|
||
pulse of the header, and an idle screen composes the same each time. pt writes no
|
||
cells, so its cursor cannot drift. The last paint already left the terminal
|
||
aligned. (This case alone caused the flicker: once each second, the app erased a
|
||
screen with an emoji and wrote it again, with no change.)
|
||
* **No wide glyph drawn → nothing to do.** The plain differential paint is exact.
|
||
* **In all other cases, only the rows that changed.** The background composes one
|
||
line for each terminal row, so its diff maps directly onto rows. Thus a header that
|
||
ticks paints the header again, not the screen below it. A float is a centred box,
|
||
and the layout places its rows, not us. Thus when a dialog changes (or closes,
|
||
refer to :meth:`_reconcile_layers`), the fallback still removes the cached frame of
|
||
pt. The same occurs for a frame whose height changed. The user causes these cases,
|
||
and they are not frequent. The timer is neither of the two.
|
||
"""
|
||
entry = (text, _has_wide_glyph(text))
|
||
previous = self._layers.get(layer)
|
||
if previous == entry:
|
||
return ANSI(text)
|
||
self._layers[layer] = entry
|
||
if any(wide for _text, wide in self._layers.values()):
|
||
rows = _changed_rows(previous[0], text) if previous is not None else None
|
||
# Line i of the background is terminal row i. No other layer can claim that.
|
||
if layer != "base" or rows is None or not self._scrub_rows(rows):
|
||
self._invalidate_last_frame()
|
||
return ANSI(text)
|
||
|
||
def _repaint_if_wide(self) -> None:
|
||
"""Change this paint into a full paint if a drawn layer has a wide glyph."""
|
||
if any(wide for _text, wide in self._layers.values()):
|
||
self._invalidate_last_frame()
|
||
|
||
def _reconcile_layers(self) -> None:
|
||
"""Forget the layers that this paint will not draw, and count their removal as a change.
|
||
|
||
A float is hidden when its ``Window`` is removed from the layout. Thus a dialog that
|
||
closes only stops its calls to :meth:`_emit`. Without this method, nothing sees that
|
||
the dialog is gone. Then the cells that it gives back to the base are written again
|
||
in small parts, over a row that the terminal may draw out of position.
|
||
The base layer calls this method at the start of the paint, because the background
|
||
is composed before the floats above it.
|
||
"""
|
||
live = {f"float{i}" for i in range(len(self._float_layers()))}
|
||
if self._base_screen() is not None:
|
||
live.add("base")
|
||
if self._overlay_visible():
|
||
live.add("overlay")
|
||
gone = [layer for layer in self._layers if layer not in live]
|
||
for layer in gone:
|
||
self._layers.pop(layer)
|
||
if gone:
|
||
self._repaint_if_wide()
|
||
|
||
def _render_base(self) -> ANSI:
|
||
"""Render the background screen with the persistent header and footer around it."""
|
||
self._reconcile_layers() # the paint starts here: the background composes first
|
||
cols, rows = self._size()
|
||
base = self._base_screen()
|
||
if base is None:
|
||
return ANSI("")
|
||
scrub = base.consume_edge_scrub()
|
||
if scrub:
|
||
self._scrub_right_columns(scrub)
|
||
# A bare base (the QR code of the share screen) is only its body on blank rows.
|
||
if base.bare:
|
||
return self._emit(frame.compose_bare(base, cols, rows))
|
||
# A chromeless base (the startup splash) has no header or footer bars. It is
|
||
# centred below its banner, instead of stretched across the terminal.
|
||
if not base.chrome:
|
||
return self._emit(frame.compose_startup(base, cols, rows))
|
||
footer = frame.fitted_footer(self.top or base, cols)
|
||
lane = self._fkey_lane(self.top or base)
|
||
return self._emit(
|
||
frame.compose_base(self._header(cols), base, footer, cols, rows, footer_lane=lane)
|
||
)
|
||
|
||
def _plain_frame(self) -> str | None:
|
||
"""The full frame as rows, when the fast path can do this paint.
|
||
|
||
The frame is the background screen, with each floating dialog composited over it
|
||
exactly where the ``FloatContainer`` of prompt_toolkit would centre it (refer to
|
||
:func:`~meshterm.ui.tui.frame.composite_float`). Thus a confirm, a picker, or the
|
||
packet viewer stays on the row-diff path, and does not pay for the full grid rebuild
|
||
of prompt_toolkit at each key press.
|
||
|
||
The method returns ``None`` (that is, "let prompt_toolkit lay this one out") for the
|
||
two frames that we do not place ourselves: the busy overlay, which is a ``Window``
|
||
sized to its content that the float container measures, and an empty stack, which
|
||
has no background.
|
||
|
||
Each layer still goes through its usual renderer (:meth:`_render_base`,
|
||
:meth:`_render_float_layer`). Thus the layer records that those renderers keep are
|
||
the same, whichever path the paint takes.
|
||
"""
|
||
if self._overlay_visible() or not self._stack:
|
||
return None
|
||
rows_out = self._render_base().value.split("\n")
|
||
layers = self._float_layers()
|
||
if layers:
|
||
cols, rows = self._size()
|
||
for index in range(len(layers)):
|
||
rows_out = frame.composite_float(
|
||
rows_out, self._render_float_layer(index).value, cols, rows
|
||
)
|
||
return "\n".join(rows_out)
|
||
|
||
@staticmethod
|
||
def _fkey_lane(active: Screen) -> Callable[[], Text] | None:
|
||
"""A deferred F-key lane row for ``active``, or ``None`` on a platform without a lane.
|
||
|
||
The row is deferred because the frame resolves it after the body renders. A lane
|
||
dims the slots whose action does nothing, and that information comes from the
|
||
scroll metrics that this paint will store (refer to
|
||
:func:`~meshterm.ui.tui.fkeys.default_lane`).
|
||
"""
|
||
deck = fkeys.active_deck()
|
||
if deck is None:
|
||
return None
|
||
return lambda: deck.lane_text(active.fkey_lane, shifted=modifier_watch.shift_down())
|
||
|
||
def _render_float_layer(self, index: int) -> ANSI:
|
||
"""Render the ``index``-th floating dialog (from bottom to top) as a centred box."""
|
||
layers = self._float_layers()
|
||
if index >= len(layers):
|
||
return ANSI("")
|
||
cols, rows = self._size()
|
||
return self._emit(frame.compose_dialog(layers[index], cols, rows), f"float{index}")
|
||
|
||
def _overlay_visible(self) -> bool:
|
||
"""Whether to paint the busy overlay in this frame.
|
||
|
||
The overlay shows only on an empty screen stack. Thus the card appears only in the
|
||
"black screen" gaps that a device operation opens between screens (for example, a
|
||
menu navigation before the first prompt of the tool). It never covers a dialog that
|
||
the user must read. It also stays unpainted during the initial hold of the overlay
|
||
(:attr:`BusyOverlay.brightness` is 0). Thus an operation that finishes during the
|
||
hold shows nothing and never flashes.
|
||
"""
|
||
return self._overlay is not None and not self._stack and self._overlay.brightness > 0
|
||
|
||
def _render_overlay(self) -> ANSI:
|
||
"""Render the skeleton card of the busy overlay (only when :meth:`_overlay_visible`)."""
|
||
if self._overlay is None:
|
||
return ANSI("")
|
||
return self._emit(self._overlay.render(), "overlay")
|
||
|
||
# --- input ---------------------------------------------------------------
|
||
|
||
def _key_bindings(self) -> KeyBindings:
|
||
"""Build the global key bindings that send normalized actions to the top screen.
|
||
|
||
Each key (navigation, Ctrl chord, typed character, pasted text) goes through
|
||
:meth:`_dispatch`. Thus the right-Ctrl rescue there covers the full app, not only the
|
||
screen actions.
|
||
"""
|
||
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.Any)
|
||
def _typed(event: Any) -> None: # noqa: ANN401
|
||
data = event.data
|
||
if not data:
|
||
return
|
||
if len(data) == 1 and data.isprintable():
|
||
self._dispatch("text", data)
|
||
elif len(data) > 1:
|
||
# A bracketed paste (Ctrl-V, right-click, Ctrl-Shift-V) arrives as one
|
||
# multi-character sequence. Give all of it to the top screen as a paste that
|
||
# the screen can confirm and insert. The old len==1 guard removed it instead.
|
||
self._dispatch("paste", data)
|
||
|
||
return kb
|
||
|
||
def _dispatch(self, action: str, data: str = "") -> None:
|
||
"""Send an action to the top screen, and paint the frame again.
|
||
|
||
In all the app, the right Ctrl key is first read as Ctrl. While the right Ctrl key is
|
||
physically held, a plain navigation key, or any bare letter that arrives as text,
|
||
changes into its Ctrl chord (refer to :func:`_right_ctrl_down`, :data:`_CTRL_CHORDS`,
|
||
:data:`_CTRL_LETTER_CHORDS`). When the console already reported the chord, this does
|
||
nothing. When a keyboard layout claimed the right Ctrl key and removed it, so that
|
||
only a bare character arrived, this is the rescue. Each binding goes through this
|
||
method. Thus the rescue also covers the chords of the session, not only the screen
|
||
actions: right Ctrl-V pastes into a compose line instead of a typed ``v``, and right
|
||
Ctrl-C quits.
|
||
|
||
This method answers the session-level actions itself, and does not send them to a
|
||
screen. No screen ever gets ``to_menu``, ``quit``, ``quit_now``, or
|
||
``paste_clipboard``.
|
||
|
||
The paint keeps the fast differential paint of prompt_toolkit. A frame with a glyph
|
||
that the terminal may draw narrower than the cells that pt reserves for it (an emoji)
|
||
changes itself into a full paint when it is composed. Refer to :meth:`_emit`.
|
||
"""
|
||
# An F-key resolves against the lane of the top screen, on the deck of the platform
|
||
# (refer to ui.tui.fkeys), into a normal screen action. With an unassigned slot, a
|
||
# key that the deck does not use, or a platform without a lane, the key press stops
|
||
# here.
|
||
if len(action) in (2, 3) and action[0] == "f" and action[1:].isdigit():
|
||
number = int(action[1:])
|
||
deck = fkeys.active_deck()
|
||
if deck is None:
|
||
return
|
||
if deck.is_shift_key(number):
|
||
# Whatever this resolves to, the keyboard sent this code only because Shift
|
||
# was physically down. Thus latch the shifted labels of the lane, to stop the
|
||
# flicker when Shift is released and pressed again (refer to modifier_watch).
|
||
modifier_watch.note_shift_bank_key()
|
||
top = self.top or self._base_screen()
|
||
resolved = deck.action_for(top.fkey_lane, number) if top else None
|
||
if resolved is None:
|
||
return
|
||
action = resolved
|
||
# The key-state probe comes last in each test. Thus it runs only for a key that can
|
||
# be a chord, not at each key press.
|
||
if action in _CTRL_CHORDS and _right_ctrl_down():
|
||
action = _CTRL_CHORDS[action]
|
||
elif action == "text" and data.lower() in _CTRL_LETTER_CHORDS and _right_ctrl_down():
|
||
action, data = _CTRL_LETTER_CHORDS[data.lower()], ""
|
||
# An Esc of a terminal that is still held down is a hold, not yet a press. Any other
|
||
# key goes after an Esc that is held back, so that the keys keep their order (refer
|
||
# to set_esc_probe).
|
||
watched = self._watched_esc
|
||
if watched is not None:
|
||
if action == "escape":
|
||
if watched.take():
|
||
self.invalidate()
|
||
return
|
||
else:
|
||
watched.ahead()
|
||
if action == "to_menu":
|
||
# Pop all the screens on the stack at one time (^W). The unwind is refused over a
|
||
# dialog or over work in progress (refer to request_pop_all). The paint below is
|
||
# still wanted in the two cases. After a refusal, it changes nothing, and an
|
||
# armed unwind will soon change everything.
|
||
self.request_pop_all()
|
||
self.invalidate()
|
||
return
|
||
if action == "quit":
|
||
# Ask first, from any screen. A second press while the question is open quits
|
||
# (refer to request_quit). "Unpair & quit" comes with the confirm that the menu
|
||
# declared.
|
||
self.request_quit()
|
||
self.invalidate()
|
||
return
|
||
if action == "quit_now":
|
||
# Quit immediately, with no question: the ``Quit!`` chip of the main menu, on the
|
||
# Shift half of the ``Quit?`` chip. It is a two-key press on purpose, on the only
|
||
# screen that offers it. It is never a chord that a wrong key press on the way to
|
||
# ^W can hit.
|
||
self._exit_app()
|
||
return
|
||
if action == "paste_clipboard":
|
||
# Some terminals send Ctrl-V as the literal control key, with no bracketed-paste
|
||
# sequence, so the event has no text. Read the OS clipboard here, and give the
|
||
# text to the top screen as a paste. Terminals that change Ctrl-V into a
|
||
# bracketed paste never get here: that paste goes to _typed as a multi-character
|
||
# sequence.
|
||
text = _read_clipboard()
|
||
if text:
|
||
self._dispatch("paste", text)
|
||
return
|
||
top = self.top
|
||
# The edge scroll and its snap-back get the key first (Screen.edge_scroll). A key
|
||
# that acts on a highlight that is not visible first brings the highlight back, and
|
||
# does nothing more. Home/End take the page to its ends. An arrow continues to the
|
||
# screen. If the arrow moved nothing, the next paint scrolls the page instead
|
||
# (Screen.note_highlight).
|
||
if top is not None and not top.edge_scroll(action):
|
||
top.handle(action, data)
|
||
self.invalidate()
|
||
|
||
|
||
__all__ = [
|
||
"TuiSession",
|
||
"Choice",
|
||
"Separator",
|
||
"SelectScreen",
|
||
"ReorderScreen",
|
||
"ScrollScreen",
|
||
]
|