node-26/drb-edge-node/app/internal/audio_trim.py

"""
Leading/trailing silence removal, as a slice of raw PCM.

WHY: P25 grants the channel, radios tune, and only then does a human start
talking; the recorder also deliberately over-captures at the tail (it closes a
call only after N seconds of silence have actually been HEARD). Both ends
therefore carry dead air. That is not just wasted Whisper spend: silence is a
well-documented trigger for Whisper hallucinating text that was never spoken,
and a hallucinated sentence poisons entity extraction and then incident
correlation downstream.

WHY IT IS SAFE: only the head and tail are touched, never the middle, and a
guard margin is kept around the detected speech so no syllable can be clipped.
If detection says the whole buffer is silent we do NOT emit a zero-length
recording — the caller is told and decides (see call_recorder: it skips the
upload and logs).

TIMING: trimming changes the audio's duration relative to the call's wall-clock
start/end, so every trim reports exactly how much was removed from each end.
Callers must carry those offsets forward — `started_at`/`ended_at` keep meaning
the CALL's bounds, and the trimmed audio's own bounds are reported separately.

HISTORY — THIS USED TO BE TWO FFMPEG PASSES. Detection was `silencedetect`
parsed out of FFmpeg's stderr, and the cut was a second FFmpeg re-encode. Both
are gone: the recorder now buffers PCM, so detection is arithmetic over the
samples and the cut is a byte-offset slice. Consequences worth keeping in mind:

  * The recording is encoded to MP3 exactly ONCE, after this runs, instead of
    being captured as MP3 and then re-encoded. One less generation of lossy
    encoding on every upload, and one less subprocess per call.
  * The threshold is now RMS over a short window (see pcm.rms_dbfs), where
    FFmpeg's silencedetect compared |sample| per sample. Same units (dBFS),
    slightly different meaning — do not port an old threshold across without
    re-reading the field logs.
  * There is no "is it worth re-encoding" minimum any more. A slice is free, so
    even a 0.05 s trim is applied.
"""
from dataclasses import dataclass
from typing import Optional, Tuple

from app.config import settings
from app.internal import pcm
from app.internal.logger import logger

# Window the head/tail scan works in. 20 ms is short enough that the guard
# margin below dwarfs the quantisation error, and long enough that RMS means
# something.
ANALYSIS_WINDOW_SECONDS = 0.02

# How far in from each end the scan is willing to look before giving up.
#
# Bounds the only unbounded cost in this module: the per-sample RMS loop. A
# normal recording resolves within a window or two at the head (the recorder
# starts on voice onset) and within the silence run at the tail, so this cap is
# never reached in practice. If it IS reached, we leave the audio untrimmed and
# say so — shipping an untrimmed recording is always better than shipping none.
MAX_SCAN_SECONDS = 30.0


@dataclass(frozen=True)
class TrimResult:
    """Outcome of a trim attempt. `lead`/`tail` are seconds actually removed."""

    lead: float = 0.0
    tail: float = 0.0
    duration_before: float = 0.0
    duration_after: float = 0.0
    all_silence: bool = False
    applied: bool = False
    # True when the scan hit MAX_SCAN_SECONDS without finding speech, so
    # `all_silence` could not be determined and nothing was trimmed.
    scan_truncated: bool = False

    @property
    def trimmed_seconds(self) -> float:
        return self.lead + self.tail


def _window_bytes() -> int:
    return max(pcm.FRAME_BYTES, pcm.byte_offset(ANALYSIS_WINDOW_SECONDS))


def first_signal_offset(
    audio: bytes,
    threshold_db: float,
    limit_seconds: float = MAX_SCAN_SECONDS,
) -> Optional[int]:
    """
    Byte offset of the first window carrying signal, scanning forward.

    None means "no signal found" — either the buffer really is all silence or
    the scan hit `limit_seconds` first; the caller distinguishes the two by
    comparing the scanned span against the buffer length.
    """
    window = _window_bytes()
    limit = min(len(audio), pcm.byte_offset(limit_seconds) or len(audio))
    offset = 0
    while offset < limit:
        chunk = audio[offset:offset + window]
        if not pcm.is_silent(chunk, threshold_db):
            return offset
        offset += window
    return None


