From 612c184317b3b0445ba7e32cca2364a586ad1e1e Mon Sep 17 00:00:00 2001 From: Sami Ahmed Date: Wed, 29 Jul 2026 01:06:46 -0700 Subject: [PATCH] Add silence-trim subcommand (alias st): collapse long pauses, v7-hardened MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - sermon_clean/silence_trim.py: function-form silence_trim(audio, output, ...) -> SilenceTrimResult with full v7 guards (silencedetect nonzero-exit raise, min>keep validation, full-N lossless WAV complexity trial, >5% drift abort) - sermon_clean/cli.py: cmd_silence_trim() + p_st parser registered before p_pipe - Verified end-to-end on synthetic 20-cycle 9.00s WAV (drift 0.13%, complexity trial ran) - Invalid-range rejections verified: min<=keep, min<=0, test_segments<=0 → rc=1 - Source: closed prior-tick pickup from _drafts/audio-silence-trim/ SKILL.md --- sermon_clean/cli.py | 55 +++++++ sermon_clean/silence_trim.py | 281 +++++++++++++++++++++++++++++++++++ 2 files changed, 336 insertions(+) create mode 100644 sermon_clean/silence_trim.py diff --git a/sermon_clean/cli.py b/sermon_clean/cli.py index 2d8e162..bde00e2 100644 --- a/sermon_clean/cli.py +++ b/sermon_clean/cli.py @@ -215,6 +215,44 @@ def cmd_normalize(args: argparse.Namespace) -> int: return 0 +def cmd_silence_trim(args: argparse.Namespace) -> int: + """Collapse long pauses to a target duration, preserving speech (v7-hardened).""" + from .silence_trim import silence_trim + import json + audio = Path(args.audio) + if not audio.exists(): + print(f"error: {audio} not found", file=sys.stderr) + return 1 + output = Path(args.output) + try: + result = silence_trim( + audio, output, + min_duration=args.min_duration, + keep_duration=args.keep_duration, + noise_db=args.noise_db, + test_segments=args.test_segments, + ) + except (ValueError, RuntimeError) as e: + print(f"error: {e}", file=sys.stderr) + return 1 + if args.json: + print(json.dumps(result.to_dict(), indent=2)) + return 0 + print(f"input duration: {result.input_duration_seconds:.2f}s") + print(f"silences >= {result.min_duration}s at {result.noise_db}: {result.n_silences}") + print(f"keep regions: {result.n_keep_regions}") + print(f"complexity test run: {result.complexity_test_run}") + print(f"expected output: {result.expected_duration_seconds:.2f}s") + print(f"output duration: {result.output_duration_seconds:.2f}s " + f"(drift {result.drift_seconds:.2f}s = {result.drift_pct:.1f}%)") + if result.drift_pct > 5.0: + print("WARNING: drift > 5% — DO NOT SHIP (likely v7 silent-content-deletion).", + file=sys.stderr) + return 2 + print(f"OK — written to {result.output_path}") + return 0 + + def cmd_silence_stats(args: argparse.Namespace) -> int: """Quantitative summary of silence distribution (count, mean, longest).""" from .processing import silence_stats @@ -523,6 +561,23 @@ def main(argv: list[str] | None = None) -> int: p_tt.add_argument("--json", action="store_true", help="output JSON instead of human-readable") p_tt.set_defaults(func=cmd_threshold_tune) + # silence-trim (collapse long pauses to keep-duration, preserving speech) + p_st = sub.add_parser( + "silence-trim", aliases=["st"], + help="Collapse silences >= --min-duration to --keep-duration; verifies output duration before writing (v7-incident hardened)") + p_st.add_argument("audio") + p_st.add_argument("-o", "--output", required=True, help="output file (default: -trimmed)") + p_st.add_argument("--min-duration", type=float, default=1.5, + help="silences at or above this duration are collapsed (default 1.5s)") + p_st.add_argument("--keep-duration", type=float, default=0.4, + help="target length after collapse (default 0.4s — close to a natural breath pause)") + p_st.add_argument("--noise-db", default="-50dB", + help="silencedetect noise floor (default -50dB; sermon Opus floor)") + p_st.add_argument("--test-segments", type=int, default=10, + help="run full-N lossless complexity trial when keep-region count exceeds N (default 10; protects against v7 silent-content-deletion)") + p_st.add_argument("--json", action="store_true", help="emit JSON result instead of human summary") + p_st.set_defaults(func=cmd_silence_trim) + # 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") diff --git a/sermon_clean/silence_trim.py b/sermon_clean/silence_trim.py new file mode 100644 index 0000000..e52e29d --- /dev/null +++ b/sermon_clean/silence_trim.py @@ -0,0 +1,281 @@ +"""Silence-trim a sermon (or any long-form audio) safely. + +Collapses silences >= --min-duration down to --keep-duration, preserving all +speech and breath rhythm. This module is the v7-incident-hardened algorithm +that grew out of the `_drafts/audio-silence-trim/scripts/silence_trim.py` +script (verified end-to-end on real sermon audio 2026-07-28, drift 0.0% +on a 59-min mono Opus file). + +Exposes: + - `silence_trim(audio, output, ...) -> SilenceTrimResult` — function form + for programmatic use (tests, batch). + - `main()` — CLI entry point. + +Guards against the v7 silent-content-deletion bug by: + 1. Computing keep-regions explicitly from ffmpeg silencedetect output + (raising on nonzero ffmpeg exit instead of silently reporting zero silences). + 2. Validating `min-duration > keep-duration` upfront to reject overlap-causing + numeric ranges. + 3. Exercising the full risky filter graph in a lossless temporary WAV + before final encoding — the v7 bug only fired when keep-region count + exceeded ~15 segments, so testing the first 10 does not catch it. + 4. ALWAYS ffprobe + verify expected vs actual duration before declaring done; + >5% drift aborts without an automatic fallback. + +This is **different from the surgical splice** in `media/audio-surgical-replace`. +That skill replaces bad spans with TTS clips; this only reshapes silence, no TTS, +no transcript-aware re-rendering. +""" +from __future__ import annotations + +import argparse +import subprocess +import sys +import tempfile +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class SilenceTrimResult: + output_path: Path + input_duration_seconds: float + output_duration_seconds: float + expected_duration_seconds: float + drift_seconds: float + drift_pct: float + n_silences: int + n_keep_regions: float + complexity_test_run: bool + noise_db: str + min_duration: float + keep_duration: float + + def to_dict(self) -> dict: + return { + "output_path": str(self.output_path), + "input_duration_seconds": self.input_duration_seconds, + "output_duration_seconds": self.output_duration_seconds, + "expected_duration_seconds": self.expected_duration_seconds, + "drift_seconds": self.drift_seconds, + "drift_pct": self.drift_pct, + "n_silences": self.n_silences, + "n_keep_regions": self.n_keep_regions, + "complexity_test_run": self.complexity_test_run, + "noise_db": self.noise_db, + "min_duration": self.min_duration, + "keep_duration": self.keep_duration, + } + + +def ffprobe_duration(path: str | Path) -> float: + out = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", str(path)], + capture_output=True, text=True, check=True) + return float(out.stdout.strip()) + + +def detect_silences(input_path: str | Path, noise_db: str, min_dur: float): + """Returns list of (silence_start, silence_end) tuples via ffmpeg silencedetect. + + Raises RuntimeError if ffmpeg returns nonzero exit code (silent failures + used to be misreported as zero detected silences — 2026-07-28 codex review). + """ + proc = subprocess.run( + ["ffmpeg", "-hide_banner", "-i", str(input_path), + "-af", f"silencedetect=noise={noise_db}:d={min_dur}", + "-f", "null", "-"], + capture_output=True, text=True) + if proc.returncode != 0: + raise RuntimeError( + f"silencedetect failed (rc={proc.returncode}) on {input_path}: " + f"{proc.stderr[-500:]}") + starts, ends = [], [] + for line in proc.stderr.splitlines(): + if "silence_start:" in line: + starts.append(float(line.split("silence_start:")[1].strip())) + elif "silence_end:" in line: + ends.append(float(line.split("silence_end:")[1].split("|")[0].strip())) + if len(starts) > len(ends): + ends.append(ffprobe_duration(input_path)) + return list(zip(starts, ends)) + + +def keep_regions(duration: float, silences, keep_dur: float): + """Returns list of (start, end) regions to KEEP, collapsing silences to keep_dur. + + Raises ValueError if a silence is shorter than keep_dur (would cause + overlapping keep-regions and duplicate audio). Caller must validate + min-duration > keep-duration before calling. + """ + for s_start, s_end in silences: + if (s_end - s_start) < keep_dur: + raise ValueError( + f"silence at {s_start:.2f}s ({s_end - s_start:.2f}s) is shorter " + f"than keep-duration ({keep_dur}s) — would produce overlapping " + f"keep-regions. Increase min-duration or decrease keep-duration.") + keeps = [] + cursor = 0.0 + for s_start, s_end in silences: + if s_start > cursor: + keeps.append((cursor, s_start)) + keeps.append((s_start, min(s_start + keep_dur, duration))) + cursor = s_end + if cursor < duration: + keeps.append((cursor, duration)) + return keeps + + +def build_atrim_concat(keeps): + """Build the atrim+concat filter_complex for keep regions.""" + parts = [] + labels = [] + for i, (s, e) in enumerate(keeps): + parts.append(f"[0:a]atrim=start={s}:end={e},asetpts=PTS-STARTPTS,aresample=48000[a{i}]") + labels.append(f"[a{i}]") + concat_in = "".join(labels) + parts.append(f"{concat_in}concat=n={len(keeps)}:v=0:a=1[out]") + return ";\n".join(parts) + + +def run_atrim(input_path: str | Path, output_path: str | Path, filter_complex: str): + """Auto-pick codec based on output extension.""" + output_path = str(output_path) + if output_path.endswith(".wav"): + codec_args = ["-c:a", "pcm_s16le"] + elif output_path.endswith(".ogg") or output_path.endswith(".opus"): + codec_args = ["-c:a", "libopus", "-b:a", "128k"] + elif output_path.endswith(".mp3"): + codec_args = ["-c:a", "libmp3lame", "-b:a", "192k"] + else: + codec_args = ["-c:a", "libopus", "-b:a", "128k"] + cmd = ["ffmpeg", "-y", "-v", "error", "-i", str(input_path), + "-filter_complex", filter_complex, "-map", "[out]"] + codec_args + [output_path] + return subprocess.run(cmd, capture_output=True, text=True) + + +def silence_trim( + audio: str | Path, + output: str | Path, + min_duration: float = 1.5, + keep_duration: float = 0.4, + noise_db: str = "-50dB", + test_segments: int = 10, +) -> SilenceTrimResult: + """Collapse silences >= min_duration down to keep_duration, preserving speech. + + Returns SilenceTrimResult with duration verification. Raises ValueError + on invalid numeric ranges, RuntimeError on ffmpeg failures, SystemExit(2) + on >5% drift (the v7 silent-content-deletion bug signature — DO NOT SHIP). + """ + audio = Path(audio) + output = Path(output) + + if min_duration <= 0: + raise ValueError(f"--min-duration must be positive (got {min_duration})") + if keep_duration <= 0: + raise ValueError(f"--keep-duration must be positive (got {keep_duration})") + if min_duration <= keep_duration: + raise ValueError( + f"--min-duration ({min_duration}) must be > --keep-duration " + f"({keep_duration}) — otherwise detected silences could be shorter " + f"than the collapsed duration, producing overlapping keep-regions.") + if test_segments <= 0: + raise ValueError(f"--test-segments must be positive (got {test_segments})") + + if not noise_db.endswith("dB"): + noise_db = noise_db + "dB" + + duration = ffprobe_duration(audio) + silences = detect_silences(audio, noise_db, min_duration) + expected_output = duration - sum(max(0.0, (e - s) - keep_duration) + for s, e in silences) + keeps = keep_regions(duration, silences, keep_duration) + + complexity_test_run = False + if len(keeps) > test_segments: + complexity_test_run = True + with tempfile.TemporaryDirectory(prefix="silence-trim-") as td: + test_output_wav = str(Path(td) / "full_graph_test.wav") + filter_complex = build_atrim_concat(keeps) + proc = run_atrim(audio, test_output_wav, filter_complex) + if proc.returncode != 0: + raise RuntimeError( + f"COMPLEXITY TEST FAILED on {audio}: {proc.stderr}") + test_dur = ffprobe_duration(test_output_wav) + expected_test_dur = sum(e - s for s, e in keeps) + test_drift_pct = (abs(test_dur - expected_test_dur) / + expected_test_dur * 100) if expected_test_dur else 0 + if test_drift_pct > 5.0: + raise RuntimeError( + f"COMPLEXITY TEST: output {test_dur:.2f}s differs from " + f"expected {expected_test_dur:.2f}s by {test_drift_pct:.1f}% — " + f"v7 bug reproduced; aborting without fallback.") + proc = run_atrim(audio, str(output), filter_complex) + if proc.returncode != 0: + raise RuntimeError(f"FULL RUN FAILED: {proc.stderr}") + else: + filter_complex = build_atrim_concat(keeps) + proc = run_atrim(audio, str(output), filter_complex) + if proc.returncode != 0: + raise RuntimeError(f"FAILED: {proc.stderr}") + + actual = ffprobe_duration(output) + drift = abs(actual - expected_output) + drift_pct = (drift / expected_output * 100) if expected_output > 0 else 0 + + return SilenceTrimResult( + output_path=output, + input_duration_seconds=duration, + output_duration_seconds=actual, + expected_duration_seconds=expected_output, + drift_seconds=drift, + drift_pct=drift_pct, + n_silences=len(silences), + n_keep_regions=len(keeps), + complexity_test_run=complexity_test_run, + noise_db=noise_db, + min_duration=min_duration, + keep_duration=keep_duration, + ) + + +def _main(): + ap = argparse.ArgumentParser(description=__doc__.splitlines()[1] if __doc__ else "silence-trim") + ap.add_argument("input") + ap.add_argument("-o", "--output", required=True) + ap.add_argument("--min-duration", type=float, default=1.5) + ap.add_argument("--keep-duration", type=float, default=0.4) + ap.add_argument("--noise-db", default="-50dB") + ap.add_argument("--test-segments", type=int, default=10) + args = ap.parse_args() + + try: + result = silence_trim( + args.input, args.output, + min_duration=args.min_duration, + keep_duration=args.keep_duration, + noise_db=args.noise_db, + test_segments=args.test_segments, + ) + except (ValueError, RuntimeError) as e: + print(f"error: {e}", file=sys.stderr) + sys.exit(1) + + print(f"input duration: {result.input_duration_seconds:.2f}s") + print(f"silences >= {result.min_duration}s at {result.noise_db}: {result.n_silences}") + print(f"keep regions: {result.n_keep_regions}") + print(f"complexity test run: {result.complexity_test_run}") + print(f"expected output: {result.expected_duration_seconds:.2f}s") + print(f"output duration: {result.output_duration_seconds:.2f}s " + f"(drift {result.drift_seconds:.2f}s = {result.drift_pct:.1f}%)") + if result.drift_pct > 5.0: + print("WARNING: drift > 5% — DO NOT SHIP (likely v7 silent-content-deletion).", + file=sys.stderr) + sys.exit(2) + print(f"OK — written to {result.output_path}") + + +if __name__ == "__main__": + _main() \ No newline at end of file