path: root/lib/tests/test_cli.py

import sys
from types import SimpleNamespace

import numpy as np
import pytest
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_processed.wav"
    rc = process_one(inp, _opts(report=True), single=True)
    assert rc == out
    assert out.exists()
    rep_path = out.with_name("e2e_processed.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


def test_default_output_suffix_processed(tmp_path):
    from producer.cli import _resolve_output

    inp = tmp_path / "song.wav"
    assert _resolve_output(inp, _opts(), single=True) == tmp_path / "song_processed.wav"
    odir = tmp_path / "out"
    opts = _opts(output=str(odir))
    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)
    assert first == tmp_path / "in_processed.wav"
    assert process_one(inp, _opts(), single=True) == tmp_path / "in_processed_1.wav"
    assert process_one(inp, _opts(), single=True) == tmp_path / "in_processed_2.wav"
    assert (tmp_path / "in_processed.wav").exists()


def test_conflict_prompt_overwrite(tmp_path, monkeypatch):
    inp = _mk_wav(tmp_path, "in.wav")
    out = tmp_path / "in_processed.wav"
    process_one(inp, _opts(), single=True)
    monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
    monkeypatch.setattr("builtins.input", lambda _prompt: "o")
    assert process_one(inp, _opts(), single=True) == out
    assert out.exists()


def test_conflict_prompt_rename(tmp_path, monkeypatch):
    inp = _mk_wav(tmp_path, "in.wav")
    out = tmp_path / "in_processed.wav"
    process_one(inp, _opts(), single=True)
    monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
    monkeypatch.setattr("builtins.input", lambda _prompt: "r")
    second = process_one(inp, _opts(), single=True)
    assert second == tmp_path / "in_processed_1.wav"
    assert second.exists()
    assert out.exists()


def test_conflict_prompt_invalid_then_overwrite(tmp_path, monkeypatch):
    inp = _mk_wav(tmp_path, "in.wav")
    out = tmp_path / "in_processed.wav"
    process_one(inp, _opts(), single=True)
    monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
    answers = iter(["maybe", "O"])
    monkeypatch.setattr("builtins.input", lambda _prompt: next(answers))
    assert process_one(inp, _opts(), single=True) == out


def test_conflict_prompt_cancel(tmp_path, monkeypatch, capsys):
    inp = _mk_wav(tmp_path, "in.wav")
    out = tmp_path / "in_processed.wav"
    process_one(inp, _opts(), single=True)
    before = out.read_bytes()
    monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))
    monkeypatch.setattr("builtins.input", lambda _prompt: "c")
    assert process_one(inp, _opts(), single=True) is None
    assert out.read_bytes() == before
    assert "skipped" in capsys.readouterr().out


def test_conflict_prompt_eof_cancels(tmp_path, monkeypatch):
    inp = _mk_wav(tmp_path, "in.wav")
    process_one(inp, _opts(), single=True)
    monkeypatch.setattr(sys, "stdin", SimpleNamespace(isatty=lambda: True))

    def _eof(_prompt):
        raise EOFError

    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()