"""The model_specs catalog, server.json building and selection views.""" import json import re import sys import wave from pathlib import Path from typing import Dict, List, Optional, Set, Tuple from backends.common import find_wav_files from converter.clients import AUDIOCPP_CLONE_ONLY_FAMILIES, audiocpp_family_spec_tasks from .constants import TASK_CLON, TASK_TTS, TASK_VDES DESIGN_PACKAGE_RE = re.compile(r"voice[\s_\-]?design", re.IGNORECASE) # The spec task vocabulary for "can this family turn text into audio": # plain synthesis ("tts"), reference-voice synthesis ("clone"; specs spell # it out, unlike the hosted "clon" task) and described-voice synthesis # ("vdes"). Families whose tasks name none of these only transform audio # (speech-to-speech, voice conversion, ...) and are kept out of the # install catalog. NARRATION_TASKS = frozenset({TASK_TTS, "clone", TASK_VDES}) def request_options_families(audiocpp_dir: Path) -> Dict[str, dict]: """Map the families whose spec defines per-request options. Reads every ``model_specs/.json`` in AUDIOCPP_DIR once and returns ``{family_key: {"display_name": ...}}`` for the specs that list request options (a non-empty ``options.request`` array) — the families a "Request options" field makes sense for. The key is the spec's ``family`` field (falling back to the file stem), matching what ``GET /v1/models`` reports, so callers can look an entry up by its family id. A missing or unreadable specs directory yields {} (every family then counts as unknown rather than unsupported). """ specs_dir = audiocpp_dir / "model_specs" if not specs_dir.is_dir(): return {} families: Dict[str, dict] = {} for spec_path in sorted(specs_dir.glob("*.json")): try: spec = json.loads(spec_path.read_text(encoding="utf-8")) except (OSError, ValueError): continue options = spec.get("options") request = options.get("request") if isinstance(options, dict) else None if not isinstance(request, list) or not request: continue family = str(spec.get("family") or spec_path.stem) families[family] = { "display_name": str(spec.get("display_name") or family), } return families def supports_request_options(families: Dict[str, dict], family: str) -> Optional[bool]: """Whether FAMILY accepts per-request options — None when unknown. True only when FAMILIES (from request_options_families) lists the family; False when it was read but does not define request options; None when support cannot be determined from the local specs (no checkout, or a family the specs do not describe). """ if not families: return None return family in families _BACKEND_DESCRIPTIONS = ( ("cuda", "NVIDIA GPUs"), ("vulkan", "cross-vendor GPU"), ("hip", "AMD GPUs"), ("cpu", "no GPU required"), ) def _backend_options(detected: Optional[str] = None ) -> Tuple[List[Tuple[str, str]], int]: """Build the aligned backend menu options and the default index. The backend names are padded to a common width so the ``-`` dashes before the descriptions line up. When DETECTED matches one of the options, that option gets ``[auto-detected]`` appended and is the default (cursor/start) selection; otherwise the first option is the default as before. Returns (options, default_index). On macOS only ``cpu`` is offered: Metal (via ``build_metal.sh``) is the only buildable inference backend there, and it is recorded as ``cpu`` (see the ``-metal-`` -> ``cpu`` mapping in ``detect_backend`` and ``built_server_binary``), so cuda/vulkan/hip — which cannot build on macOS — never reach the build step. """ if sys.platform == "darwin": label = "cpu - Apple Metal" if detected == "cpu": label += " [auto-detected]" return [(label, "cpu")], 0 width = max(len(name) for name, _ in _BACKEND_DESCRIPTIONS) options = [] default_index = 0 for index, (name, desc) in enumerate(_BACKEND_DESCRIPTIONS): label = f"{name.ljust(width)} - {desc}" if detected == name: label += " [auto-detected]" default_index = index options.append((label, name)) return options, default_index _BACKEND_TOKEN_RE = re.compile(r"-(cuda|vulkan|hip|cpu|metal)(?:-|$)") def detect_backend(audiocpp_dir: Path) -> Optional[str]: """Best-effort detection of the backend audiocpp_server was built for. Scans ``audiocpp_dir/build/*`` for build directories that contain a built ``bin/audiocpp_server`` (``.exe`` allowed on Windows) and reads the backend token out of the directory name (``-cuda-``, ``-vulkan-``, ``-hip-`` or ``-cpu-``; ``-metal-`` is mapped to ``cpu``). Returns the backend only when exactly one distinct backend was built, so a checkout with builds for several backends does not silently pick one. Returns None when there is no ``build/`` directory, no built server, or more than one distinct backend. """ build_root = audiocpp_dir / "build" if not build_root.is_dir(): return None backends: Set[str] = set() try: build_dirs = sorted(build_root.iterdir(), key=lambda p: p.name.lower()) except OSError: return None for build_dir in build_dirs: if not build_dir.is_dir(): continue server = build_dir / "bin" / "audiocpp_server" if not server.exists(): server_exe = build_dir / "bin" / "audiocpp_server.exe" if not server_exe.exists(): continue match = _BACKEND_TOKEN_RE.search(build_dir.name.lower()) if not match: continue token = match.group(1) backends.add("cpu" if token == "metal" else token) if len(backends) == 1: return next(iter(backends)) return None def _default_package(packages: List[dict]) -> Optional[dict]: """Pick the default package from a list of packages. Prefers the package flagged ``default: true``, then the first GGUF package, then the first package overall. Returns None for an empty list. """ if not packages: return None for package in packages: if package.get("default"): return package for package in packages: if package.get("format") == "gguf": return package return packages[0] def spec_gguf_rooted(spec: dict) -> bool: """True when SPEC's gguf source resolves its weights from ``$gguf``. The ``$gguf`` root is audio.cpp's single-GGUF convention: the model directory must hold exactly one top-level GGUF (or ``model.gguf``), which then carries the weights. Packages that feed such a source must therefore install their GGUF at the top of the target directory, not nested under a repository subdirectory. """ for source in spec.get("sources") or []: if not isinstance(source, dict) or source.get("format") != "gguf": continue roots = source.get("roots") if isinstance(roots, dict) and any( value == "$gguf" for value in roots.values() if isinstance(value, str)): return True return False def package_common_prefix(files: list) -> Optional[str]: """The one directory prefix every file path in FILES shares, or None. Only exact single-component matches count: every file must contain a ``/`` and start with the same first component. A flat package (no ``/`` at all) or a mixed one (some files under ``config/``, some at the root, like minimax_music3's intentional multi-file layout) yields None. """ prefixes: List[str] = [] for item in files: if not isinstance(item, str) or "/" not in item: return None prefix = item.split("/", 1)[0] if not prefix: return None prefixes.append(prefix) if not prefixes: return None first = prefixes[0] if first in (".", ".."): return None return first if all(prefix == first for prefix in prefixes) else None def sanitize_model_spec(spec: dict) -> bool: """Repair broken package ``strip_prefix`` entries in SPEC, in place. Two upstream spec bug classes have shipped in the audio.cpp checkout, and both make the model manager install files where the server cannot find them: - A dot ``strip_prefix`` ("." or "./") is meant for files written ``./``; when the package instead lists repo-root files bare (``model.gguf``), the manager rejects the whole package ("file path does not start with strip_prefix '.': ...") and nothing can be downloaded. Root-level files need no prefix at all (upstream specs like minimax_music3.json store ""), so dropping the dot prefix is the safe repair. Prefixes naming a real directory are left alone — the correct remote paths cannot be guessed. - A package with NO strip_prefix whose files all nest under one repository directory (``Text to audio (TTS)/GLM-TTS_Q8.gguf`` — glm_tts and outetts shipped like this) installs the GGUF under that subdirectory. The server then finds no top-level GGUF and falls back to the safetensors source, failing on a companion file the GGUF package never ships ("missing model package file 'tokenizer_merges'"). When the spec's gguf source resolves weights from ``$gguf`` (the single-GGUF convention) and every file shares one directory prefix, that prefix is what the strip_prefix should have been, so set it. Packages whose gguf source names tensors by explicit file paths (minimax_music3's nested multi-GGUF layout) are left untouched. Returns True when SPEC changed. """ changed = False for package in spec.get("packages") or []: if not isinstance(package, dict): continue prefix = str(package.get("strip_prefix") or "").rstrip("/") if prefix in (".", ".."): files = package.get("files") if isinstance(files, list) and files \ and all(isinstance(item, str) and item.startswith(prefix + "/") for item in files): continue package["strip_prefix"] = "" changed = True continue if prefix or package.get("format") != "gguf" \ or not spec_gguf_rooted(spec): continue common = package_common_prefix(package.get("files") or []) if common is None: continue package["strip_prefix"] = common changed = True return changed def load_model_catalog(audiocpp_dir: Path) -> List[dict]: """Read model_specs/*.json and return the TTS-capable families. Each returned entry has: family, display_name, description, languages, clone_capable, packages (the full list from the spec, with broken ``strip_prefix`` entries repaired in memory via sanitize_model_spec — the checkout's files are never written back; the download path materializes the same repair through its sanitized specs copy), install_id (recommended package id), and default_path (``models/``). All families are treated equally and listed in alphabetical order by display name. """ specs_dir = audiocpp_dir / "model_specs" if not specs_dir.is_dir(): raise NotADirectoryError( f"{audiocpp_dir} has no model_specs/ directory; re-run setup " "to refresh the audio.cpp checkout") entries: List[dict] = [] for spec_path in sorted(specs_dir.glob("*.json")): try: spec = json.loads(spec_path.read_text(encoding="utf-8")) except (OSError, ValueError): continue sanitize_model_spec(spec) tasks = spec.get("tasks") or [] if tasks: # A task list that names no text-synthesis capability means the # family cannot narrate text at all (e.g. PersonaPlex: # categorized "tts" but speech-to-speech only) — hosting one # fails every request, so it is never offered for install. if not (NARRATION_TASKS & set(tasks)): continue elif spec.get("category") != "tts": # No task list: fall back to the category as before. continue family = spec.get("family") or spec_path.stem packages = spec.get("packages") or [] package = _default_package(packages) if package is None: # No installable package: skip (cannot be hosted from a path). continue target_directory = package.get("target_directory") or family languages = spec.get("languages") or [] display_name = spec.get("display_name") or family description = spec.get("description") or "" entries.append({ "family": family, "display_name": display_name, "description": description, "languages": languages, "tasks": list(tasks), "clone_capable": "clone" in tasks, "packages": packages, "install_id": package.get("id") or family, "default_path": f"models/{target_directory}", }) # All families are treated equally: alphabetical by display name. entries.sort(key=lambda entry: entry["display_name"].lower()) return entries def is_design_package(package: dict) -> bool: """Return True when a package's name marks it a voice-design model. audio.cpp voice-design packages (whose id, display name, or target directory mentions "voice design") are the only packages that must be hosted with task "vdes"; their role is not in the schema, only in those strings, so it is detected from them. """ text = " ".join(str(package.get(key, "")) for key in ("id", "display_name", "target_directory")) return bool(DESIGN_PACKAGE_RE.search(text)) def is_clone_only_family(family: str, tasks: Optional[Set[str]] = None ) -> bool: """True when FAMILY's audio.cpp implementation rejects plain TTS. Such families can only synthesize by cloning a reference voice, so their server entries must be hosted with task "clon" — hosting them with "tts" fails every speech request at session-creation time. Families are classified from the explicit known-clone-only set (AUDIOCPP_CLONE_ONLY_FAMILIES, which also covers specs that wrongly claim "tts" — Chatterbox) or from a spec task list that names only "clone" (TASKS, when the caller has it; without one the specs are read best-effort). """ if family in AUDIOCPP_CLONE_ONLY_FAMILIES: return True if tasks is None: tasks = audiocpp_family_spec_tasks(family) return bool(tasks) and set(tasks) == {"clone"} def hosting_task(entry: dict) -> str: """The server.json task a family's non-design packages are hosted with. Clone-only families (see ``is_clone_only_family``) get "clon" so their cloning sessions can be created at all; every other family keeps "tts", which serves plain TTS and — where the family supports it — cloning through the request's voice field alike. """ if is_clone_only_family(str(entry.get("family") or ""), tasks=set(entry.get("tasks") or []) or None): return TASK_CLON return TASK_TTS def rehost_clone_only_entries(server_json: Path, data: dict) -> List[str]: """Re-host "tts"-tasked clone-only entries in DATA as "clon", in place. Server.json files written before clone-only hosting existed carry task "tts" for families whose audio.cpp implementation rejects plain TTS sessions (e.g. Chatterbox), so every speech request fails with HTTP 500. Each such entry's task is rewritten to "clon"; when anything changed, the document is written back to SERVER_JSON (same layout the wizard writes). Returns the repaired entries' ids, in order — empty when nothing needed changing (or the document is unusable). """ models = data.get("models") if not isinstance(models, list): return [] repaired: List[str] = [] for entry in models: if not isinstance(entry, dict): continue if str(entry.get("task") or "") != TASK_TTS: continue family = str(entry.get("family") or "") if not family or not is_clone_only_family(family): continue entry["task"] = TASK_CLON repaired.append(str(entry.get("id") or family)) if repaired: try: with server_json.open("w", encoding="utf-8") as handle: json.dump(data, handle, indent=2, ensure_ascii=False) handle.write("\n") except OSError: # The in-memory document is fixed either way; a failed write # only means the fix does not survive the process. pass return repaired def package_dir_options(entry: dict) -> List[dict]: """Return one option per distinct target_directory of a family's packages. Each option is a dict with: target_directory, install_id (the recommended package id inside that directory), design (voice-design package flag), and recommended (whether it holds the family's default package). Precisions that share a directory (q8_0/bf16/...) collapse to a single option. """ packages = entry.get("packages") or [] default_pkg = _default_package(packages) default_dir = (default_pkg or {}).get("target_directory") or entry["family"] by_dir: Dict[str, List[dict]] = {} order: List[str] = [] for package in packages: directory = package.get("target_directory") or entry["family"] if directory not in by_dir: by_dir[directory] = [] order.append(directory) by_dir[directory].append(package) options: List[dict] = [] for directory in order: package = _default_package(by_dir[directory]) options.append({ "target_directory": directory, "install_id": (package or {}).get("id") or directory, "design": is_design_package(package or {}), "recommended": directory == default_dir, }) # Put the recommended package first for a friendlier checklist. options.sort(key=lambda opt: not opt["recommended"]) return options def entry_model_path(entry: dict, target_directory: Optional[str] = None) -> str: """The server.json model path that hosts ENTRY's package for TARGET_DIRECTORY. ``models/`` normally (the recommended package's directory when TARGET_DIRECTORY is None). audio.cpp only loads a model directory that holds exactly one top-level GGUF, so a package that ships several GGUFs into one directory (MiniMax-H3's text-encoder / DiT / audio-VAE / video-VAE bundle) is hosted from its first GGUF file instead: the explicit file path makes the server pick the gguf source, and the directory stays the model root that the spec's named tensors resolve from. """ directory = target_directory if directory is None: default_path = str(entry.get("default_path") or "") directory = default_path[len("models/"):] \ if default_path.startswith("models/") else default_path if not directory: directory = str(entry.get("family") or "") package = next( (item for item in entry.get("packages") or [] if isinstance(item, dict) and str(item.get("target_directory") or entry.get("family")) == directory and item.get("format") == "gguf"), None) if package is None: return f"models/{directory}" prefix = str(package.get("strip_prefix") or "").rstrip("/") ggufs: List[str] = [] for item in package.get("files") or []: if not isinstance(item, str) or not item.lower().endswith(".gguf"): continue local = item if prefix and local.startswith(prefix + "/"): local = local[len(prefix) + 1:] elif prefix: continue ggufs.append(local) if len(ggufs) > 1: return f"models/{directory}/{ggufs[0]}" return f"models/{directory}" def build_model_entry(family: str, model_id: str, model_path: str, task: str = TASK_TTS, session_options: Optional[Dict[str, str]] = None) -> dict: """Assemble one server.json model entry. ``task`` defaults to "tts"; voice design packages are hosted with "vdes" so the server runs its design session for speech requests (audiobook.py then requires --instructions with that entry). SESSION_OPTIONS carries per-entry session-level options the server applies at session creation (e.g. MioTTS's codec model path, or a VoxCPM AudioVAE encoder capacity sized for long voice references); an empty mapping is omitted so the entry keeps its minimal shape. """ entry = { "id": model_id, "family": family, "path": model_path, "task": task, "mode": "offline", } if session_options: entry["session_options"] = dict(session_options) return entry def build_server_config(host: str, port: int, backend: str, lazy_load: bool, model_entries: List[dict], voice_dir: Optional[str] = None) -> dict: """Assemble the server.json document. ``voice_dir`` is a server-level cloning voice library; when set, every hosted clone-capable family can use its voices with ``--voice``. """ config_doc = { "host": host, "port": port, "backend": backend, "lazy_load": lazy_load, "models": model_entries, } if voice_dir: config_doc["voice_dir"] = voice_dir return config_doc def load_server_config(server_json: Path) -> Optional[dict]: """Read server.json into a dict, or None when it cannot be used. Returns None for a missing file, unreadable content, or a non-dict document. Used by the wizard's modify flow to pre-fill its screens from an existing config instead of prompting to overwrite it. """ if not server_json.exists(): return None try: data = json.loads(server_json.read_text(encoding="utf-8")) except (OSError, ValueError): return None if not isinstance(data, dict): return None return data def server_config_selections(server_config: dict, catalog: List[dict] ) -> Tuple[Dict[str, List[str]], Dict[Tuple[str, str], str]]: """Map an existing server.json's models back to catalog selections. Returns ``(selected_dirs, tasks)``: ``selected_dirs`` maps a catalog family to the target directories it hosts (``models/`` 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 rel = path[len("models/"):] if path.startswith("models/") else path # Entries hosted from a specific model file (e.g. MiniMax-H3's # multi-GGUF directory, "models//") belong to the # directory's catalog option, not to a made-up nested one. Unprefixed # and absolute paths keep their raw target as before (the wizard's # valid-directory check filters those it cannot offer again). target = rel.split("/", 1)[0] \ if path.startswith("models/") and "/" in rel else rel 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 # --------------------------------------------------------------------------- # Session options the setup bakes into server.json entries # --------------------------------------------------------------------------- # MioTTS loads its MioCodec companion through the ``miotts.codec_model_path`` # session option: the server's built-in default looks for the codec as a # sibling of the GGUF's materialized sidecar root (/tmp/audiocpp-gguf/...), # which is never where the model manager installs it. Pointing the option at # the installed codec package directory is the supported path, so the wizard # writes it (and downloads the codec alongside the model — it is an # audio_tools family, not a TTS one, so the model catalog never offers it). MIOTTS_CODEC_INSTALL_ID = "miocodec_q8_0" MIOTTS_CODEC_DIRECTORY = "MioCodec-25Hz-44.1kHz-v2-GGUF" MIOTTS_CODEC_MODEL_PATH = f"models/{MIOTTS_CODEC_DIRECTORY}" MIOTTS_CODEC_DISPLAY_NAME = "MioCodec 25Hz 44.1kHz v2 (required by MioTTS)" # VoxCPM-style AudioVAE encoders cap the reference audio they encode at a # fixed sample count (240000 samples ≈ 15 s at the VAE's 16 kHz rate, per # the audio.cpp runtime). A voice reference longer than that fails every # cloning request ("sample capacity exceeded"). The session option below # raises the cap; the wizard sizes it to the longest wav in the voice # directory so the configured voices all work without hand-trimming. VOXCPM_ENCODER_CAPACITY_DEFAULT_SAMPLES = 240_000 VOXCPM_ENCODER_CAPACITY_SAMPLE_RATE = 16_000 VOXCPM_ENCODER_CAPACITY_MAX_SAMPLES = 240_000 * 20 # Fallback key when a family's spec does not name the option (older # checkouts may lack voxcpm1.json; the spec is preferred when present). VOXCPM_ENCODER_CAPACITY_FALLBACK_KEYS = { "voxcpm2": "voxcpm2.audiovae_encoder_sample_capacity", "voxcpm1": "voxcpm1.audiovae_encoder_sample_capacity", } def wav_seconds(path: Path) -> Optional[float]: """A PCM WAV file's duration in seconds, or None when unreadable. The stdlib wave module handles the PCM variants voice references use; float-format or non-WAV files (mp3 renamed, exotic headers) yield None so callers skip them instead of guessing a duration. """ try: with wave.open(str(path), "rb") as handle: frames = handle.getnframes() rate = handle.getframerate() except (OSError, EOFError, wave.Error): return None if rate <= 0 or frames <= 0: return None return frames / rate def voxcpm_encoder_capacity_option(family: str, audiocpp_dir: Path) -> Optional[str]: """FAMILY's AudioVAE encoder-capacity session option key, or None. Discovered from the family's model spec (the option whose name carries ``encoder_sample_capacity``), falling back to the known VoxCPM keys so a spec-less family still gets the right option name rather than a broken server.json entry. """ try: spec = json.loads( (audiocpp_dir / "model_specs" / f"{family}.json") .read_text(encoding="utf-8")) except (OSError, ValueError): spec = None if isinstance(spec, dict): options = spec.get("options") session = options.get("session") if isinstance(options, dict) else None if isinstance(session, list): for item in session: name = item.get("name") if isinstance(item, dict) else None if isinstance(name, str) and "encoder_sample_capacity" in name: return name return VOXCPM_ENCODER_CAPACITY_FALLBACK_KEYS.get(family) def voxcpm_encoder_capacity_samples(max_seconds: Optional[float]) -> Optional[int]: """The encoder-sample capacity that fits MAX_SECONDS, or None. Rounds up to a multiple of the model's default 240000-sample capacity and clamps to a generous ceiling (8 minutes) so an absurdly long reference cannot push VRAM through the roof; anything at or below the default needs no override at all. """ if max_seconds is None or max_seconds <= 0: return None default = VOXCPM_ENCODER_CAPACITY_DEFAULT_SAMPLES needed = max_seconds * VOXCPM_ENCODER_CAPACITY_SAMPLE_RATE samples = ((int(needed) + default - 1) // default) * default if samples <= default: return None return min(samples, VOXCPM_ENCODER_CAPACITY_MAX_SAMPLES) def apply_entry_session_options(model_entries: List[dict], wav_dir: Optional[Path], audiocpp_dir: Path) -> List[str]: """Add the per-entry session options heavy families need, in place. MioTTS entries get ``miotts.codec_model_path`` (see MIOTTS_CODEC_*: the server's default codec path cannot be satisfied by a normal install). VoxCPM-family entries get an AudioVAE encoder-sample capacity sized to the longest wav in WAV_DIR, so voice references longer than the model's built-in 15 s ceiling still clone (returned as None when the voice directory is unknown, unreadable, or only holds short wavs — the default then applies as before). Existing session options are preserved; a hand-set codec path is never overridden. Returns the entry ids that gained options (for the wizard's summary line), in order. """ max_seconds: Optional[float] = None if wav_dir is not None: for path in find_wav_files(Path(wav_dir)): seconds = wav_seconds(path) if seconds is not None and (max_seconds is None or seconds > max_seconds): max_seconds = seconds applied: List[str] = [] for entry in model_entries: family = str(entry.get("family") or "") current = dict(entry.get("session_options") or {}) options = dict(current) if family == "miotts" \ and "miotts.codec_model_path" not in options: options["miotts.codec_model_path"] = MIOTTS_CODEC_MODEL_PATH if family.startswith("voxcpm") and max_seconds is not None: key = voxcpm_encoder_capacity_option(family, audiocpp_dir) samples = voxcpm_encoder_capacity_samples(max_seconds) if key and samples: options[key] = str(samples) if options != current: entry["session_options"] = options applied.append(str(entry.get("id") or family)) return applied