sami7777
612c184317
- 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
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""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() |