From ada6082cb7b2f18de9b7d97392f573edad4ef51e Mon Sep 17 00:00:00 2001 From: Ben Younes <2910651+ousamabenyounes@users.noreply.github.com> Date: Wed, 9 Sep 2026 20:35:26 +0000 Subject: [PATCH] feat: add performance benchmark suite Closes #158 --- .gitignore | 1 + BENCHMARKS.md | 83 +++++ README.md | 7 +- benchmarks/__init__.py | 1 + benchmarks/benchmark_voxcpm.py | 376 ++++++++++++++++++++++ tests/test_benchmark_voxcpm.py | 558 +++++++++++++++++++++++++++++++++ 6 files changed, 1025 insertions(+), 1 deletion(-) create mode 100644 BENCHMARKS.md create mode 100644 benchmarks/__init__.py create mode 100644 benchmarks/benchmark_voxcpm.py create mode 100644 tests/test_benchmark_voxcpm.py diff --git a/.gitignore b/.gitignore index 61bac306..1e8f4e0a 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ launch.json .venv/ __pycache__ +.coverage voxcpm.egg-info .DS_Store ./pretrained_models/ diff --git a/BENCHMARKS.md b/BENCHMARKS.md new file mode 100644 index 00000000..1a5a698d --- /dev/null +++ b/BENCHMARKS.md @@ -0,0 +1,83 @@ +# VoxCPM performance benchmarks + +The benchmark runner measures the same public `VoxCPM.from_pretrained()` and +`generate()` path used by applications. It reports model-load time, each +generation's real-time factor (RTF), aggregate audio and utterance throughput, +and peak CUDA memory allocated during measured inference. + +## Run the benchmark + +Install VoxCPM and execute the runner from the repository root: + +```bash +python benchmarks/benchmark_voxcpm.py --device cuda --iterations 3 --output benchmark.json +``` + +Two built-in English prompts are used by default. Repeat `--text` to benchmark +a custom batch, or provide a UTF-8 file containing one prompt per line: + +```bash +python benchmarks/benchmark_voxcpm.py \ + --input-file prompts.txt \ + --iterations 3 \ + --warmup-runs 1 \ + --device cuda \ + --output benchmark.json +``` + +Each warm-up run generates every configured prompt before memory tracking and +measured iterations begin. + +Use `--no-optimize` when intentionally measuring eager CUDA inference. VoxCPM +currently compiles only on the unindexed `cuda` device, so the runner records +whether optimization was requested and attempted, plus the compile-wrapper +status of every inference component. The status can be `none`, `partial`, or +`full`, so a partially failed compilation is not mislabeled as fully optimized. +Wrapper detection describes the configured model; a backend failure during lazy +compilation causes the benchmark run itself to fail. Compilation is not attempted +for CPU, MPS, or indexed devices such as `cuda:1`. Use `--denoiser` when its +model-load cost should be part of the comparison. Run `--help` for model, +revision, generation, seed, and output options. + +For Hugging Face models, `--revision` accepts a branch, tag, or commit. The +runner downloads that revision once, loads the resulting local snapshot, and +records its immutable commit in `model_commit`. Snapshot resolution and download +happen before the model-load timer starts. Reuse that commit when reproducing a +report. Local model directories have no Hub commit and record `null` instead. + +## Metrics + +- `model_load_seconds`: wall time for `VoxCPM.from_pretrained()`, including its + built-in warm-up when optimization is enabled. +- `rtf`: generation wall time divided by generated audio duration. Lower is + better; values below 1 mean faster-than-real-time generation. +- `audio_seconds_per_second`: generated audio duration divided by generation + wall time. Higher is better and is the reciprocal of aggregate RTF. +- `utterances_per_second`: measured prompts divided by generation wall time. +- `peak_cuda_memory_mb`: peak CUDA memory allocated after warm-up. It is `null` + on CPU and MPS because PyTorch does not expose the corresponding CUDA metric. + +CUDA is synchronized around timed regions so asynchronous kernels are included. +Audio files are not written during the benchmark, avoiding storage performance +in the generation measurements. + +## Hardware comparison matrix + +Keep the model revision, prompts, generation settings, and software versions +identical when comparing machines. The JSON report records every prompt plus +platform, processor, Python, PyTorch, CUDA, model, requested/resolved/actual +device, iteration, warm-up, denoiser, requested/attempted/per-component +optimization, model snapshot, CFG, timestep, and seed metadata. + +The table below is intentionally an unpopulated results template: this project +does not publish unverified or simulated performance numbers. Add a row only +from a saved JSON report produced by the command above, and link or include the +report so other users can reproduce the measurement. + +| Model | Device | PyTorch / CUDA | Optimize | Mean RTF | Audio sec/sec | Utterances/sec | Peak CUDA MiB | +| --- | --- | --- | --- | ---: | ---: | ---: | ---: | +| _Verified report required_ | _GPU / CPU_ | _versions_ | _yes / no_ | _value_ | _value_ | _value_ | _value / N/A_ | + +For stable comparisons, run the benchmark on an otherwise idle machine at least +three times and report the median result. Include the exact command and JSON +report when publishing a matrix row. diff --git a/README.md b/README.md index 2d9880ed..ae7685b1 100644 --- a/README.md +++ b/README.md @@ -274,6 +274,11 @@ python app.py --device auto Supported values are `auto`, `cpu`, `mps`, `cuda`, and `cuda:N`. On Apple Silicon Macs, `auto` uses MPS when available. +### Performance benchmarking + +Use the reproducible [benchmark runner](BENCHMARKS.md) to measure model-load +time, real-time factor, batch throughput, and peak CUDA memory on your hardware. + ### 🚢 Production Deployment (Nano-vLLM) For high-throughput serving, use **[Nano-vLLM-VoxCPM](https://github.com/a710128/nanovllm-voxcpm)** — a dedicated inference engine built on Nano-vLLM with concurrent request support and an async API. @@ -694,4 +699,4 @@ VoxCPM model weights and code are open-sourced under the [Apache-2.0](LICENSE) l ## ⭐ Star History -[Star History Chart](https://star-history.com/#OpenBMB/VoxCPM&Date) \ No newline at end of file +[Star History Chart](https://star-history.com/#OpenBMB/VoxCPM&Date) diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 00000000..cb7f6e9e --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Performance benchmarks for VoxCPM.""" diff --git a/benchmarks/benchmark_voxcpm.py b/benchmarks/benchmark_voxcpm.py new file mode 100644 index 00000000..669a89b3 --- /dev/null +++ b/benchmarks/benchmark_voxcpm.py @@ -0,0 +1,376 @@ +#!/usr/bin/env python3 +"""Measure VoxCPM model load and text-to-speech generation performance.""" + +from __future__ import annotations + +import argparse +import json +import platform +import sys +import time +from pathlib import Path +from typing import Any, Callable, Iterable + +DEFAULT_MODEL_ID = "openbmb/VoxCPM2" +DEFAULT_DEVICE = "auto" +CUDA_DEVICE = "cuda" +MPS_DEVICE = "mps" +CPU_DEVICE = "cpu" +DEFAULT_ITERATIONS = 1 +DEFAULT_WARMUP_RUNS = 1 +DEFAULT_CFG_VALUE = 2.0 +DEFAULT_INFERENCE_TIMESTEPS = 10 +DEFAULT_SEED = 42 +DEFAULT_TEXTS = ( + "VoxCPM turns written language into natural speech.", + "This second sentence measures throughput across multiple prompts.", +) +BYTES_PER_MEBIBYTE = 1024 * 1024 +EXIT_SUCCESS = 0 +EXIT_USAGE_ERROR = 2 +JSON_INDENT = 2 +SNAPSHOTS_DIRECTORY = "snapshots" +COMPILED_FUNCTION_ATTRIBUTE = "_torchdynamo_orig_callable" +COMPILED_MODULE_ATTRIBUTE = "_orig_mod" +OPTIMIZATION_NONE = "none" +OPTIMIZATION_PARTIAL = "partial" +OPTIMIZATION_FULL = "full" + + +class BenchmarkError(RuntimeError): + """Raised when a benchmark cannot produce a trustworthy measurement.""" + + +class CudaMemory: + """Synchronize CUDA timing and expose peak allocated model memory.""" + + def __init__(self, torch_module: Any): + self._torch = torch_module + self._device: str | None = None + + def bind(self, model: Any) -> None: + """Select the CUDA device actually used by the loaded model.""" + resolved_device = str(getattr(getattr(model, "tts_model", None), "device", "")) + self._device = resolved_device if resolved_device.startswith(CUDA_DEVICE) else None + + @property + def available(self) -> bool: + return self._device is not None and bool(self._torch.cuda.is_available()) + + def synchronize(self) -> None: + if self.available: + self._torch.cuda.synchronize(device=self._device) + + def reset(self) -> None: + if self.available: + self.synchronize() + self._torch.cuda.reset_peak_memory_stats(device=self._device) + + def peak_megabytes(self) -> float | None: + if not self.available: + return None + self.synchronize() + return self._torch.cuda.max_memory_allocated(device=self._device) / BYTES_PER_MEBIBYTE + + +def _validated_texts(texts: Iterable[str]) -> tuple[str, ...]: + normalized = tuple(text.strip() for text in texts if text.strip()) + if not normalized: + raise BenchmarkError("Provide at least one non-empty prompt.") + return normalized + + +def run_benchmark( + *, + model_loader: Callable[[], Any], + texts: Iterable[str], + iterations: int, + warmup_runs: int, + generation_options: dict[str, Any], + clock: Callable[[], float] = time.perf_counter, + memory: Any, +) -> dict[str, Any]: + """Run a deterministic sequence of warm-up and measured generations.""" + benchmark_texts = _validated_texts(texts) + if iterations < 1: + raise BenchmarkError("Iterations must be at least one.") + if warmup_runs < 0: + raise BenchmarkError("Warm-up runs cannot be negative.") + + load_started = clock() + try: + model = model_loader() + except Exception as exc: + raise BenchmarkError(f"Unable to load model: {exc}") from exc + memory.bind(model) + memory.synchronize() + model_load_seconds = clock() - load_started + + sample_rate = getattr(getattr(model, "tts_model", None), "sample_rate", 0) + if sample_rate <= 0: + raise BenchmarkError("The loaded model reported an invalid sample rate.") + + for _ in range(warmup_runs): + for warmup_text in benchmark_texts: + try: + model.generate(text=warmup_text, **generation_options) + except Exception as exc: + raise BenchmarkError(f"Warm-up generation failed: {exc}") from exc + + memory.reset() + runs = [] + for iteration in range(iterations): + for text_index, text in enumerate(benchmark_texts): + memory.synchronize() + generation_started = clock() + try: + audio = model.generate(text=text, **generation_options) + except Exception as exc: + raise BenchmarkError(f"Generation failed for prompt {text_index + 1}: {exc}") from exc + memory.synchronize() + generation_seconds = clock() - generation_started + if generation_seconds <= 0: + raise BenchmarkError("Generation elapsed time must be positive.") + + sample_count = len(audio) + if sample_count <= 0: + raise BenchmarkError("The model generated empty audio.") + audio_seconds = sample_count / sample_rate + runs.append( + { + "iteration": iteration + 1, + "prompt": text_index + 1, + "text": text, + "generation_seconds": generation_seconds, + "audio_seconds": audio_seconds, + "rtf": generation_seconds / audio_seconds, + } + ) + + total_generation_seconds = sum(run["generation_seconds"] for run in runs) + total_audio_seconds = sum(run["audio_seconds"] for run in runs) + run_count = len(runs) + return { + "model_load_seconds": model_load_seconds, + "runtime": { + "device": str(getattr(model.tts_model, "device", "unknown")), + "optimization": _optimization_status(model), + }, + "summary": { + "runs": run_count, + "generated_audio_seconds": total_audio_seconds, + "generation_seconds": total_generation_seconds, + "mean_rtf": sum(run["rtf"] for run in runs) / run_count, + "audio_seconds_per_second": total_audio_seconds / total_generation_seconds, + "utterances_per_second": run_count / total_generation_seconds, + "peak_cuda_memory_mb": memory.peak_megabytes(), + }, + "runs": runs, + } + + +def load_model(*, model_id: str, device: str, load_denoiser: bool, optimize: bool) -> Any: + """Load VoxCPM without adding imports to benchmark discovery.""" + try: + from voxcpm import VoxCPM + + return VoxCPM.from_pretrained( + model_id, + device=device, + load_denoiser=load_denoiser, + optimize=optimize, + ) + except Exception as exc: + raise BenchmarkError(f"Unable to initialize {model_id!r}: {exc}") from exc + + +def resolve_model_source(model_id: str, revision: str | None) -> tuple[str, str | None]: + """Download a Hub snapshot once and return its immutable commit identifier.""" + model_path = Path(model_id) + if model_path.is_dir(): + if revision is not None: + raise BenchmarkError("A local model directory cannot be combined with --revision.") + return str(model_path), None + try: + from huggingface_hub import snapshot_download + + snapshot_path = Path(snapshot_download(repo_id=model_id, revision=revision)) + except Exception as exc: + raise BenchmarkError(f"Unable to resolve model {model_id!r}: {exc}") from exc + + if snapshot_path.parent.name != SNAPSHOTS_DIRECTORY: + raise BenchmarkError(f"Unable to determine the immutable commit for model {model_id!r}.") + return str(snapshot_path), snapshot_path.name + + +def describe_environment(torch_module: Any, device: str) -> dict[str, Any]: + """Collect enough environment metadata to compare benchmark reports.""" + cuda_available = bool(torch_module.cuda.is_available()) + selected_cuda = cuda_available and device.startswith(CUDA_DEVICE) + device_name = torch_module.cuda.get_device_name(device) if selected_cuda else None + return { + "platform": platform.platform(), + "processor": platform.processor() or platform.machine(), + "python": platform.python_version(), + "torch": torch_module.__version__, + "cuda": torch_module.version.cuda, + "cuda_device": device_name, + } + + +def _resolve_device(torch_module: Any, requested_device: str) -> str: + """Resolve ``auto`` using VoxCPM's CUDA, MPS, then CPU preference.""" + normalized_device = requested_device.strip().lower() + if not normalized_device: + raise BenchmarkError("Device cannot be empty.") + if normalized_device != DEFAULT_DEVICE: + return normalized_device + if torch_module.cuda.is_available(): + return CUDA_DEVICE + mps = getattr(getattr(torch_module, "backends", None), "mps", None) + if mps is not None and mps.is_available(): + return MPS_DEVICE + return CPU_DEVICE + + +def _optimization_attempt(*, device: str, requested: bool) -> bool: + """Attempt compilation only where VoxCPM supports it.""" + return requested and device == CUDA_DEVICE + + +def _optimization_status(model: Any) -> dict[str, Any]: + """Report compile wrappers per component, including partially optimized models.""" + tts_model = getattr(model, "tts_model", None) + base_lm = getattr(tts_model, "base_lm", None) + residual_lm = getattr(tts_model, "residual_lm", None) + feature_decoder = getattr(tts_model, "feat_decoder", None) + components = { + "base_lm": hasattr(getattr(base_lm, "forward_step", None), COMPILED_FUNCTION_ATTRIBUTE), + "residual_lm": hasattr(getattr(residual_lm, "forward_step", None), COMPILED_FUNCTION_ATTRIBUTE), + "feature_encoder": hasattr(getattr(tts_model, "feat_encoder", None), COMPILED_MODULE_ATTRIBUTE), + "feature_decoder": hasattr(getattr(feature_decoder, "estimator", None), COMPILED_MODULE_ATTRIBUTE), + } + detected_count = sum(components.values()) + if detected_count == len(components): + state = OPTIMIZATION_FULL + elif detected_count: + state = OPTIMIZATION_PARTIAL + else: + state = OPTIMIZATION_NONE + return {"state": state, "components": components} + + +def _positive_integer(value: str) -> int: + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be at least one") + return parsed + + +def _non_negative_integer(value: str) -> int: + parsed = int(value) + if parsed < 0: + raise argparse.ArgumentTypeError("cannot be negative") + return parsed + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + prompt_group = parser.add_mutually_exclusive_group() + prompt_group.add_argument("--text", action="append", help="Prompt to benchmark; repeat for a batch") + prompt_group.add_argument("--input-file", type=Path, help="UTF-8 file containing one prompt per line") + parser.add_argument("--model", default=DEFAULT_MODEL_ID, help="Hugging Face model ID or local model directory") + parser.add_argument("--revision", help="Hugging Face revision; the resolved snapshot commit is recorded") + parser.add_argument( + "--device", default=DEFAULT_DEVICE, help="Runtime device such as auto, cuda, cuda:0, mps, or cpu" + ) + parser.add_argument("--iterations", type=_positive_integer, default=DEFAULT_ITERATIONS) + parser.add_argument("--warmup-runs", type=_non_negative_integer, default=DEFAULT_WARMUP_RUNS) + parser.add_argument("--cfg-value", type=float, default=DEFAULT_CFG_VALUE) + parser.add_argument("--inference-timesteps", type=_positive_integer, default=DEFAULT_INFERENCE_TIMESTEPS) + parser.add_argument("--seed", type=int, default=DEFAULT_SEED) + parser.add_argument("--denoiser", action="store_true", help="Include the optional denoiser in model loading") + parser.add_argument("--no-optimize", action="store_true", help="Disable the default compiled inference path") + parser.add_argument("--output", type=Path, help="Write JSON here instead of standard output") + return parser + + +def _read_texts(args: argparse.Namespace) -> tuple[str, ...]: + if args.input_file is not None: + try: + return _validated_texts(args.input_file.read_text(encoding="utf-8").splitlines()) + except (OSError, UnicodeError) as exc: + raise BenchmarkError(f"Unable to read prompt file {args.input_file}: {exc}") from exc + return _validated_texts(args.text or DEFAULT_TEXTS) + + +def _write_result(result: dict[str, Any], output: Path | None) -> None: + payload = json.dumps(result, indent=JSON_INDENT, sort_keys=True) + "\n" + if output is None: + print(payload, end="") + return + try: + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(payload, encoding="utf-8") + except OSError as exc: + raise BenchmarkError(f"Unable to write benchmark result {output}: {exc}") from exc + + +def main(argv: list[str] | None = None) -> int: + parser = _build_parser() + args = parser.parse_args(argv) + try: + import torch + + texts = _read_texts(args) + resolved_device = _resolve_device(torch, args.device) + optimize_requested = not args.no_optimize + optimize_attempted = _optimization_attempt(device=resolved_device, requested=optimize_requested) + resolved_model, model_commit = resolve_model_source(args.model, args.revision) + memory = CudaMemory(torch) + result = run_benchmark( + model_loader=lambda: load_model( + model_id=resolved_model, + device=resolved_device, + load_denoiser=args.denoiser, + optimize=optimize_attempted, + ), + texts=texts, + iterations=args.iterations, + warmup_runs=args.warmup_runs, + generation_options={ + "cfg_value": args.cfg_value, + "inference_timesteps": args.inference_timesteps, + "seed": args.seed, + }, + memory=memory, + ) + actual_device = result["runtime"]["device"] + optimization = result["runtime"]["optimization"] + result["environment"] = describe_environment(torch, actual_device) + result["config"] = { + "model": args.model, + "model_revision": args.revision, + "model_commit": model_commit, + "device": resolved_device, + "device_actual": actual_device, + "device_requested": args.device, + "iterations": args.iterations, + "warmup_runs": args.warmup_runs, + "denoiser": args.denoiser, + "optimize_requested": optimize_requested, + "optimize_attempted": optimize_attempted, + "optimization_wrapper_state": optimization["state"], + "cfg_value": args.cfg_value, + "inference_timesteps": args.inference_timesteps, + "seed": args.seed, + } + _write_result(result, args.output) + except BenchmarkError as exc: + print(f"Benchmark error: {exc}", file=sys.stderr) + return EXIT_USAGE_ERROR + return EXIT_SUCCESS + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_benchmark_voxcpm.py b/tests/test_benchmark_voxcpm.py new file mode 100644 index 00000000..0196e815 --- /dev/null +++ b/tests/test_benchmark_voxcpm.py @@ -0,0 +1,558 @@ +from __future__ import annotations + +import argparse +import importlib.util +import json +import runpy +import sys +import types +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +BENCHMARK_PATH = ROOT / "benchmarks" / "benchmark_voxcpm.py" +SAMPLE_RATE = 16_000 +MEBIBYTE = 1024 * 1024 + +spec = importlib.util.spec_from_file_location("benchmark_voxcpm", BENCHMARK_PATH) +benchmark_voxcpm = importlib.util.module_from_spec(spec) +sys.modules["benchmark_voxcpm"] = benchmark_voxcpm +assert spec.loader is not None +spec.loader.exec_module(benchmark_voxcpm) + + +class FakeModel: + class TTSModel: + sample_rate = SAMPLE_RATE + device = "cpu" + + tts_model = TTSModel() + + def __init__(self): + self.calls = [] + + def generate(self, **kwargs): + self.calls.append(kwargs) + return [0.0] * SAMPLE_RATE + + +class FakeCudaMemory: + def __init__(self, peak_bytes=256 * MEBIBYTE): + self.peak_bytes = peak_bytes + self.bound_model = None + self.reset_calls = 0 + + def bind(self, model): + self.bound_model = model + + def reset(self): + self.reset_calls += 1 + + def synchronize(self): + pass + + def peak_megabytes(self): + return self.peak_bytes / MEBIBYTE + + +def test_run_benchmark_reports_load_rtf_throughput_and_peak_memory(): + model = FakeModel() + clock = iter([0.0, 2.0, 3.0, 3.5, 4.0, 4.5, 5.0, 5.5, 6.0, 6.5]).__next__ + memory = FakeCudaMemory() + + result = benchmark_voxcpm.run_benchmark( + model_loader=lambda: model, + texts=("first", "second"), + iterations=2, + warmup_runs=1, + generation_options={"cfg_value": 2.0, "inference_timesteps": 10, "seed": 42}, + clock=clock, + memory=memory, + ) + + assert result["model_load_seconds"] == pytest.approx(2.0) + assert result["summary"] == { + "runs": 4, + "generated_audio_seconds": pytest.approx(4.0), + "generation_seconds": pytest.approx(2.0), + "mean_rtf": pytest.approx(0.5), + "audio_seconds_per_second": pytest.approx(2.0), + "utterances_per_second": pytest.approx(2.0), + "peak_cuda_memory_mb": pytest.approx(256.0), + } + assert [run["text"] for run in result["runs"]] == ["first", "second", "first", "second"] + assert all(run["rtf"] == pytest.approx(0.5) for run in result["runs"]) + assert memory.reset_calls == 1 + assert memory.bound_model is model + assert [call["text"] for call in model.calls[:2]] == ["first", "second"] + assert len(model.calls) == 6 + + +@pytest.mark.parametrize( + ("sample_rate", "samples", "clock_values", "message"), + [ + (0, SAMPLE_RATE, [0.0, 1.0, 2.0, 3.0], "sample rate"), + (SAMPLE_RATE, 0, [0.0, 1.0, 2.0, 3.0], "empty audio"), + (SAMPLE_RATE, SAMPLE_RATE, [0.0, 1.0, 2.0, 2.0], "elapsed time"), + ], +) +def test_run_benchmark_rejects_invalid_measurements(sample_rate, samples, clock_values, message): + model = FakeModel() + model.tts_model = type("TTSModel", (), {"sample_rate": sample_rate})() + model.generate = lambda **kwargs: [0.0] * samples + + with pytest.raises(benchmark_voxcpm.BenchmarkError, match=message): + benchmark_voxcpm.run_benchmark( + model_loader=lambda: model, + texts=("test",), + iterations=1, + warmup_runs=0, + generation_options={}, + clock=iter(clock_values).__next__, + memory=FakeCudaMemory(), + ) + + +def test_main_reads_prompt_file_and_writes_machine_readable_result(monkeypatch, tmp_path): + input_path = tmp_path / "prompts.txt" + input_path.write_text("hello\n\nworld\n", encoding="utf-8") + output_path = tmp_path / "result.json" + recorded = {} + + def fake_run_benchmark(**kwargs): + recorded.update(kwargs) + kwargs["model_loader"]() + return { + "model_load_seconds": 1.25, + "runtime": { + "device": "cpu", + "optimization": { + "state": "none", + "components": { + "base_lm": False, + "residual_lm": False, + "feature_encoder": False, + "feature_decoder": False, + }, + }, + }, + "summary": {"runs": 2}, + "runs": [], + } + + def fake_load_model(**kwargs): + recorded["model_options"] = kwargs + return FakeModel() + + monkeypatch.setattr(benchmark_voxcpm, "run_benchmark", fake_run_benchmark) + monkeypatch.setattr(benchmark_voxcpm, "load_model", fake_load_model) + monkeypatch.setattr( + benchmark_voxcpm, + "resolve_model_source", + lambda model_id, revision: ("/cache/snapshots/abc123", "abc123"), + ) + + exit_code = benchmark_voxcpm.main( + [ + "--input-file", + str(input_path), + "--output", + str(output_path), + "--model", + "local/model", + "--device", + "cpu", + "--iterations", + "2", + "--warmup-runs", + "0", + "--no-optimize", + "--denoiser", + "--cfg-value", + "3.0", + "--inference-timesteps", + "12", + "--seed", + "7", + ] + ) + + assert exit_code == 0 + assert recorded["texts"] == ("hello", "world") + assert recorded["iterations"] == 2 + assert recorded["warmup_runs"] == 0 + assert recorded["generation_options"] == { + "cfg_value": 3.0, + "inference_timesteps": 12, + "seed": 7, + } + assert recorded["model_options"] == { + "model_id": "/cache/snapshots/abc123", + "device": "cpu", + "load_denoiser": True, + "optimize": False, + } + payload = json.loads(output_path.read_text(encoding="utf-8")) + assert payload["summary"]["runs"] == 2 + assert payload["config"] == { + "model": "local/model", + "model_revision": None, + "model_commit": "abc123", + "device": "cpu", + "device_actual": "cpu", + "device_requested": "cpu", + "iterations": 2, + "warmup_runs": 0, + "denoiser": True, + "optimize_requested": False, + "optimize_attempted": False, + "optimization_wrapper_state": "none", + "cfg_value": 3.0, + "inference_timesteps": 12, + "seed": 7, + } + + +def test_main_reports_input_errors_without_loading_model(capsys): + exit_code = benchmark_voxcpm.main(["--text", " "]) + + assert exit_code == 2 + assert "non-empty prompt" in capsys.readouterr().err + + +@pytest.mark.parametrize( + ("iterations", "warmup_runs", "message"), + [(0, 0, "Iterations"), (1, -1, "Warm-up")], +) +def test_run_benchmark_rejects_invalid_run_counts(iterations, warmup_runs, message): + with pytest.raises(benchmark_voxcpm.BenchmarkError, match=message): + benchmark_voxcpm.run_benchmark( + model_loader=FakeModel, + texts=("test",), + iterations=iterations, + warmup_runs=warmup_runs, + generation_options={}, + memory=FakeCudaMemory(), + ) + + +def test_run_benchmark_wraps_model_load_and_generation_errors(): + def fail_load(): + raise OSError("checkpoint unavailable") + + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="Unable to load model"): + benchmark_voxcpm.run_benchmark( + model_loader=fail_load, + texts=("test",), + iterations=1, + warmup_runs=0, + generation_options={}, + clock=iter([0.0]).__next__, + memory=FakeCudaMemory(), + ) + + model = FakeModel() + model.generate = lambda **kwargs: (_ for _ in ()).throw(RuntimeError("generation failed")) + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="prompt 1"): + benchmark_voxcpm.run_benchmark( + model_loader=lambda: model, + texts=("test",), + iterations=1, + warmup_runs=0, + generation_options={}, + clock=iter([0.0, 1.0, 2.0]).__next__, + memory=FakeCudaMemory(), + ) + + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="Warm-up generation"): + benchmark_voxcpm.run_benchmark( + model_loader=lambda: model, + texts=("test",), + iterations=1, + warmup_runs=1, + generation_options={}, + clock=iter([0.0, 1.0]).__next__, + memory=FakeCudaMemory(), + ) + + +def test_cuda_memory_tracks_available_cuda_and_noops_without_it(): + class FakeCuda: + def __init__(self, available): + self.available = available + self.synchronize_calls = [] + self.reset_calls = [] + self.max_calls = [] + + def is_available(self): + return self.available + + def synchronize(self, device=None): + self.synchronize_calls.append(device) + + def reset_peak_memory_stats(self, device=None): + self.reset_calls.append(device) + + def max_memory_allocated(self, device=None): + self.max_calls.append(device) + return 512 * MEBIBYTE + + available_cuda = FakeCuda(True) + available_memory = benchmark_voxcpm.CudaMemory(types.SimpleNamespace(cuda=available_cuda)) + available_memory.bind(types.SimpleNamespace(tts_model=types.SimpleNamespace(device="cuda:1"))) + available_memory.synchronize() + available_memory.reset() + + assert available_memory.available is True + assert available_memory.peak_megabytes() == 512.0 + assert available_cuda.synchronize_calls == ["cuda:1", "cuda:1", "cuda:1"] + assert available_cuda.reset_calls == ["cuda:1"] + assert available_cuda.max_calls == ["cuda:1"] + + cpu_on_cuda = FakeCuda(True) + unavailable_memory = benchmark_voxcpm.CudaMemory(types.SimpleNamespace(cuda=cpu_on_cuda)) + unavailable_memory.bind(types.SimpleNamespace(tts_model=types.SimpleNamespace(device="cpu"))) + unavailable_memory.synchronize() + unavailable_memory.reset() + + assert unavailable_memory.available is False + assert unavailable_memory.peak_megabytes() is None + assert cpu_on_cuda.synchronize_calls == [] + assert cpu_on_cuda.reset_calls == [] + assert cpu_on_cuda.max_calls == [] + + +@pytest.mark.parametrize( + ("device", "requested", "expected"), + [ + ("cuda", True, True), + ("cuda:1", True, False), + ("cpu", True, False), + ("cuda", False, False), + ], +) +def test_optimization_attempt_only_enables_supported_cuda_device(device, requested, expected): + assert benchmark_voxcpm._optimization_attempt(device=device, requested=requested) is expected + + +@pytest.mark.parametrize( + ("cuda_available", "mps_available", "expected"), + [(True, True, "cuda"), (False, True, "mps"), (False, False, "cpu")], +) +def test_resolve_device_expands_auto_in_runtime_order(cuda_available, mps_available, expected): + torch_module = types.SimpleNamespace( + cuda=types.SimpleNamespace(is_available=lambda: cuda_available), + backends=types.SimpleNamespace( + mps=types.SimpleNamespace(is_available=lambda: mps_available), + ), + ) + + assert benchmark_voxcpm._resolve_device(torch_module, "auto") == expected + assert benchmark_voxcpm._resolve_device(torch_module, "cuda:1") == "cuda:1" + assert benchmark_voxcpm._resolve_device(torch_module, " CUDA:1 ") == "cuda:1" + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="Device cannot be empty"): + benchmark_voxcpm._resolve_device(torch_module, " ") + + +def test_load_model_forwards_options_and_wraps_provider_error(monkeypatch): + calls = {} + + class StubVoxCPM: + @classmethod + def from_pretrained(cls, *args, **kwargs): + calls["args"] = args + calls["kwargs"] = kwargs + return "model" + + module = types.SimpleNamespace(VoxCPM=StubVoxCPM) + monkeypatch.setitem(sys.modules, "voxcpm", module) + assert ( + benchmark_voxcpm.load_model( + model_id="model/id", + device="cuda:1", + load_denoiser=True, + optimize=False, + ) + == "model" + ) + assert calls == { + "args": ("model/id",), + "kwargs": { + "device": "cuda:1", + "load_denoiser": True, + "optimize": False, + }, + } + + class BrokenVoxCPM: + @classmethod + def from_pretrained(cls, *args, **kwargs): + raise OSError("download failed") + + module = types.SimpleNamespace(VoxCPM=BrokenVoxCPM) + monkeypatch.setitem(sys.modules, "voxcpm", module) + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="model/id"): + benchmark_voxcpm.load_model( + model_id="model/id", + device="cpu", + load_denoiser=False, + optimize=True, + ) + + +@pytest.mark.parametrize( + ("available", "device", "expected_name", "expected_calls"), + [(True, "cuda:1", "Second GPU", ["cuda:1"]), (True, "cpu", None, []), (False, "cuda:1", None, [])], +) +def test_describe_environment_reports_selected_cuda_device( + monkeypatch, available, device, expected_name, expected_calls +): + requested_devices = [] + cuda = types.SimpleNamespace( + is_available=lambda: available, + get_device_name=lambda selected_device: requested_devices.append(selected_device) or "Second GPU", + ) + torch_module = types.SimpleNamespace( + cuda=cuda, + version=types.SimpleNamespace(cuda="12.4"), + __version__="2.5.0", + ) + monkeypatch.setattr(benchmark_voxcpm.platform, "processor", lambda: "Test Processor") + + environment = benchmark_voxcpm.describe_environment(torch_module, device) + + assert environment["torch"] == "2.5.0" + assert environment["cuda"] == "12.4" + assert environment["cuda_device"] == expected_name + assert environment["processor"] == "Test Processor" + assert requested_devices == expected_calls + assert environment["platform"] + assert environment["python"] + + +def test_describe_environment_falls_back_to_machine_identifier(monkeypatch): + monkeypatch.setattr(benchmark_voxcpm.platform, "processor", lambda: "") + monkeypatch.setattr(benchmark_voxcpm.platform, "machine", lambda: "arm64") + torch_module = types.SimpleNamespace( + cuda=types.SimpleNamespace(is_available=lambda: False), + version=types.SimpleNamespace(cuda=None), + __version__="2.5.0", + ) + + assert benchmark_voxcpm.describe_environment(torch_module, "mps")["processor"] == "arm64" + + +def test_optimization_status_reports_full_partial_and_eager_components(): + compiled_function = types.SimpleNamespace(_torchdynamo_orig_callable=object()) + compiled_module = types.SimpleNamespace(_orig_mod=object()) + tts_model = types.SimpleNamespace( + base_lm=types.SimpleNamespace(forward_step=compiled_function), + residual_lm=types.SimpleNamespace(forward_step=compiled_function), + feat_encoder=compiled_module, + feat_decoder=types.SimpleNamespace(estimator=compiled_module), + ) + model = types.SimpleNamespace(tts_model=tts_model) + + expected_components = { + "base_lm": True, + "residual_lm": True, + "feature_encoder": True, + "feature_decoder": True, + } + assert benchmark_voxcpm._optimization_status(model) == { + "state": "full", + "components": expected_components, + } + + tts_model.feat_decoder.estimator = object() + expected_components["feature_decoder"] = False + assert benchmark_voxcpm._optimization_status(model) == { + "state": "partial", + "components": expected_components, + } + + assert benchmark_voxcpm._optimization_status(types.SimpleNamespace(tts_model=object())) == { + "state": "none", + "components": { + "base_lm": False, + "residual_lm": False, + "feature_encoder": False, + "feature_decoder": False, + }, + } + + +def test_resolve_model_source_records_hub_commit_and_preserves_local_path(monkeypatch, tmp_path): + commit = "a" * 40 + snapshot_path = tmp_path / "models--openbmb--VoxCPM2" / "snapshots" / commit + snapshot_path.mkdir(parents=True) + calls = [] + hub = types.SimpleNamespace( + snapshot_download=lambda **kwargs: calls.append(kwargs) or str(snapshot_path), + ) + monkeypatch.setitem(sys.modules, "huggingface_hub", hub) + + assert benchmark_voxcpm.resolve_model_source("openbmb/VoxCPM2", "release") == (str(snapshot_path), commit) + assert calls == [{"repo_id": "openbmb/VoxCPM2", "revision": "release"}] + + local_model = tmp_path / "local-model" + local_model.mkdir() + assert benchmark_voxcpm.resolve_model_source(str(local_model), None) == (str(local_model), None) + assert len(calls) == 1 + + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="local model.*revision"): + benchmark_voxcpm.resolve_model_source(str(local_model), "ignored") + + hub.snapshot_download = lambda **kwargs: str(tmp_path / "unresolved-model") + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="immutable commit"): + benchmark_voxcpm.resolve_model_source("openbmb/VoxCPM2", None) + + def fail_download(**kwargs): + raise OSError("Hub unavailable") + + hub.snapshot_download = fail_download + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="Unable to resolve model"): + benchmark_voxcpm.resolve_model_source("openbmb/VoxCPM2", None) + + +def test_argument_validators_reject_out_of_range_values(): + with pytest.raises(argparse.ArgumentTypeError, match="at least one"): + benchmark_voxcpm._positive_integer("0") + with pytest.raises(argparse.ArgumentTypeError, match="negative"): + benchmark_voxcpm._non_negative_integer("-1") + + +def test_prompt_file_and_output_boundaries_are_reported(monkeypatch, tmp_path, capsys): + args = benchmark_voxcpm._build_parser().parse_args(["--input-file", str(tmp_path / "missing.txt")]) + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="Unable to read prompt file"): + benchmark_voxcpm._read_texts(args) + + invalid_utf8_path = tmp_path / "invalid-utf8.txt" + invalid_utf8_path.write_bytes(b"\xff") + args = benchmark_voxcpm._build_parser().parse_args(["--input-file", str(invalid_utf8_path)]) + with pytest.raises(benchmark_voxcpm.BenchmarkError, match="Unable to read prompt file"): + benchmark_voxcpm._read_texts(args) + + benchmark_voxcpm._write_result({"status": "ok"}, None) + assert json.loads(capsys.readouterr().out) == {"status": "ok"} + + def fail_write(*args, **kwargs): + raise OSError("disk unavailable") + + monkeypatch.setattr(Path, "write_text", fail_write) + with pytest.raises( + benchmark_voxcpm.BenchmarkError, + match="Unable to write benchmark result", + ): + benchmark_voxcpm._write_result({"status": "ok"}, tmp_path / "result.json") + + +def test_script_entry_point_returns_main_exit_code(monkeypatch, capsys): + monkeypatch.setattr(sys, "argv", [str(BENCHMARK_PATH), "--text", " "]) + + with pytest.raises(SystemExit) as exc_info: + runpy.run_path(str(BENCHMARK_PATH), run_name="__main__") + + assert exc_info.value.code == 2 + assert "non-empty prompt" in capsys.readouterr().err