From 194c63e4d11e6de9792a736a7b99788f1db78741 Mon Sep 17 00:00:00 2001 From: historia Date: Mon, 24 Aug 2026 00:41:52 -0400 Subject: feat: running process detection, menu gating --- README.md | 8 +- audiobook.py | 8 +- backends/__init__.py | 29 +- backends/audiocpp.py | 19 +- backends/common.py | 25 ++ backends/faster.py | 17 +- backends/qwen.py | 17 +- docs/backend-qwen.md | 2 +- hub.py | 377 ------------------ requirements.txt | 2 +- tests/test_backends.py | 92 ++++- tests/test_hub.py | 150 +++++-- tests/test_tui.py | 81 +++- tui.py | 1012 ---------------------------------------------- ui/__init__.py | 9 + ui/hub.py | 383 ++++++++++++++++++ ui/tui.py | 1039 ++++++++++++++++++++++++++++++++++++++++++++++++ 17 files changed, 1806 insertions(+), 1464 deletions(-) delete mode 100644 hub.py delete mode 100644 tui.py create mode 100644 ui/__init__.py create mode 100644 ui/hub.py create mode 100644 ui/tui.py diff --git a/README.md b/README.md index f1f9ed7..6eaf8a5 100644 --- a/README.md +++ b/README.md @@ -40,11 +40,13 @@ Run the generator with no arguments in a terminal: python audiobook.py ``` -A full-screen TUI opens and detects which TTS backends are already set up. From the menu you can: +A full-screen TUI opens and shows each backend's status in a table — **unavailable** (red, name dimmed: not installed and no server running), **installed** (orange), or **running** (green, when an external server is already accepting connections on its configured port). From the menu you can: -- **Convert books…** — process the `input/` directory with a ready backend (it reads the backend's `server.json` / `voices.json` so you pick the model and voice from menus), or +- **Convert books…** — process the `input/` directory with a ready/running backend (it reads the backend's `server.json` / `voices.json` so you pick the model and voice from menus), or - **Set up a backend…** — clone, build, and configure a backend end-to-end (audio.cpp, qwen, faster), or -- **Modify a backend…** — regenerate its config (a new `server.json`, rebuild `voices.json`, change ports/speaker). +- **Configure a backend…** — regenerate its config (a new `server.json`, rebuild `voices.json`, change ports/speaker). + +**Convert books…** and **Configure a backend…** only appear once at least one backend is installed or running. Everything the TUI does can also be scripted with flags: `python audiobook.py --backend audiocpp --model higgs --voice narrator`, or `python -m backends.audiocpp --families higgs_audio_tts --clone --build-backend cuda`. diff --git a/audiobook.py b/audiobook.py index da67ac7..174d8b4 100755 --- a/audiobook.py +++ b/audiobook.py @@ -109,7 +109,7 @@ def main() -> None: except (AttributeError, ValueError): interactive = False if interactive: - import hub + from ui import hub sys.exit(hub.run()) # Non-interactive with no args: a default conversion run (cron/etc). sys.exit(convert()) @@ -129,10 +129,10 @@ Examples: python audiobook.py --backend audiocpp --model qwen-design \\ --instructions "A warm adult female narrator with a British accent" - # Use the Qwen demo server with a custom voice + # Use the qwen-tts demo server with a custom voice python audiobook.py --backend qwen - # Use the Qwen demo server with voice cloning from reference audio + # Use the qwen-tts demo server with voice cloning from reference audio python audiobook.py --backend qwen --clone path/to/reference.wav # Use the faster-qwen3-tts server (voice cloning, configured server-side) @@ -182,7 +182,7 @@ Examples: parser.add_argument( "--backend", choices=[BACKEND_AUDIOCPP, BACKEND_QWEN, BACKEND_FASTER], default=config.BACKEND, - help=("TTS server to talk to: the Qwen3-TTS demo server (qwen), the " + help=("TTS server to talk to: the qwen-tts demo server (qwen), the " "faster-qwen3-tts OpenAI-compatible server (faster), or an " "audio.cpp audiocpp_server (audiocpp) hosting any of its TTS " "model families — Qwen3-TTS, Higgs Audio, VoxCPM2, IndexTTS2, " diff --git a/backends/__init__.py b/backends/__init__.py index 9203143..629551e 100644 --- a/backends/__init__.py +++ b/backends/__init__.py @@ -5,11 +5,12 @@ its setup wizard, its status detection, and the launch command it prints once configured. This package aggregates them into a single registry so ``audiobook.py``'s TUI hub and future tools can iterate backends without hardcoding their names: ``backends.detect_all()`` reports which are set -up, and ``backends.REGISTRY`` drives the hub's setup/modify menus. +up (and whether their server is currently running), and +``backends.REGISTRY`` drives the hub's setup/configure menus. Adding a backend: create ``backends/.py`` exposing ``detect() -> BackendStatus``, ``run_tui() -> int`` and -``modify_actions: list[ModifyAction]``, then append a ``BackendInfo`` in +``configure_actions: list[ConfigureAction]``, then append a ``BackendInfo`` in ``_build_registry`` below. ``audiobook.py`` and the hub pick it up automatically. """ @@ -25,13 +26,17 @@ class BackendStatus: INSTALLED means the backend itself is present (a cloned + built checkout, or a pip package). CONFIGURED means the supporting files are in place (a server.json / voices.json and a converter/config.py that - points at the right port). DETAILS are short status lines for the hub. - LAUNCH_HINT is the exact command the user runs to start the server. + points at the right port). RUNNING means an external server is + currently accepting connections on the configured port (probed by + ``backends.common.server_running``). DETAILS are short status lines for + the hub. LAUNCH_HINT is the exact command the user runs to start the + server. """ key: str label: str installed: bool configured: bool + running: bool = False details: List[str] = field(default_factory=list) launch_hint: str = "" @@ -42,20 +47,20 @@ class BackendStatus: @dataclass -class ModifyAction: - """A per-backend "modify" menu entry (e.g. "New server.json").""" +class ConfigureAction: + """A per-backend "configure" menu entry (e.g. "New server.json").""" label: str run: Callable[[], int] @dataclass class BackendInfo: - """One registry entry: identity, detector, setup wizard, modify menu.""" + """One registry entry: identity, detector, setup wizard, configure menu.""" key: str label: str detect: Callable[[], BackendStatus] setup_tui: Callable[[], int] - modify_actions: List[ModifyAction] = field(default_factory=list) + configure_actions: List[ConfigureAction] = field(default_factory=list) REGISTRY: List[BackendInfo] = [] @@ -73,21 +78,21 @@ def _build_registry() -> None: label="audio.cpp", detect=audiocpp.detect, setup_tui=audiocpp.run_tui, - modify_actions=audiocpp.modify_actions, + configure_actions=audiocpp.configure_actions, )) REGISTRY.append(BackendInfo( key="qwen", - label="Qwen3-TTS (demo server)", + label="qwen-tts", detect=qwen.detect, setup_tui=qwen.run_tui, - modify_actions=qwen.modify_actions, + configure_actions=qwen.configure_actions, )) REGISTRY.append(BackendInfo( key="faster", label="faster-qwen3-tts", detect=faster.detect, setup_tui=faster.run_tui, - modify_actions=faster.modify_actions, + configure_actions=faster.configure_actions, )) for info in REGISTRY: _BY_KEY[info.key] = info diff --git a/backends/audiocpp.py b/backends/audiocpp.py index b401366..57636a7 100755 --- a/backends/audiocpp.py +++ b/backends/audiocpp.py @@ -41,8 +41,8 @@ from typing import Callable, Dict, List, Optional, Set, Tuple # Allow running directly (python backends/audiocpp.py) from any cwd. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -import tui -from backends import BackendStatus, ModifyAction +from ui import tui +from backends import BackendStatus, ConfigureAction from backends import common from backends.common import ( CONFIG_PATH, @@ -1645,11 +1645,14 @@ def build_parser() -> argparse.ArgumentParser: def detect() -> BackendStatus: """Detect how far audio.cpp is set up, plus the command to start it.""" checkout = find_local_checkout() + # Probe the server first: it may be running externally even with no + # local checkout, and the status table should show that. + running = common.server_running(config.AUDIOCPP_API_URL) details: List[str] = [] launch = "" if checkout is None: return BackendStatus("audiocpp", "audio.cpp", installed=False, - configured=False, + configured=False, running=running, details=["not cloned — run setup to clone " "./audio.cpp"]) details.append(f"checkout: {checkout}") @@ -1670,13 +1673,13 @@ def detect() -> BackendStatus: else: details.append("no server.json — run setup to configure models") return BackendStatus("audiocpp", "audio.cpp", installed=built, - configured=configured, details=details, - launch_hint=launch) + configured=configured, running=running, + details=details, launch_hint=launch) -modify_actions: List[ModifyAction] = [ - ModifyAction("Reconfigure audio.cpp (models, voices, server.json)", - run_tui), +configure_actions: List[ConfigureAction] = [ + ConfigureAction("Reconfigure audio.cpp (models, voices, server.json)", + run_tui), ] diff --git a/backends/common.py b/backends/common.py index 2c6437f..d707fdc 100644 --- a/backends/common.py +++ b/backends/common.py @@ -141,6 +141,31 @@ def url_with_port(url: str, port: int) -> str: (parts.scheme or "http", f"{host}:{port}", parts.path, "", "")) +def server_running(url: str, timeout: float = 0.3) -> bool: + """True when something accepts TCP connections at URL's host:port. + + A protocol-agnostic socket connect: an HTTP TTS server that is up will + accept the connection (we do not need to speak HTTP to know it is + listening). Returns False on any parse or connection error, so a + misconfigured URL never blocks the hub — it just reports the backend + as not running. Used by each backend's ``detect()`` to set + ``BackendStatus.running``. + """ + import socket + try: + parts = urllib.parse.urlsplit(url) + host = parts.hostname or "127.0.0.1" + port = parts.port or (443 if (parts.scheme or "http") == "https" + else 80) + except ValueError: + return False + try: + with socket.create_connection((host, port), timeout=timeout): + return True + except OSError: + return False + + def update_config_value(key: str, value: str, config_path: Optional[Path] = None) -> bool: """Rewrite a ``KEY = "value"`` line in converter/config.py. diff --git a/backends/faster.py b/backends/faster.py index 4a2cc6f..71be050 100755 --- a/backends/faster.py +++ b/backends/faster.py @@ -25,8 +25,8 @@ from typing import List, Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -import tui -from backends import BackendStatus, ModifyAction +from ui import tui +from backends import BackendStatus, ConfigureAction from backends import common from backends.common import TTS_ROOT, find_wav_files, normalize_dir_arg from converter import config @@ -335,6 +335,7 @@ def detect() -> BackendStatus: cloned = _is_cloned() voices_json = _checkout() / "voices.json" configured = installed and cloned and voices_json.exists() + running = common.server_running(config.FASTER_API_URL) details: List[str] = [] details.append("pip: installed" if installed else "not installed — run setup to pip install") @@ -348,12 +349,12 @@ def detect() -> BackendStatus: f"--voices {voices_json} --port {_config_port()}") return BackendStatus("faster", "faster-qwen3-tts", installed=installed and cloned, - configured=configured, details=details, - launch_hint=launch) + configured=configured, running=running, + details=details, launch_hint=launch) def _run_voices_only_tui() -> int: - """Rebuild voices.json via the TUI (the "modify" action). + """Rebuild voices.json via the TUI (the "configure" action). Runs the same wizard but skips the pip/clone prerequisites so it goes straight to picking the .wav directory and writing voices.json. @@ -364,9 +365,9 @@ def _run_voices_only_tui() -> int: return run_tui(args) -modify_actions: List[ModifyAction] = [ - ModifyAction("Rebuild voices.json", _run_voices_only_tui), - ModifyAction("Reconfigure faster-qwen3-tts", run_tui), +configure_actions: List[ConfigureAction] = [ + ConfigureAction("Rebuild voices.json", _run_voices_only_tui), + ConfigureAction("Reconfigure faster-qwen3-tts", run_tui), ] diff --git a/backends/qwen.py b/backends/qwen.py index 48e1804..60f3bb6 100644 --- a/backends/qwen.py +++ b/backends/qwen.py @@ -22,8 +22,8 @@ from typing import List, Optional sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) -import tui -from backends import BackendStatus, ModifyAction +from ui import tui +from backends import BackendStatus, ConfigureAction from backends import common from converter import config @@ -209,6 +209,10 @@ def detect() -> BackendStatus: installed = _is_installed() custom_port = _config_port(config.QWEN_API_URL, DEFAULT_CUSTOM_PORT) clone_port = _config_port(config.CLONE_API_URL, DEFAULT_CLONE_PORT) + # Running when either server is up — CustomVoice (speaker mode) or Base + # (voice clone) each suffice for a conversion on their own. + running = (common.server_running(config.QWEN_API_URL) + or common.server_running(config.CLONE_API_URL)) details: List[str] = [] details.append("pip: installed" if installed else "not installed — run setup to pip install qwen-tts") @@ -218,13 +222,14 @@ def detect() -> BackendStatus: launch = (f"qwen-tts-demo {QWEN_CUSTOMVOICE_MODEL} --ip 127.0.0.1 " f"--port {custom_port} ; qwen-tts-demo {QWEN_BASE_MODEL} " f"--ip 127.0.0.1 --port {clone_port}") - return BackendStatus("qwen", "Qwen3-TTS (demo server)", + return BackendStatus("qwen", "qwen-tts", installed=installed, configured=installed, - details=details, launch_hint=launch) + running=running, details=details, + launch_hint=launch) -modify_actions: List[ModifyAction] = [ - ModifyAction("Reconfigure Qwen3-TTS (ports/speaker)", run_tui), +configure_actions: List[ConfigureAction] = [ + ConfigureAction("Reconfigure qwen-tts (ports/speaker)", run_tui), ] diff --git a/docs/backend-qwen.md b/docs/backend-qwen.md index b564149..0c9dab0 100644 --- a/docs/backend-qwen.md +++ b/docs/backend-qwen.md @@ -1,6 +1,6 @@ # Backend Option 2: Qwen3-TTS -The TUI sets this up: run `python audiobook.py` → **Set up a backend… → Qwen3-TTS**, or `python -m backends.qwen`. It pip-installs `qwen-tts` and configures the two ports and built-in speaker in `converter/config.py`, then prints the launch commands. Manual steps: +The TUI sets this up: run `python audiobook.py` → **Set up a backend… → qwen-tts**, or `python -m backends.qwen`. It pip-installs `qwen-tts` and configures the two ports and built-in speaker in `converter/config.py`, then prints the launch commands. Manual steps: Install qwen-tts with pip: diff --git a/hub.py b/hub.py deleted file mode 100644 index a5372e2..0000000 --- a/hub.py +++ /dev/null @@ -1,377 +0,0 @@ -#!/usr/bin/env python3 -"""The TUI main menu for the audiobook generator (run via ``audiobook.py``). - -The hub is the single entry point for the whole workflow: it detects which -backends are already set up and offers to convert the input directory with -one of them, set up a new backend, or modify/reconfigure an existing one. -Each backend's setup wizard runs in its own curses session, so the hub -collects a "command" inside its own wrapper, returns to the plain terminal, -and then dispatches — no nested curses sessions. - -Esc on the main menu quits the hub. Esc inside a sub-menu falls back to the -main menu. -""" - -import json -import sys -from pathlib import Path -from typing import Optional - -import tui -import audiobook -from backends import REGISTRY, detect_all, get -from backends import audiocpp as audiocpp_backend -from backends import faster as faster_backend -from converter import config -from converter.converter import AUDIO_FORMATS -from converter.tts import AUDIOCPP_FAMILY_QWEN3_TTS, BACKEND_AUDIOCPP, \ - BACKEND_FASTER, BACKEND_QWEN - -_GO_BACK = object() - - -def run() -> int: - """Run the hub menu loop until the user quits. Returns exit code.""" - import curses - while True: - try: - command = curses.wrapper(_hub_menu) - except tui.WizardCancelled: - return 0 - except KeyboardInterrupt: - return 130 - if command is None: - return 0 - kind = command[0] - if kind == "quit": - return 0 - if kind == "setup": - info = get(command[1]) - if info is not None: - info.setup_tui() - elif kind == "modify": - info = get(command[1]) - if info is not None and command[2] < len(info.modify_actions): - info.modify_actions[command[2]].run() - elif kind == "convert": - _run_conversion(command[1], command[2]) - - -def _hub_menu(stdscr) -> Optional[tuple]: - """Show the main menu; return a command tuple, or None to quit.""" - while True: - statuses = detect_all() - summary = ["Backend status:"] - for st in statuses: - mark = "ready" if st.ready else ( - "installed" if st.installed else "not set up") - summary.append(f" {st.label}: {mark}") - choice = tui.menu( - stdscr, "tts-audiobook-generator", - [("Convert books...", "convert"), - ("Set up a backend...", "setup"), - ("Modify a backend...", "modify"), - ("Quit", "quit")], - help_lines=summary) - if choice is None or choice == "quit": - return None - if choice == "convert": - cmd = _convert_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "setup": - cmd = _setup_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "modify": - cmd = _modify_menu(stdscr, statuses) - if cmd is not None: - return cmd - - -def _setup_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a backend to set up. Returns ("setup", key) or None to go back.""" - options = [(f"{info.label} ({_status_mark(info.key, statuses)})", - info.key) for info in REGISTRY] - choice = tui.menu(stdscr, "Set up a backend", options, - back_value=_GO_BACK, - help_lines=["Clone/build/install a backend so you can " - "convert with it."]) - if choice is _GO_BACK or choice is None: - return None - return ("setup", choice) - - -def _modify_menu(stdscr, statuses) -> Optional[tuple]: - """Pick an installed backend and one of its modify actions.""" - installed = [info for info in REGISTRY - if _status_mark(info.key, statuses) != "not set up"] - if not installed: - tui.flash(stdscr, "No backend is set up yet — use 'Set up a backend' first.") - return None - options = [(info.label, info.key) for info in installed] - key = tui.menu(stdscr, "Modify a backend", options, back_value=_GO_BACK) - if key is _GO_BACK or key is None: - return None - info = get(key) - actions = info.modify_actions - choice = tui.menu( - stdscr, f"Modify {info.label}", - [(action.label, index) for index, action in enumerate(actions)], - back_value=_GO_BACK) - if choice is _GO_BACK or choice is None: - return None - return ("modify", key, choice) - - -def _status_mark(key: str, statuses) -> str: - for st in statuses: - if st.key == key: - return "ready" if st.ready else ( - "installed" if st.installed else "not set up") - return "not set up" - - -def _convert_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a ready backend and collect per-backend run settings.""" - ready = [st for st in statuses if st.ready] - options = [(f"{st.label}", st.key) for st in ready] - if not ready: - choice = tui.menu( - stdscr, "No backend is ready", - [("Set up a backend...", "__setup__")], - help_lines=["Set up a backend (clone/build/configure) before " - "converting."]) - if choice == "__setup__": - return _setup_menu(stdscr, statuses) - return None - options.append(("Set up a backend...", "__setup__")) - key = tui.menu(stdscr, "Convert books with...", options, - back_value=_GO_BACK) - if key is _GO_BACK or key is None: - return None - if key == "__setup__": - return _setup_menu(stdscr, statuses) - if key == BACKEND_AUDIOCPP: - return _convert_audiocpp(stdscr, statuses) - if key == BACKEND_QWEN: - return _convert_qwen(stdscr) - if key == BACKEND_FASTER: - return _convert_faster(stdscr) - return None - - -def _convert_audiocpp(stdscr, statuses) -> Optional[tuple]: - """Collect audio.cpp run settings by reading ./audio.cpp/server.json.""" - checkout = audiocpp_backend.find_local_checkout() - server_json = checkout / "server.json" if checkout else None - if not server_json or not server_json.exists(): - tui.flash(stdscr, "No server.json found in the audio.cpp checkout. " - "Run 'Set up a backend' first.") - return None - try: - data = json.loads(server_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - tui.flash(stdscr, f"Could not read {server_json}.") - return None - models = data.get("models") or [] - if not models: - tui.flash(stdscr, "No model entries in server.json. Reconfigure " - "audio.cpp first.") - return None - model_options = [(f"{m.get('id')} ({m.get('family')}, {m.get('task', 'tts')})", - m.get("id")) for m in models] - model_id = tui.menu(stdscr, "Select the audio.cpp model to use", - model_options, back_value=_GO_BACK) - if model_id is _GO_BACK or model_id is None: - return None - entry = next((m for m in models if m.get("id") == model_id), {}) - family = entry.get("family") - task = entry.get("task", "tts") - - # Voice: optional for qwen3_tts (built-in speaker), required otherwise. - voice = None - voice_dir = data.get("voice_dir") - voices = _list_voices(voice_dir) if voice_dir else [] - if task == "vdes": - # Voice design: no voice, instructions required. - pass - elif family == AUDIOCPP_FAMILY_QWEN3_TTS: - # Speaker mode available; voice optional. - if voices: - opts = [("(built-in speaker)", None)] + [(v, v) for v in voices] - voice = tui.menu(stdscr, "Voice", opts, back_value=_GO_BACK) - if voice is _GO_BACK: - return None - else: - voice = None - else: - if not voices: - tui.flash(stdscr, f"This model needs a --voice but voice_dir " - f"{voice_dir} has no .wav voices. Reconfigure " - "audio.cpp or add voices.") - return None - voice = tui.menu(stdscr, "Select the voice to clone", [(v, v) for v in voices], - back_value=_GO_BACK) - if voice is _GO_BACK or voice is None: - return None - - # Instructions: required for vdes, optional otherwise. - instructions = None - if task == "vdes": - instructions = tui.line_edit( - stdscr, "Voice design instructions (required for this model)", - config.AUDIOCPP_INSTRUCTIONS, - validate=lambda s: None if s.strip() - else "Describe the voice, e.g. 'A warm female narrator'", - back_value=_GO_BACK) - if instructions is _GO_BACK: - return None - else: - instructions = tui.line_edit( - stdscr, "Style instructions (optional, blank for none)", - config.AUDIOCPP_INSTRUCTIONS, back_value=_GO_BACK) - if instructions is _GO_BACK: - return None - if not instructions.strip(): - instructions = None - - common_kw = _common_options(stdscr) - if common_kw is None: - return None - return ("convert", BACKEND_AUDIOCPP, { - "model_id": model_id, "voice": voice, "instructions": instructions, - **common_kw, - }) - - -def _convert_qwen(stdscr) -> Optional[tuple]: - """Collect qwen run settings: built-in speaker or clone a .wav.""" - mode = tui.menu( - stdscr, "Qwen3-TTS mode", - [("Custom voice (built-in speaker)", "custom"), - ("Voice clone from a .wav file", "clone")], - back_value=_GO_BACK, - help_lines=[f"Speaker: {config.SPEAKER} (change it via Modify Qwen)"]) - if mode is _GO_BACK or mode is None: - return None - clone = None - if mode == "clone": - clone = tui.line_edit( - stdscr, "Path to a reference .wav (10-15s is ideal)", - "", - validate=lambda s: None if (s and Path(s).is_file() - and s.lower().endswith(".wav")) - else "Enter the path to an existing .wav file", - back_value=_GO_BACK) - if clone is _GO_BACK: - return None - common_kw = _common_options(stdscr) - if common_kw is None: - return None - return ("convert", BACKEND_QWEN, {"clone": clone, **common_kw}) - - -def _convert_faster(stdscr) -> Optional[tuple]: - """Collect faster run settings: pick a voice from voices.json.""" - checkout = faster_backend._checkout() - voices_json = checkout / "voices.json" - if not voices_json.exists(): - tui.flash(stdscr, f"No voices.json at {voices_json}. Run 'Set up a " - "backend' for faster first.") - return None - try: - voices = json.loads(voices_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - tui.flash(stdscr, f"Could not read {voices_json}.") - return None - if not voices: - tui.flash(stdscr, "voices.json has no voices. Reconfigure faster.") - return None - default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \ - next(iter(voices)) - voice = tui.menu( - stdscr, "Select the voice to clone", - [(k, k) for k in voices], - default_index=list(voices).index(default), back_value=_GO_BACK) - if voice is _GO_BACK or voice is None: - return None - common_kw = _common_options(stdscr) - if common_kw is None: - return None - return ("convert", BACKEND_FASTER, {"voice": voice, **common_kw}) - - -def _common_options(stdscr) -> Optional[dict]: - """Collect output format, speed, single-file, chunk, debug.""" - fmt_options = [(f, f) for f in AUDIO_FORMATS] - fmt_default = AUDIO_FORMATS.index(config.AUDIO_FORMAT) \ - if config.AUDIO_FORMAT in AUDIO_FORMATS else 0 - output_format = tui.menu(stdscr, "Output format", fmt_options, - default_index=fmt_default, back_value=_GO_BACK) - if output_format is _GO_BACK or output_format is None: - return None - speed_text = tui.line_edit( - stdscr, "Playback speed (1.0 = normal)", "1.0", - validate=lambda s: None if (_is_float(s) and float(s) > 0) - else "Enter a positive number, e.g. 1.0", - back_value=_GO_BACK) - if speed_text is _GO_BACK: - return None - single_file = tui.confirm(stdscr, "Combine all chapters into one file?", - default=False, cancel_value=_GO_BACK) - if single_file is _GO_BACK: - return None - chunk = tui.confirm(stdscr, "Force client-side chunking (--chunk)?", - default=False, cancel_value=_GO_BACK) - if chunk is _GO_BACK: - return None - debug = tui.confirm(stdscr, "Debug mode (dump per-chunk audio/text)?", - default=False, cancel_value=_GO_BACK) - if debug is _GO_BACK: - return None - return { - "output_format": output_format, - "speed": float(speed_text), - "single_file": single_file, - "chunk": chunk, - "debug": debug, - } - - -def _run_conversion(backend: str, kwargs: dict) -> None: - """Run a conversion in the plain console (after the TUI returns).""" - status = next((s for s in detect_all() if s.key == backend), None) - if status is not None and not status.ready: - print(f"[WARNING] {status.label} is not fully set up.") - if status is not None and status.launch_hint: - print("[INFO] Make sure the server is running. Start it with:") - print(f" {status.launch_hint}") - audiobook.convert(backend=backend, **kwargs) - - -def _list_voices(voice_dir: str) -> list: - """Return sorted .wav stems in VOICE_DIR (best-effort).""" - try: - path = Path(voice_dir) - if not path.is_dir(): - return [] - return sorted( - (p.stem for p in path.iterdir() - if p.is_file() and p.suffix.lower() == ".wav"), - key=str.lower, - ) - except OSError: - return [] - - -def _is_float(value: str) -> bool: - try: - float(value) - return True - except ValueError: - return False - - -if __name__ == "__main__": - sys.exit(run()) diff --git a/requirements.txt b/requirements.txt index 4e23dcf..4d7e9fe 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,7 +6,7 @@ ebooklib>=0.18 # Optional dependencies beautifulsoup4>=4.11.0 # better HTML cleaning for EPUB faster-whisper>=1.0.0 # reference-audio transcription for voice cloning -# windows-curses>=2.3 # Windows only: enables the TUI (audiobook.py hub + backends.* wizards) +# windows-curses>=2.3 # Windows only: enables the TUI (the audiobook.py hub + backends.* wizards) # Audio processing # Note: ffmpeg is required to concatenate and encode the final audiobook. diff --git a/tests/test_backends.py b/tests/test_backends.py index 4017cd4..8ee1be8 100644 --- a/tests/test_backends.py +++ b/tests/test_backends.py @@ -17,8 +17,8 @@ class RegistryTests(unittest.TestCase): for info in REGISTRY: self.assertTrue(callable(info.detect), info.key) self.assertTrue(callable(info.setup_tui), info.key) - self.assertIsInstance(info.modify_actions, list) - for action in info.modify_actions: + self.assertIsInstance(info.configure_actions, list) + for action in info.configure_actions: self.assertTrue(callable(action.run)) def test_get_returns_entry_by_key(self): @@ -28,7 +28,8 @@ class RegistryTests(unittest.TestCase): class DetectAllTests(unittest.TestCase): def test_detect_all_returns_one_status_per_backend(self): - statuses = detect_all() + with patch("backends.common.server_running", return_value=False): + statuses = detect_all() self.assertEqual([s.key for s in statuses], ["audiocpp", "qwen", "faster"]) for s in statuses: @@ -37,6 +38,9 @@ class DetectAllTests(unittest.TestCase): # machine none are ready. if s.ready: self.assertTrue(s.installed and s.configured) + # running is always probed; patched False here so a dev machine + # running a real server can't flake the test. + self.assertFalse(s.running) def test_audiocpp_status_when_cloned_built_configured(self): with tempfile.TemporaryDirectory() as td: @@ -52,25 +56,55 @@ class DetectAllTests(unittest.TestCase): encoding="utf-8") from backends import audiocpp with patch.object(audiocpp, "find_local_checkout", - return_value=checkout): + return_value=checkout), \ + patch("backends.common.server_running", + return_value=False): status = audiocpp.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) self.assertTrue(status.ready) + self.assertFalse(status.running) self.assertIn("audiocpp_server", status.launch_hint) + def test_audiocpp_running_when_server_probe_succeeds(self): + from backends import audiocpp + with patch.object(audiocpp, "find_local_checkout", + return_value=None), \ + patch("backends.common.server_running", return_value=True): + status = audiocpp.detect() + # Not installed (no checkout) but an external server is up. + self.assertFalse(status.installed) + self.assertTrue(status.running) + def test_qwen_status_reflects_install(self): from backends import qwen - with patch.object(qwen, "_is_installed", return_value=True): + with patch.object(qwen, "_is_installed", return_value=True), \ + patch("backends.common.server_running", return_value=False): status = qwen.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) + self.assertFalse(status.running) self.assertIn("qwen-tts-demo", status.launch_hint) - with patch.object(qwen, "_is_installed", return_value=False): + with patch.object(qwen, "_is_installed", return_value=False), \ + patch("backends.common.server_running", return_value=False): status = qwen.detect() self.assertFalse(status.installed) self.assertFalse(status.configured) + def test_qwen_running_when_either_port_is_up(self): + # Either the CustomVoice port or the Base port counts as running. + from backends import qwen + with patch.object(qwen, "_is_installed", return_value=False), \ + patch("backends.common.server_running", + side_effect=[True, False]): + status = qwen.detect() + self.assertTrue(status.running) + with patch.object(qwen, "_is_installed", return_value=False), \ + patch("backends.common.server_running", + side_effect=[False, True]): + status = qwen.detect() + self.assertTrue(status.running) + def test_faster_status_reflects_install_clone_voices(self): from backends import faster with tempfile.TemporaryDirectory() as td: @@ -80,12 +114,56 @@ class DetectAllTests(unittest.TestCase): (checkout / "voices.json").write_text('{"default":{}}', encoding="utf-8") with patch.object(faster, "_is_installed", return_value=True), \ - patch.object(faster, "_checkout", return_value=checkout): + patch.object(faster, "_checkout", + return_value=checkout), \ + patch("backends.common.server_running", + return_value=False): status = faster.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) + self.assertFalse(status.running) self.assertIn("openai_server.py", status.launch_hint) + def test_faster_running_when_server_probe_succeeds(self): + from backends import faster + with patch.object(faster, "_is_installed", return_value=False), \ + patch.object(faster, "_is_cloned", return_value=False), \ + patch("backends.common.server_running", return_value=True): + status = faster.detect() + self.assertTrue(status.running) + + +class ServerRunningTests(unittest.TestCase): + """backends.common.server_running: TCP probe against a real socket.""" + + def test_true_for_open_port(self): + import socket + from backends import common + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.bind(("127.0.0.1", 0)) + server.listen(1) + host, port = server.getsockname() + url = f"http://127.0.0.1:{port}" + try: + self.assertTrue(common.server_running(url)) + finally: + server.close() + + def test_false_for_closed_port(self): + from backends import common + # Pick an unused port by opening + closing a socket, then probe it. + import socket + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.bind(("127.0.0.1", 0)) + _, port = s.getsockname() + s.close() + self.assertFalse(common.server_running(f"http://127.0.0.1:{port}")) + + def test_false_for_invalid_url(self): + from backends import common + self.assertFalse(common.server_running("not a url")) + self.assertFalse(common.server_running("")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_hub.py b/tests/test_hub.py index 5f6d992..ce9af43 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -1,15 +1,14 @@ -"""Tests for the TUI hub (hub.py) menu and helpers. +"""Tests for the TUI hub (ui/hub.py) menu and helpers. -The hub drives the same curses widgets as tui.py, so these tests reuse the -fake curses/screen from test_tui to run the menu without a terminal. +The hub drives the same curses widgets as ui/tui.py, so these tests reuse +the fake curses/screen from test_tui to run the menu without a terminal. """ import unittest from pathlib import Path from unittest.mock import patch -import hub -import tui +from ui import hub, tui from tests.test_tui import FakeCurses, FakeScreen @@ -37,13 +36,21 @@ class HubHelperTests(unittest.TestCase): def test_status_mark(self): from backends import BackendStatus - ready = BackendStatus("k", "l", installed=True, configured=True) - half = BackendStatus("k", "l", installed=True, configured=False) + running = BackendStatus("k", "l", installed=True, configured=True, + running=True) + installed = BackendStatus("k", "l", installed=True, + configured=False) none = BackendStatus("k", "l", installed=False, configured=False) - self.assertEqual(hub._status_mark("k", [ready]), "ready") - self.assertEqual(hub._status_mark("k", [half]), "installed") - self.assertEqual(hub._status_mark("k", [none]), "not set up") - self.assertEqual(hub._status_mark("missing", []), "not set up") + # running beats installed (a server is up even if not configured); + # only a backend that is neither installed nor running is dimmed. + self.assertEqual(hub._status_mark(running), + ("running", "ok", "body")) + self.assertEqual(hub._status_mark(installed), + ("installed", "warn", "body")) + self.assertEqual(hub._status_mark(none), + ("unavailable", "err", "dim")) + self.assertEqual(hub._status_mark(None), + ("unavailable", "err", "dim")) class HubMenuTests(unittest.TestCase): @@ -58,31 +65,126 @@ class HubMenuTests(unittest.TestCase): self.addCleanup(self._patcher.stop) self.addCleanup(tui._THEME.clear) - def test_quit_returns_none(self): - # Main menu: move to "Quit" (4th option, index 3) and press Enter. - screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, - FakeCurses.KEY_DOWN, 10]) + def _none_status(self, key="k", label="l"): + from backends import BackendStatus + return BackendStatus(key, label, installed=False, configured=False) + + def test_quit_returns_none_when_no_backend(self): + # No backends installed/running: menu is [Set up, Quit]. Quit is the + # 2nd option (Down once) then Enter. + screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, 10]) with patch.object(hub, "detect_all", return_value=[]): result = hub._hub_menu(screen) self.assertIsNone(result) - def test_convert_with_no_ready_backend_offers_setup(self): - # Convert -> "Set up a backend..." is the only entry -> Enter selects - # it -> setup menu lists 3 backends; press Esc to go back -> convert - # returns None -> main menu loops. Then quit (Down x3 + Enter). + def test_menu_has_only_setup_and_quit_without_backends(self): + # Capture the options handed to tui.menu: with nothing installed or + # running, Convert/Configure must be absent. + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return "quit" + + screen = FakeScreen() + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=[]): + hub._hub_menu(screen) + labels = [label for label, _ in captured["options"]] + self.assertEqual(labels, ["Set up a backend...", "Quit"]) + + def test_menu_has_all_four_when_one_installed(self): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + captured["rows"] = kwargs.get("table_rows") + return "quit" + + screen = FakeScreen() + st = self._none_status("qwen", "qwen-tts") + st.installed = True + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=[st]): + hub._hub_menu(screen) + labels = [label for label, _ in captured["options"]] + self.assertEqual( + labels, + ["Convert books...", "Set up a backend...", + "Configure a backend...", "Quit"]) + # The status table is passed through, one row per backend. + self.assertEqual(captured["rows"], + [("qwen-tts", "installed", "warn", "body")]) + + def test_table_dims_name_when_not_installed_and_not_running(self): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["rows"] = kwargs.get("table_rows") + return "quit" + + screen = FakeScreen() + dead = self._none_status("audiocpp", "audio.cpp") + external = self._none_status("qwen", "qwen-tts") + external.running = True + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", + return_value=[dead, external]): + hub._hub_menu(screen) + # Unusable backend: dim name. Running-but-not-installed stays bright. + self.assertEqual( + captured["rows"], + [("audio.cpp", "unavailable", "err", "dim"), + ("qwen-tts", "running", "ok", "body")]) + + def test_menu_has_all_four_when_one_running_only(self): + # Running but not installed (an external server) still unlocks the + # Convert/Configure entries. + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return "quit" + + screen = FakeScreen() + st = self._none_status("qwen", "qwen-tts") + st.running = True + with patch.object(hub.tui, "menu", fake_menu), \ + patch.object(hub, "detect_all", return_value=[st]): + hub._hub_menu(screen) + labels = [label for label, _ in captured["options"]] + self.assertEqual( + labels, + ["Convert books...", "Set up a backend...", + "Configure a backend...", "Quit"]) + + def test_convert_with_no_available_backend_offers_setup(self): + # One installed-but-not-ready backend → Convert is offered. The + # convert menu lists no available backend, so only "Set up a + # backend..." is shown; Enter selects it → setup menu lists 3 + # backends; Esc goes back → convert returns None → main menu loops. + # Then quit: main menu now has 4 options, Quit is the 4th (Down x3). from backends import BackendInfo, BackendStatus - none = BackendStatus("k", "l", installed=False, configured=False) + none = BackendStatus("k", "l", installed=True, configured=False) infos = [BackendInfo("audiocpp", "audio.cpp", lambda: none, lambda: 0), - BackendInfo("qwen", "Qwen", lambda: none, lambda: 0), + BackendInfo("qwen", "qwen-tts", lambda: none, lambda: 0), BackendInfo("faster", "faster", lambda: none, lambda: 0)] - with patch.object(hub, "detect_all", return_value=[none, none, none]), \ + # installed=True so the main menu shows Convert; but ready/running + # is False so the convert menu's available list is empty. + statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=False), + BackendStatus("qwen", "qwen-tts", installed=True, + configured=False), + BackendStatus("faster", "faster", installed=True, + configured=False)] + with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub, "REGISTRY", infos): # Convert(Enter), setup-entry(Enter), Esc on setup menu, # back at main menu -> Down x3 -> Enter (Quit). screen = FakeScreen(keys=[10, 10, 27, - FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, - FakeCurses.KEY_DOWN, 10]) + FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, + FakeCurses.KEY_DOWN, 10]) result = hub._hub_menu(screen) self.assertIsNone(result) diff --git a/tests/test_tui.py b/tests/test_tui.py index ba6f99f..58d8273 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -14,7 +14,7 @@ import unittest from pathlib import Path from unittest.mock import patch -import tui +from ui import tui class FakeCurses: @@ -222,6 +222,85 @@ class MenuTests(TuiTestCase): tui.menu(screen, "Pick", self.OPTIONS, back_value=marker) +class MenuTableTests(TuiTestCase): + """The optional status table: aligned columns and colored statuses.""" + + ROWS = [("audio.cpp", "not installed", "err"), + ("qwen-tts", "installed", "warn"), + ("faster-qwen3-tts", "running", "ok")] + + def test_name_column_left_aligned_at_margin(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + x0, _ = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + for name, _, _ in self.ROWS: + x = next(x for _, x, text, _ in screen.strings + if text.rstrip() == name) + self.assertEqual(x, margin, name) + + def test_status_column_aligned_at_one_fixed_offset(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + x0, _ = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + name_w = max(len(name) for name, _, _ in self.ROWS) + expected_x = margin + name_w # the " status" segment starts here + for _, status, _ in self.ROWS: + x = next(x for _, x, text, _ in screen.strings + if text.strip() == status) + self.assertEqual(x, expected_x, status) + + def test_status_text_uses_the_theme_kind_color(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + want = {"err": tui._THEME["err"], "warn": tui._THEME["warn"], + "ok": tui._THEME["ok"]} + for _, status, kind in self.ROWS: + attr = next(a for _, _, text, a in screen.strings + if text.strip() == status) + self.assertEqual(attr, want[kind], status) + + def test_optional_name_kind_colors_the_name_column(self): + # 4-element rows: the 4th value is a theme kind for the name. + rows = [("gone", "unavailable", "err", "dim"), + ("here", "running", "ok", "body")] + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], table_rows=rows) + drawn = {text.rstrip(): attr for _, _, text, attr in screen.strings} + self.assertEqual(drawn["gone"], tui._THEME["dim"]) + self.assertEqual(drawn["here"], tui._THEME["body"]) + + def test_three_element_rows_default_to_body_names(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + for name, _, _ in self.ROWS: + attr = next(a for _, _, text, a in screen.strings + if text.rstrip() == name) + self.assertEqual(attr, tui._THEME["body"], name) + + def test_table_title_is_dim_and_left_aligned(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + x0, _ = self.dialog_box(screen) + margin = x0 + 1 + tui.Frame.LIST_MARGIN + x, attr = next((x, a) for _, x, text, a in screen.strings + if text == "Backend status") + self.assertEqual(x, margin) + self.assertEqual(attr, tui._THEME["dim"]) + + def test_table_does_not_paint_over_the_border(self): + screen = FakeScreen(keys=[10]) + tui.menu(screen, "Hub", [("Quit", "quit")], + table_title="Backend status", table_rows=self.ROWS) + self.assert_inside_border(screen) + + class ConfirmTests(TuiTestCase): def test_tab_switches_and_enter_activates(self): screen = FakeScreen(keys=[9, 10]) diff --git a/tui.py b/tui.py deleted file mode 100644 index c0c9222..0000000 --- a/tui.py +++ /dev/null @@ -1,1012 +0,0 @@ -#!/usr/bin/env python3 -"""Colorful DOS-style curses TUI widgets for the interactive tools. - -Every screen is a dialog centered on a black desktop, like an old DOS -TUI: a yellow title, colored status messages (green/yellow/red), a -bright cyan cursor bar, and Yes/No buttons you switch with Tab for -every yes/no question. Instructions and prompts are centered while -lists (directory contents, menu options, checkbox trees) are -left-justified for readability; the black background matches the -terminal default, so the full-screen repaints curses performs while -resizing a dialog never flash. One screen per decision: a directory -browser, an expandable checkbox tree, a single-line text editor, a -single-choice menu, and a yes/no confirm. There is no framework — -every widget is a function that runs its own key loop on a curses -window and returns the chosen value. - -Common key bindings: - - Up/Down (or k/j) move the cursor - Enter accept (the highlighted button or row) - Tab or Left/Right switch Yes/No buttons (confirmations) - Esc abort the whole wizard (raises WizardCancelled); - a widget passed back_value returns that sentinel - instead, so the caller can fall back a screen - (confirm() historically names this cancel_value) - -On screens without typed text (menus, confirm, tree, browser) 'q' also -aborts — even when a back_value is set, so Esc means "back" while 'q' -still means "quit". Inside text editors 'q' is an ordinary character. -When the terminal has no color support the theme degrades to -bold/reverse/dim. -""" - -import contextlib -import os -import textwrap -from pathlib import Path -from typing import Callable, List, Optional, Sequence, Tuple - -# Make Esc register quickly instead of pausing for an escape sequence. -os.environ.setdefault("ESCDELAY", "25") - - -class WizardCancelled(Exception): - """Raised when the user presses Esc to abort the wizard.""" - - -@contextlib.contextmanager -def suspend(scr): - """Temporarily leave curses to run plain-console code. - - Long-running steps that stream output to the terminal (cloning a - repository, building, pip-installing, transcribing) cannot share the - curses screen, so the wizard suspends curses for the duration of the - step and repaints the current screen afterward. ``scr`` is the curses - window returned to the wrapper callback. - """ - import curses - try: - curses.endwin() - except curses.error: - pass - try: - yield - finally: - try: - scr.redrawwin() - scr.refresh() - except Exception: - pass - try: - curses.curs_set(0) - except curses.error: - pass - - -def flash(scr, text: str, kind: str = "warn") -> None: - """Show a one-line notice until any key is pressed, then return. - - Used by the hub for "not set up yet"-style messages. KIND is a theme - key (warn/err/ok/info). Esc dismisses the notice (it does not abort). - """ - frame = Frame(scr, "Notice", "Press any key to continue Esc = back") - frame.mark(text, frame.theme.get(kind, frame.theme["body"])) - frame.cursor = None - frame.draw() - try: - key = scr.getch() - except KeyboardInterrupt: - raise WizardCancelled() from None - if key == 3: # Ctrl-C still aborts - raise WizardCancelled() - - -# Esc and 'q' both abort on screens without typed text ('q' is an -# ordinary character inside text editors). -_CANCEL_KEYS = (27, ord("q")) - - -# --------------------------------------------------------------------------- -# Theme -# --------------------------------------------------------------------------- - -_THEME: dict = {} - - -def _ensure_theme(curses) -> dict: - """Build (once) the attribute table for the classic DOS look. - - White text on a black desktop, a cyan border, yellow titles and - warnings, green success/check marks, red errors, a black-on-cyan - cursor bar and a black-on-green selected button. Black matches the - terminal's default background, so the clear-screen repaints curses - performs when a dialog changes size never flash. Without colors, - everything falls back to bold/reverse/dim attributes. - """ - if _THEME: - return _THEME - theme = { - "desktop": 0, - "border": curses.A_BOLD, - "title": curses.A_BOLD, - "body": 0, - "dim": curses.A_DIM, - "ok": curses.A_BOLD, - "warn": curses.A_BOLD, - "err": curses.A_BOLD | curses.A_REVERSE, - "info": curses.A_DIM, - "input": curses.A_BOLD, - "bar": curses.A_REVERSE, - "btn_on": curses.A_REVERSE | curses.A_BOLD, - "btn_off": curses.A_DIM, - "check": curses.A_BOLD, - "accent": curses.A_BOLD, - } - if curses.has_colors(): - try: - curses.start_color() - black = curses.COLOR_BLACK - pairs = { - "desktop": (curses.COLOR_WHITE, black), - "border": (curses.COLOR_CYAN, black), - "title": (curses.COLOR_YELLOW, black), - "ok": (curses.COLOR_GREEN, black), - "warn": (curses.COLOR_YELLOW, black), - "err": (curses.COLOR_RED, black), - "info": (curses.COLOR_WHITE, black), - "input": (curses.COLOR_WHITE, black), - "bar": (curses.COLOR_BLACK, curses.COLOR_CYAN), - "btn_on": (curses.COLOR_BLACK, curses.COLOR_GREEN), - "check": (curses.COLOR_GREEN, black), - "accent": (curses.COLOR_CYAN, black), - } - for number, (name, (fg, bg)) in enumerate(pairs.items(), 1): - curses.init_pair(number, fg, bg) - theme[name] = curses.color_pair(number) - theme["dim"] = curses.A_DIM | theme["desktop"] - theme["body"] = theme["desktop"] - theme["btn_off"] = curses.A_DIM | theme["desktop"] - for name in ("title", "ok", "warn", "err", "check", "accent", - "input"): - theme[name] |= curses.A_BOLD - theme["info"] = curses.A_DIM | theme["info"] - except curses.error: - pass - _THEME.clear() - _THEME.update(theme) - return _THEME - - -# --------------------------------------------------------------------------- -# Shared drawing helpers -# --------------------------------------------------------------------------- - -def _addstr(scr, y: int, x: int, text: str, attr: int = 0) -> None: - """addstr that ignores out-of-bounds and terminal-capability errors.""" - try: - scr.addstr(y, x, text, attr) - except Exception: - pass - - -def _addch(scr, y: int, x: int, ch, attr: int = 0) -> None: - """addch that ignores out-of-bounds and terminal-capability errors.""" - try: - scr.addch(y, x, ch, attr) - except Exception: - pass - - -def _hline(scr, y: int, x: int, n: int, attr: int = 0) -> None: - """hline of ACS_HLINE that ignores terminal-capability errors.""" - import curses - try: - scr.hline(y, x, curses.ACS_HLINE, n, attr) - except Exception: - pass - - -def _fit(text: str, width: int) -> str: - """Truncate TEXT to WIDTH columns, appending '~' when cut.""" - if width < 1: - return "" - if len(text) <= width: - return text - return text[: max(0, width - 1)] + "~" - - -class Frame: - """A dialog centered on the black desktop, DOS style. - - Widgets append logical rows with mark()/mark_segments() and call - draw() after every state change. Rows are centered by default; - list rows pass align="left" to start at a fixed margin from the - left border. Rows that are not selectable (help text, the current - directory, blank lines) are skipped by the cursor. The selected - row is drawn as a full-width bright bar. Below the rows sit the - optional Yes/No buttons, a colored one-line status, and a dim - footer. - """ - - MIN_HEIGHT = 8 - MIN_WIDTH = 30 - # Columns between the left border and align="left" rows. - LIST_MARGIN = 2 - - def __init__(self, scr, title: str, footer: str): - import curses - self.curses = curses - self.scr = scr - self.title = title - self.footer = footer - self.theme = _ensure_theme(curses) - self.rows: List[dict] = [] - self.cursor: Optional[int] = None # logical row index - self.status: Optional[Tuple[str, str]] = None # (text, kind) - self.buttons: Optional[Tuple[Sequence[str], int]] = None - self.scroll = 0 - self.page_size = 1 - try: - curses.curs_set(0) - except curses.error: - pass - try: - scr.bkgd(" ", self.theme["desktop"]) - except curses.error: - pass - - # -- content --------------------------------------------------------- - - def mark(self, text: str, attr: Optional[int] = None, indent: int = 0, - selectable: bool = False, align: str = "center") -> None: - """Append a body row (wrapped when longer than the box). - - ALIGN is "center" (the default, for instructions and prompts) - or "left" (for lists), which starts the row at a fixed margin - from the left border. - """ - if attr is None: - attr = self.theme["body"] - self.rows.append({"text": text, "segments": None, "attr": attr, - "indent": indent, "selectable": selectable, - "align": align}) - - def mark_segments(self, segments: Sequence[Tuple[str, int]], - indent: int = 0, selectable: bool = False, - align: str = "center") -> None: - """Append a row of (text, attr) segments (truncated, not wrapped).""" - self.rows.append({"text": None, "segments": list(segments), - "attr": 0, "indent": indent, - "selectable": selectable, "align": align}) - - def selectable(self) -> List[int]: - """Logical indices of the selectable rows, in order.""" - return [index for index, row in enumerate(self.rows) - if row["selectable"]] - - # -- drawing --------------------------------------------------------- - - def _row_width(self, row: dict) -> int: - """Logical width of a row, including its indent.""" - if row["segments"] is not None: - return sum(len(text) for text, _ in row["segments"]) \ - + 2 * row["indent"] - return len(row["text"]) + 2 * row["indent"] - - def _measure(self, width: int) -> int: - """Dialog width: widest row plus frame, capped to the screen.""" - longest = max(len(self.title) + 4, len(self.footer) + 4, 40) - for row in self.rows: - longest = max(longest, self._row_width(row) + 4) - if self.status: - longest = max(longest, len(self.status[0]) + 6) - if self.buttons: - labels, _ = self.buttons - longest = max(longest, - sum(len(label) + 6 for label in labels) + 4) - return min(longest + 4, width - 2) - - def _flatten(self, usable: int) -> List[Tuple[int, dict, Optional[str]]]: - """Wrap text rows into physical (logical index, row, piece) lines.""" - flat: List[Tuple[int, dict, Optional[str]]] = [] - for index, row in enumerate(self.rows): - if row["segments"] is not None: - flat.append((index, row, None)) - continue - wrap_width = usable - if row["align"] == "left": - # Leave room for the list margin, the indent and the - # right border so a wrapped line is never re-truncated. - wrap_width = usable - 1 - 2 * row["indent"] - pieces = textwrap.wrap(row["text"], max(10, wrap_width)) or [""] - for piece in pieces: - flat.append((index, row, piece)) - return flat - - def _geometry(self, height: int, width: int, dialog_w: int, - flat: List[Tuple[int, dict, Optional[str]]] - ) -> Tuple[int, int, int, int]: - """Place the dialog and scroll the cursor row into view. - - Returns (y0, x0, dialog_h, visible); also refreshes - self.scroll and self.page_size. - """ - chrome = 7 if self.buttons else 6 # title/gap/status/footer/borders - dialog_h = min(max(self.MIN_HEIGHT, len(flat) + chrome), height) - visible = max(1, dialog_h - chrome) - self.page_size = max(1, visible) - if self.cursor is not None: - positions = [i for i, (logical, _, _) in enumerate(flat) - if logical == self.cursor] - if positions: - first, last = positions[0], positions[-1] - if first < self.scroll: - self.scroll = first - elif last >= self.scroll + visible: - self.scroll = last - visible + 1 - self.scroll = max(0, min(self.scroll, max(0, len(flat) - visible))) - y0 = max(0, (height - dialog_h) // 2) - x0 = max(0, (width - dialog_w) // 2) - return y0, x0, dialog_h, visible - - def draw(self) -> None: - scr = self.scr - scr.erase() - height, width = scr.getmaxyx() - if height < self.MIN_HEIGHT or width < self.MIN_WIDTH: - msg = "Terminal too small" - _addstr(scr, height // 2, max(0, (width - len(msg)) // 2), - msg, self.curses.A_BOLD) - scr.refresh() - return - dialog_w = self._measure(width) - flat = self._flatten(dialog_w - 4) - y0, x0, dialog_h, visible = self._geometry(height, width, - dialog_w, flat) - self._draw_frame(y0, x0, dialog_h, dialog_w, len(flat), visible) - self._draw_rows(y0, x0, dialog_w, flat, visible) - self._draw_buttons(y0, x0, dialog_h, dialog_w) - self._draw_status_footer(y0, x0, dialog_h, dialog_w) - scr.refresh() - - def _draw_frame(self, y0: int, x0: int, dialog_h: int, dialog_w: int, - total_lines: int, visible: int) -> None: - curses, theme = self.curses, self.theme - scr = self.scr - border = theme["border"] - _addch(scr, y0, x0, curses.ACS_ULCORNER, border) - _addch(scr, y0, x0 + dialog_w - 1, curses.ACS_URCORNER, border) - _addch(scr, y0 + dialog_h - 1, x0, curses.ACS_LLCORNER, border) - _addch(scr, y0 + dialog_h - 1, x0 + dialog_w - 1, - curses.ACS_LRCORNER, border) - _hline(scr, y0, x0 + 1, dialog_w - 2, border) - _hline(scr, y0 + dialog_h - 1, x0 + 1, dialog_w - 2, border) - for y in range(y0 + 1, y0 + dialog_h - 1): - _addch(scr, y, x0, curses.ACS_VLINE, border) - _addch(scr, y, x0 + dialog_w - 1, curses.ACS_VLINE, border) - - inner_x = x0 + 1 - inner_w = dialog_w - 2 - title = _fit(f" {self.title} ", inner_w) - _addstr(scr, y0 + 1, inner_x + max(0, (inner_w - len(title)) // 2), - title, theme["title"]) - if total_lines > visible: - indicator = f" {self.scroll + 1}/{total_lines} " - _addstr(scr, y0, max(x0 + 1, x0 + dialog_w - 1 - len(indicator)), - indicator, theme["dim"]) - - def _draw_rows(self, y0: int, x0: int, dialog_w: int, - flat: List[Tuple[int, dict, Optional[str]]], - visible: int) -> None: - theme = self.theme - scr = self.scr - inner_x = x0 + 1 - inner_w = dialog_w - 2 - for line in range(self.scroll, min(len(flat), self.scroll + visible)): - logical, row, piece = flat[line] - y = y0 + 2 + (line - self.scroll) - selected = logical == self.cursor and row["selectable"] - if selected: - _addstr(scr, y, inner_x, " " * inner_w, theme["bar"]) - if row["segments"] is not None: - self._draw_segments_row(y, row, inner_x, inner_w, selected) - else: - self._draw_text_row(y, row, piece, inner_x, inner_w, - selected) - - def _draw_segments_row(self, y: int, row: dict, inner_x: int, - inner_w: int, selected: bool) -> None: - scr, theme = self.scr, self.theme - total = sum(len(text) for text, _ in row["segments"]) - if row["align"] == "left": - x = inner_x + self.LIST_MARGIN + 2 * row["indent"] - else: - x = inner_x + max(0, (inner_w - total) // 2) \ - + 2 * row["indent"] - # Never paint over the right border column. - room = max(0, inner_x + inner_w - 1 - x) - for text, attr in row["segments"]: - text = _fit(text, room) - if not text: - break - _addstr(scr, y, x, text, theme["bar"] if selected else attr) - x += len(text) - room -= len(text) - - def _draw_text_row(self, y: int, row: dict, piece: Optional[str], - inner_x: int, inner_w: int, selected: bool) -> None: - scr, theme = self.scr, self.theme - text = " " * row["indent"] + piece - if row["align"] == "left": - x = inner_x + self.LIST_MARGIN - limit = inner_w - 1 - self.LIST_MARGIN - 2 * row["indent"] - else: - x = inner_x + max(0, (inner_w - len(text)) // 2) - limit = inner_w - text = _fit(text, limit) - attr = theme["bar"] if selected else row["attr"] - _addstr(scr, y, x, text, attr) - - def _draw_buttons(self, y0: int, x0: int, dialog_h: int, - dialog_w: int) -> None: - if not self.buttons: - return - theme = self.theme - scr = self.scr - inner_x = x0 + 1 - inner_w = dialog_w - 2 - labels, selected = self.buttons - rendered = [f"[ {label} ]" for label in labels] - total = sum(len(r) for r in rendered) + 3 * (len(rendered) - 1) - x = inner_x + max(0, (inner_w - total) // 2) - y = y0 + dialog_h - 4 - for index, text in enumerate(rendered): - if index: - x += 3 - _addstr(scr, y, x, text, - theme["btn_on"] if index == selected - else theme["btn_off"]) - x += len(text) - - def _draw_status_footer(self, y0: int, x0: int, dialog_h: int, - dialog_w: int) -> None: - theme = self.theme - scr = self.scr - inner_x = x0 + 1 - inner_w = dialog_w - 2 - if self.status: - text, kind = self.status - attr = theme.get(kind, theme["body"]) - text = _fit(f" {text} ", inner_w) - _addstr(scr, y0 + dialog_h - 3, - inner_x + max(0, (inner_w - len(text)) // 2), - text, attr) - footer = _fit(self.footer, inner_w) - _addstr(scr, y0 + dialog_h - 2, - inner_x + max(0, (inner_w - len(footer)) // 2), - footer, theme["dim"]) - - # -- key helpers ------------------------------------------------------ - - def motion(self, key: int, cursor: int, count: int, - wrap: bool = False) -> Optional[int]: - """New cursor index for a motion KEY, or None when it moves nothing. - - Up/Down (or k/j) move one row, wrapping around at the ends when - WRAP is set (menus and trees) and clamping otherwise (the - browser); Home/End jump to the first/last row; PageUp/PageDown - move self.page_size rows. COUNT is the number of rows. - """ - curses = self.curses - if key in (curses.KEY_UP, ord("k")): - if wrap and cursor <= 0: - return count - 1 - return max(0, cursor - 1) - if key in (curses.KEY_DOWN, ord("j")): - if wrap and cursor >= count - 1: - return 0 - return min(count - 1, cursor + 1) - if key == curses.KEY_HOME: - return 0 - if key == curses.KEY_END: - return count - 1 - if key == curses.KEY_PPAGE: - return max(0, cursor - self.page_size) - if key == curses.KEY_NPAGE: - return min(count - 1, cursor + self.page_size) - return None - - def get_key(self, cancel_keys: Sequence[int] = (27,)) -> int: - """Read one key; cancel keys and Ctrl-C raise WizardCancelled.""" - try: - key = self.scr.getch() - except KeyboardInterrupt: - raise WizardCancelled() from None - if key == 3: # Ctrl-C - raise WizardCancelled() - if key in cancel_keys: - raise WizardCancelled() - return key - - def flash(self, text: str, kind: str = "err") -> None: - """Show TEXT on the status line until any key is pressed.""" - self.status = (text, kind) - self.draw() - try: - key = self.scr.getch() - if key == 3: # Ctrl-C still aborts - raise WizardCancelled() - except KeyboardInterrupt: - raise WizardCancelled() from None - self.status = None - - def edit_status(self, prompt: str = "") -> Optional[str]: - """Edit a line of text on the status line. - - Returns the edited string on Enter, or None when the user backs - out with Esc (the caller decides what that means). - """ - curses = self.curses - text = "" - while True: - self.status = (f"{prompt}{text}_", "input") - self.draw() - try: - key = self.scr.getch() - except KeyboardInterrupt: - raise WizardCancelled() from None - if key == 27: - return None - if key == 3: # Ctrl-C - raise WizardCancelled() - if key in (10, 13): - return text - if key in (curses.KEY_BACKSPACE, 8, 127): - text = text[:-1] - elif 32 <= key < 127: - text += chr(key) - - -# --------------------------------------------------------------------------- -# Widget: yes/no confirm with buttons -# --------------------------------------------------------------------------- - -def confirm(scr, question: str, default: bool = False, - body: Optional[Sequence[str]] = None, - cancel_value: object = None): - """Ask a yes/no QUESTION with centered Yes/No buttons. - - The QUESTION is the dialog title (shown exactly once); optional - BODY lines sit centered above the buttons. Tab or the arrow keys - switch the buttons, Enter activates the highlighted one (the - DEFAULT button starts highlighted, drawn bright against the dim - other one), and y/n answer directly. Esc (or 'q') aborts the - wizard — unless CANCEL_VALUE is given (not None), in which case it - is returned instead, so the caller can fall back to a previous - screen rather than aborting the whole wizard. - """ - frame = Frame(scr, question, - "Tab/arrows = switch Enter = confirm y/n Esc = cancel") - index = 0 if default else 1 - while True: - frame.rows = [] - for line in body or []: - frame.mark(line) - frame.cursor = None - frame.buttons = (["Yes", "No"], index) - frame.draw() - curses = frame.curses - key = frame.get_key(cancel_keys=()) - if key in _CANCEL_KEYS: - if cancel_value is not None: - return cancel_value - raise WizardCancelled() - if key in (9, curses.KEY_LEFT, curses.KEY_RIGHT, curses.KEY_UP, - curses.KEY_DOWN, curses.KEY_BTAB, ord("h"), ord("l")): - index = 1 - index - elif key in (ord("y"), ord("Y")): - return True - elif key in (ord("n"), ord("N")): - return False - elif key in (10, 13): - return index == 0 - - -# --------------------------------------------------------------------------- -# Widget: single-choice menu -# --------------------------------------------------------------------------- - -def menu(scr, title: str, options: Sequence[tuple], default_index: int = 0, - help_lines: Optional[Sequence[str]] = None, - back_value: object = None): - """Show OPTIONS as (label, value) pairs; return the chosen value. - - The cursor starts on DEFAULT_INDEX; Enter returns the highlighted - option's value. Options are left-justified like a DOS list; - HELP_LINES are dim, centered explanatory lines shown above them. - Esc (or 'q') aborts the wizard unless BACK_VALUE is given (not None), - in which case Esc returns it so the caller can fall back a screen. - """ - if not options: - raise ValueError("menu() needs at least one option") - frame = Frame(scr, title, - "Up/Down = move Enter = select Esc = cancel") - cursor = max(0, min(default_index, len(options) - 1)) - while True: - frame.rows = [] - for line in help_lines or []: - frame.mark(line, frame.theme["dim"]) - if help_lines: - frame.mark("") - base = len(frame.rows) - for label, _ in options: - frame.mark(label, selectable=True, align="left") - frame.cursor = base + cursor - frame.draw() - key = frame.get_key(cancel_keys=()) - if key == 27 and back_value is not None: - return back_value - if key in _CANCEL_KEYS: - raise WizardCancelled() - moved = frame.motion(key, cursor, len(options), wrap=True) - if moved is not None: - cursor = moved - elif key in (10, 13): - return options[cursor][1] - - -# --------------------------------------------------------------------------- -# Widget: single-line text editor -# --------------------------------------------------------------------------- - -def line_edit(scr, title: str, default: str, - validate: Optional[Callable[[str], Optional[str]]] = None, - help_lines: Optional[Sequence[str]] = None, - back_value: object = None) -> str: - """Edit one line of text, pre-filled with DEFAULT; Enter accepts. - - HELP_LINES are dim explanatory lines shown above the input. - VALIDATE receives the entered string and returns an error message - or None; Enter on an invalid value shows the message in red and - keeps editing. Esc aborts the wizard ('q' is an ordinary - character here) unless BACK_VALUE is given (not None), in which case - Esc returns it so the caller can fall back a screen. - """ - frame = Frame(scr, title, - "type to edit Backspace = erase Enter = accept " - "Esc = cancel") - text = default - error = None - while True: - frame.rows = [] - for line in help_lines or []: - frame.mark(line, frame.theme["dim"]) - frame.mark("") - frame.mark(f"{text}_", frame.theme["input"]) - frame.cursor = None - frame.status = (error, "err") if error else None - frame.draw() - curses = frame.curses - key = frame.get_key(cancel_keys=()) # handle Esc manually below - if key == 27 and back_value is not None: - return back_value - if key == 27: - raise WizardCancelled() - if key in (10, 13): - if validate is None: - return text - error = validate(text) - if error is None: - return text - continue - if key in (curses.KEY_BACKSPACE, 8, 127): - text = text[:-1] - elif key == 21: # Ctrl-U: clear the line - text = "" - elif 32 <= key < 127: - text += chr(key) - - -# --------------------------------------------------------------------------- -# Widget: directory browser -# --------------------------------------------------------------------------- - -def _list_dirs(path: Path) -> List[Path]: - """Return the subdirectories of PATH, sorted, dot-dirs excluded.""" - try: - entries = [child for child in path.iterdir() - if child.is_dir() and not child.name.startswith(".")] - except OSError: - return [] - return sorted(entries, key=lambda child: child.name.lower()) - - -def browse_directory(scr, title: str, - validate: Optional[Callable[[Path], Optional[str]]] = None, - start: Optional[Path] = None, - info: Optional[Callable[[Path], - Optional[Tuple[str, str]]]] = None, - preview: Optional[Callable[[Path], - Optional[Tuple[str, str]]]] = None, - help_lines: Optional[Sequence[str]] = None, - auto_select: Optional[Callable[ - [Path], Optional[Path]]] = None, - back_value: object = None - ) -> Path: - """Pick a directory DOS-browser style. - - The listing starts with a bright '[ Use this directory ]' row (the - cursor starts there; Enter accepts the directory being listed), a - dim '..' for the parent, and one row per subdirectory. List rows - are left-justified; instructions and the current path stay - centered. Enter or Right on a highlighted subdirectory opens it, - Left/Backspace goes to the parent, 'e' types a path directly, and - Home/End/PageUp/PageDown navigate long listings. Coming back out - of a directory highlights the directory you came from. - - VALIDATE receives the listed directory and returns an error message - or None; Enter on an invalid directory is refused with that message. - INFO(directory) returns a (text, kind) status shown under the - listed directory's path — kind is "ok" (green), "warn" (yellow), - "err" (red), "info" (dim) or "input". PREVIEW(directory) returns - one for the highlighted subdirectory, shown on the status line. - AUTO_SELECT receives a highlighted subdirectory when the user - opens it (Enter, Right or 'l') and may return a Path to accept - immediately — as if '[ Use this directory ]' had been pressed on - it — instead of descending; returning None keeps browsing. This - lets a subdirectory that already looks like the target (e.g. an - 'audio.cpp' checkout containing 'model_specs/') be picked in one - keystroke. Esc (or 'q') aborts the wizard unless BACK_VALUE is given - (not None), in which case Esc returns it so the caller can fall back - a screen. - """ - footer = ("Up/Down = move Enter = open/use Left = parent " - "e = type path Esc = cancel") - frame = Frame(scr, title, footer) - current = Path(start) if start is not None else Path.cwd() - try: - current = current.resolve() - except OSError: - current = Path.cwd() - sel = 0 - highlight: Optional[Path] = None - - def validation_error() -> Optional[str]: - if validate is None: - return None - try: - return validate(current) - except OSError: - return "Cannot read this directory" - - def call(callback, path: Path) -> Optional[Tuple[str, str]]: - if callback is None: - return None - try: - return callback(path) - except OSError: - return None - - while True: - entries = _list_dirs(current) - has_parent = current.parent != current - offset = 1 + (1 if has_parent else 0) - frame.rows = [] - for line in help_lines or []: - frame.mark(line, frame.theme["dim"]) - frame.mark(f"Directory: {current}", frame.theme["accent"]) - current_info = call(info, current) - if current_info: - frame.mark(current_info[0], - frame.theme.get(current_info[1], frame.theme["body"])) - frame.mark("") - frame.mark("[ Use this directory ]", frame.theme["ok"], - selectable=True, align="left") - if has_parent: - frame.mark("..", frame.theme["dim"], selectable=True, - align="left") - for entry in entries: - frame.mark(f"{entry.name}/", selectable=True, align="left") - selectable = frame.selectable() - if highlight is not None: - sel = 0 - for index, entry in enumerate(entries): - if entry == highlight: - sel = offset + index - break - highlight = None - sel = max(0, min(sel, len(selectable) - 1)) - frame.cursor = selectable[sel] if selectable else None - - if sel == 0: - frame.status = ("Enter = use this directory", "info") - elif has_parent and sel == 1: - frame.status = ("Enter = open the parent directory", "info") - else: - entry = entries[sel - offset] - frame.status = call(preview, entry) \ - or (f"Enter = open {entry.name}/", "info") - frame.draw() - curses = frame.curses - key = frame.get_key(cancel_keys=()) - if key == 27 and back_value is not None: - return back_value - if key in _CANCEL_KEYS: - raise WizardCancelled() - moved = frame.motion(key, sel, len(selectable)) - if moved is not None: - sel = moved - elif key in (10, 13, curses.KEY_RIGHT, ord("l")): - if sel == 0: - error = validation_error() - if error is None: - return current - frame.flash(f"{error} (keep browsing)", "err") - elif has_parent and sel == 1: - highlight = current - current = current.parent - else: - entry = entries[sel - offset] - if auto_select is not None: - picked = auto_select(entry) - if picked is not None: - return picked - current = entry - sel = 0 - elif key in (curses.KEY_LEFT, ord("h"), ord("u"), - curses.KEY_BACKSPACE, 8, 127): - if has_parent: - highlight = current - current = current.parent - elif key == ord("e"): - result = frame.edit_status(prompt="path: ") - if result: - candidate = Path(os.path.expanduser(result)) - if not candidate.is_absolute(): - candidate = current / candidate - try: - candidate = candidate.resolve() - except OSError: - pass - if candidate.is_dir(): - current = candidate - sel = 0 - else: - frame.flash(f"Not a directory: {candidate}", "err") - - -# --------------------------------------------------------------------------- -# Widget: expandable checkbox tree -# --------------------------------------------------------------------------- - -def checkbox_tree(scr, title: str, families: List[dict], - footer: Optional[str] = None, - expand_all: bool = False, - back_value: object = None) -> List[Tuple[int, str]]: - """Pick model families and packages from an expandable tree. - - FAMILIES is a list of dicts (one per family) shaped like:: - - { - "label": "Qwen3-TTS (qwen3_tts)", - "detail": "tts, cloning, design", - "options": [ - {"key": "Base-GGUF", "label": "base", "recommended": True}, - {"key": "VoiceDesign-GGUF", "label": "voicedesign", - "recommended": False}, - ], - } - - Space on a family row checks its recommended option (or clears every - option when one is already checked); Space on an option row toggles - that option. Tab/Right expands or collapses the family under the - cursor. Enter returns the flat list of (family_index, option_key) - pairs for every checked option, in tree order; at least one checked - option is required. Nothing is checked by default, and with - EXPAND_ALL every family starts expanded. A "[recommended]" tag is - shown only when a - family has more than one option — a single option needs no tag. - Family and option rows are left-justified like a DOS list. Esc (or - 'q') aborts the wizard unless BACK_VALUE is given (not None), in - which case Esc returns it so the caller can fall back a screen. - """ - if not families: - raise ValueError("checkbox_tree() needs at least one family") - footer = footer or ("Up/Down = move Tab/Right = expand Space = check " - "Enter = accept Esc = cancel") - frame = Frame(scr, title, footer) - expanded = {index for index in range(len(families))} if expand_all else set() - checked = set() # (family_index, option_key) - - expanded.add(0) - - def family_checked(index: int) -> bool: - return any(pair[0] == index for pair in checked) - - def accept() -> List[Tuple[int, str]]: - return [(index, option["key"]) - for index, family in enumerate(families) - for option in family["options"] - if (index, option["key"]) in checked] - - def visible_nodes() -> List[tuple]: - nodes: List[tuple] = [] # ("family", i) or ("option", i, key) - for index, family in enumerate(families): - nodes.append(("family", index)) - if index in expanded: - for option in family["options"]: - nodes.append(("option", index, option["key"])) - return nodes - - cursor = 0 - while True: - nodes = visible_nodes() - cursor = max(0, min(cursor, len(nodes) - 1)) - frame.rows = [] - for node in nodes: - if node[0] == "family": - index = node[1] - family = families[index] - on = family_checked(index) - mark = "x" if on else " " - arrow = "-" if index in expanded else "+" - frame.mark_segments( - [(f"[{mark}] ", - frame.theme["check"] if on else frame.theme["dim"]), - (f"{arrow} {family['label']}", - frame.theme["accent"] if on else frame.theme["body"])], - selectable=True, align="left") - else: - _, index, option_key = node - option = next(opt for opt in families[index]["options"] - if opt["key"] == option_key) - is_on = (index, option_key) in checked - mark = "x" if is_on else " " - segments = [(f"[{mark}] ", - frame.theme["check"] if is_on - else frame.theme["dim"]), - (option["label"], frame.theme["body"])] - if option.get("recommended") \ - and len(families[index]["options"]) > 1: - segments.append((" [recommended]", frame.theme["warn"])) - frame.mark_segments(segments, indent=2, selectable=True, - align="left") - frame.cursor = cursor - node = nodes[cursor] - frame.status = (families[node[1]].get("detail", ""), "info") - frame.draw() - curses = frame.curses - key = frame.get_key(cancel_keys=()) - if key == 27 and back_value is not None: - return back_value - if key in _CANCEL_KEYS: - raise WizardCancelled() - moved = frame.motion(key, cursor, len(nodes), wrap=True) - if moved is not None: - cursor = moved - elif key in (9, curses.KEY_RIGHT, ord("l")) and node[0] == "family": - index = node[1] - if index in expanded: - expanded.discard(index) - else: - expanded.add(index) - elif key == curses.KEY_LEFT and node[0] == "family": - expanded.discard(node[1]) - elif key == ord(" "): - if node[0] == "family": - index = node[1] - options = families[index]["options"] - if family_checked(index): - for option in options: - checked.discard((index, option["key"])) - else: - for option in options: - if option.get("recommended"): - checked.add((index, option["key"])) - break - else: - if options: - checked.add((index, options[0]["key"])) - expanded.add(index) - else: - _, index, option_key = node - if (index, option_key) in checked: - checked.discard((index, option_key)) - else: - checked.add((index, option_key)) - elif key in (10, 13): # Enter: accept the checked selection - selection = accept() - if selection: - return selection - frame.flash("Check at least one model package (Space)", "err") diff --git a/ui/__init__.py b/ui/__init__.py new file mode 100644 index 0000000..4986a10 --- /dev/null +++ b/ui/__init__.py @@ -0,0 +1,9 @@ +"""The TUI frontend for the audiobook generator. + +``ui.tui`` is the DOS-style curses widget library, and ``ui.hub`` is the +main menu the user sees when running ``audiobook.py`` with no arguments +(set up/configure backends, convert the input directory). It is the only +entry point for the interactive workflow; everything else under +``backends/`` and ``converter/`` is library code driven by it or by the +``audiobook.py`` CLI flags. +""" diff --git a/ui/hub.py b/ui/hub.py new file mode 100644 index 0000000..3115f54 --- /dev/null +++ b/ui/hub.py @@ -0,0 +1,383 @@ +#!/usr/bin/env python3 +"""The TUI main menu for the audiobook generator (run via ``audiobook.py``). + +The hub is the single entry point for the whole workflow: it detects which +backends are already set up and offers to convert the input directory with +one of them, set up a new backend, or configure an existing one. +Each backend's setup wizard runs in its own curses session, so the hub +collects a "command" inside its own wrapper, returns to the plain terminal, +and then dispatches — no nested curses sessions. + +Esc on the main menu quits the hub. Esc inside a sub-menu falls back to the +main menu. +""" + +import json +from pathlib import Path +from typing import Optional, Tuple + +from ui import tui +import audiobook +from backends import REGISTRY, BackendStatus, detect_all, get +from backends import audiocpp as audiocpp_backend +from backends import faster as faster_backend +from converter import config +from converter.converter import AUDIO_FORMATS +from converter.tts import AUDIOCPP_FAMILY_QWEN3_TTS, BACKEND_AUDIOCPP, \ + BACKEND_FASTER, BACKEND_QWEN + +_GO_BACK = object() + + +def run() -> int: + """Run the hub menu loop until the user quits. Returns exit code.""" + import curses + while True: + try: + command = curses.wrapper(_hub_menu) + except tui.WizardCancelled: + return 0 + except KeyboardInterrupt: + return 130 + if command is None: + return 0 + kind = command[0] + if kind == "quit": + return 0 + if kind == "setup": + info = get(command[1]) + if info is not None: + info.setup_tui() + elif kind == "configure": + info = get(command[1]) + if info is not None and command[2] < len(info.configure_actions): + info.configure_actions[command[2]].run() + elif kind == "convert": + _run_conversion(command[1], command[2]) + + +def _hub_menu(stdscr) -> Optional[tuple]: + """Show the main menu; return a command tuple, or None to quit.""" + while True: + statuses = detect_all() + options = [("Set up a backend...", "setup")] + if any(st.installed or st.running for st in statuses): + options.insert(0, ("Convert books...", "convert")) + options.append(("Configure a backend...", "configure")) + options.append(("Quit", "quit")) + rows = [(st.label, *_status_mark(st)) for st in statuses] + choice = tui.menu( + stdscr, "tts-audiobook-generator", options, + table_title="Backend status", table_rows=rows) + if choice is None or choice == "quit": + return None + if choice == "convert": + cmd = _convert_menu(stdscr, statuses) + if cmd is not None: + return cmd + elif choice == "setup": + cmd = _setup_menu(stdscr, statuses) + if cmd is not None: + return cmd + elif choice == "configure": + cmd = _configure_menu(stdscr, statuses) + if cmd is not None: + return cmd + + +def _setup_menu(stdscr, statuses) -> Optional[tuple]: + """Pick a backend to set up. Returns ("setup", key) or None to go back.""" + by_key = {st.key: st for st in statuses} + options = [(f"{info.label} ({_status_mark(by_key.get(info.key))[0]})", + info.key) for info in REGISTRY] + choice = tui.menu(stdscr, "Set up a backend", options, + back_value=_GO_BACK, + help_lines=["Clone/build/install a backend so you can " + "convert with it."]) + if choice is _GO_BACK or choice is None: + return None + return ("setup", choice) + + +def _configure_menu(stdscr, statuses) -> Optional[tuple]: + """Pick an installed backend and one of its configure actions.""" + by_key = {st.key: st for st in statuses} + installed = [info for info in REGISTRY + if by_key.get(info.key) is not None + and by_key[info.key].installed] + if not installed: + tui.flash(stdscr, "No backend is installed yet — use 'Set up a " + "backend' first.") + return None + options = [(info.label, info.key) for info in installed] + key = tui.menu(stdscr, "Configure a backend", options, back_value=_GO_BACK) + if key is _GO_BACK or key is None: + return None + info = get(key) + actions = info.configure_actions + choice = tui.menu( + stdscr, f"Configure {info.label}", + [(action.label, index) for index, action in enumerate(actions)], + back_value=_GO_BACK) + if choice is _GO_BACK or choice is None: + return None + return ("configure", key, choice) + + +def _status_mark(status: Optional[BackendStatus]) -> Tuple[str, str, str]: + """Map a backend's state to (status_text, status_kind, name_kind). + + 'running' (green/ok) takes priority — an external server is already up; + otherwise 'installed' (orange/warn) when the backend is present on disk, + or 'unavailable' (red/err). A backend that is neither installed nor + running is unusable, so its name is dimmed (NAME_KIND). + CURSES has no true orange, so the theme's yellow 'warn' is used; it + renders amber/orange on most terminals. + """ + if status is not None and status.running: + return ("running", "ok", "body") + if status is not None and status.installed: + return ("installed", "warn", "body") + return ("unavailable", "err", "dim") + + +def _convert_menu(stdscr, statuses) -> Optional[tuple]: + """Pick an available backend and collect per-backend run settings.""" + available = [st for st in statuses if st.ready or st.running] + options = [(st.label, st.key) for st in available] + if not available: + choice = tui.menu( + stdscr, "No backend is available", + [("Set up a backend...", "__setup__")], + help_lines=["Set up a backend (clone/build/configure) before " + "converting."]) + if choice == "__setup__": + return _setup_menu(stdscr, statuses) + return None + options.append(("Set up a backend...", "__setup__")) + key = tui.menu(stdscr, "Convert books with...", options, + back_value=_GO_BACK) + if key is _GO_BACK or key is None: + return None + if key == "__setup__": + return _setup_menu(stdscr, statuses) + if key == BACKEND_AUDIOCPP: + return _convert_audiocpp(stdscr, statuses) + if key == BACKEND_QWEN: + return _convert_qwen(stdscr) + if key == BACKEND_FASTER: + return _convert_faster(stdscr) + return None + + +def _convert_audiocpp(stdscr, statuses) -> Optional[tuple]: + """Collect audio.cpp run settings by reading ./audio.cpp/server.json.""" + checkout = audiocpp_backend.find_local_checkout() + server_json = checkout / "server.json" if checkout else None + if not server_json or not server_json.exists(): + tui.flash(stdscr, "No server.json found in the audio.cpp checkout. " + "Run 'Set up a backend' first.") + return None + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + tui.flash(stdscr, f"Could not read {server_json}.") + return None + models = data.get("models") or [] + if not models: + tui.flash(stdscr, "No model entries in server.json. Reconfigure " + "audio.cpp first.") + return None + model_options = [(f"{m.get('id')} ({m.get('family')}, {m.get('task', 'tts')})", + m.get("id")) for m in models] + model_id = tui.menu(stdscr, "Select the audio.cpp model to use", + model_options, back_value=_GO_BACK) + if model_id is _GO_BACK or model_id is None: + return None + entry = next((m for m in models if m.get("id") == model_id), {}) + family = entry.get("family") + task = entry.get("task", "tts") + + # Voice: optional for qwen3_tts (built-in speaker), required otherwise. + voice = None + voice_dir = data.get("voice_dir") + voices = _list_voices(voice_dir) if voice_dir else [] + if task == "vdes": + # Voice design: no voice, instructions required. + pass + elif family == AUDIOCPP_FAMILY_QWEN3_TTS: + # Speaker mode available; voice optional. + if voices: + opts = [("(built-in speaker)", None)] + [(v, v) for v in voices] + voice = tui.menu(stdscr, "Voice", opts, back_value=_GO_BACK) + if voice is _GO_BACK: + return None + else: + voice = None + else: + if not voices: + tui.flash(stdscr, f"This model needs a --voice but voice_dir " + f"{voice_dir} has no .wav voices. Reconfigure " + "audio.cpp or add voices.") + return None + voice = tui.menu(stdscr, "Select the voice to clone", [(v, v) for v in voices], + back_value=_GO_BACK) + if voice is _GO_BACK or voice is None: + return None + + # Instructions: required for vdes, optional otherwise. + instructions = None + if task == "vdes": + instructions = tui.line_edit( + stdscr, "Voice design instructions (required for this model)", + config.AUDIOCPP_INSTRUCTIONS, + validate=lambda s: None if s.strip() + else "Describe the voice, e.g. 'A warm female narrator'", + back_value=_GO_BACK) + if instructions is _GO_BACK: + return None + else: + instructions = tui.line_edit( + stdscr, "Style instructions (optional, blank for none)", + config.AUDIOCPP_INSTRUCTIONS, back_value=_GO_BACK) + if instructions is _GO_BACK: + return None + if not instructions.strip(): + instructions = None + + common_kw = _common_options(stdscr) + if common_kw is None: + return None + return ("convert", BACKEND_AUDIOCPP, { + "model_id": model_id, "voice": voice, "instructions": instructions, + **common_kw, + }) + + +def _convert_qwen(stdscr) -> Optional[tuple]: + """Collect qwen run settings: built-in speaker or clone a .wav.""" + mode = tui.menu( + stdscr, "qwen-tts mode", + [("Custom voice (built-in speaker)", "custom"), + ("Voice clone from a .wav file", "clone")], + back_value=_GO_BACK, + help_lines=[f"Speaker: {config.SPEAKER} (change it via Configure " + "qwen-tts)"]) + if mode is _GO_BACK or mode is None: + return None + clone = None + if mode == "clone": + clone = tui.line_edit( + stdscr, "Path to a reference .wav (10-15s is ideal)", + "", + validate=lambda s: None if (s and Path(s).is_file() + and s.lower().endswith(".wav")) + else "Enter the path to an existing .wav file", + back_value=_GO_BACK) + if clone is _GO_BACK: + return None + common_kw = _common_options(stdscr) + if common_kw is None: + return None + return ("convert", BACKEND_QWEN, {"clone": clone, **common_kw}) + + +def _convert_faster(stdscr) -> Optional[tuple]: + """Collect faster run settings: pick a voice from voices.json.""" + checkout = faster_backend._checkout() + voices_json = checkout / "voices.json" + if not voices_json.exists(): + tui.flash(stdscr, f"No voices.json at {voices_json}. Run 'Set up a " + "backend' for faster first.") + return None + try: + voices = json.loads(voices_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + tui.flash(stdscr, f"Could not read {voices_json}.") + return None + if not voices: + tui.flash(stdscr, "voices.json has no voices. Reconfigure faster.") + return None + default = config.FASTER_VOICE if config.FASTER_VOICE in voices else \ + next(iter(voices)) + voice = tui.menu( + stdscr, "Select the voice to clone", + [(k, k) for k in voices], + default_index=list(voices).index(default), back_value=_GO_BACK) + if voice is _GO_BACK or voice is None: + return None + common_kw = _common_options(stdscr) + if common_kw is None: + return None + return ("convert", BACKEND_FASTER, {"voice": voice, **common_kw}) + + +def _common_options(stdscr) -> Optional[dict]: + """Collect output format, speed, single-file, chunk, debug.""" + fmt_options = [(f, f) for f in AUDIO_FORMATS] + fmt_default = AUDIO_FORMATS.index(config.AUDIO_FORMAT) \ + if config.AUDIO_FORMAT in AUDIO_FORMATS else 0 + output_format = tui.menu(stdscr, "Output format", fmt_options, + default_index=fmt_default, back_value=_GO_BACK) + if output_format is _GO_BACK or output_format is None: + return None + speed_text = tui.line_edit( + stdscr, "Playback speed (1.0 = normal)", "1.0", + validate=lambda s: None if (_is_float(s) and float(s) > 0) + else "Enter a positive number, e.g. 1.0", + back_value=_GO_BACK) + if speed_text is _GO_BACK: + return None + single_file = tui.confirm(stdscr, "Combine all chapters into one file?", + default=False, cancel_value=_GO_BACK) + if single_file is _GO_BACK: + return None + chunk = tui.confirm(stdscr, "Force client-side chunking (--chunk)?", + default=False, cancel_value=_GO_BACK) + if chunk is _GO_BACK: + return None + debug = tui.confirm(stdscr, "Debug mode (dump per-chunk audio/text)?", + default=False, cancel_value=_GO_BACK) + if debug is _GO_BACK: + return None + return { + "output_format": output_format, + "speed": float(speed_text), + "single_file": single_file, + "chunk": chunk, + "debug": debug, + } + + +def _run_conversion(backend: str, kwargs: dict) -> None: + """Run a conversion in the plain console (after the TUI returns).""" + status = next((s for s in detect_all() if s.key == backend), None) + if status is not None and not status.ready and not status.running: + print(f"[WARNING] {status.label} is not fully set up.") + if status is not None and not status.running and status.launch_hint: + print("[INFO] Make sure the server is running. Start it with:") + print(f" {status.launch_hint}") + audiobook.convert(backend=backend, **kwargs) + + +def _list_voices(voice_dir: str) -> list: + """Return sorted .wav stems in VOICE_DIR (best-effort).""" + try: + path = Path(voice_dir) + if not path.is_dir(): + return [] + return sorted( + (p.stem for p in path.iterdir() + if p.is_file() and p.suffix.lower() == ".wav"), + key=str.lower, + ) + except OSError: + return [] + + +def _is_float(value: str) -> bool: + try: + float(value) + return True + except ValueError: + return False diff --git a/ui/tui.py b/ui/tui.py new file mode 100644 index 0000000..c73fb09 --- /dev/null +++ b/ui/tui.py @@ -0,0 +1,1039 @@ +#!/usr/bin/env python3 +"""Colorful DOS-style curses TUI widgets for the interactive tools. + +Every screen is a dialog centered on a black desktop, like an old DOS +TUI: a yellow title, colored status messages (green/yellow/red), a +bright cyan cursor bar, and Yes/No buttons you switch with Tab for +every yes/no question. Instructions and prompts are centered while +lists (directory contents, menu options, checkbox trees) are +left-justified for readability; the black background matches the +terminal default, so the full-screen repaints curses performs while +resizing a dialog never flash. One screen per decision: a directory +browser, an expandable checkbox tree, a single-line text editor, a +single-choice menu, and a yes/no confirm. There is no framework — +every widget is a function that runs its own key loop on a curses +window and returns the chosen value. + +Common key bindings: + + Up/Down (or k/j) move the cursor + Enter accept (the highlighted button or row) + Tab or Left/Right switch Yes/No buttons (confirmations) + Esc abort the whole wizard (raises WizardCancelled); + a widget passed back_value returns that sentinel + instead, so the caller can fall back a screen + (confirm() historically names this cancel_value) + +On screens without typed text (menus, confirm, tree, browser) 'q' also +aborts — even when a back_value is set, so Esc means "back" while 'q' +still means "quit". Inside text editors 'q' is an ordinary character. +When the terminal has no color support the theme degrades to +bold/reverse/dim. +""" + +import contextlib +import os +import textwrap +from pathlib import Path +from typing import Callable, List, Optional, Sequence, Tuple + +# Make Esc register quickly instead of pausing for an escape sequence. +os.environ.setdefault("ESCDELAY", "25") + + +class WizardCancelled(Exception): + """Raised when the user presses Esc to abort the wizard.""" + + +@contextlib.contextmanager +def suspend(scr): + """Temporarily leave curses to run plain-console code. + + Long-running steps that stream output to the terminal (cloning a + repository, building, pip-installing, transcribing) cannot share the + curses screen, so the wizard suspends curses for the duration of the + step and repaints the current screen afterward. ``scr`` is the curses + window returned to the wrapper callback. + """ + import curses + try: + curses.endwin() + except curses.error: + pass + try: + yield + finally: + try: + scr.redrawwin() + scr.refresh() + except Exception: + pass + try: + curses.curs_set(0) + except curses.error: + pass + + +def flash(scr, text: str, kind: str = "warn") -> None: + """Show a one-line notice until any key is pressed, then return. + + Used by the hub for "not set up yet"-style messages. KIND is a theme + key (warn/err/ok/info). Esc dismisses the notice (it does not abort). + """ + frame = Frame(scr, "Notice", "Press any key to continue Esc = back") + frame.mark(text, frame.theme.get(kind, frame.theme["body"])) + frame.cursor = None + frame.draw() + try: + key = scr.getch() + except KeyboardInterrupt: + raise WizardCancelled() from None + if key == 3: # Ctrl-C still aborts + raise WizardCancelled() + + +# Esc and 'q' both abort on screens without typed text ('q' is an +# ordinary character inside text editors). +_CANCEL_KEYS = (27, ord("q")) + + +# --------------------------------------------------------------------------- +# Theme +# --------------------------------------------------------------------------- + +_THEME: dict = {} + + +def _ensure_theme(curses) -> dict: + """Build (once) the attribute table for the classic DOS look. + + White text on a black desktop, a cyan border, yellow titles and + warnings, green success/check marks, red errors, a black-on-cyan + cursor bar and a black-on-green selected button. Black matches the + terminal's default background, so the clear-screen repaints curses + performs when a dialog changes size never flash. Without colors, + everything falls back to bold/reverse/dim attributes. + """ + if _THEME: + return _THEME + theme = { + "desktop": 0, + "border": curses.A_BOLD, + "title": curses.A_BOLD, + "body": 0, + "dim": curses.A_DIM, + "ok": curses.A_BOLD, + "warn": curses.A_BOLD, + "err": curses.A_BOLD | curses.A_REVERSE, + "info": curses.A_DIM, + "input": curses.A_BOLD, + "bar": curses.A_REVERSE, + "btn_on": curses.A_REVERSE | curses.A_BOLD, + "btn_off": curses.A_DIM, + "check": curses.A_BOLD, + "accent": curses.A_BOLD, + } + if curses.has_colors(): + try: + curses.start_color() + black = curses.COLOR_BLACK + pairs = { + "desktop": (curses.COLOR_WHITE, black), + "border": (curses.COLOR_CYAN, black), + "title": (curses.COLOR_YELLOW, black), + "ok": (curses.COLOR_GREEN, black), + "warn": (curses.COLOR_YELLOW, black), + "err": (curses.COLOR_RED, black), + "info": (curses.COLOR_WHITE, black), + "input": (curses.COLOR_WHITE, black), + "bar": (curses.COLOR_BLACK, curses.COLOR_CYAN), + "btn_on": (curses.COLOR_BLACK, curses.COLOR_GREEN), + "check": (curses.COLOR_GREEN, black), + "accent": (curses.COLOR_CYAN, black), + } + for number, (name, (fg, bg)) in enumerate(pairs.items(), 1): + curses.init_pair(number, fg, bg) + theme[name] = curses.color_pair(number) + theme["dim"] = curses.A_DIM | theme["desktop"] + theme["body"] = theme["desktop"] + theme["btn_off"] = curses.A_DIM | theme["desktop"] + for name in ("title", "ok", "warn", "err", "check", "accent", + "input"): + theme[name] |= curses.A_BOLD + theme["info"] = curses.A_DIM | theme["info"] + except curses.error: + pass + _THEME.clear() + _THEME.update(theme) + return _THEME + + +# --------------------------------------------------------------------------- +# Shared drawing helpers +# --------------------------------------------------------------------------- + +def _addstr(scr, y: int, x: int, text: str, attr: int = 0) -> None: + """addstr that ignores out-of-bounds and terminal-capability errors.""" + try: + scr.addstr(y, x, text, attr) + except Exception: + pass + + +def _addch(scr, y: int, x: int, ch, attr: int = 0) -> None: + """addch that ignores out-of-bounds and terminal-capability errors.""" + try: + scr.addch(y, x, ch, attr) + except Exception: + pass + + +def _hline(scr, y: int, x: int, n: int, attr: int = 0) -> None: + """hline of ACS_HLINE that ignores terminal-capability errors.""" + import curses + try: + scr.hline(y, x, curses.ACS_HLINE, n, attr) + except Exception: + pass + + +def _fit(text: str, width: int) -> str: + """Truncate TEXT to WIDTH columns, appending '~' when cut.""" + if width < 1: + return "" + if len(text) <= width: + return text + return text[: max(0, width - 1)] + "~" + + +class Frame: + """A dialog centered on the black desktop, DOS style. + + Widgets append logical rows with mark()/mark_segments() and call + draw() after every state change. Rows are centered by default; + list rows pass align="left" to start at a fixed margin from the + left border. Rows that are not selectable (help text, the current + directory, blank lines) are skipped by the cursor. The selected + row is drawn as a full-width bright bar. Below the rows sit the + optional Yes/No buttons, a colored one-line status, and a dim + footer. + """ + + MIN_HEIGHT = 8 + MIN_WIDTH = 30 + # Columns between the left border and align="left" rows. + LIST_MARGIN = 2 + + def __init__(self, scr, title: str, footer: str): + import curses + self.curses = curses + self.scr = scr + self.title = title + self.footer = footer + self.theme = _ensure_theme(curses) + self.rows: List[dict] = [] + self.cursor: Optional[int] = None # logical row index + self.status: Optional[Tuple[str, str]] = None # (text, kind) + self.buttons: Optional[Tuple[Sequence[str], int]] = None + self.scroll = 0 + self.page_size = 1 + try: + curses.curs_set(0) + except curses.error: + pass + try: + scr.bkgd(" ", self.theme["desktop"]) + except curses.error: + pass + + # -- content --------------------------------------------------------- + + def mark(self, text: str, attr: Optional[int] = None, indent: int = 0, + selectable: bool = False, align: str = "center") -> None: + """Append a body row (wrapped when longer than the box). + + ALIGN is "center" (the default, for instructions and prompts) + or "left" (for lists), which starts the row at a fixed margin + from the left border. + """ + if attr is None: + attr = self.theme["body"] + self.rows.append({"text": text, "segments": None, "attr": attr, + "indent": indent, "selectable": selectable, + "align": align}) + + def mark_segments(self, segments: Sequence[Tuple[str, int]], + indent: int = 0, selectable: bool = False, + align: str = "center") -> None: + """Append a row of (text, attr) segments (truncated, not wrapped).""" + self.rows.append({"text": None, "segments": list(segments), + "attr": 0, "indent": indent, + "selectable": selectable, "align": align}) + + def selectable(self) -> List[int]: + """Logical indices of the selectable rows, in order.""" + return [index for index, row in enumerate(self.rows) + if row["selectable"]] + + # -- drawing --------------------------------------------------------- + + def _row_width(self, row: dict) -> int: + """Logical width of a row, including its indent.""" + if row["segments"] is not None: + return sum(len(text) for text, _ in row["segments"]) \ + + 2 * row["indent"] + return len(row["text"]) + 2 * row["indent"] + + def _measure(self, width: int) -> int: + """Dialog width: widest row plus frame, capped to the screen.""" + longest = max(len(self.title) + 4, len(self.footer) + 4, 40) + for row in self.rows: + longest = max(longest, self._row_width(row) + 4) + if self.status: + longest = max(longest, len(self.status[0]) + 6) + if self.buttons: + labels, _ = self.buttons + longest = max(longest, + sum(len(label) + 6 for label in labels) + 4) + return min(longest + 4, width - 2) + + def _flatten(self, usable: int) -> List[Tuple[int, dict, Optional[str]]]: + """Wrap text rows into physical (logical index, row, piece) lines.""" + flat: List[Tuple[int, dict, Optional[str]]] = [] + for index, row in enumerate(self.rows): + if row["segments"] is not None: + flat.append((index, row, None)) + continue + wrap_width = usable + if row["align"] == "left": + # Leave room for the list margin, the indent and the + # right border so a wrapped line is never re-truncated. + wrap_width = usable - 1 - 2 * row["indent"] + pieces = textwrap.wrap(row["text"], max(10, wrap_width)) or [""] + for piece in pieces: + flat.append((index, row, piece)) + return flat + + def _geometry(self, height: int, width: int, dialog_w: int, + flat: List[Tuple[int, dict, Optional[str]]] + ) -> Tuple[int, int, int, int]: + """Place the dialog and scroll the cursor row into view. + + Returns (y0, x0, dialog_h, visible); also refreshes + self.scroll and self.page_size. + """ + chrome = 7 if self.buttons else 6 # title/gap/status/footer/borders + dialog_h = min(max(self.MIN_HEIGHT, len(flat) + chrome), height) + visible = max(1, dialog_h - chrome) + self.page_size = max(1, visible) + if self.cursor is not None: + positions = [i for i, (logical, _, _) in enumerate(flat) + if logical == self.cursor] + if positions: + first, last = positions[0], positions[-1] + if first < self.scroll: + self.scroll = first + elif last >= self.scroll + visible: + self.scroll = last - visible + 1 + self.scroll = max(0, min(self.scroll, max(0, len(flat) - visible))) + y0 = max(0, (height - dialog_h) // 2) + x0 = max(0, (width - dialog_w) // 2) + return y0, x0, dialog_h, visible + + def draw(self) -> None: + scr = self.scr + scr.erase() + height, width = scr.getmaxyx() + if height < self.MIN_HEIGHT or width < self.MIN_WIDTH: + msg = "Terminal too small" + _addstr(scr, height // 2, max(0, (width - len(msg)) // 2), + msg, self.curses.A_BOLD) + scr.refresh() + return + dialog_w = self._measure(width) + flat = self._flatten(dialog_w - 4) + y0, x0, dialog_h, visible = self._geometry(height, width, + dialog_w, flat) + self._draw_frame(y0, x0, dialog_h, dialog_w, len(flat), visible) + self._draw_rows(y0, x0, dialog_w, flat, visible) + self._draw_buttons(y0, x0, dialog_h, dialog_w) + self._draw_status_footer(y0, x0, dialog_h, dialog_w) + scr.refresh() + + def _draw_frame(self, y0: int, x0: int, dialog_h: int, dialog_w: int, + total_lines: int, visible: int) -> None: + curses, theme = self.curses, self.theme + scr = self.scr + border = theme["border"] + _addch(scr, y0, x0, curses.ACS_ULCORNER, border) + _addch(scr, y0, x0 + dialog_w - 1, curses.ACS_URCORNER, border) + _addch(scr, y0 + dialog_h - 1, x0, curses.ACS_LLCORNER, border) + _addch(scr, y0 + dialog_h - 1, x0 + dialog_w - 1, + curses.ACS_LRCORNER, border) + _hline(scr, y0, x0 + 1, dialog_w - 2, border) + _hline(scr, y0 + dialog_h - 1, x0 + 1, dialog_w - 2, border) + for y in range(y0 + 1, y0 + dialog_h - 1): + _addch(scr, y, x0, curses.ACS_VLINE, border) + _addch(scr, y, x0 + dialog_w - 1, curses.ACS_VLINE, border) + + inner_x = x0 + 1 + inner_w = dialog_w - 2 + title = _fit(f" {self.title} ", inner_w) + _addstr(scr, y0 + 1, inner_x + max(0, (inner_w - len(title)) // 2), + title, theme["title"]) + if total_lines > visible: + indicator = f" {self.scroll + 1}/{total_lines} " + _addstr(scr, y0, max(x0 + 1, x0 + dialog_w - 1 - len(indicator)), + indicator, theme["dim"]) + + def _draw_rows(self, y0: int, x0: int, dialog_w: int, + flat: List[Tuple[int, dict, Optional[str]]], + visible: int) -> None: + theme = self.theme + scr = self.scr + inner_x = x0 + 1 + inner_w = dialog_w - 2 + for line in range(self.scroll, min(len(flat), self.scroll + visible)): + logical, row, piece = flat[line] + y = y0 + 2 + (line - self.scroll) + selected = logical == self.cursor and row["selectable"] + if selected: + _addstr(scr, y, inner_x, " " * inner_w, theme["bar"]) + if row["segments"] is not None: + self._draw_segments_row(y, row, inner_x, inner_w, selected) + else: + self._draw_text_row(y, row, piece, inner_x, inner_w, + selected) + + def _draw_segments_row(self, y: int, row: dict, inner_x: int, + inner_w: int, selected: bool) -> None: + scr, theme = self.scr, self.theme + total = sum(len(text) for text, _ in row["segments"]) + if row["align"] == "left": + x = inner_x + self.LIST_MARGIN + 2 * row["indent"] + else: + x = inner_x + max(0, (inner_w - total) // 2) \ + + 2 * row["indent"] + # Never paint over the right border column. + room = max(0, inner_x + inner_w - 1 - x) + for text, attr in row["segments"]: + text = _fit(text, room) + if not text: + break + _addstr(scr, y, x, text, theme["bar"] if selected else attr) + x += len(text) + room -= len(text) + + def _draw_text_row(self, y: int, row: dict, piece: Optional[str], + inner_x: int, inner_w: int, selected: bool) -> None: + scr, theme = self.scr, self.theme + text = " " * row["indent"] + piece + if row["align"] == "left": + x = inner_x + self.LIST_MARGIN + limit = inner_w - 1 - self.LIST_MARGIN - 2 * row["indent"] + else: + x = inner_x + max(0, (inner_w - len(text)) // 2) + limit = inner_w + text = _fit(text, limit) + attr = theme["bar"] if selected else row["attr"] + _addstr(scr, y, x, text, attr) + + def _draw_buttons(self, y0: int, x0: int, dialog_h: int, + dialog_w: int) -> None: + if not self.buttons: + return + theme = self.theme + scr = self.scr + inner_x = x0 + 1 + inner_w = dialog_w - 2 + labels, selected = self.buttons + rendered = [f"[ {label} ]" for label in labels] + total = sum(len(r) for r in rendered) + 3 * (len(rendered) - 1) + x = inner_x + max(0, (inner_w - total) // 2) + y = y0 + dialog_h - 4 + for index, text in enumerate(rendered): + if index: + x += 3 + _addstr(scr, y, x, text, + theme["btn_on"] if index == selected + else theme["btn_off"]) + x += len(text) + + def _draw_status_footer(self, y0: int, x0: int, dialog_h: int, + dialog_w: int) -> None: + theme = self.theme + scr = self.scr + inner_x = x0 + 1 + inner_w = dialog_w - 2 + if self.status: + text, kind = self.status + attr = theme.get(kind, theme["body"]) + text = _fit(f" {text} ", inner_w) + _addstr(scr, y0 + dialog_h - 3, + inner_x + max(0, (inner_w - len(text)) // 2), + text, attr) + footer = _fit(self.footer, inner_w) + _addstr(scr, y0 + dialog_h - 2, + inner_x + max(0, (inner_w - len(footer)) // 2), + footer, theme["dim"]) + + # -- key helpers ------------------------------------------------------ + + def motion(self, key: int, cursor: int, count: int, + wrap: bool = False) -> Optional[int]: + """New cursor index for a motion KEY, or None when it moves nothing. + + Up/Down (or k/j) move one row, wrapping around at the ends when + WRAP is set (menus and trees) and clamping otherwise (the + browser); Home/End jump to the first/last row; PageUp/PageDown + move self.page_size rows. COUNT is the number of rows. + """ + curses = self.curses + if key in (curses.KEY_UP, ord("k")): + if wrap and cursor <= 0: + return count - 1 + return max(0, cursor - 1) + if key in (curses.KEY_DOWN, ord("j")): + if wrap and cursor >= count - 1: + return 0 + return min(count - 1, cursor + 1) + if key == curses.KEY_HOME: + return 0 + if key == curses.KEY_END: + return count - 1 + if key == curses.KEY_PPAGE: + return max(0, cursor - self.page_size) + if key == curses.KEY_NPAGE: + return min(count - 1, cursor + self.page_size) + return None + + def get_key(self, cancel_keys: Sequence[int] = (27,)) -> int: + """Read one key; cancel keys and Ctrl-C raise WizardCancelled.""" + try: + key = self.scr.getch() + except KeyboardInterrupt: + raise WizardCancelled() from None + if key == 3: # Ctrl-C + raise WizardCancelled() + if key in cancel_keys: + raise WizardCancelled() + return key + + def flash(self, text: str, kind: str = "err") -> None: + """Show TEXT on the status line until any key is pressed.""" + self.status = (text, kind) + self.draw() + try: + key = self.scr.getch() + if key == 3: # Ctrl-C still aborts + raise WizardCancelled() + except KeyboardInterrupt: + raise WizardCancelled() from None + self.status = None + + def edit_status(self, prompt: str = "") -> Optional[str]: + """Edit a line of text on the status line. + + Returns the edited string on Enter, or None when the user backs + out with Esc (the caller decides what that means). + """ + curses = self.curses + text = "" + while True: + self.status = (f"{prompt}{text}_", "input") + self.draw() + try: + key = self.scr.getch() + except KeyboardInterrupt: + raise WizardCancelled() from None + if key == 27: + return None + if key == 3: # Ctrl-C + raise WizardCancelled() + if key in (10, 13): + return text + if key in (curses.KEY_BACKSPACE, 8, 127): + text = text[:-1] + elif 32 <= key < 127: + text += chr(key) + + +# --------------------------------------------------------------------------- +# Widget: yes/no confirm with buttons +# --------------------------------------------------------------------------- + +def confirm(scr, question: str, default: bool = False, + body: Optional[Sequence[str]] = None, + cancel_value: object = None): + """Ask a yes/no QUESTION with centered Yes/No buttons. + + The QUESTION is the dialog title (shown exactly once); optional + BODY lines sit centered above the buttons. Tab or the arrow keys + switch the buttons, Enter activates the highlighted one (the + DEFAULT button starts highlighted, drawn bright against the dim + other one), and y/n answer directly. Esc (or 'q') aborts the + wizard — unless CANCEL_VALUE is given (not None), in which case it + is returned instead, so the caller can fall back to a previous + screen rather than aborting the whole wizard. + """ + frame = Frame(scr, question, + "Tab/arrows = switch Enter = confirm y/n Esc = cancel") + index = 0 if default else 1 + while True: + frame.rows = [] + for line in body or []: + frame.mark(line) + frame.cursor = None + frame.buttons = (["Yes", "No"], index) + frame.draw() + curses = frame.curses + key = frame.get_key(cancel_keys=()) + if key in _CANCEL_KEYS: + if cancel_value is not None: + return cancel_value + raise WizardCancelled() + if key in (9, curses.KEY_LEFT, curses.KEY_RIGHT, curses.KEY_UP, + curses.KEY_DOWN, curses.KEY_BTAB, ord("h"), ord("l")): + index = 1 - index + elif key in (ord("y"), ord("Y")): + return True + elif key in (ord("n"), ord("N")): + return False + elif key in (10, 13): + return index == 0 + + +# --------------------------------------------------------------------------- +# Widget: single-choice menu +# --------------------------------------------------------------------------- + +def menu(scr, title: str, options: Sequence[tuple], default_index: int = 0, + help_lines: Optional[Sequence[str]] = None, + back_value: object = None, + table_title: Optional[str] = None, + table_rows: Optional[Sequence[tuple]] = None): + """Show OPTIONS as (label, value) pairs; return the chosen value. + + The cursor starts on DEFAULT_INDEX; Enter returns the highlighted + option's value. Options are left-justified like a DOS list; + HELP_LINES are dim, centered explanatory lines shown above them. + + TABLE_TITLE + TABLE_ROWS render an aligned two-column table above the + options: each row is (name, status, kind) where KIND is a theme key + ("ok"/"warn"/"err"/"info"/...), optionally followed by NAME_KIND, a + theme key for the name column ("dim" to fade an unusable entry; + "body" — the default — otherwise). The name column is padded to the + widest name so every status starts at the same column — a monospace + grid. The title is dim and left-aligned with the rows. Used by the + hub to show each backend's state (unavailable / installed / running) + in matching columns with color. + + Esc (or 'q') aborts the wizard unless BACK_VALUE is given (not None), + in which case Esc returns it so the caller can fall back a screen. + """ + if not options: + raise ValueError("menu() needs at least one option") + frame = Frame(scr, title, + "Up/Down = move Enter = select Esc = cancel") + cursor = max(0, min(default_index, len(options) - 1)) + while True: + frame.rows = [] + for line in help_lines or []: + frame.mark(line, frame.theme["dim"]) + if help_lines: + frame.mark("") + if table_rows: + if table_title: + frame.mark(table_title, frame.theme["dim"], align="left") + name_w = max(len(row[0]) for row in table_rows) + for row in table_rows: + name, status, kind = row[0], row[1], row[2] + name_kind = row[3] if len(row) > 3 else "body" + frame.mark_segments( + [(name.ljust(name_w), + frame.theme.get(name_kind, frame.theme["body"])), + (" " + status, + frame.theme.get(kind, frame.theme["body"]))], + align="left") + frame.mark("") + base = len(frame.rows) + for label, _ in options: + frame.mark(label, selectable=True, align="left") + frame.cursor = base + cursor + frame.draw() + key = frame.get_key(cancel_keys=()) + if key == 27 and back_value is not None: + return back_value + if key in _CANCEL_KEYS: + raise WizardCancelled() + moved = frame.motion(key, cursor, len(options), wrap=True) + if moved is not None: + cursor = moved + elif key in (10, 13): + return options[cursor][1] + + +# --------------------------------------------------------------------------- +# Widget: single-line text editor +# --------------------------------------------------------------------------- + +def line_edit(scr, title: str, default: str, + validate: Optional[Callable[[str], Optional[str]]] = None, + help_lines: Optional[Sequence[str]] = None, + back_value: object = None) -> str: + """Edit one line of text, pre-filled with DEFAULT; Enter accepts. + + HELP_LINES are dim explanatory lines shown above the input. + VALIDATE receives the entered string and returns an error message + or None; Enter on an invalid value shows the message in red and + keeps editing. Esc aborts the wizard ('q' is an ordinary + character here) unless BACK_VALUE is given (not None), in which case + Esc returns it so the caller can fall back a screen. + """ + frame = Frame(scr, title, + "type to edit Backspace = erase Enter = accept " + "Esc = cancel") + text = default + error = None + while True: + frame.rows = [] + for line in help_lines or []: + frame.mark(line, frame.theme["dim"]) + frame.mark("") + frame.mark(f"{text}_", frame.theme["input"]) + frame.cursor = None + frame.status = (error, "err") if error else None + frame.draw() + curses = frame.curses + key = frame.get_key(cancel_keys=()) # handle Esc manually below + if key == 27 and back_value is not None: + return back_value + if key == 27: + raise WizardCancelled() + if key in (10, 13): + if validate is None: + return text + error = validate(text) + if error is None: + return text + continue + if key in (curses.KEY_BACKSPACE, 8, 127): + text = text[:-1] + elif key == 21: # Ctrl-U: clear the line + text = "" + elif 32 <= key < 127: + text += chr(key) + + +# --------------------------------------------------------------------------- +# Widget: directory browser +# --------------------------------------------------------------------------- + +def _list_dirs(path: Path) -> List[Path]: + """Return the subdirectories of PATH, sorted, dot-dirs excluded.""" + try: + entries = [child for child in path.iterdir() + if child.is_dir() and not child.name.startswith(".")] + except OSError: + return [] + return sorted(entries, key=lambda child: child.name.lower()) + + +def browse_directory(scr, title: str, + validate: Optional[Callable[[Path], Optional[str]]] = None, + start: Optional[Path] = None, + info: Optional[Callable[[Path], + Optional[Tuple[str, str]]]] = None, + preview: Optional[Callable[[Path], + Optional[Tuple[str, str]]]] = None, + help_lines: Optional[Sequence[str]] = None, + auto_select: Optional[Callable[ + [Path], Optional[Path]]] = None, + back_value: object = None + ) -> Path: + """Pick a directory DOS-browser style. + + The listing starts with a bright '[ Use this directory ]' row (the + cursor starts there; Enter accepts the directory being listed), a + dim '..' for the parent, and one row per subdirectory. List rows + are left-justified; instructions and the current path stay + centered. Enter or Right on a highlighted subdirectory opens it, + Left/Backspace goes to the parent, 'e' types a path directly, and + Home/End/PageUp/PageDown navigate long listings. Coming back out + of a directory highlights the directory you came from. + + VALIDATE receives the listed directory and returns an error message + or None; Enter on an invalid directory is refused with that message. + INFO(directory) returns a (text, kind) status shown under the + listed directory's path — kind is "ok" (green), "warn" (yellow), + "err" (red), "info" (dim) or "input". PREVIEW(directory) returns + one for the highlighted subdirectory, shown on the status line. + AUTO_SELECT receives a highlighted subdirectory when the user + opens it (Enter, Right or 'l') and may return a Path to accept + immediately — as if '[ Use this directory ]' had been pressed on + it — instead of descending; returning None keeps browsing. This + lets a subdirectory that already looks like the target (e.g. an + 'audio.cpp' checkout containing 'model_specs/') be picked in one + keystroke. Esc (or 'q') aborts the wizard unless BACK_VALUE is given + (not None), in which case Esc returns it so the caller can fall back + a screen. + """ + footer = ("Up/Down = move Enter = open/use Left = parent " + "e = type path Esc = cancel") + frame = Frame(scr, title, footer) + current = Path(start) if start is not None else Path.cwd() + try: + current = current.resolve() + except OSError: + current = Path.cwd() + sel = 0 + highlight: Optional[Path] = None + + def validation_error() -> Optional[str]: + if validate is None: + return None + try: + return validate(current) + except OSError: + return "Cannot read this directory" + + def call(callback, path: Path) -> Optional[Tuple[str, str]]: + if callback is None: + return None + try: + return callback(path) + except OSError: + return None + + while True: + entries = _list_dirs(current) + has_parent = current.parent != current + offset = 1 + (1 if has_parent else 0) + frame.rows = [] + for line in help_lines or []: + frame.mark(line, frame.theme["dim"]) + frame.mark(f"Directory: {current}", frame.theme["accent"]) + current_info = call(info, current) + if current_info: + frame.mark(current_info[0], + frame.theme.get(current_info[1], frame.theme["body"])) + frame.mark("") + frame.mark("[ Use this directory ]", frame.theme["ok"], + selectable=True, align="left") + if has_parent: + frame.mark("..", frame.theme["dim"], selectable=True, + align="left") + for entry in entries: + frame.mark(f"{entry.name}/", selectable=True, align="left") + selectable = frame.selectable() + if highlight is not None: + sel = 0 + for index, entry in enumerate(entries): + if entry == highlight: + sel = offset + index + break + highlight = None + sel = max(0, min(sel, len(selectable) - 1)) + frame.cursor = selectable[sel] if selectable else None + + if sel == 0: + frame.status = ("Enter = use this directory", "info") + elif has_parent and sel == 1: + frame.status = ("Enter = open the parent directory", "info") + else: + entry = entries[sel - offset] + frame.status = call(preview, entry) \ + or (f"Enter = open {entry.name}/", "info") + frame.draw() + curses = frame.curses + key = frame.get_key(cancel_keys=()) + if key == 27 and back_value is not None: + return back_value + if key in _CANCEL_KEYS: + raise WizardCancelled() + moved = frame.motion(key, sel, len(selectable)) + if moved is not None: + sel = moved + elif key in (10, 13, curses.KEY_RIGHT, ord("l")): + if sel == 0: + error = validation_error() + if error is None: + return current + frame.flash(f"{error} (keep browsing)", "err") + elif has_parent and sel == 1: + highlight = current + current = current.parent + else: + entry = entries[sel - offset] + if auto_select is not None: + picked = auto_select(entry) + if picked is not None: + return picked + current = entry + sel = 0 + elif key in (curses.KEY_LEFT, ord("h"), ord("u"), + curses.KEY_BACKSPACE, 8, 127): + if has_parent: + highlight = current + current = current.parent + elif key == ord("e"): + result = frame.edit_status(prompt="path: ") + if result: + candidate = Path(os.path.expanduser(result)) + if not candidate.is_absolute(): + candidate = current / candidate + try: + candidate = candidate.resolve() + except OSError: + pass + if candidate.is_dir(): + current = candidate + sel = 0 + else: + frame.flash(f"Not a directory: {candidate}", "err") + + +# --------------------------------------------------------------------------- +# Widget: expandable checkbox tree +# --------------------------------------------------------------------------- + +def checkbox_tree(scr, title: str, families: List[dict], + footer: Optional[str] = None, + expand_all: bool = False, + back_value: object = None) -> List[Tuple[int, str]]: + """Pick model families and packages from an expandable tree. + + FAMILIES is a list of dicts (one per family) shaped like:: + + { + "label": "Qwen3-TTS (qwen3_tts)", + "detail": "tts, cloning, design", + "options": [ + {"key": "Base-GGUF", "label": "base", "recommended": True}, + {"key": "VoiceDesign-GGUF", "label": "voicedesign", + "recommended": False}, + ], + } + + Space on a family row checks its recommended option (or clears every + option when one is already checked); Space on an option row toggles + that option. Tab/Right expands or collapses the family under the + cursor. Enter returns the flat list of (family_index, option_key) + pairs for every checked option, in tree order; at least one checked + option is required. Nothing is checked by default, and with + EXPAND_ALL every family starts expanded. A "[recommended]" tag is + shown only when a + family has more than one option — a single option needs no tag. + Family and option rows are left-justified like a DOS list. Esc (or + 'q') aborts the wizard unless BACK_VALUE is given (not None), in + which case Esc returns it so the caller can fall back a screen. + """ + if not families: + raise ValueError("checkbox_tree() needs at least one family") + footer = footer or ("Up/Down = move Tab/Right = expand Space = check " + "Enter = accept Esc = cancel") + frame = Frame(scr, title, footer) + expanded = {index for index in range(len(families))} if expand_all else set() + checked = set() # (family_index, option_key) + + expanded.add(0) + + def family_checked(index: int) -> bool: + return any(pair[0] == index for pair in checked) + + def accept() -> List[Tuple[int, str]]: + return [(index, option["key"]) + for index, family in enumerate(families) + for option in family["options"] + if (index, option["key"]) in checked] + + def visible_nodes() -> List[tuple]: + nodes: List[tuple] = [] # ("family", i) or ("option", i, key) + for index, family in enumerate(families): + nodes.append(("family", index)) + if index in expanded: + for option in family["options"]: + nodes.append(("option", index, option["key"])) + return nodes + + cursor = 0 + while True: + nodes = visible_nodes() + cursor = max(0, min(cursor, len(nodes) - 1)) + frame.rows = [] + for node in nodes: + if node[0] == "family": + index = node[1] + family = families[index] + on = family_checked(index) + mark = "x" if on else " " + arrow = "-" if index in expanded else "+" + frame.mark_segments( + [(f"[{mark}] ", + frame.theme["check"] if on else frame.theme["dim"]), + (f"{arrow} {family['label']}", + frame.theme["accent"] if on else frame.theme["body"])], + selectable=True, align="left") + else: + _, index, option_key = node + option = next(opt for opt in families[index]["options"] + if opt["key"] == option_key) + is_on = (index, option_key) in checked + mark = "x" if is_on else " " + segments = [(f"[{mark}] ", + frame.theme["check"] if is_on + else frame.theme["dim"]), + (option["label"], frame.theme["body"])] + if option.get("recommended") \ + and len(families[index]["options"]) > 1: + segments.append((" [recommended]", frame.theme["warn"])) + frame.mark_segments(segments, indent=2, selectable=True, + align="left") + frame.cursor = cursor + node = nodes[cursor] + frame.status = (families[node[1]].get("detail", ""), "info") + frame.draw() + curses = frame.curses + key = frame.get_key(cancel_keys=()) + if key == 27 and back_value is not None: + return back_value + if key in _CANCEL_KEYS: + raise WizardCancelled() + moved = frame.motion(key, cursor, len(nodes), wrap=True) + if moved is not None: + cursor = moved + elif key in (9, curses.KEY_RIGHT, ord("l")) and node[0] == "family": + index = node[1] + if index in expanded: + expanded.discard(index) + else: + expanded.add(index) + elif key == curses.KEY_LEFT and node[0] == "family": + expanded.discard(node[1]) + elif key == ord(" "): + if node[0] == "family": + index = node[1] + options = families[index]["options"] + if family_checked(index): + for option in options: + checked.discard((index, option["key"])) + else: + for option in options: + if option.get("recommended"): + checked.add((index, option["key"])) + break + else: + if options: + checked.add((index, options[0]["key"])) + expanded.add(index) + else: + _, index, option_key = node + if (index, option_key) in checked: + checked.discard((index, option_key)) + else: + checked.add((index, option_key)) + elif key in (10, 13): # Enter: accept the checked selection + selection = accept() + if selection: + return selection + frame.flash("Check at least one model package (Space)", "err") -- cgit v1.2.3