diff options
Diffstat (limited to 'app/backends/audiocpp.py')
| -rwxr-xr-x | app/backends/audiocpp.py | 465 |
1 files changed, 361 insertions, 104 deletions
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() |