import hashlib import io import sys from types import SimpleNamespace import numpy as np import pytest import soundfile as sf from producer import pauses, pipeline from producer.config import resolve def test_hysteresis_keeps_short_words_and_closes_at_eof(): values = [0, 0.5, 0.25, 0.1, 0, 0.8] regions = pauses.speech_regions(values, 6 * 1536 - 7, 48000, 0.35) assert regions == [(1536, 4608), (7680, 6 * 1536 - 7)] @pytest.mark.parametrize("values", [[0.5], [0.5, float("nan")], [0.5, 1.1], [0.5, "bad"]]) def test_invalid_detector_output_is_not_silently_accepted(values): with pytest.raises(RuntimeError): pauses.speech_regions(values, 3072, 48000, 0.35) def test_padding_merges_close_words_but_keeps_long_gaps(): settings = resolve() # One millisecond per sample makes these expected boundaries easy to read. speech = [(500, 800), (1000, 1300), (2000, 2300)] assert pauses.pause_regions(speech, 3000, 1000, settings) == [ (0, 380), (1420, 1880), (2420, 3000)] # No room for padding plus fade plus the requested zero core: leave alone. assert pauses.pause_regions([(0, 100), (400, 1000)], 1000, 1000, settings) == [] def test_streamed_envelope_preserves_onsets_and_duration(wav, tmp_path, progress): sr = 48000 audio = np.full(3 * sr + 7, 0.001, dtype="float32") source = wav(audio) target = tmp_path / "silenced.wav" # Deliberately straddle the 65536-sample processing boundary. a, b, fade = 60000, 100000, 480 pauses.silence_pauses(source, target, [(a, b)], 10, progress) result, rate = sf.read(target, dtype="float32") assert rate == sr and result.shape == audio.shape np.testing.assert_array_equal(result[:a + 1], audio[:a + 1]) np.testing.assert_array_equal(result[b - 1:], audio[b - 1:]) assert np.count_nonzero(result[a + fade:b - fade]) == 0 assert np.all(np.diff(result[a:a + fade]) <= 0) assert np.all(np.diff(result[b - fade:b]) >= 0) assert np.max(np.abs(np.diff(result))) < 0.00001 def test_file_edges_do_not_fade_noise_back_in(wav, tmp_path, progress): source = wav(np.full(48000, 0.001)) target = tmp_path / "edges.wav" pauses.silence_pauses(source, target, [(0, 12000), (36000, 48000)], 10, progress) data = sf.read(target)[0] assert np.count_nonzero(data[:11520]) == 0 assert np.count_nonzero(data[36480:]) == 0 np.testing.assert_array_equal(data[12000:36000], sf.read(source)[0][12000:36000]) def test_no_intervals_is_exact_bypass(wav, tmp_path, progress): source = wav(np.random.default_rng(0).normal(0, 0.01, 200001)) target = tmp_path / "out.wav" pauses.silence_pauses(source, target, [], 10, progress) np.testing.assert_array_equal(sf.read(source)[0], sf.read(target)[0]) with pytest.raises(ValueError, match="distinct"): pauses.silence_pauses(source, source, [], 10, progress) @pytest.mark.parametrize("rate", [44100, 48000]) def test_pipeline_detects_before_denoise_and_exports_zero_cores( wav, tmp_path, progress, ffmpeg, monkeypatch, rate): sr = 48000 t = np.arange(2 * sr) / sr speech = 0.15 * np.sin(2 * np.pi * 220 * t) audio = np.concatenate([speech, np.full(sr, 0.001), speech]) source = wav(audio) original = source.read_bytes() events = [] def detect(prepared, *args): events.append("detect") data = sf.read(prepared)[0] assert np.max(np.abs(data)) > 0.14 # Not the attenuated denoiser output. return [(2 * sr + 5760, 3 * sr - 5760)], {"speech_regions": 2, "pauses_silenced": 1} def denoise(prepared, target, *args, **kwargs): assert events == ["detect"] events.append("denoise") data, sr = sf.read(prepared) sf.write(target, data * 0.8, sr, subtype="FLOAT") monkeypatch.setattr(pauses, "detect_pauses", detect) monkeypatch.setattr("producer.ai.denoise", denoise) settings = resolve(overrides={"pause_silence": True, "sample_rate": rate}) target = tmp_path / "master.wav" report = pipeline.process(source, target, settings, progress) data, sr = sf.read(target) assert sr == rate and len(data) == 5 * rate # Resampling and dynamics may have tails at edges; the interior is zero, # even after full normalization and final 32-bit encoding. assert np.count_nonzero(data[int(2.3 * rate):int(2.7 * rate)]) == 0 assert report["pause_cleanup"]["pauses_silenced"] == 1 assert report["silence_removed_seconds"] == 0 assert abs(report["master_loudness"]["input_i"] - settings.target_lufs) <= 0.5 assert report["master_loudness"]["input_tp"] <= settings.true_peak_db + 0.1 assert source.read_bytes() == original def test_no_speech_fails_open_and_skips_requested_trim( wav, tmp_path, bypass, progress, ffmpeg, monkeypatch): bypass.pause_silence = bypass.trim_silence = True monkeypatch.setattr(pauses, "detect_pauses", lambda *args: ([], {"speech_regions": 0})) source = wav() report = pipeline.process(source, tmp_path / "out.wav", bypass, progress) assert report["master"]["frames"] == report["input"]["frames"] assert any("found no speech" in warning for warning in report["warnings"]) assert report["gaps_trimmed"] == 0 def test_detector_failure_does_not_publish_or_fallback( wav, tmp_path, bypass, progress, ffmpeg, monkeypatch): bypass.pause_silence = True def fail(*args): raise RuntimeError("injected detector failure") monkeypatch.setattr(pauses, "detect_pauses", fail) target = tmp_path / "master.wav" target.write_bytes(b"previous") with pytest.raises(RuntimeError, match="injected detector"): pipeline.process(wav(), target, bypass, progress, overwrite=True) assert target.read_bytes() == b"previous" assert not list(tmp_path.glob(".producer-*")) def test_vad_trimming_only_removes_zero_cores(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch): sr = 48000 bypass.pause_silence = bypass.trim_silence = True bypass.gap_threshold_db = -20 # Would eat this entire quiet recording in legacy mode. speech = np.full(sr, 0.005) source = wav(np.concatenate([speech, np.full(3 * sr, 0.001), speech])) monkeypatch.setattr(pauses, "detect_pauses", lambda *args: ([(sr + 5760, 4 * sr - 5760)], {"speech_regions": 2})) target = tmp_path / "out.wav" report = pipeline.process(source, target, bypass, progress) data = sf.read(target)[0] np.testing.assert_allclose(data[:sr], speech, atol=1e-8) np.testing.assert_allclose(data[-sr:], speech, atol=1e-8) assert report["gaps_trimmed"] == 1 assert report["silence_removed_seconds"] > 1 def test_worker_uses_state_context_and_pads_partial_last_frame(wav, tmp_path, monkeypatch): from producer.vad_worker import probabilities calls = [] class Session: def __init__(self, *args, **kwargs): assert kwargs["providers"] == ["CPUExecutionProvider"] def run(self, _, inputs): calls.append(inputs) return np.array([[0.75]]), inputs["state"] + 1 monkeypatch.setitem(sys.modules, "onnxruntime", SimpleNamespace( SessionOptions=SimpleNamespace, InferenceSession=Session)) audio = np.arange(600, dtype="float32") / 600 path = tmp_path / "vad.wav" sf.write(path, audio, 16000, subtype="FLOAT") assert probabilities(path, tmp_path / "model") == [0.75, 0.75] assert calls[0]["input"].shape == (1, 576) assert np.count_nonzero(calls[0]["input"][0, :64]) == 0 np.testing.assert_array_equal(calls[1]["input"][0, :64], audio[448:512]) assert np.count_nonzero(calls[1]["input"][0, 152:]) == 0 assert np.all(calls[1]["state"] == 1) def test_model_download_is_verified_and_cached(tmp_path, monkeypatch, progress): root = tmp_path / "runtime" monkeypatch.setattr(pauses, "data_dir", lambda: root) environment = root / "vad-onnx-1.20.1-py311-v1" (environment / "bin").mkdir(parents=True) (environment / "bin/python").touch() (environment / ".producer-ready").touch() payload = b"test model" monkeypatch.setattr(pauses, "MODEL_SHA256", hashlib.sha256(payload).hexdigest()) downloads = [] def download(*args, **kwargs): downloads.append(args) return io.BytesIO(payload) monkeypatch.setattr(pauses.urllib.request, "urlopen", download) python, model = pauses.ensure_vad(progress) assert model.read_bytes() == payload assert pauses.ensure_vad(progress) == (python, model) assert len(downloads) == 1 model.write_bytes(b"corrupt cached model") monkeypatch.setattr(pauses.urllib.request, "urlopen", lambda *a, **k: io.BytesIO(b"bad")) with pytest.raises(RuntimeError, match="checksum mismatch"): pauses.ensure_vad(progress) assert model.read_bytes() == b"corrupt cached model" assert not list(model.parent.glob(".silero-*"))