sermon-clean/sermon_clean/speech_regions.py

"""Extract only the speech regions of an audio file (skip long silences).

This makes faster-whisper ~3-5x faster on sermon-style audio because it
doesn't have to decode / transcribe silent gaps.

Usage:
    regions = extract_speech_regions(audio_path, min_silence=0.5)
    # regions = [(0.0, 6.95), (7.86, 8.44), (8.45, 9.01), ...]
    # Each tuple is (start_seconds, end_seconds) of a speech region.

    # Write the merged speech-only file
    write_speech_only(audio_path, regions, output_path)

The ffmpeg silencedetect filter emits very granular silence events
(many <1s silences between words). We merge silences within `merge_gap`
seconds to avoid cutting speech fragments apart.
"""

from __future__ import annotations

import re
import subprocess
from pathlib import Path
from typing import List, Tuple


def find_speech_regions(
    src: Path,
    *,
    noise_db: float = -30.0,
    min_silence: float = 0.5,
    merge_gap: float = 0.3,
) -> List[Tuple[float, float]]:
    """Run ffmpeg silencedetect, then return the SPEECH regions (gaps between silences).

    Adjacent silences within `merge_gap` seconds are merged before computing regions,
    so we don't end up with fragmented speech chunks.
    """
    src = Path(src)
    out = subprocess.run([
        "ffmpeg", "-v", "info",
        "-i", str(src),
        "-af", f"silencedetect=noise={noise_db}dB:d={min_silence}",
        "-f", "null", "-",
    ], capture_output=True, text=True)
    pattern = re.compile(r"silence_(start|end):\s*([0-9.]+)")
    events = []
    for line in out.stderr.splitlines():
        for m in pattern.finditer(line):
            kind = m.group(1)
            val = float(m.group(2))
            events.append((kind, val))
    # Pair up start/end events
    silences: List[Tuple[float, float]] = []
    pending_start = None
    for kind, val in events:
        if kind == "start":
            pending_start = val
        elif kind == "end" and pending_start is not None:
            silences.append((pending_start, val))
            pending_start = None
    # If audio ends in silence, no final end event — drop the dangling start
    if pending_start is not None:
        silences.append((pending_start, float("inf")))
        silences[-1] = (silences[-1][0], float(out_duration_estimate(src)))

    # Merge silences that are close together
    if not silences:
        # Audio is all speech — single region
        return [(0.0, float(out_duration_estimate(src)))]
    merged: List[Tuple[float, float]] = [silences[0]]
    for s, e in silences[1:]:
        last_s, last_e = merged[-1]
        if s - last_e <= merge_gap:
            merged[-1] = (last_s, max(last_e, e))
        else:
            merged.append((s, e))

    # Convert silence regions to speech regions (the gaps)
    total_duration = float(out_duration_estimate(src))
    speech: List[Tuple[float, float]] = []
    cursor = 0.0
    for s_start, s_end in merged:
        if s_start > cursor:
            speech.append((cursor, s_start))
        cursor = max(cursor, s_end)
    if cursor < total_duration:
        speech.append((cursor, total_duration))
    return speech


def out_duration_estimate(src: Path) -> float:
    """Get the duration of an audio file via ffprobe."""
    out = subprocess.run([
        "ffprobe", "-v", "error",
        "-show_entries", "format=duration",
        "-of", "default=noprint_wrappers=1:nokey=1",
        str(src),
    ], capture_output=True, text=True)
    try:
        return float(out.stdout.strip())
    except ValueError:
        return 0.0


def write_speech_only(
    src: Path,
    regions: List[Tuple[float, float]],
    output: Path,
) -> Path:
    """Write a new audio file containing only the speech regions, concatenated.

    Uses the ffmpeg concat demuxer with explicit per-region trim commands.
    Output preserves the source codec (libopus for OGG/Opus, libmp3lame for mp3).
    """
    src = Path(src)
    output = Path(output)
    output.parent.mkdir(parents=True, exist_ok=True)
    if not regions:
        # Nothing to write — just copy a tiny silence file
        subprocess.run([
            "ffmpeg", "-y", "-v", "error",
            "-f", "lavfi", "-i", "anullsrc=r=8000:cl=mono",
            "-t", "0.1",
            str(output),
        ], check=True)
        return output

    # Build a temporary concat list of per-region trims
    # We use the filter_complex approach with concat filter for accuracy
    # Each input: ffmpeg -ss start -to end -i src; concat them all
    if len(regions) == 1:
        s, e = regions[0]
        subprocess.run([
            "ffmpeg", "-y", "-v", "error",
            "-ss", f"{s:.3f}", "-to", f"{e:.3f}",
            "-i", str(src),
            "-c", "copy",
            str(output),
        ], check=True)
        return output

    # Multi-region: use filter_complex with concat
    inputs = []
    filter_parts = []
    for i, (s, e) in enumerate(regions):
        inputs.extend(["-ss", f"{s:.3f}", "-to", f"{e:.3f}", "-i", str(src)])
        filter_parts.append(f"[{i}:a]atrim=0:{e-s:.3f},asetpts=PTS-STARTPTS[a{i}]")
    concat_inputs = "".join(f"[a{i}]" for i in range(len(regions)))
    filter_parts.append(f"{concat_inputs}concat=n={len(regions)}:v=0:a=1[out]")
    filter_complex = ";".join(filter_parts)
    cmd = ["ffmpeg", "-y", "-v", "error"] + inputs + [
        "-filter_complex", filter_complex,
        "-map", "[out]",
        "-c:a", "libmp3lame",  # always re-encode to mp3 for the speech-only file
        "-ar", "16000", "-ac", "1",
        str(output),
    ]
    subprocess.run(cmd, check=True)
    return output


def write_speech_only_via_silenceremove(src: Path, output: Path) -> Path:
    """Alternative: use ffmpeg's silenceremove filter directly.

    Faster than per-region trim for files with many silences — single decode pass.
    """
    output = Path(output)
    subprocess.run([
        "ffmpeg", "-y", "-v", "error",
        "-i", str(src),
        "-af", (
            "silenceremove=stop_periods=-1:"
            "stop_duration=0.5:stop_threshold=-30dB,"
            "asetpts=N/SR/TB"
        ),
        "-ar", "16000", "-ac", "1",
        "-c:a", "libmp3lame",
        str(output),
    ], check=True)
    return output