From 464edbc4fd658728f06d3a7d98340aa28ff3e3ec Mon Sep 17 00:00:00 2001 From: historia Date: Sat, 12 Sep 2026 15:25:06 -0400 Subject: feat: slop web gui --- lib/project/tests/test_pauses.py | 200 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 200 insertions(+) create mode 100644 lib/project/tests/test_pauses.py (limited to 'lib/project/tests/test_pauses.py') diff --git a/lib/project/tests/test_pauses.py b/lib/project/tests/test_pauses.py new file mode 100644 index 0000000..b67f98d --- /dev/null +++ b/lib/project/tests/test_pauses.py @@ -0,0 +1,200 @@ +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-*")) -- cgit v1.2.3