aboutsummaryrefslogtreecommitdiff
path: root/lib/project/tests/test_pipeline.py
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-09-12 16:48:55 -0400
committerhistoria <historiavg@proton.me>2026-09-12 16:48:55 -0400
commit18346d8deb858cc77d166ffea8ef76d9c9d48bfe (patch)
treee11bb5951a047ed372053906dcbb97b8651cee84 /lib/project/tests/test_pipeline.py
parent464edbc4fd658728f06d3a7d98340aa28ff3e3ec (diff)
downloadproducer-18346d8deb858cc77d166ffea8ef76d9c9d48bfe.tar.gz
feat: automatic stereo mixdown
Diffstat (limited to 'lib/project/tests/test_pipeline.py')
-rw-r--r--lib/project/tests/test_pipeline.py49
1 files changed, 47 insertions, 2 deletions
diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py
index fe04519..0bc2cef 100644
--- a/lib/project/tests/test_pipeline.py
+++ b/lib/project/tests/test_pipeline.py
@@ -64,13 +64,13 @@ def test_existing_report_log_is_replaced_without_overwrite(wav, tmp_path, bypass
@pytest.mark.parametrize("kind,message", [
("empty", "empty"), ("silence", "silent or too quiet"),
("quiet", "silent or too quiet"), ("surround", "Only mono or stereo"),
- ("nonfinite", "non-finite"), ("stereo", "downmix to mono"),
+ ("nonfinite", "non-finite"),
])
def test_unusable_inputs_leave_no_outputs(wav, tmp_path, bypass, progress, ffmpeg, kind, message):
samples = {
"empty": np.empty(0), "silence": np.zeros(48000),
"quiet": np.full(48000, 1e-6), "surround": np.ones((4800, 3)) * 0.1,
- "nonfinite": np.array([0.1, np.nan]), "stereo": np.ones((4800, 2)) * 0.1,
+ "nonfinite": np.array([0.1, np.nan]),
}[kind]
source = wav(samples)
target = tmp_path / "outputs" / "master.wav"
@@ -79,6 +79,51 @@ def test_unusable_inputs_leave_no_outputs(wav, tmp_path, bypass, progress, ffmpe
assert not target.parent.exists()
+def test_downmix_to_mono_strategies(wav, tmp_path, progress):
+ from producer.audio import downmix_to_mono
+ time = np.arange(48000 * 3) / 48000
+ voice = 0.1 * np.sin(2 * np.pi * 440 * time)
+ # Identical channels: a plain average is lossless and keeps full level.
+ mono = tmp_path / "same.wav"
+ method = downmix_to_mono(wav(np.column_stack([voice, voice])), mono, progress)
+ assert method == "the channels were averaged"
+ assert sf.info(mono).channels == 1
+ assert np.abs(sf.read(mono)[0] - voice).max() == pytest.approx(0, abs=1e-7)
+ # Inverted channels: the phase-cancel case the old error guarded against.
+ mono = tmp_path / "inverted.wav"
+ method = downmix_to_mono(wav(np.column_stack([voice, -voice])), mono, progress)
+ assert "inverted" in method
+ assert np.abs(sf.read(mono)[0] - voice).max() == pytest.approx(0, abs=1e-7)
+ # One silent channel: keep the live channel at unity gain, not -6 dB.
+ mono = tmp_path / "silent.wav"
+ method = downmix_to_mono(wav(np.column_stack([voice, np.zeros_like(voice)])), mono, progress)
+ assert "right channel is silent" in method
+ assert np.abs(sf.read(mono)[0] - voice).max() == pytest.approx(0, abs=1e-7)
+ method = downmix_to_mono(wav(np.column_stack([np.zeros_like(voice), voice])),
+ tmp_path / "silent-left.wav", progress)
+ assert "left channel is silent" in method
+ with pytest.raises(ValueError, match="non-finite"):
+ downmix_to_mono(wav(np.array([[0.1, np.nan]])), tmp_path / "bad.wav", progress)
+
+
+def test_stereo_input_is_downmixed_not_rejected(wav, tmp_path, bypass, progress, ffmpeg):
+ time = np.arange(48000 * 3) / 48000
+ voice = 0.1 * np.sin(2 * np.pi * 440 * time)
+ source = wav(np.column_stack([voice, -voice]))
+ original = source.read_bytes()
+ target = tmp_path / "master.wav"
+ report = pipeline.process(source, target, bypass, progress)
+ assert any("downmixed to mono" in warning and "inverted" in warning
+ for warning in report["warnings"])
+ stats = sf.info(target)
+ assert stats.channels == 1 and stats.frames == 144000
+ # A naive (L + R) / 2 mix of these channels is digital silence; the master
+ # must carry real signal instead.
+ assert report["master"]["sample_peak_dbfs"] > -30
+ assert source.read_bytes() == original
+ assert not list(target.parent.glob(".producer-*"))
+
+
def test_measure_reports_channels_peaks_clipping_and_progress(wav):
samples = np.tile([1.0, -0.25], (4800, 1))
events = []