From c2b0f7e4fb4738afcae1705db8f983dd90a669a4 Mon Sep 17 00:00:00 2001 From: historia Date: Sun, 6 Sep 2026 15:17:57 -0400 Subject: inital commit --- lib/tests/conftest.py | 70 +++++++++++++++++++++ lib/tests/test_cli.py | 152 +++++++++++++++++++++++++++++++++++++++++++++ lib/tests/test_dsp.py | 130 ++++++++++++++++++++++++++++++++++++++ lib/tests/test_engines.py | 59 ++++++++++++++++++ lib/tests/test_loudness.py | 30 +++++++++ lib/tests/test_meters.py | 44 +++++++++++++ lib/tests/test_pipeline.py | 70 +++++++++++++++++++++ 7 files changed, 555 insertions(+) create mode 100644 lib/tests/conftest.py create mode 100644 lib/tests/test_cli.py create mode 100644 lib/tests/test_dsp.py create mode 100644 lib/tests/test_engines.py create mode 100644 lib/tests/test_loudness.py create mode 100644 lib/tests/test_meters.py create mode 100644 lib/tests/test_pipeline.py (limited to 'lib/tests') diff --git a/lib/tests/conftest.py b/lib/tests/conftest.py new file mode 100644 index 0000000..885d22b --- /dev/null +++ b/lib/tests/conftest.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np +import pytest + +SRC = Path(__file__).resolve().parents[1] / "src" +sys.path.insert(0, str(SRC)) +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +SR = 44100 + + +def speechish( + dur: float, + sr: int = SR, + level_dbfs: float = -20.0, + seed: int = 0, +) -> np.ndarray: + rng = np.random.default_rng(seed) + n = int(sr * dur) + t = np.arange(n) / sr + f0 = 110.0 * (1.0 + 0.02 * np.sin(2 * np.pi * 0.9 * t)) + phase = 2 * np.pi * np.cumsum(f0) / sr + x = np.zeros(n) + for k in range(1, 9): + x += (1.0 / k**1.3) * np.sin(k * phase + 0.3 * k) + syll = 0.5 + 0.5 * np.sin(2 * np.pi * 3.0 * t + float(rng.uniform(0, 6))) + pauses = (np.sin(2 * np.pi * 0.5 * t) > -0.6).astype(float) + env = np.clip(syll, 0.02, 1.0) ** 0.6 * np.maximum(pauses, 0.05) + x = x * env + x /= np.max(np.abs(x)) + 1e-12 + return (x * (10 ** (level_dbfs / 20.0))).astype(np.float32) + + +def sine(freq: float, dur: float, sr: int = SR, peak_dbfs: float = -20.0) -> np.ndarray: + t = np.arange(int(sr * dur)) / sr + return (10 ** (peak_dbfs / 20.0) * np.sin(2 * np.pi * freq * t)).astype(np.float32) + + +def band_db(x: np.ndarray, sr: int, lo: float, hi: float) -> float: + from scipy import signal + + sos = signal.butter(4, [lo, hi], btype="bandpass", fs=sr, output="sos") + y = signal.sosfilt(sos, x.astype(np.float64)) + r = np.sqrt(np.mean(np.square(y))) + if r <= 0: + return -120.0 + return float(20 * np.log10(r)) + + +@pytest.fixture +def sr() -> int: + return SR + + +@pytest.fixture +def speech() -> np.ndarray: + return speechish(6.0, level_dbfs=-20.0) + + +@pytest.fixture +def noisy_speech(sr, speech) -> np.ndarray: + rng = np.random.default_rng(7) + noise = rng.standard_normal(speech.size) + noise *= (10 ** (-48.0 / 20.0)) / np.sqrt(np.mean(np.square(noise))) + hum = 0.003 * np.sin(2 * np.pi * 50.0 * np.arange(speech.size) / sr) + return (speech + noise + hum).astype(np.float32) diff --git a/lib/tests/test_cli.py b/lib/tests/test_cli.py new file mode 100644 index 0000000..e43489c --- /dev/null +++ b/lib/tests/test_cli.py @@ -0,0 +1,152 @@ +import numpy as np +import soundfile as sf +from conftest import speechish + +from producer import io as pio +from producer.cli import _apply_args, build_parser, process_one +from producer.config import Options + + +def _mk_wav(tmp_path, name="in.wav", stereo=False): + x = speechish(3.0, level_dbfs=-30.0) + if stereo: + data = np.stack([x, x * 0.5], axis=1) + sf.write(str(tmp_path / name), data, 44100, subtype="PCM_16") + else: + sf.write(str(tmp_path / name), x, 44100, subtype="PCM_16") + return tmp_path / name + + +def _opts(**kw): + opts = Options() + opts.denoise = "off" + opts.enhance = "off" + for k, v in kw.items(): + setattr(opts, k, v) + return opts + + +def test_decode_stereo_mixdown(tmp_path): + p = _mk_wav(tmp_path, "st.wav", stereo=True) + x, sr = pio.decode(p) + assert sr == 44100 + assert x.dtype.name == "float32" + assert x.ndim == 1 + + +def test_encode_wav_bitdepths(tmp_path): + x = speechish(2.0, level_dbfs=-20.0) + for depth in (16, 24, 32): + out = tmp_path / f"o{depth}.wav" + pio.encode(x, 44100, out, "wav", depth) + y, sr = pio.decode(out) + assert sr == 44100 + assert ( + abs( + float(np.sqrt(np.mean(y.astype(np.float64) ** 2))) + - float(np.sqrt(np.mean(x.astype(np.float64) ** 2))) + ) + < 1e-3 + ) + + +def test_encode_flac(tmp_path): + x = speechish(2.0, level_dbfs=-20.0) + out = tmp_path / "o.flac" + pio.encode(x, 44100, out, "flac", 24) + y, sr = pio.decode(out) + assert sr == 44100 + assert np.corrcoef(x, y)[0, 1] > 0.999 + + +def test_mp3_roundtrip(tmp_path): + import pytest as _pt + + if not pio.ffmpeg_available(): + _pt.skip("ffmpeg missing") + x = speechish(4.0, level_dbfs=-20.0) + out = tmp_path / "o.mp3" + pio.encode(x, 44100, out, "mp3", 16) + y, sr = pio.decode(out) + assert sr == 44100 + from producer import meters + + assert abs(meters.rms_db(x) - meters.rms_db(y)) < 0.7 + + +def test_decode_via_ffmpeg_fallback(tmp_path): + import pytest as _pt + + if not pio.ffmpeg_available(): + _pt.skip("ffmpeg missing") + x = speechish(4.0, level_dbfs=-20.0) + out = tmp_path / "o.mp3" + pio.encode(x, 44100, out, "mp3", 16) + y, sr = pio.decode(out) + assert sr == 44100 + assert y.size > 0 + + +def test_process_one_end_to_end(tmp_path, capsys): + import json + + from producer import meters + + inp = _mk_wav(tmp_path, "e2e.wav") + out = tmp_path / "e2e_master.wav" + rc = process_one(inp, _opts(report=True), single=True) + assert rc == out + assert out.exists() + rep_path = out.with_name("e2e_master.report.json") + assert rep_path.exists() + rep = json.loads(rep_path.read_text()) + y, sr = pio.decode(out) + assert abs(meters.rms_db(y) + 20.0) < 0.6 + assert meters.true_peak_db(y, sr) <= -2.9 + assert rep["after"]["rms_db"] != 0 + + +def test_dry_run_listing(tmp_path, capsys): + from producer.cli import main + + inp = _mk_wav(tmp_path, "dry.wav") + rc = main([str(inp), "--dry-run", "--denoise", "off"]) + assert rc == 0 + out = capsys.readouterr().out + assert "denoise" in out and "dsp" in out and "levelling" in out + + +def test_arg_parsing_precedence(): + parser = build_parser() + args = parser.parse_args( + ["in.wav", "--profile", "podcast", "--warmth", "0.1", "--ceiling", "-2.0"] + ) + opts = Options() + _apply_args(opts, args) + assert opts.profile == "podcast" + assert opts.loudness_mode() == "lufs" + assert opts.strengths["warmth"] == 0.1 + assert opts.ceiling() == -2.0 + assert opts.strengths["air"] is None + + +def test_radio_profile_has_tuned_defaults(): + from producer import pipeline + + opts = Options(profile="radio") + assert opts.eff("tape") > 0 and opts.eff("soothe") > 0 + assert opts.loudness_mode() == "lufs" + stages = pipeline.build_stages(opts) + assert [st.name for st in stages] == ["denoise", "enhance", "dsp", "levelling"] + assert opts.denoise_strength is not None + + +def test_tape_and_soothe_flags_override(): + args = build_parser().parse_args( + ["in.wav", "--profile", "radio", "--tape", "0.4", "--soothe", "0.7"] + ) + opts = Options() + _apply_args(opts, args) + assert opts.profile == "radio" + assert opts.strengths["tape"] == 0.4 + assert opts.strengths["soothe"] == 0.7 diff --git a/lib/tests/test_dsp.py b/lib/tests/test_dsp.py new file mode 100644 index 0000000..b9508ab --- /dev/null +++ b/lib/tests/test_dsp.py @@ -0,0 +1,130 @@ +import numpy as np +from conftest import band_db, sine, speechish + +from producer import dsp, meters + + +def test_hpf_removes_rumble(sr): + x = sine(40, 3.0, sr, -20.0) + sine(200, 3.0, sr, -20.0) + y = dsp.hpf(x, sr, 80.0) + assert band_db(x, sr, 35, 45) - band_db(y, sr, 35, 45) > 10.0 + assert abs(band_db(x, sr, 190, 210) - band_db(y, sr, 190, 210)) < 0.5 + + +def test_peak_eq_mud_cut(sr): + x = sine(300, 3.0, sr, -20.0) + sine(1000, 3.0, sr, -20.0) + y = dsp.peak_eq(x, sr, 300.0, -3.0, 1.0) + d300 = band_db(x, sr, 280, 320) - band_db(y, sr, 280, 320) + d1k = band_db(x, sr, 950, 1050) - band_db(y, sr, 950, 1050) + assert 2.4 < d300 < 3.6 + assert abs(d1k) < 0.3 + + +def test_shelf(sr): + x = sine(60, 3.0, sr, -20.0) + sine(3000, 3.0, sr, -20.0) + y = dsp.shelf(x, sr, 150.0, 1.5, low=True) + d60 = band_db(y, sr, 50, 70) - band_db(x, sr, 50, 70) + assert 1.1 < d60 < 1.9 + + +def test_compressor_steady_sine(sr): + x = sine(440, 4.0, sr, peak_dbfs=-10.0) + y = dsp.compressor(x, sr, -20.0, 3.0, 15.0, 150.0, 6.0) + in_rms = meters.rms_db(x) + out_rms = meters.rms_db(y) + expected_gr = (1.0 - 1.0 / 3.0) * (in_rms - (-20.0)) + assert abs((in_rms - out_rms) - expected_gr) < 0.4 + + +def test_compressor_quiet_signal_unaffected(sr): + x = sine(440, 4.0, sr, peak_dbfs=-45.0) + y = dsp.compressor(x, sr, -20.0, 3.0, 15.0, 150.0, 6.0) + assert abs(meters.rms_db(x) - meters.rms_db(y)) < 0.1 + + +def test_deesser(sr): + x = sine(1000, 4.0, sr, -20.0) + sine(6500, 4.0, sr, -30.0) + y = dsp.deesser(x, sr, 5500.0, 8000.0, 8.0) + d_sib = band_db(x, sr, 5800, 7200) - band_db(y, sr, 5800, 7200) + d_mid = band_db(x, sr, 900, 1100) - band_db(y, sr, 900, 1100) + assert 1.0 < d_sib < 3.0 + assert abs(d_mid) < 0.4 + + +def test_deesser_bypass(sr): + x = sine(1000, 2.0, sr, -20.0) + y = dsp.deesser(x, sr, 5500.0, 8000.0, 0.0) + assert np.allclose(x, y) + + +def test_expander_attenuates_pauses(sr): + speech = speechish(10.0, sr, level_dbfs=-20.0) + rng = np.random.default_rng(3) + noise = rng.standard_normal(speech.size).astype(np.float64) + noise *= (10 ** (-52.0 / 20.0)) / np.sqrt(np.mean(np.square(noise))) + x = (speech + noise).astype(np.float32) + y = dsp.expander(x, sr, max_drop_db=2.1) + frame = int(0.02 * sr) + nf = x.size // frame + frms_x = np.sqrt(np.mean(np.square(x[: nf * frame].reshape(nf, frame)), axis=1)) + frms_y = np.sqrt(np.mean(np.square(y[: nf * frame].reshape(nf, frame)), axis=1)) + fdb_x = 20 * np.log10(frms_x + 1e-12) + loud = fdb_x > np.percentile(fdb_x, 75) + quiet = fdb_x < np.percentile(fdb_x, 15) + stable_loud = np.convolve(loud.astype(int), np.ones(5, dtype=int), mode="same") == 5 + d_y = 20 * np.log10(frms_y + 1e-12) + assert np.mean(fdb_x[stable_loud] - d_y[stable_loud]) < 0.6 + assert np.mean(fdb_x[quiet] - d_y[quiet]) > 1.5 + + +def test_limit_hits_ceiling(sr): + x = speechish(6.0, sr, level_dbfs=-3.0) + y = dsp.limit(x, sr, -3.0) + assert meters.true_peak_db(y, sr) <= -3.0 + 0.1 + assert meters.rms_db(x) - meters.rms_db(y) < 1.0 + + +def test_limit_below_ceiling_transparent(sr): + x = speechish(6.0, sr, level_dbfs=-20.0) + y = dsp.limit(x, sr, -3.0) + assert np.max(np.abs(x - y)) < 1e-6 + + +def test_resample_roundtrip(sr): + x = speechish(4.0, sr) + y = dsp.resample(dsp.resample(x, sr, 48000), 48000, sr) + assert abs(y.size - x.size) <= 2 + assert abs(meters.rms_db(x) - meters.rms_db(y)) < 0.2 + + +def test_tape_bypass(sr): + x = speechish(3.0, sr) + assert np.array_equal(x, dsp.tape(x, sr, 0.0)) + + +def test_tape_adds_even_harmonics(sr): + x = sine(220, 3.0, sr, -6.0) + y = dsp.tape(x, sr, 1.0) + assert np.all(np.isfinite(y)) + d_h2 = band_db(y, sr, 430, 460) - band_db(x, sr, 430, 460) + d_h3 = band_db(y, sr, 655, 690) - band_db(x, sr, 655, 690) + assert d_h2 > 1.0 + assert d_h3 > 1.0 + assert abs(meters.rms_db(x) - meters.rms_db(y)) < 2.0 + + +def test_soothe_bypass(sr): + x = speechish(3.0, sr) + assert np.array_equal(x, dsp.soothe(x, sr, 0.0)) + + +def test_soothe_reduces_resonant_bands(sr): + x = sine(300, 4.0, sr, -6.0) + sine(3500, 4.0, sr, -6.0) + sine(1000, 4.0, sr, -30.0) + y = dsp.soothe(x, sr, 1.0) + d_low = band_db(x, sr, 260, 350) - band_db(y, sr, 260, 350) + d_harsh = band_db(x, sr, 3200, 3800) - band_db(y, sr, 3200, 3800) + assert 2.0 < d_low <= 3.6 + assert 2.0 < d_harsh <= 5.2 + # frequencies outside the bands stay untouched + d_mid = band_db(x, sr, 900, 1200) - band_db(y, sr, 900, 1200) + assert abs(d_mid) < 0.4 diff --git a/lib/tests/test_engines.py b/lib/tests/test_engines.py new file mode 100644 index 0000000..0efccc6 --- /dev/null +++ b/lib/tests/test_engines.py @@ -0,0 +1,59 @@ +import os + +import numpy as np +import pytest + + +def _metrics_floor(x, sr): + from producer import meters + + return meters.noise_floor_db(x, sr) + + +@pytest.mark.slow +def test_dfn3_reduces_noise(sr, noisy_speech): + pytest.importorskip("torch") + pytest.importorskip("df") + from producer.engines import denoise_dfn + + x = noisy_speech[: sr * 4] + y, eng, _dev = denoise_dfn.denoise(x, sr, 1.0, "cpu") + assert "dfn" in eng + assert _metrics_floor(y, sr) < _metrics_floor(x, sr) - 5.0 + assert np.corrcoef(x, y.astype(np.float64))[0, 1] > 0.9 + + +@pytest.mark.slow +def test_zipenhancer_reduces_noise(sr, noisy_speech): + pytest.importorskip("torch") + pytest.importorskip("zipenhancer") + from producer.engines import denoise_zip + + x = noisy_speech[: sr * 4] + y, eng, _ = denoise_zip.denoise(x, sr, 1.0, "cpu") + assert "zipenhancer" in eng + assert _metrics_floor(y, sr) < _metrics_floor(x, sr) - 5.0 + + +@pytest.mark.slow +def test_mossformer2_enhances(sr, noisy_speech): + pytest.importorskip("torch") + pytest.importorskip("clearvoice") + from producer.engines import enhance_mossformer + + x = noisy_speech[: sr * 4] + y, eng, _ = enhance_mossformer.enhance(x, sr, 1.0, "cpu") + assert "mossformer2" in eng + assert np.all(np.isfinite(y)) + + +@pytest.mark.slow +def test_resemble_enhance(sr, noisy_speech): + if not os.environ.get("PRODUCER_TEST_RESEMBLE"): + pytest.skip("set PRODUCER_TEST_RESEMBLE=1 to run the isolated-venv generative engine") + from producer.engines import enhance_resemble + + x = noisy_speech[: sr * 4] + y, eng, _ = enhance_resemble.enhance(x, sr, 1.0, "cpu") + assert "resemble" in eng + assert np.all(np.isfinite(y)) diff --git a/lib/tests/test_loudness.py b/lib/tests/test_loudness.py new file mode 100644 index 0000000..5cd84d6 --- /dev/null +++ b/lib/tests/test_loudness.py @@ -0,0 +1,30 @@ +import numpy as np +from conftest import speechish + +from producer import loudness, meters + + +def test_rms_normalize_hits_target(sr): + x = speechish(8.0, sr, level_dbfs=-40.0) + y = loudness.normalize(x, sr, "rms", -20.0, -3.0) + assert abs(meters.rms_db(y) + 20.0) < 0.5 + assert meters.true_peak_db(y, sr) <= -2.9 + + +def test_lufs_normalize_hits_target(sr): + x = speechish(8.0, sr, level_dbfs=-35.0) + y = loudness.normalize(x, sr, "lufs", -16.0, -1.5) + assert abs(meters.lufs(y, sr) + 16.0) < 0.6 + assert meters.true_peak_db(y, sr) <= -1.4 + + +def test_silence_passthrough(): + y = loudness.normalize(np.zeros(44100 * 2, dtype=np.float32), 44100, "rms", -20.0, -3.0) + assert np.allclose(y, 0.0) + + +def test_idempotent(sr): + x = speechish(8.0, sr, level_dbfs=-35.0) + y1 = loudness.normalize(x, sr, "rms", -20.0, -3.0) + y2 = loudness.normalize(y1, sr, "rms", -20.0, -3.0) + assert abs(meters.rms_db(y1) - meters.rms_db(y2)) < 0.6 diff --git a/lib/tests/test_meters.py b/lib/tests/test_meters.py new file mode 100644 index 0000000..73f3db6 --- /dev/null +++ b/lib/tests/test_meters.py @@ -0,0 +1,44 @@ +import numpy as np +from conftest import sine, speechish + +from producer import meters + + +def test_rms_and_peak_of_sine(sr): + x = sine(1000, 3.0, sr, peak_dbfs=-17.0) + assert abs(meters.rms_db(x) + 20.0) < 0.1 + assert abs(meters.sample_peak_db(x) + 17.0) < 0.05 + + +def test_true_peak_bounds(sr): + x = sine(1000, 3.0, sr, peak_dbfs=-17.0) + tp = meters.true_peak_db(x, sr) + sp = meters.sample_peak_db(x) + assert sp - 0.01 <= tp <= sp + 0.6 + + +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 + + +def test_lufs_silence(): + assert meters.lufs(np.zeros(48000, dtype=np.float32), 48000) == meters.SILENCE_DB + + +def test_noise_floor_below_speech(sr): + x = speechish(8.0, sr, level_dbfs=-20.0) + floor = meters.noise_floor_db(x, sr) + assert floor < meters.rms_db(x) - 4.0 + + +def test_all_meters_keys(sr): + d = meters.all_meters(speechish(3.0, sr), sr) + assert set(d) == { + "rms_db", + "sample_peak_db", + "true_peak_db", + "lufs", + "noise_floor_db", + "duration_s", + } diff --git a/lib/tests/test_pipeline.py b/lib/tests/test_pipeline.py new file mode 100644 index 0000000..da2e9da --- /dev/null +++ b/lib/tests/test_pipeline.py @@ -0,0 +1,70 @@ +import numpy as np +from conftest import band_db, sine, speechish + +from producer import meters, pipeline +from producer.config import Options + + +def test_stage_order_and_names(): + opts = Options() + opts.denoise = "off" + opts.enhance = "off" + res = pipeline.run_pipeline(np.zeros(44100, dtype=np.float32), 44100, opts) + names = [s.name for s in res.stages] + assert names == ["denoise", "enhance", "dsp", "levelling"] + by_name = {s.name: s for s in res.stages} + assert by_name["denoise"].enabled is False + assert by_name["enhance"].enabled is False + assert by_name["dsp"].enabled is True + assert by_name["levelling"].enabled is True + + +def test_full_chain_profile_bounds(sr): + opts = Options() + opts.denoise = "off" + opts.enhance = "off" + x = speechish(8.0, sr, level_dbfs=-35.0) + res = pipeline.run_pipeline(x, sr, opts) + assert abs(meters.rms_db(res.audio) + 20.0) < 0.6 + assert meters.true_peak_db(res.audio, sr) <= -2.9 + assert res.timings.get("levelling", 0) >= 0 + + +def test_passthrough_when_disabled(sr): + opts = Options() + opts.denoise = "off" + opts.enhance = "off" + opts.dsp = False + opts.levelling = False + x = speechish(4.0, sr, level_dbfs=-20.0) + res = pipeline.run_pipeline(x, sr, opts) + assert np.allclose(res.audio, x) + + +def test_knob_zero_disables_eq(sr): + base = Options() + base.denoise = "off" + base.enhance = "off" + base.levelling = False + base.strengths["warmth"] = 0.0 + warm = Options() + warm.denoise = "off" + warm.enhance = "off" + warm.levelling = False + x = sine(60, 4.0, sr, -20.0) + sine(3000, 4.0, sr, -20.0) + y_flat = pipeline.run_pipeline(x, sr, base).audio + y_warm = pipeline.run_pipeline(x, sr, warm).audio + d_flat = band_db(y_flat, sr, 50, 70) - band_db(x, sr, 50, 70) + d_warm = band_db(y_warm, sr, 50, 70) - band_db(x, sr, 50, 70) + assert d_warm - d_flat > 0.8 + + +def test_podcast_profile_bounds(sr): + opts = Options() + opts.profile = "podcast" + opts.denoise = "off" + opts.enhance = "off" + x = speechish(8.0, sr, level_dbfs=-35.0) + 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 -- cgit v1.2.3