aboutsummaryrefslogtreecommitdiff
path: root/lib
diff options
context:
space:
mode:
authorhistoria <historiavg@proton.me>2026-09-11 18:06:53 -0400
committerhistoria <historiavg@proton.me>2026-09-11 18:06:53 -0400
commit3eccf822e14f5f4dd1f6b96e785c0217db647c3e (patch)
tree8c3204d3568e713ae98a5cfa59b18f37a72a1722 /lib
parent077c82d9614328ff267632298da6c88009313337 (diff)
downloadproducer-3eccf822e14f5f4dd1f6b96e785c0217db647c3e.tar.gz
Remove analyze and channel selection; move reports to logs/
Diffstat (limited to 'lib')
-rw-r--r--lib/project/examples/voice.toml1
-rw-r--r--lib/project/src/producer/cli.py13
-rw-r--r--lib/project/src/producer/config.py3
-rw-r--r--lib/project/src/producer/pipeline.py26
-rw-r--r--lib/project/tests/test_cli.py37
-rw-r--r--lib/project/tests/test_config.py1
-rw-r--r--lib/project/tests/test_pipeline.py65
7 files changed, 43 insertions, 103 deletions
diff --git a/lib/project/examples/voice.toml b/lib/project/examples/voice.toml
index 0905815..ba0fa5c 100644
--- a/lib/project/examples/voice.toml
+++ b/lib/project/examples/voice.toml
@@ -9,7 +9,6 @@ denoiser = "deepfilter" # DeepFilterNet3 only; fft is explicit non-AI fallback;
device = "auto" # auto, cpu, cuda; explicit cuda fails if unavailable
denoise_strength = 0.85 # 0..1; AI dry/wet blend; zero skips denoising and AI setup
fft_reduction_db = 8.0 # 0.01..30; FFT reduction is scaled by denoise_strength
-channel = "auto" # auto accepts mono only; stereo requires left, right or mix
highpass = true
highpass_hz = 70.0 # 20..300
diff --git a/lib/project/src/producer/cli.py b/lib/project/src/producer/cli.py
index 52cd1bd..4788fb5 100644
--- a/lib/project/src/producer/cli.py
+++ b/lib/project/src/producer/cli.py
@@ -13,9 +13,9 @@ import tomllib
from rich.console import Console
from . import __version__
-from .audio import ffmpeg_run, loudness, measure
+from .audio import ffmpeg_run
from .config import PROFILES, Settings, resolve, toml
-from .pipeline import check_target, process, targets_for
+from .pipeline import check_target, process, report_for, targets_for
from .progress import Display
from .setup import data_dir, ensure_ffmpeg
@@ -24,7 +24,7 @@ def parser():
app = argparse.ArgumentParser(prog="producer", description="Local speech cleanup and podcast mastering. Originals are never overwritten.")
app.add_argument("--version", action="version", version=__version__)
sub = app.add_subparsers(dest="command", required=True)
- for name in ("process", "preview", "config", "analyze"):
+ for name in ("process", "preview", "config"):
cmd = sub.add_parser(name)
cmd.add_argument("--config", type=Path, help="TOML configuration file")
cmd.add_argument("--profile", choices=PROFILES, default=None)
@@ -104,11 +104,6 @@ def main():
print(json.dumps(asdict(settings), indent=2) if args.json else toml(settings), end="\n" if args.json else "")
return
with Display() as display:
- if args.command == "analyze":
- stats = measure(args.inputs[0], display.update)
- stats["loudness"] = loudness(ensure_ffmpeg(), args.inputs[0], settings, stats["duration_seconds"], display.update)
- print(json.dumps(stats, indent=2, allow_nan=False))
- return
if args.command == "process":
if args.output and len(args.inputs) != 1:
raise ValueError("--output requires exactly one input")
@@ -156,7 +151,7 @@ def main():
output = output_dir / f"{source.stem}.{profile}.preview.wav"
if output.resolve() == original:
raise ValueError("Preview output would overwrite the original")
- targets = [output, output.with_suffix(".report.json")]
+ targets = [output, report_for(output)]
resolved = {target.resolve() for target in targets}
if len(resolved) < len(targets) or resolved & claimed:
raise ValueError("Preview outputs have duplicate names")
diff --git a/lib/project/src/producer/config.py b/lib/project/src/producer/config.py
index ef0cd49..b1d84cc 100644
--- a/lib/project/src/producer/config.py
+++ b/lib/project/src/producer/config.py
@@ -12,7 +12,6 @@ class Settings:
device: str = "auto"
denoise_strength: float = 0.85
fft_reduction_db: float = 8.0
- channel: str = "auto"
highpass: bool = True
highpass_hz: float = 70.0
lowpass: bool = False
@@ -97,7 +96,7 @@ def resolve(path: Path | None = None, profile: str | None = None,
raise ValueError(f"{key} must be between {RANGES[key][0]} and {RANGES[key][1]}")
for key, options in {
"denoiser": ("deepfilter", "fft", "none"), "device": ("auto", "cpu", "cuda"),
- "channel": ("auto", "left", "right", "mix"), "hum_hz": (0, 50, 60),
+ "hum_hz": (0, 50, 60),
"sample_rate": (44100, 48000), "mp3_bitrate": (128, 160, 192, 224, 256, 320),
}.items():
if merged[key] not in options:
diff --git a/lib/project/src/producer/pipeline.py b/lib/project/src/producer/pipeline.py
index 8fb5484..d7c2ced 100644
--- a/lib/project/src/producer/pipeline.py
+++ b/lib/project/src/producer/pipeline.py
@@ -31,9 +31,14 @@ def filters(settings: Settings) -> list[str]:
return result or ["anull"]
+def report_for(destination: Path) -> Path:
+ """Diagnostic reports live in a logs/ directory, out of the way of outputs."""
+ return destination.parent / "logs" / destination.with_suffix(".report.json").name
+
+
def targets_for(destination: Path, settings: Settings) -> list[Path]:
"""Every path a successful run publishes: master WAV, report, optional MP3."""
- targets = [destination, destination.with_suffix(".report.json")]
+ targets = [destination, report_for(destination)]
if settings.mp3:
targets.append(destination.with_suffix(".mp3"))
return targets
@@ -66,18 +71,17 @@ def process(source: Path, destination: Path, settings: Settings, progress,
raise ValueError(f"Input must be an existing WAV file: {source}")
if destination.suffix.lower() != ".wav":
raise ValueError("Master output must have a .wav extension")
- report_path = destination.with_suffix(".report.json")
+ report_path = report_for(destination)
for target in targets_for(destination, settings):
check_target(source, target, overwrite)
ffmpeg = ensure_ffmpeg()
original = measure(source, progress, "Analyzing original")
if original["sample_peak_dbfs"] < -100:
raise ValueError("Recording is silent or too quiet to master safely")
- channel = settings.channel
- if original["channels"] == 2 and channel == "auto":
- raise ValueError("Stereo input: select --channel left, right, or mix explicitly. Auto downmix could cancel your voice.")
- if original["channels"] == 1 and channel == "right":
- raise ValueError("Cannot select the right channel of a mono recording")
+ if original["channels"] == 2:
+ raise ValueError("Stereo input: downmix to mono first, for example "
+ "'ffmpeg -i input.wav -ac 1 mono.wav'. An automatic "
+ "downmix could cancel your voice.")
duration = original["duration_seconds"]
warnings = []
if original["near_full_scale_samples"]:
@@ -85,13 +89,11 @@ def process(source: Path, destination: Path, settings: Settings, progress,
if max(abs(x) for x in original["dc_offset"]) > 0.01:
warnings.append("Significant input DC offset detected; enable highpass filtering to remove it.")
destination.parent.mkdir(parents=True, exist_ok=True)
+ report_path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.TemporaryDirectory(prefix=".producer-", dir=destination.parent) as temporary:
work = Path(temporary)
prepared, cleaned, shaped = [work / name for name in ("prepared.wav", "cleaned.wav", "shaped.wav")]
input_filters = []
- if original["channels"] == 2:
- input_filters.append({"left": "pan=mono|c0=c0", "right": "pan=mono|c0=c1",
- "mix": "pan=mono|c0=0.5*c0+0.5*c1"}[channel])
if settings.highpass:
input_filters.append(f"highpass=f={settings.highpass_hz}:p=2")
if settings.hum_hz:
@@ -110,8 +112,8 @@ def process(source: Path, destination: Path, settings: Settings, progress,
cleaned = prepared
clean_stats = measure(cleaned, progress, "Measuring cleaned voice")
if clean_stats["sample_peak_dbfs"] < -100:
- raise ValueError("The selected audio is silent or too quiet to master safely; "
- "check --channel, the mix, and cleanup settings")
+ raise ValueError("The audio is silent or too quiet to master safely; "
+ "check the denoise and cleanup settings")
gain = 0.0
if settings.leveling:
speech_db = clean_stats["speech_level_estimate_dbfs"]
diff --git a/lib/project/tests/test_cli.py b/lib/project/tests/test_cli.py
index 300b3a8..e02aad1 100644
--- a/lib/project/tests/test_cli.py
+++ b/lib/project/tests/test_cli.py
@@ -45,12 +45,12 @@ def test_preview_real_audio_preserves_original_and_reports_provenance(
assert source.read_bytes() == original
assert len(modes) == 2
assert {path.name for path in output_dir.iterdir()} == {
- f"input.{profile}.preview{suffix}"
- for profile in ("natural", "radio") for suffix in (".wav", ".report.json")
- }
+ f"input.{profile}.preview.wav"
+ for profile in ("natural", "radio")
+ } | {"logs"}
for profile, mode in zip(("natural", "radio"), modes):
target = output_dir / f"input.{profile}.preview.wav"
- report = json.loads(target.with_suffix(".report.json").read_text())
+ report = json.loads(pipeline.report_for(target).read_text())
info = sf.info(target)
assert info.duration == pytest.approx(5, abs=0.02)
assert info.channels == 1
@@ -88,19 +88,6 @@ def test_config_stdout_json_and_set_precedence(tmp_path, invoke, capsys):
assert settings["denoiser"] == "none"
-def test_analyze_stdout_is_json(wav, ffmpeg, invoke, capsys):
- source = wav()
- original = source.read_bytes()
- invoke("analyze", source, "--no-denoise")
- stats = json.loads(capsys.readouterr().out)
- assert stats["duration_seconds"] == pytest.approx(3)
- assert stats["channels"] == 1
- assert stats["frames"] == 144000
- assert np.isfinite(stats["loudness"]["input_i"])
- assert source.read_bytes() == original
- assert list(source.parent.iterdir()) == [source]
-
-
@pytest.mark.parametrize("start", ["nan", "inf", "-inf"])
def test_preview_rejects_nonfinite_start_before_tool_setup(
tmp_path, invoke, capsys, monkeypatch, start
@@ -148,7 +135,7 @@ def test_existing_output_is_not_overwritten(wav, tmp_path, invoke, capsys, monke
assert "Output exists" in capsys.readouterr().err
assert target.read_bytes() == b"previous master"
assert source.read_bytes() == original
- assert not target.with_suffix(".report.json").exists()
+ assert not pipeline.report_for(target).exists()
def test_overwrite_cannot_replace_input(wav, invoke, capsys, monkeypatch):
@@ -183,7 +170,7 @@ def test_default_outputs_are_written_beside_each_input(wav, tmp_path, ffmpeg, in
invoke("process", *sources, "--no-denoise")
for source in sources:
assert {path.name for path in source.parent.iterdir()} == {
- "take.wav", "take.natural.wav", "take.natural.report.json"}
+ "take.wav", "take.natural.wav", "logs"}
captured = capsys.readouterr()
assert captured.out == ""
assert captured.err.count("Written:") == 2
@@ -240,8 +227,8 @@ def test_preview_default_output_is_written_beside_input(wav, tmp_path, ffmpeg, i
invoke("preview", source, "--no-denoise", "--start", "1", "--duration", "2",
"--profiles", "natural")
assert {path.name for path in tmp_path.iterdir()} == {
- "input.wav", "input.natural.preview.wav", "input.natural.preview.report.json"}
- report = json.loads((tmp_path / "input.natural.preview.report.json").read_text())
+ "input.wav", "input.natural.preview.wav", "logs"}
+ report = json.loads(pipeline.report_for(tmp_path / "input.natural.preview.wav").read_text())
assert report["settings"]["profile"] == "natural"
assert source.read_bytes() == original
@@ -275,7 +262,8 @@ def test_nonfile_output_with_overwrite_fails_before_tool_setup(
source = wav()
original = source.read_bytes()
target = tmp_path / "master.wav"
- nonfile = target.with_suffix(suffix)
+ nonfile = target.with_suffix(suffix) if suffix != ".report.json" else pipeline.report_for(target)
+ nonfile.parent.mkdir(parents=True, exist_ok=True)
nonfile.mkdir()
bypass.mp3 = True
monkeypatch.setattr(pipeline, "ensure_ffmpeg", lambda: pytest.fail("Non-file output reached FFmpeg"))
@@ -283,5 +271,8 @@ def test_nonfile_output_with_overwrite_fails_before_tool_setup(
pipeline.process(source, target, bypass, progress, overwrite=True)
assert nonfile.is_dir()
assert list(nonfile.iterdir()) == []
- assert set(tmp_path.iterdir()) == {source, nonfile}
+ if suffix == ".report.json":
+ assert {path.name for path in tmp_path.iterdir()} == {source.name, "logs"}
+ else:
+ assert set(tmp_path.iterdir()) == {source, nonfile}
assert source.read_bytes() == original
diff --git a/lib/project/tests/test_config.py b/lib/project/tests/test_config.py
index d3321d5..02b9eba 100644
--- a/lib/project/tests/test_config.py
+++ b/lib/project/tests/test_config.py
@@ -73,7 +73,6 @@ def test_numeric_ranges_are_inclusive_and_reject_outside(key):
({"target_lufs": float("inf")}, "finite number"),
({"denoiser": "unknown"}, "denoiser must be one of"),
({"device": "gpu"}, "device must be one of"),
- ({"channel": "stereo"}, "channel must be one of"),
({"hum_hz": 55}, "hum_hz must be one of"),
({"sample_rate": 96000}, "sample_rate must be one of"),
({"mp3_bitrate": 64}, "mp3_bitrate must be one of"),
diff --git a/lib/project/tests/test_pipeline.py b/lib/project/tests/test_pipeline.py
index 3f6000d..797bc12 100644
--- a/lib/project/tests/test_pipeline.py
+++ b/lib/project/tests/test_pipeline.py
@@ -41,7 +41,8 @@ def test_original_is_never_overwritten(wav, tmp_path, bypass, progress, monkeypa
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 = pipeline.report_for(target) if suffix == ".report.json" else 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"))
@@ -53,20 +54,16 @@ def test_existing_output_or_sidecar_rejected_before_processing(wav, tmp_path, by
@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", "select --channel"),
- ("mono-right", "right channel of a mono"),
+ ("nonfinite", "non-finite"), ("stereo", "downmix to mono"),
])
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,
- "mono-right": np.ones(4800) * 0.1,
}[kind]
source = wav(samples)
target = tmp_path / "outputs" / "master.wav"
- if kind == "mono-right":
- bypass.channel = "right"
with pytest.raises(ValueError, match=message):
pipeline.process(source, target, bypass, progress)
assert not target.parent.exists()
@@ -86,26 +83,6 @@ def test_measure_reports_channels_peaks_clipping_and_progress(wav):
assert events[-1][1:] == (4800, 4800)
-@pytest.mark.parametrize("channel", ["left", "right", "mix"])
-def test_explicit_stereo_selection_preserves_signal_and_duration(wav, tmp_path, bypass, progress, ffmpeg, channel):
- time = np.arange(48000 * 3) / 48000
- left = 0.2 * np.sin(2 * np.pi * 440 * time)
- right = 0.1 * np.sin(2 * np.pi * 880 * time)
- source = wav(np.column_stack([left, right]))
- bypass.channel = channel
- target = tmp_path / "master.wav"
- report = pipeline.process(source, target, bypass, progress)
- actual, rate = sf.read(target)
- expected = {"left": left, "right": right, "mix": (left + right) / 2}[channel]
- assert actual.ndim == 1
- assert rate == 48000
- assert len(actual) == len(expected)
- np.testing.assert_allclose(actual, expected, atol=5e-7, rtol=0)
- assert report["normalization_mode"] == "disabled"
- assert report["level_correction_db"] == 0
- assert sf.info(target).subtype == "PCM_32"
-
-
@pytest.mark.parametrize("rate", [44100, 48000])
def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg, rate):
source = wav()
@@ -123,7 +100,7 @@ def test_normalization_peak_ceiling_and_report(wav, tmp_path, progress, ffmpeg,
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(target.with_suffix(".report.json").read_text()) == report
+ assert json.loads(pipeline.report_for(target).read_text()) == report
assert source.read_bytes() == original
assert not list(target.parent.glob(".producer-*"))
@@ -173,30 +150,6 @@ def test_leveling_skipped_when_speech_estimate_is_unreliable(wav, tmp_path, bypa
assert any("leveling skipped" in warning for warning in report["warnings"])
-def test_antiphase_mix_is_rejected_after_selection(wav, tmp_path, bypass, progress, ffmpeg):
- time = np.arange(48000 * 3) / 48000
- tone = 0.2 * np.sin(2 * np.pi * 440 * time)
- source = wav(np.column_stack([tone, -tone]))
- bypass.channel = "mix"
- target = tmp_path / "master.wav"
- with pytest.raises(ValueError, match="selected audio is silent or too quiet"):
- pipeline.process(source, target, bypass, progress)
- assert not target.exists()
- assert not target.with_suffix(".report.json").exists()
- assert not list(tmp_path.glob(".producer-*"))
-
-
-def test_silent_selected_channel_is_rejected_after_selection(wav, tmp_path, bypass, progress, ffmpeg):
- time = np.arange(48000 * 3) / 48000
- tone = 0.2 * np.sin(2 * np.pi * 440 * time)
- source = wav(np.column_stack([tone, np.zeros_like(tone)]))
- bypass.channel = "right"
- target = tmp_path / "master.wav"
- with pytest.raises(ValueError, match="selected audio is silent or too quiet"):
- pipeline.process(source, target, bypass, progress)
- assert not target.exists()
-
-
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")
@@ -214,14 +167,15 @@ def test_fft_and_mp3_export_can_replace_generated_outputs(wav, tmp_path, bypass,
bypass.denoiser = "fft"
bypass.mp3 = True
target = tmp_path / "master.wav"
- for suffix in (".wav", ".report.json", ".mp3"):
- target.with_suffix(suffix).write_bytes(b"old output")
+ for path in (target, pipeline.report_for(target), target.with_suffix(".mp3")):
+ path.parent.mkdir(parents=True, exist_ok=True)
+ 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
- assert json.loads(target.with_suffix(".report.json").read_text()) == report
+ assert json.loads(pipeline.report_for(target).read_text()) == report
assert sf.info(target).channels == 1
assert source.read_bytes() == original
@@ -230,8 +184,9 @@ def test_render_failure_preserves_existing_outputs_and_removes_temporary_files(w
source = wav()
target = tmp_path / "master.wav"
bypass.mp3 = True
- outputs = [target.with_suffix(suffix) for suffix in (".wav", ".report.json", ".mp3")]
+ outputs = [target, pipeline.report_for(target), target.with_suffix(".mp3")]
for path in outputs:
+ path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(b"previous publication")
real_run = pipeline.ffmpeg_run