path: root/lib/project/tests/test_pipeline.py

import errno
import json
import os
from pathlib import Path

import numpy as np
import pytest
import soundfile as sf

from producer import pipeline
from producer.audio import loudness, measure
from producer.config import resolve


@pytest.mark.parametrize("kind", ["missing", "input-extension", "output-extension"])
def test_invalid_paths_fail_before_tool_setup(wav, tmp_path, bypass, progress, monkeypatch, kind):
    source = wav(name="input.txt" if kind == "input-extension" else "input.wav")
    if kind == "missing":
        source = tmp_path / "missing.wav"
    target = tmp_path / ("master.mp3" if kind == "output-extension" else "master.wav")
    monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Invalid paths must fail preflight"))
    with pytest.raises(ValueError, match="WAV file|.wav extension"):
        pipeline.process(source, target, bypass, progress)
    assert not target.exists()


@pytest.mark.parametrize("alias", ["same", "symlink", "hardlink"])
@pytest.mark.parametrize("overwrite", [False, True])
def test_original_is_never_overwritten(wav, tmp_path, bypass, progress, monkeypatch, alias, overwrite):
    source = wav()
    original = source.read_bytes()
    target = source if alias == "same" else tmp_path / "alias.wav"
    if alias == "symlink":
        target.symlink_to(source)
    elif alias == "hardlink":
        target.hardlink_to(source)
    monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject the input"))
    with pytest.raises(ValueError, match="original recording"):
        pipeline.process(source, target, bypass, progress, overwrite=overwrite)
    assert source.read_bytes() == original


@pytest.mark.parametrize("suffix", [".wav", ".mp3"])
def test_existing_output_or_sidecar_rejected_before_processing(wav, tmp_path, bypass, progress, monkeypatch, suffix):
    source = wav()
    target = tmp_path / "master.wav"
    existing = target.with_suffix(suffix)
    existing.parent.mkdir(parents=True, exist_ok=True)
    existing.write_bytes(b"existing output")
    bypass.mp3 = True
    monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Preflight must reject existing outputs"))
    with pytest.raises(FileExistsError, match="Output exists"):
        pipeline.process(source, target, bypass, progress)
    assert existing.read_bytes() == b"existing output"


def test_existing_report_log_is_kept_and_new_report_published(wav, tmp_path, bypass, progress, ffmpeg, logs):
    source = wav()
    target = tmp_path / "master.wav"
    logs.mkdir(parents=True, exist_ok=True)
    stale = logs / "master.20240101-000000.report.json"
    stale.write_bytes(b"stale report")
    report = pipeline.process(source, target, bypass, progress)
    # Timestamped reports never replace earlier ones.
    assert stale.read_bytes() == b"stale report"
    published = Path(report["report"])
    assert published.parent == logs and published != stale
    assert json.loads(published.read_text()) == report


@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"),
])
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]),
    }[kind]
    source = wav(samples)
    target = tmp_path / "outputs" / "master.wav"
    with pytest.raises(ValueError, match=message):
        pipeline.process(source, target, bypass, progress)
    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 = []
    result = measure(wav(samples), lambda *event: events.append(event))
    assert result["duration_seconds"] == 0.1
    assert result["channels"] == 2
    assert result["sample_peak_dbfs"] == 0
    assert result["channel_rms_dbfs"] == pytest.approx([0, 20 * np.log10(0.25)])
    assert result["dc_offset"] == pytest.approx([1, -0.25])
    assert result["near_full_scale_samples"] == 4800
    assert events[0][1:] == (0, 4800)
    assert events[-1][1:] == (4800, 4800)


@pytest.mark.parametrize("rate", [44100, 48000])
def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg, rate, report_of):
    source = wav()
    original = source.read_bytes()
    target = tmp_path / "nested" / "master.wav"
    settings = resolve(overrides={"denoiser": "none", "sample_rate": rate})
    report = pipeline.process(source, target, settings, progress)
    stats = sf.info(target)
    assert stats.channels == 1
    assert stats.samplerate == rate
    assert stats.subtype == "PCM_32"
    assert abs(stats.duration - 3) <= 0.02
    verified = loudness(ffmpeg, target, settings, stats.duration, progress)
    assert verified["input_i"] == pytest.approx(settings.target_lufs, abs=0.5)
    assert verified["input_tp"] <= settings.true_peak_db + 0.1
    assert report["master"]["near_full_scale_samples"] == 0
    assert report["master"]["sample_peak_dbfs"] <= settings.true_peak_db + 0.1
    assert json.loads(report_of(target).read_text()) == report
    assert source.read_bytes() == original
    assert not list(target.parent.glob(".producer-*"))


def test_limiter_without_normalization(wav, tmp_path, bypass, progress, ffmpeg):
    time = np.arange(48000 * 3) / 48000
    source = wav(0.99 * np.sin(2 * np.pi * 440 * time))
    bypass.limiter = True
    bypass.true_peak_db = -6
    report = pipeline.process(source, tmp_path / "master.wav", bypass, progress)
    assert report["normalization_mode"] == "disabled"
    assert report["master_loudness"]["input_tp"] <= -5.9
    assert report["master"]["sample_peak_dbfs"] > -7


