diff --git a/CHANGELOG.md b/CHANGELOG.md index b81df96b..d430a435 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ the frozen-backend fallback mirror it for their toolchains. **Highlights** - MOSS-TTS-v1.5, Confucius4-TTS, dots.tts, Supertonic-3 and PocketTTS install in one click, each in its own environment, so switching engines and back never breaks a working one (#2015, #2016) +- VoxCPM2 installs in one click into its own environment, with the CUDA build of PyTorch on NVIDIA GPUs (#VOXPR) - A pronunciation entry that is stored but not applied yet says so, instead of looking like it did not match (#1949) - A bare 500 report now names the backend error class, so two unrelated faults stop filing the same issue (#1773) - A rejected dubbing source language now names the code it rejected (#1960) diff --git a/backend/engines/voxcpm2_subprocess/__init__.py b/backend/engines/voxcpm2_subprocess/__init__.py new file mode 100644 index 00000000..e8886f8c --- /dev/null +++ b/backend/engines/voxcpm2_subprocess/__init__.py @@ -0,0 +1,106 @@ +"""voxcpm2-subprocess: VoxCPM2 from its own venv (one-click install). + +VoxCPM2 used to run only in-process, which meant installing ``voxcpm``, and a +torch of its choosing, into VoiceStudio's own environment. The one-click +installer now gives it a venv under ``DATA_DIR/engines/voxcpm2/``, and this +class runs the model there in a sidecar, so nothing it installs can touch the +app or another engine. + +The engine id stays ``voxcpm2``. ``tts_backend._effective_backend_class`` +resolves to this class once that venv exists and to the in-process +``VoxCPM2Backend`` otherwise, so an install made with ``pip install voxcpm`` +keeps working as it always has. What the app sees is the same: voice design, +48 kHz output, its own mastering, the same languages. The parent still +prepares the reference clip and trims the silent tail, as the in-process +engine does. +""" +from __future__ import annotations + +import math +import os +from pathlib import Path +from typing import TYPE_CHECKING + +from services.subprocess_backend import SubprocessBackend + +if TYPE_CHECKING: + import torch # noqa: F401 + +VENV_ENV_VAR = "OMNIVOICE_VOXCPM2_DIR" + + +def own_venv_python() -> "Path | None": + """The venv the one-click installer made for VoxCPM2, if any.""" + from services.sidecar_install import engine_venv_python + + return engine_venv_python(VENV_ENV_VAR) + + +class VoxCPM2SubprocessBackend(SubprocessBackend): + """VoxCPM2 in a killable sidecar running the engine's own venv.""" + + id = "voxcpm2" + display_name = "VoxCPM2 (30 langs, studio 48 kHz, voice design)" + supports_voice_design = True + applies_own_mastering = True # native 48 kHz studio output — skip apply_mastering() + gpu_compat = ("cuda", "mps", "cpu") + _DEFAULT_SAMPLE_RATE = 48_000 + + @classmethod + def is_available(cls) -> tuple[bool, str]: + if own_venv_python() is None: + return False, ( + "voxcpm package not installed. Install it from Model Catalogue → Engines." + ) + return True, "ready" + + @classmethod + def venv_python(cls) -> Path: + py = own_venv_python() + if py is None: + raise RuntimeError( + "VoxCPM2's environment is missing. Reinstall it from " + "Model Catalogue → Engines." + ) + return py + + @classmethod + def sidecar_script(cls) -> Path: + return Path(__file__).resolve().parent / "main.py" + + @property + def recv_timeout_s(self) -> float: + # A cold load downloads several GB of weights; the sidecar heartbeats + # progress frames meanwhile, and each one re-arms this deadline. + try: + v = float(os.environ.get("OMNIVOICE_VOXCPM2_RECV_TIMEOUT_S", "900")) + except (TypeError, ValueError): + return 900.0 + if not math.isfinite(v): # reject inf/nan so the deadline can't be disabled + return 900.0 + return max(30.0, v) + + @property + def sample_rate(self) -> int: + return self._DEFAULT_SAMPLE_RATE + + @property + def supported_languages(self) -> list[str]: + from services.tts_backend import VoxCPM2Backend + + return VoxCPM2Backend.supported_languages.fget(self) + + def generate(self, text: str, **kw) -> "torch.Tensor": + # The same preparation and finishing as VoxCPM2Backend.generate: the + # reference clip is trimmed and capped here (the model no longer does + # it), and the output's long silent tail is cut. + from services.audio_dsp import trim_trailing_silence + from services.tts_backend import _prepare_voxcpm_ref + + if kw.get("ref_audio"): + kw["ref_audio"] = _prepare_voxcpm_ref(kw["ref_audio"]) + wav = super().generate(text, **kw) + return trim_trailing_silence(wav, self.sample_rate) + + +__all__ = ["VENV_ENV_VAR", "VoxCPM2SubprocessBackend", "own_venv_python"] diff --git a/backend/engines/voxcpm2_subprocess/main.py b/backend/engines/voxcpm2_subprocess/main.py new file mode 100644 index 00000000..d008acf7 --- /dev/null +++ b/backend/engines/voxcpm2_subprocess/main.py @@ -0,0 +1,249 @@ +"""voxcpm2 sidecar: VoxCPM2 in the engine's own venv (one-click install). + +Launched as `` main.py`` by VoxCPM2SubprocessBackend. It +imports nothing from the app: the venv holds only ``voxcpm`` and what it +depends on (torch, torchaudio, numpy), so this file must stay importable with +the standard library plus those. The parent prepares the reference clip and +trims the output's silent tail, exactly as the in-process engine does; this +process only loads the model and synthesizes. + +Wire protocol: length-prefixed JSON over stdio, identical to the other +sidecars (engines/pockettts/main.py). A ``ready`` frame comes first, then one +``audio`` (or ``error``) frame per ``synthesize``, with ``progress`` frames +while a cold load runs so the parent's watchdog stays armed. +""" +from __future__ import annotations + +import base64 +import json +import os +import re +import struct +import sys +import threading +import time +import traceback + +# Mirrors services/subprocess_backend.py::MAX_FRAME_BYTES. +MAX_FRAME_BYTES = 64 * 1024 * 1024 +#: VoxCPM2's studio output rate; the in-process engine assumes the same. +VOXCPM2_SAMPLE_RATE = 48_000 +#: Emit a progress frame at least this often during a cold load (a multi-GB +#: first download) so the parent's recv watchdog doesn't kill a healthy sidecar. +_HEARTBEAT_S = 5.0 +#: ref_audio must be a local file path, not a URL (local-first; no SSRF). +_URL_RE = re.compile(r"^[a-z][a-z0-9+.\-]*://", re.IGNORECASE) +#: A download failure worth retrying (the HF cache resumes, so a retry +#: continues rather than restarts). Anything else propagates at once. +_TRANSIENT_MARKERS = ( + "connection", "timed out", "timeout", "peer closed", "incomplete", + "remoteprotocolerror", "temporarily unavailable", +) + +_MODEL = None + +# -- wire protocol ----------------------------------------------------------- + +#: Serializes _send across threads (the cold-load heartbeat + the main loop) so +#: concurrent length+body writes can't interleave and corrupt the framing. +_send_lock = threading.Lock() + + +def _send(stream, obj: dict) -> None: + body = json.dumps(obj, separators=(",", ":")).encode("utf-8") + with _send_lock: + stream.write(struct.pack("!I", len(body))) + stream.write(body) + stream.flush() + + +def _recv(stream): + header = stream.read(4) + if len(header) < 4: + return None # EOF + (n,) = struct.unpack("!I", header) + if n > MAX_FRAME_BYTES: + raise IOError(f"frame too large: {n}") + body = bytearray() + while len(body) < n: + chunk = stream.read(n - len(body)) + if not chunk: + raise IOError("short read") + body.extend(chunk) + return json.loads(bytes(body).decode("utf-8")) + + +def _measure_vram_mb() -> float: + try: + import torch # noqa: PLC0415 + + if torch.cuda.is_available(): + return float(torch.cuda.memory_allocated()) / (1024 * 1024) + except Exception: # noqa: BLE001 — a probe, never fatal + pass + return 0.0 + + +# -- model loading (lazy, on the first synthesize) --------------------------- + + +def _with_retries(load): + """Run ``load``, retrying a transient download failure with a short + backoff, the way the app's own loader does for in-process engines.""" + try: + attempts = max(1, int(os.environ.get("OMNIVOICE_MODEL_LOAD_RETRIES", "3"))) + except ValueError: + attempts = 3 + for attempt in range(1, attempts + 1): + try: + return load() + except Exception as exc: # noqa: BLE001 — classified below + text = f"{type(exc).__name__}: {exc}".lower() + if attempt == attempts or not any(m in text for m in _TRANSIENT_MARKERS): + raise + time.sleep(2.0 * attempt) + raise AssertionError("unreachable") + + +def _load_model(stdout): + global _MODEL + if _MODEL is not None: + return _MODEL + _send(stdout, {"op": "progress", "stage": "loading_model", "percent": 0}) + stop = threading.Event() + + def _heartbeat() -> None: + pct = 1 + while not stop.wait(_HEARTBEAT_S): + pct = min(pct + 1, 99) + _send(stdout, {"op": "progress", "stage": "loading_model", "percent": pct}) + + hb = threading.Thread(target=_heartbeat, daemon=True) + hb.start() + try: + from voxcpm import VoxCPM # type: ignore[import-not-found] # noqa: PLC0415 + + checkpoint = os.environ.get("OMNIVOICE_VOXCPM_MODEL", "openbmb/VoxCPM2") + _MODEL = _with_retries( + lambda: VoxCPM.from_pretrained(checkpoint, load_denoiser=False) + ) + finally: + stop.set() + hb.join(timeout=_HEARTBEAT_S + 1) + _send(stdout, {"op": "progress", "stage": "loading_model", "percent": 100}) + return _MODEL + + +def _sample_rate(model) -> int: + for owner in (model, getattr(model, "tts_model", None)): + sr = getattr(owner, "sample_rate", None) + if isinstance(sr, int) and sr > 0: + return sr + return VOXCPM2_SAMPLE_RATE + + +def _to_pcm_b64(wav) -> tuple[str, int]: + """A float waveform in [-1, 1] (numpy or torch) as base64 int16 PCM.""" + import numpy as np # noqa: PLC0415 + + if hasattr(wav, "detach"): + wav = wav.detach().float().cpu().numpy() + arr = np.asarray(wav, dtype=np.float32).squeeze() + if arr.ndim > 1: + raise ValueError(f"expected mono audio (1-D after squeeze), got shape {arr.shape}") + arr = np.clip(arr, -1.0, 1.0) + pcm = (arr * 32767.0).astype(np.int16).tobytes() + return base64.b64encode(pcm).decode("ascii"), int(arr.shape[-1]) + + +def _handle_synthesize(msg: dict, stdout) -> None: + """One synthesize request. The mapping mirrors VoxCPM2Backend.generate.""" + text = msg.get("text") + if not text or not isinstance(text, str): + raise ValueError("synthesize: missing or non-string 'text'") + ref_audio = msg.get("ref_audio") or None + if ref_audio and _URL_RE.match(str(ref_audio)): + raise ValueError( + "ref_audio must be a local file path; URLs are not accepted (local-first)." + ) + model = _load_model(stdout) + description = msg.get("description") + cfg_value = msg.get("guidance_scale", 2.0) + timesteps = msg.get("num_step", 10) + if description and not ref_audio: + # Voice design: a voice from a text description, no reference clip. + wav = model.generate( + text=text, + voice_description=description, + cfg_value=cfg_value, + inference_timesteps=timesteps, + ) + else: + instruct = msg.get("instruct") + ref_text = msg.get("ref_text") + wav = model.generate( + text=f"({instruct}){text}" if instruct else text, + cfg_value=cfg_value, + inference_timesteps=timesteps, + reference_wav_path=ref_audio, + prompt_wav_path=ref_audio if ref_text else None, + prompt_text=ref_text, + ) + pcm_b64, n_samples = _to_pcm_b64(wav) + _send(stdout, { + "op": "audio", + "audio_pcm_b64": pcm_b64, + "sample_rate": _sample_rate(model), + "n_samples": n_samples, + }) + + +# -- main loop --------------------------------------------------------------- + + +def main() -> int: + stdin = sys.stdin.buffer + # Frames go down a PRIVATE fd, and fd 1 is pointed at stderr (#1428): the + # libraries this loads print to fd 1 (tqdm, native torch output), and those + # bytes would otherwise interleave with the length-prefixed frames. + _frame_fd = os.dup(1) + os.dup2(2, 1) + stdout = os.fdopen(_frame_fd, "wb") + + # Ready handshake fires BEFORE any heavy import. + _send(stdout, {"op": "ready", "engine": "voxcpm2", "sample_rate": VOXCPM2_SAMPLE_RATE}) + + while True: + try: + msg = _recv(stdin) + except Exception as exc: # noqa: BLE001 + _send(stdout, { + "op": "error", + "stage": "recv", + "message": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc(), + }) + return 1 + if msg is None: + return 0 + op = msg.get("op") if isinstance(msg, dict) else None + try: + if op == "ping": + _send(stdout, {"op": "pong", "vram_mb": _measure_vram_mb()}) + elif op == "synthesize": + _handle_synthesize(msg, stdout) + elif op == "shutdown": + return 0 + else: + _send(stdout, {"op": "error", "stage": "dispatch", "message": f"unknown op: {op!r}"}) + except Exception as exc: # noqa: BLE001 + _send(stdout, { + "op": "error", + "stage": op or "unknown", + "message": f"{type(exc).__name__}: {exc}", + "traceback": traceback.format_exc(), + }) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/services/sidecar_install.py b/backend/services/sidecar_install.py index 7854de55..b407c8c9 100644 --- a/backend/services/sidecar_install.py +++ b/backend/services/sidecar_install.py @@ -149,6 +149,12 @@ class SidecarSpec: # Add PyTorch's CPU index on every host, for an engine that only ever # runs torch on the CPU (see core.torch_indexes). cpu_torch_index: bool = False + # torch/torchaudio pins for an upstream that leaves torch unpinned. Left + # to the resolver, PyPI's newest torch (CPU-only on Windows) pairs with a + # CUDA torchaudio from the other index. The host picks the build of the + # pinned pair: `+cu128` on a CUDA host, `+cpu` on other Windows and Linux + # hosts, plain on macOS. + torch_pins: tuple[str, ...] = () # Can the one-click install work on THIS machine? (ok, reason). Consulted # before an Install button is offered and again when an install starts, so # a host the upstream does not support never gets a job that can only fail. @@ -204,6 +210,16 @@ def _host_family() -> str: return "cpu" +def _torch_pin_args(spec: "SidecarSpec") -> list[str]: + from core.torch_indexes import UV_PIP_CPU_ARGS, UV_PIP_CU128_ARGS + + if _host_family() == "cuda": + return [f"{pin}+cu128" for pin in spec.torch_pins] + list(UV_PIP_CU128_ARGS) + if sys.platform in ("win32", "linux"): + return [f"{pin}+cpu" for pin in spec.torch_pins] + list(UV_PIP_CPU_ARGS) + return list(spec.torch_pins) + + def _moss_host() -> tuple[bool, str]: if _host_family() == "cuda": return True, "" @@ -232,6 +248,15 @@ def _pockettts_host() -> tuple[bool, str]: return True, "" +def _voxcpm2_host() -> tuple[bool, str]: + import platform + if sys.platform == "darwin" and platform.machine().lower() == "x86_64": + return False, ( + "VoxCPM2 needs a PyTorch version that has no Intel Mac build." + ) + return True, "" + + def _in_app_env(module: str) -> Callable[[], bool]: """An install made with ``uv sync --extra`` lives in the app's own environment. It counts as installed, so the installer never provisions a @@ -404,6 +429,28 @@ SPECS: dict[str, SidecarSpec] = { installed_probe=_in_app_env("pocket_tts"), host_supported=_pockettts_host, ), + # Run in a sidecar from its own venv (engines/voxcpm2_subprocess). voxcpm + # leaves torch unpinned, so the pair is pinned here; each build of it was + # resolved with voxcpm==2.0.3 on 2026-09-10 (Windows and Linux: +cu128 and + # +cpu; Apple Silicon: plain). Weights download on first synthesis. + "voxcpm2": SidecarSpec( + engine_id="voxcpm2", + display_name="VoxCPM2", + repo_url="", + tarball_url="", + checkout_dirname="voxcpm2", + env_var="OMNIVOICE_VOXCPM2_DIR", + probe_module="voxcpm", + has_source=False, + venv_args=("--python", "3.11"), + install_args=("voxcpm==2.0.3",), + torch_pins=("torch==2.11.0", "torchaudio==2.11.0"), + docs_path="docs/engines/voxcpm2.md", + # CUDA torch (~5 GB unpacked) + transformers. + required_bytes=10 * _GIB, + installed_probe=_in_app_env("voxcpm"), + host_supported=_voxcpm2_host, + ), } @@ -1157,7 +1204,9 @@ def _step_install_deps(spec: SidecarSpec, job: dict) -> None: uv = _locate_uv() _log(job, f"Installing {spec.display_name} into its venv (this can take several minutes) …") target = [_expand(arg, checkout) for arg in spec.install_args] - if spec.cpu_torch_index: + if spec.torch_pins: + target += _torch_pin_args(spec) + elif spec.cpu_torch_index: from core.torch_indexes import UV_PIP_CPU_ARGS target += list(UV_PIP_CPU_ARGS) elif spec.uses_cuda_index and _host_family() == "cuda": diff --git a/backend/services/tts_backend.py b/backend/services/tts_backend.py index ccd4ae98..66a03406 100644 --- a/backend/services/tts_backend.py +++ b/backend/services/tts_backend.py @@ -2663,6 +2663,12 @@ def _effective_backend_class( host_family: str | None = None, ) -> type[TTSBackend]: """Resolve host-specific containment without changing the configured id.""" + if backend_id == "voxcpm2": + # Its own venv (one-click install) runs in a sidecar; an install made + # with `pip install voxcpm` keeps running in-process. + from engines.voxcpm2_subprocess import VoxCPM2SubprocessBackend, own_venv_python + + return VoxCPM2SubprocessBackend if own_venv_python() is not None else backend_cls if backend_id != "omnivoice": return backend_cls if host_family is None: diff --git a/docs/engines/voxcpm2.md b/docs/engines/voxcpm2.md index 556c9b5d..d8a1bf0c 100644 --- a/docs/engines/voxcpm2.md +++ b/docs/engines/voxcpm2.md @@ -65,6 +65,20 @@ now retried once with a fresh client. See - Language coverage is 30 languages; for anything else use the default [OmniVoice](omnivoice.md) engine ([languages.md](../languages.md)). +## One-click install + +Click **Install** in **Model Catalogue → Engines → VoxCPM2**. VoiceStudio +puts VoxCPM2 in its own Python environment under its data directory and runs +it there, in a separate process. It installs the CUDA build of PyTorch on an +NVIDIA GPU, the CPU build on other Windows and Linux machines, and the +regular build on Apple Silicon. + +Nothing it installs touches VoiceStudio itself or any other engine, and +**Uninstall** in the same row removes only that folder. An existing +`pip install voxcpm` setup keeps working as it is. The button is not offered +on Intel Macs, where no PyTorch build it needs exists. The model weights +download on first use. + ## Troubleshooting - Engine shows unavailable: the `voxcpm` package isn't installed — run the diff --git a/tests/test_sidecar_install.py b/tests/test_sidecar_install.py index 2187c4ea..29c2364a 100644 --- a/tests/test_sidecar_install.py +++ b/tests/test_sidecar_install.py @@ -1046,7 +1046,7 @@ def test_cuda_index_is_added_only_for_cuda_pinned_specs_on_cuda_hosts( si._step_install_deps(spec, si._new_job(engine_id)) pip = next(a for a in argvs if a[1:3] == ["pip", "install"]) has_index = _PYTORCH_INDEX in pip - assert has_index == (spec.uses_cuda_index and family == "cuda") + assert has_index == (family == "cuda" and (spec.uses_cuda_index or bool(spec.torch_pins))) if has_index: i = pip.index("--extra-index-url") assert tuple(pip[i:i + len(UV_PIP_CU128_ARGS)]) == UV_PIP_CU128_ARGS @@ -1090,11 +1090,11 @@ def test_verify_probe_runs_in_the_engines_venv_and_compiles(monkeypatch, engine_ @pytest.mark.parametrize( ("family", "platform", "machine", "expected"), [ - ("cuda", "linux", "x86_64", {"moss-tts-v15", "dots-tts", "pockettts"}), - ("cuda", "win32", "AMD64", {"moss-tts-v15", "pockettts"}), - ("cpu", "win32", "AMD64", {"pockettts"}), - ("mps", "darwin", "arm64", {"dots-tts", "pockettts"}), - # Intel Mac: PyTorch publishes no build PocketTTS can use. + ("cuda", "linux", "x86_64", {"moss-tts-v15", "dots-tts", "pockettts", "voxcpm2"}), + ("cuda", "win32", "AMD64", {"moss-tts-v15", "pockettts", "voxcpm2"}), + ("cpu", "win32", "AMD64", {"pockettts", "voxcpm2"}), + ("mps", "darwin", "arm64", {"dots-tts", "pockettts", "voxcpm2"}), + # Intel Mac: PyTorch publishes no build PocketTTS or VoxCPM2 can use. ("cpu", "darwin", "x86_64", {"dots-tts"}), ], ) @@ -1287,3 +1287,34 @@ def test_a_failed_dependency_install_is_repaired_not_reported_installed(monkeypa monkeypatch.setattr(si, "_run_logged", ok_run) assert _run(spec)["state"] == "succeeded" assert si._healthy(spec) + + +@pytest.mark.parametrize( + ("family", "platform", "suffix", "index"), + [ + ("cuda", "win32", "+cu128", "https://download.pytorch.org/whl/cu128"), + ("cuda", "linux", "+cu128", "https://download.pytorch.org/whl/cu128"), + ("cpu", "win32", "+cpu", "https://download.pytorch.org/whl/cpu"), + ("rocm", "linux", "+cpu", "https://download.pytorch.org/whl/cpu"), + ("mps", "darwin", "", None), + ], +) +def test_torch_pins_follow_the_host(monkeypatch, family, platform, suffix, index): + """voxcpm leaves torch unpinned, and resolving it with the CUDA index + paired PyPI's newest torch (CPU-only on Windows) with a CUDA torchaudio. + The spec pins the pair; the host decides which build of it.""" + spec = si.get_spec("voxcpm2") + assert spec.torch_pins + argvs = _capture_install_argvs(monkeypatch, family=family) + monkeypatch.setattr(si.sys, "platform", platform) + + si._step_install_deps(spec, si._new_job("voxcpm2")) + + pip = next(a for a in argvs if a[1:3] == ["pip", "install"]) + for pin in spec.torch_pins: + assert f"{pin}{suffix}" in pip + if index: + assert pip.count("--extra-index-url") == 1 + assert pip[pip.index("--extra-index-url") + 1] == index + else: + assert "--extra-index-url" not in pip diff --git a/tests/test_voxcpm2_subprocess.py b/tests/test_voxcpm2_subprocess.py new file mode 100644 index 00000000..cbac2784 --- /dev/null +++ b/tests/test_voxcpm2_subprocess.py @@ -0,0 +1,196 @@ +"""VoxCPM2 from its own venv: the sidecar, and the switch to it. + +The sidecar runs in a venv that holds only voxcpm and its dependencies, so it +must import nothing from the app, and it must drive the model exactly as the +in-process VoxCPM2Backend does. These tests run it with a fake `voxcpm`. +""" +import importlib.util +import io +import json +import re +import struct +import sys +import types +from pathlib import Path + +import numpy as np +import pytest + +_MAIN = Path(__file__).resolve().parents[1] / "backend/engines/voxcpm2_subprocess/main.py" + + +def _load_sidecar(monkeypatch, calls): + class FakeModel: + sample_rate = 48000 + + def generate(self, **kw): + calls.append(kw) + return np.zeros(480, dtype=np.float32) + + class VoxCPM: + @classmethod + def from_pretrained(cls, checkpoint, **kw): + calls.append({"from_pretrained": checkpoint, **kw}) + return FakeModel() + + fake = types.ModuleType("voxcpm") + fake.VoxCPM = VoxCPM + monkeypatch.setitem(sys.modules, "voxcpm", fake) + spec = importlib.util.spec_from_file_location("_voxcpm2_sidecar_under_test", _MAIN) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _frames(buf): + data, out, i = buf.getvalue(), [], 0 + while i < len(data): + (n,) = struct.unpack("!I", data[i:i + 4]) + out.append(json.loads(data[i + 4:i + 4 + n])) + i += 4 + n + return out + + +def test_voice_design_maps_to_voice_description(monkeypatch): + calls = [] + sidecar = _load_sidecar(monkeypatch, calls) + out = io.BytesIO() + + sidecar._handle_synthesize({"op": "synthesize", "text": "hello", "description": "warm, low"}, out) + + assert calls[-1] == { + "text": "hello", + "voice_description": "warm, low", + "cfg_value": 2.0, + "inference_timesteps": 10, + } + audio = _frames(out)[-1] + assert audio["op"] == "audio" + assert audio["sample_rate"] == 48000 and audio["n_samples"] == 480 + + +def test_clone_mode_passes_the_reference_and_prompt(monkeypatch, tmp_path): + calls = [] + sidecar = _load_sidecar(monkeypatch, calls) + ref = tmp_path / "ref.wav" + ref.write_bytes(b"x") + + sidecar._handle_synthesize( + {"text": "hi", "ref_audio": str(ref), "ref_text": "hello there", "instruct": "calm", + "guidance_scale": 3.0, "num_step": 12}, + io.BytesIO(), + ) + + assert calls[-1] == { + "text": "(calm)hi", + "cfg_value": 3.0, + "inference_timesteps": 12, + "reference_wav_path": str(ref), + "prompt_wav_path": str(ref), + "prompt_text": "hello there", + } + + +def test_a_reference_without_its_transcript_is_not_a_prompt(monkeypatch, tmp_path): + calls = [] + sidecar = _load_sidecar(monkeypatch, calls) + ref = tmp_path / "ref.wav" + ref.write_bytes(b"x") + sidecar._handle_synthesize({"text": "hi", "ref_audio": str(ref)}, io.BytesIO()) + assert calls[-1]["reference_wav_path"] == str(ref) + assert calls[-1]["prompt_wav_path"] is None + + +def test_loads_the_configured_checkpoint_without_the_denoiser(monkeypatch): + calls = [] + monkeypatch.setenv("OMNIVOICE_VOXCPM_MODEL", "local/ckpt") + sidecar = _load_sidecar(monkeypatch, calls) + sidecar._handle_synthesize({"text": "hi"}, io.BytesIO()) + assert calls[0] == {"from_pretrained": "local/ckpt", "load_denoiser": False} + + +def test_rejects_a_url_reference(monkeypatch): + sidecar = _load_sidecar(monkeypatch, []) + with pytest.raises(ValueError, match="local file path"): + sidecar._handle_synthesize({"text": "hi", "ref_audio": "https://x.test/y.wav"}, io.BytesIO()) + + +def test_retries_a_transient_download_failure_only(monkeypatch): + sidecar = _load_sidecar(monkeypatch, []) + monkeypatch.setattr(sidecar.time, "sleep", lambda s: None) + attempts = [] + + def flaky(): + attempts.append(1) + if len(attempts) < 3: + raise OSError("peer closed connection without sending complete message body") + return "model" + + assert sidecar._with_retries(flaky) == "model" + assert len(attempts) == 3 + + def broken(): + raise ValueError("bad config") + + with pytest.raises(ValueError): + sidecar._with_retries(broken) + + +def test_the_sidecar_imports_nothing_from_the_app(): + src = _MAIN.read_text(encoding="utf-8") + for name in ("services", "core", "engines", "api", "backend", "utils"): + assert not re.search(rf"^\s*(from|import) {name}\b", src, re.M), name + + +def test_the_class_switches_to_the_sidecar_once_its_venv_exists(monkeypatch, tmp_path): + from engines.voxcpm2_subprocess import VoxCPM2SubprocessBackend + from services import tts_backend + from services.sidecar_install import _venv_python + + monkeypatch.setenv("OMNIVOICE_VOXCPM2_DIR", "") + monkeypatch.delenv("OMNIVOICE_VOXCPM2_DIR") + assert tts_backend.get_backend_class("voxcpm2") is tts_backend.VoxCPM2Backend + + py = _venv_python(tmp_path / ".venv") + py.parent.mkdir(parents=True) + py.write_text("#!fake\n") + monkeypatch.setenv("OMNIVOICE_VOXCPM2_DIR", str(tmp_path)) + + cls = tts_backend.get_backend_class("voxcpm2") + assert cls is VoxCPM2SubprocessBackend + assert cls.venv_python() == py + assert cls.is_available() == (True, "ready") + # The same engine to the rest of the app. + for attr in ("id", "display_name", "supports_voice_design", "applies_own_mastering", "gpu_compat"): + assert getattr(cls, attr) == getattr(tts_backend.VoxCPM2Backend, attr), attr + assert cls().supported_languages == tts_backend.VoxCPM2Backend().supported_languages + + +def test_the_sidecar_class_prepares_the_reference_and_trims_the_tail(monkeypatch): + import torch + + import services.audio_dsp as dsp + from engines.voxcpm2_subprocess import VoxCPM2SubprocessBackend + from services import subprocess_backend, tts_backend + + sent = {} + + def fake_generate(self, text, **kw): + sent.update(kw) + return torch.zeros(1, 4800) + + trimmed = {} + + def fake_trim(wav, sr): + trimmed["sr"] = sr + return wav[:, :10] + + monkeypatch.setattr(subprocess_backend.SubprocessBackend, "generate", fake_generate) + monkeypatch.setattr(tts_backend, "_prepare_voxcpm_ref", lambda p: p + ".prepared.wav") + monkeypatch.setattr(dsp, "trim_trailing_silence", fake_trim) + + out = VoxCPM2SubprocessBackend().generate("hi", ref_audio="/clip.wav") + + assert sent["ref_audio"] == "/clip.wav.prepared.wav" + assert trimmed["sr"] == 48000 + assert tuple(out.shape) == (1, 10)