"""The audio.cpp setup wizard: TUI screens, task lanes, CLI entry points.""" import argparse import json import sys from pathlib import Path from typing import Callable, Dict, List, Optional, Tuple # Alias kept on this module: main()'s tty check and its tests patch it # here. from backends.setup import interactive as _interactive from backends import common from backends.common import ( APP_DIR, PROMPT_TEXT_FILENAME, VOICES_DIR, find_wav_files, read_prompt_text, resolve_wav_dir_arg, write_prompt_text, ) from ui import taskview, tui from . import build as _build from . import configsync as _configsync from . import models as _models from . import prebuilt as _prebuilt from . import voices as _voices from .catalog import (_backend_options, build_model_entry, build_server_config, detect_backend, entry_model_path, hosting_task, load_model_catalog, load_server_config, MIOTTS_CODEC_DISPLAY_NAME, MIOTTS_CODEC_INSTALL_ID, package_dir_options, server_config_selections, apply_entry_session_options) from .constants import (AUDIOCPP_DIR_NAME, AUDIOCPP_GIT_URL, BACKENDS, DEFAULT_HOST, TASK_TTS, TASK_VDES) _GO_BACK = object() def _flag_build_mode(args: argparse.Namespace, backend: str) -> str: """Resolve the ``--prebuilt`` flag into a concrete build mode. ``auto`` (the default) downloads the prebuilt release when this platform/backend has one and otherwise builds from source; ``yes`` forces the download, ``no`` forces the source build. Only consulted when a build/install is actually pending. """ choice = getattr(args, "prebuilt", "auto") if choice == "no": return "source" if choice == "yes" or _prebuilt.prebuilt_supported(backend): return "prebuilt" return "source" class _GoBack(Exception): """Internal signal: Esc was pressed inside one of a screen's sub-prompts. The wizard drives a stack of screens via ``tui.Wizard``. Helpers that ask several questions through callbacks (the task/id pickers inside ``_build_entries``, the transcription plan, the download prompt) cannot themselves return the wizard's ``BACK`` sentinel, so they convert the ``_GO_BACK`` value passed to each widget into this exception. The screen that invoked the helper catches it and returns ``tui.Wizard.BACK``, which pops back to the previous screen. Esc on the first screen aborts the whole wizard. """ class _TuiError(Exception): """A fatal error raised from inside the TUI wizard. The message is reported to stderr after the terminal is restored; the process exits with code 2 (matching a parser error). """ def _build_entries(family_keys: List[str], chosen: Dict[str, List[dict]], catalog_by_family: Dict[str, dict], task_picker: Callable[[str], str], known_tasks: Optional[Dict[Tuple[str, str], str]] = None ) -> Tuple[List[dict], List[str], List[Tuple[str, str]], List[Tuple[str, str]], List[str], bool]: """Build server.json model entries from the selected families/packages. TASK_PICKER is called for each design package to choose vdes/tts. KNOWN_TASKS maps ``(family, target_directory)`` to a previously-stored task ("tts" or "vdes") so a modify run preserves how a design package was hosted instead of re-asking. Each entry's server id is its package ``target_directory`` (flattened to a token), so packages from the same family never collide; an id that does collide (across families) is auto-suffixed without prompting. Entry paths come from the catalog (``entry_model_path``): normally ``models/``, or the package's first GGUF file when the package ships several GGUFs into one directory (audio.cpp refuses multi-GGUF directories). COMPANION_GUIDANCE carries the companion packages hosted models require but the TTS catalog never offers (MioCodec for MioTTS). Returns (model_entries, entry_ids, install_guidance, companion_guidance, design_entry_ids, include_clone). """ model_entries: List[dict] = [] entry_ids: List[str] = [] install_guidance: List[Tuple[str, str]] = [] companion_guidance: List[Tuple[str, str]] = [] design_entry_ids: List[str] = [] include_clone = False for family in family_keys: entry = catalog_by_family[family] include_clone = include_clone or entry["clone_capable"] for opt in chosen[family]: if opt["design"]: task = known_tasks.get((family, opt["target_directory"])) \ if known_tasks else None if task is None: task = task_picker(opt["install_id"]) else: # Clone-only families (Chatterbox, Confucius4-TTS, # Echo-TTS) reject plain-TTS sessions, so they are hosted # with task "clon"; everything else keeps "tts". task = hosting_task(entry) base_id = opt["target_directory"].replace("/", "-") model_id = base_id if model_id in entry_ids: n = 2 while f"{base_id}-{n}" in entry_ids: n += 1 model_id = f"{base_id}-{n}" entry_ids.append(model_id) model_entries.append(build_model_entry( family, model_id, entry_model_path(entry, opt["target_directory"]), task=task)) install_guidance.append((entry["display_name"], opt["install_id"])) if task == TASK_VDES: design_entry_ids.append(model_id) if any(str(e.get("family")) == "miotts" for e in model_entries): companion_guidance.append( (MIOTTS_CODEC_DISPLAY_NAME, MIOTTS_CODEC_INSTALL_ID)) return (model_entries, entry_ids, install_guidance, companion_guidance, design_entry_ids, include_clone) def _write_and_advise(audiocpp_dir: Path, wav_dir: Optional[Path], output_path: Path, model_entries: List[dict], install_guidance: List[Tuple[str, str]], host: str, port: int, backend: str, lazy_load: bool, transcripts: Dict[str, str], write_prompt: bool) -> None: """Console phase shared by both UI modes: write files, print summary. After a successful run the console output is the path of the written server.json. The model install commands (and optional automatic download) are handled separately by _install_models, called by both UI modes once the user has decided whether to download. Before the document is written, apply_entry_session_options bakes in the per-entry session options heavy families need (MioTTS's codec path; a VoxCPM AudioVAE encoder capacity sized to the voice directory's longest reference), reported as one summary line. """ voice_dir: Optional[str] = None if transcripts: if write_prompt: prompt_path = wav_dir / PROMPT_TEXT_FILENAME write_prompt_text(wav_dir, transcripts) print(f"[OK] Wrote {prompt_path}") voice_dir = str(wav_dir.resolve()) configured = apply_entry_session_options(model_entries, wav_dir, audiocpp_dir) if configured: print(f"[OK] Added family session options to: {', '.join(configured)}") server_config = build_server_config( host=host, port=port, backend=backend, lazy_load=lazy_load, model_entries=model_entries, voice_dir=voice_dir) with output_path.open("w", encoding="utf-8") as handle: json.dump(server_config, handle, indent=2, ensure_ascii=False) handle.write("\n") count = len(model_entries) print(f"Wrote {output_path.resolve()} with {count} " f"{'entry' if count == 1 else 'entries'}.") def _plan_from_mode(mode: str, wav_files: list, existing: Dict[str, str]) -> dict: """Build the transcription PLAN for the chosen form MODE. The plan dict is what ``voices._transcribe`` consumes: "missing" carries the .wavs lacking a transcript plus the existing mapping, "all" re-transcribes everything; both reuse the mapping read while applying the form. """ if mode == "missing": missing = [wav for wav in wav_files if not existing.get(wav.stem, "").strip()] return {"mode": mode, "missing": missing, "existing": dict(existing)} return {"mode": mode, "missing": [], "existing": dict(existing)} def _wizard(stdscr, args: argparse.Namespace, parser: argparse.ArgumentParser ) -> Optional[dict]: """Run every TUI screen; return the collected settings, or None to abort. The wizard has two screens: the model tree ("Select TTS Models to Host") and one combined configuration form (backend choice when it is ambiguous, build offer when needed, clone-voice directory, transcription plan, model download/defaults/cleanup), laid out like the Generate Audiobooks screen — every option appears on one screen, and options that do not apply are hidden instead of asked separately. The bind host is always 127.0.0.1 and the port comes from AUDIOCPP_API_URL in app/converter/config.py (the Settings screen), so neither is ever asked. Esc on the first screen aborts the whole wizard; Esc on the form pops back to the model tree. """ s: dict = {} def resolve_checkout(audiocpp_dir: Path) -> None: """Validate the audio.cpp checkout and populate the wizard state ``s``.""" audiocpp_dir = Path(audiocpp_dir).resolve() try: catalog = load_model_catalog(audiocpp_dir) except NotADirectoryError as exc: raise _TuiError(str(exc)) from exc if not catalog: raise _TuiError(f"No TTS model families found in " f"{audiocpp_dir}/model_specs; check the " "checkout is up to date") catalog_by_family = {entry["family"]: entry for entry in catalog} output_path = args.output if args.output is not None \ else audiocpp_dir / "server.json" # Modify flow: an existing server.json seeds the wizard's screens # instead of being overwritten from scratch (an explicit --force # still starts fresh). existing_config = load_server_config(output_path) \ if not args.force else None if existing_config is not None: existing_selected, existing_tasks = \ server_config_selections(existing_config, catalog) else: existing_selected, existing_tasks = {}, {} s.update({ "audiocpp_dir": audiocpp_dir, "catalog": catalog, "catalog_by_family": catalog_by_family, "output_path": output_path, "existing_config": existing_config, "existing_selected": existing_selected, "existing_tasks": existing_tasks, "existing_backend": existing_config.get("backend") if existing_config else None, "existing_voice_dir": existing_config.get("voice_dir") if existing_config else None, "detected_backend": detect_backend(audiocpp_dir), }) def _families_from_flag() -> None: requested = [f.strip() for f in args.families.split(",") if f.strip()] unknown = [f for f in requested if f not in s["catalog_by_family"]] if unknown: raise _TuiError( f"Unknown family in --families: {', '.join(unknown)}. " f"Available: {', '.join(s['catalog_by_family'])}") chosen: Dict[str, List[dict]] = {} family_keys: List[str] = [] for family in requested: if family not in family_keys: family_keys.append(family) chosen[family] = [opt for opt in package_dir_options( s["catalog_by_family"][family]) if opt["recommended"]] s["chosen"] = chosen s["family_keys"] = family_keys def _compute_entries() -> None: # Design task menu. Esc raises _GoBack, which the caller turns into # Wizard.BACK (the design prompts are grouped: Esc returns to the # families tree). def task_picker(install_id: str) -> str: result = tui.menu( stdscr, f"How should the '{install_id}' package be hosted?", [ ("design (vdes) - describe the voice with " "--instructions", TASK_VDES), ("tts - normal synthesis", TASK_TTS), ], default_index=0, back_value=_GO_BACK) if result is _GO_BACK: raise _GoBack() return result model_entries, entry_ids, install_guidance, companion_guidance, \ design_entry_ids, include_clone = _build_entries( s["family_keys"], s["chosen"], s["catalog_by_family"], task_picker, known_tasks=s["existing_tasks"]) s.update({ "model_entries": model_entries, "entry_ids": entry_ids, "install_guidance": install_guidance, "companion_guidance": companion_guidance, "design_entry_ids": design_entry_ids, "include_clone": include_clone, }) def _finalize() -> dict: host = DEFAULT_HOST port = _configsync.config_port() backend = s["backend"] # Build decision: --build-backend builds when no single-backend # binary was detected; a plain --backend or a detected build never # rebuilds; the interactive answer comes from the form. build = s["build"] return { "audiocpp_dir": s["audiocpp_dir"], "catalog": s["catalog"], "catalog_by_family": s["catalog_by_family"], "output_path": s["output_path"], "family_keys": s["family_keys"], "chosen": s["chosen"], "model_entries": s["model_entries"], "entry_ids": s["entry_ids"], "install_guidance": s["install_guidance"], "companion_guidance": s["companion_guidance"], "design_entry_ids": s["design_entry_ids"], "include_clone": s["include_clone"], "host": host, "port": port, "backend": backend, "build": build, "build_mode": s.get("build_mode"), "prebuilt_forced": False, "lazy_load": True, "sync_port": None, "wav_dir": s["wav_dir"], "plan": s["plan"], "download": s["download"], "delete_unused": s["delete_unused"], "unused_entries": s["unused_entries"], } def screen_families(): """Pick TTS model families and packages (the modify tree).""" tree_families = _models._build_tree_families(s["catalog"]) # Modify flow: pre-check the models an existing server.json hosts, # so the tree opens as a "modify" list rather than a fresh one. checked_set = set() for family, dirs in s["existing_selected"].items(): if family not in s["catalog_by_family"]: continue family_index = s["catalog"].index(s["catalog_by_family"][family]) valid_dirs = {opt["target_directory"] for opt in package_dir_options( s["catalog_by_family"][family])} for target in dirs: if target in valid_dirs: checked_set.add((family_index, target)) picked = tui.checkbox_tree( stdscr, "Select TTS Models to Host", tree_families, expand_all=args.all_packages, back_value=_GO_BACK, checked=checked_set, start_on_buttons=True) if picked is _GO_BACK: return tui.Wizard.BACK chosen: Dict[str, List[dict]] = {} family_keys: List[str] = [] for family_index, option_key in picked: family = s["catalog"][family_index]["family"] if family not in chosen: chosen[family] = [] family_keys.append(family) chosen[family].append(option_key) for family in list(chosen): keyed = {opt["target_directory"]: opt for opt in package_dir_options( s["catalog_by_family"][family])} chosen[family] = [keyed[key] for key in chosen[family]] s["chosen"] = chosen s["family_keys"] = family_keys return screen_config def _after_families(): if args.families is not None: _families_from_flag() return screen_config return screen_families def _field_val(fields_list, key, default=None): return next((f["value"] for f in fields_list if f.get("key") == key), default) def _apply_form(result: dict) -> dict: """Fold the form's answers into the settings and finalize.""" # Backend/build: the interactive combination. A backend whose # binary already exists (switching to an already-built one) hides # the build row — honor that by re-checking at apply time. if s["backend"] is None: s["backend"] = result["backend"] if _build.built_server_binary(s["audiocpp_dir"], s["backend"]) is not None: s["build"] = False s["build_mode"] = None else: # The form reports every field's value, including hidden # ones, so the mode is chosen by whether this backend has # a prebuilt asset at all — not by which field was shown. # Only the visible field's answer is meaningful: the # choice field when a prebuilt asset exists, otherwise the # plain build question. if _prebuilt.prebuilt_supported(s["backend"]): mode = result.get("build_mode") if mode not in ("prebuilt", "source", "skip"): mode = "prebuilt" else: mode = "source" if bool(result.get("build")) else "skip" s["build_mode"] = mode s["build"] = mode in ("prebuilt", "source") # Clone-voice directory: only meaningful for clone-capable picks. if args.input_dir is not None: s["wav_dir"] = args.input_dir elif s["include_clone"]: raw = result.get("wav_dir") s["wav_dir"] = Path(raw) if raw else None else: s["wav_dir"] = None # Transcription plan (transcription itself runs in the tail). s["plan"] = None if s["include_clone"]: wav_files = find_wav_files(Path(s["wav_dir"])) \ if s["wav_dir"] is not None else [] prompt_path = Path(s["wav_dir"]) / PROMPT_TEXT_FILENAME \ if s["wav_dir"] is not None else None existing = {} if prompt_path is not None and prompt_path.exists() \ and not args.force: existing = read_prompt_text(prompt_path) mode = result.get("transcription") if mode not in ("missing", "all"): mode = "missing" s["plan"] = _plan_from_mode(mode, wav_files, existing) s["download"] = bool(result.get("download")) and ( _models.download_applicable( s["audiocpp_dir"], s["model_entries"], companions=s.get("companion_guidance"))) s["delete_unused"] = bool(result.get("delete_unused")) \ and bool(s["unused_entries"]) return _finalize() def screen_config(): """One combined configuration screen for everything else. The Generate-audiobooks-style form replaces the old one-question- per-screen chain (host, port, port sync, backend, build offer, wav directory, transcription plan, delete unused, download). Rows whose question does not apply are hidden rather than skipped silently. Esc or Cancel pops back to the model tree. """ try: _compute_entries() except _GoBack: return tui.Wizard.BACK # Backend: pinned by a flag or an existing build when possible; # only otherwise does it become a form question. Not built for any # pinned backend yet still asks — even on a modify run, so a user # who declined the build the first time is never stranded without # a way to build from the TUI. if args.build_backend is not None: s["backend"] = args.build_backend s["build"] = s["detected_backend"] is None s["build_mode"] = _flag_build_mode(args, s["backend"]) \ if s["build"] else None elif args.backend is not None: s["backend"] = args.backend s["build"] = False s["build_mode"] = None elif s["detected_backend"] is not None: # Already built: use the detected backend, no menu, no build. s["backend"] = s["detected_backend"] s["build"] = False s["build_mode"] = None else: s["backend"] = None # decided by the form s["build"] = None s["build_mode"] = None # Clone-voice directory seed: the voice_dir recorded by the # server.json being modified, else the project voices/ dir (the # same default the --wavs flag documents). No auto-detection: the # field must never start blank. wav_start = None if s["include_clone"]: if isinstance(s["existing_voice_dir"], str) \ and s["existing_voice_dir"]: wav_start = Path(s["existing_voice_dir"]) else: wav_start = VOICES_DIR s["wav_dir"] = wav_start fields: List[dict] = [] if s["backend"] is None: options, default_index = _backend_options(None) default_backend = options[default_index][1] if s["existing_backend"] in BACKENDS: default_backend = next( (value for _label, value in options if value == s["existing_backend"]), default_backend) def needs_build(fs) -> bool: chosen = _field_val(fs, "backend", default_backend) return _build.built_server_binary( s["audiocpp_dir"], chosen) is None fields.append({ "key": "backend", "label": "Inference backend", "kind": "choice", "value": default_backend, "choices": options, }) # How to get the binary: macOS and Windows have upstream # release binaries (no toolchain needed — see # backends.audiocpp.prebuilt), everything else builds from # source. HIP on Windows has no release asset, so it keeps # the plain build question. fields.append({ "key": "build_mode", "label": "Get audiocpp_server", "kind": "choice", "value": "prebuilt", "choices": [ ("Download prebuilt server (recommended)", "prebuilt"), ("Build from source", "source"), ("Skip for now", "skip"), ], "visible": lambda fs: ( needs_build(fs) and _prebuilt.prebuilt_supported( _field_val(fs, "backend", default_backend))), }) fields.append({ "key": "build", "label": "Build audiocpp_server now?", "kind": "bool", "value": True, "visible": lambda fs: ( needs_build(fs) and not _prebuilt.prebuilt_supported( _field_val(fs, "backend", default_backend))), }) wav_field = { "key": "wav_dir", "label": "Voice clone .wav directory", "kind": "dir", "value": Path(wav_start) if wav_start else None, "info": common.wav_dir_info, "preview": common.wav_dir_preview, "visible": lambda fs: bool(s["include_clone"]), } fields.append(wav_field) # Voice transcript handling: a plain in-place toggle, always # offered whenever any clone-capable model is hosted (no # dependency on what the picked directory currently holds). fields.append({ "key": "transcription", "label": "Voice transcripts", "kind": "toggle", "value": "missing", "choices": [("Transcribe new voices", "missing"), ("Re-transcribe all voices", "all")], "visible": lambda fs: bool(s["include_clone"]), }) if _models.download_applicable( s["audiocpp_dir"], s["model_entries"], companions=s.get("companion_guidance")): fields.append({ "key": "download", "label": "Download the selected models automatically?", "kind": "bool", "value": True, }) new_paths = {entry["path"] for entry in s["model_entries"]} s["unused_entries"] = _models.unused_installed_entries( s["output_path"], new_paths) \ if s["existing_config"] is not None else [] s["delete_unused"] = False if s["unused_entries"]: count = len(s["unused_entries"]) fields.append({ "key": "delete_unused", "label": f"Delete {count} unused downloaded model " f"{'entry' if count == 1 else 'entries'} from disk?", "kind": "bool", "value": False, }) result = tui.form( stdscr, "Configure audio.cpp", fields, buttons=("Continue", "Cancel"), start_on_buttons=True, back_value=tui.Wizard.BACK) if result is tui.Wizard.BACK: return tui.Wizard.BACK return _apply_form(result) # First screen: resolve the checkout directly when it already exists # (the modify flow), so the wizard starts on a real screen. When no # checkout exists, clone it into ./app/audio.cpp (streaming inside the # TUI task view, not by dropping to the console) without asking, then # continue the same way. audiocpp_dir = _build.find_local_checkout() if audiocpp_dir is None: target = APP_DIR / AUDIOCPP_DIR_NAME # The ggml build patches are deliberately NOT applied here: they # belong to the source-build path (build_audiocpp applies them # right before building), and a prebuilt install checks out the # release tag, which the patches may not fit. rc = taskview.run_steps(stdscr, "Clone audio.cpp", [ taskview.TaskStep( f"Cloning audio.cpp into {target}", lambda emit, cancel: common.git_clone( AUDIOCPP_GIT_URL, target, emit=emit, cancel=cancel)), ]) if rc == 130: # Cancelled from the task view: abort the wizard quietly. return None if rc != 0: raise _TuiError( f"audio.cpp setup step failed (exit {rc}). Clone " f"audio.cpp manually: git clone " f"{AUDIOCPP_GIT_URL} {target}, then re-run") audiocpp_dir = target resolve_checkout(audiocpp_dir) first = _after_families() return tui.Wizard().run(first) def _execute_lanes(settings: dict, args: argparse.Namespace) -> List[taskview.TaskLane]: """Build the ordered setup steps for the in-TUI task view, per lane. The same work ``_execute`` runs on the console, split into two lanes so the view can run the build in one pane while configuring and downloading models in the other (both progress bars visible at once). The install lane exists only when ``settings["build"]`` is set: it downloads the prebuilt release (``settings["build_mode"] == "prebuilt"``) or builds from source, per the user's choice. The models lane always exists (transcribe → write server.json → download/print commands). Shared results (the transcription mapping) travel through a small closure dict scoped to the models lane. Each step's ``work(emit, cancel)`` returns its exit code; subprocess steps stream through EMIT and abort on CANCEL, while print()-based steps are captured by the view's stdout routing. """ audiocpp_dir = settings["audiocpp_dir"] state: dict = {} build = settings.get("build") lanes: List[taskview.TaskLane] = [] if build: mode = settings.get("build_mode") or "source" forced = bool(settings.get("prebuilt_forced")) def source_build_step(emit, cancel) -> int: """Build from source, warning-and-continue like the setup.""" rc = _build.build_audiocpp(audiocpp_dir, settings["backend"], emit=emit, cancel=cancel) if rc == 124: print("[WARNING] build went silent and was stopped; the " "server.json was still written — build " "audiocpp_server manually before starting it") elif rc != 0: print(f"[WARNING] build exited with code {rc}; the " "server.json was still written — build " "audiocpp_server manually before starting it") else: print("[OK] build complete") return rc if mode == "prebuilt": def build_step(emit, cancel): rc = _prebuilt.install_prebuilt( audiocpp_dir, settings["backend"], emit=emit, cancel=cancel) if rc == 0: print("[OK] prebuilt audiocpp_server installed") return 0 if rc == 130 or forced: # A cancelled download, or one the user explicitly # forced with --prebuilt yes: fail fast. return rc # The usual failure is GitHub's API rate limit — recover # in place instead of leaving the setup half-installed. print(f"[WARNING] prebuilt download failed (exit {rc}); " "falling back to a source build...") return source_build_step(emit, cancel) build_title = f"Install audiocpp_server ({settings['backend']})" else: build_step = source_build_step build_title = f"Build audiocpp_server ({settings['backend']})" lanes.append(taskview.TaskLane( "Build", [taskview.TaskStep(build_title, build_step)])) def transcribe(emit, cancel): args.input_dir = settings["wav_dir"] if settings["include_clone"] and args.input_dir is not None: transcripts, write_prompt = _voices._transcribe( args, plan=settings["plan"], cancel=cancel) elif args.input_dir is not None: print(f"[WARNING] Ignoring {args.input_dir}: no clone-capable " "family selected, so voice presets are not used") transcripts, write_prompt = {}, False else: transcripts, write_prompt = {}, False state["transcripts"] = transcripts state["write_prompt"] = write_prompt # A blank transcript means the affected clone voices cannot work: # transcript-conditioned families (Qwen3-TTS Base ICL) reject every # request without it. Report the step as failed instead of a silent # [OK]; the setup continues (warn-and-continue) and still writes # server.json and prompt_text. blank = sorted(name for name, text in transcripts.items() if not text.strip()) if blank: print("[ERROR] No transcript for: " + ", ".join(blank)) print("[ERROR] Those clone voices will NOT work until prompt_text " "carries an accurate transcript for each one; see the " "summary below for how to fix them by hand.") return 1 return 0 def write(emit, cancel): _write_and_advise( audiocpp_dir, settings["wav_dir"], settings["output_path"], settings["model_entries"], settings["install_guidance"], settings["host"], settings["port"], settings["backend"], settings["lazy_load"], state["transcripts"], state["write_prompt"]) # Delete-unused cleanup (modify flow): remove the already-downloaded # models the new selection dropped. The regenerated server.json # already only lists the kept entries. if settings.get("delete_unused"): removed = _models.delete_model_files(settings["output_path"], settings["unused_entries"]) print(f"[OK] Deleted {removed} unused model " f"{'entry' if removed == 1 else 'entries'} from disk.") _voices.print_empty_transcript_warning(state["transcripts"]) return 0 def install(emit, cancel): _models._install_models(audiocpp_dir, settings["install_guidance"], settings["download"], emit=emit, cancel=cancel, model_entries=settings["model_entries"], companions=settings.get("companion_guidance")) _build._print_launch_hint(audiocpp_dir, settings["output_path"]) return 0 # Everything already on disk: the install step just reports it, so the # step title says so instead of promising a download. everything_installed = bool(settings["model_entries"]) \ and _models._all_models_present(audiocpp_dir, settings["model_entries"]) if everything_installed: install_title = "Verify models" elif settings.get("download"): install_title = "Download models" else: install_title = "Print model install commands" lanes.append(taskview.TaskLane( "Configure & download", [taskview.TaskStep("Transcribe reference voices", transcribe), taskview.TaskStep("Write server.json & sync config", write), taskview.TaskStep(install_title, install)])) return lanes def _execute_steps(settings: dict, args: argparse.Namespace) -> List[taskview.TaskStep]: """The ordered setup steps for the sequential console path. The lanes ``_execute_lanes`` builds, flattened into one ordered list (build first, then transcribe → write → download), so the console tail is byte-identical to the pre-lanes behavior. """ steps: List[taskview.TaskStep] = [] for lane in _execute_lanes(settings, args): steps.extend(lane.steps) return steps def _execute(settings: dict, args: argparse.Namespace) -> int: """Shared console tail: build, sync, transcribe, write, install, advise. Runs after the TUI wizard returns (or after _collect_from_flags for a non-interactive run): the terminal is plain, so subprocess output and transcription progress appear normally. The same work as ``_execute_steps``, run with no emit (console streaming). """ return taskview.run_steps_inline(_execute_steps(settings, args)) def setup_screen(stdscr) -> int: """Run the setup wizard on an existing curses screen (the hub's). The hub drives this as one screen of its own ``tui.Wizard`` stack, so Esc on the wizard's first screen simply returns here and the hub pops back to the menu that launched it. The setup tail (build, transcribe, write, download) runs inside the TUI task view on this same screen, so the hub's curses session stays intact and the user sees per-step status and progress instead of being dropped to the console. On a fresh install the build and the model setup run as two parallel lanes (a split view), so cloning → configuring → building+downloading is one continuous, one-click flow; the individual "Build" and "Download Missing Models" hub actions remain only as fallbacks when something fails or is interrupted. Returns 0 on completion, 1 when the user aborted. """ parser = build_parser() args = parser.parse_args([]) settings = _wizard(stdscr, args, parser) if settings is None: return 1 return taskview.run_lanes(stdscr, "Setting up audio.cpp", _execute_lanes(settings, args)) def build_screen(stdscr) -> int: """Install audiocpp_server from the hub when the checkout has none. Asks which backend to install for (pre-selecting the backend an existing server.json records, else cuda), then — where a prebuilt release asset exists (macOS and Windows, see ``backends.audiocpp.prebuilt``) — whether to download it or build from source. The chosen action runs inside the TUI task view — alongside a download of any missing models when server.json is already configured and those models map to an install command (the split view), or alone otherwise — then updates server.json's ``backend`` field to match. Returns 0 on success, non-zero when the user backed out, cancelled, or the install failed. This is the hub's "Build audio.cpp Server" action, so a checkout that was cloned but never built is always installable from the TUI; the standalone "Download Missing Models" action stays as the fallback when a model download fails or is interrupted. """ checkout = _build.find_local_checkout() if checkout is None: tui.flash(stdscr, "No audio.cpp checkout found — install audio.cpp " "first.", "err") return 1 if _build.find_audiocpp_server_bin(checkout) is not None: tui.flash(stdscr, "audiocpp_server is already built.", "ok") return 0 server_config = load_server_config(checkout / "server.json") or {} recorded = server_config.get("backend") options, default = _backend_options(None) if recorded in BACKENDS: default = next((i for i, (_label, value) in enumerate(options) if value == recorded), default) backend = tui.menu( stdscr, "Which inference backend should audiocpp_server be built " "for?", options, default_index=default, back_value=_GO_BACK) if backend is _GO_BACK: return 1 mode = "source" if _prebuilt.prebuilt_supported(backend): mode = tui.menu( stdscr, "Install audiocpp_server:", [("Download prebuilt server from GitHub releases " "(recommended)", "prebuilt"), ("Build from source", "source")], default_index=0, back_value=_GO_BACK) if mode is _GO_BACK: return 1 outcome: dict = {"prebuilt": False} def build_step(emit, cancel): if mode == "prebuilt": rc = _prebuilt.install_prebuilt(checkout, backend, emit=emit, cancel=cancel) if rc == 0: outcome["prebuilt"] = True return 0 if rc == 130: return rc # The usual failure is GitHub's API rate limit — recover in # place instead of sending the user back to the menus. print(f"[WARNING] prebuilt download failed (exit {rc}); " "falling back to a source build...") return _build.build_audiocpp(checkout, backend, emit=emit, cancel=cancel) action = f"Install audiocpp_server ({backend})" if mode == "prebuilt" \ else f"Build audiocpp_server ({backend})" lanes = [taskview.TaskLane("Build", [taskview.TaskStep(action, build_step)])] # Missing models this build can also fetch, so a configured backend that # lost its binary is restored to "installed" in one step. server_json = checkout / "server.json" missing = _models.missing_model_entries(server_json) if server_json.exists() else [] guidance = _models.missing_model_install_guidance(checkout, missing) \ if missing else [] if guidance: def download_step(emit, cancel): _models.install_models(checkout, guidance, emit=emit, cancel=cancel) return 0 lanes.append(taskview.TaskLane( "Download models", [taskview.TaskStep("Download missing models", download_step)])) install_title = ("Install audiocpp_server" if mode == "prebuilt" else "Build audiocpp_server") title = f"{install_title} & download models" if len(lanes) == 2 \ else install_title rc = taskview.run_lanes(stdscr, title, lanes) if rc != 0: return rc if not _configsync.update_server_backend(backend): verb = "installed" if outcome["prebuilt"] else "built" tui.flash(stdscr, f"audiocpp_server {verb} for {backend}. (Could not " "update server.json's backend field — reconfigure audio.cpp " "if it was already configured.)", "warn") # Models that can't be mapped to an install command still need hand # installation; say so now rather than leaving the user in the dark. if missing and not guidance: tui.flash(stdscr, _models.hand_install_guidance(checkout, missing), "err") return 0 def run_tui(args: Optional[argparse.Namespace] = None, parser: Optional[argparse.ArgumentParser] = None) -> int: """Run the audio.cpp setup wizard end-to-end. With no ARGS (the hub's call) a default namespace is built so the full wizard runs. Called from ``main`` after argparse when the terminal is interactive. Returns the process exit code. """ import curses if args is None: parser = build_parser() args = parser.parse_args([]) if args.input_dir is not None and not args.input_dir.is_dir(): print(f"[ERROR] --wavs not found: {args.input_dir}", file=sys.stderr) return 2 try: settings = curses.wrapper(_wizard, args, parser) except _TuiError as exc: print(f"[ERROR] {exc}", file=sys.stderr) return 2 except tui.WizardCancelled: print("\n[INFO] Cancelled; nothing was written") return 1 try: curses.curs_set(1) # restore the text cursor hidden by the TUI except curses.error: pass if settings is None: print("[INFO] Aborted; existing server.json kept") return 1 return _execute(settings, args) def _collect_from_flags(args: argparse.Namespace, parser: argparse.ArgumentParser) -> Optional[dict]: """Build the settings dict from flags for a non-interactive run. Every required value must come from a flag (there are no prompts in a non-interactive run); a missing one is a hard ``parser.error``. Returns the settings dict, or None when the user declined an overwrite (the default-location fallback then also exists). """ # Checkout: ./app/audio.cpp, else --clone clones one there. # (The ggml build patches are applied by build_audiocpp itself, so a # prebuilt install never needs them.) audiocpp_dir = _build.find_local_checkout() if audiocpp_dir is None and args.clone: target = APP_DIR / AUDIOCPP_DIR_NAME rc = common.git_clone(AUDIOCPP_GIT_URL, target) if rc != 0: parser.error(f"git clone failed (exit {rc}); clone audio.cpp " f"manually: git clone {AUDIOCPP_GIT_URL} {target}") audiocpp_dir = target if audiocpp_dir is None: parser.error( "An audio.cpp checkout is required. Pass --clone to clone " "app/audio.cpp, or run without flags for the TUI wizard.") try: catalog = load_model_catalog(audiocpp_dir) except NotADirectoryError as exc: parser.error(str(exc)) if not catalog: parser.error( f"No TTS model families found in {audiocpp_dir}/model_specs; " "check the checkout is up to date") catalog_by_family = {entry["family"]: entry for entry in catalog} # Families: required from --families in a non-interactive run. if args.families is None: parser.error("--families is required in a non-interactive run (or run " "without flags for the TUI wizard)") requested = [f.strip() for f in args.families.split(",") if f.strip()] unknown = [f for f in requested if f not in catalog_by_family] if unknown: parser.error( f"Unknown family in --families: {', '.join(unknown)}. " f"Available: {', '.join(catalog_by_family)}") family_keys: List[str] = [] for fam in requested: if fam not in family_keys: family_keys.append(fam) chosen: Dict[str, List[dict]] = {} for family in family_keys: opts = package_dir_options(catalog_by_family[family]) if args.all_packages: chosen[family] = opts else: chosen[family] = [opt for opt in opts if opt["recommended"]] # Non-interactive picker: design packages default to vdes. def task_picker(install_id: str) -> str: return TASK_VDES model_entries, entry_ids, install_guidance, companion_guidance, \ design_entry_ids, include_clone = _build_entries( family_keys, chosen, catalog_by_family, task_picker) # Server settings. Host is always 127.0.0.1 and the port comes from # AUDIOCPP_API_URL in app/converter/config.py (the Settings screen) — # neither is a CLI option. host = DEFAULT_HOST detected_backend = detect_backend(audiocpp_dir) if args.build_backend: backend = args.build_backend build = detected_backend is None elif args.backend: backend = args.backend build = False elif detected_backend is not None: backend = detected_backend build = False else: backend = "cuda" build = False # How a pending install happens: the prebuilt release by default on # macOS/Windows (``--prebuilt no`` forces the source build), always # the source build elsewhere. A ``--prebuilt yes`` download that # fails stays a failure (no source-build fallback) — the flag's # whole point is "do not build". build_mode = _flag_build_mode(args, backend) if build else None prebuilt_forced = bool(build) and getattr(args, "prebuilt", "auto") \ == "yes" port = _configsync.config_port() lazy_load = True # Output path / overwrite (decline falls back to cwd, then aborts). output_path = args.output if args.output is not None \ else audiocpp_dir / "server.json" if output_path.exists() and not args.force: if args.output is None: output_path = Path.cwd() / "server.json" if output_path.exists() and not args.force: print("[INFO] Aborted; existing server.json kept") return None else: print("[INFO] Aborted; existing server.json kept") return None # Wav dir + transcription plan (defaults to the project's voices/ dir). wav_dir = args.input_dir if args.input_dir is not None else VOICES_DIR plan: Optional[dict] = None if include_clone and wav_dir is not None: wav_files = find_wav_files(wav_dir) if wav_files: prompt_path = wav_dir / PROMPT_TEXT_FILENAME plan = _voices._flag_plan(wav_files, prompt_path, args.force) return { "audiocpp_dir": audiocpp_dir, "catalog": catalog, "catalog_by_family": catalog_by_family, "output_path": output_path, "family_keys": family_keys, "chosen": chosen, "model_entries": model_entries, "entry_ids": entry_ids, "install_guidance": install_guidance, "companion_guidance": companion_guidance, "design_entry_ids": design_entry_ids, "include_clone": include_clone, "host": host, "port": port, "backend": backend, "build": build, "build_mode": build_mode, "prebuilt_forced": prebuilt_forced, "lazy_load": lazy_load, "sync_port": None, "wav_dir": wav_dir, "plan": plan, "download": args.download, } def build_parser() -> argparse.ArgumentParser: """The audio.cpp setup CLI (also used to build a default namespace).""" parser = argparse.ArgumentParser( description="Set up the audio.cpp TTS backend: clone/build, pick " "models, write server.json, and sync the configured port.") parser.add_argument("--wavs", type=resolve_wav_dir_arg, default=None, dest="input_dir", metavar="WAV_DIR", help="Directory with .wav reference files to publish as " "a server-level voice_dir cloning library " f"(default: {VOICES_DIR}; asked for when omitted " "in the TUI)") parser.add_argument("--output", type=Path, default=None, help="Output path for server.json (default: " "server.json inside the audio.cpp checkout; an " "existing file is overwritten only with --force " "or a TUI confirm)") parser.add_argument("--clone", action="store_true", help="Non-interactive: clone audio.cpp into " "./app/audio.cpp when no checkout is found") parser.add_argument("--families", type=str, default=None, help="Comma-separated model families to host, as named " "in the audio.cpp catalog (e.g. " "qwen3_tts,higgs_audio_tts). Required in a " "non-interactive run; skips the family tree in " "the TUI") parser.add_argument("--all-packages", action="store_true", help="Host every installable package of each selected " "family (distinct target_directory) instead of " "only the recommended one. Voice-design packages " "are hosted with task 'vdes'") parser.add_argument("--backend", choices=BACKENDS, default=None, help="Inference backend recorded in server.json " "(default: auto-detected from the checkout's " "build/ directory, else cuda)") parser.add_argument("--build-backend", choices=BACKENDS, default=None, help="Build audiocpp_server for this backend when it " "is not built yet, and use it in server.json") parser.add_argument("--prebuilt", choices=("auto", "yes", "no"), default="auto", help="How to install audiocpp_server when it is " "missing: auto downloads the prebuilt release " "on macOS/Windows (falling back to a source " "build if the download fails) and builds from " "source elsewhere; yes forces the prebuilt " "download and fails if it cannot; no forces a " "source build") parser.add_argument("--whisper-model", type=str, default="base", help="Whisper model size for transcription " "(default: base)") parser.add_argument("--force", action="store_true", help="Overwrite the output file (and prompt_text) " "without prompting; in the TUI, start the " "wizard fresh instead of loading the existing " "server.json") parser.add_argument("--download", action="store_true", help="Run model_manager_v2.py install for each hosted " "model automatically (default: print the commands " "only)") return parser def main() -> int: parser = build_parser() args = parser.parse_args() if args.input_dir is not None and not args.input_dir.is_dir(): parser.error( f"WAV directory not found: {args.input_dir}\n" f" (resolved from the current working directory: " f"{Path.cwd()})\n" " --wavs must be a directory containing the .wav " "reference files to use as voice cloning presets") if _interactive(): return run_tui(args, parser) # Non-interactive (no terminal, or all flags supplied): flag-only path. settings = _collect_from_flags(args, parser) if settings is None: return 1 return _execute(settings, args)