diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/backends/__init__.py | 32 | ||||
| -rwxr-xr-x | app/backends/audiocpp.py | 465 | ||||
| -rw-r--r-- | app/backends/common.py | 10 | ||||
| -rw-r--r-- | app/backends/envs.py | 14 | ||||
| -rwxr-xr-x | app/backends/faster.py | 161 | ||||
| -rw-r--r-- | app/backends/qwen.py | 21 | ||||
| -rw-r--r-- | app/docs/backend-audiocpp.md | 2 | ||||
| -rw-r--r-- | app/docs/backend-faster.md | 2 | ||||
| -rw-r--r-- | app/docs/backend-qwen.md | 2 | ||||
| -rw-r--r-- | app/tests/test_backends.py | 6 | ||||
| -rw-r--r-- | app/tests/test_backends_audiocpp.py | 296 | ||||
| -rw-r--r-- | app/tests/test_backends_faster.py | 77 | ||||
| -rw-r--r-- | app/tests/test_hub.py | 270 | ||||
| -rw-r--r-- | app/tests/test_tui.py | 31 | ||||
| -rw-r--r-- | app/ui/hub.py | 174 | ||||
| -rw-r--r-- | app/ui/tui.py | 29 |
16 files changed, 1342 insertions, 250 deletions
diff --git a/app/backends/__init__.py b/app/backends/__init__.py index 488c36e..3dff306 100644 --- a/app/backends/__init__.py +++ b/app/backends/__init__.py @@ -1,12 +1,12 @@ """Registry of the TTS backends the audiobook generator can talk to. Each backend (audio.cpp, qwen, faster) lives in its own module and owns -its setup wizard, its status detection, and the launch command it prints -once configured. This package aggregates them into a single registry so -``audiobook.py``'s TUI hub and future tools can iterate backends without -hardcoding their names: ``backends.detect_all()`` reports which are set +its setup wizard, its status detection, its uninstaller, and the launch +command it prints once configured. This package aggregates them into a single +registry so ``audiobook.py``'s TUI hub and future tools can iterate backends +without hardcoding their names: ``backends.detect_all()`` reports which are set up (and whether their server is currently running), and the registry -drives the hub's setup/configure menus. +drives the hub's "Configure backends" menu. The registry is built lazily on the first call to ``get``/``detect_all``/ ``detect`` (not at package import time), because the backend modules pull @@ -17,9 +17,8 @@ importing this package must stay cheap and dependency-free. Adding a backend: create ``backends/<name>.py`` exposing ``detect() -> BackendStatus``, ``run_tui() -> int`` and -``configure_actions: list[ConfigureAction]``, then append a ``BackendInfo`` in -``_build_registry`` below. ``audiobook.py`` and the hub pick it up -automatically. +``uninstall() -> int``, then append a ``BackendInfo`` in ``_build_registry`` +below. ``audiobook.py`` and the hub pick it up automatically. """ import shlex @@ -131,20 +130,13 @@ def format_launch_hint(servers: List[ServerSpec]) -> str: @dataclass -class ConfigureAction: - """A per-backend "configure" menu entry (e.g. "New server.json").""" - label: str - run: Callable[[], int] - - -@dataclass class BackendInfo: - """One registry entry: identity, detector, setup wizard, configure menu.""" + """One registry entry: identity, detector, setup wizard, uninstaller.""" key: str label: str detect: Callable[[], BackendStatus] setup_tui: Callable[[], int] - configure_actions: List[ConfigureAction] = field(default_factory=list) + uninstall: Callable[[], int] = lambda: 0 REGISTRY: List[BackendInfo] = [] @@ -162,21 +154,21 @@ def _build_registry() -> None: label="audio.cpp", detect=audiocpp.detect, setup_tui=audiocpp.run_tui, - configure_actions=audiocpp.configure_actions, + uninstall=audiocpp.uninstall, )) REGISTRY.append(BackendInfo( key="qwen", label="qwen-tts", detect=qwen.detect, setup_tui=qwen.run_tui, - configure_actions=qwen.configure_actions, + uninstall=qwen.uninstall, )) REGISTRY.append(BackendInfo( key="faster", label="faster-qwen3-tts", detect=faster.detect, setup_tui=faster.run_tui, - configure_actions=faster.configure_actions, + uninstall=faster.uninstall, )) for info in REGISTRY: _BY_KEY[info.key] = info diff --git a/app/backends/audiocpp.py b/app/backends/audiocpp.py index a502c8f..f74f726 100755 --- a/app/backends/audiocpp.py +++ b/app/backends/audiocpp.py @@ -26,12 +26,20 @@ Usage: 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, lazy-load +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 json import os import re +import shutil import subprocess import sys import urllib.parse @@ -44,7 +52,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from backends import ( BackendStatus, - ConfigureAction, ServerSpec, common, format_launch_hint, @@ -721,14 +728,18 @@ def _offer_config_model_id_sync(model_id: str, accepted: Optional[bool]) -> None def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], catalog_by_family: Dict[str, dict], task_picker: Callable[[str], str], - id_picker: Callable[[str, str, str], str] + id_picker: Callable[[str, str, 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; - ID_PICKER resolves a duplicate server entry id. Returns (model_entries, - entry_ids, install_guidance, design_entry_ids, include_clone). + ID_PICKER resolves a duplicate server entry id. 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. Returns (model_entries, entry_ids, + install_guidance, design_entry_ids, include_clone). """ model_entries: List[dict] = [] entry_ids: List[str] = [] @@ -739,7 +750,13 @@ def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], entry = catalog_by_family[family] include_clone = include_clone or entry["clone_capable"] for opt in chosen[family]: - task = task_picker(opt["install_id"]) if opt["design"] else TASK_TTS + 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 = (f"{entry['preferred_id']}-design" if task == TASK_VDES else entry["preferred_id"]) model_id = base_id @@ -861,8 +878,6 @@ def _build_tree_families(catalog: List[dict]) -> List[dict]: if "design" in entry["tasks"]: capabilities.append("design") name = entry["display_name"] - if name != entry["family"]: - name = f"{name} ({entry['family']})" options = [] for opt in package_dir_options(entry): options.append({ @@ -899,21 +914,15 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser step = 0 while True: if step == 0: - # Checkout browser + the output path/overwrite confirmation. The - # browser asks for the checkout root and finds model_specs/ inside - # it (picking the model_specs directory itself works too — its - # parent is used). A highlighted subdirectory named "audio.cpp" - # that already contains model_specs/ is auto-accepted on - # Enter/Right, skipping the "[ Use this directory ]" step. - # Pressing Esc on an overwrite confirmation returns here instead - # of aborting: the browser then restarts inside the previously - # accepted checkout with auto-accept disabled, so a wrong guess - # can be corrected. An explicit --audiocpp-dir flag has no - # browser to return to, so Esc still aborts there. Esc on the - # browser itself is the first step, so it aborts the wizard. + # Checkout browser. The browser asks for the checkout root and + # finds model_specs/ inside it (picking the model_specs directory + # itself works too — its parent is used). A highlighted + # subdirectory named "audio.cpp" that already contains + # model_specs/ is auto-accepted on Enter/Right, skipping the + # "[ Use this directory ]" step. Esc on the browser is the first + # step, so it aborts the wizard. auto_accept = True browser_start: Path = Path.cwd() - force_browse = False def do_browse(): return tui.browse_directory( @@ -932,44 +941,36 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser while True: audiocpp_dir = args.audiocpp_dir - if audiocpp_dir is None and not force_browse: + if audiocpp_dir is None: audiocpp_dir = find_local_checkout() - if force_browse: - audiocpp_dir = None if audiocpp_dir is None: - if force_browse: - # Esc on an overwrite confirmation came back here: go - # straight back into the browser inside the previously - # accepted checkout (auto-accept disabled). - audiocpp_dir = do_browse() + # No checkout found anywhere: offer to clone one into + # ./app/audio.cpp or browse for an existing checkout. + # Esc on this first menu aborts the wizard. + choice = tui.menu( + stdscr, "No audio.cpp checkout found", + [(f"Clone into ./app/{AUDIOCPP_DIR_NAME} " + f"(from {AUDIOCPP_GIT_URL})", "clone"), + ("Browse for an existing checkout", "browse")], + help_lines=[ + "audio.cpp hosts the TTS model families " + "this generator uses.", + "Clone it into the project's app " + "directory, or point at an existing " + "checkout."]) + if choice == "clone": + target = APP_DIR / AUDIOCPP_DIR_NAME + with tui.suspend(stdscr): + rc = common.git_clone(AUDIOCPP_GIT_URL, + target) + if rc != 0: + raise _TuiError( + f"git clone failed (exit {rc}). Clone " + f"audio.cpp manually: git clone " + f"{AUDIOCPP_GIT_URL} {target}") + audiocpp_dir = target else: - # No checkout found anywhere: offer to clone one into - # ./app/audio.cpp or browse for an existing checkout. - # Esc on this first menu aborts the wizard. - choice = tui.menu( - stdscr, "No audio.cpp checkout found", - [(f"Clone into ./app/{AUDIOCPP_DIR_NAME} " - f"(from {AUDIOCPP_GIT_URL})", "clone"), - ("Browse for an existing checkout", "browse")], - help_lines=[ - "audio.cpp hosts the TTS model families " - "this generator uses.", - "Clone it into the project's app " - "directory, or point at an existing " - "checkout."]) - if choice == "clone": - target = APP_DIR / AUDIOCPP_DIR_NAME - with tui.suspend(stdscr): - rc = common.git_clone(AUDIOCPP_GIT_URL, - target) - if rc != 0: - raise _TuiError( - f"git clone failed (exit {rc}). Clone " - f"audio.cpp manually: git clone " - f"{AUDIOCPP_GIT_URL} {target}") - audiocpp_dir = target - else: - audiocpp_dir = do_browse() + audiocpp_dir = do_browse() audiocpp_dir = Path(audiocpp_dir).resolve() if not audiocpp_dir.is_dir(): raise _TuiError(f"audio.cpp checkout not found: " @@ -993,37 +994,28 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser output_path = args.output if args.output is not None \ else audiocpp_dir / "server.json" - esc_back = args.audiocpp_dir is None - went_back = False - if not args.force and output_path.exists(): - decision = tui.confirm( - stdscr, f"{output_path} already exists. Overwrite?", - default=True, - cancel_value=_GO_BACK if esc_back else None) - if decision is _GO_BACK: - went_back = True - elif decision is False: - if args.output is None: - output_path = Path.cwd() / "server.json" - if output_path.exists(): - decision = tui.confirm( - stdscr, - f"{output_path} already exists. " - "Overwrite?", - default=True, - cancel_value=_GO_BACK if esc_back else None) - if decision is _GO_BACK: - went_back = True - elif decision is False: - return None - else: - return None - if went_back: - auto_accept = False - browser_start = audiocpp_dir - force_browse = True - continue break + + # 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 = {}, {} + 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_lazy = existing_config.get("lazy_load") \ + if existing_config else None + existing_voice_dir = existing_config.get("voice_dir") \ + if existing_config else None detected_backend = detect_backend(audiocpp_dir) step = 1 continue @@ -1048,10 +1040,24 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser catalog_by_family[family]) if opt["recommended"]] else: tree_families = _build_tree_families(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 existing_selected.items(): + if family not in catalog_by_family: + continue + family_index = catalog.index(catalog_by_family[family]) + valid_dirs = {opt["target_directory"] + for opt in package_dir_options( + 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) + back_value=_GO_BACK, checked=checked_set) if picked is _GO_BACK: step = 0 continue @@ -1100,7 +1106,7 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser model_entries, entry_ids, install_guidance, \ design_entry_ids, include_clone = _build_entries( family_keys, chosen, catalog_by_family, - task_picker, id_picker) + task_picker, id_picker, known_tasks=existing_tasks) except _GoBack: step = 1 continue @@ -1114,7 +1120,9 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser host = args.host else: host = tui.line_edit( - stdscr, "Bind host", DEFAULT_HOST, + stdscr, "Bind host", + existing_host if isinstance(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) @@ -1125,7 +1133,9 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser port = args.port else: port_text = tui.line_edit( - stdscr, "Port", str(config_port()), + stdscr, "Port", + str(existing_port) if isinstance(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", @@ -1154,6 +1164,11 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser # Already built: use the detected backend, no menu, no build. backend = detected_backend build = False + elif existing_backend in BACKENDS: + # Modify flow: keep the backend an existing server.json + # records (already configured, no rebuild needed). + backend = existing_backend + build = False else: backend_options, backend_default = _backend_options(None) backend = tui.menu( @@ -1173,6 +1188,8 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser step = 2 continue default_lazy = len(model_entries) > 1 + if isinstance(existing_lazy, bool): + default_lazy = existing_lazy if args.lazy_load: lazy_load = True else: @@ -1192,6 +1209,10 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser wav_dir = args.input_dir elif include_clone: wav_start = detect_wav_dir(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(existing_voice_dir, str) and existing_voice_dir: + wav_start = Path(existing_voice_dir) wav_dir = tui.browse_directory( stdscr, "Select the directory with your .wav voices", info=_wav_dir_info, preview=_wav_dir_preview, @@ -1240,16 +1261,35 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser if sync_model_ids is _GO_BACK: step = 5 continue + step = 7 + continue + + if step == 7: + # Delete unused models: already-downloaded models that the new + # selection no longer hosts. Only offered in the modify flow (an + # existing config was loaded), since a fresh --force run is an + # explicit overwrite. Esc falls back to the model-id sync (6). + new_paths = {entry["path"] for entry in model_entries} + unused_entries = unused_installed_entries(output_path, new_paths) \ + if existing_config is not None else [] + delete_unused = False + if unused_entries: + delete_unused = tui.confirm( + stdscr, "Delete unused models?", default=False, + cancel_value=_GO_BACK) + if delete_unused is _GO_BACK: + step = 6 + continue step = 8 continue if step == 8: # Automatic model download (or print the install commands). Esc - # falls back to the model-id sync (step 6). + # falls back to the delete-unused step (7). try: download = _decide_download(audiocpp_dir, ask_confirm) except _GoBack: - step = 6 + step = 7 continue return { "audiocpp_dir": audiocpp_dir, @@ -1273,9 +1313,64 @@ def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser "wav_dir": wav_dir, "plan": plan, "download": download, + "delete_unused": delete_unused, + "unused_entries": unused_entries, } +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. @@ -1322,6 +1417,70 @@ def missing_model_entries(server_json: Path) -> List[dict]: 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). @@ -1331,12 +1490,7 @@ def model_install_hints(audiocpp_dir: Path, carries the exact ``model_manager_v2.py install`` command; entries whose directory matches no catalog package just name the path. """ - 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 + by_path = _install_id_by_path(audiocpp_dir) hints: List[str] = [] for item in missing: install_id = by_path.get(item["rel"]) @@ -1348,6 +1502,104 @@ def model_install_hints(audiocpp_dir: Path, return hints +def install_models(audiocpp_dir: Path, + guidance: List[Tuple[str, str]]) -> None: + """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; a failing install is reported as a + warning and does not abort the rest. Used by the hub's "Download Missing + Models" action (see ``missing_model_install_guidance`` for the mapping). + """ + _install_models(audiocpp_dir, guidance, download=True) + + +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 + AUDIOCPP_DIR 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() -> int: + """Remove the audio.cpp backend entirely: stop its server, delete the checkout. + + The checkout (``app/audio.cpp``, or wherever ``find_local_checkout`` + resolves it) 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). Returns the exit + code. + """ + servers.stop("audiocpp") + 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]: """Best-effort location of an audio.cpp checkout with model_specs. @@ -1555,6 +1807,15 @@ def _execute(settings: dict, args: argparse.Namespace) -> int: settings["host"], settings["port"], settings["backend"], settings["lazy_load"], transcripts, 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"]) @@ -1806,7 +2067,9 @@ def build_parser() -> argparse.ArgumentParser: "(default: base)") parser.add_argument("--force", action="store_true", help="Overwrite the output file (and prompt_text) " - "without prompting") + "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 " @@ -1893,12 +2156,6 @@ def _detect_remote(managed: bool = False) -> Tuple[bool, dict]: return False, {} -configure_actions: List[ConfigureAction] = [ - ConfigureAction("Reconfigure audio.cpp (models, voices, server.json)", - run_tui), -] - - def main() -> int: parser = build_parser() args = parser.parse_args() diff --git a/app/backends/common.py b/app/backends/common.py index d5e1b6b..08c8863 100644 --- a/app/backends/common.py +++ b/app/backends/common.py @@ -299,3 +299,13 @@ def pip_install(packages: List[str]) -> int: """ from backends import envs return envs.pip_install(packages) + + +def pip_uninstall(packages: List[str]) -> int: + """pip uninstall PACKAGES from the managed venv. Returns exit code. + + Delegates to ``backends.envs.pip_uninstall`` (local import to avoid a + circular import). Used by the backends' ``uninstall`` action. + """ + from backends import envs + return envs.pip_uninstall(packages) diff --git a/app/backends/envs.py b/app/backends/envs.py index 6e5b6cc..5a51a33 100644 --- a/app/backends/envs.py +++ b/app/backends/envs.py @@ -107,6 +107,20 @@ def pip_install(packages: List[str]) -> int: [str(env_python()), "-m", "pip", "install", *packages]) +def pip_uninstall(packages: List[str]) -> int: + """pip uninstall PACKAGES from the venv. Returns pip's exit code. + + Used by the backends' ``uninstall`` action to remove pip-installed TTS + packages from the managed environment. A missing env is a no-op (there + is nothing to uninstall from), reported as success. + """ + if not env_exists(): + return 0 + print(f"[INFO] pip uninstall {' '.join(packages)} from {ENV_DIR}...") + return common.run_console_subprocess( + [str(env_python()), "-m", "pip", "uninstall", "-y", *packages]) + + def module_available(module: str) -> bool: """True when MODULE imports inside the venv (e.g. qwen_tts, faster_qwen3_tts). diff --git a/app/backends/faster.py b/app/backends/faster.py index e1249ca..7e1be74 100755 --- a/app/backends/faster.py +++ b/app/backends/faster.py @@ -14,10 +14,17 @@ Usage: python app/backends/faster.py [--wavs WAV_DIR] [--output PATH] [--language LANG] [--whisper-model NAME] [--force] [--port PORT] [--voice NAME] [--skip-install] [--skip-clone] + +When the target ``voices.json`` already exists, the TUI wizard runs as a +"modify": it loads the existing voices and pre-fills the language and +wav directory from them instead of prompting to overwrite, asks whether +to only transcribe new voices or re-transcribe everything, and writes +back to the same file. """ import argparse import json +import shutil import sys from pathlib import Path from typing import List, Optional @@ -26,7 +33,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from backends import ( BackendStatus, - ConfigureAction, ServerSpec, common, envs, @@ -95,9 +101,68 @@ def build_voices(wav_files: list, language: str, whisper_model: str) -> dict: return voices +def load_voices(path: Path) -> dict: + """Read voices.json into a name -> voice-entry dict, or {} when unusable. + + Returns {} for a missing file, unreadable content, or a non-dict + document. Used by the wizard's modify flow to seed its defaults from an + existing voices.json instead of prompting to overwrite it. + """ + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + if not isinstance(data, dict): + return {} + return data + + +def _decide_faster_transcription(wav_files: list, existing_voices: dict, + confirm) -> Optional[dict]: + """Decide which voices to transcribe when a voices.json already exists. + + CONFIRM asks the yes/no question (returning True/False, or None when the + user backs out). With new .wavs present it offers to transcribe only + those (default Yes); otherwise — and always, per the modify design — it + offers to re-transcribe everything (default No), so a stale transcript + can be refreshed even when every voice is already known. Returns a plan + dict: ``{"mode": "missing"|"all"|"keep", "missing": [...], "existing": + {...}}``, or None when CONFIRM cancelled. + """ + existing = dict(existing_voices) + new_wavs = [wav for wav in wav_files if wav.stem not in existing] + if new_wavs: + choice = confirm("Existing voices.json found. Only transcribe the " + "new voices?", True) + if choice is None: + return None + if choice: + return {"mode": "missing", "missing": new_wavs, + "existing": existing} + return {"mode": "all", "missing": [], "existing": existing} + choice = confirm("All voices already in voices.json. Re-transcribe " + "anyway?", False) + if choice is None: + return None + if choice: + return {"mode": "all", "missing": [], "existing": existing} + return {"mode": "keep", "missing": [], "existing": existing} + + def _write_voices_json(output_path: Path, wav_dir: Path, language: str, - whisper_model: str, force: bool) -> Optional[dict]: - """Transcribe the wav dir and write voices.json; return the voices dict.""" + whisper_model: str, plan: Optional[dict]) -> Optional[dict]: + """Transcribe the wav dir and write voices.json; return the voices dict. + + PLAN (built by ``_decide_faster_transcription`` in the wizard, or an + "all" plan for a fresh/flag run) decides whether every voice is + re-transcribed ("all"), only the new ones ("missing" — merged into the + existing entries), or nothing changes ("keep" — the existing file is + left untouched and returned as-is). None (cancelled) writes nothing. + """ + if plan is None: + return None + if plan["mode"] == "keep": + return dict(plan["existing"]) wav_files = find_wav_files(wav_dir) if not wav_files: print(f"[ERROR] No .wav files found in {wav_dir}") @@ -106,7 +171,11 @@ def _write_voices_json(output_path: Path, wav_dir: Path, language: str, print("[WARNING] Neither faster_whisper nor whisper was found, so " "transcripts will be empty — install one or edit voices.json " "by hand.") - voices = build_voices(wav_files, language, whisper_model) + if plan["mode"] == "missing": + voices = dict(plan["existing"]) + voices.update(build_voices(plan["missing"], language, whisper_model)) + else: + voices = build_voices(wav_files, language, whisper_model) with output_path.open("w", encoding="utf-8") as handle: json.dump(voices, handle, indent=4, ensure_ascii=False) handle.write("\n") @@ -142,17 +211,39 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: return None do_clone = choice - # Step 2: voices.json — wav dir, language, whisper model, output path. + # Step 2: voices.json — an existing one seeds the defaults (modify flow) + # instead of an overwrite prompt. + existing_voices = {} + default_output = args.output + if default_output is None and _is_cloned(): + default_output = _checkout() / "voices.json" + if default_output is not None and default_output.exists() \ + and not args.force: + existing_voices = load_voices(default_output) + + wav_start = VOICES_DIR + if existing_voices: + ref_dirs = {Path(voice["ref_audio"]).parent + for voice in existing_voices.values() + if isinstance(voice, dict) and voice.get("ref_audio")} + if len(ref_dirs) == 1: + wav_start = next(iter(ref_dirs)) + wav_dir = args.input_dir if wav_dir is None: wav_dir = tui.browse_directory( stdscr, "Select the directory with your .wav voices", info=common.wav_dir_info, preview=common.wav_dir_preview, - start=VOICES_DIR) + start=wav_start) language = args.language if language is None: + default_language = config.LANGUAGE + for voice in existing_voices.values(): + if isinstance(voice, dict) and voice.get("language"): + default_language = voice["language"] + break lang_text = tui.line_edit( - stdscr, "Language", config.LANGUAGE, + stdscr, "Language", default_language, validate=lambda s: None if _try_language(s) else "Unknown language (e.g. English, en)", help_lines=["Language for every voice, as passed to the TTS " @@ -170,12 +261,16 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: # when the checkout is not present (so a flag-only run still works). output_path = (_checkout() / "voices.json") if _is_cloned() \ else (wav_dir / "voices.json") - if output_path.exists() and not args.force: - choice = confirm(f"{output_path} already exists. Overwrite?", - default=True) - if choice is None or choice is False: - # Fall back to a path in the current directory. - output_path = Path.cwd() / "voices.json" + + # Transcription plan: re-transcribe only new voices (or all of them) — + # the "re-transcribe anyway?" offer appears even when nothing is new. + plan: Optional[dict] = {"mode": "all", "missing": [], "existing": {}} + wav_files = find_wav_files(wav_dir) + if wav_files and existing_voices and not args.force: + plan = _decide_faster_transcription(wav_files, existing_voices, + confirm) + if plan is None: + return None # Step 3: port + default voice. port = args.port @@ -195,6 +290,7 @@ def _wizard(stdscr, args: argparse.Namespace) -> Optional[dict]: "output_path": output_path, "port": port, "force": args.force, + "plan": plan, } @@ -226,7 +322,7 @@ def _execute(settings: dict) -> int: voices = _write_voices_json(settings["output_path"], settings["wav_dir"], settings["language"], settings["whisper_model"], - settings["force"]) + settings["plan"]) if voices is None: return 1 @@ -308,6 +404,7 @@ def _collect_from_flags(args: argparse.Namespace, "output_path": output_path, "port": args.port if args.port is not None else _config_port(), "force": args.force, + "plan": {"mode": "all", "missing": [], "existing": {}}, } @@ -332,7 +429,8 @@ def build_parser() -> argparse.ArgumentParser: "(default: base)") parser.add_argument("--force", action="store_true", help="Overwrite an existing voices.json without " - "prompting") + "prompting; in the TUI, re-transcribe every " + "voice instead of reusing the existing file") parser.add_argument("--port", type=int, default=None, help="Server port to record in app/converter/config.py " "(default: the port in FASTER_API_URL)") @@ -394,22 +492,27 @@ def _detect_remote(managed: bool = False): return False, {} -def _run_voices_only_tui() -> int: - """Rebuild voices.json via the TUI (the "configure" action). +def uninstall() -> int: + """Remove the faster-qwen3-tts backend entirely. - Runs the same wizard but skips the pip/clone prerequisites so it goes - straight to picking the .wav directory and writing voices.json. + Uninstalls the pip package (``faster-qwen3-tts``) from the managed venv + and deletes the cloned checkout (``app/faster-qwen3-tts``, which holds + examples/openai_server.py and voices.json). A running server this tool + started is stopped first (best-effort). Returns the exit code. """ - args = build_parser().parse_args([]) - args.skip_install = True - args.skip_clone = True - return run_tui(args) - - -configure_actions: List[ConfigureAction] = [ - ConfigureAction("Rebuild voices.json", _run_voices_only_tui), - ConfigureAction("Reconfigure faster-qwen3-tts", run_tui), -] + servers.stop("faster") + rc = common.pip_uninstall(["faster-qwen3-tts"]) + if rc != 0: + print("[WARNING] pip uninstall failed (exit " + f"{rc}); remove faster-qwen3-tts from the managed venv manually") + else: + print("[OK] faster-qwen3-tts removed.") + checkout = _checkout() + if checkout.is_dir(): + print(f"[INFO] Removing checkout {checkout}...") + shutil.rmtree(checkout, ignore_errors=True) + print("[OK] checkout removed.") + return 0 def main() -> int: diff --git a/app/backends/qwen.py b/app/backends/qwen.py index 7f821fa..b170eb8 100644 --- a/app/backends/qwen.py +++ b/app/backends/qwen.py @@ -22,7 +22,6 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from backends import ( BackendStatus, - ConfigureAction, ServerSpec, common, envs, @@ -282,9 +281,23 @@ def _detect_remote(managed: bool = False): return remote_models, remote_urls -configure_actions: List[ConfigureAction] = [ - ConfigureAction("Reconfigure qwen-tts (ports/speaker)", run_tui), -] +def uninstall() -> int: + """Remove the qwen-tts backend entirely: stop its servers, pip uninstall. + + qwen-tts is a pip package (``qwen_tts`` + the ``qwen-tts-demo`` script) + installed into the managed venv, so uninstalling it removes the backend. + Any server this tool started is stopped first (best-effort). Returns the + exit code. + """ + servers.stop("qwen-custom") + servers.stop("qwen-clone") + rc = common.pip_uninstall([QWEN_PIP_PKG]) + if rc != 0: + print(f"[WARNING] pip uninstall failed (exit {rc}); remove " + f"{QWEN_PIP_PKG} from the managed venv manually") + else: + print(f"[OK] {QWEN_PIP_PKG} removed.") + return 0 def main() -> int: diff --git a/app/docs/backend-audiocpp.md b/app/docs/backend-audiocpp.md index 5ce7f49..db35c81 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 **Set up a 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). 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). 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. If you prefer to install the backend yourself (in your own environment, not the managed venv), the manual steps are below. Either way the hub detects a running server by its port, so a manually-installed backend works once its server is up. diff --git a/app/docs/backend-faster.md b/app/docs/backend-faster.md index d4b5b40..3036ca7 100644 --- a/app/docs/backend-faster.md +++ b/app/docs/backend-faster.md @@ -2,7 +2,7 @@ `--backend faster` talks to the OpenAI-compatible server from [faster-qwen3-tts](https://github.com/andimarafioti/faster-qwen3-tts), which uses CUDA graph capture for roughly 5-10x faster inference with the same models. **It requires an NVIDIA GPU**. -The easiest way is to run `python audiobook.py` → **Set up a backend… → faster-qwen3-tts** (or `python app/backends/faster.py path/to/clone/wavs`): the TUI pip-installs `faster-qwen3-tts[demo]` into its managed venv (`app/envs/tts`), clones the repo, transcribes the `.wav` files with `whisper`, and writes `voices.json` for you. You can also start the server from the hub's **Start/Stop Backend Servers** menu, or let a conversion start it automatically. +The easiest way is to run `python audiobook.py` → **Configure backends… → Install Backend → faster-qwen3-tts** (or `python app/backends/faster.py path/to/clone/wavs`): the TUI pip-installs `faster-qwen3-tts[demo]` into its managed venv (`app/envs/tts`), clones the repo, transcribes the `.wav` files with `whisper`, and writes `voices.json` for you. You can also start the server from the hub's **Start/Stop Backend Servers** menu, or let a conversion start it automatically. If you prefer to install the backend yourself (in your own environment, not the managed venv), the manual steps are below. Either way the hub detects a running server by its port, so a manually-installed backend works once its server is up. To use a server on another machine, set `FASTER_REMOTE_URL` in `app/converter/config.py` to its `host:port` (default `127.0.0.1:8000`) — the hub probes it and offers a `faster-qwen3-tts [remote]` entry — or pass `--api-url` on the CLI. diff --git a/app/docs/backend-qwen.md b/app/docs/backend-qwen.md index 0cc2e5a..800c7ae 100644 --- a/app/docs/backend-qwen.md +++ b/app/docs/backend-qwen.md @@ -1,6 +1,6 @@ # Backend Option 2: Qwen3-TTS -The easiest way is to run `python audiobook.py` → **Set up a backend… → qwen-tts** (or `python app/backends/qwen.py`): the TUI pip-installs `qwen-tts` into its managed venv (`app/envs/tts`), configures the two ports and the built-in speaker in `app/converter/config.py`, and prints the launch commands. You can also start the server from the hub's **Start/Stop Backend Servers** menu, or let a conversion start it automatically. +The easiest way is to run `python audiobook.py` → **Configure backends… → Install Backend → qwen-tts** (or `python app/backends/qwen.py`): the TUI pip-installs `qwen-tts` into its managed venv (`app/envs/tts`), configures the two ports and the built-in speaker in `app/converter/config.py`, and prints the launch commands. You can also start the server from the hub's **Start/Stop Backend Servers** menu, or let a conversion start it automatically. If you prefer to install the backend yourself (in your own environment, not the managed venv), the manual steps are below. Either way the hub detects a running server by its port, so a manually-installed backend works once its server is up. To use demo servers on another machine, set `QWEN_REMOTE_URL`/`CLONE_REMOTE_URL` in `app/converter/config.py` to their `host:port` (defaults `127.0.0.1:7860`/`:7861`) — the hub probes each and offers the matching `qwen-tts [remote]` mode — or pass `--api-url` on the CLI. diff --git a/app/tests/test_backends.py b/app/tests/test_backends.py index acee6b6..0e260be 100644 --- a/app/tests/test_backends.py +++ b/app/tests/test_backends.py @@ -31,13 +31,11 @@ class RegistryTests(unittest.TestCase): keys = [info.key for info in REGISTRY] self.assertEqual(keys, ["audiocpp", "qwen", "faster"]) - def test_every_entry_has_detect_and_setup_tui(self): + def test_every_entry_has_detect_setup_and_uninstall(self): for info in REGISTRY: self.assertTrue(callable(info.detect), info.key) self.assertTrue(callable(info.setup_tui), info.key) - self.assertIsInstance(info.configure_actions, list) - for action in info.configure_actions: - self.assertTrue(callable(action.run)) + self.assertTrue(callable(info.uninstall), info.key) def test_get_returns_entry_by_key(self): self.assertIs(get("audiocpp").key, "audiocpp") diff --git a/app/tests/test_backends_audiocpp.py b/app/tests/test_backends_audiocpp.py index 563ed78..3d042db 100644 --- a/app/tests/test_backends_audiocpp.py +++ b/app/tests/test_backends_audiocpp.py @@ -1270,5 +1270,301 @@ class DetectServerSpecTests(unittest.TestCase): for line in status.details)) +class InstalledModelEntriesTests(unittest.TestCase): + """installed_model_entries: the complement of missing_model_entries.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def _server_json(self, models): + path = self.dir / "server.json" + path.write_text(json.dumps({"models": models}), encoding="utf-8") + return path + + def test_lists_entries_whose_files_are_on_disk(self): + (self.dir / "models" / "present").mkdir(parents=True) + (self.dir / "models" / "present" / "m.gguf").write_bytes(b"x") + path = self._server_json([ + {"id": "a", "path": "models/present"}, + {"id": "b", "path": "models/absent"}, + ]) + installed = make_server.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), []) + + +class MissingModelInstallGuidanceTests(unittest.TestCase): + """missing_model_install_guidance: missing paths -> (id, install_id).""" + + def test_maps_paths_and_skips_unmapped(self): + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) + specs = checkout / "model_specs" + specs.mkdir() + (specs / "qwen3_tts.json").write_text(json.dumps({ + "family": "qwen3_tts", "category": "tts", + "tasks": ["tts"], + "packages": [{ + "id": "qwen3_tts_0_6b_base_q8_0", "format": "gguf", + "target_directory": "Qwen3-TTS-12Hz-0.6B-Base-GGUF", + }], + }), encoding="utf-8") + missing = [ + {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}, + {"id": "x", "rel": "models/nope"}, + ] + guidance = make_server.missing_model_install_guidance( + checkout, missing) + self.assertEqual(guidance, + [("qwen", "qwen3_tts_0_6b_base_q8_0")]) + + +class LoadServerConfigTests(unittest.TestCase): + """load_server_config: read server.json, or None when unusable.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + + def tearDown(self): + self._tmp.cleanup() + + def test_reads_dict_document(self): + 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), + {"host": "0.0.0.0", "models": []}) + + def test_missing_file_returns_none(self): + self.assertIsNone(make_server.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)) + + 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)) + + +class ServerConfigSelectionsTests(unittest.TestCase): + """server_config_selections: map server.json models back to the catalog.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.checkout = _make_checkout(Path(self._tmp.name)) + self.catalog = make_server.load_model_catalog(self.checkout) + + def tearDown(self): + self._tmp.cleanup() + + def test_maps_paths_to_family_dirs_and_tasks(self): + config = {"models": [ + {"id": "qwen", "family": "qwen3_tts", + "path": "models/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"}, + {"id": "qwen-design", "family": "qwen3_tts", + "path": "models/Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF", + "task": "vdes"}, + {"id": "higgs", "family": "higgs_audio_tts", + "path": "models/Higgs-Audio-v3-TTS-4B-GGUF", "task": "tts"}, + ]} + selected, tasks = make_server.server_config_selections(config, + self.catalog) + self.assertEqual(selected["qwen3_tts"], + ["Qwen3-TTS-12Hz-1.7B-Base-GGUF", + "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF"]) + self.assertEqual(selected["higgs_audio_tts"], + ["Higgs-Audio-v3-TTS-4B-GGUF"]) + self.assertEqual(tasks[("qwen3_tts", + "Qwen3-TTS-12Hz-1.7B-VoiceDesign-GGUF")], + "vdes") + self.assertEqual(tasks[("qwen3_tts", + "Qwen3-TTS-12Hz-1.7B-Base-GGUF")], "tts") + + def test_unknown_family_ignored(self): + config = {"models": [ + {"id": "x", "family": "not_a_family", "path": "models/x"}, + ]} + selected, tasks = make_server.server_config_selections(config, + self.catalog) + self.assertEqual(selected, {}) + self.assertEqual(tasks, {}) + + def test_absolute_and_unprefixed_paths_kept_as_targets(self): + config = {"models": [ + {"id": "qwen", "family": "qwen3_tts", + "path": "/abs/Qwen3-TTS-12Hz-1.7B-Base-GGUF", "task": "tts"}, + ]} + selected, tasks = make_server.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": []}, + self.catalog) + self.assertEqual(selected, {}) + self.assertEqual(tasks, {}) + + +class UnusedInstalledEntriesTests(unittest.TestCase): + """unused_installed_entries: installed models dropped by a new selection.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + (self.dir / "models" / "kept").mkdir(parents=True) + (self.dir / "models" / "kept" / "m.gguf").write_bytes(b"x") + (self.dir / "models" / "dropped").mkdir() + (self.dir / "models" / "dropped" / "m.gguf").write_bytes(b"x") + (self.dir / "models" / "missing").mkdir() # empty: not installed + + def tearDown(self): + self._tmp.cleanup() + + def _server_json(self, models): + path = self.dir / "server.json" + path.write_text(json.dumps({"models": models}), encoding="utf-8") + return path + + def test_returns_installed_entries_not_in_new_paths(self): + path = self._server_json([ + {"id": "kept", "path": "models/kept"}, + {"id": "dropped", "path": "models/dropped"}, + {"id": "missing", "path": "models/missing"}, + ]) + unused = make_server.unused_installed_entries( + path, {"models/kept"}) + self.assertEqual([entry["id"] for entry in unused], ["dropped"]) + + def test_nothing_unused_when_all_kept(self): + path = self._server_json([ + {"id": "kept", "path": "models/kept"}, + ]) + unused = make_server.unused_installed_entries( + path, {"models/kept"}) + self.assertEqual(unused, []) + + +class DeleteModelFilesTests(unittest.TestCase): + """delete_model_files: remove on-disk model files for {id, rel} entries.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + (self.dir / "models" / "a").mkdir(parents=True) + (self.dir / "models" / "a" / "m.gguf").write_bytes(b"x") + (self.dir / "models" / "b").mkdir() + (self.dir / "models" / "b" / "m.gguf").write_bytes(b"x") + (self.dir / "models" / "c").mkdir(parents=True) + self.server_json = self.dir / "server.json" + self.server_json.write_text(json.dumps({ + "models": [ + {"id": "a", "path": "models/a"}, + {"id": "b", "path": "models/b"}, + {"id": "c", "path": "models/c"}, + ], + }), encoding="utf-8") + + def tearDown(self): + self._tmp.cleanup() + + def test_removes_dirs_and_counts(self): + removed = make_server.delete_model_files( + self.server_json, + [{"id": "a", "rel": "models/a"}, {"id": "b", "rel": "models/b"}]) + self.assertEqual(removed, 2) + self.assertFalse((self.dir / "models" / "a").exists()) + self.assertFalse((self.dir / "models" / "b").exists()) + self.assertTrue((self.dir / "models" / "c").exists()) + + def test_missing_paths_ignored(self): + removed = make_server.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( + self.server_json, [{"id": "s", "rel": "models/single.gguf"}]) + self.assertEqual(removed, 1) + self.assertFalse(file_path.exists()) + + def test_absolute_rel_path_honored(self): + target = self.dir / "absolute" + target.mkdir() + (target / "m.gguf").write_bytes(b"x") + removed = make_server.delete_model_files( + self.server_json, [{"id": "a", "rel": str(target)}]) + self.assertEqual(removed, 1) + self.assertFalse(target.exists()) + + +class InstallModelsTests(unittest.TestCase): + """install_models: runs the install helper with download=True.""" + + def test_downloads_delegating_to_install_models(self): + 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) + mk.assert_called_once_with(checkout, guidance, download=True) + + +class HandInstallGuidanceTests(unittest.TestCase): + """hand_install_guidance: explains how to install models by hand.""" + + def test_lists_each_model_and_its_path(self): + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) + message = make_server.hand_install_guidance(checkout, [ + {"id": "qwen", "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}, + {"id": "higgs", "rel": "models/Higgs-Audio-4B-GGUF"}, + ]) + self.assertIn("qwen", message) + self.assertIn("models/Qwen3-TTS-12Hz-0.6B-Base-GGUF", message) + self.assertIn("higgs", message) + self.assertIn("models/Higgs-Audio-4B-GGUF", message) + self.assertIn("download", message.lower()) + + +class UninstallTests(unittest.TestCase): + """uninstall: stop the server and remove the checkout.""" + + def test_removes_checkout(self): + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) / "audio.cpp" + checkout.mkdir() + with patch.object(make_server, "find_local_checkout", + return_value=checkout), \ + patch.object(make_server.servers, "stop") as mk_stop: + rc = make_server.uninstall() + self.assertEqual(rc, 0) + self.assertFalse(checkout.exists()) + mk_stop.assert_called_once_with("audiocpp") + + def test_no_checkout_is_a_noop(self): + with patch.object(make_server, "find_local_checkout", + return_value=None), \ + patch.object(make_server.servers, "stop") as mk_stop: + rc = make_server.uninstall() + self.assertEqual(rc, 0) + mk_stop.assert_called_once_with("audiocpp") + + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_backends_faster.py b/app/tests/test_backends_faster.py index 641f6ee..21baea7 100644 --- a/app/tests/test_backends_faster.py +++ b/app/tests/test_backends_faster.py @@ -80,6 +80,83 @@ class BuildVoicesTests(unittest.TestCase): self.assertEqual(mock_transcribe.call_args.kwargs["model_name"], "large-v3") +class LoadVoicesTests(unittest.TestCase): + """load_voices: read voices.json, or {} when unusable.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.dir = Path(self._tmp.name) + self.path = self.dir / "voices.json" + + def tearDown(self): + self._tmp.cleanup() + + def test_reads_dict_document(self): + self.path.write_text(json.dumps({"narrator": {"ref_text": "hi"}}), + encoding="utf-8") + self.assertEqual(make_voices.load_voices(self.path), + {"narrator": {"ref_text": "hi"}}) + + def test_missing_file_returns_empty(self): + self.assertEqual(make_voices.load_voices(self.path), {}) + + def test_unreadable_json_returns_empty(self): + self.path.write_text("not json", encoding="utf-8") + self.assertEqual(make_voices.load_voices(self.path), {}) + + def test_non_dict_document_returns_empty(self): + self.path.write_text("[1, 2]", encoding="utf-8") + self.assertEqual(make_voices.load_voices(self.path), {}) + + +class DecideFasterTranscriptionTests(unittest.TestCase): + """_decide_faster_transcription: the re-transcribe plan questions.""" + + def setUp(self): + self._tmp = tempfile.TemporaryDirectory() + self.folder = Path(self._tmp.name) + self.narrator = self.folder / "narrator.wav" + self.narrator.write_bytes(b"x") + self.new_voice = self.folder / "new.wav" + self.new_voice.write_bytes(b"x") + + def tearDown(self): + self._tmp.cleanup() + + def test_new_voices_default_to_missing_mode(self): + confirm = lambda q, default=True: True # noqa: E731 + plan = make_voices._decide_faster_transcription( + [self.narrator, self.new_voice], + {"narrator": {"ref_text": "old"}}, confirm) + self.assertEqual(plan["mode"], "missing") + self.assertEqual([w.name for w in plan["missing"]], ["new.wav"]) + + def test_declining_new_voices_transcribes_all(self): + confirm = lambda q, default=True: False # noqa: E731 + plan = make_voices._decide_faster_transcription( + [self.narrator, self.new_voice], + {"narrator": {"ref_text": "old"}}, confirm) + self.assertEqual(plan["mode"], "all") + + def test_no_new_voices_offers_retranscribe_default_no(self): + confirm = lambda q, default=True: default # noqa: E731 + plan = make_voices._decide_faster_transcription( + [self.narrator], {"narrator": {"ref_text": "old"}}, confirm) + self.assertEqual(plan["mode"], "keep") + + def test_no_new_voices_accepted_retranscribes_all(self): + confirm = lambda q, default=True: True # noqa: E731 + plan = make_voices._decide_faster_transcription( + [self.narrator], {"narrator": {"ref_text": "old"}}, confirm) + self.assertEqual(plan["mode"], "all") + + def test_cancel_returns_none(self): + confirm = lambda q, default=True: None # noqa: E731 + plan = make_voices._decide_faster_transcription( + [self.narrator], {"narrator": {"ref_text": "old"}}, confirm) + self.assertIsNone(plan) + + class MainTests(unittest.TestCase): """The flag-only (non-TUI) path through main(), end to end.""" diff --git a/app/tests/test_hub.py b/app/tests/test_hub.py index 5f91a61..a8e3ac1 100644 --- a/app/tests/test_hub.py +++ b/app/tests/test_hub.py @@ -129,16 +129,16 @@ class HubMenuTests(unittest.TestCase): return BackendStatus(key, label, installed=False, configured=False) def test_quit_returns_none_when_no_backend(self): - # No backends installed/running: menu is [Set up, Settings, Quit]. - # Quit is the 3rd option (Down twice) then Enter. + # No backends installed/running: menu is [Configure backends, + # Settings, Quit]. Quit is the 3rd option (Down twice) then Enter. screen = FakeScreen(keys=[FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, 10]) with patch.object(hub, "detect_all", return_value=[]): result = hub._hub_menu(screen) self.assertIsNone(result) - def test_menu_has_only_setup_settings_and_quit_without_backends(self): + def test_menu_has_only_configure_settings_and_quit_without_backends(self): # Capture the options handed to tui.menu: with nothing installed or - # running, Convert/Configure must be absent. + # running, Convert/Server must be absent. captured = {} def fake_menu(stdscr, title, options, **kwargs): @@ -150,9 +150,9 @@ class HubMenuTests(unittest.TestCase): patch.object(hub, "detect_all", return_value=[]): hub._hub_menu(screen) labels = [label for label, _ in captured["options"]] - self.assertEqual(labels, ["Set up a backend", "Settings", "Quit"]) + self.assertEqual(labels, ["Configure backends", "Settings", "Quit"]) - def test_menu_has_all_six_when_one_installed(self): + def test_menu_has_all_five_when_one_installed(self): captured = {} def fake_menu(stdscr, title, options, **kwargs): @@ -169,9 +169,8 @@ class HubMenuTests(unittest.TestCase): labels = [label for label, _ in captured["options"]] self.assertEqual( labels, - ["Convert books", "Set up a backend", - "Configure a backend", "Start/Stop Backend Servers", - "Settings", "Quit"]) + ["Convert books", "Configure backends", + "Start/Stop Backend Servers", "Settings", "Quit"]) # The status table is passed through, one row per backend. self.assertEqual(captured["rows"], [("qwen-tts", "installed", "warn", "body")]) @@ -199,9 +198,9 @@ class HubMenuTests(unittest.TestCase): [("audio.cpp", "unavailable", "err", "dim"), ("qwen-tts", "running [remote]", "ok", "body")]) - def test_menu_hides_configure_and_server_when_only_running(self): + def test_menu_hides_server_when_only_running(self): # Running but not installed (an external server) still unlocks - # Convert — but Configure/Server need the backend on this machine. + # Convert — but Start/Stop needs the backend on this machine. captured = {} def fake_menu(stdscr, title, options, **kwargs): @@ -217,7 +216,7 @@ class HubMenuTests(unittest.TestCase): labels = [label for label, _ in captured["options"]] self.assertEqual( labels, - ["Convert books", "Set up a backend", "Settings", "Quit"]) + ["Convert books", "Configure backends", "Settings", "Quit"]) def test_ffmpeg_warning_shown_when_missing(self): # ffmpeg not on PATH → a red notice is passed above the table. @@ -252,9 +251,9 @@ class HubMenuTests(unittest.TestCase): def test_convert_with_no_available_backend_flashes(self): # Installed-but-not-ready backends → Convert is offered, but the - # convert flow has nothing to list: it flashes a hint (no "Set up - # a backend" detour anymore) and returns to the main menu. Then - # quit: 6 main-menu options, Quit is the 6th (Down x5). + # convert flow has nothing to list: it flashes a hint (no "Configure + # backends" detour anymore) and returns to the main menu. Then + # quit: 5 main-menu options, Quit is the 5th (Down x4). from backends import BackendStatus statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, configured=False), @@ -269,11 +268,11 @@ class HubMenuTests(unittest.TestCase): with patch.object(hub, "detect_all", return_value=statuses), \ patch.object(hub.tui, "flash", fake_flash): - # Convert(Enter) → flash → main menu; Down x5 -> Quit, Enter. + # Convert(Enter) → flash → main menu; Down x4 -> Quit, Enter. screen = FakeScreen(keys=[10, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, FakeCurses.KEY_DOWN, - FakeCurses.KEY_DOWN, 10]) + 10]) result = hub._hub_menu(screen) self.assertIsNone(result) self.assertEqual(len(flashed), 1) @@ -283,8 +282,8 @@ class HubMenuTests(unittest.TestCase): class SubmenuStatusTableTests(unittest.TestCase): """First picker screen of every flow repeats the backend status table. - Entries themselves stay clean: setup lists bare labels, and the - Start/Stop menu offers only installed backends. + Entries themselves stay clean: the configure-backends menu lists flat + actions, and the Start/Stop menu offers only installed backends. """ def _capture_menu(self, captured): @@ -305,14 +304,15 @@ class SubmenuStatusTableTests(unittest.TestCase): return fake_form - def test_setup_menu_lists_bare_labels_and_status_table(self): + def test_configure_backends_menu_lists_actions_and_status_table(self): captured = {} - infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0), - BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), + BackendInfo("faster", "faster-qwen3-tts", lambda: None, + lambda: 0)] statuses = [ - BackendStatus("audiocpp", "audio.cpp", installed=True, + BackendStatus("qwen", "qwen-tts", installed=True, configured=True), - BackendStatus("qwen", "qwen-tts", installed=False, + BackendStatus("faster", "faster-qwen3-tts", installed=False, configured=False, running=True, remote=True), ] with patch.object(hub, "REGISTRY", infos), \ @@ -320,19 +320,66 @@ class SubmenuStatusTableTests(unittest.TestCase): self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/usr/bin/ffmpeg"): - result = hub._setup_menu(None, statuses) + result = hub._configure_backends_menu(None, statuses) self.assertIsNone(result) - # No inline "(running)"-style suffix on the entries anymore... + # Install (faster uninstalled), Configure (qwen installed), then + # Uninstall (qwen installed); no audio.cpp means no model actions. self.assertEqual([label for label, _ in captured["options"]], - ["audio.cpp", "qwen-tts"]) + ["Install Backend", "Configure qwen-tts", + "Uninstall Backend"]) # ...the shared status table carries the states instead. self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( captured["table_rows"], - [("audio.cpp", "installed", "warn", "body"), - ("qwen-tts", "running [remote]", "ok", "body")]) + [("qwen-tts", "installed", "warn", "body"), + ("faster-qwen3-tts", "running [remote]", "ok", "body")]) self.assertIsNone(captured["notice_lines"]) + def test_configure_backends_menu_install_only_when_nothing_installed(self): + captured = {} + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] + statuses = [BackendStatus("qwen", "qwen-tts", installed=False, + configured=False)] + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub.tui, "menu", + self._capture_menu(captured)), \ + patch.object(hub.shutil, "which", return_value="/x"): + result = hub._configure_backends_menu(None, statuses) + self.assertIsNone(result) + # Nothing installed: only the install entry is offered. + self.assertEqual([label for label, _ in captured["options"]], + ["Install Backend"]) + + def test_configure_backends_menu_audiocpp_model_actions(self): + captured = {} + infos = [BackendInfo("audiocpp", "audio.cpp", lambda: None, lambda: 0)] + statuses = [BackendStatus("audiocpp", "audio.cpp", installed=True, + configured=True)] + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) + (checkout / "server.json").write_text(json.dumps({ + "models": [{"id": "present", "path": "models/present"}, + {"id": "absent", "path": "models/absent"}], + }), encoding="utf-8") + (checkout / "models" / "present").mkdir(parents=True) + (checkout / "models" / "present" / "m.gguf").write_bytes(b"x") + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub.tui, "menu", + self._capture_menu(captured)), \ + patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=checkout), \ + patch.object(hub.shutil, "which", return_value="/x"): + result = hub._configure_backends_menu(None, statuses) + self.assertIsNone(result) + labels = [label for label, _ in captured["options"]] + # A model is missing (download), plus the installed backend's + # configure + uninstall entries. Deleting unused models now lives + # inside the "Configure audio.cpp" wizard, not here. + self.assertEqual( + labels, + ["Configure audio.cpp", "Download Missing Models (audio.cpp)", + "Uninstall Backend"]) + def test_convert_menu_builds_one_form_with_backend_field(self): captured = {} st = BackendStatus("qwen", "qwen-tts", installed=True, @@ -373,7 +420,7 @@ class SubmenuStatusTableTests(unittest.TestCase): self.assertEqual(menus, []) self.assertIn("No backend is ready", flashed[0]) - def test_configure_menu_shows_status_table(self): + def test_configure_backends_menu_shows_status_table(self): captured = {} infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0)] statuses = [BackendStatus("qwen", "qwen-tts", installed=True, @@ -382,7 +429,7 @@ class SubmenuStatusTableTests(unittest.TestCase): patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value="/x"): - result = hub._configure_menu(None, statuses) + result = hub._configure_backends_menu(None, statuses) self.assertIsNone(result) self.assertEqual(captured["table_title"], "Backend status") self.assertEqual( @@ -433,7 +480,7 @@ class SubmenuStatusTableTests(unittest.TestCase): patch.object(hub.tui, "menu", self._capture_menu(captured)), \ patch.object(hub.shutil, "which", return_value=None): - hub._setup_menu(None, statuses) + hub._configure_backends_menu(None, statuses) self.assertEqual(captured["notice_lines"], [("Warning: ffmpeg not installed!", "err")]) @@ -1408,5 +1455,164 @@ class AudiocppServerConfigTests(unittest.TestCase): self.assertFalse(audiocpp_backend.update_server_config_port(9090)) +class ConfigureBackendsDispatchTests(unittest.TestCase): + """run() and the configure-backends submenus dispatch their commands.""" + + def test_run_dispatches_install_to_setup_tui(self): + info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) + + def fake_wrapper(cb): + fake_wrapper.calls += 1 + return ("install", "qwen") if fake_wrapper.calls == 1 else None + fake_wrapper.calls = 0 + + import curses + with patch.object(curses, "wrapper", fake_wrapper), \ + patch.object(hub, "get", return_value=info), \ + patch.object(info, "setup_tui") as mk_setup: + hub.run() + mk_setup.assert_called_once_with() + + def test_run_dispatches_uninstall(self): + info = BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0) + + def fake_wrapper(cb): + fake_wrapper.calls += 1 + return ("uninstall", "qwen") if fake_wrapper.calls == 1 else None + fake_wrapper.calls = 0 + + import curses + with patch.object(curses, "wrapper", fake_wrapper), \ + patch.object(hub, "get", return_value=info), \ + patch.object(info, "uninstall") as mk_uninstall: + hub.run() + mk_uninstall.assert_called_once_with() + + def _capture_flashes(self): + flashes = [] + + def fake_flash(stdscr, text, kind="warn"): + flashes.append((text, kind)) + + return patch.object(hub.tui, "flash", fake_flash), flashes + + def test_download_models_action_flashes_hand_install_guidance(self): + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) + (checkout / "server.json").write_text(json.dumps({"models": []}), + encoding="utf-8") + missing = [{"id": "qwen", + "rel": "models/Qwen3-TTS-12Hz-0.6B-Base-GGUF"}] + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=checkout), \ + patch.object(hub.audiocpp_backend, "missing_model_entries", + return_value=missing), \ + patch.object(hub.audiocpp_backend, + "missing_model_install_guidance", + return_value=[]), \ + patch.object(hub.audiocpp_backend, "hand_install_guidance", + return_value="do it by hand") as mk_hand, \ + patch_flash: + hub._download_models_action(None) + self.assertEqual(flashes, [("do it by hand", "err")]) + mk_hand.assert_called_once() + + def test_download_models_action_flashes_ok_when_nothing_missing(self): + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) + (checkout / "server.json").write_text(json.dumps({"models": []}), + encoding="utf-8") + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=checkout), \ + patch.object(hub.audiocpp_backend, "missing_model_entries", + return_value=[]), \ + patch_flash: + hub._download_models_action(None) + self.assertEqual(len(flashes), 1) + self.assertEqual(flashes[0][1], "ok") + + def test_download_models_action_suspends_and_installs(self): + import contextlib + + @contextlib.contextmanager + def fake_suspend(scr): + yield + + with tempfile.TemporaryDirectory() as td: + checkout = Path(td) + (checkout / "server.json").write_text(json.dumps({"models": []}), + encoding="utf-8") + missing = [{"id": "qwen", "rel": "models/q"}] + guidance = [("qwen", "qwen3_tts_0_6b_base_q8_0")] + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=checkout), \ + patch.object(hub.audiocpp_backend, "missing_model_entries", + return_value=missing), \ + patch.object(hub.audiocpp_backend, + "missing_model_install_guidance", + return_value=guidance), \ + patch.object(hub.tui, "suspend", fake_suspend), \ + patch.object(hub.audiocpp_backend, "install_models") as mk, \ + patch_flash: + hub._download_models_action(None) + mk.assert_called_once_with(checkout, guidance) + self.assertEqual(len(flashes), 1) + self.assertEqual(flashes[0][1], "ok") + + def test_download_models_action_flashes_error_when_no_checkout(self): + patch_flash, flashes = self._capture_flashes() + with patch.object(hub.audiocpp_backend, "find_local_checkout", + return_value=None), patch_flash: + hub._download_models_action(None) + self.assertEqual(len(flashes), 1) + self.assertEqual(flashes[0][1], "err") + + def test_pick_backend_menu_install_lists_uninstalled_only(self): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return hub._GO_BACK + + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), + BackendInfo("faster", "faster-qwen3-tts", lambda: None, + lambda: 0)] + statuses = [BackendStatus("qwen", "qwen-tts", installed=True, + configured=True), + BackendStatus("faster", "faster-qwen3-tts", + installed=False, configured=False)] + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub.tui, "menu", fake_menu): + result = hub._pick_backend_menu(None, statuses, "Install Backend", + installed_only=False) + self.assertIsNone(result) + self.assertEqual([label for label, _ in captured["options"]], + ["faster-qwen3-tts"]) + + def test_pick_backend_menu_uninstall_lists_installed_only(self): + captured = {} + + def fake_menu(stdscr, title, options, **kwargs): + captured["options"] = options + return "qwen" + + infos = [BackendInfo("qwen", "qwen-tts", lambda: None, lambda: 0), + BackendInfo("faster", "faster-qwen3-tts", lambda: None, + lambda: 0)] + statuses = [BackendStatus("qwen", "qwen-tts", installed=True, + configured=True), + BackendStatus("faster", "faster-qwen3-tts", + installed=False, configured=False)] + with patch.object(hub, "REGISTRY", infos), \ + patch.object(hub.tui, "menu", fake_menu): + result = hub._pick_backend_menu(None, statuses, "Uninstall Backend", + installed_only=True) + self.assertEqual(result, ("uninstall", "qwen")) + self.assertEqual([label for label, _ in captured["options"]], + ["qwen-tts"]) + if __name__ == "__main__": unittest.main() diff --git a/app/tests/test_tui.py b/app/tests/test_tui.py index ce408af..49960a1 100644 --- a/app/tests/test_tui.py +++ b/app/tests/test_tui.py @@ -857,6 +857,37 @@ class CheckboxTreeTests(TuiTestCase): picked = tui.checkbox_tree(screen, "Pick models", self.FAMILIES) self.assertEqual(picked, [(0, "pkg-a")]) + def test_prechecked_selection_accepted_directly(self): + # checked= seeds the tree (modify flow): Enter alone accepts the + # pre-checked option without any key presses in between. + screen = FakeScreen(keys=[10]) + picked = tui.checkbox_tree( + screen, "Pick models", self.FAMILIES, + checked={(0, "pkg-b"), (1, "pkg-c")}) + self.assertEqual(picked, [(0, "pkg-b"), (1, "pkg-c")]) + + def test_prechecked_options_draw_as_checked(self): + screen = FakeScreen(keys=[10]) + tui.checkbox_tree(screen, "Pick models", self.FAMILIES, + checked={(0, "pkg-b")}) + texts = [text for _, _, text, _ in screen.strings] + # The pre-checked option row (indented) draws its box checked. + self.assertIn("[x] ", texts) + # ...and the family row is expanded (its options are listed). + self.assertIn("- Family one", texts) + + def test_prechecked_family_cursor_starts_on_it(self): + # Only the second family is pre-checked, so the cursor starts on it: + # Space clears then re-checks that family (the cursor never moves). + # If the cursor were still on the first family, the two Spaces would + # check then clear family one and Enter would flash instead of + # accepting anything. + screen = FakeScreen(keys=[ord(" "), ord(" "), 10]) + picked = tui.checkbox_tree( + screen, "Pick models", self.FAMILIES, + checked={(1, "pkg-c")}) + self.assertEqual(picked, [(1, "pkg-c")]) + def test_empty_families_rejected(self): with self.assertRaises(ValueError): tui.checkbox_tree(self.screen, "Pick", []) diff --git a/app/ui/hub.py b/app/ui/hub.py index 95ac06a..2ac1551 100644 --- a/app/ui/hub.py +++ b/app/ui/hub.py @@ -3,7 +3,8 @@ The hub is the single entry point for the whole workflow: it detects which backends are already set up and offers to convert the input directory with -one of them, set up a new backend, or configure an existing one. +one of them, or install/configure/remove a backend via the "Configure +backends" menu. Each backend's setup wizard runs in its own curses session, so the hub collects a "command" inside its own wrapper, returns to the plain terminal, and then dispatches — no nested curses sessions. @@ -70,14 +71,14 @@ def run() -> int: kind = command[0] if kind == "quit": return 0 - if kind == "setup": + if kind in ("install", "configure"): info = get(command[1]) if info is not None: info.setup_tui() - elif kind == "configure": + elif kind == "uninstall": info = get(command[1]) - if info is not None and command[2] < len(info.configure_actions): - info.configure_actions[command[2]].run() + if info is not None: + info.uninstall() elif kind == "convert": _dispatch_conversion(command[1], command[2]) elif kind == "server": @@ -88,14 +89,13 @@ def _hub_menu(stdscr) -> Optional[tuple]: """Show the main menu; return a command tuple, or None to quit.""" while True: statuses = detect_all() - options = [("Set up a backend", "setup")] + options = [("Configure backends", "configure_backends")] # Converting works against an external (remote) server too, but # configuring one and starting/stopping its servers need it on # this machine. if any(st.installed or st.running for st in statuses): options.insert(0, ("Convert books", "convert")) if any(st.installed for st in statuses): - options.append(("Configure a backend", "configure")) options.append(("Start/Stop Backend Servers", "server")) options.append(("Settings", "settings")) options.append(("Quit", "quit")) @@ -109,12 +109,8 @@ def _hub_menu(stdscr) -> Optional[tuple]: cmd = _convert_menu(stdscr, statuses) if cmd is not None: return cmd - elif choice == "setup": - cmd = _setup_menu(stdscr, statuses) - if cmd is not None: - return cmd - elif choice == "configure": - cmd = _configure_menu(stdscr, statuses) + elif choice == "configure_backends": + cmd = _configure_backends_menu(stdscr, statuses) if cmd is not None: return cmd elif choice == "server": @@ -125,48 +121,130 @@ def _hub_menu(stdscr) -> Optional[tuple]: _settings_menu(stdscr) -def _setup_menu(stdscr, statuses) -> Optional[tuple]: - """Pick a backend to set up. Returns ("setup", key) or None to go back.""" - options = [(info.label, info.key) for info in REGISTRY] - choice = tui.menu(stdscr, "Set up a backend", options, - back_value=_GO_BACK, - help_lines=["Clone/build/install a backend so you can " - "convert with it."], - table_title="Backend status", - table_rows=_status_rows(statuses), - notice_lines=_notice_lines()) - if choice is _GO_BACK or choice is None: - return None - return ("setup", choice) - +def _configure_backends_menu(stdscr, statuses) -> Optional[tuple]: + """One flat menu of backend setup/configure/cleanup actions. -def _configure_menu(stdscr, statuses) -> Optional[tuple]: - """Pick an installed backend and one of its configure actions.""" + Replaces the old "Set up a backend" + "Configure a backend" pair with a + single screen whose options are populated from the detected statuses: + install (any uninstalled backend), configure each installed backend, + download/delete audio.cpp models (when a server.json references models + on/off disk), and uninstall. Each option returns a command tuple that + ``run`` dispatches after the curses session ends. + """ by_key = {st.key: st for st in statuses} installed = [info for info in REGISTRY if by_key.get(info.key) is not None and by_key[info.key].installed] - if not installed: - tui.flash(stdscr, "No backend is installed yet — use 'Set up a " - "backend' first.") + options: list = [] + if any(info.key not in by_key or not by_key[info.key].installed + for info in REGISTRY): + options.append(("Install Backend", "install")) + for info in installed: + options.append((f"Configure {info.label}", ("configure", info.key))) + + audiocpp_status = by_key.get("audiocpp") + missing = [] + if audiocpp_status is not None and audiocpp_status.installed: + checkout = audiocpp_backend.find_local_checkout() + server_json = checkout / "server.json" if checkout else None + if server_json is not None and server_json.exists(): + missing = audiocpp_backend.missing_model_entries(server_json) + if missing: + options.append(("Download Missing Models (audio.cpp)", + "download_models")) + + if installed: + options.append(("Uninstall Backend", "uninstall")) + + choice = tui.menu( + stdscr, "Configure backends", options, + back_value=_GO_BACK, + help_lines=["Install, configure, or remove a TTS backend."], + table_title="Backend status", + table_rows=_status_rows(statuses), + notice_lines=_notice_lines()) + if choice is _GO_BACK or choice is None: return None - options = [(info.label, info.key) for info in installed] - key = tui.menu(stdscr, "Configure a backend", options, + if choice == "install": + return _pick_backend_menu(stdscr, statuses, "Install Backend", + installed_only=False) + if choice == "uninstall": + return _pick_backend_menu(stdscr, statuses, "Uninstall Backend", + installed_only=True) + if choice == "download_models": + _download_models_action(stdscr) + return None + kind, key = choice + return (kind, key) + + +def _pick_backend_menu(stdscr, statuses, title: str, + installed_only: bool) -> Optional[tuple]: + """Pick a backend for the Install/Uninstall actions. + + With INSTALLED_ONLY False every backend is listed (the install list); + with it True only the currently-installed ones are (the uninstall list). + Returns ``(action, key)`` where action is "install" or "uninstall". + """ + by_key = {st.key: st for st in statuses} + if installed_only: + candidates = [info for info in REGISTRY + if by_key.get(info.key) is not None + and by_key[info.key].installed] + else: + candidates = [info for info in REGISTRY + if by_key.get(info.key) is None + or not by_key[info.key].installed] + if not candidates: + tui.flash(stdscr, "No backends to list here.") + return None + options = [(info.label, info.key) for info in candidates] + key = tui.menu(stdscr, title, options, back_value=_GO_BACK, table_title="Backend status", table_rows=_status_rows(statuses), notice_lines=_notice_lines()) if key is _GO_BACK or key is None: return None - info = get(key) - actions = info.configure_actions - choice = tui.menu( - stdscr, f"Configure {info.label}", - [(action.label, index) for index, action in enumerate(actions)], - back_value=_GO_BACK) - if choice is _GO_BACK or choice is None: - return None - return ("configure", key, choice) + action = "uninstall" if installed_only else "install" + return (action, key) + + +def _download_models_action(stdscr) -> None: + """Run the "Download Missing Models (audio.cpp)" action inside the TUI. + + Computes the missing models; when they map to install commands it + suspends curses to stream the downloads, then flashes a result — instead + of silently returning to the main menu. When the checkout/server.json is + missing, nothing is missing, or the models do not map to an install + command, it flashes an explanatory notice (the latter explaining how to + install each model by hand). + """ + checkout = audiocpp_backend.find_local_checkout() + if checkout is None: + tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp " + "first.", "err") + return + server_json = checkout / "server.json" + if not server_json.exists(): + tui.flash(stdscr, "No audio.cpp server.json found — configure " + "audio.cpp first.", "err") + return + missing = audiocpp_backend.missing_model_entries(server_json) + if not missing: + tui.flash(stdscr, "Every configured audio.cpp model is already " + "downloaded.", "ok") + return + guidance = audiocpp_backend.missing_model_install_guidance( + checkout, missing) + if not guidance: + tui.flash(stdscr, audiocpp_backend.hand_install_guidance( + checkout, missing), "err") + return + with tui.suspend(stdscr): + audiocpp_backend.install_models(checkout, guidance) + tui.flash(stdscr, "Model download finished. See the output above for " + "any warnings.", "ok") def _status_mark(status: Optional[BackendStatus]) -> Tuple[str, str, str]: @@ -244,7 +322,7 @@ def _convert_menu(stdscr, statuses) -> Optional[tuple]: st, True)) if not entries: tui.flash(stdscr, "No backend is ready to convert with yet — use " - "'Set up a backend' first.") + "'Configure backends' first.") return None builders = {} for key, _label, st, remote in entries: @@ -439,7 +517,7 @@ def _audiocpp_fields(stdscr, api_url: Optional[str] = None) -> Optional[tuple]: server_json = checkout / "server.json" if checkout else None if not (server_json and server_json.exists()): tui.flash(stdscr, "No audio.cpp server.json found — run " - "'Set up a backend' first.") + "'Configure backends' first.") return None try: data = json.loads(server_json.read_text(encoding="utf-8")) @@ -1108,8 +1186,8 @@ def _server_menu(stdscr, statuses) -> Optional[tuple]: # file), so listing it would dead-end. candidates = [st for st in statuses if st.installed] if not candidates: - tui.flash(stdscr, "No backend is installed yet — use 'Set up a " - "backend' first.") + tui.flash(stdscr, "No backend is installed yet — use " + "'Configure backends' first.") return None options = [(st.label, st.key) for st in candidates] key = tui.menu(stdscr, "Start / Stop a server", options, @@ -1130,7 +1208,7 @@ def _server_actions(stdscr, status) -> Optional[tuple]: specs = status.servers if not specs: tui.flash(stdscr, f"{status.label} has no server configured. " - "Run 'Set up a backend' first.") + "Run 'Configure backends' first.") return None if len(specs) == 1: spec = specs[0] diff --git a/app/ui/tui.py b/app/ui/tui.py index 83ec43d..95130cf 100644 --- a/app/ui/tui.py +++ b/app/ui/tui.py @@ -1113,7 +1113,8 @@ def browse_directory(scr, title: str, def checkbox_tree(scr, title: str, families: List[dict], footer: Optional[str] = None, expand_all: bool = False, - back_value: object = None) -> List[Tuple[int, str]]: + back_value: object = None, + checked: Optional[set] = None) -> List[Tuple[int, str]]: """Pick model families and packages from an expandable tree. FAMILIES is a list of dicts (one per family) shaped like:: @@ -1134,9 +1135,12 @@ def checkbox_tree(scr, title: str, families: List[dict], cursor. Enter returns the flat list of (family_index, option_key) pairs for every checked option, in tree order; at least one checked option is required. Nothing is checked by default, and with - EXPAND_ALL every family starts expanded. A "[recommended]" tag is - shown only when a - family has more than one option — a single option needs no tag. + EXPAND_ALL every family starts expanded. CHECKED (a set of + (family_index, option_key) pairs) pre-checks those options instead, + expanding every family that holds a checked option and placing the + cursor on the first such family — the "modify an existing config" + entry point. A "[recommended]" tag is shown only when a family has + more than one option — a single option needs no tag. Family and option rows are left-justified like a DOS list. Esc (or 'q') aborts the wizard unless BACK_VALUE is given (not None), in which case either key returns it so the caller can fall back a @@ -1148,9 +1152,14 @@ def checkbox_tree(scr, title: str, families: List[dict], "Enter = accept Esc = cancel") frame = Frame(scr, title, footer) expanded = {index for index in range(len(families))} if expand_all else set() - checked = set() # (family_index, option_key) + checked = set(checked or ()) # (family_index, option_key) - expanded.add(0) + if checked: + for index, _option_key in checked: + expanded.add(index) + expanded.add(0) + else: + expanded.add(0) def family_checked(index: int) -> bool: return any(pair[0] == index for pair in checked) @@ -1170,9 +1179,17 @@ def checkbox_tree(scr, title: str, families: List[dict], nodes.append(("option", index, option["key"])) return nodes + first_checked = min((index for index, _option_key in checked), + default=None) cursor = 0 while True: nodes = visible_nodes() + if first_checked is not None: + for position, node in enumerate(nodes): + if node[0] == "family" and node[1] == first_checked: + cursor = position + break + first_checked = None cursor = max(0, min(cursor, len(nodes) - 1)) frame.rows = [] for node in nodes: |