diff options
| author | historia <historiavg@proton.me> | 2026-09-11 18:06:53 -0400 |
|---|---|---|
| committer | historia <historiavg@proton.me> | 2026-09-11 18:06:53 -0400 |
| commit | 3eccf822e14f5f4dd1f6b96e785c0217db647c3e (patch) | |
| tree | 8c3204d3568e713ae98a5cfa59b18f37a72a1722 /lib | |
| parent | 077c82d9614328ff267632298da6c88009313337 (diff) | |
| download | producer-3eccf822e14f5f4dd1f6b96e785c0217db647c3e.tar.gz | |
Remove analyze and channel selection; move reports to logs/
Diffstat (limited to 'lib')
| -rw-r--r-- | lib/project/examples/voice.toml | 1 | ||||
| -rw-r--r-- | lib/project/src/producer/cli.py | 13 | ||||
| -rw-r--r-- | lib/project/src/producer/config.py | 3 | ||||
| -rw-r--r-- | lib/project/src/producer/pipeline.py | 26 | ||||
| -rw-r--r-- | lib/project/tests/test_cli.py | 37 | ||||
| -rw-r--r-- | lib/project/tests/test_config.py | 1 | ||||
| -rw-r--r-- | lib/project/tests/test_pipeline.py | 65 |
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 |
