diff --git a/README.md b/README.md index 2d9880ed..d2cfecf1 100644 --- a/README.md +++ b/README.md @@ -173,6 +173,7 @@ wav = model.generate( cfg_value=2.0, inference_timesteps=10, seed=42, + volume_multiplier=1.0, # linear output gain relative to the reference level; e.g. 3.0 ≈ 3x louder ) sf.write("controllable_clone.wav", wav, model.tts_model.sample_rate) ``` diff --git a/README_zh.md b/README_zh.md index 88cd891d..f57243d6 100644 --- a/README_zh.md +++ b/README_zh.md @@ -175,6 +175,7 @@ wav = model.generate( cfg_value=2.0, inference_timesteps=10, seed=42, + volume_multiplier=1.0, # 相对参考音量的线性增益,例如 3.0 表示约 3 倍音量 ) sf.write("controllable_clone.wav", wav, model.tts_model.sample_rate) ``` diff --git a/src/voxcpm/core.py b/src/voxcpm/core.py index 1a1d8398..dc8d51dd 100644 --- a/src/voxcpm/core.py +++ b/src/voxcpm/core.py @@ -10,6 +10,32 @@ from .model.voxcpm2 import VoxCPM2Model from .model.utils import next_and_close +# Volume control (issue #362): a numeric, reproducible alternative to shaping +# loudness only through the prompt text. +_DEFAULT_VOLUME_MULTIPLIER = 1.0 # identity: leave the model's natural level untouched +_AUDIO_PEAK_LIMIT = 1.0 # full-scale amplitude for float32 PCM waveforms + + +def _apply_volume(wav: np.ndarray, multiplier: float) -> np.ndarray: + """Linearly scale a waveform by ``multiplier``, guarding against clipping. + + The model's natural output already matches the reference/prompt audio + loudness, so ``multiplier`` is effectively expressed relative to the + reference volume (e.g. ``3.0`` ≈ three times the reference volume). Peaks + that would exceed full scale are attenuated to avoid hard clipping. + + ponytail: streaming applies the peak limiter per chunk, so a single very + loud chunk is attenuated more than its neighbours; switch to a two-pass + scale over the full utterance if cross-chunk gain drift becomes audible. + """ + if multiplier == _DEFAULT_VOLUME_MULTIPLIER or wav.size == 0: + return wav + scaled = wav * multiplier + peak = float(np.max(np.abs(scaled))) + if peak > _AUDIO_PEAK_LIMIT: + scaled = scaled * (_AUDIO_PEAK_LIMIT / peak) + return scaled.astype(wav.dtype, copy=False) + class VoxCPM: def __init__( @@ -197,6 +223,7 @@ def _generate( retry_badcase_ratio_threshold: float = 6.0, streaming: bool = False, seed: Optional[int] = None, + volume_multiplier: float = _DEFAULT_VOLUME_MULTIPLIER, ) -> Generator[np.ndarray, None, None]: """Synthesize speech for the given text and return a single waveform. @@ -220,6 +247,11 @@ def _generate( retry_badcase_ratio_threshold: Threshold for audio-to-text ratio. streaming: Whether to return a generator of audio chunks. seed: Optional random seed for reproducibility. + volume_multiplier: Linear gain applied to the output waveform, + relative to the model's natural (reference-matched) level. + ``1.0`` leaves it unchanged; ``3.0`` is roughly three times the + reference volume. Must be positive. Peaks are limited to avoid + clipping. Returns: Generator of numpy.ndarray: 1D waveform array (float32) on CPU. Yields audio chunks for each generation step if ``streaming=True``, @@ -239,6 +271,9 @@ def _generate( if (prompt_wav_path is None) != (prompt_text is None): raise ValueError("prompt_wav_path and prompt_text must both be provided or both be None") + if volume_multiplier <= 0: + raise ValueError(f"volume_multiplier must be positive, got {volume_multiplier}") + is_v2 = isinstance(self.tts_model, VoxCPM2Model) if reference_wav_path is not None and not is_v2: raise ValueError("reference_wav_path is only supported with VoxCPM2 models") @@ -302,12 +337,12 @@ def _generate( if streaming: try: for wav, _, _ in generate_result: - yield wav.squeeze(0).cpu().numpy() + yield _apply_volume(wav.squeeze(0).cpu().numpy(), volume_multiplier) finally: generate_result.close() else: wav, _, _ = next_and_close(generate_result) - yield wav.squeeze(0).cpu().numpy() + yield _apply_volume(wav.squeeze(0).cpu().numpy(), volume_multiplier) finally: for tmp_path in temp_files: diff --git a/tests/test_volume.py b/tests/test_volume.py new file mode 100644 index 00000000..8a3f3169 --- /dev/null +++ b/tests/test_volume.py @@ -0,0 +1,128 @@ +"""Tests for the volume_multiplier control added for issue #362. + +Loads ``_apply_volume`` and the module constants directly from +``src/voxcpm/core.py`` without importing the heavy model dependencies. +""" +from __future__ import annotations + +import importlib.util +import sys +import types +from pathlib import Path + +import numpy as np +import pytest + +ROOT = Path(__file__).resolve().parents[1] +CORE_PATH = ROOT / "src" / "voxcpm" / "core.py" + +# Stub the model submodules that core.py imports at module load time, so we can +# exercise the pure volume helper without pulling in torch / the real weights. +hf_stub = types.ModuleType("huggingface_hub") +hf_stub.snapshot_download = lambda *a, **k: "" +sys.modules.setdefault("huggingface_hub", hf_stub) + +pkg = types.ModuleType("voxcpm") +pkg.__path__ = [str(ROOT / "src" / "voxcpm")] +sys.modules.setdefault("voxcpm", pkg) + +model_pkg = types.ModuleType("voxcpm.model") +model_pkg.__path__ = [str(ROOT / "src" / "voxcpm" / "model")] +sys.modules.setdefault("voxcpm.model", model_pkg) + +voxcpm_stub = types.ModuleType("voxcpm.model.voxcpm") +voxcpm_stub.VoxCPMModel = type("VoxCPMModel", (), {}) +voxcpm_stub.LoRAConfig = type("LoRAConfig", (), {}) +sys.modules["voxcpm.model.voxcpm"] = voxcpm_stub + +voxcpm2_stub = types.ModuleType("voxcpm.model.voxcpm2") +voxcpm2_stub.VoxCPM2Model = type("VoxCPM2Model", (), {}) +sys.modules["voxcpm.model.voxcpm2"] = voxcpm2_stub + +utils_stub = types.ModuleType("voxcpm.model.utils") +utils_stub.next_and_close = lambda gen: next(gen) +sys.modules["voxcpm.model.utils"] = utils_stub + +spec = importlib.util.spec_from_file_location("voxcpm.core", CORE_PATH) +core = importlib.util.module_from_spec(spec) +sys.modules["voxcpm.core"] = core +assert spec.loader is not None +spec.loader.exec_module(core) + + +def _rms(wav: np.ndarray) -> float: + return float(np.sqrt(np.mean(np.square(wav)))) + + +def test_identity_multiplier_is_noop(): + wav = np.array([0.1, -0.2, 0.3, -0.05], dtype=np.float32) + out = core._apply_volume(wav, core._DEFAULT_VOLUME_MULTIPLIER) + assert np.array_equal(out, wav) + + +def test_multiplier_scales_rms_linearly(): + wav = (np.array([0.1, -0.2, 0.15, -0.05], dtype=np.float32)) + out = core._apply_volume(wav, 2.0) + # Peak is 0.4 < 1.0, so no clipping limiter kicks in: RMS doubles exactly. + ratio = _rms(out) / _rms(wav) + print(f"rms in={_rms(wav):.6f} out={_rms(out):.6f} ratio={ratio:.6f}") + assert ratio == pytest.approx(2.0, rel=1e-5) + + +def test_peak_limited_to_avoid_clipping(): + wav = np.array([0.6, -0.5, 0.4], dtype=np.float32) + out = core._apply_volume(wav, 3.0) # would reach 1.8, must be limited to <= 1.0 + peak = float(np.max(np.abs(out))) + print(f"peak after 3x = {peak:.6f}") + assert peak <= core._AUDIO_PEAK_LIMIT + 1e-6 + assert peak == pytest.approx(core._AUDIO_PEAK_LIMIT, rel=1e-5) + + +def test_empty_waveform_untouched(): + wav = np.array([], dtype=np.float32) + out = core._apply_volume(wav, 5.0) + assert out.size == 0 + + +def _make_pipeline(monkeypatch): + """Build a VoxCPM instance whose generation yields a fixed unit waveform.""" + inst = core.VoxCPM.__new__(core.VoxCPM) + inst.text_normalizer = None + inst.denoiser = None + base = np.array([0.1, -0.2, 0.15, -0.05], dtype=np.float32) + + class _Tensor: + def __init__(self, arr): + self._arr = arr + + def squeeze(self, _dim): + return self + + def cpu(self): + return self + + def numpy(self): + return self._arr + + class _FakeModel: + def _generate_with_prompt_cache(self, **kwargs): + yield (_Tensor(base), None, None) + + inst.tts_model = _FakeModel() + # core.py branches on isinstance(..., VoxCPM2Model); force the v1 path. + monkeypatch.setattr(core, "VoxCPM2Model", type("Other", (), {})) + return inst, base + + +def test_generate_applies_multiplier_end_to_end(monkeypatch): + inst, base = _make_pipeline(monkeypatch) + out = inst.generate(text="hello world", volume_multiplier=2.0) + ratio = _rms(out) / _rms(base) + print(f"end-to-end ratio = {ratio:.6f}") + assert ratio == pytest.approx(2.0, rel=1e-5) + + +def test_generate_rejects_nonpositive_multiplier(monkeypatch): + inst, _ = _make_pipeline(monkeypatch) + with pytest.raises(ValueError, match="volume_multiplier must be positive"): + inst.generate(text="hello world", volume_multiplier=0.0)