aboutsummaryrefslogtreecommitdiff
path: root/lib/tests
diff options
context:
space:
mode:
Diffstat (limited to 'lib/tests')
-rw-r--r--lib/tests/test_chunking.py250
-rw-r--r--lib/tests/test_cli.py127
-rw-r--r--lib/tests/test_engines.py396
-rw-r--r--lib/tests/test_lazy.py286
-rw-r--r--lib/tests/test_meters.py21
-rw-r--r--lib/tests/test_pipeline.py44
-rw-r--r--lib/tests/test_ui.py169
-rw-r--r--lib/tests/test_updates.py221
8 files changed, 1514 insertions, 0 deletions
diff --git a/lib/tests/test_chunking.py b/lib/tests/test_chunking.py
new file mode 100644
index 0000000..f00c04e
--- /dev/null
+++ b/lib/tests/test_chunking.py
@@ -0,0 +1,250 @@
+import numpy as np
+import pytest
+
+from producer.engines.base import blend
+from producer.engines.chunking import apply_chunked, plan_chunks, stitch
+
+
+def test_plan_chunks_basic():
+ assert plan_chunks(0, 5, 2) == []
+ assert plan_chunks(4, 10, 2) == [(0, 4)]
+ assert plan_chunks(10, 5, 0) == [(0, 5), (5, 10)]
+ assert plan_chunks(10, 5, 2) == [(0, 5), (3, 8), (6, 10)]
+ assert plan_chunks(7, 5, 2) == [(0, 5), (3, 7)]
+
+
+def test_plan_chunks_last_span_covers_tail():
+ spans = plan_chunks(11, 5, 4)
+ assert spans[-1][1] == 11
+ assert spans[-1][1] - spans[-1][0] >= 5
+
+
+def test_plan_chunks_rejects_overlap_ge_chunk():
+ with pytest.raises(ValueError):
+ plan_chunks(100, 5, 5)
+ with pytest.raises(ValueError):
+ plan_chunks(100, 5, 6)
+
+
+def test_stitch_identity_pieces_reconstruct_signal():
+ rng = np.random.default_rng(0)
+ x = rng.standard_normal(10_000).astype(np.float32)
+ spans = plan_chunks(x.size, 2500, 300)
+ y = stitch(spans, x.size, (x[a:b] for a, b in spans))
+ assert y.dtype == np.float32
+ assert y.shape == x.shape
+ np.testing.assert_allclose(y, x, atol=1e-5)
+
+
+def test_stitch_single_span_passthrough():
+ x = np.arange(100, dtype=np.float32)
+ y = stitch([(0, 100)], 100, [x])
+ np.testing.assert_array_equal(y, x)
+
+
+def test_stitch_crossfade_of_complementary_pieces():
+ # piece A is silence, piece B is a full-scale ramp; the crossfade zone
+ # must be a smooth blend, not a jump.
+ spans = [(0, 10), (5, 15)]
+ pieces = [np.zeros(10, dtype=np.float32), np.ones(10, dtype=np.float32)]
+ y = stitch(spans, 15, pieces)
+ mid = y[7] # 50% through the overlap
+ assert 0.4 < mid < 0.6
+ assert y[0] == 0.0
+ assert y[14] == 1.0
+
+
+def test_stitch_pads_short_pieces():
+ spans = [(0, 10), (5, 15)]
+ pieces = [np.arange(10, dtype=np.float32), np.arange(8, dtype=np.float32)]
+ y = stitch(spans, 15, pieces)
+ assert y.shape == (15,)
+ assert np.all(np.isfinite(y))
+
+
+def test_apply_chunked_identity_matches_whole_file():
+ rng = np.random.default_rng(1)
+ x = rng.standard_normal(50_000).astype(np.float32)
+ sr = 8000
+ calls = []
+
+ def fn(chunk):
+ calls.append(chunk.size)
+ return chunk
+
+ y = apply_chunked(x, sr, 7.0, 0.05, fn)
+ assert calls == [50_000]
+ np.testing.assert_array_equal(y, x)
+
+ calls.clear()
+ y = apply_chunked(x, sr, 0.5, 0.05, fn)
+ assert len(calls) >= 10
+ assert all(c <= 4000 for c in calls)
+ np.testing.assert_allclose(y, x, atol=1e-5)
+
+
+def test_apply_chunked_zero_disables_chunking():
+ x = np.zeros(10, dtype=np.float32)
+ calls = []
+
+ def fn(chunk):
+ calls.append(chunk.size)
+ return chunk
+
+ apply_chunked(x, 8000, 0.0, 0.5, fn)
+ assert calls == [10]
+
+
+def test_apply_chunked_context_reconstructs_identity():
+ rng = np.random.default_rng(3)
+ x = rng.standard_normal(30_000).astype(np.float32)
+ y = apply_chunked(x, 8000, 1.5, 0.25, lambda c: c, context_s=0.5)
+ np.testing.assert_allclose(y, x, atol=1e-5)
+
+
+def test_apply_chunked_context_feeds_padded_chunks():
+ x = np.ones(5_000, dtype=np.float32)
+ sizes = []
+
+ def fn(chunk):
+ sizes.append(chunk.size)
+ return chunk
+
+ apply_chunked(x, 1000, 2.0, 0.0, fn, context_s=0.5)
+ # plan_chunks(5000, 2000, 0) -> (0,2000), (2000,4000), (4000,5000)
+ assert sizes == [2500, 3000, 1500]
+
+
+def test_apply_chunked_context_trims_cold_start_artifacts():
+ x = np.ones(6_000, dtype=np.float32)
+
+ def fn(chunk):
+ out = chunk.copy()
+ out[0] = 0.0 # cold-start artifact at the start of every model call
+ return out
+
+ broken = apply_chunked(x, 1000, 2.0, 0.0, fn)
+ assert int(np.sum(broken == 0.0)) > 1
+
+ y = apply_chunked(x, 1000, 2.0, 0.0, fn, context_s=0.5)
+ assert y[0] == 0.0 # only the true file start stays degraded
+ assert np.all(y[1:] == 1.0)
+
+
+def test_apply_chunked_context_survives_oom_retry():
+ x = np.ones(8_000, dtype=np.float32)
+ attempts = []
+
+ def fn(chunk):
+ attempts.append(chunk.size)
+ if chunk.size > 3000:
+ raise MemoryError("simulated oom")
+ out = chunk.copy()
+ out[0] = 0.0
+ return out
+
+ y = apply_chunked(x, 1000, 6.0, 0.0, fn, min_chunk_s=1.0, context_s=0.5)
+ assert max(attempts) > 3000
+ assert y[0] == 0.0
+ assert np.all(y[1:] == 1.0)
+
+
+def test_apply_chunked_retries_smaller_on_oom():
+ x = np.random.default_rng(2).standard_normal(40_000).astype(np.float32)
+ sr = 8000
+ attempts = []
+
+ def fn(chunk):
+ attempts.append(chunk.size)
+ if chunk.size > 3000:
+ raise MemoryError("simulated oom")
+ return chunk
+
+ y = apply_chunked(x, sr, 5.0, 0.0, fn, min_chunk_s=0.25)
+ assert attempts[0] == 40_000
+ assert attempts[-1] == 2500
+ assert max(a for a in attempts if a <= 3000) == 2500
+ np.testing.assert_allclose(y, x, atol=1e-5)
+
+
+def test_apply_chunked_oom_gives_up_at_min_chunk():
+ x = np.zeros(40_000, dtype=np.float32)
+
+ def fn(chunk):
+ raise MemoryError("always")
+
+ with pytest.raises(MemoryError):
+ apply_chunked(x, 8000, 1.0, 0.0, fn, min_chunk_s=0.5)
+
+
+def test_apply_chunked_progress_callback():
+ x = np.random.default_rng(3).standard_normal(20_000).astype(np.float32)
+ sr = 8000
+ seen: list[tuple[int, int]] = []
+ y = apply_chunked(x, sr, 0.5, 0.0, lambda c: c, on_progress=lambda d, t: seen.append((d, t)))
+ total = len(plan_chunks(x.size, round(0.5 * sr), 0))
+ assert seen[0] == (0, total)
+ assert seen[-1] == (total, total)
+ assert [d for d, _t in seen[1:]] == list(range(1, total + 1))
+ np.testing.assert_allclose(y, x, atol=1e-6)
+
+
+def test_apply_chunked_progress_resets_after_oom_retry():
+ x = np.random.default_rng(4).standard_normal(40_000).astype(np.float32)
+ sr = 8000
+ seen: list[tuple[int, int]] = []
+
+ def fn(chunk):
+ if chunk.size > 3000:
+ raise MemoryError("simulated oom")
+ return chunk
+
+ y = apply_chunked(
+ x, sr, 5.0, 0.0, fn, min_chunk_s=0.25, on_progress=lambda d, t: seen.append((d, t))
+ )
+ assert seen[0] == (0, 1) # first attempt is one whole-file chunk
+ done, total = seen[-1]
+ assert done == total and total > 2 # retried into smaller chunks
+ np.testing.assert_allclose(y, x, atol=1e-5)
+
+
+def test_apply_chunked_progress_absent_when_disabled():
+ x = np.zeros(10_000, dtype=np.float32)
+ y = apply_chunked(x, 8000, 0.5, 0.0, lambda c: c, on_progress=None)
+ assert y.shape == x.shape
+
+
+def test_apply_chunked_whole_file_mode_falls_back_on_oom():
+ x = np.random.default_rng(5).standard_normal(10_000).astype(np.float32)
+ sr = 1000
+
+ def fn(chunk):
+ if chunk.size > 1500:
+ raise MemoryError("simulated oom")
+ return chunk * 2
+
+ y = apply_chunked(x, sr, 0.0, 0.0, fn, min_chunk_s=0.5)
+ np.testing.assert_allclose(y, x * 2, atol=1e-6)
+
+
+def test_blend_float32_no_float64_temporaries():
+ x = np.full(1000, 0.25, dtype=np.float32)
+ y = np.full(1000, 0.75, dtype=np.float32)
+ out = blend(x, y, 0.5)
+ assert out.dtype == np.float32
+ np.testing.assert_allclose(out, 0.5, atol=1e-7)
+ assert blend(x, y, 1.0) is y
+ np.testing.assert_allclose(blend(x, y, 0.0), x, atol=1e-7)
+
+
+def test_blend_realigns_resample_drift():
+ # a 48k -> 16k -> 48k round trip can come back a sample or two long
+ # (resample_poly emits ceil(n * up/down) per hop); blend must cope
+ x = np.ones(5, dtype=np.float32)
+ long_y = np.full(7, 0.5, dtype=np.float32)
+ np.testing.assert_allclose(blend(x, long_y, 0.5), np.full(5, 0.75), atol=1e-7)
+
+ short_y = np.full(4, 0.5, dtype=np.float32)
+ out = blend(x, short_y, 0.5)
+ assert out.shape == x.shape
+ np.testing.assert_allclose(out, [0.75, 0.75, 0.75, 0.75, 0.5], atol=1e-7)
diff --git a/lib/tests/test_cli.py b/lib/tests/test_cli.py
index 1a157f7..76952ba 100644
--- a/lib/tests/test_cli.py
+++ b/lib/tests/test_cli.py
@@ -2,6 +2,7 @@ import sys
from types import SimpleNamespace
import numpy as np
+import pytest
import soundfile as sf
from conftest import speechish
@@ -165,6 +166,74 @@ def test_default_output_suffix_processed(tmp_path):
assert _resolve_output(inp, opts, single=False) == odir / "song_processed.wav"
+def test_engine_chunk_flags():
+ args = build_parser().parse_args(["in.wav", "--engine-chunk", "15", "--engine-overlap", "1.0"])
+ opts = Options()
+ _apply_args(opts, args)
+ assert opts.engine_chunk_s == 15.0
+ assert opts.engine_overlap_s == 1.0
+
+
+def test_default_options_whole_file_and_soft_denoise():
+ opts = Options()
+ assert opts.engine_chunk_s == 0.0
+ assert opts.denoise_strength == 0.9
+
+
+def test_engine_chunk_validation():
+ with pytest.raises(SystemExit):
+ args = build_parser().parse_args(["in.wav", "--engine-chunk", "5", "--engine-overlap", "5"])
+ _apply_args(Options(), args)
+ with pytest.raises(SystemExit):
+ args = build_parser().parse_args(["in.wav", "--engine-chunk", "-1"])
+ _apply_args(Options(), args)
+
+
+def test_engine_chunk_config_override(tmp_path):
+ from producer import config as cfgmod
+
+ cfg = tmp_path / "config.toml"
+ cfg.write_text("engine_chunk = 10.0\nengine_overlap = 1.0\n")
+ opts = Options()
+ cfgmod.apply_config(opts, cfgmod.load_config(cfg))
+ assert opts.engine_chunk_s == 10.0
+ assert opts.engine_overlap_s == 1.0
+
+
+def test_denoise_pf_flag_and_config_plumbing(tmp_path):
+ from producer import config as cfgmod
+
+ opts = Options()
+ _apply_args(opts, build_parser().parse_args(["in.wav"]))
+ assert opts.denoise_pf is False
+
+ opts = Options()
+ _apply_args(opts, build_parser().parse_args(["in.wav", "--denoise-pf"]))
+ assert opts.denoise_pf is True
+
+ cfg = tmp_path / "config.toml"
+ cfg.write_text('[denoise]\nengine = "dfn3"\nstrength = 0.8\npf = true\n')
+ opts = Options()
+ cfgmod.apply_config(opts, cfgmod.load_config(cfg))
+ assert opts.denoise_strength == 0.8
+ assert opts.denoise_pf is True
+
+
+def test_spectral_denoise_choice(tmp_path):
+ from producer import config as cfgmod
+
+ opts = Options()
+ _apply_args(opts, build_parser().parse_args(["in.wav", "--denoise", "spectral"]))
+ assert opts.denoise == "spectral"
+
+ cfg = tmp_path / "config.toml"
+ cfg.write_text('[denoise]\nengine = "spectral"\nstrength = 0.8\n')
+ opts = Options()
+ cfgmod.apply_config(opts, cfgmod.load_config(cfg))
+ assert opts.denoise == "spectral"
+ assert opts.denoise_strength == 0.8
+
+
def test_conflict_auto_renames_when_not_a_tty(tmp_path):
inp = _mk_wav(tmp_path, "in.wav")
first = process_one(inp, _opts(), single=True)
@@ -228,3 +297,61 @@ def test_conflict_prompt_eof_cancels(tmp_path, monkeypatch):
monkeypatch.setattr("builtins.input", _eof)
assert process_one(inp, _opts(), single=True) is None
+
+
+def test_process_one_reports_stage_status(tmp_path, capsys):
+ inp = _mk_wav(tmp_path, "status.wav")
+ out_path = process_one(inp, _opts(), single=True)
+ assert out_path is not None
+ out = capsys.readouterr().out
+ assert f"[producer] {inp} (" in out
+ assert "dsp done in" in out
+ assert "levelling done in" in out
+ assert f"[producer] wrote {out_path}" in out
+
+
+def test_multi_file_run_gets_position_prefixes(tmp_path, capsys):
+ from producer.cli import main
+
+ a = _mk_wav(tmp_path, "a.wav")
+ b = _mk_wav(tmp_path, "b.wav")
+ rc = main([str(a), str(b), "--denoise", "off"])
+ assert rc == 0
+ out = capsys.readouterr().out
+ assert "[1/2]" in out and "[2/2]" in out
+ assert (tmp_path / "a_processed.wav").exists()
+ assert (tmp_path / "b_processed.wav").exists()
+
+
+def test_output_file_rejected_for_multiple_inputs(tmp_path):
+ from producer.cli import main
+
+ a = _mk_wav(tmp_path, "a.wav")
+ b = _mk_wav(tmp_path, "b.wav")
+ with pytest.raises(SystemExit):
+ main([str(a), str(b), "-o", str(tmp_path / "out.wav"), "--denoise", "off"])
+
+
+def test_output_dir_accepted_for_multiple_inputs(tmp_path):
+ from producer.cli import main
+
+ a = _mk_wav(tmp_path, "a.wav")
+ b = _mk_wav(tmp_path, "b.wav")
+ outdir = tmp_path / "masters"
+ rc = main([str(a), str(b), "-o", str(outdir), "--denoise", "off"])
+ assert rc == 0
+ assert (outdir / "a_processed.wav").exists()
+ assert (outdir / "b_processed.wav").exists()
+
+
+def test_batch_failure_continues_to_next_file(tmp_path, capsys):
+ from producer.cli import main
+
+ good = _mk_wav(tmp_path, "good.wav")
+ missing = tmp_path / "missing.wav"
+ rc = main([str(missing), str(good), "--denoise", "off"])
+ assert rc == 1
+ captured = capsys.readouterr()
+ assert "[2/2]" in captured.out
+ assert "[producer] ERROR" in captured.err
+ assert (tmp_path / "good_processed.wav").exists()
diff --git a/lib/tests/test_engines.py b/lib/tests/test_engines.py
index 781b46e..9881859 100644
--- a/lib/tests/test_engines.py
+++ b/lib/tests/test_engines.py
@@ -12,6 +12,365 @@ AUDIO_META_MSG = (
)
+class _FakeTensor:
+ def __init__(self, arr):
+ self.arr = np.asarray(arr)
+
+ def unsqueeze(self, axis):
+ return _FakeTensor(self.arr[None, ...])
+
+ def detach(self):
+ return self
+
+ def cpu(self):
+ return self
+
+ def numpy(self):
+ return self.arr
+
+
+def _install_torch_stub(monkeypatch):
+ fake = types.ModuleType("torch")
+ fake.Tensor = _FakeTensor
+ fake.from_numpy = lambda a: _FakeTensor(a)
+ fake.cuda = types.SimpleNamespace(is_available=lambda: False)
+ monkeypatch.setitem(sys.modules, "torch", fake)
+
+
+def _install_df_enhance_stub(monkeypatch, transform, calls, records=None):
+ pkg = types.ModuleType("df")
+
+ class _State:
+ def sr(self):
+ return 44100
+
+ class _Model:
+ def to(self, dev):
+ return self
+
+ def init_df(*_args, **kwargs):
+ if records is not None:
+ records["init_df"] = kwargs
+ return _Model(), _State(), "fake"
+
+ def enhance(_model, _state, audio, **kwargs):
+ calls.append(audio.arr.shape[-1])
+ if records is not None:
+ records["enhance"] = kwargs
+ return _FakeTensor(transform(audio.arr.copy()))
+
+ pkg.enhance = types.ModuleType("df.enhance")
+ pkg.enhance.init_df = init_df
+ pkg.enhance.enhance = enhance
+ for sub in ("df.io", "df.logger", "df.utils"):
+ mod = types.ModuleType(sub)
+ setattr(pkg, sub.split(".")[1], mod)
+ monkeypatch.setitem(sys.modules, sub, mod)
+ monkeypatch.setitem(sys.modules, "df", pkg)
+ monkeypatch.setitem(sys.modules, "df.enhance", pkg.enhance)
+
+
+def test_dfn3_chunked_engine_stitches_full_length(monkeypatch, sr, noisy_speech):
+ from producer.engines import denoise_dfn
+
+ monkeypatch.setattr("producer.lazy.ensure", lambda *_a, **_k: None)
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: None)
+ _install_torch_stub(monkeypatch)
+ calls: list[int] = []
+ _install_df_enhance_stub(monkeypatch, lambda a: a * 0.5 + 0.001, calls)
+
+ x = noisy_speech[: sr * 4]
+ y, eng, dev = denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=1.0, overlap_s=0.1)
+ assert "dfn" in eng
+ assert "cpu" in dev
+ assert len(calls) > 2
+ assert y.shape == x.shape
+ np.testing.assert_allclose(y, x * 0.5 + 0.001, atol=1e-6)
+
+
+def test_dfn3_whole_file_mode_single_call(monkeypatch, sr, noisy_speech):
+ from producer.engines import denoise_dfn
+
+ monkeypatch.setattr("producer.lazy.ensure", lambda *_a, **_k: None)
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: None)
+ _install_torch_stub(monkeypatch)
+ calls: list[int] = []
+ _install_df_enhance_stub(monkeypatch, lambda a: a * 0.5, calls)
+
+ x = noisy_speech[: sr * 2]
+ y, _eng, _dev = denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=0.0)
+ assert calls == [x.size]
+ np.testing.assert_allclose(y, x * 0.5, atol=1e-7)
+
+
+def test_dfn3_post_filter_opt_in_no_atten_lim(monkeypatch, sr, noisy_speech):
+ from producer.engines import denoise_dfn
+
+ monkeypatch.setattr("producer.lazy.ensure", lambda *_a, **_k: None)
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: None)
+ _install_torch_stub(monkeypatch)
+ calls: list[int] = []
+ records: dict = {}
+ _install_df_enhance_stub(monkeypatch, lambda a: a * 0.5, calls, records)
+
+ x = noisy_speech[: sr * 2]
+ denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=0.0)
+ assert records["init_df"]["post_filter"] is False
+ assert records["enhance"] == {} # stock df_enhance call, no atten-lim override
+
+ calls.clear()
+ records.clear()
+ denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=0.0, post_filter=True)
+ assert records["init_df"]["post_filter"] is True
+
+
+def test_dfn3_chunked_feeds_context_padding(monkeypatch, sr, noisy_speech):
+ from producer.engines import denoise_dfn
+
+ monkeypatch.setattr("producer.lazy.ensure", lambda *_a, **_k: None)
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: None)
+ _install_torch_stub(monkeypatch)
+ calls: list[int] = []
+ _install_df_enhance_stub(monkeypatch, lambda a: a * 0.5, calls)
+
+ x = noisy_speech[: sr * 10]
+ y, _eng, _dev = denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=3.0, overlap_s=0.5)
+ # chunks are widened with context, then trimmed back to the spans
+ assert max(calls) > 3.0 * sr
+ assert min(calls) >= 3.0 * sr
+ np.testing.assert_allclose(y, x * 0.5, atol=1e-7)
+
+
+_Z_MODEL_REPO = "iic/speech_zipenhancer_ans_multiloss_16k_base"
+
+
+def _install_zipenhancer_stub(monkeypatch, calls):
+ fake = types.ModuleType("zipenhancer")
+ fake.MODEL_ZIPENHANCER = _Z_MODEL_REPO
+
+ def denoise(chunk, sample_rate, model=_Z_MODEL_REPO, normalize=True, strength=1.0, **_kw):
+ calls.append(
+ {"n": chunk.size, "model": model, "normalize": normalize, "strength": strength}
+ )
+ scale = 0.1 if len(calls) % 2 == 1 else 1.0
+ return (chunk * scale, 0.0, chunk.size / sample_rate)
+
+ fake.denoise = denoise
+ monkeypatch.setitem(sys.modules, "zipenhancer", fake)
+
+
+def test_zipenhancer_chunked_normalizes_once(monkeypatch, sr):
+ from producer.engines import denoise_zip
+
+ seq: list[str] = []
+ monkeypatch.setattr(
+ "producer.lazy.ensure",
+ lambda pkgs, **_k: seq.append("ensure:" + ",".join(str(p) for p in pkgs)),
+ )
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: seq.append("torch"))
+ monkeypatch.setattr(
+ "producer.lazy.ensure_import", lambda mod, **_k: seq.append("import:" + mod)
+ )
+ monkeypatch.setattr("producer.lazy.ensure_call", lambda fn, **_k: (seq.append("call"), fn())[1])
+ calls: list[dict] = []
+ _install_zipenhancer_stub(monkeypatch, calls)
+
+ n = 4 * 44100
+ t = np.arange(n) / sr
+ x = (0.5 * np.sin(2 * np.pi * 160.0 * t)).astype(np.float32)
+ y, eng, _dev = denoise_zip.denoise(x, sr, 1.0, "cpu", chunk_s=1.0, overlap_s=0.0)
+ assert "zipenhancer" in eng
+ # torch is pinned before package installs; the undeclared modelscope
+ # import ships with the engine; import and call probes run after both
+ assert seq == [
+ "torch",
+ "ensure:zipenhancer==0.3.2,modelscope",
+ "import:zipenhancer",
+ "call",
+ "call",
+ "call",
+ "call",
+ ]
+ assert len(calls) == 4
+ # the library API takes the full modelscope repo id; the short name
+ # would be treated as a repo id and fail with modelscope E3021
+ assert all(c["model"] == _Z_MODEL_REPO for c in calls)
+ assert all(c["normalize"] is False for c in calls)
+ assert y.shape == x.shape
+ peak = float(np.max(np.abs(y)))
+ assert abs(peak - 10 ** (-3.0 / 20.0)) < 0.01
+ even_rms = float(np.sqrt(np.mean(y[: n // 4].astype(np.float64) ** 2)))
+ odd_rms = float(np.sqrt(np.mean(y[n // 4 : n // 2].astype(np.float64) ** 2)))
+ assert 8.0 < odd_rms / even_rms < 12.0
+
+
+def test_zipenhancer_resample_roundtrip_length_realigned(monkeypatch):
+ # 120s @ 48k round-tripped through the 16k engine can come back a sample
+ # or two long (resample_poly emits ceil(n * up/down) per hop); the blend
+ # used to crash on the mismatch instead of realigning
+ from producer.engines import denoise_zip
+
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: None)
+ monkeypatch.setattr("producer.lazy.ensure", lambda *_a, **_k: None)
+ calls: list[dict] = []
+ _install_zipenhancer_stub(monkeypatch, calls)
+
+ sr = 48000
+ n = 100001 # 48k -> 16k -> 48k drifts +1 for this length
+ x = (0.3 * np.sin(2 * np.pi * 220.0 * np.arange(n) / sr)).astype(np.float32)
+ y, _eng, _dev = denoise_zip.denoise(x, sr, 0.5, "cpu")
+ assert y.shape == x.shape
+
+
+def test_mossformer_clearvoice_probe(monkeypatch, sr, noisy_speech):
+ from producer.engines import enhance_mossformer
+
+ seq: list[str] = []
+ monkeypatch.setattr("producer.lazy.ensure_torch", lambda: seq.append("torch"))
+ monkeypatch.setattr(
+ "producer.lazy.ensure",
+ lambda pkgs, **_k: seq.append("ensure:" + ",".join(str(p) for p in pkgs)),
+ )
+ monkeypatch.setattr(
+ "producer.lazy.ensure_import", lambda mod, **_k: seq.append("import:" + mod)
+ )
+
+ fake = types.ModuleType("clearvoice")
+
+ class _FakeCV:
+ def __init__(self, task=None, model_name=None):
+ pass
+
+ def __call__(self, chunk):
+ return (chunk * 0.5, 48000)
+
+ fake.ClearVoice = _FakeCV
+ monkeypatch.setitem(sys.modules, "clearvoice", fake)
+
+ x = noisy_speech[: sr * 4]
+ y, eng, _dev = enhance_mossformer.enhance(x, 48000, 1.0, "cpu", chunk_s=2.0)
+ assert "mossformer2" in eng
+ assert seq == ["torch", "ensure:clearvoice==0.1.2", "import:clearvoice"]
+ assert y.shape == x.shape
+ np.testing.assert_allclose(y, x * 0.5, atol=1e-6)
+
+
+def _hissy_speech(sr, dur=12.0, noise_db=-42.0, seed=3):
+ from conftest import speechish
+
+ x = speechish(dur, sr, level_dbfs=-20.0, seed=seed)
+ rng = np.random.default_rng(seed)
+ noise = rng.standard_normal(x.size)
+ noise *= (10 ** (noise_db / 20.0)) / np.sqrt(np.mean(np.square(noise)))
+ return (x + noise).astype(np.float32)
+
+
+def test_spectral_reduces_steady_noise(sr):
+ from producer import meters
+ from producer.engines import denoise_spectral
+
+ x = _hissy_speech(sr)
+ y, eng, dev = denoise_spectral.denoise(x, sr, 1.0, "cpu")
+ assert "spectral" in eng and dev == "cpu"
+ assert y.shape == x.shape
+ assert np.all(np.isfinite(y))
+ assert meters.noise_floor_db(y, sr) < meters.noise_floor_db(x, sr) - 8.0
+ assert np.corrcoef(x.astype(np.float64), y.astype(np.float64))[0, 1] > 0.8
+
+
+def test_spectral_speech_level_flat(sr):
+ from producer.engines import denoise_spectral
+
+ x = _hissy_speech(sr)
+ y, _eng, _dev = denoise_spectral.denoise(x, sr, 1.0, "cpu")
+ frame = int(0.03 * sr)
+ nf = x.size // frame
+
+ def frms_db(z):
+ rms = np.sqrt(np.mean(z[: nf * frame].reshape(nf, frame).astype(np.float64) ** 2, axis=1))
+ return 20.0 * np.log10(rms + 1e-12)
+
+ xdb, ydb = frms_db(x), frms_db(y)
+ speech = xdb > np.percentile(xdb, 10) + 12.0
+ assert speech.sum() > 20
+ # the deterministic engine must not pump the speech level (dfn3's failure mode)
+ swing = np.abs(ydb[speech] - xdb[speech])
+ assert np.percentile(swing, 95) < 2.0
+
+
+def test_spectral_strength_zero_is_identity(sr):
+ from producer.engines import denoise_spectral
+
+ x = _hissy_speech(sr)
+ y, _eng, _dev = denoise_spectral.denoise(x, sr, 0.0, "cpu")
+ np.testing.assert_array_equal(y, x)
+
+
+def test_spectral_sample_aligned(sr):
+ from producer.engines import denoise_spectral
+
+ x = _hissy_speech(sr)
+ y, _eng, _dev = denoise_spectral.denoise(x, sr, 1.0, "cpu")
+ lo, hi = int(sr * 2.0), int(sr * 9.0)
+ corrs = {k: float(np.corrcoef(x[lo:hi], y[lo + k : hi + k])[0, 1]) for k in range(-3, 4)}
+ assert max(corrs, key=corrs.get) == 0
+ assert corrs[0] > 0.8
+
+
+def test_spectral_suppression_capped(sr):
+ from producer import meters
+ from producer.engines import denoise_spectral
+
+ noise = (0.008 * np.random.default_rng(5).standard_normal(sr * 6)).astype(np.float32)
+ y, _eng, _dev = denoise_spectral.denoise(noise, sr, 1.0, "cpu")
+ drop = meters.noise_floor_db(noise, sr) - meters.noise_floor_db(y, sr)
+ # bounded: deep enough to matter, never gated to digital silence
+ assert 20.0 < drop < 40.0
+
+
+def test_spectral_chunked_matches_whole_file(sr):
+ from producer.engines import denoise_spectral
+
+ x = _hissy_speech(sr, dur=20.0)
+ y1, _e, _d = denoise_spectral.denoise(x, sr, 1.0, "cpu", chunk_s=60.0, overlap_s=0.5)
+ y2, _e, _d = denoise_spectral.denoise(x, sr, 1.0, "cpu", chunk_s=8.0, overlap_s=0.5)
+ assert y1.shape == y2.shape == x.shape
+ corr = np.corrcoef(y1.astype(np.float64), y2.astype(np.float64))[0, 1]
+ assert corr > 0.99
+
+
+def test_spectral_profile_global_not_per_chunk(sr):
+ """Tail hiss must get the same suppression with or without speech up front.
+
+ The old per-chunk percentile leaked speech into the noise estimate and
+ under-suppressed exactly where it matters (between sentences).
+ """
+ from conftest import speechish
+
+ from producer import meters
+ from producer.engines import denoise_spectral
+
+ speech = speechish(8.0, sr, level_dbfs=-20.0, seed=11)
+ rng = np.random.default_rng(9)
+ noise = rng.standard_normal(sr * 16).astype(np.float32)
+ noise *= (10 ** (-40.0 / 20.0)) / np.sqrt(np.mean(np.square(noise)))
+ tail_lo, tail_hi = sr * 10, sr * 16
+ with_speech = np.concatenate([speech + noise[: speech.size], noise[speech.size :]]).astype(
+ np.float32
+ )
+ hiss_only = noise.copy()
+ y1, _e, _d = denoise_spectral.denoise(with_speech, sr, 1.0, "cpu")
+ y2, _e, _d = denoise_spectral.denoise(hiss_only, sr, 1.0, "cpu")
+ drop1 = meters.noise_floor_db(with_speech[tail_lo:tail_hi], sr) - meters.noise_floor_db(
+ y1[tail_lo:tail_hi], sr
+ )
+ drop2 = meters.noise_floor_db(hiss_only[tail_lo:tail_hi], sr) - meters.noise_floor_db(
+ y2[tail_lo:tail_hi], sr
+ )
+ assert drop1 > 14.0
+ assert drop1 > drop2 - 4.0
+
+
def _stub_df_modules(calls: list[str]) -> tuple[types.ModuleType, list[types.ModuleType]]:
pkg = types.ModuleType("df")
mods = []
@@ -84,6 +443,43 @@ def test_dfn3_reduces_noise(sr, noisy_speech):
@pytest.mark.slow
+def test_dfn3_speech_level_flat(sr, noisy_speech):
+ """Guard against dfn3's reported failure mode: volume wobble in sentences."""
+ pytest.importorskip("torch")
+ pytest.importorskip("df")
+ from producer.engines import denoise_dfn
+
+ x = noisy_speech[: sr * 8]
+ y, _eng, _dev = denoise_dfn.denoise(x, sr, 0.9, "cpu")
+ frame = int(0.03 * sr)
+ nf = x.size // frame
+
+ def frms_db(z):
+ rms = np.sqrt(np.mean(z[: nf * frame].reshape(nf, frame).astype(np.float64) ** 2, axis=1))
+ return 20.0 * np.log10(rms + 1e-12)
+
+ xdb, ydb = frms_db(x), frms_db(y)
+ speech = xdb > np.percentile(xdb, 10) + 12.0
+ swing = np.abs(ydb[speech] - xdb[speech])
+ assert np.percentile(swing, 95) < 2.5
+
+
+@pytest.mark.slow
+def test_dfn3_chunked_matches_whole_file(sr, noisy_speech):
+ pytest.importorskip("torch")
+ pytest.importorskip("df")
+ from producer.engines import denoise_dfn
+
+ x = noisy_speech[: sr * 60]
+ y_full, _, _ = denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=0.0)
+ y_chunk, _, _ = denoise_dfn.denoise(x, sr, 1.0, "cpu", chunk_s=15.0, overlap_s=0.5)
+ assert y_full.shape == y_chunk.shape == x.shape
+ corr = np.corrcoef(y_full.astype(np.float64), y_chunk.astype(np.float64))[0, 1]
+ assert corr > 0.99
+ assert float(np.max(np.abs(y_full - y_chunk))) < 0.1
+
+
+@pytest.mark.slow
def test_zipenhancer_reduces_noise(sr, noisy_speech):
pytest.importorskip("torch")
pytest.importorskip("zipenhancer")
diff --git a/lib/tests/test_lazy.py b/lib/tests/test_lazy.py
new file mode 100644
index 0000000..5ab5b95
--- /dev/null
+++ b/lib/tests/test_lazy.py
@@ -0,0 +1,286 @@
+import types
+
+import pytest
+
+from producer import lazy
+
+PYPI_JSON = {
+ "urls": [
+ {
+ "filename": "torch-2.7.1-cp311-cp311-manylinux_2_28_x86_64.whl",
+ "url": "https://files.pythonhosted.org/packages/xx/torch-2.7.1-cp311-cp311-manylinux_2_28_x86_64.whl",
+ "digests": {"sha256": "abc123"},
+ "size": 766_668_798,
+ },
+ {
+ "filename": "torch-2.7.1-cp311-cp311-win_amd64.whl",
+ "url": "https://example.com/win.whl",
+ "digests": {},
+ "size": 1,
+ },
+ {
+ "filename": "torch-2.7.1.tar.gz",
+ "url": "https://example.com/src.tar.gz",
+ "digests": {},
+ "size": 1,
+ },
+ ]
+}
+
+_SHA = "e1a846516570851234567890abcdef1234567890abcdef1234567890abcdef12"
+GPU_HTML = (
+ "<html><body>"
+ '<a href="https://download-r2.pytorch.org/whl/cu126/torch-2.6.0%2Bcu126-cp311-cp311-manylinux_2_28_x86_64.whl#sha256=00'
+ f'00">old</a><a href="https://download-r2.pytorch.org/whl/cu126/torch-2.7.1%2Bcu126-cp311-cp311-manylinux_2_28_x86_64.whl#sha256={_SHA}">new</a>'
+ '<a href="https://download-r2.pytorch.org/whl/cu126/torch-2.7.1%2Bcu126-cp311-cp311-win_amd64.whl#sha256=1111">win</a>'
+ '<a href="torch-2.7.1%2Bcu126-cp311-cp311-manylinux_2_28_x86_64.whl">relative-dup</a>'
+ "</body></html>"
+)
+
+
+def test_choose_pypi_picks_linux_wheel():
+ got = lazy._choose_pypi(PYPI_JSON)
+ assert got is not None
+ name, url, sha, size = got
+ assert name.endswith("manylinux_2_28_x86_64.whl")
+ assert "pythonhosted" in url
+ assert sha == "abc123"
+ assert size == 766_668_798
+
+
+def test_choose_gpu_matches_version_and_arch():
+ got = lazy._choose_gpu(GPU_HTML, "torch", "2.7.1")
+ assert got is not None
+ name, url, sha, size = got
+ assert name == "torch-2.7.1+cu126-cp311-cp311-manylinux_2_28_x86_64.whl"
+ assert url == (
+ "https://download-r2.pytorch.org/whl/cu126/"
+ "torch-2.7.1%2Bcu126-cp311-cp311-manylinux_2_28_x86_64.whl"
+ )
+ assert sha == _SHA
+ assert size is None
+
+
+def test_choose_gpu_no_match_returns_none():
+ assert lazy._choose_gpu("<html></html>", "torch", "2.7.1") is None
+
+
+def test_resolve_torch_wheels_needs_both_packages(monkeypatch):
+ def fake(pkg, ver):
+ return None if pkg == "torchaudio" else ("t.whl", "u", None, 1)
+
+ monkeypatch.setattr(lazy, "_pypi_wheel", fake)
+ assert lazy._resolve_torch_wheels(gpu=False) == []
+
+
+def test_resolve_torch_wheels_returns_both(monkeypatch):
+ wheels = [("torch.whl", "u1", "s1", 1), ("torchaudio.whl", "u2", None, None)]
+ monkeypatch.setattr(lazy, "_pypi_wheel", lambda pkg, ver: wheels.pop(0))
+ got = lazy._resolve_torch_wheels(gpu=False)
+ assert [w[0] for w in got] == ["torch.whl", "torchaudio.whl"]
+
+
+def test_ensure_torch_downloads_wheels_then_installs(monkeypatch, tmp_path):
+ wheels = [
+ ("torch-2.7.1-cp311.whl", "https://x/torch.whl", "sha", 10),
+ ("torchaudio-2.7.1-cp311.whl", "https://x/ta.whl", None, 5),
+ ]
+ monkeypatch.setattr(lazy, "has_module", lambda name: False)
+ monkeypatch.setattr(lazy, "_resolve_torch_wheels", lambda gpu: wheels)
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(lazy, "WHEEL_CACHE", tmp_path / "wheels")
+ downloads: list[str] = []
+ runs: list[tuple[list[str], str]] = []
+
+ def fake_download(url, dest, label=None, expected_size=None, sha256=None, timeout=60.0):
+ downloads.append(dest.name)
+ return dest
+
+ monkeypatch.setattr(lazy.ui, "download", fake_download)
+ monkeypatch.setattr(lazy.ui, "run", lambda cmd, label, check=True: runs.append((cmd, label)))
+ lazy.ensure_torch()
+ assert downloads == ["torch-2.7.1-cp311.whl", "torchaudio-2.7.1-cp311.whl"]
+ assert len(runs) == 1
+ cmd = runs[0][0]
+ assert cmd[0] == "uv" and cmd[1] == "pip" and cmd[2] == "install"
+ assert str(tmp_path / "wheels" / "torch-2.7.1-cp311.whl") in cmd
+ assert str(tmp_path / "wheels" / "torchaudio-2.7.1-cp311.whl") in cmd
+
+
+def test_ensure_torch_falls_back_to_uv_index(monkeypatch):
+ monkeypatch.setattr(lazy, "has_module", lambda name: False)
+ monkeypatch.setattr(lazy, "_resolve_torch_wheels", lambda gpu: [])
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(lazy, "gpu_present", lambda: True)
+ runs: list[list[str]] = []
+ monkeypatch.setattr(lazy.ui, "run", lambda cmd, label, check=True: runs.append(cmd))
+ lazy.ensure_torch()
+ cmd = runs[0]
+ assert "torch==2.7.1+cu126" in cmd
+ assert "--index-url" in cmd
+ assert lazy.TORCH_GPU_INDEX in cmd
+
+
+def test_ensure_torch_falls_back_to_pypi_cpu(monkeypatch):
+ monkeypatch.setattr(lazy, "has_module", lambda name: False)
+ monkeypatch.setattr(lazy, "_resolve_torch_wheels", lambda gpu: [])
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(lazy, "gpu_present", lambda: False)
+ runs: list[list[str]] = []
+ monkeypatch.setattr(lazy.ui, "run", lambda cmd, label, check=True: runs.append(cmd))
+ lazy.ensure_torch()
+ cmd = runs[0]
+ assert "torch==2.7.1" in cmd
+ assert "--index-url" not in cmd
+
+
+def test_ensure_torch_noop_when_installed(monkeypatch):
+ monkeypatch.setattr(lazy, "has_module", lambda name: True)
+ runs: list = []
+ monkeypatch.setattr(lazy.ui, "run", lambda cmd, label, check=True: runs.append(cmd))
+ lazy.ensure_torch()
+ assert runs == []
+
+
+def test_ensure_import_noop_when_importable(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+ lazy.ensure_import("numpy", purpose="test")
+ assert installs == []
+
+
+def test_ensure_import_installs_missing_module(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+ state = {"round": 0}
+
+ def fake_import(name):
+ if name == "fakeengine":
+ if state["round"] == 0:
+ state["round"] = 1
+ raise ModuleNotFoundError("No module named 'addict'", name="addict")
+ return types.ModuleType("fakeengine")
+ raise ModuleNotFoundError(f"No module named {name!r}", name=name)
+
+ monkeypatch.setattr(lazy.importlib, "import_module", fake_import)
+ lazy.ensure_import("fakeengine", purpose="test")
+ assert installs == [["addict"]]
+
+
+def test_ensure_import_dotted_missing_reraises_without_install(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+
+ def fake_import(name):
+ raise ModuleNotFoundError("No module named 'pkg.sub'", name="pkg.sub")
+
+ monkeypatch.setattr(lazy.importlib, "import_module", fake_import)
+ with pytest.raises(ModuleNotFoundError):
+ lazy.ensure_import("whatever", purpose="test")
+ assert installs == []
+
+
+def test_ensure_import_gives_up_after_rounds(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+
+ def fake_import(name):
+ raise ModuleNotFoundError("No module named 'ghost'", name="ghost")
+
+ monkeypatch.setattr(lazy.importlib, "import_module", fake_import)
+ with pytest.raises(lazy.EngineUnavailable, match="ghost"):
+ lazy.ensure_import("whatever", purpose="test")
+ assert len(installs) == 4
+ assert all(pkgs == ["ghost"] for pkgs in installs)
+
+
+def test_ensure_import_maps_pil_to_pillow(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+ state = {"round": 0}
+
+ def fake_import(name):
+ if name == "fakeengine":
+ if state["round"] == 0:
+ state["round"] = 1
+ raise ModuleNotFoundError("No module named 'PIL'", name="PIL")
+ return types.ModuleType("fakeengine")
+ raise ModuleNotFoundError(f"No module named {name!r}", name=name)
+
+ monkeypatch.setattr(lazy.importlib, "import_module", fake_import)
+ lazy.ensure_import("fakeengine", purpose="test")
+ assert installs == [["pillow"]]
+
+
+def test_ensure_call_passthrough_without_install(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+ assert lazy.ensure_call(lambda: 42, purpose="test") == 42
+ assert installs == []
+
+
+def test_ensure_call_installs_missing_module(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+ state = {"round": 0}
+
+ def fake_fn():
+ if state["round"] == 0:
+ state["round"] = 1
+ raise ModuleNotFoundError("No module named 'addict'", name="addict")
+ return "ok"
+
+ assert lazy.ensure_call(fake_fn, purpose="test") == "ok"
+ assert installs == [["addict"]]
+
+
+def test_ensure_call_dotted_missing_reraises_without_install(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+
+ def fake_fn():
+ raise ModuleNotFoundError("No module named 'pkg.sub'", name="pkg.sub")
+
+ with pytest.raises(ModuleNotFoundError):
+ lazy.ensure_call(fake_fn, purpose="test")
+ assert installs == []
+
+
+def test_ensure_call_unnamed_missing_reraises_without_install(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+
+ def fake_fn():
+ raise ModuleNotFoundError("boom")
+
+ with pytest.raises(ModuleNotFoundError):
+ lazy.ensure_call(fake_fn, purpose="test")
+ assert installs == []
+
+
+def test_ensure_call_maps_pil_to_pillow(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+ state = {"round": 0}
+
+ def fake_fn():
+ if state["round"] == 0:
+ state["round"] = 1
+ raise ModuleNotFoundError("No module named 'PIL'", name="PIL")
+ return "ok"
+
+ assert lazy.ensure_call(fake_fn, purpose="test") == "ok"
+ assert installs == [["pillow"]]
+
+
+def test_ensure_call_gives_up_after_rounds(monkeypatch):
+ installs: list[list[str]] = []
+ monkeypatch.setattr(lazy, "ensure", lambda pkgs, **_k: installs.append(list(pkgs)))
+
+ def fake_fn():
+ raise ModuleNotFoundError("No module named 'ghost'", name="ghost")
+
+ with pytest.raises(lazy.EngineUnavailable, match="ghost"):
+ lazy.ensure_call(fake_fn, purpose="test")
+ assert len(installs) == 8
+ assert all(pkgs == ["ghost"] for pkgs in installs)
diff --git a/lib/tests/test_meters.py b/lib/tests/test_meters.py
index 73f3db6..b013ec9 100644
--- a/lib/tests/test_meters.py
+++ b/lib/tests/test_meters.py
@@ -17,6 +17,20 @@ def test_true_peak_bounds(sr):
assert sp - 0.01 <= tp <= sp + 0.6
+def test_true_peak_blockwise_matches_whole_file(sr):
+ from scipy import signal
+
+ rng = np.random.default_rng(3)
+ n = sr * 75
+ x = (
+ 0.5 * np.sin(2 * np.pi * 997.0 * np.arange(n) / sr) + 0.02 * rng.standard_normal(n)
+ ).astype(np.float32)
+ half = 10 * 4
+ h = signal.firwin(2 * half + 1, 0.25, window=("kaiser", 8.0)).astype(np.float32)
+ ref = meters.sample_peak_db(signal.resample_poly(x, 4, 1, window=h))
+ assert abs(meters.true_peak_db(x, sr) - ref) < 0.01
+
+
def test_lufs_of_sine(sr):
x = sine(1000, 3.0, sr, peak_dbfs=-17.0)
assert abs(meters.lufs(x, sr) + 20.05) < 0.2
@@ -32,6 +46,13 @@ def test_noise_floor_below_speech(sr):
assert floor < meters.rms_db(x) - 4.0
+def test_speech_level_db_ignores_leading_silence(sr):
+ x = speechish(6.0, sr, level_dbfs=-20.0)
+ y = np.concatenate([np.zeros(sr * 4, dtype=np.float32), x])
+ assert abs(meters.speech_level_db(x, sr) - meters.speech_level_db(y, sr)) < 1.0
+ assert meters.speech_level_db(x, sr) > meters.rms_db(x)
+
+
def test_all_meters_keys(sr):
d = meters.all_meters(speechish(3.0, sr), sr)
assert set(d) == {
diff --git a/lib/tests/test_pipeline.py b/lib/tests/test_pipeline.py
index da2e9da..b3f91f1 100644
--- a/lib/tests/test_pipeline.py
+++ b/lib/tests/test_pipeline.py
@@ -19,6 +19,16 @@ def test_stage_order_and_names():
assert by_name["levelling"].enabled is True
+def test_denoise_stage_detail_shows_pf():
+ opts = Options()
+ stages = pipeline.build_stages(opts)
+ assert "pf=off" in stages[0].detail
+ assert "leveling" not in stages[0].detail
+ opts.denoise_pf = True
+ stages = pipeline.build_stages(opts)
+ assert "pf=on" in stages[0].detail
+
+
def test_full_chain_profile_bounds(sr):
opts = Options()
opts.denoise = "off"
@@ -68,3 +78,37 @@ def test_podcast_profile_bounds(sr):
res = pipeline.run_pipeline(x, sr, opts)
assert abs(meters.lufs(res.audio, sr) + 16.0) < 0.8
assert meters.true_peak_db(res.audio, sr) <= -1.4
+
+
+def test_dsp_pregain_makes_chain_level_invariant(sr):
+ """The voice chain must treat a quiet and a hot take identically.
+
+ Denoised files often arrive far below the chain's design level; without
+ the pre-gain the absolute comp thresholds would idle (or slam) depending
+ on input level alone.
+ """
+ hot = Options()
+ hot.denoise = "off"
+ hot.enhance = "off"
+ hot.levelling = False
+ quiet = Options()
+ quiet.denoise = "off"
+ quiet.enhance = "off"
+ quiet.levelling = False
+ x_hot = speechish(6.0, sr, level_dbfs=-14.0)
+ x_quiet = speechish(6.0, sr, level_dbfs=-38.0)
+ y_hot = pipeline.run_pipeline(x_hot, sr, hot).audio
+ y_quiet = pipeline.run_pipeline(x_quiet, sr, quiet).audio
+ diff = abs(meters.rms_db(y_hot) - meters.rms_db(y_quiet))
+ assert diff < 1.0
+
+
+def test_dsp_pregain_note_recorded(sr):
+ opts = Options()
+ opts.denoise = "off"
+ opts.enhance = "off"
+ notes: list[str] = []
+ stages = pipeline.build_stages(opts, notes)
+ dsp_stage = next(s for s in stages if s.name == "dsp")
+ dsp_stage.fn(speechish(4.0, sr, level_dbfs=-30.0), sr)
+ assert any("pregain" in n for n in notes)
diff --git a/lib/tests/test_ui.py b/lib/tests/test_ui.py
new file mode 100644
index 0000000..156b3e8
--- /dev/null
+++ b/lib/tests/test_ui.py
@@ -0,0 +1,169 @@
+import hashlib
+import subprocess
+import sys
+
+import pytest
+
+from producer import ui
+
+
+class _FakeResp:
+ def __init__(self, payload: bytes, length: str | None = None):
+ self._payload = payload
+ self.headers = {"Content-Length": length} if length else {}
+
+ def read(self, n=-1):
+ if not self._payload:
+ return b""
+ out, self._payload = self._payload[:n], self._payload[n:]
+ return out
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *_a):
+ return False
+
+
+def test_fmt_bytes():
+ assert ui.fmt_bytes(512) == "512 B"
+ assert ui.fmt_bytes(2048) == "2.0 KB"
+ assert ui.fmt_bytes(8 << 20) == "8.0 MB"
+ assert ui.fmt_bytes(None) == "?"
+
+
+def test_fmt_secs():
+ assert ui.fmt_secs(0) == "0:00"
+ assert ui.fmt_secs(59) == "0:59"
+ assert ui.fmt_secs(61) == "1:01"
+ assert ui.fmt_secs(3700) == "1:01:40"
+ assert ui.fmt_secs(None) == "?"
+
+
+def test_bar_fills():
+ assert ui._bar(0.0) == "[" + "-" * 22 + "]"
+ assert ui._bar(1.0) == "[" + "#" * 22 + "]"
+ half = ui._bar(0.5)
+ assert half.count("#") == 11 and half.count("-") == 11
+
+
+def test_download_writes_file(tmp_path, monkeypatch, capsys):
+ payload = b"x" * (2 << 20)
+ sha = hashlib.sha256(payload).hexdigest()
+ monkeypatch.setattr(
+ "urllib.request.urlopen", lambda req, timeout=None: _FakeResp(payload, str(len(payload)))
+ )
+ dest = tmp_path / "big.bin"
+ out = ui.download("https://example.com/big.bin", dest, sha256=sha)
+ assert out == dest
+ assert dest.read_bytes() == payload
+ assert "done" in capsys.readouterr().out
+
+
+def test_download_checksum_mismatch(tmp_path, monkeypatch):
+ monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=None: _FakeResp(b"abc"))
+ dest = tmp_path / "f.bin"
+ with pytest.raises(RuntimeError):
+ ui.download("https://example.com/f.bin", dest, sha256="0" * 64)
+ assert not dest.exists()
+ assert not dest.with_name(dest.name + ".part").exists()
+
+
+def test_download_cached_skips(tmp_path, monkeypatch, capsys):
+ dest = tmp_path / "cached.bin"
+ dest.write_bytes(b"hello")
+
+ def boom(*_a, **_k):
+ raise AssertionError("should not download")
+
+ monkeypatch.setattr("urllib.request.urlopen", boom)
+ ui.download("https://example.com/cached.bin", dest)
+ assert dest.read_bytes() == b"hello"
+ assert "cached" in capsys.readouterr().out
+
+
+def test_download_size_mismatch_redownloads(tmp_path, monkeypatch):
+ dest = tmp_path / "short.bin"
+ dest.write_bytes(b"too short")
+ monkeypatch.setattr("urllib.request.urlopen", lambda req, timeout=None: _FakeResp(b"abcd"))
+ ui.download("https://example.com/short.bin", dest, expected_size=4)
+ assert dest.read_bytes() == b"abcd"
+
+
+def test_download_milestones_non_tty(tmp_path, monkeypatch, capsys):
+ payload = b"y" * (8 << 20)
+ monkeypatch.setattr(
+ "urllib.request.urlopen", lambda req, timeout=None: _FakeResp(payload, str(len(payload)))
+ )
+ ui.download("https://example.com/m.bin", tmp_path / "m.bin")
+ out = capsys.readouterr().out
+ assert "8.0 MB/8.0 MB" in out
+ assert "done" in out
+
+
+def test_run_piped_forwards_output(monkeypatch, capsys):
+ monkeypatch.setattr(ui, "is_tty", lambda: False)
+ rc = ui.run([sys.executable, "-c", "print('hello-uv-output')"], "test run")
+ assert rc == 0
+ assert "hello-uv-output" in capsys.readouterr().out
+
+
+def test_run_piped_raises_on_failure(monkeypatch):
+ monkeypatch.setattr(ui, "is_tty", lambda: False)
+ with pytest.raises(subprocess.CalledProcessError):
+ ui.run([sys.executable, "-c", "raise SystemExit(3)"], "test run")
+
+
+def test_status_non_tty_milestones_and_done(capsys):
+ s = ui.Status(prefix="[producer] in.wav — ")
+ s.stage("denoise")
+ for i in range(1, 11):
+ s.tick(i, 10)
+ s.stage_done("denoise", 2.5)
+ s.stage("dsp")
+ s.stage_done("dsp", 0.3)
+ s.finish()
+ out = capsys.readouterr().out
+ assert "denoise 10/10 chunks (100%)" in out
+ assert "denoise done in 2.5s (10/10 chunks)" in out
+ assert "dsp done in 0.3s" in out
+
+
+def test_status_tty_live_line(monkeypatch, capsys):
+ monkeypatch.setattr(ui, "is_tty", lambda: True)
+ s = ui.Status(prefix="p — ")
+ s.stage("denoise")
+ s.tick(10, 10) # completion redraws immediately (bypasses the 0.1 s throttle)
+ out = capsys.readouterr().out
+ assert "denoise" in out and "10/10" in out
+ s.finish()
+
+
+def test_progress_milestones_non_tty(capsys):
+ p = ui.Progress("downloading x", total=1000)
+ for i in (100, 300, 1000):
+ p.update(i)
+ p.close()
+ out = capsys.readouterr().out
+ assert "1000 B/1000 B" in out
+ assert "done (1000 B" in out
+
+
+def test_progress_tty_bar(monkeypatch, capsys):
+ monkeypatch.setattr(ui, "is_tty", lambda: True)
+ p = ui.Progress("downloading torch", total=100)
+ p.update(50)
+ p.close()
+ out = capsys.readouterr().out
+ assert "50 B/100 B (50%)" in out
+
+
+def test_log_clears_live_line(monkeypatch, capsys):
+ monkeypatch.setattr(ui, "is_tty", lambda: True)
+ p = ui.Progress("downloading", total=100)
+ p.update(50)
+ ui.log("[producer] some message")
+ out = capsys.readouterr().out
+ assert "some message" in out
+ assert out.index("some message") > out.index("downloading")
+ p.close()
diff --git a/lib/tests/test_updates.py b/lib/tests/test_updates.py
new file mode 100644
index 0000000..7ad1727
--- /dev/null
+++ b/lib/tests/test_updates.py
@@ -0,0 +1,221 @@
+import subprocess
+import sys
+from types import SimpleNamespace
+
+from producer import lazy, updates
+
+
+def test_parse_would_install():
+ out = "\n".join(
+ [
+ "Using Python 3.11.16 environment at: /x",
+ "Resolved 2 packages in 1ms",
+ "Would install 2 packages",
+ " + scipy==1.14.1",
+ " - numpy==1.26.4",
+ " + numpy==2.2.6",
+ " ~ cffi==2.1.1",
+ "",
+ ]
+ )
+ assert updates._parse_would_install(out) == {
+ "scipy": "1.14.1",
+ "numpy": "2.2.6",
+ "cffi": "2.1.1",
+ }
+
+
+def test_vkey_ordering():
+ assert updates._vkey("1.26.4") < updates._vkey("2.2.6")
+ assert updates._vkey("0.5.6") < updates._vkey("0.5.7")
+ assert updates._vkey("2.7.1") < updates._vkey("2.7.1+cu126")
+ assert updates._vkey("1.2.10") > updates._vkey("1.2.9")
+ assert not updates._vkey("0.5.6") > updates._vkey("0.5.6")
+
+
+def test_core_names(tmp_path, monkeypatch):
+ reqs = tmp_path / "requirements-core.txt"
+ reqs.write_text("numpy==1.26.4\nscipy>=1.11,<1.15\n\n# comment\nsoundfile>=0.12,<0.13\n")
+ monkeypatch.setattr(updates, "CORE_FILE", reqs)
+ assert updates._core_names() == ["numpy", "scipy", "soundfile"]
+
+
+def test_probe_filters_to_keep(monkeypatch):
+ calls = []
+
+ def fake_run(cmd, capture_output, text, timeout):
+ calls.append(cmd)
+ return SimpleNamespace(
+ returncode=0,
+ stdout="Resolved 2 packages\n + scipy==1.14.1\n + numpy==2.2.6\n",
+ )
+
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(updates.subprocess, "run", fake_run)
+ got = updates._probe(["-U", "-r", "reqs.txt"], None, {"scipy"})
+ assert got == {"scipy": "1.14.1"}
+ assert calls[0][1:4] == ["pip", "install", "--dry-run"]
+
+
+def test_probe_failure_returns_empty(monkeypatch):
+ def boom(cmd, capture_output, text, timeout):
+ raise subprocess.TimeoutExpired(cmd, timeout)
+
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(updates.subprocess, "run", boom)
+ assert updates._probe(["torch"], None, {"torch"}) == {}
+
+
+def test_collect_full_matrix(monkeypatch, tmp_path):
+ installed = {
+ "numpy": "1.26.4",
+ "scipy": "1.13.1",
+ "soundfile": "0.12.1",
+ "pyloudnorm": "0.2.0",
+ "torch": "2.7.1+cu126",
+ "torchaudio": "2.7.1+cu126",
+ "deepfilternet": "0.5.6",
+ "zipenhancer": None,
+ "clearvoice": None,
+ }
+
+ def fake_probe(args, python, keep):
+ if "torch" in args and "torchaudio" in args and "--index-url" in args:
+ return {"torch": "2.9.0+cu126", "torchaudio": "2.9.0+cu126"}
+ if "deepfilternet" in args:
+ return {"deepfilternet": "0.5.7", "torch": "2.14.0"}
+ if "zipenhancer" in args:
+ return {}
+ return {"scipy": "1.14.1"}
+
+ monkeypatch.setattr(updates, "_installed", lambda d: installed.get(d))
+ monkeypatch.setattr(updates, "_probe", fake_probe)
+ monkeypatch.setattr(lazy, "gpu_present", lambda: True)
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(updates, "RESEMBLE_PY", tmp_path / "missing" / "python")
+
+ got = updates.collect()
+ labels = [u.label for u in got]
+ assert labels == ["scipy", "torch + torchaudio", "deepfilternet"]
+
+ scipy = got[0]
+ assert (scipy.old, scipy.new) == ("1.13.1", "1.14.1")
+ assert "-U" in scipy.cmd and str(updates.CORE_FILE) in scipy.cmd
+
+ torch_u = got[1]
+ assert (torch_u.old, torch_u.new) == ("2.7.1+cu126", "2.9.0+cu126")
+ assert "CUDA" in torch_u.note
+ assert "torch==2.9.0+cu126" in torch_u.cmd
+ assert "torchaudio==2.9.0+cu126" in torch_u.cmd
+ assert lazy.TORCH_GPU_INDEX in torch_u.cmd
+
+ dfn = got[2]
+ assert (dfn.old, dfn.new) == ("0.5.6", "0.5.7")
+ assert "torch==2.7.1" in dfn.cmd
+ assert "deepfilternet==0.5.7" in dfn.cmd
+ assert "torch==2.14.0" not in dfn.cmd
+
+
+def test_collect_skips_uninstalled_engines_and_missing_torch(monkeypatch, tmp_path):
+ monkeypatch.setattr(updates, "_installed", lambda d: None)
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(updates, "RESEMBLE_PY", tmp_path / "missing" / "python")
+
+ probed = []
+ monkeypatch.setattr(updates, "_probe", lambda args, python, keep: probed.append(args) or {})
+ assert updates.collect() == []
+ assert len(probed) == 1
+ assert "deepfilternet" not in probed[0] and "torch" not in probed[0]
+
+
+def test_collect_resemble_isolated_venv(monkeypatch, tmp_path):
+ py = tmp_path / "resemble" / "bin" / "python"
+ py.parent.mkdir(parents=True)
+ py.write_text("")
+ monkeypatch.setattr(updates, "_installed", lambda d: "2.7.1+cu126" if d == "torch" else None)
+ monkeypatch.setattr(updates, "_installed_in", lambda p, d: "0.0.1")
+ monkeypatch.setattr(lazy, "find_uv", lambda: "uv")
+ monkeypatch.setattr(updates, "RESEMBLE_PY", py)
+
+ def fake_probe(args, python, keep):
+ if "resemble-enhance" in args:
+ assert python == str(py)
+ return {"resemble-enhance": "0.0.2"}
+ return {}
+
+ monkeypatch.setattr(updates, "_probe", fake_probe)
+ got = updates.collect()
+ assert len(got) == 1
+ u = got[0]
+ assert u.label == "resemble-enhance"
+ assert "resemble-enhance==0.0.2" in u.cmd
+ assert str(py) in u.cmd
+
+
+def test_confirm(monkeypatch):
+ answers = iter(["", "no", "y", "yes"])
+ monkeypatch.setattr("builtins.input", lambda prompt: next(answers))
+ assert not updates._confirm("install updates?")
+ assert not updates._confirm("install updates?")
+ assert updates._confirm("install updates?")
+ assert updates._confirm("install updates?")
+
+
+def test_check_and_prompt_non_tty_notices_only(monkeypatch, capsys):
+ monkeypatch.setattr(
+ updates,
+ "collect",
+ lambda: [updates.Update("scipy", "1.13.1", "1.14.1", "", ["uv"])],
+ )
+ monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: False))
+ assert updates.check_and_prompt() is False
+ out = capsys.readouterr().out
+ assert "updates available" in out
+ assert "./producer update" in out
+
+
+def test_check_and_prompt_assume_yes_applies(monkeypatch, capsys):
+ u = updates.Update("scipy", "1.13.1", "1.14.1", "", ["uv", "pip", "install", "scipy"])
+ monkeypatch.setattr(updates, "collect", lambda: [u])
+ runs = []
+ monkeypatch.setattr(updates.ui, "run", lambda cmd, label, check=True: runs.append(cmd))
+ assert updates.check_and_prompt(assume_yes=True) is True
+ assert runs == [["uv", "pip", "install", "scipy"]]
+ assert "updates installed" in capsys.readouterr().out
+
+
+def test_check_and_prompt_declined(monkeypatch, capsys):
+ u = updates.Update("scipy", "1.13.1", "1.14.1", "", ["uv"])
+ monkeypatch.setattr(updates, "collect", lambda: [u])
+ monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
+ monkeypatch.setattr(updates, "_confirm", lambda prompt: False)
+ runs = []
+ monkeypatch.setattr(updates.ui, "run", lambda cmd, label, check=True: runs.append(cmd))
+ assert updates.check_and_prompt() is False
+ assert runs == []
+ assert "skipped updates" in capsys.readouterr().out
+
+
+def test_check_and_prompt_up_to_date(monkeypatch, capsys):
+ monkeypatch.setattr(updates, "collect", lambda: [])
+ assert updates.check_and_prompt(force=True) is False
+ assert "up to date" in capsys.readouterr().out
+
+
+def test_update_command_requires_yes_when_not_interactive(monkeypatch):
+ monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: False))
+ monkeypatch.setattr(
+ updates,
+ "collect",
+ lambda: [updates.Update("scipy", "1.13.1", "1.14.1", "", ["uv"])],
+ )
+ try:
+ updates.run_update_command([])
+ except SystemExit as e:
+ assert "--yes" in str(e)
+ else:
+ raise AssertionError("expected SystemExit")
+ runs = []
+ monkeypatch.setattr(updates.ui, "run", lambda cmd, label, check=True: runs.append(cmd))
+ assert updates.run_update_command(["--yes"]) == 0
+ assert len(runs) == 1
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 09:29:54 +0000