diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/make_audiocpp_server_json.py | 1029 | ||||
| -rw-r--r-- | tools/tui.py | 843 |
2 files changed, 1384 insertions, 488 deletions
diff --git a/tools/make_audiocpp_server_json.py b/tools/make_audiocpp_server_json.py index fb16a43..3446428 100755 --- a/tools/make_audiocpp_server_json.py +++ b/tools/make_audiocpp_server_json.py @@ -8,15 +8,24 @@ converter itself is family-agnostic (it detects the family of the selected entry from ``GET /v1/models`` at startup), so any TTS family listed in the catalog works without further changes. -By default the tool runs as a minimal full-screen TUI (curses): a file -browser for the audio.cpp checkout and the .wav directory, an expandable -checkbox tree of model families and their installable packages, and a -series of single-question screens for the server settings. Pass ``--notui`` -to use the classic numbered line prompts instead (also selected +By default the tool runs as a colorful full-screen TUI (curses): every +screen is a centered DOS-style dialog on a black desktop — a file +browser for the audio.cpp checkout and the .wav directory, an +expandable checkbox tree of model families and their installable +packages, centered single-question screens for the server settings, +and Yes/No buttons for every confirmation. In the checkout browser, +pressing Enter (or Right) on a subdirectory named ``audio.cpp`` that +already contains ``model_specs/`` picks it directly, skipping the +``[ Use this directory ]`` step; pressing Esc on the overwrite +confirmation then returns to the browser inside that checkout (with +the auto-pick disabled), instead of aborting the wizard. Esc on any +other wizard screen falls back to the previous screen group (only the +first screen, the checkout browser, exits on Esc). Pass ``--notui`` to +use the classic numbered line prompts instead (also selected automatically when stdin/stdout is not a terminal, or when curses is -unavailable such as on Windows without ``windows-curses``). Every value -can also be supplied as a command-line flag, which skips the corresponding -screen or prompt. +unavailable such as on Windows without ``windows-curses``). Every +value can also be supplied as a command-line flag, which skips the +corresponding screen or prompt. Each family is hosted through its recommended package by default; the TUI tree always lists every installable package (distinct ``target_directory`` @@ -24,7 +33,12 @@ values) as checkboxes, while ``--all-packages`` in prompt mode offers a per-family package checklist (and pre-expands every family in the TUI). Packages whose name marks them as voice-design models are asked whether to host them with task "vdes" (describe the voice with ``--instructions``) or -plain "tts". +plain "tts". All families are treated equally and listed alphabetically. + +The .wav directory browser (and the prompt default) starts in the single +directory that directly contains .wav files across the audio.cpp checkout +and the tts-audiobook-generator root, if exactly one exists; the +generator's ``output/`` directory is never offered. Cloning reference .wav files (``--wavs DIR``) are transcribed with a local Whisper backend (faster_whisper or whisper) and published as a server-level @@ -50,19 +64,29 @@ with its resolved absolute path if it does not exist. server.json is written into the audio.cpp checkout by default (next to model_specs/). If that file already exists you are prompted [Y/n] before overwriting; answering "n" writes server.json in the current working -directory instead. +directory instead (in the TUI, Esc on that prompt returns to the +checkout browser rather than aborting). After a successful run the +console output is the written file plus one copy-pasteable +model_manager_v2.py install command per hosted model; you are also +asked whether to run those downloads automatically. --audiocpp-dir defaults to a detected audio.cpp checkout (the AUDIOCPP_DIR environment variable, or an ``audio.cpp`` directory next to or above the current working directory); if none is found it is asked interactively. The checkout must contain a ``model_specs/`` directory. A leading ``~`` in a path argument or prompt answer is expanded. + +--backend is the inference backend audiocpp_server was built for. When the +checkout contains a build directory (``build/<platform>-<backend>-<type>`` +with a built ``bin/audiocpp_server``), that backend is auto-detected, +selected by default and marked ``[auto-detected]`` in the menu. """ import argparse import json import os import re +import subprocess import sys import urllib.parse from pathlib import Path @@ -78,27 +102,42 @@ DEFAULT_HOST = "127.0.0.1" FALLBACK_PORT = 8080 CONFIG_PATH = Path(__file__).resolve().parent.parent / "converter" / "config.py" +# The tts-audiobook-generator checkout root (where audiobook.py lives), used +# to default the .wav directory browser. The audio.cpp checkout is detected +# separately (see detect_audiocpp_dir). +TTS_ROOT = Path(__file__).resolve().parent.parent +# Output directory of tts-audiobook-generator; never offered as a .wav source. +TTS_OUTPUT_DIR = "output" + BACKENDS = ("cuda", "vulkan", "hip", "cpu") PROMPT_TEXT_FILENAME = "prompt_text" TASK_TTS = "tts" TASK_VDES = "vdes" +# Sentinel returned by tui.confirm (via its cancel_value) when the user +# presses Esc on an overwrite prompt to go back to the checkout browser +# instead of aborting the wizard. +_GO_BACK = object() + + +class _GoBack(Exception): + """Raised inside the TUI wizard to fall back to the previous screen group. + + Every wizard widget is passed ``back_value=_GO_BACK`` so Esc returns the + sentinel instead of aborting; pickers and confirmations that call into + callbacks (task/id pickers, the transcription plan, the download prompt) + convert that sentinel into this exception so the enclosing step can catch + it and step back. Only the first screen (the checkout browser) lets Esc + abort the whole wizard. + """ + # Package names that mark a voice-design model (hosted with task "vdes"). DESIGN_PACKAGE_RE = re.compile(r"voice[\s_\-]?design", re.IGNORECASE) -# Families explicitly tested with this converter, in display order. These are -# listed first in the checklist and marked "[tested]"; every other TTS family -# in the catalog is offered too through the converter's generic profile. -TESTED_FAMILIES = ( - "qwen3_tts", - "higgs_audio_tts", - "voxcpm2", - "index_tts2", -) - -# Short, friendly default entry ids for tested families. Other families derive -# an id from their family name (see default_model_id). +# Short, friendly default entry ids for selected families. Other families +# derive an id from their family name (see default_model_id). All families +# are listed equally, in alphabetical order. PREFERRED_IDS = { "qwen3_tts": "qwen", "higgs_audio_tts": "higgs", @@ -172,6 +211,127 @@ def find_wav_files(input_dir: Path) -> list: ) +def _count_wavs(directory: Path) -> int: + """Count the .wav files in DIRECTORY (0 when it cannot be read).""" + try: + return sum(1 for path in directory.iterdir() + if path.is_file() and path.suffix.lower() == ".wav") + except OSError: + return 0 + + +def detect_wav_dir(audiocpp_dir: Path, tts_root: Path) -> Optional[Path]: + """Find a unique directory that directly contains .wav files. + + Looks shallowly (the root itself and its immediate subdirectories) in + both the audio.cpp checkout and the tts-audiobook-generator root (where + audiobook.py lives), since clone reference .wavs commonly live in either. + The tts-audiobook-generator ``output/`` directory is excluded. When + exactly one candidate is found it is returned (as a starting directory + for the .wav browser); when none or several are found None is returned + so the caller falls back to its default start location. + """ + candidates: List[Path] = [] + seen: Set[Path] = set() + + def consider(directory: Path) -> None: + try: + resolved = directory.resolve() + except OSError: + return + if resolved in seen: + return + seen.add(resolved) + if _count_wavs(directory) > 0: + candidates.append(directory) + + for root in (audiocpp_dir, tts_root): + if not root.is_dir(): + continue + consider(root) + try: + children = sorted(root.iterdir(), key=lambda p: p.name.lower()) + except OSError: + continue + for child in children: + if not child.is_dir() or child.name.startswith("."): + continue + # Exclude the tts-audiobook-generator output directory. + if root == tts_root and child.name == TTS_OUTPUT_DIR: + continue + consider(child) + + if len(candidates) == 1: + return candidates[0] + return None + + +def _wav_dir_info(directory: Path) -> Tuple[str, str]: + """TUI status describing the directory listed in the wav browser.""" + count = _count_wavs(directory) + if count: + wavs = ".wav" if count == 1 else ".wavs" + return (f"{count} {wavs} found in this directory. Press Enter.", + "ok") + return ("No .wav files found in this directory", "warn") + + +def _wav_dir_preview(directory: Path) -> Tuple[str, str]: + """TUI status describing a highlighted subdirectory in the wav browser.""" + count = _count_wavs(directory) + if count: + wavs = ".wav" if count == 1 else ".wavs" + return (f"{count} {wavs}", "ok") + return ("no .wav files", "info") + + +def _resolve_audiocpp_root(directory: Path) -> Optional[Path]: + """Return the audio.cpp checkout root for DIRECTORY, or None. + + Accepts either the checkout root itself (it must contain a + ``model_specs`` directory) or the ``model_specs`` directory inside + it (the parent is used), so the file browser cannot pick the wrong + one of the two. + """ + if (directory / "model_specs").is_dir(): + return directory + if directory.name == "model_specs" and directory.is_dir(): + return directory.parent + return None + + +def _audiocpp_root_status(directory: Path) -> Tuple[str, str]: + """TUI status describing the directory listed in the checkout browser.""" + if _resolve_audiocpp_root(directory) is not None: + return ("model_specs/ found here", "ok") + return ("No model_specs/ directory here", "warn") + + +def _audiocpp_root_preview(directory: Path) -> Optional[Tuple[str, str]]: + """TUI status for a highlighted subdirectory in the checkout browser.""" + if (directory / "model_specs").is_dir(): + return ("contains model_specs/", "ok") + return None + + +def _checkout_auto_select(entry: Path) -> Optional[Path]: + """Auto-accept a highlighted checkout in the TUI browser. + + A subdirectory named ``audio.cpp`` that already contains a + ``model_specs`` directory is the audio.cpp checkout root, so it is + accepted immediately on Enter/Right (as if ``[ Use this directory ]`` + had been pressed) instead of being descended into. Anything else + returns None so the user keeps browsing. This is only consulted + while auto-accepting is still enabled; after the user presses Esc to + go back, the browser is restarted inside the previously accepted + checkout and this callback is no longer passed, so a wrong guess can + be corrected. + """ + if entry.name == "audio.cpp" and (entry / "model_specs").is_dir(): + return entry + return None + + def ask(prompt: str, default: Optional[str] = None) -> Optional[str]: """Prompt for a free-text value with a default; EOF returns the default.""" suffix = f" [{default}]" if default is not None else "" @@ -267,15 +427,43 @@ def ask_checklist(title: str, options: list, default: Set[str]) -> Set[str]: print(f"Please enter comma-separated numbers between 1 and {len(options)}.") -def ask_backend() -> str: +# Backend display order, with short descriptions. The backend name is padded +# so the descriptions' dashes line up in the menu. +_BACKEND_DESCRIPTIONS = ( + ("cuda", "NVIDIA GPUs (fastest)"), + ("vulkan", "cross-vendor GPU"), + ("hip", "AMD GPUs"), + ("cpu", "no GPU required"), +) + + +def _backend_options(detected: Optional[str] = None + ) -> Tuple[List[Tuple[str, str]], int]: + """Build the aligned backend menu options and the default index. + + The backend names are padded to a common width so the ``-`` dashes + before the descriptions line up. When DETECTED matches one of the + options, that option gets ``[auto-detected]`` appended and is the + default (cursor/start) selection; otherwise the first option is the + default as before. Returns (options, default_index). + """ + width = max(len(name) for name, _ in _BACKEND_DESCRIPTIONS) + options: List[Tuple[str, str]] = [] + default_index = 0 + for index, (name, desc) in enumerate(_BACKEND_DESCRIPTIONS): + label = f"{name.ljust(width)} - {desc}" + if detected == name: + label += " [auto-detected]" + default_index = index + options.append((label, name)) + return options, default_index + + +def ask_backend(detected: Optional[str] = None) -> str: + options, default_index = _backend_options(detected) return ask_menu( "Which inference backend was audiocpp_server built for?", - [ - ("cuda - NVIDIA GPUs (fastest)", "cuda"), - ("vulkan - cross-vendor GPU", "vulkan"), - ("hip - AMD GPUs", "hip"), - ("cpu - no GPU required", "cpu"), - ]) + options, default_index=default_index + 1) def config_port() -> int: @@ -384,6 +572,53 @@ def detect_audiocpp_dir() -> Optional[Path]: return None +# audio.cpp build directories are named ``<platform>-<backend>-<type>`` (e.g. +# ``linux-cuda-release``, ``windows-vulkan-debug``, ``macos-metal-release``) +# and the built server lands in ``<that>/bin/audiocpp_server``. The Metal +# macOS backend is reported as "cpu" here since it is not a separate +# --backend choice for audiocpp_server. +_BACKEND_TOKEN_RE = re.compile(r"-(cuda|vulkan|hip|cpu|metal)(?:-|$)") + + +def detect_backend(audiocpp_dir: Path) -> Optional[str]: + """Best-effort detection of the backend audiocpp_server was built for. + + Scans ``audiocpp_dir/build/*`` for build directories that contain a + built ``bin/audiocpp_server`` (``.exe`` allowed on Windows) and reads + the backend token out of the directory name (``-cuda-``, ``-vulkan-``, + ``-hip-`` or ``-cpu-``; ``-metal-`` is mapped to ``cpu``). Returns the + backend only when exactly one distinct backend was built, so a checkout + with builds for several backends does not silently pick one. Returns + None when there is no ``build/`` directory, no built server, or more + than one distinct backend. + """ + build_root = audiocpp_dir / "build" + if not build_root.is_dir(): + return None + backends: Set[str] = set() + try: + build_dirs = sorted(build_root.iterdir(), + key=lambda p: p.name.lower()) + except OSError: + return None + for build_dir in build_dirs: + if not build_dir.is_dir(): + continue + server = build_dir / "bin" / "audiocpp_server" + if not server.exists(): + server_exe = build_dir / "bin" / "audiocpp_server.exe" + if not server_exe.exists(): + continue + match = _BACKEND_TOKEN_RE.search(build_dir.name.lower()) + if not match: + continue + token = match.group(1) + backends.add("cpu" if token == "metal" else token) + if len(backends) == 1: + return next(iter(backends)) + return None + + def _default_package(packages: List[dict]) -> Optional[dict]: """Pick the default package from a list of packages. @@ -407,8 +642,8 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: Each returned entry has: family, display_name, description, languages, clone_capable, packages (the full list from the spec), install_id (recommended package id), default_path (``models/<target_directory>``), - tested, and preferred_id. Tested families come first (in TESTED_FAMILIES - order), the rest follow alphabetically by display name. + and preferred_id. All families are treated equally and listed in + alphabetical order by display name. """ specs_dir = audiocpp_dir / "model_specs" if not specs_dir.is_dir(): @@ -444,17 +679,11 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: "packages": packages, "install_id": package.get("id") or family, "default_path": f"models/{target_directory}", - "tested": family in TESTED_FAMILIES, "preferred_id": default_model_id(family), }) - def sort_key(entry: dict) -> tuple: - family = entry["family"] - if family in TESTED_FAMILIES: - return (0, TESTED_FAMILIES.index(family), "") - return (1, 0, entry["display_name"].lower()) - - entries.sort(key=sort_key) + # All families are treated equally: alphabetical by display name. + entries.sort(key=lambda entry: entry["display_name"].lower()) return entries @@ -708,9 +937,7 @@ def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None: def _apply_port_sync(port: int, accepted: bool) -> None: """Write the port into converter/config.py, or report when declined.""" if accepted: - if update_config_api_url_port(port): - print(f"[OK] Updated AUDIOCPP_API_URL in {CONFIG_PATH}") - else: + if not update_config_api_url_port(port): print(f"[WARNING] Could not update {CONFIG_PATH}; edit " "AUDIOCPP_API_URL by hand so audiobook.py uses the " "new port") @@ -720,9 +947,14 @@ def _apply_port_sync(port: int, accepted: bool) -> None: def _ask_host_port_backend_lazy(args: argparse.Namespace, - default_lazy: bool + default_lazy: bool, + detected_backend: Optional[str] = None ) -> Tuple[str, int, str, bool]: - """Ask for (or take from flags) the shared server settings.""" + """Ask for (or take from flags) the shared server settings. + + DETECTED_BACKEND (from detect_backend) is offered as the default backend + selection when --backend is not given. + """ host = args.host if args.host else ask("Bind host", DEFAULT_HOST) port = args.port if args.port is not None else ask_port(config_port()) if port != config_port(): @@ -731,7 +963,8 @@ def _ask_host_port_backend_lazy(args: argparse.Namespace, _apply_port_sync(port, True) else: _apply_port_sync(port, False) - backend = args.backend if args.backend else ask_backend() + backend = args.backend if args.backend else \ + ask_backend(detected_backend) lazy_load = args.lazy_load or ask_bool( "Load models lazily (on first use instead of at startup)", default_lazy) return host, port, backend, lazy_load @@ -742,7 +975,10 @@ def _decide_transcription(wav_files: list, existing: Dict[str, str], confirm: Callable[[str, bool], bool]) -> dict: """Decide which voices to transcribe; CONFIRM asks the plan questions. - Returns a plan dict: {"mode": "all"|"missing"|"keep", "missing": [...]}. + Returns a plan dict: {"mode": "all"|"missing"|"keep", "missing": + [...], "existing": {...}} — "existing" carries the prompt_text + mapping read while deciding, so the caller can reuse it instead of + reading the file again. """ mode = "all" missing: List[Path] = [] @@ -760,7 +996,7 @@ def _decide_transcription(wav_files: list, existing: Dict[str, str], mode = "missing" else: mode = "all" - return {"mode": mode, "missing": missing} + return {"mode": mode, "missing": missing, "existing": existing} def _transcribe(args: argparse.Namespace, include_clone: bool, @@ -771,7 +1007,8 @@ def _transcribe(args: argparse.Namespace, include_clone: bool, Returns the mapping and a flag indicating whether it should be written to prompt_text (False when an existing, complete prompt_text is kept as-is). When PLAN is given (pre-collected by the TUI) no further questions are - asked; otherwise the plan is decided with the line prompts. + asked and the prompt_text mapping it already read is reused; otherwise + the plan is decided with the line prompts. """ if not include_clone: print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family " @@ -785,13 +1022,14 @@ def _transcribe(args: argparse.Namespace, include_clone: bool, return {}, False prompt_path = args.input_dir / PROMPT_TEXT_FILENAME - existing = read_prompt_text(prompt_path) if ( - prompt_path.exists() and not args.force) else {} - if plan is None: + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} plan = _decide_transcription( wav_files, existing, prompt_path.exists(), args.force, lambda question, default: ask_bool(question, default)) + else: + existing = plan.get("existing") or {} if plan["mode"] == "keep": print(f"[INFO] Kept existing {prompt_path}; all voices were " @@ -832,9 +1070,7 @@ def _offer_config_model_id_sync(model_id: str, f"in converter/config.py to '{model_id}' so " "audiobook.py uses this model", True) if accepted: - if update_config_model_ids(model_id, model_id): - print(f"[OK] Updated the model ids in {CONFIG_PATH}") - else: + if not update_config_model_ids(model_id, model_id): print(f"[WARNING] Could not update {CONFIG_PATH}; edit " "AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID by hand so " "audiobook.py uses this model") @@ -843,14 +1079,6 @@ def _offer_config_model_id_sync(model_id: str, f"still request model '{config.AUDIOCPP_MODEL_ID}'") -def _print_multi_model_model_id_note(entry_ids: List[str]) -> None: - """Tell the user how to select one entry per run for a multi-model server.""" - print("[INFO] Several model entries were configured. audiobook.py uses one " - "entry per run: pass --model <id> when converting, or set " - "AUDIOCPP_MODEL_ID in converter/config.py to one of: " - f"{', '.join(entry_ids)}") - - def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], catalog_by_family: Dict[str, dict], task_picker: Callable[[str], str], @@ -890,14 +1118,18 @@ def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], design_entry_ids, include_clone) -def _write_and_advise(wav_dir: Optional[Path], output_path: Path, - model_entries: List[dict], entry_ids: List[str], - install_guidance: List[Tuple[str, str]], - design_entry_ids: List[str], family_keys: List[str], - catalog_by_family: Dict[str, dict], host: str, port: int, - backend: str, lazy_load: bool, +def _write_and_advise(audiocpp_dir: Path, wav_dir: Optional[Path], + output_path: Path, model_entries: List[dict], + install_guidance: List[Tuple[str, str]], host: str, + port: int, backend: str, lazy_load: bool, transcripts: Dict[str, str], write_prompt: bool) -> None: - """Console phase shared by both UI modes: write files and print guidance.""" + """Console phase shared by both UI modes: write files, print summary. + + After a successful run the console output is the path of the written + server.json. The model install commands (and optional automatic + download) are handled separately by _install_models, called by both + UI modes once the user has decided whether to download. + """ voice_dir: Optional[str] = None if transcripts: if write_prompt: @@ -910,32 +1142,74 @@ def _write_and_advise(wav_dir: Optional[Path], output_path: Path, host=host, port=port, backend=backend, lazy_load=lazy_load, model_entries=model_entries, voice_dir=voice_dir) - print("\nGenerated server.json:") - print(json.dumps(server_config, indent=2, ensure_ascii=False)) - with output_path.open("w", encoding="utf-8") as handle: json.dump(server_config, handle, indent=2, ensure_ascii=False) handle.write("\n") - print(f"\n[OK] Wrote {output_path} with {len(model_entries)} model " - f"entry/entries" + (f" and voice_dir '{voice_dir}'" if voice_dir else "")) - for display_name, install_id in install_guidance: - print(f"[INFO] Install {display_name} from the audio.cpp checkout: " - f"python3 tools/model_manager_v2.py install {install_id}") - if len(model_entries) > 1: - print("[INFO] Models load lazily and stay in memory until the server " - "exits; restart the server (or POST /v1/tasks/unload_models) " - "before switching to a large model to free VRAM.") - for family in family_keys: - if catalog_by_family[family]["clone_capable"]: - print(f"[INFO] {catalog_by_family[family]['display_name']} supports " - "voice cloning: run audiobook.py with --backend audiocpp " - "--voice <preset name>") - for design_id in design_entry_ids: - print(f"[INFO] Voice design entry '{design_id}' hosted with task " - "'vdes': convert with python audiobook.py --backend audiocpp " - f"--model {design_id} " - '--instructions "A warm adult female narrator"') + count = len(model_entries) + print(f"Wrote {output_path.resolve()} with {count} " + f"{'entry' if count == 1 else 'entries'}.") + + +def _install_models(audiocpp_dir: Path, + install_guidance: List[Tuple[str, str]], + download: bool) -> None: + """Print and optionally run the model install commands. + + One ``python <manager> install <id>`` command per hosted model (de-duped + by install id). When DOWNLOAD is True each command is run in the audio.cpp + checkout via ``subprocess.run`` so the models are downloaded automatically; + a failing install is reported as a warning and does not abort the remaining + downloads. When DOWNLOAD is False (or the model manager is missing) the + commands are only printed, copy-pasteable as before. + """ + manager = audiocpp_dir / "tools" / "model_manager_v2.py" + seen: Set[str] = set() + install_ids: List[str] = [] + for _, install_id in install_guidance: + if install_id not in seen: + seen.add(install_id) + install_ids.append(install_id) + + if download and not manager.is_file(): + print(f"[WARNING] {manager} not found; printing the install commands " + "instead of running them") + download = False + + for install_id in install_ids: + command = f"python {manager} install {install_id}" + if not download: + print(command) + continue + print(f"[INFO] Downloading {install_id}...") + try: + result = subprocess.run( + [sys.executable, str(manager), "install", install_id], + cwd=str(audiocpp_dir)) + except OSError as exc: + print(f"[WARNING] Could not run {command}: {exc}") + continue + if result.returncode != 0: + print(f"[WARNING] install {install_id} exited with code " + f"{result.returncode}; the model may need to be downloaded " + "by hand") + + +def _decide_download(audiocpp_dir: Path, + confirm: Callable[[str, bool], bool]) -> bool: + """Ask whether to download the selected models now. + + CONFIRM asks the yes/no question (ask_bool for the line prompts, a TUI + confirm for the wizard). When the audio.cpp model manager is missing the + prompt is skipped and False is returned, so the install commands are only + printed rather than offered to run. + """ + manager = audiocpp_dir / "tools" / "model_manager_v2.py" + if not manager.is_file(): + return False + return confirm( + "Automatically download the selected models with model_manager_v2.py " + "now?", False) def _build_tree_families(catalog: List[dict]) -> List[dict]: @@ -950,16 +1224,11 @@ def _build_tree_families(catalog: List[dict]) -> List[dict]: name = entry["display_name"] if name != entry["family"]: name = f"{name} ({entry['family']})" - if entry["tested"]: - name = f"{name} [tested]" options = [] for opt in package_dir_options(entry): - label = opt["install_id"] - if opt["design"]: - label = f"{label} (voice design)" options.append({ "key": opt["target_directory"], - "label": label, + "label": opt["install_id"], "recommended": opt["recommended"], }) families.append({ @@ -972,194 +1241,348 @@ def _build_tree_families(catalog: List[dict]) -> List[dict]: def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser ) -> Optional[dict]: - """Run every TUI screen; return the collected settings, or None to abort.""" + """Run every TUI screen; return the collected settings, or None to abort. + + The wizard is a step state machine; each screen group is one step, and + Esc anywhere but the first step falls back to the previous group (the + widget returns the _GO_BACK sentinel, or a callback raises _GoBack). On + the first screen (the audio.cpp checkout browser) Esc aborts the whole + wizard as before. + """ tui = _load_tui() - # 1. audio.cpp checkout (flag, detected, or browsed). - audiocpp_dir = args.audiocpp_dir - if audiocpp_dir is None: - audiocpp_dir = detect_audiocpp_dir() - if audiocpp_dir is None: - audiocpp_dir = tui.browse_directory( - stdscr, "Locate your audio.cpp checkout", - validate=lambda p: None if (p / "model_specs").is_dir() - else "No model_specs/ directory here", - start=Path.cwd()) - audiocpp_dir = Path(audiocpp_dir).resolve() - if not audiocpp_dir.is_dir(): - raise _TuiError(f"audio.cpp checkout not found: {audiocpp_dir}") - try: - catalog = load_model_catalog(audiocpp_dir) - except NotADirectoryError as exc: - raise _TuiError(str(exc)) - if not catalog: - raise _TuiError(f"No TTS model families found in " - f"{audiocpp_dir}/model_specs; check the checkout is " - "up to date") - catalog_by_family = {entry["family"]: entry for entry in catalog} + def ask_confirm(question: str, default: bool) -> bool: + result = tui.confirm(stdscr, question, default=default, + cancel_value=_GO_BACK) + if result is _GO_BACK: + raise _GoBack() + return result - # 2. Output path + overwrite confirmation. - output_path = args.output if args.output is not None \ - else audiocpp_dir / "server.json" - if not args.force and output_path.exists() \ - and not tui.confirm(stdscr, - f"{output_path} already exists. Overwrite?", - default=True): - if args.output is None: - output_path = Path.cwd() / "server.json" - if output_path.exists() and not tui.confirm( - stdscr, f"{output_path} already exists. Overwrite?", - default=True): - return None - else: - return None + step = 0 + while True: + if step == 0: + # Checkout browser + the output path/overwrite confirmation. The + # browser asks for the checkout root and finds model_specs/ inside + # it (picking the model_specs directory itself works too — its + # parent is used). A highlighted subdirectory named "audio.cpp" + # that already contains model_specs/ is auto-accepted on + # Enter/Right, skipping the "[ Use this directory ]" step. + # Pressing Esc on an overwrite confirmation returns here instead + # of aborting: the browser then restarts inside the previously + # accepted checkout with auto-accept disabled, so a wrong guess + # can be corrected. An explicit --audiocpp-dir flag has no + # browser to return to, so Esc still aborts there. Esc on the + # browser itself is the first step, so it aborts the wizard. + auto_accept = True + browser_start: Path = Path.cwd() + force_browse = False + while True: + audiocpp_dir = args.audiocpp_dir + if audiocpp_dir is None: + audiocpp_dir = detect_audiocpp_dir() + if force_browse: + audiocpp_dir = None + if audiocpp_dir is None: + audiocpp_dir = tui.browse_directory( + stdscr, "Select your audio.cpp directory", + validate=lambda p: None if _resolve_audiocpp_root(p) + else "No model_specs/ directory here", + info=_audiocpp_root_status, + preview=_audiocpp_root_preview, + help_lines=["The root folder of your audio.cpp " + "checkout;", + "it is the one that contains " + "model_specs/"], + start=browser_start, + auto_select=_checkout_auto_select if auto_accept + else None) + audiocpp_dir = Path(audiocpp_dir).resolve() + if not audiocpp_dir.is_dir(): + raise _TuiError(f"audio.cpp checkout not found: " + f"{audiocpp_dir}") + root = _resolve_audiocpp_root(audiocpp_dir) + if root is None: + raise _TuiError( + f"{audiocpp_dir} has no model_specs/ directory; " + "select the root of your audio.cpp checkout") + audiocpp_dir = root + try: + catalog = load_model_catalog(audiocpp_dir) + except NotADirectoryError as exc: + raise _TuiError(str(exc)) + if not catalog: + raise _TuiError(f"No TTS model families found in " + f"{audiocpp_dir}/model_specs; check the " + "checkout is up to date") + catalog_by_family = {entry["family"]: entry + for entry in catalog} + + output_path = args.output if args.output is not None \ + else audiocpp_dir / "server.json" + esc_back = args.audiocpp_dir is None + went_back = False + if not args.force and output_path.exists(): + decision = tui.confirm( + stdscr, f"{output_path} already exists. Overwrite?", + default=True, + cancel_value=_GO_BACK if esc_back else None) + if decision is _GO_BACK: + went_back = True + elif decision is False: + if args.output is None: + output_path = Path.cwd() / "server.json" + if output_path.exists(): + decision = tui.confirm( + stdscr, + f"{output_path} already exists. " + "Overwrite?", + default=True, + cancel_value=_GO_BACK if esc_back else None) + if decision is _GO_BACK: + went_back = True + elif decision is False: + return None + else: + return None + if went_back: + auto_accept = False + browser_start = audiocpp_dir + force_browse = True + continue + break + detected_backend = detect_backend(audiocpp_dir) + step = 1 + continue - # 3. Families and packages (flag or tree). - chosen: Dict[str, List[dict]] = {} - if args.families is not None: - requested = [f.strip() for f in args.families.split(",") if f.strip()] - unknown = [f for f in requested if f not in catalog_by_family] - if unknown: - raise _TuiError( - f"Unknown family in --families: {', '.join(unknown)}. " - f"Available: {', '.join(catalog_by_family)}") - family_keys: List[str] = [] - for family in requested: - if family not in family_keys: - family_keys.append(family) - chosen[family] = [opt for opt in package_dir_options( - catalog_by_family[family]) if opt["recommended"]] - else: - tree_families = _build_tree_families(catalog) - picked = tui.checkbox_tree( - stdscr, "Select TTS model families to host", - tree_families, expand_all=args.all_packages) - family_keys = [] - for family_index, option_key in picked: - family = catalog[family_index]["family"] - if family not in chosen: - chosen[family] = [] - family_keys.append(family) - chosen[family].append(option_key) - for family in list(chosen): - keyed = {opt["target_directory"]: opt - for opt in package_dir_options(catalog_by_family[family])} - chosen[family] = [keyed[key] for key in chosen[family]] - - # 4. Design task menus and duplicate-id renames. - def task_picker(install_id: str) -> str: - return tui.menu( - stdscr, f"How should the '{install_id}' package be hosted?", - [ - ("design (vdes) - describe the voice with --instructions", - TASK_VDES), - ("tts - normal synthesis", TASK_TTS), - ], default_index=0) - - def id_picker(display_name: str, install_id: str, default: str) -> str: - return tui.line_edit( - stdscr, - f"Server model id for {display_name} package '{install_id}'", - default) + if step == 1: + # Families and packages (flag or tree). Esc returns to the + # checkout browser (step 0). + chosen: Dict[str, List[dict]] = {} + if args.families is not None: + requested = [f.strip() for f in args.families.split(",") + if f.strip()] + unknown = [f for f in requested if f not in catalog_by_family] + if unknown: + raise _TuiError( + f"Unknown family in --families: {', '.join(unknown)}. " + f"Available: {', '.join(catalog_by_family)}") + family_keys: List[str] = [] + for family in requested: + if family not in family_keys: + family_keys.append(family) + chosen[family] = [opt for opt in package_dir_options( + catalog_by_family[family]) if opt["recommended"]] + else: + tree_families = _build_tree_families(catalog) + picked = tui.checkbox_tree( + stdscr, "Select TTS model families to host", + tree_families, expand_all=args.all_packages, + back_value=_GO_BACK) + if picked is _GO_BACK: + step = 0 + continue + family_keys = [] + for family_index, option_key in picked: + family = catalog[family_index]["family"] + if family not in chosen: + chosen[family] = [] + family_keys.append(family) + chosen[family].append(option_key) + for family in list(chosen): + keyed = {opt["target_directory"]: opt + for opt in package_dir_options( + catalog_by_family[family])} + chosen[family] = [keyed[key] for key in chosen[family]] + step = 2 + continue - model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ - _build_entries(family_keys, chosen, catalog_by_family, - task_picker, id_picker) + if step == 2: + # Design task menus and duplicate-id renames. Esc anywhere here + # falls back to the families tree (step 1). + def task_picker(install_id: str) -> str: + result = tui.menu( + stdscr, + f"How should the '{install_id}' package be hosted?", + [ + ("design (vdes) - describe the voice with " + "--instructions", TASK_VDES), + ("tts - normal synthesis", TASK_TTS), + ], default_index=0, back_value=_GO_BACK) + if result is _GO_BACK: + raise _GoBack() + return result + + def id_picker(display_name: str, install_id: str, + default: str) -> str: + result = tui.line_edit( + stdscr, + f"Server model id for {display_name} package " + f"'{install_id}'", default, back_value=_GO_BACK) + if result is _GO_BACK: + raise _GoBack() + return result + + try: + model_entries, entry_ids, install_guidance, \ + design_entry_ids, include_clone = _build_entries( + family_keys, chosen, catalog_by_family, + task_picker, id_picker) + except _GoBack: + step = 1 + continue + step = 3 + continue - # 5. Server settings. - host = args.host if args.host else tui.line_edit(stdscr, "Bind host", - DEFAULT_HOST) - if args.port is not None: - port = args.port - else: - port_text = tui.line_edit( - stdscr, "Port", str(config_port()), - validate=lambda s: None if (s.isdigit() and 1 <= int(s) <= 65535) - else "Enter a port number between 1 and 65535") - port = int(port_text) - sync_port: Optional[bool] = None - if port != config_port(): - sync_port = tui.confirm( - stdscr, f"Update AUDIOCPP_API_URL in converter/config.py to port " - f"{port} so audiobook.py talks to this server", default=True) - backend = args.backend if args.backend else tui.menu( - stdscr, "Which inference backend was audiocpp_server built for?", - [ - ("cuda - NVIDIA GPUs (fastest)", "cuda"), - ("vulkan - cross-vendor GPU", "vulkan"), - ("hip - AMD GPUs", "hip"), - ("cpu - no GPU required", "cpu"), - ], default_index=0) - default_lazy = len(model_entries) > 1 - lazy_load = args.lazy_load or tui.confirm( - stdscr, "Load models lazily (on first use instead of at startup)", - default=default_lazy) - - # 6. Wav directory (flag, browsed when cloning, else skipped). - if args.input_dir is not None: - wav_dir = args.input_dir - elif include_clone: - wav_dir = tui.browse_directory( - stdscr, "Directory with .wav voice cloning files", - start=Path.cwd()) - else: - wav_dir = None + if step == 3: + # Server settings (host, port, port-sync, backend, lazy). Esc on + # any of them falls back to the previous group (step 2). + if args.host: + host = args.host + else: + host = tui.line_edit( + stdscr, "Bind host", DEFAULT_HOST, + help_lines=["The IP address audiocpp will be hosted on", + "127.0.0.1 (this machine) is probably " + "correct"], back_value=_GO_BACK) + if host is _GO_BACK: + step = 2 + continue + if args.port is not None: + port = args.port + else: + port_text = tui.line_edit( + stdscr, "Port", str(config_port()), + validate=lambda s: None if (s.isdigit() + and 1 <= int(s) <= 65535) + else "Enter a port number between 1 and 65535", + help_lines=["The port audiocpp will be hosted on"], + back_value=_GO_BACK) + if port_text is _GO_BACK: + step = 2 + continue + port = int(port_text) + sync_port: Optional[bool] = None + if port != config_port(): + sync_port = tui.confirm( + stdscr, f"Update AUDIOCPP_API_URL in converter/config.py " + f"to port {port} so audiobook.py talks to this server", + default=True, cancel_value=_GO_BACK) + if sync_port is _GO_BACK: + step = 2 + continue + if args.backend: + backend = args.backend + else: + backend_options, backend_default = \ + _backend_options(detected_backend) + backend = tui.menu( + stdscr, "Which inference backend was audiocpp_server " + "built for?", backend_options, + default_index=backend_default, back_value=_GO_BACK) + if backend is _GO_BACK: + step = 2 + continue + default_lazy = len(model_entries) > 1 + if args.lazy_load: + lazy_load = True + else: + lazy_load = tui.confirm( + stdscr, "Load models lazily (on first use instead of at " + "startup)", default=default_lazy, cancel_value=_GO_BACK) + if lazy_load is _GO_BACK: + step = 2 + continue + step = 4 + continue - # 7. Transcription plan (questions only; transcription runs after). - plan: Optional[dict] = None - if include_clone and wav_dir is not None: - wav_files = find_wav_files(wav_dir) - if wav_files: - prompt_path = wav_dir / PROMPT_TEXT_FILENAME - existing = read_prompt_text(prompt_path) if ( - prompt_path.exists() and not args.force) else {} - plan = _decide_transcription( - wav_files, existing, prompt_path.exists(), args.force, - lambda question, default: tui.confirm(stdscr, question, default)) - - # 8. Single-model id sync decision. - sync_model_ids: Optional[bool] = None - if len(entry_ids) == 1 and not ( - config.AUDIOCPP_MODEL_ID == entry_ids[0] - and config.AUDIOCPP_CLONE_MODEL_ID == entry_ids[0]): - sync_model_ids = tui.confirm( - stdscr, "Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID in " - f"converter/config.py to '{entry_ids[0]}' so audiobook.py uses " - "this model", default=True) - - # 9. Summary and final confirmation. - summary_lines = [ - f"Output: {output_path}", - f"Server: {host}:{port} ({backend}, lazy_load={'on' if lazy_load else 'off'})", - f"Models: {', '.join(entry_ids)}", - ] - if wav_dir is not None: - summary_lines.append(f"Voices: {wav_dir}") - if not tui.confirm(stdscr, "Generate server.json?", default=True, - body=summary_lines): - return None + if step == 4: + # Wav directory (flag, browsed when cloning, else skipped). Esc + # falls back to the server settings (step 3). + if args.input_dir is not None: + wav_dir = args.input_dir + elif include_clone: + wav_start = detect_wav_dir(audiocpp_dir, TTS_ROOT) + wav_dir = tui.browse_directory( + stdscr, "Select the directory with your .wav voices", + info=_wav_dir_info, preview=_wav_dir_preview, + start=wav_start if wav_start is not None else Path.cwd(), + back_value=_GO_BACK) + if wav_dir is _GO_BACK: + step = 3 + continue + else: + wav_dir = None + step = 5 + continue - return { - "audiocpp_dir": audiocpp_dir, - "catalog": catalog, - "catalog_by_family": catalog_by_family, - "output_path": output_path, - "family_keys": family_keys, - "chosen": chosen, - "model_entries": model_entries, - "entry_ids": entry_ids, - "install_guidance": install_guidance, - "design_entry_ids": design_entry_ids, - "include_clone": include_clone, - "host": host, - "port": port, - "backend": backend, - "lazy_load": lazy_load, - "sync_port": sync_port, - "sync_model_ids": sync_model_ids, - "wav_dir": wav_dir, - "plan": plan, - } + if step == 5: + # Transcription plan (questions only; transcription runs after). + # Esc falls back to the wav browser (step 4). + plan: Optional[dict] = None + if include_clone and wav_dir is not None: + wav_files = find_wav_files(wav_dir) + if wav_files: + prompt_path = wav_dir / PROMPT_TEXT_FILENAME + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + try: + plan = _decide_transcription( + wav_files, existing, prompt_path.exists(), + args.force, ask_confirm) + except _GoBack: + step = 4 + continue + step = 6 + continue + + if step == 6: + # Single-model id sync decision. Esc falls back to the + # transcription plan (step 5). + sync_model_ids: Optional[bool] = None + if len(entry_ids) == 1 and not ( + config.AUDIOCPP_MODEL_ID == entry_ids[0] + and config.AUDIOCPP_CLONE_MODEL_ID == entry_ids[0]): + sync_model_ids = tui.confirm( + stdscr, "Update AUDIOCPP_MODEL_ID and " + "AUDIOCPP_CLONE_MODEL_ID in converter/config.py to " + f"'{entry_ids[0]}' so audiobook.py uses this model", + default=True, cancel_value=_GO_BACK) + if sync_model_ids is _GO_BACK: + step = 5 + continue + step = 8 + continue + + if step == 8: + # Automatic model download (or print the install commands). Esc + # falls back to the model-id sync (step 6). + try: + download = _decide_download(audiocpp_dir, ask_confirm) + except _GoBack: + step = 6 + continue + return { + "audiocpp_dir": audiocpp_dir, + "catalog": catalog, + "catalog_by_family": catalog_by_family, + "output_path": output_path, + "family_keys": family_keys, + "chosen": chosen, + "model_entries": model_entries, + "entry_ids": entry_ids, + "install_guidance": install_guidance, + "design_entry_ids": design_entry_ids, + "include_clone": include_clone, + "host": host, + "port": port, + "backend": backend, + "lazy_load": lazy_load, + "sync_port": sync_port, + "sync_model_ids": sync_model_ids, + "wav_dir": wav_dir, + "plan": plan, + "download": download, + } def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: @@ -1174,6 +1597,10 @@ def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: except tui.WizardCancelled: print("\n[INFO] Cancelled; nothing was written") return 1 + try: + curses.curs_set(1) # restore the text cursor hidden by the TUI + except curses.error: + pass if settings is None: print("[INFO] Aborted; existing server.json kept") return 1 @@ -1196,18 +1623,17 @@ def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: transcripts, write_prompt = {}, False _write_and_advise( - settings["wav_dir"], settings["output_path"], settings["model_entries"], - settings["entry_ids"], settings["install_guidance"], - settings["design_entry_ids"], settings["family_keys"], - settings["catalog_by_family"], settings["host"], settings["port"], - settings["backend"], settings["lazy_load"], transcripts, write_prompt) + settings["audiocpp_dir"], settings["wav_dir"], settings["output_path"], + settings["model_entries"], settings["install_guidance"], + settings["host"], settings["port"], settings["backend"], + settings["lazy_load"], transcripts, write_prompt) if len(settings["entry_ids"]) == 1: _offer_config_model_id_sync(settings["entry_ids"][0], settings["sync_model_ids"]) - elif len(settings["entry_ids"]) > 1: - _print_multi_model_model_id_note(settings["entry_ids"]) print_empty_transcript_warning(transcripts) + _install_models(settings["audiocpp_dir"], settings["install_guidance"], + settings["download"]) return 0 @@ -1250,7 +1676,8 @@ def main() -> int: "AUDIOCPP_API_URL from converter/config.py)") parser.add_argument("--backend", choices=BACKENDS, default=None, help="Inference backend audiocpp_server was built " - "for (default: cuda)") + "for (default: auto-detected from the checkout's " + "build/ directory, else cuda)") parser.add_argument("--lazy-load", action="store_true", help="Load models on first use instead of at startup " "(default: on when more than one model is hosted)") @@ -1279,9 +1706,16 @@ def main() -> int: # ---- Line-prompt flow (original behaviour). --------------------------- - # Resolve the wav directory (flag, else prompt). + # Resolve the wav directory (flag, else prompt). The prompt default is + # the unique directory that directly contains .wav files across the + # audio.cpp checkout (best-effort detected here) and the + # tts-audiobook-generator root, so the user usually just presses Enter. if args.input_dir is None: - answer = ask("Directory with .wav reference files", "") + tentative_checkout = args.audiocpp_dir or detect_audiocpp_dir() + wav_start = detect_wav_dir(tentative_checkout, TTS_ROOT) \ + if tentative_checkout is not None else None + default = str(wav_start) if wav_start is not None else "" + answer = ask("Directory with .wav reference files", default) args.input_dir = resolve_wav_dir_arg(answer) if answer else None if args.input_dir is None: parser.error("--wavs is required: a directory containing the .wav " @@ -1312,6 +1746,11 @@ def main() -> int: audiocpp_dir = audiocpp_dir.resolve() if not audiocpp_dir.is_dir(): parser.error(f"audio.cpp checkout not found: {audiocpp_dir}") + root = _resolve_audiocpp_root(audiocpp_dir) + if root is None: + parser.error(f"{audiocpp_dir} has no model_specs/ directory; point " + "--audiocpp-dir at the root of an audio.cpp checkout") + audiocpp_dir = root try: catalog = load_model_catalog(audiocpp_dir) except NotADirectoryError as exc: @@ -1377,20 +1816,24 @@ def main() -> int: # single-entry server loads at startup, while a multi-entry server avoids # loading every model until it is actually used. default_lazy = len(model_entries) > 1 - host, port, backend, lazy_load = _ask_host_port_backend_lazy(args, default_lazy) + detected_backend = detect_backend(audiocpp_dir) + host, port, backend, lazy_load = _ask_host_port_backend_lazy( + args, default_lazy, detected_backend) transcripts, write_prompt = _transcribe(args, include_clone) _write_and_advise( - args.input_dir, output_path, model_entries, entry_ids, - install_guidance, design_entry_ids, family_keys, catalog_by_family, - host, port, backend, lazy_load, transcripts, write_prompt) + audiocpp_dir, args.input_dir, output_path, model_entries, + install_guidance, host, port, backend, lazy_load, transcripts, + write_prompt) if len(entry_ids) == 1: _offer_config_model_id_sync(entry_ids[0]) - elif len(entry_ids) > 1: - _print_multi_model_model_id_note(entry_ids) print_empty_transcript_warning(transcripts) + + download = _decide_download( + audiocpp_dir, lambda question, default: ask_bool(question, default)) + _install_models(audiocpp_dir, install_guidance, download) return 0 diff --git a/tools/tui.py b/tools/tui.py index e1eda05..906aec5 100644 --- a/tools/tui.py +++ b/tools/tui.py @@ -1,22 +1,38 @@ #!/usr/bin/env python3 -"""Minimal curses TUI widgets for the interactive tools. - -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. +"""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 - Esc abort the whole wizard (raises WizardCancelled) + 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; inside text editors it is an ordinary character. +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 os +import textwrap from pathlib import Path from typing import Callable, List, Optional, Sequence, Tuple @@ -28,6 +44,82 @@ class WizardCancelled(Exception): """Raised when the user presses Esc to abort the wizard.""" +# 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 # --------------------------------------------------------------------------- @@ -40,6 +132,23 @@ def _addstr(scr, y: int, x: int, text: str, attr: int = 0) -> None: 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: @@ -50,75 +159,306 @@ def _fit(text: str, width: int) -> str: class Frame: - """A screen frame: title, scrolling body rows, message and footer. - - Widgets append styled body rows via mark(), call draw() after every - state change, and read keys through get_key()/edit_line(). + """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.message = "" # transient status line - self.message_attr = None # None -> bold reverse video - self.rows: List[dict] = [] # {text, attr, indent} + 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.cursor = 0 # highlighted row index + 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). - def mark(self, text: str, attr: int = 0, indent: int = 0) -> None: - self.rows.append({"text": text, "attr": attr, "indent": indent}) + 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: - curses = self.curses scr = self.scr scr.erase() height, width = scr.getmaxyx() - if height < 6 or width < 20: - _addstr(scr, 0, 0, _fit("Terminal too small", width - 1), - curses.A_BOLD) + 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 - top = 2 - visible = height - 3 - top - if visible < 1: - visible = 1 - # Keep the cursor inside the viewport. - if self.cursor < self.scroll: - self.scroll = self.cursor - elif self.cursor >= self.scroll + visible: - self.scroll = self.cursor - visible + 1 - if self.scroll + visible > len(self.rows): - self.scroll = max(0, len(self.rows) - visible) - scrolling = len(self.rows) > visible - indicator = f" {self.cursor + 1}/{len(self.rows)} " if scrolling else "" - title_width = width - 1 - (len(indicator) if indicator else 0) - _addstr(scr, 0, 0, _fit(self.title, title_width), - curses.A_BOLD | curses.A_UNDERLINE) - for index in range(self.scroll, - min(len(self.rows), self.scroll + visible)): - row = self.rows[index] - line = " " * row["indent"] + row["text"] - attr = row["attr"] - if index == self.cursor: - attr |= curses.A_REVERSE - _addstr(scr, top + index - self.scroll, 0, - _fit(line, width - 1), attr) - if indicator: - _addstr(scr, 0, max(0, width - len(indicator)), indicator, - curses.A_DIM) - if self.message: - attr = self.message_attr - if attr is None: - attr = curses.A_BOLD | curses.A_REVERSE - _addstr(scr, height - 2, 0, _fit(self.message, width - 1), attr) - _addstr(scr, height - 1, 0, _fit(self.footer, width - 1), curses.A_DIM) + 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: @@ -131,31 +471,38 @@ class Frame: raise WizardCancelled() return key - def edit_line(self, start: str, prompt: str = "" - ) -> Optional[str]: - """Run an inline editor on the message line. + 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 = start + text = "" while True: - height, width = self.scr.getmaxyx() - self.message = "" + self.status = (f"{prompt}{text}_", "input") self.draw() - room = max(1, width - 2 - len(prompt)) - shown = text if len(text) < room else ">" + text[-(room - 2):] - _addstr(self.scr, height - 2, 0, - _fit(f"{prompt}{shown}_", width - 1), curses.A_BOLD) - self.scr.refresh() try: key = self.scr.getch() except KeyboardInterrupt: raise WizardCancelled() from None if key == 27: return None - if key in (10, 13): # Enter + 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] @@ -164,63 +511,89 @@ class Frame: # --------------------------------------------------------------------------- -# Widget: yes/no confirm +# Widget: yes/no confirm with buttons # --------------------------------------------------------------------------- def confirm(scr, question: str, default: bool = False, - body: Optional[Sequence[str]] = None) -> bool: - """Ask a yes/no QUESTION; Enter takes DEFAULT, Esc aborts. - - BODY lines are shown above the question (a summary, for example). + 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, - "y = yes n = no Enter = default Esc = cancel") - cancel = (27, ord("q")) + "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) - if body: - frame.mark("") - hint = "[Y/n]" if default else "[y/N]" - frame.mark(f"{question} {hint}") - frame.cursor = len(frame.rows) - 1 + frame.cursor = None + frame.buttons = (["Yes", "No"], index) frame.draw() - key = frame.get_key(cancel) - if key in (ord("y"), ord("Y")): + 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 - if key in (ord("n"), ord("N")): + elif key in (ord("n"), ord("N")): return False - if key in (10, 13): - return default + elif key in (10, 13): + return index == 0 # --------------------------------------------------------------------------- # Widget: single-choice menu # --------------------------------------------------------------------------- -def menu(scr, title: str, options: Sequence[tuple], default_index: int = 0): +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. + 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") - cancel = (27, ord("q")) 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) - frame.cursor = cursor + frame.mark(label, selectable=True, align="left") + frame.cursor = base + cursor frame.draw() - curses = frame.curses - key = frame.get_key(cancel) - if key in (curses.KEY_UP, ord("k")): - cursor = (cursor - 1) % len(options) - elif key in (curses.KEY_DOWN, ord("j")): - cursor = (cursor + 1) % len(options) + 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] @@ -230,38 +603,49 @@ def menu(scr, title: str, options: Sequence[tuple], default_index: int = 0): # --------------------------------------------------------------------------- def line_edit(scr, title: str, default: str, - validate: Optional[Callable[[str], Optional[str]]] = None - ) -> 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. - VALIDATE receives the entered string and returns an error message or - None; Enter on an invalid value shows the message and keeps editing. - Esc aborts the wizard ('q' is an ordinary character here). + 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 = "" + 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.cursor = 1 - frame.message = error + 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() # Esc only; 'q' must stay typeable + 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 - error = f"{error} (edit, then Enter)" 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) @@ -282,72 +666,144 @@ def _list_dirs(path: Path) -> List[Path]: def browse_directory(scr, title: str, validate: Optional[Callable[[Path], Optional[str]]] = None, - start: Optional[Path] = 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; Enter accepts the directory being listed. - - Right (or l) descends into the highlighted entry, Left/Backspace/u - goes to the parent, and e edits the path directly. VALIDATE receives - the listed directory and returns an error message or None; Enter on - an invalid directory is refused with that message. Esc aborts the - wizard. + """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 Right = open Left = parent e = edit " - "path Enter = choose this directory Esc = cancel") + footer = ("Up/Down = move Enter = open/use Left = parent " + "e = type path Esc = cancel") frame = Frame(scr, title, footer) - cancel = (27, ord("q")) current = Path(start) if start is not None else Path.cwd() try: current = current.resolve() except OSError: current = Path.cwd() - cursor = 0 + sel = 0 + highlight: Optional[Path] = None def validation_error() -> Optional[str]: if validate is None: return None - return validate(current) + 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) - cursor = max(0, min(cursor, max(0, len(entries) - 1))) + has_parent = current.parent != current + offset = 1 + (1 if has_parent else 0) frame.rows = [] - frame.mark(f"Directory: {current}", frame.curses.A_BOLD) - error = validation_error() - if error is None: - frame.mark(" This directory is a valid choice. Press Enter.", - frame.curses.A_DIM) - else: - frame.mark(f" {error}", frame.curses.A_BOLD) + 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("") - if not entries: - frame.mark(" (no subdirectories)") + 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}/") - header = 3 # directory line, validity line, blank separator - frame.cursor = header + (cursor if entries else 0) - frame.message = "" + 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) - if key in (curses.KEY_UP, ord("k")): - cursor = max(0, cursor - 1) - elif key in (curses.KEY_DOWN, ord("j")): - if entries: - cursor = min(len(entries) - 1, cursor + 1) - elif key in (curses.KEY_RIGHT, ord("l")): - if entries: - current = entries[cursor] - cursor = 0 + 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): - parent = current.parent - if parent != current: - current = parent - cursor = 0 + if has_parent: + highlight = current + current = current.parent elif key == ord("e"): - result = frame.edit_line("", prompt="path: ") - if result is not None: + result = frame.edit_status(prompt="path: ") + if result: candidate = Path(os.path.expanduser(result)) if not candidate.is_absolute(): candidate = current / candidate @@ -357,19 +813,9 @@ def browse_directory(scr, title: str, pass if candidate.is_dir(): current = candidate - cursor = 0 + sel = 0 else: - frame.message = f"Not a directory: {candidate}" - frame.draw() - frame.get_key(cancel) - frame.get_key(cancel) - elif key in (10, 13): # Enter: accept the listed directory - error = validation_error() - if error is None: - return current - frame.message = f"{error} (keep browsing)" - frame.draw() - frame.get_key(cancel) + frame.flash(f"Not a directory: {candidate}", "err") # --------------------------------------------------------------------------- @@ -378,7 +824,8 @@ def browse_directory(scr, title: str, def checkbox_tree(scr, title: str, families: List[dict], footer: Optional[str] = None, - expand_all: bool = False) -> List[Tuple[int, str]]: + 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:: @@ -398,27 +845,23 @@ def checkbox_tree(scr, title: str, families: List[dict], 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. The first family's recommended option starts - checked (the prompt flow's default), and with EXPAND_ALL every - family starts expanded. + 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) - cancel = (27, ord("q")) expanded = {index for index in range(len(families))} if expand_all else set() checked = set() # (family_index, option_key) - if families: - expanded.add(0) - first = families[0]["options"] - for option in first: - if option.get("recommended"): - checked.add((0, option["key"])) - break - else: - if first: - checked.add((0, first[0]["key"])) + expanded.add(0) def family_checked(index: int) -> bool: return any(pair[0] == index for pair in checked) @@ -447,29 +890,43 @@ def checkbox_tree(scr, title: str, families: List[dict], if node[0] == "family": index = node[1] family = families[index] - mark = "x" if family_checked(index) else " " + on = family_checked(index) + mark = "x" if on else " " arrow = "-" if index in expanded else "+" - attr = frame.curses.A_BOLD if family_checked(index) else 0 - frame.mark(f"[{mark}] {arrow} {family['label']}", attr) + 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 " " - note = " [recommended]" if option.get("recommended") else "" - frame.mark(f" [{mark}] {option['label']}{note}") + 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.message = families[node[1]].get("detail", "") - frame.message_attr = frame.curses.A_DIM + frame.status = (families[node[1]].get("detail", ""), "info") frame.draw() curses = frame.curses - key = frame.get_key(cancel) - if key in (curses.KEY_UP, ord("k")): - cursor = (cursor - 1) % len(nodes) - elif key in (curses.KEY_DOWN, ord("j")): - cursor = (cursor + 1) % len(nodes) + 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: @@ -504,8 +961,4 @@ def checkbox_tree(scr, title: str, families: List[dict], selection = accept() if selection: return selection - frame.message = "Check at least one model package (Space)" - frame.message_attr = None - frame.draw() - frame.get_key(cancel) - frame.message_attr = frame.curses.A_DIM + frame.flash("Check at least one model package (Space)", "err") |