diff options
Diffstat (limited to 'tools')
| -rwxr-xr-x | tools/make_audiocpp_server_json.py | 1097 | ||||
| -rw-r--r-- | tools/tui.py | 511 |
2 files changed, 1327 insertions, 281 deletions
diff --git a/tools/make_audiocpp_server_json.py b/tools/make_audiocpp_server_json.py index bda50e3..fb16a43 100755 --- a/tools/make_audiocpp_server_json.py +++ b/tools/make_audiocpp_server_json.py @@ -2,36 +2,61 @@ """Interactively generate a server.json for the audio.cpp audiocpp_server. Reads the model catalog (``model_specs/*.json``) from a local audio.cpp -checkout and offers every TTS model family audio.cpp supports as a -multi-select checklist, so one server.json can host several lazily-loaded -model entries at once. The 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. - -Cloning reference .wav files (the required WAV_DIR argument) are transcribed -with a local Whisper backend (faster_whisper or whisper) and published as a -server-level ``voice_dir`` plus a ``prompt_text`` mapping file written into -WAV_DIR, so every hosted clone-capable family can use them with ``--voice``. - -Every value can also be supplied as a command-line flag; anything missing is -asked interactively with the default shown in brackets. Pressing Enter accepts -the default (the Qwen3-TTS built-in-speakers + voice-cloning flow). +checkout and offers every TTS model family audio.cpp supports, so one +server.json can host several lazily-loaded model entries at once. The +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 +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. + +Each family is hosted through its recommended package by default; the TUI +tree always lists every installable package (distinct ``target_directory`` +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". + +Cloning reference .wav files (``--wavs DIR``) are transcribed with a local +Whisper backend (faster_whisper or whisper) and published as a server-level +``voice_dir`` plus a ``prompt_text`` mapping file written into the wav +directory, so every hosted clone-capable family can use them with +``--voice``. If ``prompt_text`` already exists, only voices that are missing +(or have an empty transcript) are re-transcribed, and you are asked first +when everything is already transcribed or when a mix of existing and new +voices is detected. Transcription runs in the plain console after the TUI +has gathered every setting. Usage: - python tools/make_audiocpp_server_json.py WAV_DIR [--output PATH] - [--audiocpp-dir PATH] [--families FAM1,FAM2] - [--models {both,custom,clone}] [--host HOST] [--port PORT] + python tools/make_audiocpp_server_json.py [--wavs WAV_DIR] + [--output PATH] [--audiocpp-dir PATH] [--families FAM1,FAM2] + [--all-packages] [--host HOST] [--port PORT] [--backend {cuda,vulkan,hip,cpu}] [--lazy-load] - [--whisper-model NAME] [--force] + [--whisper-model NAME] [--force] [--notui] -WAV_DIR is required: a directory of .wav reference files used as voice -cloning presets. It is checked up front and reported with its resolved -absolute path if it does not exist. +--wavs is the directory of .wav reference files used as voice cloning +presets; when omitted it is asked for. It is checked up front and reported +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. --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. +checkout must contain a ``model_specs/`` directory. A leading ``~`` in a +path argument or prompt answer is expanded. """ import argparse @@ -41,7 +66,7 @@ import re import sys import urllib.parse from pathlib import Path -from typing import Dict, List, Optional, Tuple +from typing import Callable, Dict, List, Optional, Set, Tuple # Allow running from any working directory. sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) @@ -51,15 +76,16 @@ from converter.tts import transcribe_reference_audio, whisper_backend_available DEFAULT_HOST = "127.0.0.1" FALLBACK_PORT = 8080 -DEFAULT_CUSTOM_VOICE_PATH = "models/Qwen3-TTS-12Hz-1.7B-CustomVoice-GGUF" -DEFAULT_BASE_PATH = "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF" CONFIG_PATH = Path(__file__).resolve().parent.parent / "converter" / "config.py" -MODEL_SELECTIONS = ("both", "custom", "clone") BACKENDS = ("cuda", "vulkan", "hip", "cpu") -FAMILY_QWEN3_TTS = "qwen3_tts" PROMPT_TEXT_FILENAME = "prompt_text" +TASK_TTS = "tts" +TASK_VDES = "vdes" + +# 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 @@ -81,8 +107,46 @@ PREFERRED_IDS = { } -def resolve_wav_dir_arg(value: str) -> Path: - """Normalize a user-supplied wav directory argument. +class _TuiError(Exception): + """A fatal error raised from inside the TUI wizard. + + The message is reported to stderr after the terminal is restored; the + process exits with code 2 (matching a parser error). + """ + + +def _curses_importable() -> bool: + """Return True when the curses module can be imported.""" + try: + import curses # noqa: F401 + return True + except ImportError: + return False + + +def _load_tui(): + """Import the TUI widget module (tools/tui.py).""" + try: + from tools import tui + except ImportError: # executed directly from the tools/ directory + import tui + return tui + + +def _tui_enabled(args: argparse.Namespace) -> bool: + """Decide whether to run the TUI or fall back to line prompts.""" + if args.notui: + return False + if not _curses_importable(): + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +def normalize_dir_arg(value: str) -> Path: + """Normalize a user-supplied path argument. Strips surrounding quotes (a common copy-paste artifact), expands a leading ``~``, and resolves the result to an absolute path so relative @@ -94,6 +158,11 @@ def resolve_wav_dir_arg(value: str) -> Path: return Path(os.path.expanduser(cleaned)).resolve() +def resolve_wav_dir_arg(value: str) -> Path: + """Normalize a user-supplied wav directory argument.""" + return normalize_dir_arg(value) + + def find_wav_files(input_dir: Path) -> list: """Return the .wav files in INPUT_DIR, sorted alphabetically by name.""" return sorted( @@ -103,21 +172,6 @@ def find_wav_files(input_dir: Path) -> list: ) -def prompt_overwrite(output_path: Path) -> bool: - """Ask whether to overwrite an existing output file.""" - while True: - try: - answer = input(f"{output_path} already exists. Overwrite? (y/n): ").strip().lower() - except EOFError: - print("\n[WARNING] No interactive input available; keeping existing file") - return False - if answer in ("y", "yes"): - return True - if answer in ("n", "no"): - return False - print("Please answer 'y' or 'n'.") - - 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 "" @@ -180,14 +234,37 @@ def ask_menu(title: str, options: list, default_index: int = 1) -> str: print(f"Please enter a number between 1 and {len(options)}.") -def ask_models() -> str: - return ask_menu( - "Which Qwen3-TTS models should the server host?", - [ - ("Both (recommended) - built-in speakers + voice cloning", "both"), - ("CustomVoice only - built-in speakers", "custom"), - ("Base only - voice cloning (converting then requires --voice)", "clone"), - ]) +def ask_checklist(title: str, options: list, default: Set[str]) -> Set[str]: + """Show a numbered multi-select checklist and return the chosen values. + + Input is comma/space-separated numbers; Enter or EOF selects every option + in DEFAULT. At least one option is required. + """ + print(title) + for number, (label, _) in enumerate(options, 1): + print(f" {number}) {label}") + default_numbers = [str(number) for number, (_, value) in enumerate(options, 1) + if value in default] + suffix = f" [{', '.join(default_numbers)}]" + while True: + try: + answer = input(f"Choice{suffix}: ").strip() + except EOFError: + return set(default) + if not answer: + return set(default) + parts = [p for p in re.split(r"[,\s]+", answer) if p] + indices: List[int] = [] + valid = True + for part in parts: + if part.isdigit() and 1 <= int(part) <= len(options): + indices.append(int(part)) + else: + valid = False + break + if valid and indices: + return {options[index - 1][1] for index in indices} + print(f"Please enter comma-separated numbers between 1 and {len(options)}.") def ask_backend() -> str: @@ -201,25 +278,6 @@ def ask_backend() -> str: ]) -def ask_distinct_clone_id(primary_id: str) -> str: - """Prompt until a non-empty id different from PRIMARY_ID is entered.""" - prompt = (f"Enter a new id for the cloning (Base) model " - f"(must differ from '{primary_id}'): ") - while True: - try: - answer = input(prompt).strip() - except EOFError: - print() - sys.exit("[FATAL] No interactive input available to resolve the " - "duplicate model id; give the two models distinct " - "AUDIOCPP_MODEL_ID / AUDIOCPP_CLONE_MODEL_ID values in " - "converter/config.py first") - if answer and answer != primary_id: - return answer - print(f"[WARNING] The id must be unique; it cannot be empty or " - f"equal to '{primary_id}'.") - - def config_port() -> int: """Return the port of AUDIOCPP_API_URL in converter/config.py.""" try: @@ -311,7 +369,7 @@ def detect_audiocpp_dir() -> Optional[Path]: candidates: List[Path] = [] env_dir = os.environ.get("AUDIOCPP_DIR") if env_dir: - candidates.append(Path(env_dir)) + candidates.append(Path(os.path.expanduser(env_dir))) cwd = Path.cwd() candidates.append(cwd / "audio.cpp") candidates.append(cwd.parent / "audio.cpp") @@ -326,14 +384,12 @@ def detect_audiocpp_dir() -> Optional[Path]: return None -def _default_package(spec: dict) -> Optional[dict]: - """Pick the default installable package from a model spec. +def _default_package(packages: List[dict]) -> Optional[dict]: + """Pick the default package from a list of packages. Prefers the package flagged ``default: true``, then the first GGUF - package, then the first package overall. Returns None if the spec - declares no packages. + package, then the first package overall. Returns None for an empty list. """ - packages = spec.get("packages") or [] if not packages: return None for package in packages: @@ -349,10 +405,10 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: """Read model_specs/*.json and return the TTS-capable families. Each returned entry has: family, display_name, description, languages, - clone_capable, install_id (default 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. + 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. """ specs_dir = audiocpp_dir / "model_specs" if not specs_dir.is_dir(): @@ -369,7 +425,8 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: if "tts" not in tasks and spec.get("category") != "tts": continue family = spec.get("family") or spec_path.stem - package = _default_package(spec) + packages = spec.get("packages") or [] + package = _default_package(packages) if package is None: # No installable package: skip (cannot be hosted from a path). continue @@ -382,7 +439,9 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: "display_name": display_name, "description": description, "languages": languages, + "tasks": list(tasks), "clone_capable": "clone" in tasks, + "packages": packages, "install_id": package.get("id") or family, "default_path": f"models/{target_directory}", "tested": family in TESTED_FAMILIES, @@ -399,26 +458,118 @@ def load_model_catalog(audiocpp_dir: Path) -> List[dict]: return entries +def is_design_package(package: dict) -> bool: + """Return True when a package's name marks it a voice-design model. + + audio.cpp voice-design packages (whose id, display name, or target + directory mentions "voice design") are the only packages that must be + hosted with task "vdes"; their role is not in the schema, only in those + strings, so it is detected from them. + """ + text = " ".join(str(package.get(key, "")) + for key in ("id", "display_name", "target_directory")) + return bool(DESIGN_PACKAGE_RE.search(text)) + + +def package_dir_options(entry: dict) -> List[dict]: + """Return one option per distinct target_directory of a family's packages. + + Each option is a dict with: target_directory, install_id (the recommended + package id inside that directory), design (voice-design package flag), and + recommended (whether it holds the family's default package). Precisions + that share a directory (q8_0/bf16/...) collapse to a single option. + """ + packages = entry.get("packages") or [] + default_pkg = _default_package(packages) + default_dir = (default_pkg or {}).get("target_directory") or entry["family"] + by_dir: Dict[str, List[dict]] = {} + order: List[str] = [] + for package in packages: + directory = package.get("target_directory") or entry["family"] + if directory not in by_dir: + by_dir[directory] = [] + order.append(directory) + by_dir[directory].append(package) + options: List[dict] = [] + for directory in order: + package = _default_package(by_dir[directory]) + options.append({ + "target_directory": directory, + "install_id": (package or {}).get("id") or directory, + "design": is_design_package(package or {}), + "recommended": directory == default_dir, + }) + # Put the recommended package first for a friendlier checklist. + options.sort(key=lambda opt: not opt["recommended"]) + return options + + +def ask_package_dirs(entry: dict) -> List[dict]: + """Choose which of a family's packages to host (multi-select checklist). + + Enter selects the recommended package only, matching the default flow. + """ + options = package_dir_options(entry) + if len(options) <= 1: + return options + default = {opt["target_directory"] for opt in options if opt["recommended"]} + labels = [] + for opt in options: + marker = " [recommended]" if opt["recommended"] else "" + labels.append((f"{opt['install_id']} -> {opt['target_directory']}{marker}", + opt["target_directory"])) + chosen = ask_checklist( + f"Which {entry['display_name']} packages should the server host?", + labels, default=default) + return [opt for opt in options if opt["target_directory"] in chosen] + + +def ask_package_task(install_id: str) -> str: + """Ask how to host a voice-design package: vdes or tts.""" + return ask_menu( + 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=1) + + def ask_families(catalog: List[dict]) -> List[str]: - """Show a numbered checklist and return the chosen family keys. + """Show a numbered table and return the chosen family keys. Input is comma/space-separated numbers; Enter alone selects the first - entry (the default Qwen3-TTS flow). At least one family is required. + entry. At least one family is required. """ + rows: List[Tuple[str, str]] = [] + for entry in catalog: + capabilities = ["tts"] + if "clone" in entry["tasks"]: + capabilities.append("cloning") + if "design" in entry["tasks"]: + capabilities.append("design") + name = entry["display_name"] + if name != entry["family"]: + name = f"{name} ({entry['family']})" + rows.append((name, ", ".join(capabilities))) + number_width = len(str(len(rows))) + name_width = max([len("Model family")] + [len(name) for name, _ in rows]) + tasks_width = max([len("Tasks")] + [len(tasks) for _, tasks in rows]) + header = (f"{'#'.ljust(number_width)} | " + f"{'Model family'.ljust(name_width)} | " + f"{'Tasks'.ljust(tasks_width)}") + divider = (f"{'-' * number_width}-+-" + f"{'-' * name_width}-+-" + f"{'-' * tasks_width}") print("Select TTS model families to host (comma-separated numbers,") - print("or press Enter for the default Qwen3-TTS flow):") - for number, entry in enumerate(catalog, 1): - marker = " [tested with this converter]" if entry["tested"] else "" - langs = entry["languages"] - lang_text = ", ".join(langs[:6]) + ("..." if len(langs) > 6 else "") - if entry["family"] == FAMILY_QWEN3_TTS: - caps = "built-in speakers + voice cloning" - elif entry["clone_capable"]: - caps = "voice cloning" - else: - caps = "TTS (no cloning)" - detail = f"({lang_text}; {caps})" if lang_text else f"({caps})" - print(f" {number}) {entry['display_name']}{marker} {detail}") + print("or press Enter for the first family):") + print(header) + print(divider) + for number, (name, tasks) in enumerate(rows, 1): + print(f"{str(number).ljust(number_width)} | " + f"{name.ljust(name_width)} | " + f"{tasks.ljust(tasks_width)}") while True: try: answer = input("Choice [1]: ").strip() @@ -447,13 +598,19 @@ def ask_families(catalog: List[dict]) -> List[str]: print(f"Please enter comma-separated numbers between 1 and {len(catalog)}.") -def build_model_entry(family: str, model_id: str, model_path: str) -> dict: - """Assemble one server.json model entry.""" +def build_model_entry(family: str, model_id: str, model_path: str, + task: str = TASK_TTS) -> dict: + """Assemble one server.json model entry. + + ``task`` defaults to "tts"; voice design packages are hosted with + "vdes" so the server runs its design session for speech requests + (audiobook.py then requires --instructions with that entry). + """ return { "id": model_id, "family": family, "path": model_path, - "task": "tts", + "task": task, "mode": "offline", } @@ -495,6 +652,27 @@ def transcribe_wav_dir(wav_files: list, whisper_model: str) -> Dict[str, str]: return transcripts +def read_prompt_text(prompt_path: Path) -> Dict[str, str]: + """Parse a prompt_text file into a stem -> transcript mapping. + + Lines are ``<name>|<transcript>``; blank lines are skipped and a line + without a ``|`` separator is treated as a name with an empty transcript. + Returns an empty mapping when the file does not exist. + """ + if not prompt_path.exists(): + return {} + mapping: Dict[str, str] = {} + for line in prompt_path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + if "|" in line: + name, _, text = line.partition("|") + else: + name, text = line, "" + mapping[name.strip()] = text + return mapping + + def write_prompt_text(wav_dir: Path, transcripts: Dict[str, str]) -> Path: """Write the voice_dir prompt_text mapping into WAV_DIR. @@ -527,6 +705,20 @@ def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None: print(bar) +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: + print(f"[WARNING] Could not update {CONFIG_PATH}; edit " + "AUDIOCPP_API_URL by hand so audiobook.py uses the " + "new port") + else: + print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " + f"will still use port {config_port()}") + + def _ask_host_port_backend_lazy(args: argparse.Namespace, default_lazy: bool ) -> Tuple[str, int, str, bool]: @@ -536,62 +728,110 @@ def _ask_host_port_backend_lazy(args: argparse.Namespace, if port != config_port(): if ask_bool(f"Update AUDIOCPP_API_URL in converter/config.py to port " f"{port} so audiobook.py talks to this server", True): - if update_config_api_url_port(port): - print(f"[OK] Updated AUDIOCPP_API_URL in {CONFIG_PATH}") - else: - print(f"[WARNING] Could not update {CONFIG_PATH}; edit " - "AUDIOCPP_API_URL by hand so audiobook.py uses the " - "new port") + _apply_port_sync(port, True) else: - print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " - f"will still use port {config_port()}") + _apply_port_sync(port, False) backend = args.backend if args.backend else ask_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 -def _collect_transcripts(args: argparse.Namespace, - include_clone: bool) -> Dict[str, str]: +def _decide_transcription(wav_files: list, existing: Dict[str, str], + prompt_exists: bool, force: bool, + 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": [...]}. + """ + mode = "all" + missing: List[Path] = [] + if prompt_exists and not force: + missing = [wav for wav in wav_files + if not existing.get(wav.stem, "").strip()] + if not missing: + if confirm("All voices already transcribed in prompt_text. " + "Re-transcribe anyway?", False): + mode = "all" + else: + mode = "keep" + elif confirm("Existing transcription and new .wavs detected, " + "only transcribe new voices?", True): + mode = "missing" + else: + mode = "all" + return {"mode": mode, "missing": missing} + + +def _transcribe(args: argparse.Namespace, include_clone: bool, + plan: Optional[dict] = None + ) -> Tuple[Dict[str, str], bool]: """Transcribe the wav directory into a stem -> transcript mapping. - Returns the mapping (empty when no wavs were found or cloning is not - used by any selected family). Runs only when a cloning voice library is - needed; a run without any clone-capable family ignores the wav directory - entirely. + 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. """ if not include_clone: print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family " "selected, so voice presets are not used") - return {} + return {}, False wav_files = find_wav_files(args.input_dir) if not wav_files: print(f"[WARNING] No .wav files found in {args.input_dir}; writing the " "config without a voice_dir") - return {} + 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: + plan = _decide_transcription( + wav_files, existing, prompt_path.exists(), args.force, + lambda question, default: ask_bool(question, default)) + + if plan["mode"] == "keep": + print(f"[INFO] Kept existing {prompt_path}; all voices were " + "already transcribed, nothing new to transcribe") + return existing, False + if whisper_backend_available() is None: print("[WARNING] Neither faster_whisper nor whisper was found, so " "reference .wav files cannot be transcribed automatically and " "every transcript will be empty.") - print(' Did you remember to "conda activate qwen3-tts"? ' - "Transcripts must be added by hand (see the warning at the end).") - return transcribe_wav_dir(wav_files, args.whisper_model) + print(" Install whisper (or faster_whisper) in your " + "audiobook environment to transcribe automatically; otherwise " + "transcripts must be added by hand (see the warning at the end).") + + if plan["mode"] == "missing": + new_transcripts = transcribe_wav_dir(plan["missing"], args.whisper_model) + transcripts = dict(existing) + transcripts.update(new_transcripts) + else: + transcripts = transcribe_wav_dir(wav_files, args.whisper_model) + return transcripts, True -def _offer_config_model_id_sync(model_id: str) -> None: - """Offer to point converter/config.py at a single non-Qwen model entry. +def _offer_config_model_id_sync(model_id: str, + accepted: Optional[bool] = None) -> None: + """Offer to point converter/config.py at a single hosted model entry. The converter requests the model id configured in AUDIOCPP_MODEL_ID, and single-model servers use the same id for the clone entry, so both - ids are rewritten together. + ids are rewritten together. When ACCEPTED is None the user is asked + (line prompt); otherwise the given decision is applied. """ if config.AUDIOCPP_MODEL_ID == model_id \ and config.AUDIOCPP_CLONE_MODEL_ID == model_id: return - if ask_bool("Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID in " - f"converter/config.py to '{model_id}' so audiobook.py uses " - "this model", True): + if accepted is None: + accepted = ask_bool("Update AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID " + 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: @@ -611,30 +851,398 @@ def _print_multi_model_model_id_note(entry_ids: List[str]) -> None: 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], + id_picker: Callable[[str, str, str], str] + ) -> Tuple[List[dict], List[str], List[Tuple[str, str]], + List[str], bool]: + """Build server.json model entries from the selected families/packages. + + TASK_PICKER is called for each design package to choose vdes/tts; + ID_PICKER resolves a duplicate server entry id. Returns (model_entries, + entry_ids, install_guidance, design_entry_ids, include_clone). + """ + model_entries: List[dict] = [] + entry_ids: List[str] = [] + install_guidance: List[Tuple[str, str]] = [] + design_entry_ids: List[str] = [] + include_clone = False + for family in family_keys: + entry = catalog_by_family[family] + include_clone = include_clone or entry["clone_capable"] + for opt in chosen[family]: + task = task_picker(opt["install_id"]) if opt["design"] else TASK_TTS + base_id = (f"{entry['preferred_id']}-design" + if task == TASK_VDES else entry["preferred_id"]) + model_id = base_id + if model_id in entry_ids: + model_id = id_picker(entry["display_name"], opt["install_id"], + f"{base_id}-2") + entry_ids.append(model_id) + model_entries.append(build_model_entry( + family, model_id, f"models/{opt['target_directory']}", + task=task)) + install_guidance.append((entry["display_name"], opt["install_id"])) + if task == TASK_VDES: + design_entry_ids.append(model_id) + return (model_entries, entry_ids, install_guidance, + 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, + transcripts: Dict[str, str], write_prompt: bool) -> None: + """Console phase shared by both UI modes: write files and print guidance.""" + voice_dir: Optional[str] = None + if transcripts: + if write_prompt: + prompt_path = wav_dir / PROMPT_TEXT_FILENAME + write_prompt_text(wav_dir, transcripts) + print(f"[OK] Wrote {prompt_path}") + voice_dir = str(wav_dir.resolve()) + + server_config = build_server_config( + 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"') + + +def _build_tree_families(catalog: List[dict]) -> List[dict]: + """Shape the catalog into the checkbox_tree widget's family list.""" + families: List[dict] = [] + for entry in catalog: + capabilities = ["tts"] + if "clone" in entry["tasks"]: + capabilities.append("cloning") + if "design" in entry["tasks"]: + capabilities.append("design") + 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, + "recommended": opt["recommended"], + }) + families.append({ + "label": name, + "detail": ", ".join(capabilities), + "options": options, + }) + return families + + +def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser + ) -> Optional[dict]: + """Run every TUI screen; return the collected settings, or None to abort.""" + 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} + + # 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 + + # 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) + + model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ + _build_entries(family_keys, chosen, catalog_by_family, + task_picker, id_picker) + + # 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 + + # 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 + + 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, + } + + +def _run_tui(args: argparse.Namespace, parser: argparse.ArgumentParser) -> int: + """Run the TUI wizard, then the shared console phase.""" + import curses + tui = _load_tui() + try: + settings = curses.wrapper(_wizard, args, parser) + except _TuiError as exc: + print(f"[ERROR] {exc}", file=sys.stderr) + return 2 + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + if settings is None: + print("[INFO] Aborted; existing server.json kept") + return 1 + + # Port sync (applied now that the terminal is back). + if settings["sync_port"] is True: + _apply_port_sync(settings["port"], True) + elif settings["sync_port"] is False: + _apply_port_sync(settings["port"], False) + + # Transcription (console; the questions were already answered in the TUI). + args.input_dir = settings["wav_dir"] + if settings["include_clone"]: + transcripts, write_prompt = _transcribe(args, True, plan=settings["plan"]) + elif args.input_dir is not None: + print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable family " + "selected, so voice presets are not used") + transcripts, write_prompt = {}, False + else: + 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) + + 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) + return 0 + + def main() -> int: parser = argparse.ArgumentParser( description="Generate a server.json for the audio.cpp audiocpp_server " "hosting one or more TTS model families used by this converter.") - parser.add_argument("input_dir", type=resolve_wav_dir_arg, metavar="WAV_DIR", + parser.add_argument("--wavs", type=resolve_wav_dir_arg, default=None, + dest="input_dir", metavar="WAV_DIR", help="Directory with .wav reference files to publish as " - "a server-level voice_dir cloning library (required)") - parser.add_argument("--output", type=Path, default=Path("server.json"), + "a server-level voice_dir cloning library (asked " + "for when omitted)") + parser.add_argument("--output", type=Path, default=None, help="Output path for server.json (default: " - "server.json in the current directory)") - parser.add_argument("--audiocpp-dir", type=Path, default=None, + "server.json inside the audio.cpp checkout; if it " + "already exists you are asked [Y/n] to overwrite, " + "and answering 'n' writes server.json in the " + "current directory instead)") + parser.add_argument("--audiocpp-dir", type=normalize_dir_arg, default=None, help="Path to a local audio.cpp checkout containing a " "model_specs/ directory (default: detected from " "AUDIOCPP_DIR or an audio.cpp directory next to/above " "the current working directory; prompted otherwise)") parser.add_argument("--families", type=str, default=None, help="Comma-separated model families to host, as named " - "in the audio.cpp catalog (e.g. " - "qwen3_tts,higgs_audio_tts). Skips the family checklist") - parser.add_argument("--models", choices=MODEL_SELECTIONS, default=None, - help="Which Qwen3-TTS models to host: both (default), " - "custom (CustomVoice speakers only), or clone " - "(Base voice cloning only). Only valid when the " - "qwen3_tts family is selected") + "in the audio.cpp catalog (e.g. " + "qwen3_tts,higgs_audio_tts). Skips the family " + "checklist") + parser.add_argument("--all-packages", action="store_true", + help="Instead of hosting each family's recommended " + "package, offer a checklist of every installable " + "package (distinct target_directory) so several " + "packages of one family can be hosted at once. " + "In the TUI this pre-expands every family in the " + "tree (which always lists all packages)") parser.add_argument("--host", type=str, default=None, help="Bind host for the server (default: 127.0.0.1)") parser.add_argument("--port", type=int, default=None, @@ -652,14 +1260,38 @@ def main() -> int: parser.add_argument("--force", action="store_true", help="Overwrite the output file (and prompt_text) " "without prompting") + parser.add_argument("--notui", action="store_true", + help="Use the classic line prompts instead of the " + "full-screen TUI (automatic when curses is " + "unavailable or stdin/stdout is not a terminal)") args = parser.parse_args() + if args.input_dir is not None and not args.input_dir.is_dir(): + parser.error( + f"WAV directory not found: {args.input_dir}\n" + f" (resolved from the current working directory: " + f"{Path.cwd()})\n" + " --wavs must be a directory containing the .wav " + "reference files to use as voice cloning presets") + + if _tui_enabled(args): + return _run_tui(args, parser) + + # ---- Line-prompt flow (original behaviour). --------------------------- + + # Resolve the wav directory (flag, else prompt). + if args.input_dir is None: + answer = ask("Directory with .wav reference files", "") + 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 " + "reference files to use as voice cloning presets") if not args.input_dir.is_dir(): parser.error( f"WAV directory not found: {args.input_dir}\n" f" (resolved from the current working directory: " f"{Path.cwd()})\n" - " WAV_DIR must be a directory containing the .wav " + " --wavs must be a directory containing the .wav " "reference files to use as voice cloning presets") # Resolve the audio.cpp checkout and load its model catalog. @@ -667,7 +1299,10 @@ def main() -> int: if audiocpp_dir is None: audiocpp_dir = detect_audiocpp_dir() if audiocpp_dir is None: - audiocpp_dir = Path(ask("Path to your audio.cpp checkout", "") or "") + print("[INFO] Could not find an audio.cpp checkout next to or above " + "the current directory.") + answer = ask("Path to your audio.cpp checkout", "") + audiocpp_dir = normalize_dir_arg(answer) if answer else None if not audiocpp_dir: parser.error( "An audio.cpp checkout is required to read the model catalog. " @@ -686,10 +1321,23 @@ def main() -> int: f"No TTS model families found in {audiocpp_dir}/model_specs; " "check the checkout is up to date") - if args.output.exists() and not args.force \ - and not prompt_overwrite(args.output): - print("[INFO] Aborted; existing server.json kept") - return 1 + # Resolve the server.json output path. It defaults to the audio.cpp + # checkout; an existing file is overwritten only with confirmation, and a + # declined overwrite of the default location falls back to the current + # working directory. + 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 ask_bool(f"{output_path} already exists. Overwrite?", True): + if args.output is None: + output_path = Path.cwd() / "server.json" + if output_path.exists() and not ask_bool( + f"{output_path} already exists. Overwrite?", True): + print("[INFO] Aborted; existing server.json kept") + return 1 + else: + print("[INFO] Aborted; existing server.json kept") + return 1 # Select families. if args.families is not None: @@ -708,151 +1356,38 @@ def main() -> int: family_keys = ask_families(catalog) catalog_by_family = {entry["family"]: entry for entry in catalog} - is_qwen = FAMILY_QWEN3_TTS in family_keys - if not is_qwen and args.models is not None: - parser.error("--models only applies to the qwen3_tts family") - if is_qwen and args.models is not None and len(family_keys) > 1 \ - and args.models != "both": - parser.error( - "--models custom/clone selects Qwen3-TTS sub-entries and is only " - "valid when qwen3_tts is the sole selected family") - - print("[INFO] Model ids from converter/config.py:") - print(f" built-in speakers (CustomVoice): '{config.AUDIOCPP_MODEL_ID}'") - print(f" voice cloning (Base): '{config.AUDIOCPP_CLONE_MODEL_ID}'") - model_entries: List[dict] = [] - entry_ids: List[str] = [] - non_qwen_single_id: Optional[str] = None - - if is_qwen: - selection = args.models if args.models is not None else ask_models() - # When qwen3_tts is selected with other families, keep both entries so - # speaker mode and cloning are both available; custom/clone sub-choice - # is only honored when qwen3_tts is the sole family. - if len(family_keys) > 1 and args.models is None: - selection = "both" - include_custom = selection in ("both", "custom") - include_clone = selection in ("both", "clone") - - custom_voice_id = config.AUDIOCPP_MODEL_ID - clone_model_id = config.AUDIOCPP_CLONE_MODEL_ID - if include_custom and include_clone and custom_voice_id == clone_model_id: - print(f"[WARNING] AUDIOCPP_MODEL_ID and AUDIOCPP_CLONE_MODEL_ID are " - f"both '{custom_voice_id}' in converter/config.py, but server " - "model ids must be unique.") - clone_model_id = ask_distinct_clone_id(custom_voice_id) - - custom_voice_path = base_path = None - if include_custom: - custom_voice_path = ask("Path to the Qwen3-TTS CustomVoice GGUF package", - DEFAULT_CUSTOM_VOICE_PATH) - model_entries.append(build_model_entry( - FAMILY_QWEN3_TTS, custom_voice_id, custom_voice_path)) - entry_ids.append(custom_voice_id) - if include_clone: - base_path = ask("Path to the Qwen3-TTS Base GGUF package", - DEFAULT_BASE_PATH) - model_entries.append(build_model_entry( - FAMILY_QWEN3_TTS, clone_model_id, base_path)) - entry_ids.append(clone_model_id) - qwen_include_clone = include_clone - else: - qwen_include_clone = False - - # Non-Qwen families: one entry each. + chosen: Dict[str, List[dict]] = {} for family in family_keys: - if family == FAMILY_QWEN3_TTS: - continue entry = catalog_by_family[family] - model_id = entry["preferred_id"] - # Ensure uniqueness against already-chosen ids. - if model_id in entry_ids: - model_id = ask(f"Server model id for {entry['display_name']}", - f"{model_id}-2") - model_path = entry["default_path"] - # For a single non-Qwen family, ask the path (matching the old flow); - # for several, use the catalog default to keep the prompt count sane. - if len(family_keys) == 1: - model_path = ask(f"Path to the {entry['display_name']} package", - model_path) - model_entries.append(build_model_entry(family, model_id, model_path)) - entry_ids.append(model_id) - if len(family_keys) == 1: - non_qwen_single_id = model_id - - # Whether any selected family can clone (drives voice_dir / wav transcription). - include_clone = qwen_include_clone or any( - catalog_by_family[f]["clone_capable"] - for f in family_keys if f != FAMILY_QWEN3_TTS) - - # Default to lazy loading only when hosting more than one family: a - # single-family server (including the Qwen3-TTS CustomVoice+Base pair) - # loads at startup as before, while a multi-family server avoids loading - # every model until it is actually used. - default_lazy = len(family_keys) > 1 - host, port, backend, lazy_load = _ask_host_port_backend_lazy(args, default_lazy) - - transcripts = _collect_transcripts(args, include_clone) - - voice_dir: Optional[str] = None - if transcripts: - prompt_path = args.input_dir / PROMPT_TEXT_FILENAME - if prompt_path.exists() and not args.force: - if not ask_bool(f"Overwrite existing {prompt_path}", True): - print(f"[INFO] Kept existing {prompt_path}; new transcripts " - "were not written") - else: - write_prompt_text(args.input_dir, transcripts) - print(f"[OK] Wrote {prompt_path}") + if args.all_packages: + chosen[family] = ask_package_dirs(entry) else: - write_prompt_text(args.input_dir, transcripts) - print(f"[OK] Wrote {prompt_path}") - voice_dir = str(args.input_dir.resolve()) - - server_config = build_server_config( - host=host, port=port, backend=backend, lazy_load=lazy_load, - model_entries=model_entries, voice_dir=voice_dir) + chosen[family] = [opt for opt in package_dir_options(entry) + if opt["recommended"]] + + model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ + _build_entries(family_keys, chosen, catalog_by_family, + task_picker=lambda install_id: ask_package_task(install_id), + id_picker=lambda display_name, install_id, base_id: ask( + f"Server model id for {display_name} package " + f"'{install_id}'", f"{base_id}-2")) + + # Default to lazy loading when hosting more than one model entry: a + # 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) - print("\nGenerated server.json:") - print(json.dumps(server_config, indent=2, ensure_ascii=False)) - if not ask_bool(f"\nWrite this to {args.output}", True): - print("[INFO] Aborted; nothing written") - return 1 + transcripts, write_prompt = _transcribe(args, include_clone) - with args.output.open("w", encoding="utf-8") as handle: - json.dump(server_config, handle, indent=2, ensure_ascii=False) - handle.write("\n") + _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) - # Post-generation guidance. - print(f"\n[OK] Wrote {args.output} with {len(model_entries)} model entry/entries" - + (f" and voice_dir '{voice_dir}'" if voice_dir else "")) - for family in family_keys: - entry = catalog_by_family[family] - if family == FAMILY_QWEN3_TTS: - print("[INFO] Install the Qwen3-TTS packages from the audio.cpp " - "checkout:") - print(" python3 tools/model_manager_v2.py install " - "qwen3_tts_1_7b_customvoice_q8_0") - print(" python3 tools/model_manager_v2.py install " - "qwen3_tts_1_7b_base_q8_0") - else: - print(f"[INFO] Install {entry['display_name']} from the audio.cpp " - f"checkout: python3 tools/model_manager_v2.py install " - f"{entry['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.") - if family_keys != [FAMILY_QWEN3_TTS]: - for family in family_keys: - if family == FAMILY_QWEN3_TTS: - continue - entry = catalog_by_family[family] - print(f"[INFO] Clone-only family {entry['display_name']}: run " - "audiobook.py with --backend audiocpp --voice <preset name>") - if len(entry_ids) == 1 and non_qwen_single_id is not None: - _offer_config_model_id_sync(non_qwen_single_id) + 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) diff --git a/tools/tui.py b/tools/tui.py new file mode 100644 index 0000000..e1eda05 --- /dev/null +++ b/tools/tui.py @@ -0,0 +1,511 @@ +#!/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. + +Common key bindings: + + Up/Down (or k/j) move the cursor + Enter accept + Esc abort the whole wizard (raises WizardCancelled) + +On screens without typed text (menus, confirm, tree, browser) 'q' also +aborts; inside text editors it is an ordinary character. +""" + +import os +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.""" + + +# --------------------------------------------------------------------------- +# 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 _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 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(). + """ + + 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.scroll = 0 + self.cursor = 0 # highlighted row index + + def mark(self, text: str, attr: int = 0, indent: int = 0) -> None: + self.rows.append({"text": text, "attr": attr, "indent": indent}) + + 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) + 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) + scr.refresh() + + # -- key helpers ------------------------------------------------------ + + 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 edit_line(self, start: str, prompt: str = "" + ) -> Optional[str]: + """Run an inline editor on the message 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 + while True: + height, width = self.scr.getmaxyx() + self.message = "" + 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 + return text + if key in (curses.KEY_BACKSPACE, 8, 127): + text = text[:-1] + elif 32 <= key < 127: + text += chr(key) + + +# --------------------------------------------------------------------------- +# Widget: yes/no confirm +# --------------------------------------------------------------------------- + +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). + """ + frame = Frame(scr, question, + "y = yes n = no Enter = default Esc = cancel") + cancel = (27, ord("q")) + 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.draw() + key = frame.get_key(cancel) + if key in (ord("y"), ord("Y")): + return True + if key in (ord("n"), ord("N")): + return False + if key in (10, 13): + return default + + +# --------------------------------------------------------------------------- +# Widget: single-choice menu +# --------------------------------------------------------------------------- + +def menu(scr, title: str, options: Sequence[tuple], default_index: int = 0): + """Show OPTIONS as (label, value) pairs; return the chosen value. + + The cursor starts on DEFAULT_INDEX; Enter returns the highlighted + option's value. + """ + 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 label, _ in options: + frame.mark(label) + frame.cursor = 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) + 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 + ) -> 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). + """ + frame = Frame(scr, title, + "type to edit Backspace = erase Enter = accept " + "Esc = cancel") + text = default + error = "" + while True: + frame.rows = [] + frame.mark("") + frame.mark(f" {text}_") + frame.cursor = 1 + frame.message = error + frame.draw() + curses = frame.curses + key = frame.get_key() # Esc only; 'q' must stay typeable + 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 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 + ) -> 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. + """ + footer = ("Up/Down = move Right = open Left = parent e = edit " + "path Enter = choose this directory 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 + + def validation_error() -> Optional[str]: + if validate is None: + return None + return validate(current) + + while True: + entries = _list_dirs(current) + cursor = max(0, min(cursor, max(0, len(entries) - 1))) + 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) + frame.mark("") + if not entries: + frame.mark(" (no subdirectories)") + 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.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 + 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 + elif key == ord("e"): + result = frame.edit_line("", prompt="path: ") + if result is not None: + 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 + cursor = 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) + + +# --------------------------------------------------------------------------- +# Widget: expandable checkbox tree +# --------------------------------------------------------------------------- + +def checkbox_tree(scr, title: str, families: List[dict], + footer: Optional[str] = None, + expand_all: bool = False) -> 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. The first family's recommended option starts + checked (the prompt flow's default), and with EXPAND_ALL every + family starts expanded. + """ + 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"])) + + 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] + mark = "x" if family_checked(index) 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) + 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}") + frame.cursor = cursor + node = nodes[cursor] + frame.message = families[node[1]].get("detail", "") + frame.message_attr = frame.curses.A_DIM + 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) + 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.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 |