def last_signal_offset(
    audio: bytes,
    threshold_db: float,
    limit_seconds: float = MAX_SCAN_SECONDS,
) -> Optional[int]:
    """
    Byte offset of the END of the last window carrying signal, scanning back.

    Returns the offset one past the last signal-bearing window, so it can be
    used directly as a slice bound.
    """
    window = _window_bytes()
    total = pcm.align(len(audio))
    floor = max(0, total - (pcm.byte_offset(limit_seconds) or total))
    offset = total
    while offset > floor:
        start = max(floor, offset - window)
        if not pcm.is_silent(audio[start:offset], threshold_db):
            return offset
        offset = start
    return None


def keep_window(
    first_signal: Optional[int],
    last_signal: Optional[int],
    total_bytes: int,
    guard_bytes: int,
) -> Tuple[int, int]:
    """
    Turn detected signal bounds into the byte range to keep.

    Pure and side-effect free so the decision that can destroy a transmission
    stays unit-testable without any audio. Offsets are sample-aligned and
    clamped to the buffer.
    """
    total = pcm.align(total_bytes)
    start = 0 if first_signal is None else max(0, first_signal - guard_bytes)
    end = total if last_signal is None else min(total, last_signal + guard_bytes)
    start = pcm.align(start)
    end = pcm.align(end)
    if end <= start:
        return 0, total
    return start, end


def trim_pcm(
    audio: bytes,
    threshold_db: Optional[float] = None,
    guard: Optional[float] = None,
) -> Tuple[bytes, TrimResult]:
    """
    Return (kept_audio, result). Never raises and never returns empty audio.

    An all-silence buffer is returned UNCHANGED with `all_silence=True`: the
    caller decides what to do with a recording that contains no speech at all —
    that is itself a signal (squelch misconfigured, wrong sink, dead audio
    path), not something to silently truncate to nothing.
    """
    threshold = settings.trim_silence_threshold_db if threshold_db is None else threshold_db
    margin = settings.trim_silence_guard_seconds if guard is None else guard

    total = pcm.align(len(audio))
    duration = pcm.seconds(total)
    if total <= 0:
        return audio, TrimResult()

    first = first_signal_offset(audio, threshold)
    if first is None:
        scanned = min(total, pcm.byte_offset(MAX_SCAN_SECONDS) or total)
        if scanned < total:
            # Could not prove it is all silence; refuse to guess.
            logger.warning(
                f"Silence scan gave up after {MAX_SCAN_SECONDS:.0f}s without finding speech in a "
                f"{duration:.1f}s recording — leaving it untrimmed."
            )
            return audio, TrimResult(
                duration_before=duration, duration_after=duration, scan_truncated=True
            )
        logger.warning(
            f"Recording is entirely silence ({duration:.2f}s, threshold {threshold:.1f}dBFS RMS) — "
            "no speech detected."
        )
        return audio, TrimResult(duration_before=duration, duration_after=duration, all_silence=True)

    last = last_signal_offset(audio, threshold)
    guard_bytes = pcm.byte_offset(margin)
    keep_start, keep_end = keep_window(first, last, total, guard_bytes)

    lead = pcm.seconds(keep_start)
    tail = pcm.seconds(total - keep_end)
    if keep_start <= 0 and keep_end >= total:
        return audio[:total], TrimResult(duration_before=duration, duration_after=duration)

    kept = audio[keep_start:keep_end]
    after = pcm.seconds(len(kept))
    logger.info(
        f"Trimmed recording: -{lead:.2f}s lead, -{tail:.2f}s tail "
        f"({duration:.2f}s -> {after:.2f}s, threshold {threshold:.1f}dBFS RMS)"
    )
    return kept, TrimResult(
        lead=lead,
        tail=tail,
        duration_before=duration,
        duration_after=after,
        applied=True,
    )