diff options
Diffstat (limited to 'app')
| -rwxr-xr-x | app/backends/audiocpp.py | 2556 | ||||
| -rw-r--r-- | app/backends/audiocpp/__init__.py | 107 | ||||
| -rw-r--r-- | app/backends/audiocpp/__main__.py | 8 | ||||
| -rw-r--r-- | app/backends/audiocpp/build.py | 339 | ||||
| -rw-r--r-- | app/backends/audiocpp/catalog.py | 293 | ||||
| -rw-r--r-- | app/backends/audiocpp/configsync.py | 163 | ||||
| -rw-r--r-- | app/backends/audiocpp/constants.py | 32 | ||||
| -rw-r--r-- | app/backends/audiocpp/models.py | 384 | ||||
| -rw-r--r-- | app/backends/audiocpp/patches/ggml-top-k-cuda-iterator.patch (renamed from app/backends/patches/ggml-top-k-cuda-iterator.patch) | 0 | ||||
| -rw-r--r-- | app/backends/audiocpp/remote.py | 59 | ||||
| -rw-r--r-- | app/backends/audiocpp/status.py | 90 | ||||
| -rw-r--r-- | app/backends/audiocpp/voices.py | 146 | ||||
| -rw-r--r-- | app/backends/audiocpp/wizard.py | 1081 | ||||
| -rwxr-xr-x | app/backends/faster.py | 49 | ||||
| -rw-r--r-- | app/backends/qwen.py | 53 | ||||
| -rw-r--r-- | app/backends/setup.py | 73 | ||||
| -rw-r--r-- | app/docs/backend-audiocpp.md | 2 | ||||
| -rw-r--r-- | app/tests/test_backends.py | 10 | ||||
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 497 | ||||
| -rw-r--r-- | app/tests/test_backends_faster.py | 3 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 9 |
21 files changed, 3060 insertions, 2894 deletions
diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py deleted file mode 100755 index 3cca94c..0000000 --- a/app/backends/audiocpp.py +++ /dev/null @@ -1,2556 +0,0 @@ -#!/usr/bin/env python3 -"""Set up the audio.cpp TTS backend for the audiobook generator. - -This does the whole audio.cpp setup end-to-end as a full-screen DOS-style -TUI: locate or clone an audio.cpp checkout into ``app/audio.cpp``, optionally -build ``audiocpp_server``, pick model families/packages from the checkout's -``model_specs`` catalog, transcribe reference .wav voices, write -``server.json``, sync ``app/converter/config.py``, download the models, and -print the exact command to start the server. It is driven by -``audiobook.py``'s TUI hub (``backends.REGISTRY``) but can also be run -directly for scripting — every value has a flag, and a non-interactive run -with all flags supplied never opens the TUI. - -The converter 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. - -Usage: - python app/backends/audiocpp.py [--wavs WAV_DIR] [--output PATH] - [--clone] [--families FAM1,FAM2] - [--all-packages] [--host HOST] [--port PORT] - [--build-backend {cuda,vulkan,hip,cpu}] [--backend {cuda,vulkan,hip,cpu}] - [--whisper-model NAME] [--force] - [--download] [--no-sync-port] [--no-sync-model-ids] - -With no flags and a terminal, the TUI wizard runs. Without a terminal -(or with all flags supplied), it runs non-interactively from the flags; -any missing required value is a hard error with a remediation hint. - -When the target ``server.json`` already exists, the TUI wizard runs as a -"modify": it loads the existing models, host, port, backend and voice -directory and pre-fills the screens with them (the model tree -opens with the installed models already checked) instead of prompting to -overwrite, and offers to delete already-downloaded models that are no -longer selected. -""" - -import argparse -import contextlib -import io -import json -import os -import re -import shlex -import shutil -import sys -import tempfile -import urllib.parse -import urllib.request -from datetime import datetime -from pathlib import Path -from typing import Callable, Dict, List, Optional, Set, Tuple - -# Allow running directly (python app/backends/audiocpp.py) from any cwd. -sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) - -from backends import ( - BackendStatus, - ServerSpec, - common, - format_launch_hint, - probe, - servers, -) -from backends.common import ( - APP_DIR, - CONFIG_PATH, - PROMPT_TEXT_FILENAME, - TTS_ROOT, - VOICES_DIR, - detect_wav_dir, - find_wav_files, - read_prompt_text, - resolve_wav_dir_arg, - url_with_port, - write_prompt_text, -) -from backends.common import ( - wav_dir_info as _wav_dir_info, -) -from backends.common import ( - wav_dir_preview as _wav_dir_preview, -) -from converter import config -from converter.clients import transcribe_reference_audio, whisper_backend_available -from ui import taskview, tui - -DEFAULT_HOST = "127.0.0.1" -FALLBACK_PORT = 8080 - -BACKENDS = ("cuda", "vulkan", "hip", "cpu") - -TASK_TTS = "tts" -TASK_VDES = "vdes" - -# audio.cpp is cloned into the app directory of the audiobook generator. -AUDIOCPP_DIR_NAME = "audio.cpp" -AUDIOCPP_GIT_URL = "https://github.com/0xShug0/audio.cpp" - -# ggml build patches shipped in this repo and applied to the (gitignored) -# audio.cpp checkout before building, so a fresh clone survives known ggml -# build bugs the audio.cpp fork has not re-vendored yet. See -# apply_ggml_patches() below. -PATCH_DIR = Path(__file__).resolve().parent / "patches" - -# Sentinel returned by tui.confirm (via its cancel_value) when the user -# presses Esc on an overwrite prompt to go back to the wav-directory browser -# instead of aborting the wizard. -_GO_BACK = object() - - -class _GoBack(Exception): - """Internal signal: Esc was pressed inside one of a screen's sub-prompts. - - The wizard drives a stack of screens via ``tui.Wizard``. Helpers that ask - several questions through callbacks (the task/id pickers inside - ``_build_entries``, the transcription plan, the download prompt) cannot - themselves return the wizard's ``BACK`` sentinel, so they convert the - ``_GO_BACK`` value passed to each widget into this exception. The screen - that invoked the helper catches it and returns ``tui.Wizard.BACK``, which - pops back to the previous screen. Esc on the first screen aborts 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) - - -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 _interactive() -> bool: - """True when the TUI wizard can run (curses importable + tty).""" - try: - import curses # noqa: F401 - except ImportError: - return False - try: - return sys.stdin.isatty() and sys.stdout.isatty() - except (AttributeError, ValueError): - return False - - -# 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 config_port() -> int: - """Return the port of AUDIOCPP_API_URL in app/converter/config.py.""" - try: - return urllib.parse.urlsplit(config.AUDIOCPP_API_URL).port or FALLBACK_PORT - except ValueError: - return FALLBACK_PORT - - -def update_config_api_url_port(port: int, config_path: Optional[Path] = None) -> bool: - """Rewrite the port inside AUDIOCPP_API_URL in app/converter/config.py. - - Reads the configured URL from the file (not from the imported module, - which a long hub session can leave behind), swaps its port for PORT, - and writes it back through ``common.update_config_value`` so the - imported module mirrors the change immediately. Returns True when the - file now holds the new URL. - """ - path = Path(config_path) if config_path is not None else CONFIG_PATH - try: - text = path.read_text(encoding="utf-8") - except OSError: - return False - match = re.search(r'(?m)^\s*AUDIOCPP_API_URL\s*=\s*"([^"]*)"', text) - if not match: - return False - return common.update_config_value("AUDIOCPP_API_URL", - url_with_port(match.group(1), port), - config_path=path) - - -def update_server_config_port(port: int) -> bool: - """Rewrite the 'port' in the audio.cpp checkout's server.json. - - Loads ``<checkout>/server.json``, sets its ``port`` to PORT, and - rewrites it with the same ``json.dump`` formatting the wizard uses. - Returns True when the file now carries PORT (a no-op when it already - does), and False when there is no checkout/server.json or the file - cannot be read or written. - """ - checkout = find_local_checkout() - if checkout is None: - return False - server_json = checkout / "server.json" - if not server_json.exists(): - return False - try: - data = json.loads(server_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - return False - if not isinstance(data, dict): - return False - if data.get("port") == port: - return True - data["port"] = port - try: - with server_json.open("w", encoding="utf-8") as handle: - json.dump(data, handle, indent=2, ensure_ascii=False) - handle.write("\n") - except OSError: - return False - return True - - -def update_config_model_ids(model_id: str, - clone_model_id: Optional[str] = None, - config_path: Optional[Path] = None) -> bool: - """Rewrite AUDIOCPP_MODEL_ID (and AUDIOCPP_CLONE_MODEL_ID when given). - - Goes through ``common.update_config_value`` so the imported config - module mirrors the change immediately. Returns True when every named - key now holds its value in the file. - """ - path = Path(config_path) if config_path is not None else CONFIG_PATH - ok = common.update_config_value("AUDIOCPP_MODEL_ID", model_id, - config_path=path) - if clone_model_id is not None: - ok = common.update_config_value("AUDIOCPP_CLONE_MODEL_ID", - clone_model_id, - config_path=path) and ok - return ok - - -# 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. - - Prefers the package flagged ``default: true``, then the first GGUF - package, then the first package overall. Returns None for an empty list. - """ - if not packages: - return None - for package in packages: - if package.get("default"): - return package - for package in packages: - if package.get("format") == "gguf": - return package - return packages[0] - - -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, packages (the full list from the spec), install_id - (recommended package id), and default_path (``models/<target_directory>``). - 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(): - raise NotADirectoryError( - f"{audiocpp_dir} has no model_specs/ directory; re-run setup " - "to refresh the audio.cpp checkout") - entries: List[dict] = [] - for spec_path in sorted(specs_dir.glob("*.json")): - try: - spec = json.loads(spec_path.read_text(encoding="utf-8")) - except (OSError, ValueError): - continue - tasks = spec.get("tasks") or [] - if "tts" not in tasks and spec.get("category") != "tts": - continue - family = spec.get("family") or spec_path.stem - packages = spec.get("packages") or [] - package = _default_package(packages) - if package is None: - # No installable package: skip (cannot be hosted from a path). - continue - target_directory = package.get("target_directory") or family - languages = spec.get("languages") or [] - display_name = spec.get("display_name") or family - description = spec.get("description") or "" - entries.append({ - "family": family, - "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}", - }) - - # All families are treated equally: alphabetical by display name. - entries.sort(key=lambda entry: entry["display_name"].lower()) - 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 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": task, - "mode": "offline", - } - - -def build_server_config(host: str, port: int, backend: str, lazy_load: bool, - model_entries: List[dict], - voice_dir: Optional[str] = None) -> dict: - """Assemble the server.json document. - - ``voice_dir`` is a server-level cloning voice library; when set, every - hosted clone-capable family can use its voices with ``--voice``. - """ - config_doc = { - "host": host, - "port": port, - "backend": backend, - "lazy_load": lazy_load, - "models": model_entries, - } - if voice_dir: - config_doc["voice_dir"] = voice_dir - return config_doc - - -def transcribe_wav_dir(wav_files: list, whisper_model: str, - cancel=None) -> Dict[str, str]: - """Transcribe each wav file and return a mapping of stem -> transcript. - - CANCEL (a ``threading.Event``) is checked between files so the in-TUI - task view can stop a long transcription early. - """ - transcripts: Dict[str, str] = {} - for wav_file in wav_files: - if cancel is not None and cancel.is_set(): - print("[INFO] Transcription cancelled") - break - name = wav_file.stem - print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") - text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) - if text: - print(f"[OK] {name}: {text}") - else: - print(f"[WARNING] No transcript for '{name}'; cloning works best " - "with an accurate transcript — consider editing prompt_text " - "by hand before starting the server") - transcripts[name] = text or "" - return transcripts - - -def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None: - """Print a loud, final warning for voices whose transcript is empty.""" - empty = sorted(name for name, text in transcripts.items() if not text) - if not empty: - return - bar = "=" * 70 - print() - print(bar) - print("[WARNING] MANUAL TRANSCRIPTION REQUIRED") - print(bar) - listing = " - " + "\n - ".join(empty) if len(empty) > 1 else f" - {empty[0]}" - print(f"The following voice(s) have an EMPTY transcript in prompt_text:\n" - f"{listing}") - print("Those voices will NOT work until you add an accurate transcript.") - print(f"Edit {PROMPT_TEXT_FILENAME} in your voice directory and fill in the " - "text after '|' for each voice above.") - print(bar) - - -def _apply_port_sync(port: int, accepted: bool) -> None: - """Write the port into app/converter/config.py, or report when declined.""" - if accepted: - 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") - else: - print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " - f"will still use port {config_port()}") - - -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": - [...], "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] = [] - 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, "existing": existing} - - -def _transcribe(args: argparse.Namespace, plan: Optional[dict], - cancel=None) -> Tuple[Dict[str, str], bool]: - """Transcribe the wav directory into a stem -> transcript mapping. - - 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). - PLAN is always pre-collected — by the TUI (via _decide_transcription and - its confirm callbacks) or by _flag_plan for a non-interactive run — so no - questions are asked here; a None PLAN defaults to "transcribe everything". - CANCEL is checked between files. - """ - 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 {}, False - - prompt_path = args.input_dir / PROMPT_TEXT_FILENAME - existing = dict((plan or {}).get("existing") or {}) - mode = plan["mode"] if plan else "all" - - if 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(" 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, - cancel=cancel) - transcripts = dict(existing) - transcripts.update(new_transcripts) - else: - transcripts = transcribe_wav_dir(wav_files, args.whisper_model, - cancel=cancel) - return transcripts, True - - -def _flag_plan(wav_files: list, prompt_path: Path, force: bool) -> dict: - """Build a transcription plan for a non-interactive (flag-only) run. - - With --force everything is re-transcribed; otherwise an existing - prompt_text is reused and only voices with an empty transcript are - re-transcribed, mirroring what the TUI confirms interactively. - """ - if prompt_path.exists() and not force: - existing = read_prompt_text(prompt_path) - missing = [wav for wav in wav_files - if not existing.get(wav.stem, "").strip()] - if not missing: - return {"mode": "keep", "missing": [], "existing": existing} - return {"mode": "missing", "missing": missing, "existing": existing} - return {"mode": "all", "missing": [], "existing": {}} - - -def _offer_config_model_id_sync(model_id: str, accepted: Optional[bool]) -> None: - """Point app/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. ACCEPTED is True/False (apply/skip the - rewrite) or None when no single-entry sync applies (nothing to do). - """ - if config.AUDIOCPP_MODEL_ID == model_id \ - and config.AUDIOCPP_CLONE_MODEL_ID == model_id: - return - if accepted is None: - return - if accepted: - 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") - else: - print("[WARNING] Left the model ids unchanged; audiobook.py will " - f"still request model '{config.AUDIOCPP_MODEL_ID}'") - - -def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], - catalog_by_family: Dict[str, dict], - task_picker: Callable[[str], str], - known_tasks: Optional[Dict[Tuple[str, str], str]] = None - ) -> 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. - KNOWN_TASKS maps ``(family, target_directory)`` to a previously-stored - task ("tts" or "vdes") so a modify run preserves how a design package - was hosted instead of re-asking. Each entry's server id is its package - ``target_directory`` (flattened to a token), so packages from the same - family never collide; an id that does collide (across families) is - auto-suffixed without prompting. 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]: - if opt["design"]: - task = known_tasks.get((family, opt["target_directory"])) \ - if known_tasks else None - if task is None: - task = task_picker(opt["install_id"]) - else: - task = TASK_TTS - base_id = opt["target_directory"].replace("/", "-") - model_id = base_id - if model_id in entry_ids: - n = 2 - while f"{base_id}-{n}" in entry_ids: - n += 1 - model_id = f"{base_id}-{n}" - 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(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, 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: - 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) - - with output_path.open("w", encoding="utf-8") as handle: - json.dump(server_config, handle, indent=2, ensure_ascii=False) - handle.write("\n") - - 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, emit=None, cancel=None) -> int: - """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`` 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. - - With EMIT given (the in-TUI task view) each download streams its output - to EMIT and — when the checkout's ``model_manager_v2.py`` supports it — - runs with ``--progress --cancel-file`` so the view can show a real byte - progress bar and cancel gracefully. CANCEL aborts a running download. - - Returns 0 when every command succeeded (or nothing needed running), - 130 when cancelled, 1 when any download failed. - """ - 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) - - supports_progress = emit is not None and _manager_supports_progress(manager) - - if download and not manager.is_file(): - print(f"[WARNING] {manager} not found; printing the install commands " - "instead of running them") - download = False - - failed = 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}...") - argv = [sys.executable, str(manager), "install", install_id] - cancel_file: Optional[Path] = None - on_cancel = None - if supports_progress: - fd, cancel_path = tempfile.mkstemp( - prefix="audiocpp_cancel_", suffix=".cancel") - os.close(fd) - cancel_file = Path(cancel_path) - cancel_file.unlink() # absent = not cancelled - argv += ["--progress", "--cancel-file", str(cancel_file)] - on_cancel = cancel_file.touch - try: - rc = common.run_console_subprocess( - argv, cwd=str(audiocpp_dir), emit=emit, cancel=cancel, - on_cancel=on_cancel) - except OSError as exc: - print(f"[WARNING] Could not run {command}: {exc}") - rc = 1 - finally: - if cancel_file is not None: - try: - cancel_file.unlink() - except OSError: - pass - if rc == 130 or (cancel is not None and cancel.is_set()): - return 130 - if rc != 0: - failed = True - print(f"[WARNING] install {install_id} exited with code " - f"{rc}; the model may need to be downloaded " - "by hand") - return 1 if failed else 0 - - -def _manager_supports_progress(manager: Path) -> bool: - """True when MANAGER (model_manager_v2.py) supports --progress output. - - The ``--progress``/``--cancel-file`` flags are relatively recent; an - older audio.cpp checkout may not have them, so probe the script source - once instead of failing the download with an unknown flag. - """ - try: - text = manager.read_text(encoding="utf-8", errors="ignore") - except OSError: - return False - return "AUDIOCPP_PROGRESS" in text and "--cancel-file" in text - - -def _decide_download(audiocpp_dir: Path, - model_entries: List[dict], - 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. The prompt is also skipped (False) - when every selected model is already on disk (see ``_all_models_present``), - so an already-configured checkout is not asked to re-download models it - already has. - """ - manager = audiocpp_dir / "tools" / "model_manager_v2.py" - if not manager.is_file(): - return False - if _all_models_present(audiocpp_dir, model_entries): - return False - return confirm( - "Automatically download the selected models with model_manager_v2.py " - "now?", True) - - -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"] - options = [] - for opt in package_dir_options(entry): - options.append({ - "key": opt["target_directory"], - "label": opt["install_id"], - "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. - - The wizard is driven by ``tui.Wizard`` as a stack of screen closures: - each screen shows one interactive widget and returns the next screen - (a closure), ``Wizard.BACK`` (Esc/q pressed — pop to the previous - screen), or the final settings dict. Only screens that actually render - are pushed, so Esc always lands on the previous real screen. A step - whose value is already provided by a flag (``--host``, ``--port``, - ``--families``, ...) or does not apply (e.g. the port-sync prompt when - the port did not change) is folded into the ``_after_*`` guards and - never becomes a screen. Esc on the first screen aborts the whole - wizard. - """ - - s: dict = {} - - 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 - - def resolve_checkout(audiocpp_dir: Path) -> None: - """Validate the audio.cpp checkout and populate the wizard state ``s``.""" - audiocpp_dir = Path(audiocpp_dir).resolve() - 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" - # Modify flow: an existing server.json seeds the wizard's screens - # instead of being overwritten from scratch (an explicit --force - # still starts fresh). - existing_config = load_server_config(output_path) \ - if not args.force else None - if existing_config is not None: - existing_selected, existing_tasks = \ - server_config_selections(existing_config, catalog) - else: - existing_selected, existing_tasks = {}, {} - s.update({ - "audiocpp_dir": audiocpp_dir, - "catalog": catalog, - "catalog_by_family": catalog_by_family, - "output_path": output_path, - "existing_config": existing_config, - "existing_selected": existing_selected, - "existing_tasks": existing_tasks, - "existing_host": existing_config.get("host") - if existing_config else None, - "existing_port": existing_config.get("port") - if existing_config else None, - "existing_backend": existing_config.get("backend") - if existing_config else None, - "existing_voice_dir": existing_config.get("voice_dir") - if existing_config else None, - "detected_backend": detect_backend(audiocpp_dir), - }) - - def _families_from_flag() -> None: - requested = [f.strip() for f in args.families.split(",") if f.strip()] - unknown = [f for f in requested if f not in s["catalog_by_family"]] - if unknown: - raise _TuiError( - f"Unknown family in --families: {', '.join(unknown)}. " - f"Available: {', '.join(s['catalog_by_family'])}") - chosen: Dict[str, List[dict]] = {} - 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( - s["catalog_by_family"][family]) if opt["recommended"]] - s["chosen"] = chosen - s["family_keys"] = family_keys - - def _compute_entries() -> None: - # Design task menu. Esc raises _GoBack, which the caller turns into - # Wizard.BACK (the design prompts are grouped: Esc returns to the - # families tree). - 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 - - model_entries, entry_ids, install_guidance, \ - design_entry_ids, include_clone = _build_entries( - s["family_keys"], s["chosen"], s["catalog_by_family"], - task_picker, known_tasks=s["existing_tasks"]) - s.update({ - "model_entries": model_entries, - "entry_ids": entry_ids, - "install_guidance": install_guidance, - "design_entry_ids": design_entry_ids, - "include_clone": include_clone, - }) - - def _finalize() -> dict: - return { - "audiocpp_dir": s["audiocpp_dir"], - "catalog": s["catalog"], - "catalog_by_family": s["catalog_by_family"], - "output_path": s["output_path"], - "family_keys": s["family_keys"], - "chosen": s["chosen"], - "model_entries": s["model_entries"], - "entry_ids": s["entry_ids"], - "install_guidance": s["install_guidance"], - "design_entry_ids": s["design_entry_ids"], - "include_clone": s["include_clone"], - "host": s["host"], - "port": s["port"], - "backend": s["backend"], - "build": s["build"], - "lazy_load": s["lazy_load"], - "sync_port": s["sync_port"], - "sync_model_ids": s["sync_model_ids"], - "wav_dir": s["wav_dir"], - "plan": s["plan"], - "download": s["download"], - "delete_unused": s["delete_unused"], - "unused_entries": s["unused_entries"], - } - - def screen_families(): - """Pick TTS model families and packages (the modify tree).""" - tree_families = _build_tree_families(s["catalog"]) - # Modify flow: pre-check the models an existing server.json hosts, - # so the tree opens as a "modify" list rather than a fresh one. - checked_set = set() - for family, dirs in s["existing_selected"].items(): - if family not in s["catalog_by_family"]: - continue - family_index = s["catalog"].index(s["catalog_by_family"][family]) - valid_dirs = {opt["target_directory"] - for opt in package_dir_options( - s["catalog_by_family"][family])} - for target in dirs: - if target in valid_dirs: - checked_set.add((family_index, target)) - picked = tui.checkbox_tree( - stdscr, "Select TTS model families to host", - tree_families, expand_all=args.all_packages, - back_value=_GO_BACK, checked=checked_set) - if picked is _GO_BACK: - return tui.Wizard.BACK - chosen: Dict[str, List[dict]] = {} - family_keys: List[str] = [] - for family_index, option_key in picked: - family = s["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( - s["catalog_by_family"][family])} - chosen[family] = [keyed[key] for key in chosen[family]] - s["chosen"] = chosen - s["family_keys"] = family_keys - return screen_host - - def _after_families(): - if args.families is not None: - _families_from_flag() - return screen_host - return screen_families - - def screen_host(): - """Build the model entries, then ask the bind host. - - The task/id pickers (when any) run here too and are grouped with - this screen: Esc on one of them (or on the host field) returns to - the families tree. - """ - try: - _compute_entries() - except _GoBack: - return tui.Wizard.BACK - if args.host is not None: - s["host"] = args.host - return _after_host() - host = tui.line_edit( - stdscr, "Bind host", - s["existing_host"] if isinstance(s["existing_host"], str) - else 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: - return tui.Wizard.BACK - s["host"] = host - return _after_host() - - def _after_host(): - if args.port is None: - return screen_port - s["port"] = args.port - return _after_port() - - def screen_port(): - port_text = tui.line_edit( - stdscr, "Port", - str(s["existing_port"]) if isinstance(s["existing_port"], int) - else 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: - return tui.Wizard.BACK - s["port"] = int(port_text) - return _after_port() - - def _after_port(): - s["sync_port"] = None - if s["port"] != config_port(): - return screen_sync_port - return _after_sync() - - def screen_sync_port(): - sync_port = tui.confirm( - stdscr, "Update AUDIOCPP_API_URL in app/converter/config.py " - f"to port {s['port']} so audiobook.py talks to this server", - default=True, cancel_value=_GO_BACK) - if sync_port is _GO_BACK: - return tui.Wizard.BACK - s["sync_port"] = sync_port - return _after_sync() - - def _after_sync(): - if args.build_backend: - s["backend"] = args.build_backend - s["build"] = s["detected_backend"] is None - return _after_backend() - if args.backend: - s["backend"] = args.backend - s["build"] = False - return _after_backend() - if s["detected_backend"] is not None: - # Already built: use the detected backend, no menu, no build. - s["backend"] = s["detected_backend"] - s["build"] = False - return _after_backend() - # Not built for any backend yet: always ask which backend the server - # should use and offer to build it — even on a modify run, so a user - # who declined the build the first time is never stranded without a - # way to build from the TUI. - return screen_backend - - def screen_backend(): - # Pre-select the backend an existing server.json records (modify - # flow), so re-running setup lands on the previous choice. - backend_options, backend_default = _backend_options(None) - if s["existing_backend"] in BACKENDS: - backend_default = next( - (index for index, (_label, value) in enumerate(backend_options) - if value == s["existing_backend"]), backend_default) - backend = tui.menu( - stdscr, "Which inference backend should audiocpp_server " - "use?", backend_options, - default_index=backend_default, back_value=_GO_BACK) - if backend is _GO_BACK: - return tui.Wizard.BACK - s["backend"] = backend - if built_server_binary(s["audiocpp_dir"], backend) is not None: - # A checkout with builds for several backends: this one is - # already built, so there is nothing to build. - s["build"] = False - return _after_backend() - return screen_build - - def screen_build(): - # Not built for the chosen backend yet: offer to build it now. The - # build itself runs in the TUI task view (or the console tail for - # CLI runs) after the wizard. - build = tui.confirm( - stdscr, f"audiocpp_server is not built for {s['backend']}. " - f"Build it now (runs scripts/build_*)?", - default=True, cancel_value=_GO_BACK) - if build is _GO_BACK: - return tui.Wizard.BACK - s["build"] = build - return _after_backend() - - def _after_backend(): - s["lazy_load"] = True - return _after_lazy() - - def _after_lazy(): - if args.input_dir is not None: - s["wav_dir"] = args.input_dir - return _after_wav() - if s["include_clone"]: - return screen_wav - s["wav_dir"] = None - return _after_wav() - - def screen_wav(): - wav_start = detect_wav_dir(s["audiocpp_dir"], TTS_ROOT) - # Modify flow: an existing voice_dir seeds the browser so the user - # can accept it on Enter instead of re-navigating. - if isinstance(s["existing_voice_dir"], str) and s["existing_voice_dir"]: - wav_start = Path(s["existing_voice_dir"]) - 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 VOICES_DIR, - back_value=_GO_BACK) - if wav_dir is _GO_BACK: - return tui.Wizard.BACK - s["wav_dir"] = wav_dir - return _after_wav() - - def _after_wav(): - s["plan"] = None - if s["include_clone"] and s["wav_dir"] is not None: - wav_files = find_wav_files(s["wav_dir"]) - if wav_files: - prompt_path = s["wav_dir"] / PROMPT_TEXT_FILENAME - if prompt_path.exists() and not args.force: - return screen_transcription - existing = read_prompt_text(prompt_path) if ( - prompt_path.exists() and not args.force) else {} - s["plan"] = _decide_transcription( - wav_files, existing, prompt_path.exists(), - args.force, ask_confirm) - return _after_transcription() - - def screen_transcription(): - # Transcription plan (questions only; transcription runs after). - wav_files = find_wav_files(s["wav_dir"]) - prompt_path = s["wav_dir"] / PROMPT_TEXT_FILENAME - existing = read_prompt_text(prompt_path) if ( - prompt_path.exists() and not args.force) else {} - try: - s["plan"] = _decide_transcription( - wav_files, existing, prompt_path.exists(), - args.force, ask_confirm) - except _GoBack: - return tui.Wizard.BACK - return _after_transcription() - - def _after_transcription(): - s["sync_model_ids"] = None - if len(s["entry_ids"]) == 1 and not ( - config.AUDIOCPP_MODEL_ID == s["entry_ids"][0] - and config.AUDIOCPP_CLONE_MODEL_ID == s["entry_ids"][0]): - return screen_model_sync - return _after_model_sync() - - def screen_model_sync(): - sync_model_ids = tui.confirm( - stdscr, "Update AUDIOCPP_MODEL_ID and " - "AUDIOCPP_CLONE_MODEL_ID in app/converter/config.py to " - f"'{s['entry_ids'][0]}' so audiobook.py uses this model", - default=True, cancel_value=_GO_BACK) - if sync_model_ids is _GO_BACK: - return tui.Wizard.BACK - s["sync_model_ids"] = sync_model_ids - return _after_model_sync() - - def _after_model_sync(): - new_paths = {entry["path"] for entry in s["model_entries"]} - s["unused_entries"] = unused_installed_entries( - s["output_path"], new_paths) \ - if s["existing_config"] is not None else [] - s["delete_unused"] = False - if s["unused_entries"]: - return screen_delete_unused - return _after_delete() - - def screen_delete_unused(): - delete_unused = tui.confirm( - stdscr, "Delete unused models?", default=False, - cancel_value=_GO_BACK) - if delete_unused is _GO_BACK: - return tui.Wizard.BACK - s["delete_unused"] = delete_unused - return _after_delete() - - def _after_delete(): - manager = s["audiocpp_dir"] / "tools" / "model_manager_v2.py" - if manager.is_file(): - return screen_download - s["download"] = False - return _finalize() - - def screen_download(): - # Automatic model download (or print the install commands). - try: - s["download"] = _decide_download( - s["audiocpp_dir"], s["model_entries"], ask_confirm) - except _GoBack: - return tui.Wizard.BACK - return _finalize() - - # First screen: resolve the checkout directly when it already exists - # (the modify flow), so the wizard starts on a real screen. When no - # checkout exists, clone it into ./app/audio.cpp (streaming inside the - # TUI task view, not by dropping to the console) without asking, then - # continue the same way. - audiocpp_dir = find_local_checkout() - if audiocpp_dir is None: - target = APP_DIR / AUDIOCPP_DIR_NAME - rc = taskview.run_steps(stdscr, "Clone audio.cpp", [ - taskview.TaskStep( - f"Cloning audio.cpp into {target}", - lambda emit, cancel: common.git_clone( - AUDIOCPP_GIT_URL, target, emit=emit, cancel=cancel)), - taskview.TaskStep( - "Apply ggml build patches", - lambda emit, cancel: apply_ggml_patches( - target, emit=emit, cancel=cancel)), - ]) - if rc == 130: - # Cancelled from the task view: abort the wizard quietly. - return None - if rc != 0: - raise _TuiError( - f"audio.cpp setup step failed (exit {rc}). Clone " - f"audio.cpp manually: git clone " - f"{AUDIOCPP_GIT_URL} {target}, then re-run") - audiocpp_dir = target - resolve_checkout(audiocpp_dir) - first = _after_families() - return tui.Wizard().run(first) - - -def load_server_config(server_json: Path) -> Optional[dict]: - """Read server.json into a dict, or None when it cannot be used. - - Returns None for a missing file, unreadable content, or a non-dict - document. Used by the wizard's modify flow to pre-fill its screens - from an existing config instead of prompting to overwrite it. - """ - if not server_json.exists(): - return None - try: - data = json.loads(server_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - return None - if not isinstance(data, dict): - return None - return data - - -def server_config_selections(server_config: dict, - catalog: List[dict] - ) -> Tuple[Dict[str, List[str]], - Dict[Tuple[str, str], str]]: - """Map an existing server.json's models back to catalog selections. - - Returns ``(selected_dirs, tasks)``: ``selected_dirs`` maps a catalog - family to the target directories it hosts (``models/<target>`` paths - with the ``models/`` prefix stripped, in server.json order), and - ``tasks`` maps ``(family, target_directory)`` to the entry's task - (``"tts"`` or ``"vdes"``) so the wizard can preserve how design - packages were hosted. Entries whose family is not in the CATALOG are - ignored — the wizard cannot offer them again. - """ - families = {entry["family"] for entry in catalog} - selected_dirs: Dict[str, List[str]] = {} - tasks: Dict[Tuple[str, str], str] = {} - for entry in server_config.get("models") or []: - if not isinstance(entry, dict): - continue - family = entry.get("family") - if not isinstance(family, str) or family not in families: - continue - path = entry.get("path") - if not isinstance(path, str): - continue - target = path[len("models/"):] if path.startswith("models/") else path - if family not in selected_dirs: - selected_dirs[family] = [] - if target not in selected_dirs[family]: - selected_dirs[family].append(target) - tasks[(family, target)] = str(entry.get("task") or TASK_TTS) - return selected_dirs, tasks - - -def _model_path_present(path: Path) -> bool: - """True when a server.json model path holds actual model files. - - A present path is either a file (a single-model package) or a non-empty - directory (the usual GGUF package target directory; an empty one means a - download that never ran or was cleaned up halfway). - """ - try: - if path.is_file(): - return True - if path.is_dir(): - return any(path.iterdir()) - except OSError: - return False - return False - - -def _all_models_present(audiocpp_dir: Path, model_entries: List[dict]) -> bool: - """True when every selected model entry's path already holds files on disk. - - Paths resolve against the checkout (where model_manager_v2.py installs - them), honoring absolute paths. Used by the wizard to skip the - "Automatically download the selected models" prompt when nothing is - actually missing. An empty selection is treated as not-present. - """ - if not model_entries: - return False - for entry in model_entries: - rel = entry.get("path") - if not isinstance(rel, str) or not rel: - return False - path = Path(rel) if Path(rel).is_absolute() else audiocpp_dir / rel - if not _model_path_present(path): - return False - return True - - -def missing_model_entries(server_json: Path) -> List[dict]: - """Return the server.json model entries whose files are not on disk. - - Paths resolve exactly like audiocpp_server resolves them (relative paths - against the server.json's directory). Each returned entry carries the - entry ``id`` and ``rel`` (the configured path string); used by ``detect`` - to warn that a conversion would fail until the models are installed. - """ - try: - data = json.loads(server_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - return [] - if not isinstance(data, dict): - return [] - base = server_json.parent - missing: List[dict] = [] - for entry in data.get("models") or []: - if not isinstance(entry, dict): - continue - rel = entry.get("path") - if not isinstance(rel, str) or not rel: - continue - path = Path(rel) if Path(rel).is_absolute() else base / rel - if _model_path_present(path): - continue - missing.append({"id": str(entry.get("id") or rel), "rel": rel}) - return missing - - -def _install_id_by_path(audiocpp_dir: Path) -> Dict[str, str]: - """Map ``models/<target_directory>`` -> catalog install id. - - The catalog package that installs a model is derived from the - ``default_path`` of each TTS family; an entry whose path matches no - catalog package has no install id. - """ - by_path: Dict[str, str] = {} - try: - for entry in load_model_catalog(audiocpp_dir): - by_path[entry["default_path"]] = entry["install_id"] - except (NotADirectoryError, OSError): - pass - return by_path - - -def installed_model_entries(server_json: Path) -> List[dict]: - """Return the server.json model entries whose files ARE on disk. - - The complement of ``missing_model_entries``: each returned entry carries - the entry ``id`` and ``rel`` (the configured path string), resolved - exactly like ``missing_model_entries`` (relative against the server.json's - directory). Used by the wizard's "Delete unused models?" step to find - already-downloaded models that were unselected. - """ - try: - data = json.loads(server_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - return [] - if not isinstance(data, dict): - return [] - base = server_json.parent - installed: List[dict] = [] - for entry in data.get("models") or []: - if not isinstance(entry, dict): - continue - rel = entry.get("path") - if not isinstance(rel, str) or not rel: - continue - path = Path(rel) if Path(rel).is_absolute() else base / rel - if _model_path_present(path): - installed.append({"id": str(entry.get("id") or rel), "rel": rel}) - return installed - - -def missing_model_install_guidance(audiocpp_dir: Path, - missing: List[dict]) -> List[Tuple[str, str]]: - """Map MISSING model entries to (display name, install id) pairs. - - The install id is derived from each entry's configured path via the - catalog (see ``_install_id_by_path``); entries whose path matches no - catalog package are skipped (there is no ``model_manager_v2.py install`` - command for them). Feeds ``_install_models`` for the "Download Missing - Models" action. - """ - by_path = _install_id_by_path(audiocpp_dir) - guidance: List[Tuple[str, str]] = [] - for item in missing: - install_id = by_path.get(item["rel"]) - if install_id: - guidance.append((item["id"], install_id)) - return guidance - - -def model_install_hints(audiocpp_dir: Path, - missing: List[dict]) -> List[str]: - """Remediation lines for MISSING model entries (see missing_model_entries). - - Maps each entry's configured path back to the catalog package that - installs it (``models/<target_directory>`` -> install id) so the line - carries the exact ``model_manager_v2.py install`` command; entries whose - directory matches no catalog package just name the path. - """ - by_path = _install_id_by_path(audiocpp_dir) - hints: List[str] = [] - for item in missing: - install_id = by_path.get(item["rel"]) - hint = f"model not downloaded: {item['id']} ({item['rel']})" - if install_id: - hint += (f" — install with: python tools/model_manager_v2.py " - f"install {install_id}") - hints.append(hint) - return hints - - -def install_models(audiocpp_dir: Path, - guidance: List[Tuple[str, str]], - emit=None, cancel=None) -> int: - """Download the (display name, install id) models via the helper script. - - Runs ``model_manager_v2.py install`` for each de-duped install id in the - checkout, streaming to the console (or to EMIT, the in-TUI task view); a - failing install is reported as a warning and does not abort the rest. - Returns 0 when every download succeeded, 130 when cancelled, 1 when any - failed. Used by the hub's "Download Missing Models" action (see - ``missing_model_install_guidance`` for the mapping). - """ - return _install_models(audiocpp_dir, guidance, download=True, - emit=emit, cancel=cancel) - - -def hand_install_guidance(audiocpp_dir: Path, - missing: List[dict]) -> str: - """Explain how to install MISSING model entries by hand. - - Returns a multi-line message listing each missing model's id and the - path its files must be placed in (``rel``, resolved against the - checkout). Used when the missing models cannot be mapped to - a ``model_manager_v2.py install`` command, so the user still knows what - to download and where to put it. - """ - lines = [ - "None of the missing models map to a model_manager_v2.py install " - "command.", - "Download them by hand and place the files at these paths:", - ] - for item in missing: - lines.append(f" {item['id']} -> {item['rel']}") - lines.append(f"(paths are relative to {audiocpp_dir})") - return "\n".join(lines) - - -def unused_installed_entries(server_json: Path, - new_paths: Set[str]) -> List[dict]: - """Return installed server.json entries whose path is not in NEW_PATHS. - - The already-downloaded models (see ``installed_model_entries``) that the - new selection does not host any more — the candidates for the wizard's - "Delete unused models?" prompt. Entries whose files are not on disk are - never listed (there is nothing to delete). - """ - return [entry for entry in installed_model_entries(server_json) - if entry["rel"] not in new_paths] - - -def delete_model_files(server_json: Path, entries: List[dict]) -> int: - """Remove the on-disk model files for ENTRIES ({id, rel}) from disk. - - Each entry's ``rel`` is resolved exactly like the server resolves it - (relative against ``server_json``'s directory; absolute paths honored), - then removed as a directory tree or a single file. Missing entries are - ignored. Returns the number of paths removed. Used by the wizard's - "Delete unused models?" step — the regenerated server.json already only - lists the kept models, so no entry cleanup is needed here. - """ - base = server_json.parent - removed = 0 - for item in entries: - rel = item.get("rel") - if not isinstance(rel, str) or not rel: - continue - path = Path(rel) if Path(rel).is_absolute() else base / rel - try: - if not path.exists(): - continue - if path.is_dir(): - shutil.rmtree(path, ignore_errors=True) - else: - path.unlink() - except OSError as exc: - print(f"[WARNING] Could not remove {path}: {exc}") - continue - print(f"[OK] Removed unused model {path}") - removed += 1 - return removed - - -def uninstall(*, emit=None, cancel=None) -> int: - """Remove the audio.cpp backend entirely: stop its server, delete the checkout. - - The checkout (``app/audio.cpp``) holds the built binary, the downloaded - models, and the server.json, so removing the directory uninstalls the - backend. A running server this tool started is stopped first - (best-effort). - - EMIT is accepted for registry symmetry with the other backends but is - unused here — this uninstall has no subprocess phase, and its prints are - captured by the task view when run in the TUI. CANCEL is a - ``threading.Event`` honored between phases only (after the server has - been stopped, before the checkout is deleted), so a started phase always - completes and the uninstall never tears halfway. Returns the exit code - (130 when cancelled before a remaining phase). - """ - # Only stop when a pid file exists: without one this tool never - # started the server, so the "not started by this tool" notice would - # be uninstall-time noise. - if servers.pid_for("audiocpp") is not None: - servers.stop("audiocpp") - if common.cancel_requested(cancel): - return 130 - checkout = find_local_checkout() - if checkout is None: - print("[INFO] No audio.cpp checkout to remove.") - return 0 - print(f"[INFO] Removing audio.cpp checkout {checkout}...") - shutil.rmtree(checkout, ignore_errors=True) - print("[OK] audio.cpp removed.") - return 0 - - -def find_local_checkout() -> Optional[Path]: - """Return the managed audio.cpp checkout at ``app/audio.cpp``. - - Returns the path only when it contains a ``model_specs`` directory; - the checkout is installed there by the setup wizard and nowhere else. - """ - try: - resolved = (APP_DIR / AUDIOCPP_DIR_NAME).resolve() - except OSError: - return None - if (resolved / "model_specs").is_dir(): - return resolved - return None - - -def fetch_server_models(api_url: str) -> Optional[List[Dict[str, str]]]: - """List a running audiocpp_server's model entries via GET /v1/models. - - Returns ``[{id, family, task}, ...]`` — the same shape the converter's - client resolves at startup — or None when URL does not answer with a - valid document (wrong server, still starting, older audio.cpp). Used by - the hub to drive the convert menus against a remote server that has no - local server.json describing it. - """ - try: - with urllib.request.urlopen( - f"{api_url.rstrip('/')}/v1/models", timeout=10) as response: - payload = json.loads(response.read().decode("utf-8")) - except (OSError, ValueError): - # URLError/HTTPError/socket errors are OSErrors; a non-JSON body is - # a ValueError. Anything else means "not an audiocpp_server". - return None - entries = payload.get("data") if isinstance(payload, dict) else None - models: List[Dict[str, str]] = [] - for entry in entries or []: - if isinstance(entry, dict) and entry.get("id"): - models.append({ - "id": str(entry["id"]), - "family": str(entry.get("family") or ""), - "task": str(entry.get("task") or ""), - }) - return models - - -def fetch_server_voices(api_url: str, model_id: str) -> Optional[List[str]]: - """List a running audiocpp_server's voices for MODEL_ID. - - Queries ``GET /v1/audio/voices?model=<id>`` — the endpoint the converter - validates ``--voice`` against — and returns its voice-name list, or None - when the server cannot be queried. Lets the hub offer a remote server's - voices without reading its configuration locally. - """ - query = urllib.parse.urlencode({"model": model_id}) - try: - with urllib.request.urlopen( - f"{api_url.rstrip('/')}/v1/audio/voices?{query}", - timeout=10) as response: - payload = json.loads(response.read().decode("utf-8")) - except (OSError, ValueError): - return None - voices = payload.get("voices") if isinstance(payload, dict) else None - if not isinstance(voices, list): - return None - return [str(voice) for voice in voices] - - -def find_audiocpp_server_bin(audiocpp_dir: Path) -> Optional[Path]: - """Return the built audiocpp_server binary, or None when not built. - - Scans ``audiocpp_dir/build/*`` for a build directory containing - ``bin/audiocpp_server`` (``.exe`` allowed on Windows). When several - builds exist the first (alphabetical) is returned. - """ - build_root = audiocpp_dir / "build" - if not build_root.is_dir(): - return None - 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 - for name in ("audiocpp_server", "audiocpp_server.exe"): - server = build_dir / "bin" / name - if server.exists(): - return server - return None - - -def built_server_binary(audiocpp_dir: Path, backend: str) -> Optional[Path]: - """Return the built audiocpp_server for BACKEND, or None. - - Like ``find_audiocpp_server_bin`` but limited to build directories whose - name carries the BACKEND token (``-cuda-``, ``-vulkan-``, ``-hip-``, - ``-cpu-``; ``-metal-`` counts as ``cpu``). A checkout with builds for - several backends is asked which one to use without re-offering a build - for a backend that is already built. - """ - build_root = audiocpp_dir / "build" - if not build_root.is_dir(): - return None - 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 - match = _BACKEND_TOKEN_RE.search(build_dir.name.lower()) - if not match: - continue - token = "cpu" if match.group(1) == "metal" else match.group(1) - if token != backend: - continue - for name in ("audiocpp_server", "audiocpp_server.exe"): - server = build_dir / "bin" / name - if server.exists(): - return server - return None - - -def find_build_script(audiocpp_dir: Path) -> Optional[Path]: - """Return the audio.cpp build helper script to run, or None. - - Prefers ``scripts/build_linux.sh``; otherwise the first - ``scripts/build_*.sh`` it finds. (Windows ``.bat`` scripts are not run - automatically — build manually there.) - """ - scripts = audiocpp_dir / "scripts" - if not scripts.is_dir(): - return None - preferred = scripts / "build_linux.sh" - if preferred.exists(): - return preferred - try: - candidates = sorted(scripts.glob("build_*.sh"), - key=lambda p: p.name.lower()) - except OSError: - return None - return candidates[0] if candidates else None - - -# Each entry pairs a shipped patch with the vendored file it touches and a -# regex marker proving the fix is already present (so the patch is skipped -# idempotently once applied, or once the fork re-vendors a fixed ggml). -GGML_PATCHES = [ - { - "file": "ggml-top-k-cuda-iterator.patch", - "target": "external/ggml/src/ggml-cuda/top-k.cu", - "marker": r"#\s*include\s*<cuda/iterator>", - "label": "top-k.cu: add #include <cuda/iterator> (CCCL 3.x build fix)", - }, -] - - -def apply_ggml_patches(audiocpp_dir: Path, *, emit=None, cancel=None) -> int: - """Apply the shipped ggml build patches to an audio.cpp checkout. - - Idempotent: a patch whose marker already matches its target is skipped - (it is either already applied, or the fork re-vendored a fixed ggml). A - patch that no longer applies because the vendored file changed shape is a - loud, non-interactive failure — the build is aborted so the user - re-evaluates the patch instead of hitting a known nvcc break minutes - later. Returns 0 when every patch is applied or already present, 1 on - drift, 130 when cancelled. - """ - for patch in GGML_PATCHES: - if cancel is not None and cancel.is_set(): - return 130 - target = audiocpp_dir / patch["target"] - if not target.is_file(): - print(f"[INFO] {patch['file']}: target {patch['target']} not " - f"present in this checkout; skipping") - continue - try: - text = target.read_text(encoding="utf-8", errors="ignore") - except OSError as exc: - print(f"[WARNING] {patch['file']}: could not read {target}: " - f"{exc}; skipping") - continue - if re.search(patch["marker"], text): - print(f"[OK] {patch['file']}: fix already present, skipping") - continue - patch_path = PATCH_DIR / patch["file"] - if not patch_path.is_file(): - print(f"[ERROR] {patch['file']}: patch file not found at " - f"{patch_path}; cannot apply") - return 1 - check_argv = ["git", "-C", str(audiocpp_dir), "apply", "--check", - "--whitespace=nowarn", str(patch_path)] - check_rc = common.run_console_subprocess( - check_argv, emit=emit, cancel=cancel) - if check_rc == 130 or (cancel is not None and cancel.is_set()): - return 130 - if check_rc != 0: - print(f"[ERROR] {patch['file']}: no longer applies to " - f"{patch['target']} (git apply --check exit {check_rc}). " - f"The audio.cpp fork's vendored ggml changed shape and " - f"still lacks the fix. Re-evaluate {patch_path}: " - f"regenerate the patch, or drop this entry if the fork " - f"now ships the fix.") - return 1 - apply_argv = ["git", "-C", str(audiocpp_dir), "apply", - "--whitespace=nowarn", str(patch_path)] - rc = common.run_console_subprocess( - apply_argv, emit=emit, cancel=cancel) - if rc == 130 or (cancel is not None and cancel.is_set()): - return 130 - if rc != 0: - print(f"[ERROR] {patch['file']}: git apply failed (exit {rc})") - return rc - print(f"[OK] {patch['file']}: applied ({patch['label']})") - return 0 - - -def build_audiocpp(audiocpp_dir: Path, backend: str, *, - emit=None, cancel=None) -> int: - """Build audiocpp_server for BACKEND, streaming output. - - With EMIT None the build script runs on the console (inherits the - terminal); with EMIT given (the in-TUI task view) its output streams line - by line to EMIT so the view can show progress, and CANCEL aborts it. - - On the EMIT (TUI) path the build output is also tee'd to - ``app/logs/audiocpp_build_<timestamp>.log`` so it survives the curses - session; when the build fails (and was not cancelled) a post-TUI notice - with the copy-pastable command and the log path is queued for the console - (see ``backends.common.record_post_tui_notice``). - - Returns the build script's exit code (non-zero when the script is - missing). - """ - script = find_build_script(audiocpp_dir) - if script is None: - message = (f"[ERROR] No build script found in {audiocpp_dir}/scripts; " - "build audiocpp_server manually (see the audio.cpp README)") - print(message) - if emit is not None: - common.record_post_tui_notice(message) - return 1 - argv = ["sh", str(script), "--backend", backend, "--target", - "audiocpp_server", "--deployment-build"] - command = f"cd {audiocpp_dir} && {shlex.join(argv)}" - if emit is None: - print(f"[INFO] Building audiocpp_server for {backend} ({command})...") - patch_rc = apply_ggml_patches(audiocpp_dir, cancel=cancel) - if patch_rc == 130 or (cancel is not None and cancel.is_set()): - return 130 - if patch_rc != 0: - print("[ERROR] ggml build patches could not be applied; " - "aborting audiocpp_server build. See the messages above " - "and re-evaluate app/backends/patches/.") - return patch_rc - return common.run_console_subprocess(argv, cwd=audiocpp_dir) - return _build_audiocpp_tui(emit, cancel, argv, command, audiocpp_dir) - - -def _build_audiocpp_tui(emit, cancel, argv: List[str], command: str, - audiocpp_dir: Path) -> int: - """Run the build on the TUI path: tee output to a log file. - - The ggml patch step runs first, inside the same log: every emitted - line (patch status, build output) is also written (and flushed) to - ``app/logs/audiocpp_build_<timestamp>.log``. On failure a summary (the - copy-pastable COMMAND and the log path) is emitted into the TUI, - written to the log, and queued as a post-TUI console notice. A - cancelled build (CANCEL set) is not reported as a failure, but its - partial output stays in the log file. - """ - log_path = common.LOG_DIR / ( - f"audiocpp_build_{datetime.now():%Y%m%d_%H%M%S}.log") - log_path.parent.mkdir(parents=True, exist_ok=True) - log_handle = log_path.open("w", encoding="utf-8") - - def tee(line: str) -> None: - log_handle.write(line + "\n") - log_handle.flush() - emit(line) - - class _TeeWriter(io.TextIOBase): - """Route print() output from the patch step into the log too.""" - - def write(self, s: str) -> int: - for line in s.splitlines(): - if line: - tee(line) - return len(s) - - try: - with contextlib.redirect_stdout(_TeeWriter()): - patch_rc = apply_ggml_patches(audiocpp_dir, emit=tee, - cancel=cancel) - if patch_rc == 130 or (cancel is not None and cancel.is_set()): - return 130 - if patch_rc != 0: - notice = ("[ERROR] ggml build patches could not be applied; " - "aborting audiocpp_server build. See the messages " - "above and re-evaluate app/backends/patches/.") - tee(notice) - common.record_post_tui_notice(notice) - return patch_rc - tee(f"[INFO] Building audiocpp_server ({command})...") - rc = common.run_console_subprocess( - argv, cwd=audiocpp_dir, emit=tee, cancel=cancel) - if rc != 0 and (cancel is None or not cancel.is_set()): - notice = (f"[ERROR] audio.cpp build failed (exit code {rc}).\n" - f" Build log: {log_path}\n" - f" Troubleshoot by re-running this command:\n" - f" {command}") - for line in notice.splitlines(): - tee(line) - common.record_post_tui_notice(notice) - finally: - log_handle.close() - return rc - - -def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None: - """Print remediation when audiocpp_server is missing (troubleshooting). - - The hub starts and stops the server itself, so a working install gets - no manual launch instructions. When no binary was built, though, the - user needs to know how to build and run it by hand. The commands are - prefixed with ``cd <checkout> &&`` because the server discovers - model_specs/<family>.json relative to its working directory. - """ - if find_audiocpp_server_bin(audiocpp_dir) is not None: - return - print("\n[INFO] audiocpp_server binary not found. Build it first, e.g.:") - script = find_build_script(audiocpp_dir) - if script is not None: - print(f" sh {script} --backend <cuda|vulkan|hip|cpu> " - "--target audiocpp_server --deployment-build") - print(f" then run: cd {audiocpp_dir} && ./build/<platform>-<backend>" - f"-release/bin/audiocpp_server --config {output_path}") - - -def _execute_lanes(settings: dict, - args: argparse.Namespace) -> List[taskview.TaskLane]: - """Build the ordered setup steps for the in-TUI task view, per lane. - - The same work ``_execute`` runs on the console, split into two lanes so - the view can run the build in one pane while configuring and downloading - models in the other (both progress bars visible at once). The build lane - exists only when ``settings["build"]`` is set; the models lane always - exists (transcribe → write server.json → download/print commands). - Shared results (the transcription mapping) travel through a small closure - dict scoped to the models lane. Each step's ``work(emit, cancel)`` - returns its exit code; subprocess steps stream through EMIT and abort on - CANCEL, while print()-based steps are captured by the view's stdout - routing. - """ - audiocpp_dir = settings["audiocpp_dir"] - state: dict = {} - build = settings.get("build") - lanes: List[taskview.TaskLane] = [] - - if build: - def build_step(emit, cancel): - rc = build_audiocpp(audiocpp_dir, settings["backend"], - emit=emit, cancel=cancel) - if rc != 0: - print(f"[WARNING] build exited with code {rc}; the server.json " - "was still written — build audiocpp_server manually " - "before starting it") - else: - print("[OK] build complete") - return rc - lanes.append(taskview.TaskLane( - "Build", - [taskview.TaskStep( - f"Build audiocpp_server ({settings['backend']})", - build_step)])) - - def transcribe(emit, cancel): - args.input_dir = settings["wav_dir"] - if settings["include_clone"] and args.input_dir is not None: - transcripts, write_prompt = _transcribe( - args, plan=settings["plan"], cancel=cancel) - 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 - state["transcripts"] = transcripts - state["write_prompt"] = write_prompt - return 0 - - def write(emit, cancel): - # 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) - - _write_and_advise( - audiocpp_dir, settings["wav_dir"], settings["output_path"], - settings["model_entries"], settings["install_guidance"], - settings["host"], settings["port"], settings["backend"], - settings["lazy_load"], state["transcripts"], state["write_prompt"]) - - # Delete-unused cleanup (modify flow): remove the already-downloaded - # models the new selection dropped. The regenerated server.json - # already only lists the kept entries. - if settings.get("delete_unused"): - removed = delete_model_files(settings["output_path"], - settings["unused_entries"]) - print(f"[OK] Deleted {removed} unused model " - f"{'entry' if removed == 1 else 'entries'} from disk.") - - if len(settings["entry_ids"]) == 1: - _offer_config_model_id_sync(settings["entry_ids"][0], - settings["sync_model_ids"]) - print_empty_transcript_warning(state["transcripts"]) - return 0 - - def install(emit, cancel): - _install_models(audiocpp_dir, settings["install_guidance"], - settings["download"], emit=emit, cancel=cancel) - _print_launch_hint(audiocpp_dir, settings["output_path"]) - return 0 - install_title = "Download models" if settings.get("download") \ - else "Print model install commands" - - lanes.append(taskview.TaskLane( - "Configure & download", - [taskview.TaskStep("Transcribe reference voices", transcribe), - taskview.TaskStep("Write server.json & sync config", write), - taskview.TaskStep(install_title, install)])) - - return lanes - - -def _execute_steps(settings: dict, - args: argparse.Namespace) -> List[taskview.TaskStep]: - """The ordered setup steps for the sequential console path. - - The lanes ``_execute_lanes`` builds, flattened into one ordered list - (build first, then transcribe → write → download), so the console tail - is byte-identical to the pre-lanes behavior. - """ - steps: List[taskview.TaskStep] = [] - for lane in _execute_lanes(settings, args): - steps.extend(lane.steps) - return steps - - -def _execute(settings: dict, args: argparse.Namespace) -> int: - """Shared console tail: build, sync, transcribe, write, install, advise. - - Runs after the TUI wizard returns (or after _collect_from_flags for a - non-interactive run): the terminal is plain, so subprocess output and - transcription progress appear normally. The same work as - ``_execute_steps``, run with no emit (console streaming). - """ - return taskview.run_steps_inline(_execute_steps(settings, args)) - - -def setup_screen(stdscr) -> int: - """Run the setup wizard on an existing curses screen (the hub's). - - The hub drives this as one screen of its own ``tui.Wizard`` stack, so - Esc on the wizard's first screen simply returns here and the hub pops - back to the menu that launched it. The setup tail (build, transcribe, - write, download) runs inside the TUI task view on this same screen, so - the hub's curses session stays intact and the user sees per-step status - and progress instead of being dropped to the console. On a fresh install - the build and the model setup run as two parallel lanes (a split view), - so cloning → configuring → building+downloading is one continuous, - one-click flow; the individual "Build" and "Download Missing Models" hub - actions remain only as fallbacks when something fails or is interrupted. - Returns 0 on completion, 1 when the user aborted. - """ - parser = build_parser() - args = parser.parse_args([]) - settings = _wizard(stdscr, args, parser) - if settings is None: - return 1 - return taskview.run_lanes(stdscr, "Setting up audio.cpp", - _execute_lanes(settings, args)) - - -def build_screen(stdscr) -> int: - """Build audiocpp_server from the hub when the checkout has no binary. - - Asks which backend to build for (pre-selecting the backend an existing - server.json records, else cuda), runs the build inside the TUI task view - — alongside a download of any missing models when server.json is already - configured and those models map to an install command (the split view), - or just the build otherwise — then updates server.json's ``backend`` - field to match. Returns 0 on success, non-zero when the user backed out, - cancelled, or the build failed. This is the hub's "Build audio.cpp - server" action, so a checkout that was cloned but never built is always - buildable from the TUI; the standalone "Download Missing Models" action - stays as the fallback when the download fails or is interrupted. - """ - checkout = find_local_checkout() - if checkout is None: - tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp " - "first.", "err") - return 1 - if find_audiocpp_server_bin(checkout) is not None: - tui.flash(stdscr, "audiocpp_server is already built.", "ok") - return 0 - server_config = load_server_config(checkout / "server.json") or {} - recorded = server_config.get("backend") - options, default = _backend_options(None) - if recorded in BACKENDS: - default = next((i for i, (_label, value) in enumerate(options) - if value == recorded), default) - backend = tui.menu( - stdscr, "Which inference backend should audiocpp_server be built " - "for?", options, default_index=default, back_value=_GO_BACK) - if backend is _GO_BACK: - return 1 - - def build_step(emit, cancel): - return build_audiocpp(checkout, backend, emit=emit, cancel=cancel) - - lanes = [taskview.TaskLane( - "Build", [taskview.TaskStep( - f"Build audiocpp_server ({backend})", build_step)])] - - # Missing models this build can also fetch, so a configured backend that - # lost its binary is restored to "installed" in one step. - server_json = checkout / "server.json" - missing = missing_model_entries(server_json) if server_json.exists() else [] - guidance = missing_model_install_guidance(checkout, missing) \ - if missing else [] - - if guidance: - def download_step(emit, cancel): - install_models(checkout, guidance, emit=emit, cancel=cancel) - return 0 - lanes.append(taskview.TaskLane( - "Download models", - [taskview.TaskStep("Download missing models", download_step)])) - - title = "Build & download models" if len(lanes) == 2 \ - else "Build audiocpp_server" - rc = taskview.run_lanes(stdscr, title, lanes) - if rc != 0: - return rc - if update_server_backend(backend): - tui.flash(stdscr, f"audiocpp_server built for {backend}.", "ok") - else: - tui.flash(stdscr, f"audiocpp_server built for {backend}. (Could not " - "update server.json's backend field — reconfigure audio.cpp " - "if it was already configured.)", "warn") - # Models that can't be mapped to an install command still need hand - # installation; say so now rather than leaving the user in the dark. - if missing and not guidance: - tui.flash(stdscr, hand_install_guidance(checkout, missing), "err") - return 0 - - -def update_server_backend(backend: str) -> bool: - """Rewrite the 'backend' in the checkout's server.json, or True when none. - - Sets ``backend`` to BACKEND in ``<checkout>/server.json`` (same - ``json.dump`` formatting as the wizard). Returns True when the file now - carries BACKEND, when there is no server.json (nothing to sync), or when - it already does; False when the file exists but cannot be read/written. - """ - checkout = find_local_checkout() - if checkout is None: - return True - server_json = checkout / "server.json" - if not server_json.exists(): - return True - try: - data = json.loads(server_json.read_text(encoding="utf-8")) - except (OSError, ValueError): - return False - if not isinstance(data, dict): - return False - if data.get("backend") == backend: - return True - data["backend"] = backend - try: - with server_json.open("w", encoding="utf-8") as handle: - json.dump(data, handle, indent=2, ensure_ascii=False) - handle.write("\n") - except OSError: - return False - return True - - -def run_tui(args: Optional[argparse.Namespace] = None, - parser: Optional[argparse.ArgumentParser] = None) -> int: - """Run the audio.cpp setup wizard end-to-end. - - With no ARGS (the hub's call) a default namespace is built so the full - wizard runs. Called from ``main`` after argparse when the terminal is - interactive. Returns the process exit code. - """ - import curses - if args is None: - parser = build_parser() - args = parser.parse_args([]) - if args.input_dir is not None and not args.input_dir.is_dir(): - print(f"[ERROR] --wavs not found: {args.input_dir}", - file=sys.stderr) - return 2 - 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 - 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 - return _execute(settings, args) - - -def _collect_from_flags(args: argparse.Namespace, - parser: argparse.ArgumentParser) -> Optional[dict]: - """Build the settings dict from flags for a non-interactive run. - - Every required value must come from a flag (there are no prompts in a - non-interactive run); a missing one is a hard ``parser.error``. Returns - the settings dict, or None when the user declined an overwrite (the - default-location fallback then also exists). - """ - # Checkout: ./app/audio.cpp, else --clone clones one there. - audiocpp_dir = find_local_checkout() - if audiocpp_dir is None and args.clone: - target = APP_DIR / AUDIOCPP_DIR_NAME - rc = common.git_clone(AUDIOCPP_GIT_URL, target) - if rc != 0: - parser.error(f"git clone failed (exit {rc}); clone audio.cpp " - f"manually: git clone {AUDIOCPP_GIT_URL} {target}") - patch_rc = apply_ggml_patches(target) - if patch_rc != 0: - parser.error( - f"ggml build patches could not be applied to {target} " - f"(exit {patch_rc}); see messages above. The audio.cpp " - f"fork's vendored ggml may have changed — re-evaluate " - f"app/backends/patches/.") - audiocpp_dir = target - if audiocpp_dir is None: - parser.error( - "An audio.cpp checkout is required. Pass --clone to clone " - "app/audio.cpp, or run without flags for the TUI wizard.") - try: - catalog = load_model_catalog(audiocpp_dir) - except NotADirectoryError as exc: - parser.error(str(exc)) - if not catalog: - parser.error( - f"No TTS model families found in {audiocpp_dir}/model_specs; " - "check the checkout is up to date") - catalog_by_family = {entry["family"]: entry for entry in catalog} - - # Families: required from --families in a non-interactive run. - if args.families is None: - parser.error("--families is required in a non-interactive run (or run " - "without flags for the TUI wizard)") - 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: - parser.error( - f"Unknown family in --families: {', '.join(unknown)}. " - f"Available: {', '.join(catalog_by_family)}") - family_keys: List[str] = [] - for fam in requested: - if fam not in family_keys: - family_keys.append(fam) - - chosen: Dict[str, List[dict]] = {} - for family in family_keys: - opts = package_dir_options(catalog_by_family[family]) - if args.all_packages: - chosen[family] = opts - else: - chosen[family] = [opt for opt in opts if opt["recommended"]] - - # Non-interactive picker: design packages default to vdes. - def task_picker(install_id: str) -> str: - return TASK_VDES - - model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ - _build_entries(family_keys, chosen, catalog_by_family, - task_picker) - - # Server settings. - host = args.host or DEFAULT_HOST - detected_backend = detect_backend(audiocpp_dir) - if args.build_backend: - backend = args.build_backend - build = detected_backend is None - elif args.backend: - backend = args.backend - build = False - elif detected_backend is not None: - backend = detected_backend - build = False - else: - backend = "cuda" - build = False - port = args.port if args.port is not None else config_port() - lazy_load = True - - # Output path / overwrite (decline falls back to cwd, then aborts). - output_path = args.output if args.output is not None \ - else audiocpp_dir / "server.json" - if output_path.exists() and not args.force: - if args.output is None: - output_path = Path.cwd() / "server.json" - if output_path.exists() and not args.force: - print("[INFO] Aborted; existing server.json kept") - return None - else: - print("[INFO] Aborted; existing server.json kept") - return None - - # Config sync decisions (auto-apply unless explicitly declined). - sync_port: Optional[bool] = None - if port != config_port(): - sync_port = not args.no_sync_port - 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 = not args.no_sync_model_ids - - # Wav dir + transcription plan (defaults to the project's voices/ dir). - wav_dir = args.input_dir if args.input_dir is not None else VOICES_DIR - 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 - plan = _flag_plan(wav_files, prompt_path, args.force) - - 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, - "build": build, - "lazy_load": lazy_load, - "sync_port": sync_port, - "sync_model_ids": sync_model_ids, - "wav_dir": wav_dir, - "plan": plan, - "download": args.download, - } - - -def build_parser() -> argparse.ArgumentParser: - """The audio.cpp setup CLI (also used to build a default namespace).""" - parser = argparse.ArgumentParser( - description="Set up the audio.cpp TTS backend: clone/build, pick " - "models, write server.json, and sync app/converter/config.py.") - 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 " - f"(default: {VOICES_DIR}; asked for when omitted " - "in the TUI)") - parser.add_argument("--output", type=Path, default=None, - help="Output path for server.json (default: " - "server.json inside the audio.cpp checkout; an " - "existing file is overwritten only with --force " - "or a TUI confirm)") - parser.add_argument("--clone", action="store_true", - help="Non-interactive: clone audio.cpp into " - "./app/audio.cpp when no checkout is found") - 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). Required in a " - "non-interactive run; skips the family tree in " - "the TUI") - parser.add_argument("--all-packages", action="store_true", - help="Host every installable package of each selected " - "family (distinct target_directory) instead of " - "only the recommended one. Voice-design packages " - "are hosted with task 'vdes'") - 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, - help="Port for the server (default: the port in " - "AUDIOCPP_API_URL from app/converter/config.py)") - parser.add_argument("--backend", choices=BACKENDS, default=None, - help="Inference backend recorded in server.json " - "(default: auto-detected from the checkout's " - "build/ directory, else cuda)") - parser.add_argument("--build-backend", choices=BACKENDS, default=None, - help="Build audiocpp_server for this backend when it " - "is not built yet, and use it in server.json") - parser.add_argument("--whisper-model", type=str, default="base", - help="Whisper model size for transcription " - "(default: base)") - parser.add_argument("--force", action="store_true", - help="Overwrite the output file (and prompt_text) " - "without prompting; in the TUI, start the " - "wizard fresh instead of loading the existing " - "server.json") - parser.add_argument("--download", action="store_true", - help="Run model_manager_v2.py install for each hosted " - "model automatically (default: print the commands " - "only)") - parser.add_argument("--no-sync-port", action="store_true", - help="Do not rewrite AUDIOCPP_API_URL in " - "app/converter/config.py when --port differs") - parser.add_argument("--no-sync-model-ids", action="store_true", - help="Do not rewrite AUDIOCPP_MODEL_ID/" - "AUDIOCPP_CLONE_MODEL_ID for a single-entry server") - return parser - - -def detect() -> BackendStatus: - """Detect how far audio.cpp is set up, plus the command to start it.""" - checkout = find_local_checkout() - details: List[str] = [] - launch = "" - if checkout is None: - # No local checkout: only a remote server can make this usable. - remote = _detect_remote() - return BackendStatus("audiocpp", "audio.cpp", installed=False, - configured=False, running=remote[0], - remote=remote[0], remote_urls=remote[1], - details=["not cloned — run setup to clone " - "./app/audio.cpp"]) - details.append(f"checkout: {checkout}") - binary = find_audiocpp_server_bin(checkout) - built = binary is not None - if built: - details.append(f"built: {binary}") - else: - details.append("not built — run setup to build audiocpp_server") - server_json = checkout / "server.json" - configured = server_json.exists() - specs: List[ServerSpec] = [] - missing = missing_model_entries(server_json) if configured else [] - if configured: - details.append(f"config: {server_json}") - if missing: - # The config references model files that are not on disk; a - # conversion would fail at model-load time, so say so now. - details.extend(model_install_hints(checkout, missing)) - if built: - # Spawned from the checkout: audiocpp_server discovers - # model_specs/<family>.json relative to its working directory. - specs = [ServerSpec( - "audiocpp", config.AUDIOCPP_API_URL, - [str(binary), "--config", str(server_json)], - cwd=checkout, identity=probe.IDENTITY_AUDIOCPP)] - else: - launch = (f"cd {checkout} && ./build/<platform>-<backend>-release" - f"/bin/audiocpp_server --config {server_json}") - else: - details.append("no server.json — run setup to configure models") - if specs: - launch = format_launch_hint(specs) - managed = servers.manages(specs) - remote_running, remote_urls = _detect_remote(managed) - # A more specific "part-way set up" label than unavailable/installed: - # cloned but never built, or built but not configured. - partial = "" - if not built: - partial = "downloaded (not built)" - elif not configured: - partial = "built (not configured)" - return BackendStatus("audiocpp", "audio.cpp", installed=built, - configured=configured, - running=managed or remote_running, - details=details, launch_hint=launch, - servers=specs, managed=managed, - remote=remote_running, remote_urls=remote_urls, - models_missing=bool(missing), partial=partial) - - -def _detect_remote(managed: bool = False) -> Tuple[bool, dict]: - """Detect an externally-run audiocpp_server at the remote URL. - - Returns ``(running, {spec_name: url})``. The remote URL is probed only - when configured (non-empty); a server answering there is ignored when it - is this tool's own managed server (remote URL == local URL and our pid is - still alive) — that instance is already reported as "[local]". - """ - url = (config.AUDIOCPP_REMOTE_URL or "").strip() - if not url: - return False, {} - if managed and probe.same_endpoint(url, config.AUDIOCPP_API_URL): - return False, {} - if probe.identify_server(url) == probe.IDENTITY_AUDIOCPP: - return True, {"audiocpp": url} - return False, {} - - -def main() -> int: - parser = build_parser() - 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 _interactive(): - return run_tui(args, parser) - - # Non-interactive (no terminal, or all flags supplied): flag-only path. - settings = _collect_from_flags(args, parser) - if settings is None: - return 1 - return _execute(settings, args) - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/app/backends/audiocpp/__init__.py b/app/backends/audiocpp/__init__.py new file mode 100644 index 0000000..84166af --- /dev/null +++ b/app/backends/audiocpp/__init__.py @@ -0,0 +1,107 @@ +"""audio.cpp backend — setup wizard, server config, model management. + +A package of focused modules behind one import surface: everything public +is re-exported here, so callers (the hub, the CLI, tests) keep using +``backends.audiocpp.<name>`` regardless of which module implements it. + +Modules: + constants shared constants (backends, tasks, checkout location) + catalog the model_specs catalog + server.json building/selections + models install state on disk, missing-model guidance, downloads + voices reference-.wav transcription planning and execution + configsync app/converter/config.py + server.json port/id/backend sync + build checkout lifecycle: ggml patches, binary build, uninstall + remote querying a running server for its models/voices + status detect() for the hub's backend menu + wizard the TUI wizard and the CLI entry points +""" + +from .constants import ( + AUDIOCPP_DIR_NAME, + AUDIOCPP_GIT_URL, + BACKENDS, + DEFAULT_HOST, + FALLBACK_PORT, + PATCH_DIR, + TASK_TTS, + TASK_VDES, +) +from .catalog import ( + _backend_options, + _default_package, + detect_backend, + is_design_package, + load_model_catalog, + package_dir_options, + build_model_entry, + build_server_config, + load_server_config, + server_config_selections, +) +from .models import ( + delete_model_files, + hand_install_guidance, + install_models, + installed_model_entries, + missing_model_entries, + missing_model_install_guidance, + model_install_hints, + unused_installed_entries, +) +from .voices import ( + print_empty_transcript_warning, + transcribe_wav_dir, +) +from .configsync import ( + config_port, + update_config_api_url_port, + update_config_model_ids, + update_server_backend, + update_server_config_port, +) +from .build import ( + apply_ggml_patches, + build_audiocpp, + find_build_script, + find_local_checkout, + find_audiocpp_server_bin, + uninstall, +) +from .remote import fetch_server_models, fetch_server_voices +from .wizard import ( + build_parser, + build_screen, + main, + run_tui, + setup_screen, +) +from .status import detect + +__all__ = [ + # constants + "AUDIOCPP_DIR_NAME", "AUDIOCPP_GIT_URL", "BACKENDS", "DEFAULT_HOST", + "FALLBACK_PORT", "PATCH_DIR", "TASK_TTS", "TASK_VDES", + # catalog + "detect_backend", "load_model_catalog", "is_design_package", + "package_dir_options", "build_model_entry", "build_server_config", + "load_server_config", "server_config_selections", + # models + "missing_model_entries", "installed_model_entries", + "unused_installed_entries", "delete_model_files", "install_models", + "hand_install_guidance", "missing_model_install_guidance", + "model_install_hints", + # voices + "transcribe_wav_dir", "print_empty_transcript_warning", + # configsync + "config_port", "update_config_api_url_port", + "update_config_model_ids", "update_server_config_port", + "update_server_backend", + # build + "find_local_checkout", "find_audiocpp_server_bin", "find_build_script", + "apply_ggml_patches", "build_audiocpp", "uninstall", + # remote + "fetch_server_models", "fetch_server_voices", + # wizard / status + "setup_screen", "build_screen", "run_tui", "build_parser", "detect", + "main", +] diff --git a/app/backends/audiocpp/__main__.py b/app/backends/audiocpp/__main__.py new file mode 100644 index 0000000..457cddc --- /dev/null +++ b/app/backends/audiocpp/__main__.py @@ -0,0 +1,8 @@ +"""Direct CLI execution: ``python -m backends.audiocpp`` (from app/).""" + +import sys + +from backends.audiocpp import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/app/backends/audiocpp/build.py b/app/backends/audiocpp/build.py new file mode 100644 index 0000000..a4b307a --- /dev/null +++ b/app/backends/audiocpp/build.py @@ -0,0 +1,339 @@ +"""Checkout lifecycle: clone location, ggml patches, binary build, uninstall.""" + +import contextlib +import io +import re +import shlex +import shutil +from datetime import datetime +from pathlib import Path +from typing import List, Optional + +from backends import common, servers +from backends.common import APP_DIR +from .catalog import _BACKEND_TOKEN_RE +from .constants import ( + AUDIOCPP_DIR_NAME, + AUDIOCPP_GIT_URL, + PATCH_DIR, +) + +def uninstall(*, emit=None, cancel=None) -> int: + """Remove the audio.cpp backend entirely: stop its server, delete the checkout. + + The checkout (``app/audio.cpp``) holds the built binary, the downloaded + models, and the server.json, so removing the directory uninstalls the + backend. A running server this tool started is stopped first + (best-effort). + + EMIT is accepted for registry symmetry with the other backends but is + unused here — this uninstall has no subprocess phase, and its prints are + captured by the task view when run in the TUI. CANCEL is a + ``threading.Event`` honored between phases only (after the server has + been stopped, before the checkout is deleted), so a started phase always + completes and the uninstall never tears halfway. Returns the exit code + (130 when cancelled before a remaining phase). + """ + # Only stop when a pid file exists: without one this tool never + # started the server, so the "not started by this tool" notice would + # be uninstall-time noise. + if servers.pid_for("audiocpp") is not None: + servers.stop("audiocpp") + if common.cancel_requested(cancel): + return 130 + checkout = find_local_checkout() + if checkout is None: + print("[INFO] No audio.cpp checkout to remove.") + return 0 + print(f"[INFO] Removing audio.cpp checkout {checkout}...") + shutil.rmtree(checkout, ignore_errors=True) + print("[OK] audio.cpp removed.") + return 0 + + +def find_local_checkout() -> Optional[Path]: + """Return the managed audio.cpp checkout at ``app/audio.cpp``. + + Returns the path only when it contains a ``model_specs`` directory; + the checkout is installed there by the setup wizard and nowhere else. + """ + try: + resolved = (APP_DIR / AUDIOCPP_DIR_NAME).resolve() + except OSError: + return None + if (resolved / "model_specs").is_dir(): + return resolved + return None + + +def find_audiocpp_server_bin(audiocpp_dir: Path) -> Optional[Path]: + """Return the built audiocpp_server binary, or None when not built. + + Scans ``audiocpp_dir/build/*`` for a build directory containing + ``bin/audiocpp_server`` (``.exe`` allowed on Windows). When several + builds exist the first (alphabetical) is returned. + """ + build_root = audiocpp_dir / "build" + if not build_root.is_dir(): + return None + 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 + for name in ("audiocpp_server", "audiocpp_server.exe"): + server = build_dir / "bin" / name + if server.exists(): + return server + return None + + +def built_server_binary(audiocpp_dir: Path, backend: str) -> Optional[Path]: + """Return the built audiocpp_server for BACKEND, or None. + + Like ``find_audiocpp_server_bin`` but limited to build directories whose + name carries the BACKEND token (``-cuda-``, ``-vulkan-``, ``-hip-``, + ``-cpu-``; ``-metal-`` counts as ``cpu``). A checkout with builds for + several backends is asked which one to use without re-offering a build + for a backend that is already built. + """ + build_root = audiocpp_dir / "build" + if not build_root.is_dir(): + return None + 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 + match = _BACKEND_TOKEN_RE.search(build_dir.name.lower()) + if not match: + continue + token = "cpu" if match.group(1) == "metal" else match.group(1) + if token != backend: + continue + for name in ("audiocpp_server", "audiocpp_server.exe"): + server = build_dir / "bin" / name + if server.exists(): + return server + return None + + +def find_build_script(audiocpp_dir: Path) -> Optional[Path]: + """Return the audio.cpp build helper script to run, or None. + + Prefers ``scripts/build_linux.sh``; otherwise the first + ``scripts/build_*.sh`` it finds. (Windows ``.bat`` scripts are not run + automatically — build manually there.) + """ + scripts = audiocpp_dir / "scripts" + if not scripts.is_dir(): + return None + preferred = scripts / "build_linux.sh" + if preferred.exists(): + return preferred + try: + candidates = sorted(scripts.glob("build_*.sh"), + key=lambda p: p.name.lower()) + except OSError: + return None + return candidates[0] if candidates else None + + +GGML_PATCHES = [ + { + "file": "ggml-top-k-cuda-iterator.patch", + "target": "external/ggml/src/ggml-cuda/top-k.cu", + "marker": r"#\s*include\s*<cuda/iterator>", + "label": "top-k.cu: add #include <cuda/iterator> (CCCL 3.x build fix)", + }, +] + + +def apply_ggml_patches(audiocpp_dir: Path, *, emit=None, cancel=None) -> int: + """Apply the shipped ggml build patches to an audio.cpp checkout. + + Idempotent: a patch whose marker already matches its target is skipped + (it is either already applied, or the fork re-vendored a fixed ggml). A + patch that no longer applies because the vendored file changed shape is a + loud, non-interactive failure — the build is aborted so the user + re-evaluates the patch instead of hitting a known nvcc break minutes + later. Returns 0 when every patch is applied or already present, 1 on + drift, 130 when cancelled. + """ + for patch in GGML_PATCHES: + if cancel is not None and cancel.is_set(): + return 130 + target = audiocpp_dir / patch["target"] + if not target.is_file(): + print(f"[INFO] {patch['file']}: target {patch['target']} not " + f"present in this checkout; skipping") + continue + try: + text = target.read_text(encoding="utf-8", errors="ignore") + except OSError as exc: + print(f"[WARNING] {patch['file']}: could not read {target}: " + f"{exc}; skipping") + continue + if re.search(patch["marker"], text): + print(f"[OK] {patch['file']}: fix already present, skipping") + continue + patch_path = PATCH_DIR / patch["file"] + if not patch_path.is_file(): + print(f"[ERROR] {patch['file']}: patch file not found at " + f"{patch_path}; cannot apply") + return 1 + check_argv = ["git", "-C", str(audiocpp_dir), "apply", "--check", + "--whitespace=nowarn", str(patch_path)] + check_rc = common.run_console_subprocess( + check_argv, emit=emit, cancel=cancel) + if check_rc == 130 or (cancel is not None and cancel.is_set()): + return 130 + if check_rc != 0: + print(f"[ERROR] {patch['file']}: no longer applies to " + f"{patch['target']} (git apply --check exit {check_rc}). " + f"The audio.cpp fork's vendored ggml changed shape and " + f"still lacks the fix. Re-evaluate {patch_path}: " + f"regenerate the patch, or drop this entry if the fork " + f"now ships the fix.") + return 1 + apply_argv = ["git", "-C", str(audiocpp_dir), "apply", + "--whitespace=nowarn", str(patch_path)] + rc = common.run_console_subprocess( + apply_argv, emit=emit, cancel=cancel) + if rc == 130 or (cancel is not None and cancel.is_set()): + return 130 + if rc != 0: + print(f"[ERROR] {patch['file']}: git apply failed (exit {rc})") + return rc + print(f"[OK] {patch['file']}: applied ({patch['label']})") + return 0 + + +def build_audiocpp(audiocpp_dir: Path, backend: str, *, + emit=None, cancel=None) -> int: + """Build audiocpp_server for BACKEND, streaming output. + + With EMIT None the build script runs on the console (inherits the + terminal); with EMIT given (the in-TUI task view) its output streams line + by line to EMIT so the view can show progress, and CANCEL aborts it. + + On the EMIT (TUI) path the build output is also tee'd to + ``app/logs/audiocpp_build_<timestamp>.log`` so it survives the curses + session; when the build fails (and was not cancelled) a post-TUI notice + with the copy-pastable command and the log path is queued for the console + (see ``backends.common.record_post_tui_notice``). + + Returns the build script's exit code (non-zero when the script is + missing). + """ + script = find_build_script(audiocpp_dir) + if script is None: + message = (f"[ERROR] No build script found in {audiocpp_dir}/scripts; " + "build audiocpp_server manually (see the audio.cpp README)") + print(message) + if emit is not None: + common.record_post_tui_notice(message) + return 1 + argv = ["sh", str(script), "--backend", backend, "--target", + "audiocpp_server", "--deployment-build"] + command = f"cd {audiocpp_dir} && {shlex.join(argv)}" + if emit is None: + print(f"[INFO] Building audiocpp_server for {backend} ({command})...") + patch_rc = apply_ggml_patches(audiocpp_dir, cancel=cancel) + if patch_rc == 130 or (cancel is not None and cancel.is_set()): + return 130 + if patch_rc != 0: + print("[ERROR] ggml build patches could not be applied; " + "aborting audiocpp_server build. See the messages above " + "and re-evaluate app/backends/patches/.") + return patch_rc + return common.run_console_subprocess(argv, cwd=audiocpp_dir) + return _build_audiocpp_tui(emit, cancel, argv, command, audiocpp_dir) + + +def _build_audiocpp_tui(emit, cancel, argv: List[str], command: str, + audiocpp_dir: Path) -> int: + """Run the build on the TUI path: tee output to a log file. + + The ggml patch step runs first, inside the same log: every emitted + line (patch status, build output) is also written (and flushed) to + ``app/logs/audiocpp_build_<timestamp>.log``. On failure a summary (the + copy-pastable COMMAND and the log path) is emitted into the TUI, + written to the log, and queued as a post-TUI console notice. A + cancelled build (CANCEL set) is not reported as a failure, but its + partial output stays in the log file. + """ + log_path = common.LOG_DIR / ( + f"audiocpp_build_{datetime.now():%Y%m%d_%H%M%S}.log") + log_path.parent.mkdir(parents=True, exist_ok=True) + log_handle = log_path.open("w", encoding="utf-8") + + def tee(line: str) -> None: + log_handle.write(line + "\n") + log_handle.flush() + emit(line) + + class _TeeWriter(io.TextIOBase): + """Route print() output from the patch step into the log too.""" + + def write(self, s: str) -> int: + for line in s.splitlines(): + if line: + tee(line) + return len(s) + + try: + with contextlib.redirect_stdout(_TeeWriter()): + patch_rc = apply_ggml_patches(audiocpp_dir, emit=tee, + cancel=cancel) + if patch_rc == 130 or (cancel is not None and cancel.is_set()): + return 130 + if patch_rc != 0: + notice = ("[ERROR] ggml build patches could not be applied; " + "aborting audiocpp_server build. See the messages " + "above and re-evaluate app/backends/patches/.") + tee(notice) + common.record_post_tui_notice(notice) + return patch_rc + tee(f"[INFO] Building audiocpp_server ({command})...") + rc = common.run_console_subprocess( + argv, cwd=audiocpp_dir, emit=tee, cancel=cancel) + if rc != 0 and (cancel is None or not cancel.is_set()): + notice = (f"[ERROR] audio.cpp build failed (exit code {rc}).\n" + f" Build log: {log_path}\n" + f" Troubleshoot by re-running this command:\n" + f" {command}") + for line in notice.splitlines(): + tee(line) + common.record_post_tui_notice(notice) + finally: + log_handle.close() + return rc + + +def _print_launch_hint(audiocpp_dir: Path, output_path: Path) -> None: + """Print remediation when audiocpp_server is missing (troubleshooting). + + The hub starts and stops the server itself, so a working install gets + no manual launch instructions. When no binary was built, though, the + user needs to know how to build and run it by hand. The commands are + prefixed with ``cd <checkout> &&`` because the server discovers + model_specs/<family>.json relative to its working directory. + """ + if find_audiocpp_server_bin(audiocpp_dir) is not None: + return + print("\n[INFO] audiocpp_server binary not found. Build it first, e.g.:") + script = find_build_script(audiocpp_dir) + if script is not None: + print(f" sh {script} --backend <cuda|vulkan|hip|cpu> " + "--target audiocpp_server --deployment-build") + print(f" then run: cd {audiocpp_dir} && ./build/<platform>-<backend>" + f"-release/bin/audiocpp_server --config {output_path}") + + diff --git a/app/backends/audiocpp/catalog.py b/app/backends/audiocpp/catalog.py new file mode 100644 index 0000000..c672232 --- /dev/null +++ b/app/backends/audiocpp/catalog.py @@ -0,0 +1,293 @@ +"""The model_specs catalog, server.json building and selection views.""" + +import json +import re +from pathlib import Path +from typing import Dict, List, Optional, Set, Tuple + +from .. import common +from .constants import ( + BACKENDS, + DEFAULT_HOST, + FALLBACK_PORT, + TASK_TTS, +) + +DESIGN_PACKAGE_RE = re.compile(r"voice[\s_\-]?design", re.IGNORECASE) + + +_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 + + +_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. + + Prefers the package flagged ``default: true``, then the first GGUF + package, then the first package overall. Returns None for an empty list. + """ + if not packages: + return None + for package in packages: + if package.get("default"): + return package + for package in packages: + if package.get("format") == "gguf": + return package + return packages[0] + + +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, packages (the full list from the spec), install_id + (recommended package id), and default_path (``models/<target_directory>``). + 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(): + raise NotADirectoryError( + f"{audiocpp_dir} has no model_specs/ directory; re-run setup " + "to refresh the audio.cpp checkout") + entries: List[dict] = [] + for spec_path in sorted(specs_dir.glob("*.json")): + try: + spec = json.loads(spec_path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + tasks = spec.get("tasks") or [] + if "tts" not in tasks and spec.get("category") != "tts": + continue + family = spec.get("family") or spec_path.stem + packages = spec.get("packages") or [] + package = _default_package(packages) + if package is None: + # No installable package: skip (cannot be hosted from a path). + continue + target_directory = package.get("target_directory") or family + languages = spec.get("languages") or [] + display_name = spec.get("display_name") or family + description = spec.get("description") or "" + entries.append({ + "family": family, + "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}", + }) + + # All families are treated equally: alphabetical by display name. + entries.sort(key=lambda entry: entry["display_name"].lower()) + 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 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": task, + "mode": "offline", + } + + +def build_server_config(host: str, port: int, backend: str, lazy_load: bool, + model_entries: List[dict], + voice_dir: Optional[str] = None) -> dict: + """Assemble the server.json document. + + ``voice_dir`` is a server-level cloning voice library; when set, every + hosted clone-capable family can use its voices with ``--voice``. + """ + config_doc = { + "host": host, + "port": port, + "backend": backend, + "lazy_load": lazy_load, + "models": model_entries, + } + if voice_dir: + config_doc["voice_dir"] = voice_dir + return config_doc + + +def load_server_config(server_json: Path) -> Optional[dict]: + """Read server.json into a dict, or None when it cannot be used. + + Returns None for a missing file, unreadable content, or a non-dict + document. Used by the wizard's modify flow to pre-fill its screens + from an existing config instead of prompting to overwrite it. + """ + if not server_json.exists(): + return None + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + if not isinstance(data, dict): + return None + return data + + +def server_config_selections(server_config: dict, + catalog: List[dict] + ) -> Tuple[Dict[str, List[str]], + Dict[Tuple[str, str], str]]: + """Map an existing server.json's models back to catalog selections. + + Returns ``(selected_dirs, tasks)``: ``selected_dirs`` maps a catalog + family to the target directories it hosts (``models/<target>`` paths + with the ``models/`` prefix stripped, in server.json order), and + ``tasks`` maps ``(family, target_directory)`` to the entry's task + (``"tts"`` or ``"vdes"``) so the wizard can preserve how design + packages were hosted. Entries whose family is not in the CATALOG are + ignored — the wizard cannot offer them again. + """ + families = {entry["family"] for entry in catalog} + selected_dirs: Dict[str, List[str]] = {} + tasks: Dict[Tuple[str, str], str] = {} + for entry in server_config.get("models") or []: + if not isinstance(entry, dict): + continue + family = entry.get("family") + if not isinstance(family, str) or family not in families: + continue + path = entry.get("path") + if not isinstance(path, str): + continue + target = path[len("models/"):] if path.startswith("models/") else path + if family not in selected_dirs: + selected_dirs[family] = [] + if target not in selected_dirs[family]: + selected_dirs[family].append(target) + tasks[(family, target)] = str(entry.get("task") or TASK_TTS) + return selected_dirs, tasks + + diff --git a/app/backends/audiocpp/configsync.py b/app/backends/audiocpp/configsync.py new file mode 100644 index 0000000..0a161c0 --- /dev/null +++ b/app/backends/audiocpp/configsync.py @@ -0,0 +1,163 @@ +"""Keep app/converter/config.py and server.json in sync with setup choices.""" + +import json +import re +import urllib.parse +from pathlib import Path +from typing import Optional + +from backends import common +from backends.common import CONFIG_PATH, url_with_port +from converter import config +from . import build +from .constants import FALLBACK_PORT + +def config_port() -> int: + """Return the port of AUDIOCPP_API_URL in app/converter/config.py.""" + try: + return urllib.parse.urlsplit(config.AUDIOCPP_API_URL).port or FALLBACK_PORT + except ValueError: + return FALLBACK_PORT + + +def update_config_api_url_port(port: int, config_path: Optional[Path] = None) -> bool: + """Rewrite the port inside AUDIOCPP_API_URL in app/converter/config.py. + + Reads the configured URL from the file (not from the imported module, + which a long hub session can leave behind), swaps its port for PORT, + and writes it back through ``common.update_config_value`` so the + imported module mirrors the change immediately. Returns True when the + file now holds the new URL. + """ + path = Path(config_path) if config_path is not None else CONFIG_PATH + try: + text = path.read_text(encoding="utf-8") + except OSError: + return False + match = re.search(r'(?m)^\s*AUDIOCPP_API_URL\s*=\s*"([^"]*)"', text) + if not match: + return False + return common.update_config_value("AUDIOCPP_API_URL", + url_with_port(match.group(1), port), + config_path=path) + + +def update_server_config_port(port: int) -> bool: + """Rewrite the 'port' in the audio.cpp checkout's server.json. + + Loads ``<checkout>/server.json``, sets its ``port`` to PORT, and + rewrites it with the same ``json.dump`` formatting the wizard uses. + Returns True when the file now carries PORT (a no-op when it already + does), and False when there is no checkout/server.json or the file + cannot be read or written. + """ + checkout = build.find_local_checkout() + if checkout is None: + return False + server_json = checkout / "server.json" + if not server_json.exists(): + return False + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + return False + if not isinstance(data, dict): + return False + if data.get("port") == port: + return True + data["port"] = port + try: + with server_json.open("w", encoding="utf-8") as handle: + json.dump(data, handle, indent=2, ensure_ascii=False) + handle.write("\n") + except OSError: + return False + return True + + +def update_config_model_ids(model_id: str, + clone_model_id: Optional[str] = None, + config_path: Optional[Path] = None) -> bool: + """Rewrite AUDIOCPP_MODEL_ID (and AUDIOCPP_CLONE_MODEL_ID when given). + + Goes through ``common.update_config_value`` so the imported config + module mirrors the change immediately. Returns True when every named + key now holds its value in the file. + """ + path = Path(config_path) if config_path is not None else CONFIG_PATH + ok = common.update_config_value("AUDIOCPP_MODEL_ID", model_id, + config_path=path) + if clone_model_id is not None: + ok = common.update_config_value("AUDIOCPP_CLONE_MODEL_ID", + clone_model_id, + config_path=path) and ok + return ok + + +def _apply_port_sync(port: int, accepted: bool) -> None: + """Write the port into app/converter/config.py, or report when declined.""" + if accepted: + 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") + else: + print("[WARNING] Left AUDIOCPP_API_URL unchanged; audiobook.py " + f"will still use port {config_port()}") + + +def _offer_config_model_id_sync(model_id: str, accepted: Optional[bool]) -> None: + """Point app/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. ACCEPTED is True/False (apply/skip the + rewrite) or None when no single-entry sync applies (nothing to do). + """ + if config.AUDIOCPP_MODEL_ID == model_id \ + and config.AUDIOCPP_CLONE_MODEL_ID == model_id: + return + if accepted is None: + return + if accepted: + 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") + else: + print("[WARNING] Left the model ids unchanged; audiobook.py will " + f"still request model '{config.AUDIOCPP_MODEL_ID}'") + + +def update_server_backend(backend: str) -> bool: + """Rewrite the 'backend' in the checkout's server.json, or True when none. + + Sets ``backend`` to BACKEND in ``<checkout>/server.json`` (same + ``json.dump`` formatting as the wizard). Returns True when the file now + carries BACKEND, when there is no server.json (nothing to sync), or when + it already does; False when the file exists but cannot be read/written. + """ + checkout = build.find_local_checkout() + if checkout is None: + return True + server_json = checkout / "server.json" + if not server_json.exists(): + return True + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + return False + if not isinstance(data, dict): + return False + if data.get("backend") == backend: + return True + data["backend"] = backend + try: + with server_json.open("w", encoding="utf-8") as handle: + json.dump(data, handle, indent=2, ensure_ascii=False) + handle.write("\n") + except OSError: + return False + return True + + diff --git a/app/backends/audiocpp/constants.py b/app/backends/audiocpp/constants.py new file mode 100644 index 0000000..aaa1eed --- /dev/null +++ b/app/backends/audiocpp/constants.py @@ -0,0 +1,32 @@ +"""Constants shared across the audio.cpp backend modules.""" + +import re +from pathlib import Path + +DEFAULT_HOST = "127.0.0.1" + + +FALLBACK_PORT = 8080 + + +BACKENDS = ("cuda", "vulkan", "hip", "cpu") + + +TASK_TTS = "tts" + + +TASK_VDES = "vdes" + + +AUDIOCPP_DIR_NAME = "audio.cpp" + + +AUDIOCPP_GIT_URL = "https://github.com/0xShug0/audio.cpp" + + + +# ggml build patches shipped in this repo and applied to the (gitignored) +# audio.cpp checkout before building, so a fresh clone survives known ggml +# build bugs the audio.cpp fork has not re-vendored yet. See +# apply_ggml_patches() in backends.audiocpp.build. +PATCH_DIR = Path(__file__).resolve().parent / "patches" diff --git a/app/backends/audiocpp/models.py b/app/backends/audiocpp/models.py new file mode 100644 index 0000000..4e6b8bb --- /dev/null +++ b/app/backends/audiocpp/models.py @@ -0,0 +1,384 @@ +"""Model install state: what is on disk, what is missing, how to fetch it.""" + +import json +import os +import shutil +import sys +import tempfile +from pathlib import Path +from typing import Callable, Dict, List, Optional, Set, Tuple + +from backends import common +from . import catalog as _catalog + +def _install_models(audiocpp_dir: Path, + install_guidance: List[Tuple[str, str]], + download: bool, emit=None, cancel=None) -> int: + """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`` 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. + + With EMIT given (the in-TUI task view) each download streams its output + to EMIT and — when the checkout's ``model_manager_v2.py`` supports it — + runs with ``--progress --cancel-file`` so the view can show a real byte + progress bar and cancel gracefully. CANCEL aborts a running download. + + Returns 0 when every command succeeded (or nothing needed running), + 130 when cancelled, 1 when any download failed. + """ + 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) + + supports_progress = emit is not None and _manager_supports_progress(manager) + + if download and not manager.is_file(): + print(f"[WARNING] {manager} not found; printing the install commands " + "instead of running them") + download = False + + failed = 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}...") + argv = [sys.executable, str(manager), "install", install_id] + cancel_file: Optional[Path] = None + on_cancel = None + if supports_progress: + fd, cancel_path = tempfile.mkstemp( + prefix="audiocpp_cancel_", suffix=".cancel") + os.close(fd) + cancel_file = Path(cancel_path) + cancel_file.unlink() # absent = not cancelled + argv += ["--progress", "--cancel-file", str(cancel_file)] + on_cancel = cancel_file.touch + try: + rc = common.run_console_subprocess( + argv, cwd=str(audiocpp_dir), emit=emit, cancel=cancel, + on_cancel=on_cancel) + except OSError as exc: + print(f"[WARNING] Could not run {command}: {exc}") + rc = 1 + finally: + if cancel_file is not None: + try: + cancel_file.unlink() + except OSError: + pass + if rc == 130 or (cancel is not None and cancel.is_set()): + return 130 + if rc != 0: + failed = True + print(f"[WARNING] install {install_id} exited with code " + f"{rc}; the model may need to be downloaded " + "by hand") + return 1 if failed else 0 + + +def _manager_supports_progress(manager: Path) -> bool: + """True when MANAGER (model_manager_v2.py) supports --progress output. + + The ``--progress``/``--cancel-file`` flags are relatively recent; an + older audio.cpp checkout may not have them, so probe the script source + once instead of failing the download with an unknown flag. + """ + try: + text = manager.read_text(encoding="utf-8", errors="ignore") + except OSError: + return False + return "AUDIOCPP_PROGRESS" in text and "--cancel-file" in text + + +def _decide_download(audiocpp_dir: Path, + model_entries: List[dict], + 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. The prompt is also skipped (False) + when every selected model is already on disk (see ``_all_models_present``), + so an already-configured checkout is not asked to re-download models it + already has. + """ + manager = audiocpp_dir / "tools" / "model_manager_v2.py" + if not manager.is_file(): + return False + if _all_models_present(audiocpp_dir, model_entries): + return False + return confirm( + "Automatically download the selected models with model_manager_v2.py " + "now?", True) + + +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"] + options = [] + for opt in _catalog.package_dir_options(entry): + options.append({ + "key": opt["target_directory"], + "label": opt["install_id"], + "recommended": opt["recommended"], + }) + families.append({ + "label": name, + "detail": ", ".join(capabilities), + "options": options, + }) + return families + + +def _model_path_present(path: Path) -> bool: + """True when a server.json model path holds actual model files. + + A present path is either a file (a single-model package) or a non-empty + directory (the usual GGUF package target directory; an empty one means a + download that never ran or was cleaned up halfway). + """ + try: + if path.is_file(): + return True + if path.is_dir(): + return any(path.iterdir()) + except OSError: + return False + return False + + +def _all_models_present(audiocpp_dir: Path, model_entries: List[dict]) -> bool: + """True when every selected model entry's path already holds files on disk. + + Paths resolve against the checkout (where model_manager_v2.py installs + them), honoring absolute paths. Used by the wizard to skip the + "Automatically download the selected models" prompt when nothing is + actually missing. An empty selection is treated as not-present. + """ + if not model_entries: + return False + for entry in model_entries: + rel = entry.get("path") + if not isinstance(rel, str) or not rel: + return False + path = Path(rel) if Path(rel).is_absolute() else audiocpp_dir / rel + if not _model_path_present(path): + return False + return True + + +def missing_model_entries(server_json: Path) -> List[dict]: + """Return the server.json model entries whose files are not on disk. + + Paths resolve exactly like audiocpp_server resolves them (relative paths + against the server.json's directory). Each returned entry carries the + entry ``id`` and ``rel`` (the configured path string); used by ``detect`` + to warn that a conversion would fail until the models are installed. + """ + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + return [] + if not isinstance(data, dict): + return [] + base = server_json.parent + missing: List[dict] = [] + for entry in data.get("models") or []: + if not isinstance(entry, dict): + continue + rel = entry.get("path") + if not isinstance(rel, str) or not rel: + continue + path = Path(rel) if Path(rel).is_absolute() else base / rel + if _model_path_present(path): + continue + missing.append({"id": str(entry.get("id") or rel), "rel": rel}) + return missing + + +def _install_id_by_path(audiocpp_dir: Path) -> Dict[str, str]: + """Map ``models/<target_directory>`` -> catalog install id. + + The catalog package that installs a model is derived from the + ``default_path`` of each TTS family; an entry whose path matches no + catalog package has no install id. + """ + by_path: Dict[str, str] = {} + try: + for entry in _catalog.load_model_catalog(audiocpp_dir): + by_path[entry["default_path"]] = entry["install_id"] + except (NotADirectoryError, OSError): + pass + return by_path + + +def installed_model_entries(server_json: Path) -> List[dict]: + """Return the server.json model entries whose files ARE on disk. + + The complement of ``missing_model_entries``: each returned entry carries + the entry ``id`` and ``rel`` (the configured path string), resolved + exactly like ``missing_model_entries`` (relative against the server.json's + directory). Used by the wizard's "Delete unused models?" step to find + already-downloaded models that were unselected. + """ + try: + data = json.loads(server_json.read_text(encoding="utf-8")) + except (OSError, ValueError): + return [] + if not isinstance(data, dict): + return [] + base = server_json.parent + installed: List[dict] = [] + for entry in data.get("models") or []: + if not isinstance(entry, dict): + continue + rel = entry.get("path") + if not isinstance(rel, str) or not rel: + continue + path = Path(rel) if Path(rel).is_absolute() else base / rel + if _model_path_present(path): + installed.append({"id": str(entry.get("id") or rel), "rel": rel}) + return installed + + +def missing_model_install_guidance(audiocpp_dir: Path, + missing: List[dict]) -> List[Tuple[str, str]]: + """Map MISSING model entries to (display name, install id) pairs. + + The install id is derived from each entry's configured path via the + catalog (see ``_install_id_by_path``); entries whose path matches no + catalog package are skipped (there is no ``model_manager_v2.py install`` + command for them). Feeds ``_install_models`` for the "Download Missing + Models" action. + """ + by_path = _install_id_by_path(audiocpp_dir) + guidance: List[Tuple[str, str]] = [] + for item in missing: + install_id = by_path.get(item["rel"]) + if install_id: + guidance.append((item["id"], install_id)) + return guidance + + +def model_install_hints(audiocpp_dir: Path, + missing: List[dict]) -> List[str]: + """Remediation lines for MISSING model entries (see missing_model_entries). + + Maps each entry's configured path back to the catalog package that + installs it (``models/<target_directory>`` -> install id) so the line + carries the exact ``model_manager_v2.py install`` command; entries whose + directory matches no catalog package just name the path. + """ + by_path = _install_id_by_path(audiocpp_dir) + hints: List[str] = [] + for item in missing: + install_id = by_path.get(item["rel"]) + hint = f"model not downloaded: {item['id']} ({item['rel']})" + if install_id: + hint += (f" — install with: python tools/model_manager_v2.py " + f"install {install_id}") + hints.append(hint) + return hints + + +def install_models(audiocpp_dir: Path, + guidance: List[Tuple[str, str]], + emit=None, cancel=None) -> int: + """Download the (display name, install id) models via the helper script. + + Runs ``model_manager_v2.py install`` for each de-duped install id in the + checkout, streaming to the console (or to EMIT, the in-TUI task view); a + failing install is reported as a warning and does not abort the rest. + Returns 0 when every download succeeded, 130 when cancelled, 1 when any + failed. Used by the hub's "Download Missing Models" action (see + ``missing_model_install_guidance`` for the mapping). + """ + return _install_models(audiocpp_dir, guidance, download=True, + emit=emit, cancel=cancel) + + +def hand_install_guidance(audiocpp_dir: Path, + missing: List[dict]) -> str: + """Explain how to install MISSING model entries by hand. + + Returns a multi-line message listing each missing model's id and the + path its files must be placed in (``rel``, resolved against the + checkout). Used when the missing models cannot be mapped to + a ``model_manager_v2.py install`` command, so the user still knows what + to download and where to put it. + """ + lines = [ + "None of the missing models map to a model_manager_v2.py install " + "command.", + "Download them by hand and place the files at these paths:", + ] + for item in missing: + lines.append(f" {item['id']} -> {item['rel']}") + lines.append(f"(paths are relative to {audiocpp_dir})") + return "\n".join(lines) + + +def unused_installed_entries(server_json: Path, + new_paths: Set[str]) -> List[dict]: + """Return installed server.json entries whose path is not in NEW_PATHS. + + The already-downloaded models (see ``installed_model_entries``) that the + new selection does not host any more — the candidates for the wizard's + "Delete unused models?" prompt. Entries whose files are not on disk are + never listed (there is nothing to delete). + """ + return [entry for entry in installed_model_entries(server_json) + if entry["rel"] not in new_paths] + + +def delete_model_files(server_json: Path, entries: List[dict]) -> int: + """Remove the on-disk model files for ENTRIES ({id, rel}) from disk. + + Each entry's ``rel`` is resolved exactly like the server resolves it + (relative against ``server_json``'s directory; absolute paths honored), + then removed as a directory tree or a single file. Missing entries are + ignored. Returns the number of paths removed. Used by the wizard's + "Delete unused models?" step — the regenerated server.json already only + lists the kept models, so no entry cleanup is needed here. + """ + base = server_json.parent + removed = 0 + for item in entries: + rel = item.get("rel") + if not isinstance(rel, str) or not rel: + continue + path = Path(rel) if Path(rel).is_absolute() else base / rel + try: + if not path.exists(): + continue + if path.is_dir(): + shutil.rmtree(path, ignore_errors=True) + else: + path.unlink() + except OSError as exc: + print(f"[WARNING] Could not remove {path}: {exc}") + continue + print(f"[OK] Removed unused model {path}") + removed += 1 + return removed + + diff --git a/app/backends/patches/ggml-top-k-cuda-iterator.patch b/app/backends/audiocpp/patches/ggml-top-k-cuda-iterator.patch index 0eb89a5..0eb89a5 100644 --- a/app/backends/patches/ggml-top-k-cuda-iterator.patch +++ b/app/backends/audiocpp/patches/ggml-top-k-cuda-iterator.patch diff --git a/app/backends/audiocpp/remote.py b/app/backends/audiocpp/remote.py new file mode 100644 index 0000000..42b3872 --- /dev/null +++ b/app/backends/audiocpp/remote.py @@ -0,0 +1,59 @@ +"""Query a running audiocpp_server for its models and voices.""" + +import json +import urllib.request +from typing import Dict, List, Optional + +from .constants import FALLBACK_PORT + +def fetch_server_models(api_url: str) -> Optional[List[Dict[str, str]]]: + """List a running audiocpp_server's model entries via GET /v1/models. + + Returns ``[{id, family, task}, ...]`` — the same shape the converter's + client resolves at startup — or None when URL does not answer with a + valid document (wrong server, still starting, older audio.cpp). Used by + the hub to drive the convert menus against a remote server that has no + local server.json describing it. + """ + try: + with urllib.request.urlopen( + f"{api_url.rstrip('/')}/v1/models", timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + except (OSError, ValueError): + # URLError/HTTPError/socket errors are OSErrors; a non-JSON body is + # a ValueError. Anything else means "not an audiocpp_server". + return None + entries = payload.get("data") if isinstance(payload, dict) else None + models: List[Dict[str, str]] = [] + for entry in entries or []: + if isinstance(entry, dict) and entry.get("id"): + models.append({ + "id": str(entry["id"]), + "family": str(entry.get("family") or ""), + "task": str(entry.get("task") or ""), + }) + return models + + +def fetch_server_voices(api_url: str, model_id: str) -> Optional[List[str]]: + """List a running audiocpp_server's voices for MODEL_ID. + + Queries ``GET /v1/audio/voices?model=<id>`` — the endpoint the converter + validates ``--voice`` against — and returns its voice-name list, or None + when the server cannot be queried. Lets the hub offer a remote server's + voices without reading its configuration locally. + """ + query = urllib.parse.urlencode({"model": model_id}) + try: + with urllib.request.urlopen( + f"{api_url.rstrip('/')}/v1/audio/voices?{query}", + timeout=10) as response: + payload = json.loads(response.read().decode("utf-8")) + except (OSError, ValueError): + return None + voices = payload.get("voices") if isinstance(payload, dict) else None + if not isinstance(voices, list): + return None + return [str(voice) for voice in voices] + + diff --git a/app/backends/audiocpp/status.py b/app/backends/audiocpp/status.py new file mode 100644 index 0000000..9ccf8cc --- /dev/null +++ b/app/backends/audiocpp/status.py @@ -0,0 +1,90 @@ +"""detect() — the BackendStatus report for the hub's backend menu.""" + +from typing import List, Tuple + +from backends import (BackendStatus, ServerSpec, format_launch_hint, + probe, servers) +from converter import config +from . import build as _build +from . import models as _models + +def detect() -> BackendStatus: + """Detect how far audio.cpp is set up, plus the command to start it.""" + checkout = _build.find_local_checkout() + details: List[str] = [] + launch = "" + if checkout is None: + # No local checkout: only a remote server can make this usable. + remote = _detect_remote() + return BackendStatus("audiocpp", "audio.cpp", installed=False, + configured=False, running=remote[0], + remote=remote[0], remote_urls=remote[1], + details=["not cloned — run setup to clone " + "./app/audio.cpp"]) + details.append(f"checkout: {checkout}") + binary = _build.find_audiocpp_server_bin(checkout) + built = binary is not None + if built: + details.append(f"built: {binary}") + else: + details.append("not built — run setup to build audiocpp_server") + server_json = checkout / "server.json" + configured = server_json.exists() + specs: List[ServerSpec] = [] + missing = _models.missing_model_entries(server_json) if configured else [] + if configured: + details.append(f"config: {server_json}") + if missing: + # The config references model files that are not on disk; a + # conversion would fail at model-load time, so say so now. + details.extend(_models.model_install_hints(checkout, missing)) + if built: + # Spawned from the checkout: audiocpp_server discovers + # model_specs/<family>.json relative to its working directory. + specs = [ServerSpec( + "audiocpp", config.AUDIOCPP_API_URL, + [str(binary), "--config", str(server_json)], + cwd=checkout, identity=probe.IDENTITY_AUDIOCPP)] + else: + launch = (f"cd {checkout} && ./build/<platform>-<backend>-release" + f"/bin/audiocpp_server --config {server_json}") + else: + details.append("no server.json — run setup to configure models") + if specs: + launch = format_launch_hint(specs) + managed = servers.manages(specs) + remote_running, remote_urls = _detect_remote(managed) + # A more specific "part-way set up" label than unavailable/installed: + # cloned but never built, or built but not configured. + partial = "" + if not built: + partial = "downloaded (not built)" + elif not configured: + partial = "built (not configured)" + return BackendStatus("audiocpp", "audio.cpp", installed=built, + configured=configured, + running=managed or remote_running, + details=details, launch_hint=launch, + servers=specs, managed=managed, + remote=remote_running, remote_urls=remote_urls, + models_missing=bool(missing), partial=partial) + + +def _detect_remote(managed: bool = False) -> Tuple[bool, dict]: + """Detect an externally-run audiocpp_server at the remote URL. + + Returns ``(running, {spec_name: url})``. The remote URL is probed only + when configured (non-empty); a server answering there is ignored when it + is this tool's own managed server (remote URL == local URL and our pid is + still alive) — that instance is already reported as "[local]". + """ + url = (config.AUDIOCPP_REMOTE_URL or "").strip() + if not url: + return False, {} + if managed and probe.same_endpoint(url, config.AUDIOCPP_API_URL): + return False, {} + if probe.identify_server(url) == probe.IDENTITY_AUDIOCPP: + return True, {"audiocpp": url} + return False, {} + + diff --git a/app/backends/audiocpp/voices.py b/app/backends/audiocpp/voices.py new file mode 100644 index 0000000..2f0fdd7 --- /dev/null +++ b/app/backends/audiocpp/voices.py @@ -0,0 +1,146 @@ +"""Reference-.wav transcription planning and execution.""" + +import argparse +from pathlib import Path +from typing import Callable, Dict, List, Optional, Tuple + +from backends.common import (PROMPT_TEXT_FILENAME, find_wav_files, + read_prompt_text) +from converter.clients import (transcribe_reference_audio, + whisper_backend_available) + +def transcribe_wav_dir(wav_files: list, whisper_model: str, + cancel=None) -> Dict[str, str]: + """Transcribe each wav file and return a mapping of stem -> transcript. + + CANCEL (a ``threading.Event``) is checked between files so the in-TUI + task view can stop a long transcription early. + """ + transcripts: Dict[str, str] = {} + for wav_file in wav_files: + if cancel is not None and cancel.is_set(): + print("[INFO] Transcription cancelled") + break + name = wav_file.stem + print(f"[INFO] Transcribing {wav_file.name} (voice '{name}')...") + text = transcribe_reference_audio(str(wav_file), model_name=whisper_model) + if text: + print(f"[OK] {name}: {text}") + else: + print(f"[WARNING] No transcript for '{name}'; cloning works best " + "with an accurate transcript — consider editing prompt_text " + "by hand before starting the server") + transcripts[name] = text or "" + return transcripts + + +def print_empty_transcript_warning(transcripts: Dict[str, str]) -> None: + """Print a loud, final warning for voices whose transcript is empty.""" + empty = sorted(name for name, text in transcripts.items() if not text) + if not empty: + return + bar = "=" * 70 + print() + print(bar) + print("[WARNING] MANUAL TRANSCRIPTION REQUIRED") + print(bar) + listing = " - " + "\n - ".join(empty) if len(empty) > 1 else f" - {empty[0]}" + print(f"The following voice(s) have an EMPTY transcript in prompt_text:\n" + f"{listing}") + print("Those voices will NOT work until you add an accurate transcript.") + print(f"Edit {PROMPT_TEXT_FILENAME} in your voice directory and fill in the " + "text after '|' for each voice above.") + print(bar) + + +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": + [...], "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] = [] + 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, "existing": existing} + + +def _transcribe(args: argparse.Namespace, plan: Optional[dict], + cancel=None) -> Tuple[Dict[str, str], bool]: + """Transcribe the wav directory into a stem -> transcript mapping. + + 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). + PLAN is always pre-collected — by the TUI (via _decide_transcription and + its confirm callbacks) or by _flag_plan for a non-interactive run — so no + questions are asked here; a None PLAN defaults to "transcribe everything". + CANCEL is checked between files. + """ + 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 {}, False + + prompt_path = args.input_dir / PROMPT_TEXT_FILENAME + existing = dict((plan or {}).get("existing") or {}) + mode = plan["mode"] if plan else "all" + + if 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(" 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, + cancel=cancel) + transcripts = dict(existing) + transcripts.update(new_transcripts) + else: + transcripts = transcribe_wav_dir(wav_files, args.whisper_model, + cancel=cancel) + return transcripts, True + + +def _flag_plan(wav_files: list, prompt_path: Path, force: bool) -> dict: + """Build a transcription plan for a non-interactive (flag-only) run. + + With --force everything is re-transcribed; otherwise an existing + prompt_text is reused and only voices with an empty transcript are + re-transcribed, mirroring what the TUI confirms interactively. + """ + if prompt_path.exists() and not force: + existing = read_prompt_text(prompt_path) + missing = [wav for wav in wav_files + if not existing.get(wav.stem, "").strip()] + if not missing: + return {"mode": "keep", "missing": [], "existing": existing} + return {"mode": "missing", "missing": missing, "existing": existing} + return {"mode": "all", "missing": [], "existing": {}} + + diff --git a/app/backends/audiocpp/wizard.py b/app/backends/audiocpp/wizard.py new file mode 100644 index 0000000..dcab273 --- /dev/null +++ b/app/backends/audiocpp/wizard.py @@ -0,0 +1,1081 @@ +"""The audio.cpp setup wizard: TUI screens, task lanes, CLI entry points.""" + +import argparse +import json +import sys +from pathlib import Path +from typing import Callable, Dict, List, Optional, Tuple + +from backends import common +from backends.common import ( + APP_DIR, + PROMPT_TEXT_FILENAME, + TTS_ROOT, + VOICES_DIR, + detect_wav_dir, + find_wav_files, + read_prompt_text, + resolve_wav_dir_arg, + wav_dir_info as _wav_dir_info, + wav_dir_preview as _wav_dir_preview, + write_prompt_text, +) +from converter import config +from ui import taskview, tui +from . import build as _build +from . import configsync as _configsync +from . import models as _models +from . import voices as _voices +from .catalog import (BACKENDS, DEFAULT_HOST, _backend_options, + build_model_entry, build_server_config, detect_backend, + load_model_catalog, load_server_config, + package_dir_options, server_config_selections) +from .constants import (AUDIOCPP_DIR_NAME, AUDIOCPP_GIT_URL, + TASK_TTS, TASK_VDES) + +_GO_BACK = object() + + +class _GoBack(Exception): + """Internal signal: Esc was pressed inside one of a screen's sub-prompts. + + The wizard drives a stack of screens via ``tui.Wizard``. Helpers that ask + several questions through callbacks (the task/id pickers inside + ``_build_entries``, the transcription plan, the download prompt) cannot + themselves return the wizard's ``BACK`` sentinel, so they convert the + ``_GO_BACK`` value passed to each widget into this exception. The screen + that invoked the helper catches it and returns ``tui.Wizard.BACK``, which + pops back to the previous screen. Esc on the first screen aborts the + whole wizard. + """ + + +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). + """ + + +# Alias kept on this module: main()'s tty check and its tests patch it +# here. +from backends.setup import interactive as _interactive + + +def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], + catalog_by_family: Dict[str, dict], + task_picker: Callable[[str], str], + known_tasks: Optional[Dict[Tuple[str, str], str]] = None + ) -> 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. + KNOWN_TASKS maps ``(family, target_directory)`` to a previously-stored + task ("tts" or "vdes") so a modify run preserves how a design package + was hosted instead of re-asking. Each entry's server id is its package + ``target_directory`` (flattened to a token), so packages from the same + family never collide; an id that does collide (across families) is + auto-suffixed without prompting. 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]: + if opt["design"]: + task = known_tasks.get((family, opt["target_directory"])) \ + if known_tasks else None + if task is None: + task = task_picker(opt["install_id"]) + else: + task = TASK_TTS + base_id = opt["target_directory"].replace("/", "-") + model_id = base_id + if model_id in entry_ids: + n = 2 + while f"{base_id}-{n}" in entry_ids: + n += 1 + model_id = f"{base_id}-{n}" + 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(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, 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: + 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) + + with output_path.open("w", encoding="utf-8") as handle: + json.dump(server_config, handle, indent=2, ensure_ascii=False) + handle.write("\n") + + count = len(model_entries) + print(f"Wrote {output_path.resolve()} with {count} " + f"{'entry' if count == 1 else 'entries'}.") + + +def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser + ) -> Optional[dict]: + """Run every TUI screen; return the collected settings, or None to abort. + + The wizard is driven by ``tui.Wizard`` as a stack of screen closures: + each screen shows one interactive widget and returns the next screen + (a closure), ``Wizard.BACK`` (Esc/q pressed — pop to the previous + screen), or the final settings dict. Only screens that actually render + are pushed, so Esc always lands on the previous real screen. A step + whose value is already provided by a flag (``--host``, ``--port``, + ``--families``, ...) or does not apply (e.g. the port-sync prompt when + the port did not change) is folded into the ``_after_*`` guards and + never becomes a screen. Esc on the first screen aborts the whole + wizard. + """ + + s: dict = {} + + 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 + + def resolve_checkout(audiocpp_dir: Path) -> None: + """Validate the audio.cpp checkout and populate the wizard state ``s``.""" + audiocpp_dir = Path(audiocpp_dir).resolve() + 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" + # Modify flow: an existing server.json seeds the wizard's screens + # instead of being overwritten from scratch (an explicit --force + # still starts fresh). + existing_config = load_server_config(output_path) \ + if not args.force else None + if existing_config is not None: + existing_selected, existing_tasks = \ + server_config_selections(existing_config, catalog) + else: + existing_selected, existing_tasks = {}, {} + s.update({ + "audiocpp_dir": audiocpp_dir, + "catalog": catalog, + "catalog_by_family": catalog_by_family, + "output_path": output_path, + "existing_config": existing_config, + "existing_selected": existing_selected, + "existing_tasks": existing_tasks, + "existing_host": existing_config.get("host") + if existing_config else None, + "existing_port": existing_config.get("port") + if existing_config else None, + "existing_backend": existing_config.get("backend") + if existing_config else None, + "existing_voice_dir": existing_config.get("voice_dir") + if existing_config else None, + "detected_backend": detect_backend(audiocpp_dir), + }) + + def _families_from_flag() -> None: + requested = [f.strip() for f in args.families.split(",") if f.strip()] + unknown = [f for f in requested if f not in s["catalog_by_family"]] + if unknown: + raise _TuiError( + f"Unknown family in --families: {', '.join(unknown)}. " + f"Available: {', '.join(s['catalog_by_family'])}") + chosen: Dict[str, List[dict]] = {} + 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( + s["catalog_by_family"][family]) if opt["recommended"]] + s["chosen"] = chosen + s["family_keys"] = family_keys + + def _compute_entries() -> None: + # Design task menu. Esc raises _GoBack, which the caller turns into + # Wizard.BACK (the design prompts are grouped: Esc returns to the + # families tree). + 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 + + model_entries, entry_ids, install_guidance, \ + design_entry_ids, include_clone = _build_entries( + s["family_keys"], s["chosen"], s["catalog_by_family"], + task_picker, known_tasks=s["existing_tasks"]) + s.update({ + "model_entries": model_entries, + "entry_ids": entry_ids, + "install_guidance": install_guidance, + "design_entry_ids": design_entry_ids, + "include_clone": include_clone, + }) + + def _finalize() -> dict: + return { + "audiocpp_dir": s["audiocpp_dir"], + "catalog": s["catalog"], + "catalog_by_family": s["catalog_by_family"], + "output_path": s["output_path"], + "family_keys": s["family_keys"], + "chosen": s["chosen"], + "model_entries": s["model_entries"], + "entry_ids": s["entry_ids"], + "install_guidance": s["install_guidance"], + "design_entry_ids": s["design_entry_ids"], + "include_clone": s["include_clone"], + "host": s["host"], + "port": s["port"], + "backend": s["backend"], + "build": s["build"], + "lazy_load": s["lazy_load"], + "sync_port": s["sync_port"], + "sync_model_ids": s["sync_model_ids"], + "wav_dir": s["wav_dir"], + "plan": s["plan"], + "download": s["download"], + "delete_unused": s["delete_unused"], + "unused_entries": s["unused_entries"], + } + + def screen_families(): + """Pick TTS model families and packages (the modify tree).""" + tree_families = _models._build_tree_families(s["catalog"]) + # Modify flow: pre-check the models an existing server.json hosts, + # so the tree opens as a "modify" list rather than a fresh one. + checked_set = set() + for family, dirs in s["existing_selected"].items(): + if family not in s["catalog_by_family"]: + continue + family_index = s["catalog"].index(s["catalog_by_family"][family]) + valid_dirs = {opt["target_directory"] + for opt in package_dir_options( + s["catalog_by_family"][family])} + for target in dirs: + if target in valid_dirs: + checked_set.add((family_index, target)) + picked = tui.checkbox_tree( + stdscr, "Select TTS model families to host", + tree_families, expand_all=args.all_packages, + back_value=_GO_BACK, checked=checked_set) + if picked is _GO_BACK: + return tui.Wizard.BACK + chosen: Dict[str, List[dict]] = {} + family_keys: List[str] = [] + for family_index, option_key in picked: + family = s["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( + s["catalog_by_family"][family])} + chosen[family] = [keyed[key] for key in chosen[family]] + s["chosen"] = chosen + s["family_keys"] = family_keys + return screen_host + + def _after_families(): + if args.families is not None: + _families_from_flag() + return screen_host + return screen_families + + def screen_host(): + """Build the model entries, then ask the bind host. + + The task/id pickers (when any) run here too and are grouped with + this screen: Esc on one of them (or on the host field) returns to + the families tree. + """ + try: + _compute_entries() + except _GoBack: + return tui.Wizard.BACK + if args.host is not None: + s["host"] = args.host + return _after_host() + host = tui.line_edit( + stdscr, "Bind host", + s["existing_host"] if isinstance(s["existing_host"], str) + else 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: + return tui.Wizard.BACK + s["host"] = host + return _after_host() + + def _after_host(): + if args.port is None: + return screen_port + s["port"] = args.port + return _after_port() + + def screen_port(): + port_text = tui.line_edit( + stdscr, "Port", + str(s["existing_port"]) if isinstance(s["existing_port"], int) + else str(_configsync.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: + return tui.Wizard.BACK + s["port"] = int(port_text) + return _after_port() + + def _after_port(): + s["sync_port"] = None + if s["port"] != _configsync.config_port(): + return screen_sync_port + return _after_sync() + + def screen_sync_port(): + sync_port = tui.confirm( + stdscr, "Update AUDIOCPP_API_URL in app/converter/config.py " + f"to port {s['port']} so audiobook.py talks to this server", + default=True, cancel_value=_GO_BACK) + if sync_port is _GO_BACK: + return tui.Wizard.BACK + s["sync_port"] = sync_port + return _after_sync() + + def _after_sync(): + if args.build_backend: + s["backend"] = args.build_backend + s["build"] = s["detected_backend"] is None + return _after_backend() + if args.backend: + s["backend"] = args.backend + s["build"] = False + return _after_backend() + if s["detected_backend"] is not None: + # Already built: use the detected backend, no menu, no build. + s["backend"] = s["detected_backend"] + s["build"] = False + return _after_backend() + # Not built for any backend yet: always ask which backend the server + # should use and offer to build it — even on a modify run, so a user + # who declined the build the first time is never stranded without a + # way to build from the TUI. + return screen_backend + + def screen_backend(): + # Pre-select the backend an existing server.json records (modify + # flow), so re-running setup lands on the previous choice. + backend_options, backend_default = _backend_options(None) + if s["existing_backend"] in BACKENDS: + backend_default = next( + (index for index, (_label, value) in enumerate(backend_options) + if value == s["existing_backend"]), backend_default) + backend = tui.menu( + stdscr, "Which inference backend should audiocpp_server " + "use?", backend_options, + default_index=backend_default, back_value=_GO_BACK) + if backend is _GO_BACK: + return tui.Wizard.BACK + s["backend"] = backend + if _build.built_server_binary(s["audiocpp_dir"], backend) is not None: + # A checkout with builds for several backends: this one is + # already built, so there is nothing to build. + s["build"] = False + return _after_backend() + return screen_build + + def screen_build(): + # Not built for the chosen backend yet: offer to build it now. The + # build itself runs in the TUI task view (or the console tail for + # CLI runs) after the wizard. + build = tui.confirm( + stdscr, f"audiocpp_server is not built for {s['backend']}. " + f"Build it now (runs scripts/build_*)?", + default=True, cancel_value=_GO_BACK) + if build is _GO_BACK: + return tui.Wizard.BACK + s["build"] = build + return _after_backend() + + def _after_backend(): + s["lazy_load"] = True + return _after_lazy() + + def _after_lazy(): + if args.input_dir is not None: + s["wav_dir"] = args.input_dir + return _after_wav() + if s["include_clone"]: + return screen_wav + s["wav_dir"] = None + return _after_wav() + + def screen_wav(): + wav_start = detect_wav_dir(s["audiocpp_dir"], TTS_ROOT) + # Modify flow: an existing voice_dir seeds the browser so the user + # can accept it on Enter instead of re-navigating. + if isinstance(s["existing_voice_dir"], str) and s["existing_voice_dir"]: + wav_start = Path(s["existing_voice_dir"]) + 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 VOICES_DIR, + back_value=_GO_BACK) + if wav_dir is _GO_BACK: + return tui.Wizard.BACK + s["wav_dir"] = wav_dir + return _after_wav() + + def _after_wav(): + s["plan"] = None + if s["include_clone"] and s["wav_dir"] is not None: + wav_files = find_wav_files(s["wav_dir"]) + if wav_files: + prompt_path = s["wav_dir"] / PROMPT_TEXT_FILENAME + if prompt_path.exists() and not args.force: + return screen_transcription + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + s["plan"] = _voices._decide_transcription( + wav_files, existing, prompt_path.exists(), + args.force, ask_confirm) + return _after_transcription() + + def screen_transcription(): + # Transcription plan (questions only; transcription runs after). + wav_files = find_wav_files(s["wav_dir"]) + prompt_path = s["wav_dir"] / PROMPT_TEXT_FILENAME + existing = read_prompt_text(prompt_path) if ( + prompt_path.exists() and not args.force) else {} + try: + s["plan"] = _voices._decide_transcription( + wav_files, existing, prompt_path.exists(), + args.force, ask_confirm) + except _GoBack: + return tui.Wizard.BACK + return _after_transcription() + + def _after_transcription(): + s["sync_model_ids"] = None + if len(s["entry_ids"]) == 1 and not ( + config.AUDIOCPP_MODEL_ID == s["entry_ids"][0] + and config.AUDIOCPP_CLONE_MODEL_ID == s["entry_ids"][0]): + return screen_model_sync + return _after_model_sync() + + def screen_model_sync(): + sync_model_ids = tui.confirm( + stdscr, "Update AUDIOCPP_MODEL_ID and " + "AUDIOCPP_CLONE_MODEL_ID in app/converter/config.py to " + f"'{s['entry_ids'][0]}' so audiobook.py uses this model", + default=True, cancel_value=_GO_BACK) + if sync_model_ids is _GO_BACK: + return tui.Wizard.BACK + s["sync_model_ids"] = sync_model_ids + return _after_model_sync() + + def _after_model_sync(): + new_paths = {entry["path"] for entry in s["model_entries"]} + s["unused_entries"] = _models.unused_installed_entries( + s["output_path"], new_paths) \ + if s["existing_config"] is not None else [] + s["delete_unused"] = False + if s["unused_entries"]: + return screen_delete_unused + return _after_delete() + + def screen_delete_unused(): + delete_unused = tui.confirm( + stdscr, "Delete unused models?", default=False, + cancel_value=_GO_BACK) + if delete_unused is _GO_BACK: + return tui.Wizard.BACK + s["delete_unused"] = delete_unused + return _after_delete() + + def _after_delete(): + manager = s["audiocpp_dir"] / "tools" / "model_manager_v2.py" + if manager.is_file(): + return screen_download + s["download"] = False + return _finalize() + + def screen_download(): + # Automatic model download (or print the install commands). + try: + s["download"] = _models._decide_download( + s["audiocpp_dir"], s["model_entries"], ask_confirm) + except _GoBack: + return tui.Wizard.BACK + return _finalize() + + # First screen: resolve the checkout directly when it already exists + # (the modify flow), so the wizard starts on a real screen. When no + # checkout exists, clone it into ./app/audio.cpp (streaming inside the + # TUI task view, not by dropping to the console) without asking, then + # continue the same way. + audiocpp_dir = _build.find_local_checkout() + if audiocpp_dir is None: + target = APP_DIR / AUDIOCPP_DIR_NAME + rc = taskview.run_steps(stdscr, "Clone audio.cpp", [ + taskview.TaskStep( + f"Cloning audio.cpp into {target}", + lambda emit, cancel: common.git_clone( + AUDIOCPP_GIT_URL, target, emit=emit, cancel=cancel)), + taskview.TaskStep( + "Apply ggml build patches", + lambda emit, cancel: _build.apply_ggml_patches( + target, emit=emit, cancel=cancel)), + ]) + if rc == 130: + # Cancelled from the task view: abort the wizard quietly. + return None + if rc != 0: + raise _TuiError( + f"audio.cpp setup step failed (exit {rc}). Clone " + f"audio.cpp manually: git clone " + f"{AUDIOCPP_GIT_URL} {target}, then re-run") + audiocpp_dir = target + resolve_checkout(audiocpp_dir) + first = _after_families() + return tui.Wizard().run(first) + + +def _execute_lanes(settings: dict, + args: argparse.Namespace) -> List[taskview.TaskLane]: + """Build the ordered setup steps for the in-TUI task view, per lane. + + The same work ``_execute`` runs on the console, split into two lanes so + the view can run the build in one pane while configuring and downloading + models in the other (both progress bars visible at once). The build lane + exists only when ``settings["build"]`` is set; the models lane always + exists (transcribe → write server.json → download/print commands). + Shared results (the transcription mapping) travel through a small closure + dict scoped to the models lane. Each step's ``work(emit, cancel)`` + returns its exit code; subprocess steps stream through EMIT and abort on + CANCEL, while print()-based steps are captured by the view's stdout + routing. + """ + audiocpp_dir = settings["audiocpp_dir"] + state: dict = {} + build = settings.get("build") + lanes: List[taskview.TaskLane] = [] + + if build: + def build_step(emit, cancel): + rc = _build.build_audiocpp(audiocpp_dir, settings["backend"], + emit=emit, cancel=cancel) + if rc != 0: + print(f"[WARNING] build exited with code {rc}; the server.json " + "was still written — build audiocpp_server manually " + "before starting it") + else: + print("[OK] build complete") + return rc + lanes.append(taskview.TaskLane( + "Build", + [taskview.TaskStep( + f"Build audiocpp_server ({settings['backend']})", + build_step)])) + + def transcribe(emit, cancel): + args.input_dir = settings["wav_dir"] + if settings["include_clone"] and args.input_dir is not None: + transcripts, write_prompt = _voices._transcribe( + args, plan=settings["plan"], cancel=cancel) + 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 + state["transcripts"] = transcripts + state["write_prompt"] = write_prompt + return 0 + + def write(emit, cancel): + # Port sync (applied now that the terminal is back). + if settings["sync_port"] is True: + _configsync._apply_port_sync(settings["port"], True) + elif settings["sync_port"] is False: + _configsync._apply_port_sync(settings["port"], False) + + _write_and_advise( + audiocpp_dir, settings["wav_dir"], settings["output_path"], + settings["model_entries"], settings["install_guidance"], + settings["host"], settings["port"], settings["backend"], + settings["lazy_load"], state["transcripts"], state["write_prompt"]) + + # Delete-unused cleanup (modify flow): remove the already-downloaded + # models the new selection dropped. The regenerated server.json + # already only lists the kept entries. + if settings.get("delete_unused"): + removed = _models.delete_model_files(settings["output_path"], + settings["unused_entries"]) + print(f"[OK] Deleted {removed} unused model " + f"{'entry' if removed == 1 else 'entries'} from disk.") + + if len(settings["entry_ids"]) == 1: + _configsync._offer_config_model_id_sync(settings["entry_ids"][0], + settings["sync_model_ids"]) + _voices.print_empty_transcript_warning(state["transcripts"]) + return 0 + + def install(emit, cancel): + _models._install_models(audiocpp_dir, settings["install_guidance"], + settings["download"], emit=emit, cancel=cancel) + _build._print_launch_hint(audiocpp_dir, settings["output_path"]) + return 0 + install_title = "Download models" if settings.get("download") \ + else "Print model install commands" + + lanes.append(taskview.TaskLane( + "Configure & download", + [taskview.TaskStep("Transcribe reference voices", transcribe), + taskview.TaskStep("Write server.json & sync config", write), + taskview.TaskStep(install_title, install)])) + + return lanes + + +def _execute_steps(settings: dict, + args: argparse.Namespace) -> List[taskview.TaskStep]: + """The ordered setup steps for the sequential console path. + + The lanes ``_execute_lanes`` builds, flattened into one ordered list + (build first, then transcribe → write → download), so the console tail + is byte-identical to the pre-lanes behavior. + """ + steps: List[taskview.TaskStep] = [] + for lane in _execute_lanes(settings, args): + steps.extend(lane.steps) + return steps + + +def _execute(settings: dict, args: argparse.Namespace) -> int: + """Shared console tail: build, sync, transcribe, write, install, advise. + + Runs after the TUI wizard returns (or after _collect_from_flags for a + non-interactive run): the terminal is plain, so subprocess output and + transcription progress appear normally. The same work as + ``_execute_steps``, run with no emit (console streaming). + """ + return taskview.run_steps_inline(_execute_steps(settings, args)) + + +def setup_screen(stdscr) -> int: + """Run the setup wizard on an existing curses screen (the hub's). + + The hub drives this as one screen of its own ``tui.Wizard`` stack, so + Esc on the wizard's first screen simply returns here and the hub pops + back to the menu that launched it. The setup tail (build, transcribe, + write, download) runs inside the TUI task view on this same screen, so + the hub's curses session stays intact and the user sees per-step status + and progress instead of being dropped to the console. On a fresh install + the build and the model setup run as two parallel lanes (a split view), + so cloning → configuring → building+downloading is one continuous, + one-click flow; the individual "Build" and "Download Missing Models" hub + actions remain only as fallbacks when something fails or is interrupted. + Returns 0 on completion, 1 when the user aborted. + """ + parser = build_parser() + args = parser.parse_args([]) + settings = _wizard(stdscr, args, parser) + if settings is None: + return 1 + return taskview.run_lanes(stdscr, "Setting up audio.cpp", + _execute_lanes(settings, args)) + + +def build_screen(stdscr) -> int: + """Build audiocpp_server from the hub when the checkout has no binary. + + Asks which backend to build for (pre-selecting the backend an existing + server.json records, else cuda), runs the build inside the TUI task view + — alongside a download of any missing models when server.json is already + configured and those models map to an install command (the split view), + or just the build otherwise — then updates server.json's ``backend`` + field to match. Returns 0 on success, non-zero when the user backed out, + cancelled, or the build failed. This is the hub's "Build audio.cpp + server" action, so a checkout that was cloned but never built is always + buildable from the TUI; the standalone "Download Missing Models" action + stays as the fallback when the download fails or is interrupted. + """ + checkout = _build.find_local_checkout() + if checkout is None: + tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp " + "first.", "err") + return 1 + if _build.find_audiocpp_server_bin(checkout) is not None: + tui.flash(stdscr, "audiocpp_server is already built.", "ok") + return 0 + server_config = load_server_config(checkout / "server.json") or {} + recorded = server_config.get("backend") + options, default = _backend_options(None) + if recorded in BACKENDS: + default = next((i for i, (_label, value) in enumerate(options) + if value == recorded), default) + backend = tui.menu( + stdscr, "Which inference backend should audiocpp_server be built " + "for?", options, default_index=default, back_value=_GO_BACK) + if backend is _GO_BACK: + return 1 + + def build_step(emit, cancel): + return _build.build_audiocpp(checkout, backend, emit=emit, cancel=cancel) + + lanes = [taskview.TaskLane( + "Build", [taskview.TaskStep( + f"Build audiocpp_server ({backend})", build_step)])] + + # Missing models this build can also fetch, so a configured backend that + # lost its binary is restored to "installed" in one step. + server_json = checkout / "server.json" + missing = _models.missing_model_entries(server_json) if server_json.exists() else [] + guidance = _models.missing_model_install_guidance(checkout, missing) \ + if missing else [] + + if guidance: + def download_step(emit, cancel): + _models.install_models(checkout, guidance, emit=emit, cancel=cancel) + return 0 + lanes.append(taskview.TaskLane( + "Download models", + [taskview.TaskStep("Download missing models", download_step)])) + + title = "Build & download models" if len(lanes) == 2 \ + else "Build audiocpp_server" + rc = taskview.run_lanes(stdscr, title, lanes) + if rc != 0: + return rc + if _configsync.update_server_backend(backend): + tui.flash(stdscr, f"audiocpp_server built for {backend}.", "ok") + else: + tui.flash(stdscr, f"audiocpp_server built for {backend}. (Could not " + "update server.json's backend field — reconfigure audio.cpp " + "if it was already configured.)", "warn") + # Models that can't be mapped to an install command still need hand + # installation; say so now rather than leaving the user in the dark. + if missing and not guidance: + tui.flash(stdscr, _models.hand_install_guidance(checkout, missing), "err") + return 0 + + +def run_tui(args: Optional[argparse.Namespace] = None, + parser: Optional[argparse.ArgumentParser] = None) -> int: + """Run the audio.cpp setup wizard end-to-end. + + With no ARGS (the hub's call) a default namespace is built so the full + wizard runs. Called from ``main`` after argparse when the terminal is + interactive. Returns the process exit code. + """ + import curses + if args is None: + parser = build_parser() + args = parser.parse_args([]) + if args.input_dir is not None and not args.input_dir.is_dir(): + print(f"[ERROR] --wavs not found: {args.input_dir}", + file=sys.stderr) + return 2 + 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 + 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 + return _execute(settings, args) + + +def _collect_from_flags(args: argparse.Namespace, + parser: argparse.ArgumentParser) -> Optional[dict]: + """Build the settings dict from flags for a non-interactive run. + + Every required value must come from a flag (there are no prompts in a + non-interactive run); a missing one is a hard ``parser.error``. Returns + the settings dict, or None when the user declined an overwrite (the + default-location fallback then also exists). + """ + # Checkout: ./app/audio.cpp, else --clone clones one there. + audiocpp_dir = _build.find_local_checkout() + if audiocpp_dir is None and args.clone: + target = APP_DIR / AUDIOCPP_DIR_NAME + rc = common.git_clone(AUDIOCPP_GIT_URL, target) + if rc != 0: + parser.error(f"git clone failed (exit {rc}); clone audio.cpp " + f"manually: git clone {AUDIOCPP_GIT_URL} {target}") + patch_rc = _build.apply_ggml_patches(target) + if patch_rc != 0: + parser.error( + f"ggml build patches could not be applied to {target} " + f"(exit {patch_rc}); see messages above. The audio.cpp " + f"fork's vendored ggml may have changed — re-evaluate " + f"app/backends/patches/.") + audiocpp_dir = target + if audiocpp_dir is None: + parser.error( + "An audio.cpp checkout is required. Pass --clone to clone " + "app/audio.cpp, or run without flags for the TUI wizard.") + try: + catalog = load_model_catalog(audiocpp_dir) + except NotADirectoryError as exc: + parser.error(str(exc)) + if not catalog: + parser.error( + f"No TTS model families found in {audiocpp_dir}/model_specs; " + "check the checkout is up to date") + catalog_by_family = {entry["family"]: entry for entry in catalog} + + # Families: required from --families in a non-interactive run. + if args.families is None: + parser.error("--families is required in a non-interactive run (or run " + "without flags for the TUI wizard)") + 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: + parser.error( + f"Unknown family in --families: {', '.join(unknown)}. " + f"Available: {', '.join(catalog_by_family)}") + family_keys: List[str] = [] + for fam in requested: + if fam not in family_keys: + family_keys.append(fam) + + chosen: Dict[str, List[dict]] = {} + for family in family_keys: + opts = package_dir_options(catalog_by_family[family]) + if args.all_packages: + chosen[family] = opts + else: + chosen[family] = [opt for opt in opts if opt["recommended"]] + + # Non-interactive picker: design packages default to vdes. + def task_picker(install_id: str) -> str: + return TASK_VDES + + model_entries, entry_ids, install_guidance, design_entry_ids, include_clone = \ + _build_entries(family_keys, chosen, catalog_by_family, + task_picker) + + # Server settings. + host = args.host or DEFAULT_HOST + detected_backend = detect_backend(audiocpp_dir) + if args.build_backend: + backend = args.build_backend + build = detected_backend is None + elif args.backend: + backend = args.backend + build = False + elif detected_backend is not None: + backend = detected_backend + build = False + else: + backend = "cuda" + build = False + port = args.port if args.port is not None else _configsync.config_port() + lazy_load = True + + # Output path / overwrite (decline falls back to cwd, then aborts). + output_path = args.output if args.output is not None \ + else audiocpp_dir / "server.json" + if output_path.exists() and not args.force: + if args.output is None: + output_path = Path.cwd() / "server.json" + if output_path.exists() and not args.force: + print("[INFO] Aborted; existing server.json kept") + return None + else: + print("[INFO] Aborted; existing server.json kept") + return None + + # Config sync decisions (auto-apply unless explicitly declined). + sync_port: Optional[bool] = None + if port != _configsync.config_port(): + sync_port = not args.no_sync_port + 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 = not args.no_sync_model_ids + + # Wav dir + transcription plan (defaults to the project's voices/ dir). + wav_dir = args.input_dir if args.input_dir is not None else VOICES_DIR + 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 + plan = _voices._flag_plan(wav_files, prompt_path, args.force) + + 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, + "build": build, + "lazy_load": lazy_load, + "sync_port": sync_port, + "sync_model_ids": sync_model_ids, + "wav_dir": wav_dir, + "plan": plan, + "download": args.download, + } + + +def build_parser() -> argparse.ArgumentParser: + """The audio.cpp setup CLI (also used to build a default namespace).""" + parser = argparse.ArgumentParser( + description="Set up the audio.cpp TTS backend: clone/build, pick " + "models, write server.json, and sync app/converter/config.py.") + 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 " + f"(default: {VOICES_DIR}; asked for when omitted " + "in the TUI)") + parser.add_argument("--output", type=Path, default=None, + help="Output path for server.json (default: " + "server.json inside the audio.cpp checkout; an " + "existing file is overwritten only with --force " + "or a TUI confirm)") + parser.add_argument("--clone", action="store_true", + help="Non-interactive: clone audio.cpp into " + "./app/audio.cpp when no checkout is found") + 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). Required in a " + "non-interactive run; skips the family tree in " + "the TUI") + parser.add_argument("--all-packages", action="store_true", + help="Host every installable package of each selected " + "family (distinct target_directory) instead of " + "only the recommended one. Voice-design packages " + "are hosted with task 'vdes'") + 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, + help="Port for the server (default: the port in " + "AUDIOCPP_API_URL from app/converter/config.py)") + parser.add_argument("--backend", choices=BACKENDS, default=None, + help="Inference backend recorded in server.json " + "(default: auto-detected from the checkout's " + "build/ directory, else cuda)") + parser.add_argument("--build-backend", choices=BACKENDS, default=None, + help="Build audiocpp_server for this backend when it " + "is not built yet, and use it in server.json") + parser.add_argument("--whisper-model", type=str, default="base", + help="Whisper model size for transcription " + "(default: base)") + parser.add_argument("--force", action="store_true", + help="Overwrite the output file (and prompt_text) " + "without prompting; in the TUI, start the " + "wizard fresh instead of loading the existing " + "server.json") + parser.add_argument("--download", action="store_true", + help="Run model_manager_v2.py install for each hosted " + "model automatically (default: print the commands " + "only)") + parser.add_argument("--no-sync-port", action="store_true", + help="Do not rewrite AUDIOCPP_API_URL in " + "app/converter/config.py when --port differs") + parser.add_argument("--no-sync-model-ids", action="store_true", + help="Do not rewrite AUDIOCPP_MODEL_ID/" + "AUDIOCPP_CLONE_MODEL_ID for a single-entry server") + return parser + + +def main() -> int: + parser = build_parser() + 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 _interactive(): + return run_tui(args, parser) + + # Non-interactive (no terminal, or all flags supplied): flag-only path. + settings = _collect_from_flags(args, parser) + if settings is None: + return 1 + return _execute(settings, args) + + diff --git a/app/backends/faster.py b/app/backends/faster.py index cec59a6..6e2735c 100755 --- a/app/backends/faster.py +++ b/app/backends/faster.py @@ -39,6 +39,7 @@ from backends import ( format_launch_hint, probe, servers, + setup, ) from backends.common import ( APP_DIR, @@ -447,40 +448,21 @@ def _print_launch_hint(voices_path: Path, port: int) -> None: def setup_screen(stdscr) -> int: """Run the setup wizard on an existing curses screen (the hub's). - The hub drives this as one screen of its own ``tui.Wizard`` stack, so - Esc on the wizard's first screen simply returns here and the hub pops - back to the menu that launched it. The setup tail (install/clone/ - transcribe/write) runs inside the TUI task view on this same screen, so - the hub's curses session stays intact and the user sees per-step status - instead of being dropped to the console. Returns 0 on completion, 1 when - the user aborted. + See backends.setup.screen_flow for the shared flow. Returns 0 on + completion, 1 when the user aborted. """ - args = build_parser().parse_args([]) - settings = _wizard(stdscr, args) - if settings is None: - return 1 - return taskview.run_steps(stdscr, "Setting up faster-qwen3-tts", - _execute_steps(settings)) + return setup.screen_flow(stdscr, wizard=_wizard, + steps_of=_execute_steps, + title="Setting up faster-qwen3-tts", + parser_factory=build_parser) def run_tui(args: Optional[argparse.Namespace] = None) -> int: """Run the faster setup wizard end-to-end.""" - import curses if args is None: args = build_parser().parse_args([]) - try: - settings = curses.wrapper(_wizard, args) - except tui.WizardCancelled: - print("\n[INFO] Cancelled; nothing was written") - return 1 - try: - curses.curs_set(1) - except curses.error: - pass - if settings is None: - print("[INFO] Aborted") - return 1 - return _execute(settings) + return setup.tui_flow(_wizard, _execute, args=args, + aborted_message="[INFO] Aborted") def _collect_from_flags(args: argparse.Namespace, @@ -646,7 +628,7 @@ def main() -> int: parser = build_parser() args = parser.parse_args() - if _interactive(): + if setup.interactive(): return run_tui(args) settings = _collect_from_flags(args, parser) @@ -655,16 +637,5 @@ def main() -> int: return _execute(settings) -def _interactive() -> bool: - try: - import curses # noqa: F401 - except ImportError: - return False - try: - return sys.stdin.isatty() and sys.stdout.isatty() - except (AttributeError, ValueError): - return False - - if __name__ == "__main__": sys.exit(main()) diff --git a/app/backends/qwen.py b/app/backends/qwen.py index 501d929..c0aafef 100644 --- a/app/backends/qwen.py +++ b/app/backends/qwen.py @@ -28,6 +28,7 @@ from backends import ( format_launch_hint, probe, servers, + setup, ) from converter import config from converter.clients import QWEN3_TTS_SPEAKERS @@ -40,8 +41,8 @@ DEFAULT_CUSTOM_PORT = 7860 DEFAULT_CLONE_PORT = 7861 # Built-in CustomVoice speakers (see app/converter/config.py SPEAKER). The -# canonical list lives in converter.tts (shared with the audiocpp backend's -# Convert-form Speaker picker). +# canonical list lives in converter.clients.speakers (shared with the +# audio.cpp backend's Convert-form Speaker picker). QWEN_SPEAKERS = QWEN3_TTS_SPEAKERS @@ -203,40 +204,21 @@ def _execute(settings: dict) -> int: def setup_screen(stdscr) -> int: """Run the setup wizard on an existing curses screen (the hub's). - The hub drives this as one screen of its own ``tui.Wizard`` stack, so - Esc on the wizard's first screen simply returns here and the hub pops - back to the menu that launched it. The setup tail (pip install / config - sync) runs inside the TUI task view on this same screen, so the hub's - curses session stays intact and the user sees per-step status instead of - being dropped to the console. Returns 0 on completion, 1 when the user - aborted. + See backends.setup.screen_flow for the shared flow. Returns 0 on + completion, 1 when the user aborted. """ - args = build_parser().parse_args([]) - settings = _wizard(stdscr, args) - if settings is None: - return 1 - return taskview.run_steps(stdscr, "Setting up qwen-tts", - _execute_steps(settings)) + return setup.screen_flow(stdscr, wizard=_wizard, + steps_of=_execute_steps, + title="Setting up qwen-tts", + parser_factory=build_parser) def run_tui(args: Optional[argparse.Namespace] = None) -> int: """Run the qwen setup wizard end-to-end.""" - import curses if args is None: args = build_parser().parse_args([]) - try: - settings = curses.wrapper(_wizard, args) - except tui.WizardCancelled: - print("\n[INFO] Cancelled; nothing was written") - return 1 - try: - curses.curs_set(1) - except curses.error: - pass - if settings is None: - print("[INFO] Aborted") - return 1 - return _execute(settings) + return setup.tui_flow(_wizard, _execute, args=args, + aborted_message="[INFO] Aborted") def _collect_from_flags(args: argparse.Namespace, @@ -375,23 +357,12 @@ def main() -> int: parser = build_parser() args = parser.parse_args() - if _interactive(): + if setup.interactive(): return run_tui(args) settings = _collect_from_flags(args, parser) return _execute(settings) -def _interactive() -> bool: - try: - import curses # noqa: F401 - except ImportError: - return False - try: - return sys.stdin.isatty() and sys.stdout.isatty() - except (AttributeError, ValueError): - return False - - if __name__ == "__main__": sys.exit(main()) diff --git a/app/backends/setup.py b/app/backends/setup.py new file mode 100644 index 0000000..bd43b92 --- /dev/null +++ b/app/backends/setup.py @@ -0,0 +1,73 @@ +"""Shared entry-point plumbing for the backend setup wizards. + +Every backend module exposes the same surface — ``_wizard`` (TUI screens), +``_execute_steps``/``_execute`` (the work, as task-view steps), +``setup_screen`` (hub-embedded flow), ``run_tui`` (standalone curses +flow), ``main`` (CLI) — and this module holds the parts of that surface +that are identical everywhere. The backend keeps the thin named wrappers +so its public names (and test seams) stay on the backend module. +""" + +import sys + +from ui import taskview, tui + + +def interactive() -> bool: + """True when the TUI wizard can run (curses importable + tty).""" + try: + import curses # noqa: F401 + except ImportError: + return False + try: + return sys.stdin.isatty() and sys.stdout.isatty() + except (AttributeError, ValueError): + return False + + +def screen_flow(stdscr, *, wizard, steps_of, title, + parser_factory) -> int: + """Run the setup wizard on an existing curses screen (the hub's). + + The hub drives this as one screen of its own ``tui.Wizard`` stack, so + Esc on the wizard's first screen simply returns here and the hub pops + back to the menu that launched it. The setup tail runs inside the TUI + task view on this same screen, so the hub's curses session stays intact + and the user sees per-step status instead of being dropped to the + console. Returns 0 on completion, 1 when the user aborted. + + WIZARD is ``(stdscr, args) -> settings | None``; STEPS_OF turns the + settings into task-view steps; PARSER_FACTORY builds the argparse + parser whose empty namespace seeds the wizard. + """ + args = parser_factory().parse_args([]) + settings = wizard(stdscr, args) + if settings is None: + return 1 + return taskview.run_steps(stdscr, title, steps_of(settings)) + + +def tui_flow(wizard, execute, *, args, execute_takes_args=False, + aborted_message="[INFO] Aborted") -> int: + """Run the setup wizard end-to-end in its own curses session. + + WIZARD is ``(args) -> settings | None`` via ``curses.wrapper``; + EXECUTE performs the settings (called with ARGS too when + EXECUTE_TAKES_ARGS). Restores the text cursor afterwards and reports + cancellation/abort consistently. Returns the process exit code. + """ + import curses + try: + settings = curses.wrapper(wizard, args) + except tui.WizardCancelled: + print("\n[INFO] Cancelled; nothing was written") + return 1 + try: + curses.curs_set(1) + except curses.error: + pass + if settings is None: + print(aborted_message) + return 1 + return execute(settings) if not execute_takes_args \ + else execute(settings, args) diff --git a/app/docs/backend-audiocpp.md b/app/docs/backend-audiocpp.md index b292728..c098d57 100644 --- a/app/docs/backend-audiocpp.md +++ b/app/docs/backend-audiocpp.md @@ -2,7 +2,7 @@ `--backend audiocpp` talks to `audiocpp_server` from [audio.cpp](https://github.com/0xShug0/audio.cpp), which hosts numerous TTS model families. -The easiest way is the TUI: run `python audiobook.py`, choose **Configure backends… → Install Backend → audio.cpp**, and it clones `audio.cpp` into `app/audio.cpp` (or reuses an existing checkout), builds `audiocpp_server`, lets you pick model families/packages from an expandable checkbox tree (reading the checkout's `model_specs/`), transcribes `.wav` voices with `whisper`, writes `server.json` into the checkout, syncs `app/converter/config.py`, and prints the launch command (the hub can also start the server for you via the **Start/Stop Backend Servers** menu or automatically when converting). The clone, build, transcription and model downloads all run inside the TUI — each shows a status (and, where the tool can measure it, a progress bar), and can be cancelled — instead of dropping to console output. Run it directly with `python app/backends/audiocpp.py` (flags like `--wavs`, `--families`, `--build-backend`, `--clone` skip the corresponding screens for scripting). The TUI runs in the managed `app/envs/tts` venv, which includes `whisper` via `requirements.txt`; for a manual setup, make sure `whisper` (or `faster_whisper`) is installed in the environment you run the wizard from. The Qwen3-TTS model tree also offers hosting the VoiceDesign package as a `vdes` entry. +The easiest way is the TUI: run `python audiobook.py`, choose **Configure backends… → Install Backend → audio.cpp**, and it clones `audio.cpp` into `app/audio.cpp` (or reuses an existing checkout), builds `audiocpp_server`, lets you pick model families/packages from an expandable checkbox tree (reading the checkout's `model_specs/`), transcribes `.wav` voices with `whisper`, writes `server.json` into the checkout, syncs `app/converter/config.py`, and prints the launch command (the hub can also start the server for you via the **Start/Stop Backend Servers** menu or automatically when converting). The clone, build, transcription and model downloads all run inside the TUI — each shows a status (and, where the tool can measure it, a progress bar), and can be cancelled — instead of dropping to console output. Run it directly with `python -m backends.audiocpp` (from `app/`) (flags like `--wavs`, `--families`, `--build-backend`, `--clone` skip the corresponding screens for scripting). The TUI runs in the managed `app/envs/tts` venv, which includes `whisper` via `requirements.txt`; for a manual setup, make sure `whisper` (or `faster_whisper`) is installed in the environment you run the wizard from. The Qwen3-TTS model tree also offers hosting the VoiceDesign package as a `vdes` entry. The hub's backend status table distinguishes how far audio.cpp is set up: `unavailable` (nothing present), `downloaded (not built)` (checkout cloned, `audiocpp_server` not built), `built (not configured)` (binary built, no `server.json`), `installed` (ready; or `installed (models missing)` when the config references undownloaded models), and `running` once its server answers. Whenever the checkout exists but `audiocpp_server` is missing, **Configure backends… → Build audio.cpp server** builds it from the TUI (the wizard offers the build during setup too), so a backend whose build you skipped is never stuck as "unavailable". On a fresh install the setup is one continuous flow: clone → configure → and then the build and the model downloads run **simultaneously** in a split view (half building, half downloading). The setup steps are therefore ordered build > configure > download, and **Build audio.cpp server** and **Download Missing Models (audio.cpp)** are never offered at the same time; **Build audio.cpp server** downloads any missing models alongside the build, and **Download Missing Models (audio.cpp)** remains only as a fallback for when a download fails or is interrupted. diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index 47dc5d4..4ff6f6e 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -71,7 +71,7 @@ class DetectAllTests(unittest.TestCase): (checkout / "server.json").write_text('{"models":[]}', encoding="utf-8") from backends import audiocpp - with patch.object(audiocpp, "find_local_checkout", + with patch.object(audiocpp.build, "find_local_checkout", return_value=checkout), \ patch("backends.common.server_running", return_value=False): @@ -84,9 +84,9 @@ class DetectAllTests(unittest.TestCase): def test_audiocpp_running_when_remote_server_identified(self): from backends import audiocpp - with patch.object(audiocpp, "find_local_checkout", + with patch.object(audiocpp.build, "find_local_checkout", return_value=None), \ - patch.object(audiocpp.probe, "identify_server", + patch.object(audiocpp.status.probe, "identify_server", return_value="audiocpp"): status = audiocpp.detect() # Not installed (no checkout) but a remote server answers. @@ -280,12 +280,12 @@ class RemoteSuppressionTests(unittest.TestCase): encoding="utf-8") (Path(td) / "audiocpp-server.pid").write_text( "4242", encoding="utf-8") - with patch.object(audiocpp, "find_local_checkout", + with patch.object(audiocpp.build, "find_local_checkout", return_value=checkout), \ patch.object(servers_mod, "LOG_DIR", Path(td)), \ patch.object(servers_mod, "_pid_alive", return_value=True), \ - patch.object(audiocpp.probe, "identify_server", + patch.object(audiocpp.status.probe, "identify_server", return_value="audiocpp"): status = audiocpp.detect() self.assertTrue(status.managed) diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 05b47bd..59bd039 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -13,7 +13,10 @@ from unittest.mock import MagicMock, patch from converter import config from backends import audiocpp as make_server -from backends import common +import os + +from backends import common, servers +from ui import taskview from ui import tui FAKE_CONFIG = ( @@ -132,17 +135,17 @@ class FindWavFilesTests(unittest.TestCase): self._touch("notes.txt") (self.folder / "sub").mkdir() (self.folder / "sub" / "c.wav").write_bytes(b"x") - names = [path.name for path in make_server.find_wav_files(self.folder)] + names = [path.name for path in common.find_wav_files(self.folder)] self.assertEqual(names, ["a.WAV", "b.wav"]) def test_sorted_alphabetically_case_insensitive(self): for name in ("Zed.wav", "alpha.wav", "Beta.wav"): self._touch(name) - names = [path.name for path in make_server.find_wav_files(self.folder)] + names = [path.name for path in common.find_wav_files(self.folder)] self.assertEqual(names, ["alpha.wav", "Beta.wav", "Zed.wav"]) def test_empty_directory_returns_empty_list(self): - self.assertEqual(make_server.find_wav_files(self.folder), []) + self.assertEqual(common.find_wav_files(self.folder), []) class DetectWavDirTests(unittest.TestCase): @@ -167,42 +170,42 @@ class DetectWavDirTests(unittest.TestCase): def test_unique_wav_dir_in_tts_root_returned(self): found = self._wav_dir(self.tts_root, "voices") - self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.assertEqual(common.detect_wav_dir(self.audiocpp, self.tts_root), found) def test_unique_wav_dir_in_audiocpp_root_returned(self): found = self._wav_dir(self.audiocpp, "reference") - self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.assertEqual(common.detect_wav_dir(self.audiocpp, self.tts_root), found) def test_root_itself_containing_wavs_returned(self): (self.tts_root / "direct.wav").write_bytes(b"x") - self.assertEqual(make_server.detect_wav_dir(self.audiocpp, + self.assertEqual(common.detect_wav_dir(self.audiocpp, self.tts_root), self.tts_root) def test_multiple_wav_dirs_returns_none(self): self._wav_dir(self.tts_root, "one") self._wav_dir(self.audiocpp, "two") - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.assertIsNone(common.detect_wav_dir(self.audiocpp, self.tts_root)) def test_output_dir_of_tts_root_excluded(self): self._wav_dir(self.tts_root, "output") - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.assertIsNone(common.detect_wav_dir(self.audiocpp, self.tts_root)) def test_no_wavs_returns_none(self): - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.assertIsNone(common.detect_wav_dir(self.audiocpp, self.tts_root)) def test_nested_wav_dir_not_seen(self): nested = self.tts_root / "outer" / "inner" nested.mkdir(parents=True) (nested / "voice.wav").write_bytes(b"x") - self.assertIsNone(make_server.detect_wav_dir(self.audiocpp, + self.assertIsNone(common.detect_wav_dir(self.audiocpp, self.tts_root)) @@ -210,27 +213,27 @@ class ConfigPortTests(unittest.TestCase): def test_port_parsed_from_config_url(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1:8080"): - self.assertEqual(make_server.config_port(), 8080) + self.assertEqual(make_server.configsync.config_port(), 8080) def test_missing_port_falls_back(self): with patch.object(config, "AUDIOCPP_API_URL", "http://127.0.0.1"): - self.assertEqual(make_server.config_port(), + self.assertEqual(make_server.configsync.config_port(), make_server.FALLBACK_PORT) def test_invalid_url_falls_back(self): with patch.object(config, "AUDIOCPP_API_URL", "not a url"): - self.assertEqual(make_server.config_port(), + self.assertEqual(make_server.configsync.config_port(), make_server.FALLBACK_PORT) def test_url_with_port_replaces_port(self): # audiocpp reuses the shared helper (backends.common.url_with_port). self.assertEqual( - make_server.url_with_port("http://127.0.0.1:8080", 9000), + common.url_with_port("http://127.0.0.1:8080", 9000), "http://127.0.0.1:9000") def test_url_without_port_adds_port(self): self.assertEqual( - make_server.url_with_port("http://localhost", 8080), + common.url_with_port("http://localhost", 8080), "http://localhost:8080") @@ -247,7 +250,7 @@ class UpdateConfigPortTests(unittest.TestCase): self._tmp.cleanup() def test_rewrites_port_preserving_comment(self): - changed = make_server.update_config_api_url_port( + changed = make_server.configsync.update_config_api_url_port( 8080, config_path=self.config_path) self.assertTrue(changed) text = self.config_path.read_text(encoding="utf-8") @@ -260,18 +263,18 @@ class UpdateConfigPortTests(unittest.TestCase): def test_returns_false_when_no_url_line(self): path = Path(self._tmp.name) / "other.py" path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") - self.assertFalse(make_server.update_config_api_url_port( + self.assertFalse(make_server.configsync.update_config_api_url_port( 8080, config_path=path)) def test_port_unchanged_is_a_success_noop(self): # The file already holds the port: success, nothing rewritten. - self.assertTrue(make_server.update_config_api_url_port( + self.assertTrue(make_server.configsync.update_config_api_url_port( 9999, config_path=self.config_path)) self.assertEqual(self.config_path.read_text(encoding="utf-8"), FAKE_CONFIG) def test_returns_false_when_file_missing(self): - self.assertFalse(make_server.update_config_api_url_port( + self.assertFalse(make_server.configsync.update_config_api_url_port( 8080, config_path=Path(self._tmp.name) / "nope.py")) @@ -291,7 +294,7 @@ class UpdateConfigModelIdsTests(unittest.TestCase): self._tmp.cleanup() def test_rewrites_both_ids_preserving_lines(self): - changed = make_server.update_config_model_ids( + changed = make_server.configsync.update_config_model_ids( "higgs", "higgs", config_path=self.config_path) self.assertTrue(changed) text = self.config_path.read_text(encoding="utf-8") @@ -301,7 +304,7 @@ class UpdateConfigModelIdsTests(unittest.TestCase): self.assertIn('AUDIOCPP_API_URL = "http://127.0.0.1:9999"', text) def test_clone_id_optional(self): - changed = make_server.update_config_model_ids( + changed = make_server.configsync.update_config_model_ids( "voxcpm2", config_path=self.config_path) self.assertTrue(changed) text = self.config_path.read_text(encoding="utf-8") @@ -310,7 +313,7 @@ class UpdateConfigModelIdsTests(unittest.TestCase): def test_ids_unchanged_is_a_success_noop(self): # Both ids already hold their values: success, nothing rewritten. - changed = make_server.update_config_model_ids( + changed = make_server.configsync.update_config_model_ids( "qwen", "qwen-clone", config_path=self.config_path) self.assertTrue(changed) self.assertEqual(self.config_path.read_text(encoding="utf-8"), @@ -319,11 +322,11 @@ class UpdateConfigModelIdsTests(unittest.TestCase): def test_returns_false_when_lines_missing(self): path = Path(self._tmp.name) / "other.py" path.write_text('CHUNK_SIZE = 250\n', encoding="utf-8") - self.assertFalse(make_server.update_config_model_ids( + self.assertFalse(make_server.configsync.update_config_model_ids( "higgs", "higgs", config_path=path)) def test_returns_false_when_file_missing(self): - self.assertFalse(make_server.update_config_model_ids( + self.assertFalse(make_server.configsync.update_config_model_ids( "higgs", "higgs", config_path=Path(self._tmp.name) / "nope.py")) @@ -337,32 +340,32 @@ class ResolveWavDirArgTests(unittest.TestCase): self._tmp.cleanup() def test_resolves_to_absolute(self): - self.assertEqual(make_server.resolve_wav_dir_arg(str(self.folder)), + self.assertEqual(common.resolve_wav_dir_arg(str(self.folder)), self.folder.resolve()) def test_strips_surrounding_quotes(self): quoted = f'"{self.folder}"' - self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.assertEqual(common.resolve_wav_dir_arg(quoted), self.folder.resolve()) def test_strips_single_quotes(self): quoted = f"'{self.folder}'" - self.assertEqual(make_server.resolve_wav_dir_arg(quoted), + self.assertEqual(common.resolve_wav_dir_arg(quoted), self.folder.resolve()) def test_strips_whitespace(self): - self.assertEqual(make_server.resolve_wav_dir_arg(f" {self.folder} "), + self.assertEqual(common.resolve_wav_dir_arg(f" {self.folder} "), self.folder.resolve()) def test_expands_tilde(self): - with patch.object(make_server.os.path, "expanduser", + with patch.object(os.path, "expanduser", return_value=str(self.folder)) as mock_expand: - result = make_server.resolve_wav_dir_arg("~/voices") + result = common.resolve_wav_dir_arg("~/voices") mock_expand.assert_called_once_with("~/voices") self.assertEqual(result, self.folder.resolve()) def test_trailing_slash_preserved_as_dir(self): - self.assertEqual(make_server.resolve_wav_dir_arg(f"{self.folder}/"), + self.assertEqual(common.resolve_wav_dir_arg(f"{self.folder}/"), self.folder.resolve()) @@ -388,20 +391,20 @@ class FindLocalCheckoutTests(unittest.TestCase): def test_none_when_no_checkout_in_app_dir(self): with tempfile.TemporaryDirectory() as td, \ - patch.object(make_server, "APP_DIR", Path(td)): - self.assertIsNone(make_server.find_local_checkout()) + patch.object(make_server.build, "APP_DIR", Path(td)): + self.assertIsNone(make_server.build.find_local_checkout()) def test_returns_the_managed_checkout(self): with tempfile.TemporaryDirectory() as td, \ - patch.object(make_server, "APP_DIR", Path(td)): + patch.object(make_server.build, "APP_DIR", Path(td)): checkout = _make_checkout(Path(td)) - self.assertEqual(make_server.find_local_checkout(), checkout) + self.assertEqual(make_server.build.find_local_checkout(), checkout) def test_none_when_checkout_lacks_model_specs(self): with tempfile.TemporaryDirectory() as td, \ - patch.object(make_server, "APP_DIR", Path(td)): + patch.object(make_server.build, "APP_DIR", Path(td)): (Path(td) / "audio.cpp").mkdir() - self.assertIsNone(make_server.find_local_checkout()) + self.assertIsNone(make_server.build.find_local_checkout()) class LoadModelCatalogTests(unittest.TestCase): @@ -413,7 +416,7 @@ class LoadModelCatalogTests(unittest.TestCase): self._td.cleanup() def test_includes_tts_families_excludes_asr(self): - catalog = make_server.load_model_catalog(self.checkout) + catalog = make_server.catalog.load_model_catalog(self.checkout) families = [entry["family"] for entry in catalog] self.assertIn("qwen3_tts", families) self.assertIn("higgs_audio_tts", families) @@ -422,19 +425,19 @@ class LoadModelCatalogTests(unittest.TestCase): self.assertNotIn("qwen3_asr", families) def test_skips_families_with_no_packages(self): - catalog = make_server.load_model_catalog(self.checkout) + catalog = make_server.catalog.load_model_catalog(self.checkout) self.assertNotIn("empty_tts", [entry["family"] for entry in catalog]) def test_families_sorted_alphabetically_by_display_name(self): - catalog = make_server.load_model_catalog(self.checkout) + catalog = make_server.catalog.load_model_catalog(self.checkout) names = [entry["display_name"].lower() for entry in catalog] self.assertEqual(names, sorted(names)) self.assertNotIn("tested", catalog[0]) self.assertNotIn("TESTED_FAMILIES", dir(make_server)) def test_default_package_and_target_directory_resolved(self): - catalog = make_server.load_model_catalog(self.checkout) + catalog = make_server.catalog.load_model_catalog(self.checkout) by_family = {entry["family"]: entry for entry in catalog} higgs = by_family["higgs_audio_tts"] self.assertEqual(higgs["install_id"], "higgs_audio_tts_4b_q8_0") @@ -449,12 +452,12 @@ class LoadModelCatalogTests(unittest.TestCase): {"id": "voxcpm2_q8_0", "format": "gguf", "target_directory": "VoxCPM2-GGUF"}, ]) - catalog = make_server.load_model_catalog(self.checkout) + catalog = make_server.catalog.load_model_catalog(self.checkout) by_family = {entry["family"]: entry for entry in catalog} self.assertEqual(by_family["voxcpm2"]["install_id"], "voxcpm2_bf16") def test_clone_capability_from_tasks(self): - catalog = make_server.load_model_catalog(self.checkout) + catalog = make_server.catalog.load_model_catalog(self.checkout) by_family = {entry["family"]: entry for entry in catalog} self.assertTrue(by_family["higgs_audio_tts"]["clone_capable"]) self.assertFalse(by_family["supertonic"]["clone_capable"]) @@ -463,7 +466,7 @@ class LoadModelCatalogTests(unittest.TestCase): empty = Path(self._td.name) / "empty" empty.mkdir() with self.assertRaises(NotADirectoryError): - make_server.load_model_catalog(empty) + make_server.catalog.load_model_catalog(empty) class DetectBackendTests(unittest.TestCase): @@ -485,58 +488,58 @@ class DetectBackendTests(unittest.TestCase): return build_dir def test_no_build_dir_returns_none(self): - self.assertIsNone(make_server.detect_backend(self.checkout)) + self.assertIsNone(make_server.catalog.detect_backend(self.checkout)) def test_unique_linux_backend_detected(self): self._build("linux-cuda-release") - self.assertEqual(make_server.detect_backend(self.checkout), "cuda") + self.assertEqual(make_server.catalog.detect_backend(self.checkout), "cuda") def test_windows_exe_backend_detected(self): self._build("windows-vulkan-debug", binary="audiocpp_server.exe") - self.assertEqual(make_server.detect_backend(self.checkout), "vulkan") + self.assertEqual(make_server.catalog.detect_backend(self.checkout), "vulkan") def test_hip_backend_detected(self): self._build("linux-hip-release") - self.assertEqual(make_server.detect_backend(self.checkout), "hip") + self.assertEqual(make_server.catalog.detect_backend(self.checkout), "hip") def test_cpu_backend_detected(self): self._build("linux-cpu-release") - self.assertEqual(make_server.detect_backend(self.checkout), "cpu") + self.assertEqual(make_server.catalog.detect_backend(self.checkout), "cpu") def test_metal_maps_to_cpu(self): self._build("macos-metal-release") - self.assertEqual(make_server.detect_backend(self.checkout), "cpu") + self.assertEqual(make_server.catalog.detect_backend(self.checkout), "cpu") def test_multiple_backends_returns_none(self): self._build("linux-cuda-release") self._build("linux-cpu-release") - self.assertIsNone(make_server.detect_backend(self.checkout)) + self.assertIsNone(make_server.catalog.detect_backend(self.checkout)) def test_multiple_builds_same_backend_detected(self): self._build("linux-cuda-release") self._build("windows-cuda-debug") - self.assertEqual(make_server.detect_backend(self.checkout), "cuda") + self.assertEqual(make_server.catalog.detect_backend(self.checkout), "cuda") def test_build_dir_without_binary_ignored(self): (self.checkout / "build" / "linux-cuda-release").mkdir(parents=True) - self.assertIsNone(make_server.detect_backend(self.checkout)) + self.assertIsNone(make_server.catalog.detect_backend(self.checkout)) def test_non_matching_build_dir_name_ignored(self): self._build("linux-mybuild-release") - self.assertIsNone(make_server.detect_backend(self.checkout)) + self.assertIsNone(make_server.catalog.detect_backend(self.checkout)) class BackendOptionsTests(unittest.TestCase): """Aligned backend menu labels and the [auto-detected] default.""" def test_options_have_aligned_dashes(self): - options, default_index = make_server._backend_options() + options, default_index = make_server.catalog._backend_options() dash_columns = {label.index(" - ") for label, _ in options} self.assertEqual(len(dash_columns), 1) self.assertEqual(default_index, 0) def test_detected_backend_marked_and_defaulted(self): - options, default_index = make_server._backend_options("vulkan") + options, default_index = make_server.catalog._backend_options("vulkan") labels = [label for label, _ in options] self.assertEqual(default_index, labels.index(next( label for label, value in options @@ -545,22 +548,22 @@ class BackendOptionsTests(unittest.TestCase): self.assertEqual(options[default_index][1], "vulkan") def test_unknown_detected_backend_is_ignored(self): - options, default_index = make_server._backend_options("opencl") + options, default_index = make_server.catalog._backend_options("opencl") self.assertEqual(default_index, 0) self.assertFalse(any("[auto-detected]" in label for label, _ in options)) def test_labels_keep_backend_values(self): - options, _ = make_server._backend_options() + options, _ = make_server.catalog._backend_options() self.assertEqual([value for _, value in options], list(make_server.BACKENDS)) class BuildServerConfigTests(unittest.TestCase): def test_single_entry_without_voice_dir(self): - entry = make_server.build_model_entry( + entry = make_server.catalog.build_model_entry( "higgs_audio_tts", "higgs", "models/Higgs-GGUF") - cfg = make_server.build_server_config( + cfg = make_server.catalog.build_server_config( "127.0.0.1", 8080, "cuda", False, [entry]) self.assertEqual(cfg["host"], "127.0.0.1") self.assertEqual(cfg["port"], 8080) @@ -570,15 +573,15 @@ class BuildServerConfigTests(unittest.TestCase): self.assertNotIn("voice_dir", cfg) def test_voice_dir_added_when_given(self): - entry = make_server.build_model_entry("voxcpm2", "voxcpm2", "models/V") - cfg = make_server.build_server_config( + entry = make_server.catalog.build_model_entry("voxcpm2", "voxcpm2", "models/V") + cfg = make_server.catalog.build_server_config( "0.0.0.0", 9000, "cpu", True, [entry], voice_dir="/abs/voices") self.assertTrue(cfg["lazy_load"]) self.assertEqual(cfg["voice_dir"], "/abs/voices") def test_model_entry_shape(self): - entry = make_server.build_model_entry("index_tts2", "indextts2", "p") + entry = make_server.catalog.build_model_entry("index_tts2", "indextts2", "p") self.assertEqual(entry["id"], "indextts2") self.assertEqual(entry["family"], "index_tts2") self.assertEqual(entry["path"], "p") @@ -586,7 +589,7 @@ class BuildServerConfigTests(unittest.TestCase): self.assertEqual(entry["mode"], "offline") def test_model_entry_design_task(self): - entry = make_server.build_model_entry( + entry = make_server.catalog.build_model_entry( "qwen3_tts", "qwen-design", "p", task="vdes") self.assertEqual(entry["task"], "vdes") self.assertEqual(entry["mode"], "offline") @@ -612,9 +615,9 @@ class InstallModelsTests(unittest.TestCase): def test_declined_download_prints_commands_deduped(self): buf = io.StringIO() with redirect_stdout(buf), \ - patch.object(make_server.common, + patch.object(common, "run_console_subprocess") as run: - make_server._install_models(self.checkout, self.guidance, + make_server.models._install_models(self.checkout, self.guidance, download=False) out = buf.getvalue() self.assertEqual(out.count("install higgs_audio_tts_4b_q8_0"), 1) @@ -622,9 +625,9 @@ class InstallModelsTests(unittest.TestCase): run.assert_not_called() def test_accepted_download_runs_each_command(self): - with patch.object(make_server.common, + with patch.object(common, "run_console_subprocess", return_value=0) as run: - make_server._install_models(self.checkout, self.guidance, + make_server.models._install_models(self.checkout, self.guidance, download=True) self.assertEqual(run.call_count, 2) commands = [call[0][0] for call in run.call_args_list] @@ -641,9 +644,9 @@ class InstallModelsTests(unittest.TestCase): self.manager.unlink() buf = io.StringIO() with redirect_stdout(buf), \ - patch.object(make_server.common, + patch.object(common, "run_console_subprocess") as run: - make_server._install_models(self.checkout, self.guidance, + make_server.models._install_models(self.checkout, self.guidance, download=True) self.assertIn("install higgs_audio_tts_4b_q8_0", buf.getvalue()) run.assert_not_called() @@ -652,9 +655,9 @@ class InstallModelsTests(unittest.TestCase): results = iter([1, 0]) buf = io.StringIO() with redirect_stdout(buf), \ - patch.object(make_server.common, "run_console_subprocess", + patch.object(common, "run_console_subprocess", side_effect=lambda *a, **k: next(results)) as run: - make_server._install_models(self.checkout, self.guidance, + make_server.models._install_models(self.checkout, self.guidance, download=True) self.assertEqual(run.call_count, 2) self.assertIn("exited with code 1", buf.getvalue()) @@ -662,17 +665,17 @@ class InstallModelsTests(unittest.TestCase): def test_decide_download_skips_prompt_without_manager(self): self.manager.unlink() confirm = MagicMock() - self.assertFalse(make_server._decide_download(self.checkout, [], confirm)) + self.assertFalse(make_server.models._decide_download(self.checkout, [], confirm)) confirm.assert_not_called() def test_decide_download_asks_when_manager_present(self): confirm = MagicMock(return_value=True) - self.assertTrue(make_server._decide_download(self.checkout, [], confirm)) + self.assertTrue(make_server.models._decide_download(self.checkout, [], confirm)) confirm.assert_called_once() def test_decide_download_defaults_to_yes(self): confirm = MagicMock(return_value=True) - make_server._decide_download(self.checkout, [], confirm) + make_server.models._decide_download(self.checkout, [], confirm) self.assertIs(confirm.call_args[0][1], True) def test_decide_download_skips_prompt_when_all_models_present(self): @@ -680,7 +683,7 @@ class InstallModelsTests(unittest.TestCase): target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") confirm = MagicMock() - self.assertFalse(make_server._decide_download( + self.assertFalse(make_server.models._decide_download( self.checkout, [{"path": "models/higgs"}], confirm)) confirm.assert_not_called() @@ -689,7 +692,7 @@ class InstallModelsTests(unittest.TestCase): target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") confirm = MagicMock(return_value=True) - self.assertTrue(make_server._decide_download( + self.assertTrue(make_server.models._decide_download( self.checkout, [{"path": "models/higgs"}, {"path": "models/absent"}], confirm)) @@ -699,30 +702,30 @@ class InstallModelsTests(unittest.TestCase): target = self.checkout / "models" / "higgs" target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") - self.assertTrue(make_server._all_models_present( + self.assertTrue(make_server.models._all_models_present( self.checkout, [{"path": "models/higgs"}])) def test_all_models_present_false_when_one_missing(self): target = self.checkout / "models" / "higgs" target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") - self.assertFalse(make_server._all_models_present( + self.assertFalse(make_server.models._all_models_present( self.checkout, [{"path": "models/higgs"}, {"path": "models/absent"}])) def test_all_models_present_false_for_empty_selection(self): - self.assertFalse(make_server._all_models_present(self.checkout, [])) + self.assertFalse(make_server.models._all_models_present(self.checkout, [])) def test_all_models_present_honors_absolute_paths(self): target = self.checkout / "models" / "higgs" target.mkdir(parents=True) (target / "model.gguf").write_bytes(b"x") - self.assertTrue(make_server._all_models_present( + self.assertTrue(make_server.models._all_models_present( self.checkout, [{"path": str(target)}])) def test_all_models_present_false_for_empty_dir(self): (self.checkout / "models" / "higgs").mkdir(parents=True) - self.assertFalse(make_server._all_models_present( + self.assertFalse(make_server.models._all_models_present( self.checkout, [{"path": "models/higgs"}])) @@ -741,31 +744,31 @@ class TranscribeWavDirTests(unittest.TestCase): def test_transcribes_to_stem_map_with_absolute_paths(self): transcripts = {str(self.narrator): "First.", str(self.other): "Second."} - with patch.object(make_server, "transcribe_reference_audio", + with patch.object(make_server.voices, "transcribe_reference_audio", side_effect=lambda path, model_name="base": transcripts[path]): - result = make_server.transcribe_wav_dir( + result = make_server.voices.transcribe_wav_dir( [self.narrator, self.other], "base") self.assertEqual(list(result), ["narrator", "other"]) self.assertEqual(result["narrator"], "First.") def test_failed_transcription_keeps_empty_string(self): - with patch.object(make_server, "transcribe_reference_audio", + with patch.object(make_server.voices, "transcribe_reference_audio", return_value=None): - result = make_server.transcribe_wav_dir([self.narrator], "base") + result = make_server.voices.transcribe_wav_dir([self.narrator], "base") self.assertEqual(result["narrator"], "") def test_whisper_model_name_passed_through(self): - with patch.object(make_server, "transcribe_reference_audio", + with patch.object(make_server.voices, "transcribe_reference_audio", return_value="text") as mock_transcribe: - make_server.transcribe_wav_dir([self.narrator], "large-v3") + make_server.voices.transcribe_wav_dir([self.narrator], "large-v3") self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") def test_write_prompt_text_format(self): - path = make_server.write_prompt_text( + path = common.write_prompt_text( self.folder, {"narrator": "Hello.", "other": "World."}) - self.assertEqual(path, self.folder / make_server.PROMPT_TEXT_FILENAME) + self.assertEqual(path, self.folder / common.PROMPT_TEXT_FILENAME) text = path.read_text(encoding="utf-8") self.assertIn("narrator|Hello.", text) self.assertIn("other|World.", text) @@ -775,21 +778,21 @@ class DesignPackageTests(unittest.TestCase): """Voice-design package detection.""" def test_detects_voicedesign_in_id(self): - self.assertTrue(make_server.is_design_package( + self.assertTrue(make_server.catalog.is_design_package( {"id": "qwen3_tts_1_7b_voicedesign_q8_0"})) def test_detects_voicedesign_in_directory(self): - self.assertTrue(make_server.is_design_package( + self.assertTrue(make_server.catalog.is_design_package( {"target_directory": "Foo-VoiceDesign-GGUF"})) def test_detects_separated_voice_design(self): - self.assertTrue(make_server.is_design_package( + self.assertTrue(make_server.catalog.is_design_package( {"display_name": "Voice Design Q8_0"})) def test_ignores_other_packages(self): - self.assertFalse(make_server.is_design_package( + self.assertFalse(make_server.catalog.is_design_package( {"id": "higgs_audio_tts_4b_q8_0"})) - self.assertFalse(make_server.is_design_package({})) + self.assertFalse(make_server.catalog.is_design_package({})) class PackageDirOptionsTests(unittest.TestCase): @@ -807,7 +810,7 @@ class PackageDirOptionsTests(unittest.TestCase): "target_directory": "VoiceDesign-GGUF"}, ], } - options = make_server.package_dir_options(entry) + options = make_server.catalog.package_dir_options(entry) self.assertEqual([o["target_directory"] for o in options], ["Base-GGUF", "VoiceDesign-GGUF"]) self.assertTrue(options[0]["recommended"]) @@ -826,7 +829,7 @@ class PackageDirOptionsTests(unittest.TestCase): "target_directory": "Default-GGUF"}, ], } - options = make_server.package_dir_options(entry) + options = make_server.catalog.package_dir_options(entry) self.assertEqual([o["target_directory"] for o in options], ["Default-GGUF", "Other-GGUF"]) @@ -848,25 +851,25 @@ class FindAudiocppServerBinTests(unittest.TestCase): (bin_dir / binary).write_bytes(b"x") def test_no_build_dir_returns_none(self): - self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) + self.assertIsNone(make_server.build.find_audiocpp_server_bin(self.checkout)) def test_finds_built_binary(self): self._build("linux-cuda-release") self.assertEqual( - make_server.find_audiocpp_server_bin(self.checkout), + make_server.build.find_audiocpp_server_bin(self.checkout), self.checkout / "build" / "linux-cuda-release" / "bin" / "audiocpp_server") def test_finds_windows_exe(self): self._build("windows-vulkan-debug", binary="audiocpp_server.exe") self.assertEqual( - make_server.find_audiocpp_server_bin(self.checkout).name, + make_server.build.find_audiocpp_server_bin(self.checkout).name, "audiocpp_server.exe") def test_build_dir_without_binary_returns_none(self): (self.checkout / "build" / "linux-cuda-release" / "bin").mkdir( parents=True) - self.assertIsNone(make_server.find_audiocpp_server_bin(self.checkout)) + self.assertIsNone(make_server.build.find_audiocpp_server_bin(self.checkout)) class BuiltServerBinaryTests(unittest.TestCase): @@ -889,24 +892,24 @@ class BuiltServerBinaryTests(unittest.TestCase): self._build("linux-cuda-release") self._build("linux-cpu-release") self.assertEqual( - make_server.built_server_binary(self.checkout, "cpu"), + make_server.build.built_server_binary(self.checkout, "cpu"), self.checkout / "build" / "linux-cpu-release" / "bin" / "audiocpp_server") def test_returns_none_for_unbuilt_backend(self): self._build("linux-cuda-release") self.assertIsNone( - make_server.built_server_binary(self.checkout, "vulkan")) + make_server.build.built_server_binary(self.checkout, "vulkan")) def test_metal_counts_as_cpu(self): self._build("macos-metal-release") self.assertEqual( - make_server.built_server_binary(self.checkout, "cpu"), + make_server.build.built_server_binary(self.checkout, "cpu"), self.checkout / "build" / "macos-metal-release" / "bin" / "audiocpp_server") def test_no_build_dir_returns_none(self): - self.assertIsNone(make_server.built_server_binary(self.checkout, "cpu")) + self.assertIsNone(make_server.build.built_server_binary(self.checkout, "cpu")) class BuildAudiocppTests(unittest.TestCase): @@ -921,7 +924,7 @@ class BuildAudiocppTests(unittest.TestCase): (self.scripts / "build_linux.sh").write_text("#!/bin/sh\n", encoding="utf-8") self.log_dir = Path(self._td.name) / "logs" - self.addCleanup(make_server.common.drain_post_tui_notices) + self.addCleanup(common.drain_post_tui_notices) def tearDown(self): self._td.cleanup() @@ -938,9 +941,9 @@ class BuildAudiocppTests(unittest.TestCase): return sorted(self.log_dir.glob("audiocpp_build_*.log")) def test_runs_build_script_with_backend_and_target(self): - with patch.object(make_server.common, "run_console_subprocess", + with patch.object(common, "run_console_subprocess", return_value=0) as run: - rc = make_server.build_audiocpp(self.checkout, "cuda") + rc = make_server.build.build_audiocpp(self.checkout, "cuda") self.assertEqual(rc, 0) argv = run.call_args[0][0] self.assertEqual(argv[:3], ["sh", str(self.scripts / "build_linux.sh"), @@ -954,24 +957,24 @@ class BuildAudiocppTests(unittest.TestCase): def test_missing_script_returns_nonzero(self): for f in self.scripts.iterdir(): f.unlink() - rc = make_server.build_audiocpp(self.checkout, "cuda") + rc = make_server.build.build_audiocpp(self.checkout, "cuda") self.assertNotEqual(rc, 0) def test_console_path_writes_no_log_and_no_notice(self): - with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ - patch.object(make_server.common, "run_console_subprocess", + with patch.object(common, "LOG_DIR", self.log_dir), \ + patch.object(common, "run_console_subprocess", return_value=0): - rc = make_server.build_audiocpp(self.checkout, "cuda") + rc = make_server.build.build_audiocpp(self.checkout, "cuda") self.assertEqual(rc, 0) self.assertEqual(self._log_files(), []) - self.assertEqual(make_server.common.drain_post_tui_notices(), []) + self.assertEqual(common.drain_post_tui_notices(), []) def test_tui_success_writes_log_and_no_notice(self): emitted, emit = self._emit() - with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ - patch.object(make_server.common, "run_console_subprocess", + with patch.object(common, "LOG_DIR", self.log_dir), \ + patch.object(common, "run_console_subprocess", return_value=0): - rc = make_server.build_audiocpp(self.checkout, "cuda", + rc = make_server.build.build_audiocpp(self.checkout, "cuda", emit=emit) self.assertEqual(rc, 0) self.assertEqual(len(self._log_files()), 1) @@ -979,21 +982,21 @@ class BuildAudiocppTests(unittest.TestCase): self.assertIn("[INFO] Building audiocpp_server", log_text) self.assertIn("--backend cuda", log_text) self.assertTrue(emitted) - self.assertEqual(make_server.common.drain_post_tui_notices(), []) + self.assertEqual(common.drain_post_tui_notices(), []) def test_tui_failure_writes_log_and_records_notice(self): emitted, emit = self._emit() - with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ - patch.object(make_server.common, "run_console_subprocess", + with patch.object(common, "LOG_DIR", self.log_dir), \ + patch.object(common, "run_console_subprocess", return_value=3): - rc = make_server.build_audiocpp(self.checkout, "cuda", + rc = make_server.build.build_audiocpp(self.checkout, "cuda", emit=emit) self.assertEqual(rc, 3) logs = self._log_files() self.assertEqual(len(logs), 1) log_text = logs[0].read_text(encoding="utf-8") self.assertIn("failed (exit code 3)", log_text) - notices = make_server.common.drain_post_tui_notices() + notices = common.drain_post_tui_notices() self.assertEqual(len(notices), 1) notice = notices[0] self.assertIn("failed (exit code 3)", notice) @@ -1010,25 +1013,25 @@ class BuildAudiocppTests(unittest.TestCase): emitted, emit = self._emit() cancel = threading.Event() cancel.set() - with patch.object(make_server.common, "LOG_DIR", self.log_dir), \ - patch.object(make_server.common, "run_console_subprocess", + with patch.object(common, "LOG_DIR", self.log_dir), \ + patch.object(common, "run_console_subprocess", return_value=130): - rc = make_server.build_audiocpp(self.checkout, "cuda", + rc = make_server.build.build_audiocpp(self.checkout, "cuda", emit=emit, cancel=cancel) self.assertEqual(rc, 130) self.assertEqual(len(self._log_files()), 1) - self.assertEqual(make_server.common.drain_post_tui_notices(), []) + self.assertEqual(common.drain_post_tui_notices(), []) def test_tui_missing_script_records_guidance_notice(self): for f in self.scripts.iterdir(): f.unlink() emitted, emit = self._emit() - with patch.object(make_server.common, "LOG_DIR", self.log_dir): - rc = make_server.build_audiocpp(self.checkout, "cuda", + with patch.object(common, "LOG_DIR", self.log_dir): + rc = make_server.build.build_audiocpp(self.checkout, "cuda", emit=emit) self.assertNotEqual(rc, 0) self.assertEqual(self._log_files(), []) - notices = make_server.common.drain_post_tui_notices() + notices = common.drain_post_tui_notices() self.assertEqual(len(notices), 1) self.assertIn("No build script found", notices[0]) @@ -1045,18 +1048,18 @@ class AudiocppDetectTests(unittest.TestCase): self._td.cleanup() def test_not_cloned(self): - with patch.object(make_server, "find_local_checkout", return_value=None): - status = make_server.detect() + with patch.object(make_server.build, "find_local_checkout", return_value=None): + status = make_server.status.detect() self.assertFalse(status.installed) self.assertFalse(status.configured) self.assertIn("not cloned", status.details[0]) def test_cloned_not_built_not_configured(self): - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=self.checkout), \ - patch.object(make_server, "find_audiocpp_server_bin", + patch.object(make_server.build, "find_audiocpp_server_bin", return_value=None): - status = make_server.detect() + status = make_server.status.detect() self.assertFalse(status.installed) self.assertFalse(status.configured) self.assertEqual(status.launch_hint, "") @@ -1069,9 +1072,9 @@ class AudiocppDetectTests(unittest.TestCase): binary.write_bytes(b"x") server_json = self.checkout / "server.json" server_json.write_text('{"models":[]}', encoding="utf-8") - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=self.checkout): - status = make_server.detect() + status = make_server.status.detect() self.assertTrue(status.installed) self.assertTrue(status.configured) self.assertIn(str(binary), status.launch_hint) @@ -1083,9 +1086,9 @@ class AudiocppDetectTests(unittest.TestCase): / "audiocpp_server" binary.parent.mkdir(parents=True) binary.write_bytes(b"x") - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=self.checkout): - status = make_server.detect() + status = make_server.status.detect() self.assertTrue(status.installed) self.assertFalse(status.configured) self.assertEqual(status.partial, "built (not configured)") @@ -1104,11 +1107,11 @@ class NonInteractiveMainTests(unittest.TestCase): # Isolate config.py rewrites so no test touches the real one. self.fake_config = self.root / "config.py" self.fake_config.write_text(FAKE_CONFIG, encoding="utf-8") - patcher = patch.object(make_server, "CONFIG_PATH", self.fake_config) + patcher = patch.object(make_server.configsync, "CONFIG_PATH", self.fake_config) patcher.start() self.addCleanup(patcher.stop) # Tests run without a tty -> main() takes the non-interactive path. - patcher = patch.object(make_server, "_interactive", return_value=False) + patcher = patch.object(make_server.wizard, "_interactive", return_value=False) patcher.start() self.addCleanup(patcher.stop) @@ -1121,14 +1124,14 @@ class NonInteractiveMainTests(unittest.TestCase): transcribe_effect = transcribe if transcribe is not None \ else MagicMock() with patch.object(sys, "argv", argv), \ - patch.object(make_server, "find_local_checkout", + patch.object(make_server.build, "find_local_checkout", return_value=None if no_checkout else self.checkout), \ - patch.object(make_server, "transcribe_reference_audio", + patch.object(make_server.voices, "transcribe_reference_audio", side_effect=transcribe_effect), \ - patch.object(make_server, "whisper_backend_available", + patch.object(make_server.voices, "whisper_backend_available", return_value=whisper): - return make_server.main() + return make_server.wizard.main() def _args(self, *extra): return ["--wavs", str(self.folder), "--output", str(self.output)] \ @@ -1140,7 +1143,7 @@ class NonInteractiveMainTests(unittest.TestCase): self.assertEqual(exit_code, 0) data = json.loads(self.output.read_text(encoding="utf-8")) self.assertEqual(data["host"], "127.0.0.1") - self.assertEqual(data["port"], make_server.config_port()) + self.assertEqual(data["port"], make_server.configsync.config_port()) self.assertEqual(data["backend"], "cuda") self.assertTrue(data["lazy_load"]) self.assertEqual([m["id"] for m in data["models"]], @@ -1192,7 +1195,7 @@ class NonInteractiveMainTests(unittest.TestCase): "Higgs-Audio-v3-TTS-4B-GGUF"]) self.assertTrue(data["lazy_load"]) self.assertEqual(data["voice_dir"], str(self.folder.resolve())) - prompt = (self.folder / make_server.PROMPT_TEXT_FILENAME).read_text( + prompt = (self.folder / common.PROMPT_TEXT_FILENAME).read_text( encoding="utf-8") self.assertIn("narrator|a transcript", prompt) @@ -1282,8 +1285,8 @@ class FetchServerEndpointsTests(unittest.TestCase): "data": [{"id": "qwen", "family": "qwen3_tts", "task": "tts"}, {"id": "legacy"}], }).encode("utf-8")]) - with patch.object(make_server.urllib.request, "urlopen", urlopen): - models = make_server.fetch_server_models("http://127.0.0.1:8080") + with patch("urllib.request.urlopen", urlopen): + models = make_server.remote.fetch_server_models("http://127.0.0.1:8080") # Missing fields mirror the converter's client: empty strings. self.assertEqual(models, [ {"id": "qwen", "family": "qwen3_tts", "task": "tts"}, @@ -1294,36 +1297,36 @@ class FetchServerEndpointsTests(unittest.TestCase): def test_fetch_models_trailing_slash_url(self): urlopen, calls = self._urlopen_responding( [b'{"data": [{"id": "m"}]}']) - with patch.object(make_server.urllib.request, "urlopen", urlopen): - make_server.fetch_server_models("http://host:8080/") + with patch("urllib.request.urlopen", urlopen): + make_server.remote.fetch_server_models("http://host:8080/") self.assertEqual(calls, ["http://host:8080/v1/models"]) def test_fetch_models_connection_error_returns_none(self): import urllib.error urlopen, _ = self._urlopen_responding( [], errors=[urllib.error.URLError("Connection refused")]) - with patch.object(make_server.urllib.request, "urlopen", urlopen): + with patch("urllib.request.urlopen", urlopen): self.assertIsNone( - make_server.fetch_server_models("http://127.0.0.1:8080")) + make_server.remote.fetch_server_models("http://127.0.0.1:8080")) def test_fetch_models_non_json_body_returns_none(self): # A port answering TCP but not speaking audiocpp_server JSON. urlopen, _ = self._urlopen_responding([b"<html>not json</html>"]) - with patch.object(make_server.urllib.request, "urlopen", urlopen): + with patch("urllib.request.urlopen", urlopen): self.assertIsNone( - make_server.fetch_server_models("http://127.0.0.1:8080")) + make_server.remote.fetch_server_models("http://127.0.0.1:8080")) def test_fetch_models_unexpected_document_yields_empty_list(self): urlopen, _ = self._urlopen_responding([b'{"foo": 1}']) - with patch.object(make_server.urllib.request, "urlopen", urlopen): + with patch("urllib.request.urlopen", urlopen): self.assertEqual( - make_server.fetch_server_models("http://127.0.0.1:8080"), []) + make_server.remote.fetch_server_models("http://127.0.0.1:8080"), []) def test_fetch_voices_parses_names_and_encodes_model(self): urlopen, calls = self._urlopen_responding( [b'{"voices": ["narrator", "obama"]}']) - with patch.object(make_server.urllib.request, "urlopen", urlopen): - voices = make_server.fetch_server_voices( + with patch("urllib.request.urlopen", urlopen): + voices = make_server.remote.fetch_server_voices( "http://127.0.0.1:8080", "qwen") self.assertEqual(voices, ["narrator", "obama"]) self.assertEqual(calls, @@ -1333,15 +1336,15 @@ class FetchServerEndpointsTests(unittest.TestCase): import urllib.error urlopen, _ = self._urlopen_responding( [], errors=[urllib.error.URLError("boom")]) - with patch.object(make_server.urllib.request, "urlopen", urlopen): + with patch("urllib.request.urlopen", urlopen): self.assertIsNone( - make_server.fetch_server_voices("http://h", "qwen")) + make_server.remote.fetch_server_voices("http://h", "qwen")) def test_fetch_voices_non_list_shape_returns_none(self): urlopen, _ = self._urlopen_responding([b'{"voices": 5}']) - with patch.object(make_server.urllib.request, "urlopen", urlopen): + with patch("urllib.request.urlopen", urlopen): self.assertIsNone( - make_server.fetch_server_voices("http://h", "qwen")) + make_server.remote.fetch_server_voices("http://h", "qwen")) class MissingModelEntriesTests(unittest.TestCase): @@ -1366,29 +1369,29 @@ class MissingModelEntriesTests(unittest.TestCase): {"id": "a", "path": "models/present"}, {"id": "b", "path": "models/absent"}, ]) - missing = make_server.missing_model_entries(path) + missing = make_server.models.missing_model_entries(path) self.assertEqual([m["id"] for m in missing], ["b"]) def test_empty_directory_counts_as_missing(self): (self.dir / "models" / "empty").mkdir(parents=True) path = self._server_json([{"id": "a", "path": "models/empty"}]) - self.assertEqual(len(make_server.missing_model_entries(path)), 1) + self.assertEqual(len(make_server.models.missing_model_entries(path)), 1) def test_absolute_paths_honored(self): target = self.dir / "absolute" target.mkdir() (target / "m.gguf").write_bytes(b"x") path = self._server_json([{"id": "a", "path": str(target)}]) - self.assertEqual(make_server.missing_model_entries(path), []) + self.assertEqual(make_server.models.missing_model_entries(path), []) def test_unreadable_json_returns_empty(self): path = self.dir / "server.json" path.write_text("not json", encoding="utf-8") - self.assertEqual(make_server.missing_model_entries(path), []) + self.assertEqual(make_server.models.missing_model_entries(path), []) def test_no_models_returns_empty(self): path = self._server_json([]) - self.assertEqual(make_server.missing_model_entries(path), []) + self.assertEqual(make_server.models.missing_model_entries(path), []) class ModelInstallHintsTests(unittest.TestCase): @@ -1409,7 +1412,7 @@ class ModelInstallHintsTests(unittest.TestCase): }], }), encoding="utf-8") missing = [{"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}] - hints = make_server.model_install_hints(checkout, missing) + hints = make_server.models.model_install_hints(checkout, missing) self.assertEqual(len(hints), 1) self.assertIn("qwen3_tts_0_6b_base_q8_0", hints[0]) @@ -1418,7 +1421,7 @@ class ModelInstallHintsTests(unittest.TestCase): with tempfile.TemporaryDirectory() as td: checkout = Path(td) (checkout / "model_specs").mkdir() - hints = make_server.model_install_hints( + hints = make_server.models.model_install_hints( checkout, [{"id": "x", "rel": "models/nope"}]) self.assertIn("models/nope", hints[0]) self.assertNotIn("install", hints[0]) @@ -1443,11 +1446,11 @@ class DetectServerSpecTests(unittest.TestCase): def test_spec_has_cwd_and_identity(self): checkout = self._checkout() - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ - patch.object(make_server, "_detect_remote", + patch.object(make_server.status, "_detect_remote", return_value=(False, {})): - status = make_server.detect() + status = make_server.status.detect() self.assertEqual(len(status.servers), 1) spec = status.servers[0] self.assertEqual(spec.cwd, checkout) @@ -1456,11 +1459,11 @@ class DetectServerSpecTests(unittest.TestCase): def test_models_missing_flag_and_details(self): checkout = self._checkout() - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ - patch.object(make_server, "_detect_remote", + patch.object(make_server.status, "_detect_remote", return_value=(False, {})): - status = make_server.detect() + status = make_server.status.detect() self.assertTrue(status.models_missing) self.assertTrue(any("not downloaded" in line for line in status.details)) @@ -1488,13 +1491,13 @@ class InstalledModelEntriesTests(unittest.TestCase): {"id": "a", "path": "models/present"}, {"id": "b", "path": "models/absent"}, ]) - installed = make_server.installed_model_entries(path) + installed = make_server.models.installed_model_entries(path) self.assertEqual([m["id"] for m in installed], ["a"]) def test_unreadable_json_returns_empty(self): path = self.dir / "server.json" path.write_text("not json", encoding="utf-8") - self.assertEqual(make_server.installed_model_entries(path), []) + self.assertEqual(make_server.models.installed_model_entries(path), []) class MissingModelInstallGuidanceTests(unittest.TestCase): @@ -1517,7 +1520,7 @@ class MissingModelInstallGuidanceTests(unittest.TestCase): {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}, {"id": "x", "rel": "models/nope"}, ] - guidance = make_server.missing_model_install_guidance( + guidance = make_server.models.missing_model_install_guidance( checkout, missing) self.assertEqual(guidance, [("qwen", "qwen3_tts_0_6b_base_q8_0")]) @@ -1537,22 +1540,22 @@ class LoadServerConfigTests(unittest.TestCase): path = self.dir / "server.json" path.write_text(json.dumps({"host": "0.0.0.0", "models": []}), encoding="utf-8") - self.assertEqual(make_server.load_server_config(path), + self.assertEqual(make_server.catalog.load_server_config(path), {"host": "0.0.0.0", "models": []}) def test_missing_file_returns_none(self): - self.assertIsNone(make_server.load_server_config( + self.assertIsNone(make_server.catalog.load_server_config( self.dir / "nope.json")) def test_unreadable_json_returns_none(self): path = self.dir / "server.json" path.write_text("not json", encoding="utf-8") - self.assertIsNone(make_server.load_server_config(path)) + self.assertIsNone(make_server.catalog.load_server_config(path)) def test_non_dict_document_returns_none(self): path = self.dir / "server.json" path.write_text("[1, 2, 3]", encoding="utf-8") - self.assertIsNone(make_server.load_server_config(path)) + self.assertIsNone(make_server.catalog.load_server_config(path)) class ServerConfigSelectionsTests(unittest.TestCase): @@ -1561,7 +1564,7 @@ class ServerConfigSelectionsTests(unittest.TestCase): def setUp(self): self._tmp = tempfile.TemporaryDirectory() self.checkout = _make_checkout(Path(self._tmp.name)) - self.catalog = make_server.load_model_catalog(self.checkout) + self.catalog = make_server.catalog.load_model_catalog(self.checkout) def tearDown(self): self._tmp.cleanup() @@ -1576,7 +1579,7 @@ class ServerConfigSelectionsTests(unittest.TestCase): {"id": "higgs", "family": "higgs_audio_tts", "path": "models/Higgs-Audio-v3-TTS-4B-GGUF", "task": "tts"}, ]} - selected, tasks = make_server.server_config_selections(config, + selected, tasks = make_server.catalog.server_config_selections(config, self.catalog) self.assertEqual(selected["qwen3_tts"], ["Qwen3-TTS-12Hz-1.7B-Base-GGUF", @@ -1593,7 +1596,7 @@ class ServerConfigSelectionsTests(unittest.TestCase): config = {"models": [ {"id": "x", "family": "not_a_family", "path": "models/x"}, ]} - selected, tasks = make_server.server_config_selections(config, + selected, tasks = make_server.catalog.server_config_selections(config, self.catalog) self.assertEqual(selected, {}) self.assertEqual(tasks, {}) @@ -1603,13 +1606,13 @@ class ServerConfigSelectionsTests(unittest.TestCase): {"id": "qwen", "family": "qwen3_tts", "path": "/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"}, ]} - selected, tasks = make_server.server_config_selections(config, + selected, tasks = make_server.catalog.server_config_selections(config, self.catalog) self.assertEqual(selected["qwen3_tts"], ["/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF"]) def test_empty_models_yield_empty_selections(self): - selected, tasks = make_server.server_config_selections({"models": []}, + selected, tasks = make_server.catalog.server_config_selections({"models": []}, self.catalog) self.assertEqual(selected, {}) self.assertEqual(tasks, {}) @@ -1641,7 +1644,7 @@ class UnusedInstalledEntriesTests(unittest.TestCase): {"id": "dropped", "path": "models/dropped"}, {"id": "missing", "path": "models/missing"}, ]) - unused = make_server.unused_installed_entries( + unused = make_server.models.unused_installed_entries( path, {"models/kept"}) self.assertEqual([entry["id"] for entry in unused], ["dropped"]) @@ -1649,7 +1652,7 @@ class UnusedInstalledEntriesTests(unittest.TestCase): path = self._server_json([ {"id": "kept", "path": "models/kept"}, ]) - unused = make_server.unused_installed_entries( + unused = make_server.models.unused_installed_entries( path, {"models/kept"}) self.assertEqual(unused, []) @@ -1678,7 +1681,7 @@ class DeleteModelFilesTests(unittest.TestCase): self._tmp.cleanup() def test_removes_dirs_and_counts(self): - removed = make_server.delete_model_files( + removed = make_server.models.delete_model_files( self.server_json, [{"id": "a", "rel": "models/a"}, {"id": "b", "rel": "models/b"}]) self.assertEqual(removed, 2) @@ -1687,14 +1690,14 @@ class DeleteModelFilesTests(unittest.TestCase): self.assertTrue((self.dir / "models" / "c").exists()) def test_missing_paths_ignored(self): - removed = make_server.delete_model_files( + removed = make_server.models.delete_model_files( self.server_json, [{"id": "ghost", "rel": "models/ghost"}]) self.assertEqual(removed, 0) def test_removes_single_file(self): file_path = self.dir / "models" / "single.gguf" file_path.write_bytes(b"x") - removed = make_server.delete_model_files( + removed = make_server.models.delete_model_files( self.server_json, [{"id": "s", "rel": "models/single.gguf"}]) self.assertEqual(removed, 1) self.assertFalse(file_path.exists()) @@ -1703,7 +1706,7 @@ class DeleteModelFilesTests(unittest.TestCase): target = self.dir / "absolute" target.mkdir() (target / "m.gguf").write_bytes(b"x") - removed = make_server.delete_model_files( + removed = make_server.models.delete_model_files( self.server_json, [{"id": "a", "rel": str(target)}]) self.assertEqual(removed, 1) self.assertFalse(target.exists()) @@ -1716,8 +1719,8 @@ class InstallModelsApiTests(unittest.TestCase): with tempfile.TemporaryDirectory() as td: checkout = Path(td) guidance = [("qwen", "qwen3_tts_0_6b_base_q8_0")] - with patch.object(make_server, "_install_models") as mk: - make_server.install_models(checkout, guidance) + with patch.object(make_server.models, "_install_models") as mk: + make_server.models.install_models(checkout, guidance) mk.assert_called_once_with(checkout, guidance, download=True, emit=None, cancel=None) @@ -1728,7 +1731,7 @@ class HandInstallGuidanceTests(unittest.TestCase): def test_lists_each_model_and_its_path(self): with tempfile.TemporaryDirectory() as td: checkout = Path(td) - message = make_server.hand_install_guidance(checkout, [ + message = make_server.models.hand_install_guidance(checkout, [ {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}, {"id": "higgs", "rel": "models/Higgs-Audio-4B-GGUF"}, ]) @@ -1743,7 +1746,7 @@ class WizardNavigationTests(unittest.TestCase): """Esc in the audio.cpp wizard goes back one screen (via tui.Wizard).""" def _args(self): - return make_server.build_parser().parse_args([]) + return make_server.wizard.build_parser().parse_args([]) def _checkout(self): tmp = tempfile.TemporaryDirectory() @@ -1754,12 +1757,12 @@ class WizardNavigationTests(unittest.TestCase): # Configure audio.cpp (modify flow): the families tree is the first # screen, so Esc on it must abort the wizard — not re-show itself. checkout = self._checkout() - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", - return_value=make_server._GO_BACK): - settings = make_server._wizard(None, self._args(), - make_server.build_parser()) + return_value=make_server.wizard._GO_BACK): + settings = make_server.wizard._wizard(None, self._args(), + make_server.wizard.build_parser()) self.assertIsNone(settings) def test_modify_flow_offers_build_when_not_built(self): @@ -1771,7 +1774,7 @@ class WizardNavigationTests(unittest.TestCase): (checkout / "server.json").write_text( json.dumps({"models": [], "backend": "vulkan"}), encoding="utf-8") - catalog = make_server.load_model_catalog(checkout) + catalog = make_server.catalog.load_model_catalog(checkout) supertonic = next(i for i, entry in enumerate(catalog) if entry["family"] == "supertonic") confirm_questions = [] @@ -1793,14 +1796,14 @@ class WizardNavigationTests(unittest.TestCase): confirm_questions.append(question) return False # decline the build - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ patch.object(tui, "line_edit", side_effect=fake_line_edit), \ patch.object(tui, "menu", side_effect=fake_menu), \ patch.object(tui, "confirm", side_effect=fake_confirm): - settings = make_server._wizard(None, self._args(), - make_server.build_parser()) + settings = make_server.wizard._wizard(None, self._args(), + make_server.wizard.build_parser()) self.assertIsNotNone(settings) self.assertEqual(settings["backend"], "vulkan") self.assertFalse(settings["build"]) @@ -1813,11 +1816,11 @@ class WizardNavigationTests(unittest.TestCase): # Esc on "Bind host" must fall back to the model-family tree, then # re-selecting proceeds through the rest of the wizard. checkout = self._checkout() - catalog = make_server.load_model_catalog(checkout) + catalog = make_server.catalog.load_model_catalog(checkout) supertonic = next(i for i, entry in enumerate(catalog) if entry["family"] == "supertonic") tree_calls = [] - hosts = iter([make_server._GO_BACK, "127.0.0.1"]) + hosts = iter([make_server.wizard._GO_BACK, "127.0.0.1"]) def fake_tree(*args, **kwargs): tree_calls.append(1) @@ -1830,7 +1833,7 @@ class WizardNavigationTests(unittest.TestCase): return "8080" return default - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ patch.object(tui, "checkbox_tree", side_effect=fake_tree), \ @@ -1838,8 +1841,8 @@ class WizardNavigationTests(unittest.TestCase): side_effect=fake_line_edit), \ patch.object(tui, "menu", return_value="cuda"), \ patch.object(tui, "confirm", return_value=True): - settings = make_server._wizard(None, self._args(), - make_server.build_parser()) + settings = make_server.wizard._wizard(None, self._args(), + make_server.wizard.build_parser()) self.assertIsNotNone(settings) # The tree was re-shown after the host screen's Esc. self.assertEqual(len(tree_calls), 2) @@ -1855,12 +1858,12 @@ class UninstallTests(unittest.TestCase): with tempfile.TemporaryDirectory() as td: checkout = Path(td) / "audio.cpp" checkout.mkdir() - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ - patch.object(make_server.servers, "pid_for", + patch.object(servers, "pid_for", return_value=1234), \ - patch.object(make_server.servers, "stop") as mk_stop: - rc = make_server.uninstall() + patch.object(servers, "stop") as mk_stop: + rc = make_server.build.uninstall() self.assertEqual(rc, 0) self.assertFalse(checkout.exists()) mk_stop.assert_called_once_with("audiocpp") @@ -1868,12 +1871,12 @@ class UninstallTests(unittest.TestCase): def test_skips_stop_without_a_pid_file(self): # No pid file: the server was never started by this tool, so # stop (and its "stop it manually" noise) is skipped. - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=None), \ - patch.object(make_server.servers, "pid_for", + patch.object(servers, "pid_for", return_value=None), \ - patch.object(make_server.servers, "stop") as mk_stop: - rc = make_server.uninstall() + patch.object(servers, "stop") as mk_stop: + rc = make_server.build.uninstall() self.assertEqual(rc, 0) mk_stop.assert_not_called() @@ -1883,12 +1886,12 @@ class UninstallTests(unittest.TestCase): with tempfile.TemporaryDirectory() as td: checkout = Path(td) / "audio.cpp" checkout.mkdir() - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ - patch.object(make_server.servers, "pid_for", + patch.object(servers, "pid_for", return_value=1234), \ - patch.object(make_server.servers, "stop"): - rc = make_server.uninstall(emit=lambda line: None, + patch.object(servers, "stop"): + rc = make_server.build.uninstall(emit=lambda line: None, cancel=None) self.assertEqual(rc, 0) self.assertFalse(checkout.exists()) @@ -1901,12 +1904,12 @@ class UninstallTests(unittest.TestCase): checkout.mkdir() cancel = threading.Event() cancel.set() - with patch.object(make_server, "find_local_checkout", + with patch.object(make_server.build, "find_local_checkout", return_value=checkout), \ - patch.object(make_server.servers, "pid_for", + patch.object(servers, "pid_for", return_value=1234), \ - patch.object(make_server.servers, "stop"): - rc = make_server.uninstall(cancel=cancel) + patch.object(servers, "stop"): + rc = make_server.build.uninstall(cancel=cancel) self.assertEqual(rc, 130) self.assertTrue(checkout.exists()) @@ -1920,23 +1923,23 @@ class SetupScreenTests(unittest.TestCase): in-TUI task view (two parallel lanes on a fresh install).""" def test_abort_returns_one_without_executing(self): - with patch.object(make_server, "_wizard", return_value=None) as mk_wizard, \ - patch.object(make_server, "_execute_lanes") as mk_lanes: - rc = make_server.setup_screen(None) + with patch.object(make_server.wizard, "_wizard", return_value=None) as mk_wizard, \ + patch.object(make_server.wizard, "_execute_lanes") as mk_lanes: + rc = make_server.wizard.setup_screen(None) self.assertEqual(rc, 1) mk_wizard.assert_called_once() mk_lanes.assert_not_called() def test_success_runs_the_tail_in_the_task_view(self): settings = {"audiocpp_dir": Path("/x")} - lanes = [make_server.taskview.TaskLane( - "Build", [make_server.taskview.TaskStep("t", lambda emit, cancel: 0)])] - with patch.object(make_server, "_wizard", return_value=settings), \ - patch.object(make_server, "_execute_lanes", + lanes = [taskview.TaskLane( + "Build", [taskview.TaskStep("t", lambda emit, cancel: 0)])] + with patch.object(make_server.wizard, "_wizard", return_value=settings), \ + patch.object(make_server.wizard, "_execute_lanes", return_value=lanes) as mk_lanes, \ - patch.object(make_server.taskview, "run_lanes", + patch.object(taskview, "run_lanes", return_value=0) as mk_run: - rc = make_server.setup_screen(None) + rc = make_server.wizard.setup_screen(None) self.assertEqual(rc, 0) mk_lanes.assert_called_once() self.assertIs(mk_lanes.call_args[0][0], settings) @@ -1973,8 +1976,8 @@ class ExecuteLanesTests(unittest.TestCase): return settings def test_two_lanes_when_building(self): - args = make_server.build_parser().parse_args([]) - lanes = make_server._execute_lanes(self._settings(), args) + args = make_server.wizard.build_parser().parse_args([]) + lanes = make_server.wizard._execute_lanes(self._settings(), args) self.assertEqual([lane.title for lane in lanes], ["Build", "Configure & download"]) self.assertEqual([s.title for s in lanes[0].steps], @@ -1985,15 +1988,15 @@ class ExecuteLanesTests(unittest.TestCase): "Download models"]) def test_single_lane_when_not_building(self): - args = make_server.build_parser().parse_args([]) - lanes = make_server._execute_lanes( + args = make_server.wizard.build_parser().parse_args([]) + lanes = make_server.wizard._execute_lanes( self._settings(build=False), args) self.assertEqual([lane.title for lane in lanes], ["Configure & download"]) def test_flattened_console_steps_keep_the_build_first(self): - args = make_server.build_parser().parse_args([]) - steps = make_server._execute_steps(self._settings(), args) + args = make_server.wizard.build_parser().parse_args([]) + steps = make_server.wizard._execute_steps(self._settings(), args) self.assertEqual([s.title for s in steps], ["Build audiocpp_server (cuda)", "Transcribe reference voices", @@ -2001,11 +2004,11 @@ class ExecuteLanesTests(unittest.TestCase): "Download models"]) def test_download_step_prints_the_launch_hint(self): - args = make_server.build_parser().parse_args([]) - lanes = make_server._execute_lanes(self._settings(), args) + args = make_server.wizard.build_parser().parse_args([]) + lanes = make_server.wizard._execute_lanes(self._settings(), args) install_step = lanes[1].steps[2] - with patch.object(make_server, "_install_models"), \ - patch.object(make_server, "_print_launch_hint") as mk_hint: + with patch.object(make_server.models, "_install_models"), \ + patch.object(make_server.build, "_print_launch_hint") as mk_hint: install_step.work(lambda line: None, threading.Event()) mk_hint.assert_called_once_with(Path("/x"), Path("/x/server.json")) @@ -2016,9 +2019,9 @@ class LaunchHintTests(unittest.TestCase): def _capture(self, audiocpp_dir, output_path, binary=None): buf = io.StringIO() with redirect_stdout(buf), \ - patch.object(make_server, "find_audiocpp_server_bin", + patch.object(make_server.build, "find_audiocpp_server_bin", return_value=binary): - make_server._print_launch_hint(audiocpp_dir, output_path) + make_server.build._print_launch_hint(audiocpp_dir, output_path) return buf.getvalue() def test_built_server_prints_nothing(self): diff --git a/app/tests/test_backends_faster.py b/app/tests/test_backends_faster.py index 55d617c..800acb0 100644 --- a/app/tests/test_backends_faster.py +++ b/app/tests/test_backends_faster.py @@ -173,7 +173,8 @@ class MainTests(unittest.TestCase): return_value=False) patcher.start() self.addCleanup(patcher.stop) - patcher = patch.object(make_voices, "_interactive", return_value=False) + patcher = patch.object(make_voices.setup, "interactive", + return_value=False) patcher.start() self.addCleanup(patcher.stop) diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index 9f72ad9..d8a21bc 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -1972,7 +1972,7 @@ class AudiocppServerConfigTests(unittest.TestCase): import json with tempfile.TemporaryDirectory() as td: mod, checkout, server_json = self._make_checkout(td, port=8080) - with patch.object(mod, "find_local_checkout", + with patch.object(mod.build, "find_local_checkout", return_value=checkout): self.assertTrue(mod.update_server_config_port(9090)) data = json.loads(server_json.read_text(encoding="utf-8")) @@ -1986,16 +1986,17 @@ class AudiocppServerConfigTests(unittest.TestCase): with tempfile.TemporaryDirectory() as td: mod, checkout, server_json = self._make_checkout(td, port=8080) before = server_json.read_text(encoding="utf-8") - with patch.object(mod, "find_local_checkout", + with patch.object(mod.build, "find_local_checkout", return_value=checkout): self.assertTrue(mod.update_server_config_port(8080)) self.assertEqual(server_json.read_text(encoding="utf-8"), before) def test_false_when_no_checkout(self): from backends import audiocpp as audiocpp_backend - with patch.object(audiocpp_backend, "find_local_checkout", + with patch.object(audiocpp_backend.build, "find_local_checkout", return_value=None): - self.assertFalse(audiocpp_backend.update_server_config_port(9090)) + self.assertFalse( + audiocpp_backend.update_server_config_port(9090)) class ConfigureBackendsDispatchTests(unittest.TestCase): |