1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
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-*"))
|