def test_speech_estimate_survives_pause_dominated_recordings(wav):
    time = np.arange(48000 * 3) / 48000
    audio = np.concatenate([np.zeros(48000 * 27), 0.02 * np.sin(2 * np.pi * 440 * time)])
    stats = measure(wav(audio), lambda *_: None)
    assert stats["quiet_blocks_dbfs"] <= -110
    assert stats["speech_level_estimate_dbfs"] == pytest.approx(-37, abs=2)


def test_leveling_uses_active_speech_not_pauses(wav, tmp_path, progress, ffmpeg):
    time = np.arange(48000 * 3) / 48000
    audio = np.concatenate([np.zeros(48000 * 27), 0.02 * np.sin(2 * np.pi * 440 * time)])
    source = wav(audio)
    settings = resolve(profile="cleanup-only", overrides={"denoiser": "none", "leveling": True})
    report = pipeline.process(source, tmp_path / "master.wav", settings, progress)
    assert 10 < report["level_correction_db"] < settings.max_gain_db
    assert not any("gain limit" in warning for warning in report["warnings"])


def test_leveling_skipped_when_speech_estimate_is_unreliable(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
    real_measure = pipeline.measure

    def measured(path, callback, stage="Analyzing audio"):
        stats = real_measure(path, callback, stage)
        if stage == "Measuring cleaned voice":
            stats["speech_level_estimate_dbfs"] = -120
        return stats

    monkeypatch.setattr(pipeline, "measure", measured)
    bypass.leveling = True
    report = pipeline.process(wav(), tmp_path / "master.wav", bypass, progress)
    assert report["level_correction_db"] == 0
    assert any("leveling skipped" in warning for warning in report["warnings"])


def test_zero_denoise_strength_skips_ai(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
    def unexpected(*args, **kwargs):
        pytest.fail("Zero denoise strength must not invoke AI")

    monkeypatch.setattr("producer.ai.denoise", unexpected)
    bypass.denoiser = "deepfilter"
    bypass.denoise_strength = 0
    report = pipeline.process(wav(), tmp_path / "master.wav", bypass, progress)
    assert report["cleaned"]["frames"] == report["input"]["frames"]


def test_fft_and_mp3_export_can_replace_generated_outputs(wav, tmp_path, bypass, progress, ffmpeg, logs):
    source = wav()
    original = source.read_bytes()
    bypass.denoiser = "fft"
    bypass.mp3 = True
    target = tmp_path / "master.wav"
    logs.mkdir(parents=True, exist_ok=True)
    stale_report = logs / "master.20240101-000000.report.json"
    for path in (target, target.with_suffix(".mp3"), stale_report):
        path.write_bytes(b"old output")
    events = []
    report = pipeline.process(source, target, bypass, lambda *event: events.append(event), overwrite=True)
    assert any(event[0] == "FFT noise reduction" for event in events)
    assert target.with_suffix(".mp3").stat().st_size > 1000
    assert report["mp3_loudness"]["input_i"] is not None
    # The old report stays; the new one is timestamped and returned.
    assert stale_report.read_bytes() == b"old output"
    assert json.loads(Path(report["report"]).read_text()) == report
    assert Path(report["report"]).parent == logs
    assert sf.info(target).channels == 1
    assert source.read_bytes() == original


def test_render_failure_preserves_existing_outputs_and_removes_temporary_files(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch, logs):
    source = wav()
    target = tmp_path / "master.wav"
    bypass.mp3 = True
    outputs = [target, target.with_suffix(".mp3")]
    logs.mkdir(parents=True, exist_ok=True)
    stale_report = logs / "master.20240101-000000.report.json"
    for path in outputs + [stale_report]:
        path.write_bytes(b"previous publication")
    real_run = pipeline.ffmpeg_run

    def fail_encoding(executable, args, stage, duration, callback):
        if stage == "Encoding MP3":
            raise RuntimeError("injected encoder failure")
        return real_run(executable, args, stage, duration, callback)

    monkeypatch.setattr(pipeline, "ffmpeg_run", fail_encoding)
    with pytest.raises(RuntimeError, match="injected encoder failure"):
        pipeline.process(source, target, bypass, progress, overwrite=True)
    assert all(path.read_bytes() == b"previous publication" for path in outputs)
    assert stale_report.read_bytes() == b"previous publication"
    assert not list(tmp_path.glob(".producer-*"))


@pytest.mark.parametrize("overwrite", [False, True])
def test_publish_collision_respects_overwrite(tmp_path, overwrite):
    staged = tmp_path / "staged"
    target = tmp_path / "target"
    staged.write_bytes(b"new")
    target.write_bytes(b"concurrent output")
    if overwrite:
        pipeline.publish(staged, target, overwrite=True)
        assert target.read_bytes() == b"new"
        assert not staged.exists()
    else:
        with pytest.raises(FileExistsError):
            pipeline.publish(staged, target, overwrite=False)
        assert target.read_bytes() == b"concurrent output"
        assert staged.read_bytes() == b"new"


def test_publish_new_output_moves_staged_file(tmp_path):
    staged = tmp_path / "staged"
    target = tmp_path / "target"
    staged.write_bytes(b"finished")
    pipeline.publish(staged, target, overwrite=False)
    assert target.read_bytes() == b"finished"
    assert not staged.exists()


def test_publish_across_filesystems_falls_back_to_copy(tmp_path, monkeypatch):
    """Audio outputs and the shared logs directory can be on different mounts:
    publication then copies atomically into the target directory instead of
    failing with EXDEV."""
    real_link, real_replace = os.link, os.replace

    def refuse_cross_dir(src, dst=None):
        src, dst = Path(src), Path(dst)
        # Simulate two mounts: links and renames only work within one parent.
        if dst.parent != src.parent:
            raise OSError(errno.EXDEV, os.strerror(errno.EXDEV))
        return real_link(src, dst)

    def refuse_cross_rename(src, dst):
        src, dst = Path(src), Path(dst)
        if dst.parent != src.parent:
            raise OSError(errno.EXDEV, os.strerror(errno.EXDEV))
        return real_replace(src, dst)

    monkeypatch.setattr(os, "link", refuse_cross_dir)
    monkeypatch.setattr(os, "replace", refuse_cross_rename)
    outgoing = tmp_path / "out"
    incoming = tmp_path / "in"
    outgoing.mkdir()
    incoming.mkdir()
    # Non-overwrite: copy path must still refuse an existing destination.
    staged = outgoing / "report.json"
    staged.write_bytes(b"new")
    (incoming / "report.json").write_bytes(b"old")
    with pytest.raises(FileExistsError):
        pipeline.publish(staged, incoming / "report.json", overwrite=False)
    assert (incoming / "report.json").read_bytes() == b"old"
    # Successful non-overwrite publication moves the staged file.
    fresh = outgoing / "fresh.json"
    fresh.write_bytes(b"new")
    pipeline.publish(fresh, incoming / "fresh.json", overwrite=False)
    assert (incoming / "fresh.json").read_bytes() == b"new"
    assert not list(incoming.glob(".fresh.json*")) and not fresh.exists()
    # Overwrite replaces the existing destination.
    staged.write_bytes(b"newer")
    pipeline.publish(staged, incoming / "report.json", overwrite=True)
    assert (incoming / "report.json").read_bytes() == b"newer"
    assert not list(incoming.glob(".report.json*")) and not staged.exists()


def test_trim_gaps_shortens_silence_and_keeps_speech_exact(wav, tmp_path, progress):
    from producer.audio import trim_gaps
    sr = 48000
    speech_a = 0.2 * np.sin(2 * np.pi * 440 * np.arange(2 * sr) / sr)
    room_tone = 0.001 * np.random.default_rng(7).normal(0, 1, int(3 * sr))  # -60 dBFS
    speech_b = 0.2 * np.sin(2 * np.pi * 880 * np.arange(2 * sr) / sr)
    source = tmp_path / "gaps.wav"
    sf.write(source, np.concatenate([speech_a, room_tone, speech_b]), sr, format="WAV", subtype="FLOAT")
    target = tmp_path / "trimmed.wav"
    trimmed, removed = trim_gaps(source, target, threshold_db=-50, max_gap_s=1.5,
                                 pad_s=0.6, progress=progress)
    result, rate = sf.read(target, dtype="float32")
    assert rate == 48000 and trimmed == 1
    assert removed == pytest.approx(3 - 0.6, abs=0.02)  # 0.3 s pad kept at each edge
    # Speech is copied through unattenuated and sample-exact.
    np.testing.assert_allclose(result[:len(speech_a)], speech_a, atol=1e-7)
    np.testing.assert_allclose(result[-len(speech_b):], speech_b, atol=1e-6)
    kept_gap = result[len(speech_a):len(speech_a) + int(0.6 * sr)]
    assert np.count_nonzero(kept_gap) == int(0.6 * sr)  # room tone preserved, no gate


def test_trim_and_clearvoice_dispatch_flow(wav, tmp_path, bypass, progress, ffmpeg, monkeypatch):
    def fake_denoise(source, target, device, strength, progress, backend):
        assert backend == "clearvoice"
        import shutil
        shutil.copyfile(source, target)

    monkeypatch.setattr("producer.ai.denoise", fake_denoise)
    bypass.denoiser = "clearvoice"
    bypass.denoise_strength = 1.0
    bypass.trim_silence = True
    bypass.max_gap_s = 1.0
    bypass.gap_pad_s = 0.5
    sr = 48000
    speech = 0.15 * np.sin(2 * np.pi * 440 * np.arange(sr) / sr)
    room_tone = 0.001 * np.random.default_rng(7).normal(0, 1, int(2 * sr))
    source = wav(np.concatenate([speech, room_tone, speech]))
    report = pipeline.process(source, tmp_path / "master.wav", bypass, progress)
    assert report["gaps_trimmed"] == 1
    assert report["silence_removed_seconds"] > 0
    assert report["master"]["duration_seconds"] == pytest.approx(
        report["input"]["duration_seconds"] - report["silence_removed_seconds"], abs=0.02)
    assert report["warnings"] == []