diff --git a/README.md b/README.md index 0624f08..00a4b47 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,22 @@ sermon-clean si sermon.ogg --silence-threshold -35 --silence-min-duration 0.5 # 1c. Or: multiband waveform — N seconds per row, makes timestamp counting trivial sermon-clean mb sermon.ogg --band-seconds 60 --width 80 +# 1d. Or: silence stats — count/mean/longest silence. Use this to pick the right +# threshold for the next recording of the same speaker/room setup. +sermon-clean silence-stats sermon.ogg + +# 1e. Or: auto-tune the silence threshold based on expected pause density. +# Useful when the same speaker records in different rooms and the silence +# profile changes week to week. +sermon-clean threshold-tune sermon.ogg --target-spm 4.0 + +# 1f. Or: pre-flight normalization — bring the recording to broadcast-standard +# loudness before doing any other processing. +sermon-clean normalize sermon.ogg -o sermon-normalized.ogg + +# 1g. Or: light denoise (FFT-based) if the recording has hiss / mic preamp noise. +sermon-clean denoise sermon.ogg -o sermon-denoised.ogg + # 2. Or: extract overlapping slices for manual review sermon-clean slices sermon.ogg --output-dir ./slices \ --slice-seconds 5 --overlap-seconds 1 @@ -90,11 +106,33 @@ sermon-clean auto sermon.ogg --bad-words "fuck,shit,damn" --output segs.json | `scan` | ASCII waveform + silence marks (no transcription) | | `silence-index` (`si`) | Tabular list of silence runs with timestamps + position bar | | `multiband` (`mb`) | Multi-row ASCII waveform with band-start labels for timestamp counting | +| `silence-stats` (`ss`) | Quantitative summary: count, mean, median, longest silence + density per minute | +| `threshold-tune` (`tt`) | Auto-pick the silence threshold that matches expected pause density | +| `normalize` (`norm`) | Apply EBU R128 loudness normalization (target LUFS, true peak, LRA) | +| `denoise` | Apply light FFT-based noise reduction (`afftdn`) for hiss / mic preamp noise | | `slices` | Extract overlapping audio chunks for manual review | | `cut` | Trim the original around bad windows (no splice) | | `paste` | Splice pre-rendered replacements into the trimmed original | | `auto` | Transcribe + find bad-word timestamps | | `pipe` | Run cut + paste in one command | +| `batch` | Run any of the above across many files | + +### Batch processing + +Apply the same operation to many files at once: + +```bash +# Normalize every sermon from this Sunday +sermon-clean batch normalize 'sermons/*.ogg' --output-dir fixed/ --suffix=-normalized + +# Denoise a batch of older recordings +sermon-clean batch denoise 'archive/*.wav' --output-dir fixed/ --suffix=-dn + +# Silence stats for every recording (no output files — runs the stats print) +sermon-clean batch silence-stats 'sermons/*.ogg' --output-dir stats/ +``` + +Failures are collected, not raised: if one file is corrupt, the rest still process. ## How it works @@ -112,6 +150,10 @@ The previous workflow (`audio-splice-workflow.md` in krystie-profile) was a 5-st - `scan` on a 27-min audio: ~50s (Opus decode is the bottleneck on this CPU) - `slices` on a 27-min audio with 5-sec slices: ~50s - `auto` on a 27-min audio: ~10-20 min with `tiny.en` model on CPU. Use `base.en` or `small.en` for better accuracy at 2-4x the time. Whisper `tiny.en` is the right model for finding a known bad-word list — you don't need higher accuracy than "did the word 'fuck' appear at all." +- `normalize` on a 27-min audio: ~30-60s (two-pass EBU R128 measurement + apply) +- `denoise` on a 27-min audio: ~45-90s (FFT pass over the whole file) +- `silence-stats` and `threshold-tune` on a 27-min audio: ~5s (just runs silencedetect with multiple thresholds) +- `batch`: adds ~5s of subprocess overhead per file on top of the operation cost ## Development diff --git a/pyproject.toml b/pyproject.toml index aafff0e..e7ebc77 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta" [project] name = "sermon-clean" -version = "0.1.0" -description = "Find bad segments in sermon audio, cut them out, splice in ElevenLabs replacements." +version = "0.2.0" +description = "Find bad segments in sermon audio, cut them out, splice in ElevenLabs replacements. Includes normalize, denoise, silence-stats, threshold-tune, and batch operations." readme = "README.md" requires-python = ">=3.9" license = {text = "MIT"} diff --git a/sermon_clean/batch.py b/sermon_clean/batch.py new file mode 100644 index 0000000..0511803 --- /dev/null +++ b/sermon_clean/batch.py @@ -0,0 +1,160 @@ +"""Batch processing — run any sermon-clean subcommand across many files. + +Real workflow: a preacher records 4-5 sermons on Sunday, all need the same +post-processing. Running each one by hand is tedious and error-prone. + +`batch` takes a glob pattern and runs the requested subcommand on each +file. Output goes into a sibling directory tree with `-` +filenames. Errors are collected, not raised, so one bad file doesn't +kill the whole batch. +""" + +from __future__ import annotations + +import glob +import re +import shutil +import subprocess +from dataclasses import dataclass, field +from pathlib import Path +from typing import List, Optional + + +@dataclass +class BatchItem: + src: Path + output: Path + returncode: int = 0 + stderr: str = "" + stdout: str = "" + + @property + def ok(self) -> bool: + return self.returncode == 0 + + +@dataclass +class BatchResult: + items: List[BatchItem] = field(default_factory=list) + n_ok: int = 0 + n_failed: int = 0 + + def summary(self) -> str: + lines = [f"batch: {self.n_ok} ok, {self.n_failed} failed (of {len(self.items)} total)"] + for item in self.items: + marker = "✓" if item.ok else "✗" + lines.append(f" {marker} {item.src.name} -> {item.output}") + if not item.ok and item.stderr: + # First non-empty stderr line for diagnosis + for line in item.stderr.splitlines(): + if line.strip(): + lines.append(f" stderr: {line.strip()}") + break + return "\n".join(lines) + + +def _expand_globs(patterns: List[str]) -> List[Path]: + """Expand a list of glob patterns into a sorted, deduped list of files.""" + seen = set() + out = [] + for pattern in patterns: + for path in sorted(glob.glob(pattern, recursive=True)): + p = Path(path) + if p.is_file() and p.resolve() not in seen: + seen.add(p.resolve()) + out.append(p) + return out + + +def _make_output_path(src: Path, output_dir: Path, suffix: str, ext: Optional[str]) -> Path: + """Compute the output path for a given input. + + Filename pattern: , where ext defaults to src.suffix. + """ + out_ext = ext if ext is not None else src.suffix + out_name = f"{src.stem}{suffix}{out_ext}" + return output_dir / out_name + + +# Valid subcommands for batch. Keep this in sync with cli.py. +BATCH_SUBCOMMANDS = {"find", "scan", "si", "silence-index", "mb", "multiband", + "slices", "cut", "paste", "auto", "pipe", + "normalize", "silence-stats", "threshold-tune", "denoise"} + + +def run_batch( + subcommand: str, + sources: List[Path], + output_dir: Path, + *, + output_suffix: str = "-fixed", + output_extension: Optional[str] = None, + extra_args: Optional[List[str]] = None, +) -> BatchResult: + """Run `sermon-clean ` over each source file. + + Args: + subcommand: which sermon-clean subcommand to invoke. + sources: list of input audio files. + output_dir: directory to write outputs to. Created if missing. + output_suffix: filename suffix for outputs (e.g. '-normalized'). + Ignored for subcommands that don't produce an output file + (find/scan/si/mb/silence-stats/threshold-tune/slices). + output_extension: extension override (e.g. '.wav'). Defaults to + src.suffix. + extra_args: additional args to pass to the subcommand. Note that + `--suffix` is reserved by `batch` itself — strip it if you + forward it via this param. (We don't currently forward anything + from the batch CLI to subcommands other than the implicit + `-o `.) + + Returns: + BatchResult with per-file success/failure. + + Raises: + ValueError: if subcommand is not a recognized sermon-clean command. + """ + if subcommand not in BATCH_SUBCOMMANDS: + raise ValueError( + f"unknown subcommand for batch: {subcommand!r}. " + f"valid: {sorted(BATCH_SUBCOMMANDS)}" + ) + + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + # Find the sermon-clean executable in PATH or fall back to python -m + sc_bin = shutil.which("sermon-clean") + if sc_bin: + cmd_base = [sc_bin] + else: + import sys + cmd_base = [sys.executable, "-m", "sermon_clean.cli"] + + result = BatchResult() + extra_args = extra_args or [] + + for src in sources: + src = Path(src) + out = _make_output_path(src, output_dir, output_suffix, output_extension) + # Build the per-file argv. -o / --output is added only for subcommands + # that produce a single output file. + output_subcommands = {"cut", "paste", "normalize", "denoise", "pipe"} + argv = cmd_base + [subcommand, str(src)] + extra_args + if subcommand in output_subcommands: + argv += ["-o", str(out)] + + proc = subprocess.run(argv, capture_output=True, text=True) + result.items.append(BatchItem( + src=src, + output=out, + returncode=proc.returncode, + stdout=proc.stdout, + stderr=proc.stderr, + )) + if proc.returncode == 0: + result.n_ok += 1 + else: + result.n_failed += 1 + + return result diff --git a/sermon_clean/cli.py b/sermon_clean/cli.py index 06d7530..2d8e162 100644 --- a/sermon_clean/cli.py +++ b/sermon_clean/cli.py @@ -194,6 +194,124 @@ def cmd_multiband(args: argparse.Namespace) -> int: return 0 +def cmd_normalize(args: argparse.Namespace) -> int: + """Apply EBU R128 loudness normalization.""" + from .processing import normalize_loudness + audio = Path(args.audio) + if not audio.exists(): + print(f"error: {audio} not found", file=sys.stderr) + return 1 + output = Path(args.output) if args.output else audio.with_name(f"{audio.stem}-normalized{audio.suffix}") + result = normalize_loudness( + audio, output, + target_lufs=args.target_lufs, + true_peak_db=args.true_peak, + loudness_range_lu=args.loudness_range, + ) + print(f"normalized: {result['output_path']}") + print(f"target: {result['target_lufs']} LUFS, true peak {result['true_peak_db']}dB") + print(f"measured input loudness: {result['measured_input_i']} LUFS") + print(f"applied gain offset: {result['applied_offset']} dB") + return 0 + + +def cmd_silence_stats(args: argparse.Namespace) -> int: + """Quantitative summary of silence distribution (count, mean, longest).""" + from .processing import silence_stats + import json + audio = Path(args.audio) + if not audio.exists(): + print(f"error: {audio} not found", file=sys.stderr) + return 1 + stats = silence_stats( + audio, + threshold_db=args.silence_threshold, + min_duration_seconds=args.silence_min_duration, + ) + if args.json: + print(json.dumps(stats.to_dict(), indent=2)) + return 0 + # Human-readable summary + print(f"file: {stats.audio_path}") + print(f"duration: {stats.duration_seconds:.2f}s") + print(f"threshold: {stats.threshold_db}dB, min duration: {stats.min_duration_seconds}s") + print() + print(f"silence runs: {stats.n_silence_runs}") + print(f"total silence: {stats.total_silence_seconds:.2f}s ({stats.silence_fraction*100:.1f}% of audio)") + print(f"mean silence: {stats.mean_silence_seconds:.2f}s") + print(f"median silence: {stats.median_silence_seconds:.2f}s") + print(f"longest silence: {stats.longest_silence_seconds:.2f}s " + f"@ {stats.longest_silence_range[0]:.1f}-{stats.longest_silence_range[1]:.1f}s") + print(f"silence density: {stats.silence_per_minute:.2f} runs/min") + return 0 + + +def cmd_threshold_tune(args: argparse.Namespace) -> int: + """Auto-pick the silence threshold for this audio based on expected pause density.""" + from .processing import threshold_tune + import json + audio = Path(args.audio) + if not audio.exists(): + print(f"error: {audio} not found", file=sys.stderr) + return 1 + candidates = [float(x) for x in args.candidates.split(",")] + result = threshold_tune( + audio, + candidates=candidates, + target_silence_per_minute=args.target_spm, + target_min_duration=args.min_duration, + ) + if args.json: + print(json.dumps(result, indent=2)) + return 0 + print(f"target: {result['target_silence_per_minute']:.1f} silence runs/min") + print(f"picked threshold: {result['picked_threshold_db']}dB " + f"(gives {result['picked_silence_per_minute']:.2f} runs/min)") + print() + print(f"{'threshold_dB':>12} {'runs':>6} {'runs/min':>10} {'distance':>10}") + for row in result["candidates"]: + marker = " ← picked" if row["threshold_db"] == result["picked_threshold_db"] else "" + print(f"{row['threshold_db']:>12.0f} {row['n_runs']:>6} {row['silence_per_minute']:>10.2f} " + f"{row['distance_from_target']:>10.2f}{marker}") + return 0 + + +def cmd_denoise(args: argparse.Namespace) -> int: + """Apply light FFT-based noise reduction (good for tape hiss, mic preamp noise).""" + from .denoise import denoise + audio = Path(args.audio) + if not audio.exists(): + print(f"error: {audio} not found", file=sys.stderr) + return 1 + output = Path(args.output) if args.output else audio.with_name(f"{audio.stem}-denoised{audio.suffix}") + result = denoise( + audio, output, + noise_reduction_db=args.noise_reduction, + noise_floor_db=args.noise_floor, + ) + print(f"denoised: {result.output_path}") + print(f"noise reduction: {result.noise_reduction_db}dB (assumed floor: {result.noise_floor_db}dB)") + return 0 + + +def cmd_batch(args: argparse.Namespace) -> int: + """Run a sermon-clean subcommand across many files.""" + from .batch import run_batch + output_dir = Path(args.output_dir) + result = run_batch( + args.subcommand, + [Path(p) for p in args.sources], + output_dir, + output_suffix=args.suffix or "-fixed", + output_extension=args.extension, + # Don't forward --suffix or --extension to the subcommand — they're + # batch-only knobs that control output naming, not subcommand args. + extra_args=[], + ) + print(result.summary()) + return 0 if result.n_failed == 0 else 1 + + def cmd_slices(args: argparse.Namespace) -> int: """Extract overlapping audio slices for manual review. @@ -376,6 +494,52 @@ def main(argv: list[str] | None = None) -> int: p_slices.add_argument("--overlap-seconds", type=float, default=1.0, help="overlap between adjacent slices") p_slices.set_defaults(func=cmd_slices) + # normalize (EBU R128 loudness normalization) + p_norm = sub.add_parser("normalize", aliases=["norm"], help="Apply EBU R128 loudness normalization") + p_norm.add_argument("audio") + p_norm.add_argument("-o", "--output", help="output file (default: -normalized)") + p_norm.add_argument("--target-lufs", type=float, default=-16.0, help="target integrated loudness (LUFS, default -16 for podcast/YouTube)") + p_norm.add_argument("--true-peak", type=float, default=-1.5, help="true peak ceiling in dB (default -1.5)") + p_norm.add_argument("--loudness-range", type=float, default=11.0, help="loudness range target in LU (default 11)") + p_norm.set_defaults(func=cmd_normalize) + + # silence-stats (quantitative summary of silence distribution) + p_ss = sub.add_parser("silence-stats", aliases=["ss"], help="Quantitative summary of silence distribution") + p_ss.add_argument("audio") + p_ss.add_argument("--silence-threshold", type=float, default=-35.0, help="noise_db threshold") + p_ss.add_argument("--silence-min-duration", type=float, default=0.3, help="minimum silence duration in seconds") + p_ss.add_argument("--json", action="store_true", help="output JSON instead of human-readable") + p_ss.set_defaults(func=cmd_silence_stats) + + # threshold-tune (auto-pick silence threshold for this audio) + p_tt = sub.add_parser("threshold-tune", aliases=["tt"], help="Auto-pick the silence threshold based on expected pause density") + p_tt.add_argument("audio") + p_tt.add_argument("--candidates", default="-25,-30,-35,-40,-45,-50", + help="comma-separated threshold candidates in dB (default: -25,-30,-35,-40,-45,-50)") + p_tt.add_argument("--target-spm", type=float, default=4.0, + help="target silence runs per minute (default 4.0 — typical sermon pause rate)") + p_tt.add_argument("--min-duration", type=float, default=0.3, + help="minimum silence duration in seconds") + p_tt.add_argument("--json", action="store_true", help="output JSON instead of human-readable") + p_tt.set_defaults(func=cmd_threshold_tune) + + # denoise (light FFT-based noise reduction) + p_dn = sub.add_parser("denoise", help="Apply light FFT-based noise reduction (afftdn)") + p_dn.add_argument("audio") + p_dn.add_argument("-o", "--output", help="output file (default: -denoised)") + p_dn.add_argument("--noise-reduction", type=float, default=12.0, help="noise attenuation in dB (default 12; higher = more aggressive)") + p_dn.add_argument("--noise-floor", type=float, default=-50.0, help="expected noise floor in dB (default -50)") + p_dn.set_defaults(func=cmd_denoise) + + # batch (run any subcommand across many files) + p_batch = sub.add_parser("batch", help="Run a sermon-clean subcommand across many files") + p_batch.add_argument("subcommand", help="which sermon-clean subcommand to run (normalize, scan, etc.)") + p_batch.add_argument("sources", nargs="+", help="input files or glob patterns (e.g. 'sermons/*.ogg')") + p_batch.add_argument("--output-dir", required=True, help="directory to write outputs") + p_batch.add_argument("--suffix", default="-fixed", help="filename suffix for outputs (default '-fixed')") + p_batch.add_argument("--extension", help="extension override (e.g. '.wav'); defaults to input extension") + p_batch.set_defaults(func=cmd_batch) + # scan (ASCII waveform + silence marks, no whisper) p_scan = sub.add_parser("scan", help="Print an ASCII waveform + silence marks for the audio") p_scan.add_argument("audio") diff --git a/sermon_clean/denoise.py b/sermon_clean/denoise.py new file mode 100644 index 0000000..992a9e4 --- /dev/null +++ b/sermon_clean/denoise.py @@ -0,0 +1,76 @@ +"""Light noise reduction via ffmpeg's afftdn filter. + +Use cases: +- Recordings with hiss (cheap mic preamp, RF interference) +- Recordings with AC hum (50/60Hz line noise — there's a separate `hum` filter + for that, not implemented here) +- Tapes digitized with analog tape hiss + +The `afftdn` filter is ffmpeg's built-in adaptive FFT denoiser. It's +lightweight (CPU-friendly) and good for gentle hiss removal without +artifacting speech. For heavy noise, consider a real denoiser like +RNNoise or DeepFilterNet — out of scope for sermon-clean. +""" + +from __future__ import annotations + +import subprocess +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class DenoiseResult: + output_path: Path + noise_floor_db: float + noise_reduction_db: float + + def to_dict(self) -> dict: + return { + "output_path": str(self.output_path), + "noise_floor_db": self.noise_floor_db, + "noise_reduction_db": self.noise_reduction_db, + } + + +def denoise( + src: Path, + out: Path, + *, + noise_reduction_db: float = 12.0, + noise_floor_db: float = -50.0, +) -> DenoiseResult: + """Apply FFT-based noise reduction. + + Args: + src: input audio file + out: output audio file + noise_reduction_db: how much to attenuate the noise component (dB). + Higher = more aggressive. 12dB is a reasonable default; 20dB + starts to artifact speech. + noise_floor_db: expected level of the noise floor below which everything + is considered noise. -50dB is conservative. If your recording has + louder noise (e.g. AC hum at -35dB), set this higher. + + Returns: + DenoiseResult with the measured parameters. + """ + src = Path(src) + out = Path(out) + + proc = subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-i", str(src), + "-af", f"afftdn=nr={noise_reduction_db}:nf={noise_floor_db}", + "-ar", "48000", # output 48kHz to match normalize convention + str(out), + ], capture_output=True, text=True) + + if proc.returncode != 0: + raise RuntimeError(f"denoise failed: {proc.stderr}") + + return DenoiseResult( + output_path=out, + noise_floor_db=noise_floor_db, + noise_reduction_db=noise_reduction_db, + ) diff --git a/sermon_clean/processing.py b/sermon_clean/processing.py new file mode 100644 index 0000000..67295ef --- /dev/null +++ b/sermon_clean/processing.py @@ -0,0 +1,271 @@ +"""Audio processing — normalize, silence stats, threshold auto-tune. + +Three related operations that don't belong in engine.py (which is about +splicing segments) but also don't deserve their own module each. + +- normalize: Apply EBU R128 two-pass loudness normalization so different + recordings have consistent volume. +- silence_stats: Quantitative summary of silence distribution — useful + for picking the right threshold for a given recording. +- threshold_tune: Auto-pick the silence threshold that best separates + speech from breath pauses for the given audio. +""" + +from __future__ import annotations + +import math +import re +import subprocess +from dataclasses import dataclass, asdict +from pathlib import Path +from typing import List, Tuple + + +# --------------------------------------------------------------------------- +# normalize +# --------------------------------------------------------------------------- + + +def normalize_loudness( + src: Path, + out: Path, + *, + target_lufs: float = -16.0, + true_peak_db: float = -1.5, + loudness_range_lu: float = 11.0, +) -> dict: + """Normalize audio to broadcast-standard EBU R128 loudness. + + Two-pass: first pass measures the integrated loudness, second pass + applies the corrective gain. Default target is -16 LUFS (podcast/ + YouTube standard). For speech-only sermon audio, -19 LUFS (Spotify + standard) is also reasonable. + + Returns a dict with measured/achieved loudness metadata so callers + can log it. + """ + src = Path(src) + out = Path(out) + + # Pass 1: measure. ffmpeg's loudnorm filter in print-only mode. + measure = subprocess.run([ + "ffmpeg", "-v", "info", + "-i", str(src), + "-af", f"loudnorm=I={target_lufs}:TP={true_peak_db}:LRA={loudness_range_lu}:print_format=json", + "-f", "null", "-", + ], capture_output=True, text=True) + + # Parse the JSON block ffmpeg prints at the end of stderr. + # Looks like: + # { + # "input_i" : "-23.81", + # "input_tp" : "-3.02", + # ... + # } + measured = _parse_loudnorm_json(measure.stderr) + + # Pass 2: apply. Reuse the measured values for linear mode (single-pass + # would do, but linear mode produces slightly better results for highly + # dynamic content). + apply = subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-i", str(src), + "-af", ( + f"loudnorm=I={target_lufs}:TP={true_peak_db}:LRA={loudness_range_lu}" + f":measured_I={measured['input_i']}" + f":measured_TP={measured['input_tp']}" + f":measured_LRA={measured['input_lra']}" + f":measured_thresh={measured['input_thresh']}" + f":offset={measured['target_offset']}" + f":linear=true:print_format=summary" + ), + "-ar", "48000", # output 48kHz (industry standard for normalized audio) + str(out), + ], capture_output=True, text=True) + + if apply.returncode != 0: + raise RuntimeError(f"normalize failed: {apply.stderr}") + + return { + "target_lufs": target_lufs, + "true_peak_db": true_peak_db, + "measured_input_i": measured.get("input_i"), + "measured_input_tp": measured.get("input_tp"), + "measured_input_lra": measured.get("input_lra"), + "applied_offset": measured.get("target_offset"), + "output_path": str(out), + } + + +_LOUDNORM_JSON_RE = re.compile(r"\{[^{}]*\"input_i\"[^{}]*\}", re.DOTALL) + + +def _parse_loudnorm_json(stderr: str) -> dict: + """Extract the loudnorm JSON block from ffmpeg stderr. + + ffmpeg prints this at the end after the audio is 'rendered' to null. + """ + m = _LOUDNORM_JSON_RE.search(stderr) + if not m: + raise RuntimeError( + f"could not find loudnorm JSON in ffmpeg output. " + f"stderr tail: {stderr[-500:]!r}" + ) + import json + return json.loads(m.group(0)) + + +# --------------------------------------------------------------------------- +# silence_stats +# --------------------------------------------------------------------------- + + +@dataclass +class SilenceStats: + """Quantitative summary of silence distribution in an audio file.""" + audio_path: str + duration_seconds: float + threshold_db: float + min_duration_seconds: float + n_silence_runs: int + total_silence_seconds: float + silence_fraction: float # total_silence / duration + mean_silence_seconds: float + median_silence_seconds: float + longest_silence_seconds: float + longest_silence_range: Tuple[float, float] # (start, end) of longest + silence_per_minute: float # mean silence runs per 60 seconds + + def to_dict(self) -> dict: + d = asdict(self) + d["longest_silence_range"] = list(self.longest_silence_range) + return d + + +def silence_stats( + src: Path, + *, + threshold_db: float = -35.0, + min_duration_seconds: float = 0.3, +) -> SilenceStats: + """Compute quantitative silence statistics for the audio. + + Useful for: + - Comparing recordings (consistent silence fraction = consistent recording setup) + - Picking the right silence threshold (dense audio with mean silence <0.5s + probably needs a lower threshold than sparse audio with mean silence >2s) + - Detecting "dead air" recordings where the speaker lost their place + """ + from .waveform import find_silence_runs, _extract_pcm + + src = Path(src) + samples = _extract_pcm(src) + duration = len(samples) / 8000.0 + if duration <= 0: + raise ValueError(f"audio {src} has zero duration") + + runs = find_silence_runs( + src, + threshold_db=threshold_db, + min_duration_seconds=min_duration_seconds, + ) + + if not runs: + return SilenceStats( + audio_path=str(src), + duration_seconds=duration, + threshold_db=threshold_db, + min_duration_seconds=min_duration_seconds, + n_silence_runs=0, + total_silence_seconds=0.0, + silence_fraction=0.0, + mean_silence_seconds=0.0, + median_silence_seconds=0.0, + longest_silence_seconds=0.0, + longest_silence_range=(0.0, 0.0), + silence_per_minute=0.0, + ) + + durations = sorted(b - a for a, b in runs) + total = sum(durations) + longest = durations[-1] + longest_range = next((a, b) for a, b in runs if b - a == longest) + + return SilenceStats( + audio_path=str(src), + duration_seconds=duration, + threshold_db=threshold_db, + min_duration_seconds=min_duration_seconds, + n_silence_runs=len(runs), + total_silence_seconds=total, + silence_fraction=total / duration, + mean_silence_seconds=total / len(runs), + median_silence_seconds=_median(durations), + longest_silence_seconds=longest, + longest_silence_range=longest_range, + silence_per_minute=len(runs) / (duration / 60.0), + ) + + +def _median(xs: List[float]) -> float: + if not xs: + return 0.0 + n = len(xs) + mid = n // 2 + if n % 2 == 0: + return (xs[mid - 1] + xs[mid]) / 2.0 + return xs[mid] + + +# --------------------------------------------------------------------------- +# threshold_tune +# --------------------------------------------------------------------------- + + +def threshold_tune( + src: Path, + *, + candidates: List[float] = (-25.0, -30.0, -35.0, -40.0, -45.0, -50.0), + target_silence_per_minute: float = 4.0, + target_min_duration: float = 0.3, +) -> dict: + """Auto-pick the silence threshold that best matches expected pause density. + + For typical sermon audio, a speaker pauses ~3-5 times per minute for + breath/sentence-end. This scans a set of candidate thresholds and + picks the one whose silence count is closest to `target_silence_per_minute`. + + Returns the picked threshold plus the per-candidate stats so callers + can show a tuning report. + """ + from .waveform import find_silence_runs + + src = Path(src) + rows = [] + for thresh in candidates: + runs = find_silence_runs( + src, + threshold_db=thresh, + min_duration_seconds=target_min_duration, + ) + # We need duration to compute silence_per_minute + from .waveform import _extract_pcm + samples = _extract_pcm(src) + duration = len(samples) / 8000.0 + spm = len(runs) / (duration / 60.0) if duration > 0 else 0.0 + rows.append({ + "threshold_db": thresh, + "n_runs": len(runs), + "silence_per_minute": round(spm, 2), + "distance_from_target": round(abs(spm - target_silence_per_minute), 2), + }) + + rows.sort(key=lambda r: r["distance_from_target"]) + best = rows[0] + + return { + "picked_threshold_db": best["threshold_db"], + "picked_silence_per_minute": best["silence_per_minute"], + "target_silence_per_minute": target_silence_per_minute, + "candidates": rows, + } diff --git a/tests/test_denoise_batch.py b/tests/test_denoise_batch.py new file mode 100644 index 0000000..726127e --- /dev/null +++ b/tests/test_denoise_batch.py @@ -0,0 +1,133 @@ +"""Tests for sermon_clean.denoise + sermon_clean.batch.""" + +from pathlib import Path + +import pytest + +from sermon_clean.denoise import denoise, DenoiseResult +from sermon_clean.batch import run_batch, _expand_globs, _make_output_path, BATCH_SUBCOMMANDS + + +def _make_wav(path: Path, *, freq: float = 440.0, duration: float = 5.0) -> Path: + import subprocess + subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-f", "lavfi", "-i", f"sine=frequency={freq}:duration={duration}", + str(path), + ], check=True) + return path + + +# --------------------------------------------------------------------------- +# denoise +# --------------------------------------------------------------------------- + + +class TestDenoise: + def test_produces_output_file(self, tmp_path): + src = _make_wav(tmp_path / "src.wav") + out = tmp_path / "out.wav" + result = denoise(src, out) + assert isinstance(result, DenoiseResult) + assert out.exists() + assert out.stat().st_size > 1000 + + def test_custom_reduction_db(self, tmp_path): + src = _make_wav(tmp_path / "src.wav") + out = tmp_path / "out.wav" + result = denoise(src, out, noise_reduction_db=20.0, noise_floor_db=-40.0) + assert result.noise_reduction_db == 20.0 + assert result.noise_floor_db == -40.0 + + +# --------------------------------------------------------------------------- +# batch helpers +# --------------------------------------------------------------------------- + + +class TestBatchHelpers: + def test_expand_globs_dedupes(self, tmp_path): + (tmp_path / "a.wav").write_bytes(b"x") + (tmp_path / "b.wav").write_bytes(b"x") + result = _expand_globs([str(tmp_path / "*.wav"), str(tmp_path / "a.wav")]) + # Should be 2 unique files, not 3 + assert len(result) == 2 + names = sorted(p.name for p in result) + assert names == ["a.wav", "b.wav"] + + def test_expand_globs_recursive(self, tmp_path): + sub = tmp_path / "sub" + sub.mkdir() + (sub / "c.wav").write_bytes(b"x") + (tmp_path / "a.wav").write_bytes(b"x") + result = _expand_globs([str(tmp_path / "**" / "*.wav")]) + assert len(result) == 2 + + def test_make_output_path_with_suffix(self, tmp_path): + src = Path("/some/where/sermon.ogg") + out = _make_output_path(src, tmp_path, "-normalized", None) + assert out == tmp_path / "sermon-normalized.ogg" + + def test_make_output_path_with_extension_override(self, tmp_path): + src = Path("/some/where/sermon.ogg") + out = _make_output_path(src, tmp_path, "-normalized", ".wav") + assert out == tmp_path / "sermon-normalized.wav" + + def test_batch_subcommands_includes_new_ones(self): + for cmd in ["normalize", "denoise", "silence-stats", "threshold-tune"]: + assert cmd in BATCH_SUBCOMMANDS + # batch itself is NOT in BATCH_SUBCOMMANDS — recursive batch would be weird. + assert "batch" not in BATCH_SUBCOMMANDS + + +# --------------------------------------------------------------------------- +# batch.run_batch end-to-end +# --------------------------------------------------------------------------- + + +class TestRunBatch: + def test_batch_normalize_two_files(self, tmp_path): + a = _make_wav(tmp_path / "sermon-a.wav", duration=3.0) + b = _make_wav(tmp_path / "sermon-b.wav", duration=3.0) + out_dir = tmp_path / "out" + + result = run_batch( + "normalize", + [a, b], + out_dir, + output_suffix="-normalized", + ) + assert result.n_ok == 2 + assert result.n_failed == 0 + assert (out_dir / "sermon-a-normalized.wav").exists() + assert (out_dir / "sermon-b-normalized.wav").exists() + + def test_batch_denoise_with_extension_override(self, tmp_path): + a = _make_wav(tmp_path / "sermon.wav", duration=3.0) + out_dir = tmp_path / "out" + + result = run_batch( + "denoise", + [a], + out_dir, + output_suffix="-dn", + output_extension=".wav", + ) + assert result.n_ok == 1 + assert (out_dir / "sermon-dn.wav").exists() + + def test_batch_unknown_subcommand_raises(self, tmp_path): + with pytest.raises(ValueError, match="unknown subcommand"): + run_batch("nonexistent", [], tmp_path) + + def test_batch_one_failure_does_not_stop_batch(self, tmp_path): + """One bad file shouldn't kill the batch — failures are collected.""" + a = _make_wav(tmp_path / "good.wav", duration=3.0) + bad = tmp_path / "does-not-exist.wav" # doesn't exist + out_dir = tmp_path / "out" + + result = run_batch("normalize", [a, bad], out_dir, output_suffix="-n") + assert result.n_ok == 1 + assert result.n_failed == 1 + assert (out_dir / "good-n.wav").exists() + assert not (out_dir / "does-not-exist-n.wav").exists() diff --git a/tests/test_processing.py b/tests/test_processing.py new file mode 100644 index 0000000..b66ef52 --- /dev/null +++ b/tests/test_processing.py @@ -0,0 +1,140 @@ +"""Tests for sermon_clean.processing — normalize, silence_stats, threshold_tune.""" + +from pathlib import Path + +import pytest + +from sermon_clean.processing import ( + normalize_loudness, + silence_stats, + threshold_tune, + SilenceStats, +) + + +def _make_wav(path: Path, *, freq: float = 440.0, duration: float = 5.0) -> Path: + """Create a synthetic WAV file for testing. Returns the path.""" + import subprocess + subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-f", "lavfi", "-i", f"sine=frequency={freq}:duration={duration}", + str(path), + ], check=True) + return path + + +def _make_silent_wav(path: Path, duration: float = 5.0) -> Path: + """Create a silent WAV file.""" + import subprocess + subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-f", "lavfi", "-i", f"anullsrc=r=8000:cl=mono", + "-t", str(duration), + str(path), + ], check=True) + return path + + +# --------------------------------------------------------------------------- +# silence_stats +# --------------------------------------------------------------------------- + + +class TestSilenceStats: + def test_silent_audio_all_silence(self, tmp_path): + audio = _make_silent_wav(tmp_path / "silent.wav", duration=5.0) + stats = silence_stats(audio, threshold_db=-30.0, min_duration_seconds=0.3) + assert isinstance(stats, SilenceStats) + assert stats.duration_seconds == pytest.approx(5.0, abs=0.5) + # A silent file should have 1 big silence run + assert stats.n_silence_runs == 1 + assert stats.silence_fraction > 0.9 + assert stats.longest_silence_seconds > 4.0 + assert stats.silence_per_minute > 10.0 # very dense silence + + def test_loud_tone_no_silences(self, tmp_path): + audio = _make_wav(tmp_path / "tone.wav", freq=440.0, duration=5.0) + stats = silence_stats(audio, threshold_db=-30.0, min_duration_seconds=0.3) + # 440Hz tone has no silences above the threshold + assert stats.n_silence_runs == 0 + assert stats.total_silence_seconds == 0.0 + assert stats.silence_fraction == 0.0 + assert stats.silence_per_minute == 0.0 + + def test_to_dict_roundtrip(self, tmp_path): + audio = _make_silent_wav(tmp_path / "silent.wav") + stats = silence_stats(audio) + d = stats.to_dict() + assert d["audio_path"] == str(audio) + assert isinstance(d["longest_silence_range"], list) + assert d["threshold_db"] == -35.0 + + +# --------------------------------------------------------------------------- +# threshold_tune +# --------------------------------------------------------------------------- + + +class TestThresholdTune: + def test_picks_threshold_closest_to_target(self, tmp_path): + # 5s silence + 1s tone alternating — moderate silence density + import subprocess + audio = tmp_path / "mixed.wav" + # 5s silence, 1s tone, 5s silence, 1s tone (12s total, 10s silence, ~83% silence) + subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-f", "lavfi", "-i", "anullsrc=r=8000:cl=mono", + "-f", "lavfi", "-i", "sine=frequency=440", + "-filter_complex", "[0:a]atrim=0:5[s1];[1:a]atrim=0:1[t1];[0:a]atrim=0:5[s2];[1:a]atrim=0:1[t2];[s1][t1][s2][t2]concat=n=4:v=0:a=1[out]", + "-map", "[out]", "-t", "12", + str(audio), + ], check=True) + + result = threshold_tune(audio, candidates=[-25.0, -35.0, -50.0], target_silence_per_minute=4.0) + assert "picked_threshold_db" in result + assert "candidates" in result + assert len(result["candidates"]) == 3 + # The picked threshold should have the smallest distance + picked = result["picked_threshold_db"] + picked_row = next(r for r in result["candidates"] if r["threshold_db"] == picked) + assert picked_row["distance_from_target"] == min(r["distance_from_target"] for r in result["candidates"]) + + def test_handles_no_silences_at_all(self, tmp_path): + # A pure tone has zero silences regardless of threshold + audio = _make_wav(tmp_path / "tone.wav", freq=440.0, duration=3.0) + result = threshold_tune(audio, candidates=[-25.0, -40.0], target_silence_per_minute=4.0) + # All candidates give 0 silence/min, all distance == 4.0. Picked is just the first. + assert result["picked_threshold_db"] in [-25.0, -40.0] + + +# --------------------------------------------------------------------------- +# normalize +# --------------------------------------------------------------------------- + + +class TestNormalize: + def test_produces_output_file(self, tmp_path): + src = _make_wav(tmp_path / "src.wav", freq=440.0, duration=5.0) + out = tmp_path / "out.wav" + result = normalize_loudness(src, out, target_lufs=-16.0) + assert out.exists() + assert out.stat().st_size > 1000 # not a silent stub + assert result["target_lufs"] == -16.0 + assert result["measured_input_i"] is not None + assert result["output_path"] == str(out) + + def test_normalized_audio_is_quieter_than_loud_input(self, tmp_path): + # Generate a LOUD sine (0.9 amplitude) — should be normalized DOWN + import subprocess + src = tmp_path / "loud.wav" + subprocess.run([ + "ffmpeg", "-y", "-v", "error", + "-f", "lavfi", "-i", "sine=frequency=440:duration=5", + "-af", "volume=0.9", + str(src), + ], check=True) + out = tmp_path / "out.wav" + result = normalize_loudness(src, out, target_lufs=-16.0) + # The applied offset should be NEGATIVE (reducing volume) for a hot input + offset = float(result["applied_offset"]) + assert offset < 0.0, f"expected negative gain offset for loud input, got {offset}"