diff --git a/.env.example b/.env.example index 196c057..0ff5bca 100644 --- a/.env.example +++ b/.env.example @@ -31,8 +31,27 @@ PULSE_WAIT_TIMEOUT=30 # Call segmentation: seconds of radio silence before the current recording is # closed. Grants on the same talkgroup within this window stay in ONE recording. +# Tune ONLY from the "measured control-channel idle" line the edge node logs on +# every idle-timeout close — silence measured in the audio is a different clock +# (it also contains the ~1.9s P25 grant-to-speech delay). CALL_IDLE_TIMEOUT=3 +# Seconds of audio kept after the last transmission ends. This is the only +# headroom protecting the final word of a transmission — usually the disposition +# or the address. Measured at 0.5s it left ~0.3s of real margin and one recording +# ended mid-word, hence 1.0. +CALL_TAIL_PAD_SECONDS=1.0 + +# Strip leading/trailing dead air before upload. ~63% of an untrimmed recording +# is silence, which costs Whisper spend and makes it hallucinate text that was +# never spoken. Only the head and tail are touched, with a guard margin so no +# syllable is clipped. Set to false to upload raw audio. +TRIM_SILENCE=true +# dBFS below which audio counts as silence for detection. +TRIM_SILENCE_THRESHOLD_DB=-40 +# Seconds of audio kept either side of detected speech. +TRIM_SILENCE_GUARD_SECONDS=0.25 + # OP25 container (usually no need to change) OP25_API_URL=http://localhost:8001 OP25_TERMINAL_URL=http://localhost:8081 diff --git a/drb-edge-node/app/config.py b/drb-edge-node/app/config.py index 4c5a1a9..a550dfc 100644 --- a/drb-edge-node/app/config.py +++ b/drb-edge-node/app/config.py @@ -37,8 +37,31 @@ class Settings(BaseSettings): # Call segmentation — seconds with no active transmission before the current # recording is closed out. Consecutive grants on the SAME talkgroup inside this # window are kept in one recording so back-and-forth traffic stays together. + # + # Do NOT tune this against measured *audio* silence: audio gaps also contain + # the ~1.9 s P25 grant→speech delay, so they are always longer than the + # control-channel idle this timer measures. metadata_watcher logs the real + # measured idle on every idle-timeout close — tune from that. call_idle_timeout: float = 3.0 + # Audio kept after the observed end of the last transmission. The srcaddr + # 1→0 edge can be up to one poll (0.5 s) late and the encoder adds its own + # latency, so this is the only headroom protecting the last word of a + # transmission — which is usually the disposition or the address. Field + # measurement at 0.5 s left only 0.29–0.37 s of real trailing margin and one + # recording ended mid-word, hence 1.0 s. + call_tail_pad_seconds: float = 1.0 + + # Strip leading/trailing dead air before upload. ~63% of a typical recording + # is silence (the grant→speech delay plus the tail pad), which inflates + # Whisper cost and is a well-documented trigger for hallucinated transcript + # text. Trimming is conservative — see internal/audio_trim.py. + trim_silence: bool = True + # Anything quieter than this counts as silence for detection purposes. + trim_silence_threshold_db: float = -40.0 + # Guard margin kept around detected speech so no syllable is clipped. + trim_silence_guard_seconds: float = 0.25 + # OP25 container op25_api_url: str = "http://localhost:8001" op25_terminal_url: str = "http://localhost:8081" diff --git a/drb-edge-node/app/internal/audio_trim.py b/drb-edge-node/app/internal/audio_trim.py new file mode 100644 index 0000000..840da6c --- /dev/null +++ b/drb-edge-node/app/internal/audio_trim.py @@ -0,0 +1,270 @@ +""" +Conservative leading/trailing silence removal for finished call recordings. + +WHY: P25 grants the channel, radios tune, and only then does a human start +talking. Measured on six real recordings from a live node, that costs 1.71–2.45 s +of dead air at the head of every single file, and the tail pad adds its own +~0.3–1.0 s. Roughly 63% of every uploaded MP3 was silence. 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 file is silent we do NOT emit a zero-length file — +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. + +Implementation is two FFmpeg passes (detect, then cut). FFmpeg is already a hard +dependency of this container and is already running the capture, so this adds no +new moving parts. `silenceremove` in one pass was rejected on purpose: it gives +no way to learn how much it removed, which would make the timing metadata above +impossible to produce. +""" +import asyncio +import re +from dataclasses import dataclass +from pathlib import Path +from typing import List, Optional, Tuple + +from app.config import settings +from app.internal.logger import logger + +# Minimum run of quiet before FFmpeg calls it a silence region. Below ~0.3 s this +# starts firing on the natural pauses between words, which is not what we want — +# we only care about the big block of dead air at each end. +MIN_SILENCE_SECONDS = 0.3 + +# A silence region only counts as "leading" if it begins essentially at the file +# head. One chunk of slop. +HEAD_EPSILON_SECONDS = 0.15 + +# ...and only as "trailing" if it reaches the end of the audio. Do NOT assume a +# missing `silence_end` marks that case: FFmpeg 6.x flushes a closing +# `silence_end` at EOF, so an all-silence file looks exactly like a file with one +# closed silence region. Verified against ffmpeg 6.1.1 — the reported end lands +# ~0.03 s short of the (offset-corrected) duration, hence this epsilon. The +# residual risk is clipping <=0.1 s of audio that follows a >=0.3 s gap right at +# EOF, which the guard margin below more than covers. +TAIL_EPSILON_SECONDS = 0.10 + +# Don't bother re-encoding to reclaim less than this — a re-encode costs a CPU +# spike on a Pi and a generation of MP3 quality, which is a bad trade for +# a fraction of a second. +MIN_TRIM_SECONDS = 0.20 + +# Bounded so a wedged FFmpeg can never stall the upload path. +FFMPEG_TIMEOUT_SECONDS = 30.0 + +_SILENCE_START_RE = re.compile(r"silence_start:\s*(-?[0-9.]+)") +_SILENCE_END_RE = re.compile(r"silence_end:\s*(-?[0-9.]+)") +_DURATION_RE = re.compile(r"Duration:\s*(\d+):(\d\d):(\d\d\.\d+)") +_START_RE = re.compile(r"Duration:[^\n]*?start:\s*(-?[0-9.]+)") + + +@dataclass(frozen=True) +class TrimResult: + """Outcome of a trim attempt. `lead`/`tail` are seconds actually removed.""" + + path: Optional[Path] + 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 + + @property + def trimmed_seconds(self) -> float: + return self.lead + self.tail + + +async def _run(cmd: List[str]) -> Tuple[int, str]: + """Run FFmpeg and return (returncode, stderr). FFmpeg reports on stderr.""" + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.DEVNULL, + stderr=asyncio.subprocess.PIPE, + ) + try: + _, stderr = await asyncio.wait_for(proc.communicate(), timeout=FFMPEG_TIMEOUT_SECONDS) + except asyncio.TimeoutError: + try: + proc.kill() + except Exception: + pass + raise + return proc.returncode or 0, stderr.decode(errors="replace") + + +def _parse_duration(stderr: str) -> Optional[float]: + """ + Length of the decodable audio, in silencedetect's coordinates. + + MP3 carries encoder delay/padding, which FFmpeg reports as a container + `start:` offset — the container duration is that much longer than the audio + silencedetect actually timestamps. Subtracting it is what lets + TAIL_EPSILON_SECONDS stay tight enough to be safe. + """ + match = _DURATION_RE.search(stderr) + if not match: + return None + hours, minutes, seconds = match.groups() + duration = int(hours) * 3600 + int(minutes) * 60 + float(seconds) + + start = _START_RE.search(stderr) + if start: + offset = float(start.group(1)) + if 0.0 <= offset < duration: + duration -= offset + return duration + + +def _parse_silences(stderr: str) -> List[Tuple[float, Optional[float]]]: + """ + Extract silence regions as (start, end) with end=None when it runs to EOF. + + FFmpeg emits `silence_start:` and a later `silence_end:` per region, and + simply never emits the closing line for a region that reaches EOF. + """ + regions: List[Tuple[float, Optional[float]]] = [] + pending: Optional[float] = None + for line in stderr.splitlines(): + if "silencedetect" not in line: + continue + start = _SILENCE_START_RE.search(line) + if start: + pending = float(start.group(1)) + continue + end = _SILENCE_END_RE.search(line) + if end and pending is not None: + regions.append((pending, float(end.group(1)))) + pending = None + if pending is not None: + regions.append((pending, None)) + return regions + + +def speech_bounds( + regions: List[Tuple[float, Optional[float]]], + duration: float, + guard: float, +) -> Tuple[float, float, bool]: + """ + Turn detected silence regions into the [start, end] window to keep. + + Pure and side-effect free so the decision logic is unit-testable without + FFmpeg. Returns (start, end, all_silence). + """ + speech_start = 0.0 + speech_end = duration + + if regions: + head_start, head_end = regions[0] + if head_start <= HEAD_EPSILON_SECONDS and head_end is not None: + speech_start = head_end + + tail_start, tail_end = regions[-1] + # `None` = pre-6.x FFmpeg, which simply stopped reporting at EOF. + reaches_eof = tail_end is None or (duration - tail_end) <= TAIL_EPSILON_SECONDS + if reaches_eof: + speech_end = min(speech_end, tail_start) + + if speech_end <= speech_start: + # Detection says there is no speech anywhere in the file. + return 0.0, duration, True + + # Guard margin: never trim right up against the first/last syllable. + keep_start = max(0.0, speech_start - guard) + keep_end = min(duration, speech_end + guard) + return keep_start, keep_end, False + + +async def trim_silence( + path: Path, + sample_rate: str, + bitrate: str, + threshold_db: Optional[float] = None, + guard: Optional[float] = None, +) -> TrimResult: + """ + Trim leading/trailing silence in place, preserving the original on failure. + + Never raises: any problem degrades to "leave the file exactly as it was", + because shipping an untrimmed recording is far better than shipping none. + """ + 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 + + try: + _, stderr = await _run([ + "ffmpeg", "-hide_banner", "-nostdin", "-nostats", + "-i", str(path), + "-af", f"silencedetect=noise={threshold}dB:d={MIN_SILENCE_SECONDS}", + "-f", "null", "-", + ]) + except Exception as e: + logger.warning(f"Silence detection failed for {path.name} ({e}) — uploading untrimmed.") + return TrimResult(path=path) + + duration = _parse_duration(stderr) + if duration is None or duration <= 0: + logger.warning(f"Could not determine duration of {path.name} — uploading untrimmed.") + return TrimResult(path=path) + + regions = _parse_silences(stderr) + keep_start, keep_end, all_silence = speech_bounds(regions, duration, margin) + + if all_silence: + # Deliberately NOT trimmed to nothing. 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). + logger.warning( + f"{path.name} is entirely silence ({duration:.2f}s, threshold {threshold}dB) — " + "no speech detected." + ) + return TrimResult(path=path, duration_before=duration, duration_after=duration, all_silence=True) + + lead = keep_start + tail = duration - keep_end + if (lead + tail) < MIN_TRIM_SECONDS: + return TrimResult(path=path, duration_before=duration, duration_after=duration) + + trimmed = path.with_name(f"{path.stem}_trimmed{path.suffix}") + try: + code, err = await _run([ + "ffmpeg", "-hide_banner", "-nostdin", "-nostats", + "-loglevel", "warning", "-y", + "-ss", f"{keep_start:.3f}", + "-i", str(path), + "-t", f"{keep_end - keep_start:.3f}", + "-ac", "1", "-ar", sample_rate, "-b:a", bitrate, + "-f", "mp3", str(trimmed), + ]) + except Exception as e: + logger.warning(f"Silence trim failed for {path.name} ({e}) — uploading untrimmed.") + trimmed.unlink(missing_ok=True) + return TrimResult(path=path, duration_before=duration, duration_after=duration) + + if code != 0 or not trimmed.exists() or trimmed.stat().st_size == 0: + logger.warning(f"Silence trim produced no output for {path.name} ({err.strip()}) — uploading untrimmed.") + trimmed.unlink(missing_ok=True) + return TrimResult(path=path, duration_before=duration, duration_after=duration) + + trimmed.replace(path) + logger.info( + f"Trimmed {path.name}: -{lead:.2f}s lead, -{tail:.2f}s tail " + f"({duration:.2f}s → {keep_end - keep_start:.2f}s)" + ) + return TrimResult( + path=path, + lead=lead, + tail=tail, + duration_before=duration, + duration_after=keep_end - keep_start, + applied=True, + ) diff --git a/drb-edge-node/app/internal/call_recorder.py b/drb-edge-node/app/internal/call_recorder.py index a199e91..1f8c18c 100644 --- a/drb-edge-node/app/internal/call_recorder.py +++ b/drb-edge-node/app/internal/call_recorder.py @@ -1,12 +1,24 @@ """ -Continuous PulseAudio ring buffer + per-call slicing. +Continuous PulseAudio capture: a ring buffer for PRE-ROLL, a per-call +accumulator for the call itself. -The ring-buffer design is deliberate and load-bearing: a persistent capture -process runs for the lifetime of the node and every call is cut out of the -buffer after the fact. Spawning FFmpeg per call used to lose the first 1-2 s to -process startup, which meant short transmissions produced empty files. Nothing -about that changes here — only the INPUT changes, from an HTTP GET on Icecast to -a PulseAudio monitor capture. +A persistent capture process runs for the lifetime of the node. Spawning FFmpeg +per call used to lose the first 1-2 s to process startup, which meant short +transmissions produced empty files, so capture never stops. + +TWO BUFFERS, TWO JOBS — this split is load-bearing: + + RING BUFFER holds the last RING_BUFFER_SECONDS of audio at all times. Its + only job is PRE-ROLL: however late we notice a grant, we can + still seek back to OP25's exact timestamp. It is sized for + detection latency, nothing else. + + ACCUMULATOR opened by start_recording(), fed by every subsequent chunk, and + closed by stop_recording(). Call length is therefore bounded by + MAX_RECORDING_SECONDS alone — NOT by the ring buffer size. The + old design sliced the finished call back out of the ring buffer, + which silently clamped the front of any call longer than + RING_BUFFER_SECONDS. Why not Icecast: it lags ~1 s at connect and drifts progressively to 100 s+, so slice timestamps and audio content diverge without bound. Icecast stays in the @@ -23,13 +35,14 @@ RING_BUFFER_SECONDS.) import asyncio import time from collections import deque +from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import List, Optional, Tuple import httpx from app.config import settings -from app.internal import credentials, pulse +from app.internal import audio_trim, credentials, pulse from app.internal.logger import logger # Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher. @@ -37,15 +50,17 @@ MAX_RECORDING_SECONDS = 600 # Audio included ahead of OP25's call_log timestamp. The grant is logged when the # channel is granted, so the first syllable can land marginally before it. +# +# Kept small on purpose: measurement on a live node shows 1.71–2.45 s of real +# grant→speech delay on every call, so there is no clipping risk at the head and +# a larger pre-roll would only add dead air. PRE_ROLL_SECONDS = 0.25 -# Rolling history kept when no call is active. Budget for the worst realistic +# Rolling history kept for PRE-ROLL ONLY. Budget for the worst realistic # detection latency: 0.5 s poll interval + ~0.2 s http_server blocking floor + # up to 3 s httpx timeout on a stalled poll + callback work ≈ 4 s from grant to # start_recording(). 30 s is ~7x that margin, and at 16 kbps costs only ~60 KB of -# RAM, so there is no reason to trim it closer. The buffer is what makes -# detection latency harmless: however late we notice, we seek back to OP25's -# exact timestamp. +# RAM. This value does NOT bound call length — the accumulator does. RING_BUFFER_SECONDS = 30 # ~128 ms of audio per chunk at 16 kbps. Chunk size IS the timestamp resolution of @@ -61,18 +76,74 @@ READ_CHUNK_BYTES = 256 MP3_BITRATE = "16k" MP3_SAMPLE_RATE = "22050" +# Hard memory ceiling for one call's accumulator. 16 kbps is 2 KB/s, so 600 s of +# call is ~1.2 MB; 4x that is the ceiling, which both leaves room for encoder +# overshoot and guarantees a runaway call can never eat a Pi's RAM. +_MP3_BYTES_PER_SECOND = 16_000 // 8 +MAX_RECORDING_BYTES = MAX_RECORDING_SECONDS * _MP3_BYTES_PER_SECOND * 4 + +# How long stop_recording() will wait for captured audio to actually reach the +# call's end timestamp. PulseAudio → FFmpeg → MP3 encoder → muxer → our pipe read +# is a pipeline with latency, so at the instant a call ends the newest buffered +# chunk is typically a few hundred ms OLDER than the end epoch. Slicing +# immediately therefore cuts the tail short — which costs the last word of the +# transmission, usually the disposition or the address. Bounded so a dead capture +# can never hang the upload path. +TAIL_WAIT_TIMEOUT_SECONDS = 2.0 +TAIL_WAIT_POLL_SECONDS = 0.05 + # Backoff bounds for restarting a dead capture process. RESTART_BACKOFF_MIN = 1.0 RESTART_BACKOFF_MAX = 15.0 +@dataclass +class _ActiveRecording: + """Audio accumulating for the call currently being recorded.""" + + call_id: str + call_start: float # OP25 grant epoch + slice_start: float # call_start - PRE_ROLL_SECONDS + chunks: List[Tuple[float, bytes]] = field(default_factory=list) + total_bytes: int = 0 + # Seconds of requested pre-roll that were not in the buffer at open time. + clamped_seconds: float = 0.0 + # True once the byte ceiling was hit and audio started being dropped. + truncated_by_cap: bool = False + + +@dataclass +class Recording: + """ + A finished recording plus the timing metadata needed to map audio position + back to wall clock. + + `started_at`/`ended_at` upstream keep meaning the CALL's bounds. These are + the AUDIO's bounds, which differ once silence is trimmed: + + wall_clock_of(audio_offset_t) == audio_start_epoch + t + + Bounds are accurate to ±one capture chunk (~128 ms). + """ + + call_id: str + path: Optional[Path] + audio_start_epoch: float + audio_end_epoch: float + lead_trimmed: float = 0.0 + tail_trimmed: float = 0.0 + clamped_seconds: float = 0.0 + all_silence: bool = False + + class CallRecorder: - """Continuous PulseAudio capture into a ring buffer, sliced per call.""" + """Continuous PulseAudio capture: ring buffer for pre-roll, accumulator per call.""" def __init__(self): self._recordings_dir = Path(settings.recordings_path) - # Ring buffer: deque of (wall_clock_epoch_at_arrival, mp3_bytes) + # Ring buffer: deque of (wall_clock_epoch_at_arrival, mp3_bytes). + # Pre-roll only — see the module docstring. self._buffer: deque[Tuple[float, bytes]] = deque() self._buffer_bytes: int = 0 @@ -80,10 +151,8 @@ class CallRecorder: self._proc: Optional[asyncio.subprocess.Process] = None self._capturing: bool = False - # Active recording state - self._call_id: Optional[str] = None - self._call_start: Optional[float] = None # OP25 grant epoch - self._slice_start: Optional[float] = None # _call_start - PRE_ROLL_SECONDS + # Active recording state (None when idle) + self._active: Optional[_ActiveRecording] = None # ------------------------------------------------------------------ # Lifecycle @@ -213,26 +282,35 @@ class CallRecorder: pass def _ingest(self, chunk: bytes) -> None: - """Append a chunk and trim stale audio off the front of the buffer.""" + """Append a chunk to the ring buffer and, if recording, the accumulator.""" now = time.time() self._buffer.append((now, chunk)) self._buffer_bytes += len(chunk) - # While recording, never trim anything the current slice still needs — but - # keep a hard ceiling so a recording that somehow never closes can't grow - # the buffer without bound. - if self._slice_start is not None: - keep_from = max( - self._slice_start, - now - (MAX_RECORDING_SECONDS + RING_BUFFER_SECONDS), - ) - else: - keep_from = now - RING_BUFFER_SECONDS - + # The ring buffer serves pre-roll only, so it is trimmed to a fixed + # window unconditionally — an open recording no longer pins it, because + # the accumulator owns that audio. + keep_from = now - RING_BUFFER_SECONDS while self._buffer and self._buffer[0][0] < keep_from: _, old = self._buffer.popleft() self._buffer_bytes -= len(old) + active = self._active + if active is None: + return + + if active.total_bytes + len(chunk) > MAX_RECORDING_BYTES: + if not active.truncated_by_cap: + active.truncated_by_cap = True + logger.warning( + f"Recording {active.call_id} hit the {MAX_RECORDING_BYTES} byte memory ceiling " + f"after {now - active.slice_start:.1f}s — further audio is being dropped." + ) + return + + active.chunks.append((now, chunk)) + active.total_bytes += len(chunk) + # ------------------------------------------------------------------ # Recording API # ------------------------------------------------------------------ @@ -243,52 +321,73 @@ class CallRecorder: clock); the slice begins PRE_ROLL_SECONDS before it. Omit it only when no OP25 timestamp is available — then we fall back to "now", losing precision. """ - if self._call_id: - logger.warning(f"Recording already active ({self._call_id}) — ignoring start for {call_id}.") + if self._active is not None: + logger.warning(f"Recording already active ({self._active.call_id}) — ignoring start for {call_id}.") return False - self._call_id = call_id - self._call_start = start_epoch if start_epoch else time.time() - self._slice_start = self._call_start - PRE_ROLL_SECONDS + call_start = start_epoch if start_epoch else time.time() + slice_start = call_start - PRE_ROLL_SECONDS if not self._capturing: logger.warning(f"Recording {call_id} opened while PulseAudio capture is down — audio may be missing.") + # Seed the accumulator with the pre-roll already sitting in the ring + # buffer. No await between reading the buffer and publishing _active, so + # the capture task cannot slip a chunk in between and double-count it. + clamped = 0.0 oldest = self._buffer[0][0] if self._buffer else None - if oldest is not None and self._slice_start < oldest: + if oldest is not None and slice_start < oldest: # Pre-roll predates the buffer: node just started, capture restarted, - # or OP25's timestamp is far in the past. Clamp and say so. + # or OP25's timestamp is far in the past. Clamp and say so LOUDLY — + # this is silent audio loss otherwise. + clamped = oldest - slice_start logger.warning( - f"Pre-roll for call {call_id} predates buffered audio by " - f"{oldest - self._slice_start:.2f}s — recording starts at the buffer head." + f"BUFFER CLAMP: pre-roll for call {call_id} predates buffered audio by " + f"{clamped:.2f}s — recording starts at the buffer head and that audio is lost." ) - logger.info(f"Recording started: {call_id} (slice from {self._slice_start:.3f})") + seeded = [(ts, chunk) for ts, chunk in self._buffer if ts >= slice_start] + self._active = _ActiveRecording( + call_id=call_id, + call_start=call_start, + slice_start=slice_start, + chunks=seeded, + total_bytes=sum(len(c) for _, c in seeded), + clamped_seconds=clamped, + ) + + logger.info(f"Recording started: {call_id} (slice from {slice_start:.3f})") return True - async def stop_recording(self, end_epoch: Optional[float] = None) -> Optional[Path]: - """Close the recording and write the slice. `end_epoch` is host wall clock.""" - if not self._call_id: + async def stop_recording(self, end_epoch: Optional[float] = None) -> Optional[Recording]: + """ + Close the recording and write the file. `end_epoch` is host wall clock. + + Waits (bounded) for captured audio to actually cover `end_epoch` before + slicing — see TAIL_WAIT_TIMEOUT_SECONDS. Returns None only when there was + no recording open or no audio at all. + """ + active = self._active + if active is None: return None - call_id = self._call_id - call_start = self._call_start - slice_start = self._slice_start - self._call_id = None - self._call_start = None - self._slice_start = None + call_id = active.call_id + slice_start = active.slice_start end = end_epoch if end_epoch else time.time() - if call_start is not None: - end = min(end, call_start + MAX_RECORDING_SECONDS) - if slice_start is None: - return None + end = min(end, active.call_start + MAX_RECORDING_SECONDS) + + # The accumulator keeps filling during this wait — that is the point. + await self._await_tail(end, call_id) + self._active = None chunks: List[bytes] = [] - for ts, chunk in self._buffer: + last_ts = slice_start + for ts, chunk in active.chunks: if ts < slice_start: continue chunks.append(chunk) + last_ts = ts if ts >= end: # Include the chunk straddling `end` so the tail is never clipped, # then stop. @@ -301,6 +400,12 @@ class CallRecorder: ) return None + if last_ts < end - TAIL_WAIT_POLL_SECONDS: + logger.warning( + f"BUFFER CLAMP: call {call_id} ends {end - last_ts:.2f}s after the newest captured " + "audio — the tail is short. Capture may be stalled or restarting." + ) + self._recordings_dir.mkdir(parents=True, exist_ok=True) ts_str = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") output_path = self._recordings_dir / f"{ts_str}_{call_id}.mp3" @@ -308,13 +413,88 @@ class CallRecorder: output_path.write_bytes(b"".join(chunks)) size = output_path.stat().st_size - if size > 0: - logger.info(f"Recording saved: {output_path.name} ({size} bytes, {end - slice_start:.2f}s window)") - return output_path + if size <= 0: + output_path.unlink(missing_ok=True) + logger.warning(f"Recording for call {call_id} produced an empty file.") + return None - output_path.unlink(missing_ok=True) - logger.warning(f"Recording for call {call_id} produced an empty file.") - return None + audio_end = min(end, last_ts) + logger.info(f"Recording saved: {output_path.name} ({size} bytes, {audio_end - slice_start:.2f}s window)") + + recording = Recording( + call_id=call_id, + path=output_path, + audio_start_epoch=slice_start, + audio_end_epoch=audio_end, + clamped_seconds=active.clamped_seconds, + ) + return await self._apply_trim(recording) + + async def _await_tail(self, end: float, call_id: str) -> float: + """ + Block until captured audio reaches `end`, or the bounded timeout expires. + + Returns seconds waited. Logs whenever a wait was actually needed so the + real pipeline latency is observable in the field. + """ + if not self._buffer: + return 0.0 + if self._buffer[-1][0] >= end: + return 0.0 + + started = time.monotonic() + deadline = started + TAIL_WAIT_TIMEOUT_SECONDS + while time.monotonic() < deadline: + await asyncio.sleep(TAIL_WAIT_POLL_SECONDS) + if self._buffer and self._buffer[-1][0] >= end: + waited = time.monotonic() - started + logger.info(f"Waited {waited:.2f}s for the tail of call {call_id} to reach the buffer.") + return waited + if not self._capturing: + break # capture died mid-wait; nothing more is coming + + waited = time.monotonic() - started + newest = self._buffer[-1][0] if self._buffer else end + logger.warning( + f"Tail wait for call {call_id} gave up after {waited:.2f}s — captured audio is still " + f"{max(0.0, end - newest):.2f}s short of the call end. Tail may be clipped." + ) + return waited + + async def _apply_trim(self, recording: Recording) -> Recording: + """ + Strip leading/trailing dead air and keep the timing metadata honest. + + An all-silence recording is NOT uploaded: it carries no information and + silence is exactly what makes Whisper hallucinate. It is logged instead, + because it also means something is wrong with the audio path. + """ + if not settings.trim_silence or recording.path is None: + return recording + + result = await audio_trim.trim_silence( + recording.path, + sample_rate=MP3_SAMPLE_RATE, + bitrate=MP3_BITRATE, + ) + + if result.all_silence: + logger.warning( + f"Call {recording.call_id} contains no speech at all — skipping upload. " + "Check squelch, the Liquidsoap output and the drb_sink monitor." + ) + recording.path.unlink(missing_ok=True) + recording.path = None + recording.all_silence = True + return recording + + if result.applied: + recording.lead_trimmed = result.lead + recording.tail_trimmed = result.tail + # Wall clock of the trimmed audio's first and last sample. + recording.audio_start_epoch += result.lead + recording.audio_end_epoch -= result.tail + return recording # ------------------------------------------------------------------ # Upload (unchanged interface) @@ -327,6 +507,8 @@ class CallRecorder: talkgroup_id: Optional[int] = None, talkgroup_name: Optional[str] = None, system_id: Optional[str] = None, + audio_start_epoch: Optional[float] = None, + audio_end_epoch: Optional[float] = None, ) -> Optional[str]: if not settings.c2_url: logger.info("No C2_URL configured — skipping upload.") @@ -343,6 +525,13 @@ class CallRecorder: form["talkgroup_name"] = talkgroup_name if system_id: form["system_id"] = system_id + # Where this audio really sits on the wall clock once silence is trimmed. + # C2 does not declare these Form fields yet, so FastAPI ignores them — + # they cost nothing and are here for when playback/correlation want them. + if audio_start_epoch is not None: + form["audio_start_epoch"] = f"{audio_start_epoch:.3f}" + if audio_end_epoch is not None: + form["audio_end_epoch"] = f"{audio_end_epoch:.3f}" try: async with httpx.AsyncClient(timeout=120) as client: @@ -372,7 +561,7 @@ class CallRecorder: @property def is_recording(self) -> bool: - return self._call_id is not None + return self._active is not None @property def is_capturing(self) -> bool: diff --git a/drb-edge-node/app/internal/logger.py b/drb-edge-node/app/internal/logger.py index e14e2c1..9e40354 100644 --- a/drb-edge-node/app/internal/logger.py +++ b/drb-edge-node/app/internal/logger.py @@ -7,4 +7,12 @@ logging.basicConfig( handlers=[logging.StreamHandler(sys.stdout)], ) +# The metadata watcher polls the OP25 terminal twice a second and httpx logs +# every one of those requests at INFO ("HTTP Request: POST http://... 200 OK"). +# That is ~170k lines/day of pure noise which buries real events and makes field +# log-reading useless. WARNING keeps genuine transport failures visible. +# httpcore is the transport layer underneath httpx and is just as chatty. +for _noisy in ("httpx", "httpcore"): + logging.getLogger(_noisy).setLevel(logging.WARNING) + logger = logging.getLogger("drb-edge-node") diff --git a/drb-edge-node/app/internal/metadata_watcher.py b/drb-edge-node/app/internal/metadata_watcher.py index dad1cff..4108588 100644 --- a/drb-edge-node/app/internal/metadata_watcher.py +++ b/drb-edge-node/app/internal/metadata_watcher.py @@ -47,15 +47,23 @@ POLL_INTERVAL = 0.5 # Seconds of unreachable OP25 before an open segment is force-closed. OP25_OFFLINE_GRACE = 3.0 -# Audio kept after the observed end of the last transmission, so the srcaddr edge -# (up to one poll late) never clips the tail. -TAIL_PAD_SECONDS = 0.5 - # Hard ceiling on a single segment; mirrors MAX_RECORDING_SECONDS in call_recorder # so a talkgroup that never goes quiet cannot produce an unbounded recording. MAX_SEGMENT_SECONDS = 600 +def _tail_pad() -> float: + """ + Audio kept after the observed end of the last transmission, so the srcaddr + edge (up to one poll late) plus encoder latency never clips the tail. + + Read live from settings (env CALL_TAIL_PAD_SECONDS) rather than frozen into a + module constant, so it is tunable per node. See the setting for why the + default moved 0.5 → 1.0. + """ + return settings.call_tail_pad_seconds + + def _as_int(value: Any) -> Optional[int]: """Coerce an OP25 field to a positive int, or None. Rejects 0/""/"None".""" if value is None: @@ -235,7 +243,25 @@ class MetadataWatcher: # STOP: quiet for long enough. End the audio at the last transmission # plus a short pad, not at "now" — otherwise every recording carries # call_idle_timeout seconds of silence. - end = (self._last_tx_end + TAIL_PAD_SECONDS) if self._last_tx_end is not None else now + # + # The measured idle below is the CONTROL-CHANNEL idle (srcaddr 1→0 + # edge → now). It is NOT comparable to silence measured in the audio, + # which additionally contains the ~1.9 s P25 grant→speech delay. + # Tune settings.call_idle_timeout from THIS number and nothing else. + if self._last_tx_end is not None: + measured_idle = now - self._last_tx_end + end = self._last_tx_end + _tail_pad() + logger.info( + f"Idle timeout for tgid {self._current_tgid}: measured control-channel idle " + f"{measured_idle:.2f}s (threshold {settings.call_idle_timeout:.2f}s, " + f"tail pad {_tail_pad():.2f}s)." + ) + else: + end = now + logger.info( + f"Idle timeout for tgid {self._current_tgid}: no srcaddr end edge observed, " + f"idle {now - self._last_activity:.2f}s measured from last activity." + ) await self._close_segment(min(end, now), reason="idle_timeout") return diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index 5f4f181..f22caf5 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -1,5 +1,8 @@ import asyncio from contextlib import asynccontextmanager +from datetime import datetime, timezone +from typing import Optional + from fastapi import FastAPI from app.config import settings from app.models import SystemConfig @@ -20,6 +23,13 @@ from app.routers import api, ui # Event handlers wired up at startup # --------------------------------------------------------------------------- +def _iso(epoch: Optional[float]) -> Optional[str]: + """Epoch → UTC ISO-8601, matching metadata_watcher's timestamp format.""" + if epoch is None: + return None + return datetime.fromtimestamp(epoch, timezone.utc).isoformat() + + async def on_call_start(data: dict): radio_bot.start_stream() await mqtt_manager.publish_status("recording") @@ -35,20 +45,44 @@ async def on_call_start(data: dict): async def on_call_end(data: dict): radio_bot.stop_stream() - file_path = await call_recorder.stop_recording(end_epoch=data.get("ended_at_epoch")) - if file_path: + recording = await call_recorder.stop_recording(end_epoch=data.get("ended_at_epoch")) + + if recording is not None and recording.path is not None: + # Silence trimming shortens the audio, so the audio's own bounds no + # longer equal the call's. `started_at`/`ended_at` keep meaning the CALL + # (what OP25 observed on the control channel) — these extra fields carry + # the AUDIO's wall-clock bounds so playback, correlation and incident + # timelines can still map an audio offset back to real time: + # wall_clock_of(audio_offset_t) == audio_start_epoch + t + data["audio_start_at"] = _iso(recording.audio_start_epoch) + data["audio_end_at"] = _iso(recording.audio_end_epoch) + data["audio_start_epoch"] = recording.audio_start_epoch + data["audio_end_epoch"] = recording.audio_end_epoch + data["audio_lead_trimmed"] = round(recording.lead_trimmed, 3) + data["audio_tail_trimmed"] = round(recording.tail_trimmed, 3) + if recording.clamped_seconds: + data["audio_clamped_seconds"] = round(recording.clamped_seconds, 3) + + if recording is not None and recording.path is not None: node_cfg = load_node_config() audio_url = await call_recorder.upload_recording( - file_path, + recording.path, data["call_id"], talkgroup_id=data.get("tgid"), talkgroup_name=data.get("tgid_name"), system_id=node_cfg.assigned_system_id, + audio_start_epoch=recording.audio_start_epoch, + audio_end_epoch=recording.audio_end_epoch, ) if audio_url: data["audio_url"] = audio_url else: logger.error(f"Audio upload failed for call {data['call_id']}. Verify C2_URL and Node API Key.") + elif recording is not None and recording.all_silence: + # Explicit policy: an all-silence recording is not uploaded. It has no + # transcript value and silence is what makes Whisper invent text. + data["audio_skipped"] = "all_silence" + logger.warning(f"Call {data['call_id']} was pure silence — no upload. Investigate the audio path.") else: logger.warning( f"No recording file generated for call {data['call_id']} " diff --git a/drb-edge-node/tests/test_audio_trim.py b/drb-edge-node/tests/test_audio_trim.py new file mode 100644 index 0000000..c19fb6a --- /dev/null +++ b/drb-edge-node/tests/test_audio_trim.py @@ -0,0 +1,160 @@ +""" +Unit tests for silence-trim decision logic. + +`speech_bounds` is pure on purpose so the "what do we keep" decision — the part +that can destroy a transmission if it is wrong — is testable without FFmpeg. +The numbers below come from ffmpeg silencedetect run against six real recordings +off a live P25 node: 1.71–2.45 s of leading silence and 0.00–1.11 s trailing. +""" +import pytest + +from app.internal.audio_trim import ( + TrimResult, + _parse_duration, + _parse_silences, + speech_bounds, +) + +GUARD = 0.25 + + +def test_leading_silence_is_trimmed_with_a_guard_margin(): + # Real shape of file f4bfaa1f: 1.85s lead, 0.34s trail, 4.54s total. + regions = [(0.0, 1.85), (4.20, None)] + start, end, all_silence = speech_bounds(regions, duration=4.54, guard=GUARD) + + assert not all_silence + assert start == pytest.approx(1.85 - GUARD) + assert end == pytest.approx(4.20 + GUARD) + # The guard must never eat into detected speech. + assert start < 1.85 and end > 4.20 + + +def test_guard_margin_never_runs_past_the_file_bounds(): + regions = [(0.0, 0.10), (3.95, None)] + start, end, _ = speech_bounds(regions, duration=4.0, guard=1.0) + + assert start == 0.0 + assert end == 4.0 + + +def test_trailing_silence_is_trimmed_when_ffmpeg_closes_the_region_at_eof(): + """ + FFmpeg 6.x flushes a `silence_end` at EOF, so a trailing region looks closed. + Treating "no silence_end" as the only trailing signal silently disabled tail + trimming entirely — verified against ffmpeg 6.1.1. + """ + # Real ffmpeg 6.1.1 output for a 2s-silence + 1.5s-tone + 1s-silence file. + regions = [(0.0, 2.05361), (3.56367, 4.63102)] + start, end, all_silence = speech_bounds(regions, duration=4.65, guard=GUARD) + + assert not all_silence + assert start == pytest.approx(2.05361 - GUARD) + assert end == pytest.approx(3.56367 + GUARD), "the trailing second must be trimmed" + + +def test_all_silence_survives_ffmpeg_closing_the_region_at_eof(): + # Real ffmpeg 6.1.1 output for a 4s file of pure silence. + _, _, all_silence = speech_bounds([(0.0, 4.0)], duration=4.03, guard=GUARD) + assert all_silence + + +def test_trailing_silence_that_does_not_reach_eof_is_left_alone(): + """ + A silence region with a closing silence_end is an internal pause between + transmissions, not dead air at the tail. Trimming it would cut the middle + out of a conversation. + """ + regions = [(0.0, 1.9), (5.0, 7.5)] + start, end, _ = speech_bounds(regions, duration=12.0, guard=GUARD) + + assert start == pytest.approx(1.9 - GUARD) + assert end == 12.0, "an internal pause must not shorten the file" + + +def test_no_silence_detected_keeps_the_whole_file(): + start, end, all_silence = speech_bounds([], duration=6.0, guard=GUARD) + + assert (start, end) == (0.0, 6.0) + assert not all_silence + + +def test_silence_starting_late_is_not_treated_as_leading(): + """Only a region at the very head counts as leading silence.""" + regions = [(1.20, 2.00)] + start, end, _ = speech_bounds(regions, duration=5.0, guard=GUARD) + + assert start == 0.0, "speech before 1.20s must not be trimmed away" + assert end == 5.0 + + +def test_all_silence_is_reported_not_trimmed_to_nothing(): + # One region covering the whole file and running to EOF. + regions = [(0.0, None)] + start, end, all_silence = speech_bounds(regions, duration=4.0, guard=GUARD) + + assert all_silence + assert (start, end) == (0.0, 4.0), "an all-silence file must not become zero-length" + + +def test_all_silence_when_head_and_tail_regions_overlap(): + regions = [(0.0, 3.2), (3.0, None)] + _, _, all_silence = speech_bounds(regions, duration=4.0, guard=GUARD) + + assert all_silence + + +# --------------------------------------------------------------------------- +# FFmpeg output parsing +# --------------------------------------------------------------------------- + +# Verbatim shape of ffmpeg 6.1.1 output. +FFMPEG_STDERR = """ +Input #0, mp3, from '/recordings/x.mp3': + Duration: 00:00:04.70, start: 0.050113, bitrate: 16 kb/s +[silencedetect @ 0000029160e63f40] silence_start: 0 +[silencedetect @ 0000029160e63f40] silence_end: 2.05361 | silence_duration: 2.05361 +[silencedetect @ 0000029160e63f40] silence_start: 3.56367 +[silencedetect @ 0000029160e63f40] silence_end: 4.63102 | silence_duration: 1.06735 +[out#0/null @ 0x2] video:0kB audio:97kB +""" + +FFMPEG_STDERR_OPEN_TAIL = """ + Duration: 00:00:04.54, start: 0.000000, bitrate: 16 kb/s +[silencedetect @ 0x1] silence_start: 0 +[silencedetect @ 0x1] silence_end: 1.85042 | silence_duration: 1.85042 +[silencedetect @ 0x1] silence_start: 4.20134 +""" + + +def test_duration_is_corrected_for_the_mp3_container_start_offset(): + """ + MP3 encoder delay makes the container duration longer than the audio + silencedetect timestamps. Without this correction the trailing-region test + needs a slack epsilon big enough to clip real speech. + """ + assert _parse_duration(FFMPEG_STDERR) == pytest.approx(4.70 - 0.050113) + + +def test_duration_is_none_when_absent(): + assert _parse_duration("no duration here") is None + + +def test_silence_regions_are_parsed(): + regions = _parse_silences(FFMPEG_STDERR) + + assert len(regions) == 2 + assert regions[0] == (pytest.approx(0.0), pytest.approx(2.05361)) + assert regions[1] == (pytest.approx(3.56367), pytest.approx(4.63102)) + + +def test_a_region_with_no_silence_end_is_still_parsed(): + """Older FFmpeg simply stopped reporting at EOF — keep handling that.""" + regions = _parse_silences(FFMPEG_STDERR_OPEN_TAIL) + + assert regions[-1][1] is None + + +def test_trim_result_reports_total_trimmed(): + result = TrimResult(path=None, lead=1.9, tail=0.35) + assert result.trimmed_seconds == pytest.approx(2.25) diff --git a/drb-edge-node/tests/test_call_recorder.py b/drb-edge-node/tests/test_call_recorder.py index febd2dc..a307fdb 100644 --- a/drb-edge-node/tests/test_call_recorder.py +++ b/drb-edge-node/tests/test_call_recorder.py @@ -1,16 +1,24 @@ """ -Unit tests for the CallRecorder ring buffer and per-call slicing. +Unit tests for the CallRecorder pre-roll ring buffer and per-call accumulator. -No FFmpeg and no PulseAudio: the buffer is filled directly with timestamped -chunks, which is exactly what _ingest() produces at runtime. +No FFmpeg and no PulseAudio: chunks are pushed through _ingest() with a patched +clock, which is exactly what the capture loop does at runtime. Silence trimming +is disabled by default here and exercised separately with a stubbed trimmer. """ +import asyncio +import itertools import time +from typing import List from unittest.mock import patch import pytest +from app.config import settings +from app.internal import call_recorder as recorder_mod +from app.internal.audio_trim import TrimResult from app.internal.call_recorder import ( CallRecorder, + MAX_RECORDING_BYTES, MAX_RECORDING_SECONDS, PRE_ROLL_SECONDS, RING_BUFFER_SECONDS, @@ -21,51 +29,120 @@ CHUNK_INTERVAL = 0.1 # seconds of audio per synthetic chunk @pytest.fixture -def recorder(tmp_path): +def recorder(tmp_path, monkeypatch): + monkeypatch.setattr(settings, "trim_silence", False) r = CallRecorder() r._recordings_dir = tmp_path r._capturing = True return r -def fill(recorder, start: float, end: float, marker: bytes = b"A"): - """Append one chunk every CHUNK_INTERVAL seconds over [start, end).""" +def ingest(recorder, start: float, end: float, marker: bytes = b"A", index: int = 0) -> int: + """Feed one chunk every CHUNK_INTERVAL seconds over [start, end) through _ingest.""" + stamps: List[float] = [] + chunks: List[bytes] = [] ts = start - index = 0 while ts < end: - recorder._buffer.append((ts, marker + str(index).encode() + b";")) + stamps.append(ts) + chunks.append(marker + str(index).encode() + b";") index += 1 ts = round(ts + CHUNK_INTERVAL, 6) + with patch("app.internal.call_recorder.time.time", side_effect=stamps): + for chunk in chunks: + recorder._ingest(chunk) + return index + + +def fill(recorder, start: float, end: float, marker: bytes = b"A", index: int = 0) -> int: + """Alias kept for readability where the accumulator is not the point.""" + return ingest(recorder, start, end, marker=marker, index=index) + def timestamps(recorder): return [ts for ts, _ in recorder._buffer] +def markers(path) -> List[str]: + return path.read_bytes().decode().strip(";").split(";") + + +def indices(path) -> List[int]: + return [int(m[1:]) for m in markers(path) if m[1:].isdigit()] + + # --------------------------------------------------------------------------- -# Ring buffer trimming +# Ring buffer trimming (pre-roll duty only) # --------------------------------------------------------------------------- def test_idle_buffer_keeps_only_the_rolling_window(recorder): - with patch("app.internal.call_recorder.time.time", return_value=T0): - for offset in range(0, int(RING_BUFFER_SECONDS) + 20): - with patch("app.internal.call_recorder.time.time", return_value=T0 + offset): - recorder._ingest(b"x" * 16) + for offset in range(0, int(RING_BUFFER_SECONDS) + 20): + with patch("app.internal.call_recorder.time.time", return_value=T0 + offset): + recorder._ingest(b"x" * 16) assert len(recorder._buffer) <= RING_BUFFER_SECONDS + 1 assert min(timestamps(recorder)) >= (T0 + RING_BUFFER_SECONDS + 19) - RING_BUFFER_SECONDS @pytest.mark.asyncio -async def test_buffer_is_not_trimmed_below_the_active_slice(recorder): - fill(recorder, T0, T0 + 5.0) +async def test_ring_buffer_is_trimmed_even_while_recording(recorder): + """ + The ring buffer serves PRE-ROLL only. An open recording must no longer pin + it — that was the mechanism that made call length depend on buffer size. + """ + index = ingest(recorder, T0, T0 + 2.0) await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + ingest(recorder, T0 + 2.0, T0 + 2.0 + RING_BUFFER_SECONDS + 10, index=index) - # Ingest far past the normal rolling window; the slice start must survive. - with patch("app.internal.call_recorder.time.time", return_value=T0 + RING_BUFFER_SECONDS + 10): - recorder._ingest(b"z") + assert recorder.buffered_seconds <= RING_BUFFER_SECONDS + 1 + # ...and the audio the ring buffer dropped is safe in the accumulator. + assert recorder._active is not None + assert recorder._active.chunks[0][0] == pytest.approx(T0 + 1.0 - PRE_ROLL_SECONDS, abs=CHUNK_INTERVAL) - assert min(timestamps(recorder)) <= (T0 + 1.0) - PRE_ROLL_SECONDS + CHUNK_INTERVAL + +# --------------------------------------------------------------------------- +# Call length must not be bounded by the ring buffer +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_call_longer_than_the_ring_buffer_is_captured_whole(recorder): + call_length = RING_BUFFER_SECONDS * 2 + 5 # 65s against a 30s ring buffer + grant = T0 + 1.0 + end = grant + call_length + + index = ingest(recorder, T0, grant) + await recorder.start_recording("call-long", start_epoch=grant) + ingest(recorder, grant, end + 1.0, index=index) + + rec = await recorder.stop_recording(end_epoch=end) + assert rec is not None and rec.path is not None + + kept = indices(rec.path) + # Contiguous: no hole anywhere in the middle of a 65s call. + assert kept == list(range(kept[0], kept[-1] + 1)) + span = (kept[-1] - kept[0]) * CHUNK_INTERVAL + assert span > RING_BUFFER_SECONDS, "call length must not be clamped by the ring buffer" + assert span == pytest.approx(call_length + PRE_ROLL_SECONDS, abs=2 * CHUNK_INTERVAL) + + +@pytest.mark.asyncio +async def test_accumulator_stops_growing_at_the_memory_ceiling(recorder, caplog): + """A runaway call must not be able to exhaust RAM on a Pi.""" + await recorder.start_recording("call-runaway", start_epoch=T0) + + big = b"z" * 64_000 + needed = (MAX_RECORDING_BYTES // len(big)) + 5 + ticks = itertools.count() + with caplog.at_level("WARNING", logger="drb-edge-node"): + with patch("app.internal.call_recorder.time.time", + side_effect=lambda: T0 + next(ticks) * 0.1): + for _ in range(needed): + recorder._ingest(big) + + assert recorder._active is not None + assert recorder._active.total_bytes <= MAX_RECORDING_BYTES + assert recorder._active.truncated_by_cap + assert any("memory ceiling" in r.message for r in caplog.records) # --------------------------------------------------------------------------- @@ -78,15 +155,12 @@ async def test_slice_starts_pre_roll_before_the_op25_timestamp(recorder): grant_time = T0 + 5.0 await recorder.start_recording("call-1", start_epoch=grant_time) - assert recorder._slice_start == pytest.approx(grant_time - PRE_ROLL_SECONDS) + assert recorder._active.slice_start == pytest.approx(grant_time - PRE_ROLL_SECONDS) - path = await recorder.stop_recording(end_epoch=grant_time + 2.0) - assert path is not None and path.exists() + rec = await recorder.stop_recording(end_epoch=grant_time + 2.0) + assert rec is not None and rec.path is not None and rec.path.exists() - # Reconstruct which chunks landed in the file. - kept = path.read_bytes().decode().strip(";").split(";") - first_index = int(kept[0][1:]) - first_ts = T0 + first_index * CHUNK_INTERVAL + first_ts = T0 + indices(rec.path)[0] * CHUNK_INTERVAL # A chunk stamped `ts` holds the audio that arrived over [ts - interval, ts], # so the audio actually covered must begin at or before the requested slice @@ -102,24 +176,104 @@ async def test_tail_chunk_straddling_the_end_is_included(recorder): await recorder.start_recording("call-1", start_epoch=T0 + 1.0) # End halfway through a chunk interval. - path = await recorder.stop_recording(end_epoch=T0 + 3.05) - kept = path.read_bytes().decode().strip(";").split(";") - last_ts = T0 + int(kept[-1][1:]) * CHUNK_INTERVAL + rec = await recorder.stop_recording(end_epoch=T0 + 3.05) + last_ts = T0 + indices(rec.path)[-1] * CHUNK_INTERVAL assert last_ts >= T0 + 3.05, "the chunk covering the end instant must be kept" @pytest.mark.asyncio -async def test_pre_roll_earlier_than_buffer_start_is_clamped(recorder, caplog): - """A grant older than anything buffered must still produce a file.""" - fill(recorder, T0 + 5.0, T0 + 10.0) # buffer only covers T0+5 onwards +async def test_max_recording_seconds_caps_the_slice(recorder): + index = fill(recorder, T0, T0 + 1.0) + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + ingest(recorder, T0 + 1.0, T0 + MAX_RECORDING_SECONDS + 60, index=index) - await recorder.start_recording("call-1", start_epoch=T0) # 5 s before the head - path = await recorder.stop_recording(end_epoch=T0 + 8.0) + rec = await recorder.stop_recording(end_epoch=T0 + MAX_RECORDING_SECONDS + 50) + last_ts = T0 + indices(rec.path)[-1] * CHUNK_INTERVAL - assert path is not None and path.stat().st_size > 0 - kept = path.read_bytes().decode().strip(";").split(";") - assert kept[0] == "A0", "slice should begin at the buffer head, not fail" + assert last_ts <= T0 + 1.0 + MAX_RECORDING_SECONDS + CHUNK_INTERVAL + + +# --------------------------------------------------------------------------- +# Tail wait — the fix for recordings that ended mid-word +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_stop_waits_for_captured_audio_to_reach_the_call_end(recorder, caplog): + """ + PulseAudio → FFmpeg → encoder → muxer → our pipe read has latency, so at the + instant a call ends the newest captured chunk is OLDER than the end epoch. + Slicing immediately cuts the last word off. stop_recording must wait for it. + """ + index = ingest(recorder, T0, T0 + 4.0) + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + + async def late_tail(): + await asyncio.sleep(0.15) + with patch("app.internal.call_recorder.time.time", return_value=T0 + 4.6): + recorder._ingest(b"TAIL;") + + task = asyncio.create_task(late_tail()) + with caplog.at_level("INFO", logger="drb-edge-node"): + rec = await recorder.stop_recording(end_epoch=T0 + 4.5) + await task + + assert rec is not None and rec.path is not None + assert b"TAIL" in rec.path.read_bytes(), "the late-arriving tail must be in the file" + assert any("Waited" in r.message and "tail" in r.message for r in caplog.records), \ + "a tail wait must be observable in the field logs" + assert index # sanity: the pre-roll fill actually ran + + +@pytest.mark.asyncio +async def test_tail_wait_is_bounded_and_warns_when_audio_never_arrives(recorder, caplog, monkeypatch): + monkeypatch.setattr(recorder_mod, "TAIL_WAIT_TIMEOUT_SECONDS", 0.2) + ingest(recorder, T0, T0 + 4.0) + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + + started = time.monotonic() + with caplog.at_level("WARNING", logger="drb-edge-node"): + rec = await recorder.stop_recording(end_epoch=T0 + 10.0) + elapsed = time.monotonic() - started + + assert elapsed < 2.0, "the wait must be bounded, never open-ended" + assert rec is not None and rec.path is not None, "a short tail still beats no recording" + messages = [r.message for r in caplog.records] + assert any("Tail wait" in m and "gave up" in m for m in messages) + assert any("BUFFER CLAMP" in m for m in messages), "silent truncation must be loud" + + +@pytest.mark.asyncio +async def test_no_wait_when_the_buffer_already_covers_the_end(recorder, caplog): + ingest(recorder, T0, T0 + 10.0) + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + + started = time.monotonic() + with caplog.at_level("INFO", logger="drb-edge-node"): + await recorder.stop_recording(end_epoch=T0 + 3.0) + assert (time.monotonic() - started) < 0.1 + assert not any("Waited" in r.message for r in caplog.records) + + +# --------------------------------------------------------------------------- +# Clamping must be loud +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_pre_roll_earlier_than_buffer_start_is_clamped_and_warned(recorder, caplog): + """A grant older than anything buffered must still produce a file, loudly.""" + ingest(recorder, T0 + 5.0, T0 + 10.0) # buffer only covers T0+5 onwards + + with caplog.at_level("WARNING", logger="drb-edge-node"): + await recorder.start_recording("call-1", start_epoch=T0) # 5 s before the head + rec = await recorder.stop_recording(end_epoch=T0 + 8.0) + + assert rec is not None and rec.path is not None and rec.path.stat().st_size > 0 + assert markers(rec.path)[0] == "A0", "slice should begin at the buffer head, not fail" + # Buffer head is T0+5.0, requested slice start is T0-PRE_ROLL: everything in + # between is audio we can never recover, and the number must be reported. + assert rec.clamped_seconds == pytest.approx(5.0 + PRE_ROLL_SECONDS, abs=CHUNK_INTERVAL) + assert any("BUFFER CLAMP" in r.message for r in caplog.records) @pytest.mark.asyncio @@ -134,19 +288,79 @@ async def test_start_epoch_omitted_falls_back_to_now(recorder): fill(recorder, now - 5.0, now) await recorder.start_recording("call-1") - assert recorder._slice_start == pytest.approx(now - PRE_ROLL_SECONDS, abs=1.0) + assert recorder._active.slice_start == pytest.approx(now - PRE_ROLL_SECONDS, abs=1.0) + + +# --------------------------------------------------------------------------- +# Silence trimming and timing metadata +# --------------------------------------------------------------------------- + +async def _recorded(recorder, end_offset: float = 3.0): + ingest(recorder, T0, T0 + 10.0) + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + return await recorder.stop_recording(end_epoch=T0 + end_offset) @pytest.mark.asyncio -async def test_max_recording_seconds_caps_the_slice(recorder): - fill(recorder, T0, T0 + MAX_RECORDING_SECONDS + 60, marker=b"A") - await recorder.start_recording("call-1", start_epoch=T0 + 1.0) +async def test_trimming_is_off_when_the_setting_is_off(recorder, monkeypatch): + monkeypatch.setattr(settings, "trim_silence", False) + called = False - path = await recorder.stop_recording(end_epoch=T0 + MAX_RECORDING_SECONDS + 50) - kept = path.read_bytes().decode().strip(";").split(";") - last_ts = T0 + int(kept[-1][1:]) * CHUNK_INTERVAL + async def _never(*args, **kwargs): + nonlocal called + called = True + return TrimResult(path=None) - assert last_ts <= T0 + 1.0 + MAX_RECORDING_SECONDS + CHUNK_INTERVAL + monkeypatch.setattr(recorder_mod.audio_trim, "trim_silence", _never) + rec = await _recorded(recorder) + assert rec is not None and not called + + +@pytest.mark.asyncio +async def test_trim_shifts_the_audio_bounds_but_not_the_call_bounds(recorder, monkeypatch): + """ + Trimming changes audio duration, so the AUDIO's wall-clock bounds move. + The call's own started_at/ended_at (owned by metadata_watcher) must not be + redefined — the recorder only reports where the audio now sits. + """ + monkeypatch.setattr(settings, "trim_silence", True) + + async def _trim(path, **kwargs): + return TrimResult(path=path, lead=1.9, tail=0.4, duration_before=3.3, + duration_after=1.0, applied=True) + + monkeypatch.setattr(recorder_mod.audio_trim, "trim_silence", _trim) + + rec = await _recorded(recorder) + assert rec is not None and rec.path is not None + assert rec.lead_trimmed == pytest.approx(1.9) + assert rec.tail_trimmed == pytest.approx(0.4) + # Untrimmed slice was [T0+0.75, T0+3.0]; the audio now starts 1.9s later and + # ends 0.4s earlier, which is exactly what downstream needs to map an audio + # offset back to wall clock. + assert rec.audio_start_epoch == pytest.approx(T0 + 1.0 - PRE_ROLL_SECONDS + 1.9, abs=CHUNK_INTERVAL) + assert rec.audio_end_epoch == pytest.approx(T0 + 3.0 - 0.4, abs=CHUNK_INTERVAL) + + +@pytest.mark.asyncio +async def test_all_silence_recording_is_dropped_and_logged(recorder, monkeypatch, caplog): + monkeypatch.setattr(settings, "trim_silence", True) + seen = {} + + async def _trim(path, **kwargs): + seen["path"] = path + return TrimResult(path=path, duration_before=4.0, duration_after=4.0, all_silence=True) + + monkeypatch.setattr(recorder_mod.audio_trim, "trim_silence", _trim) + + with caplog.at_level("WARNING", logger="drb-edge-node"): + rec = await _recorded(recorder) + + assert rec is not None + assert rec.all_silence is True + assert rec.path is None, "an all-silence recording must not be uploaded" + assert not seen["path"].exists(), "the file must be cleaned up, not left on disk" + assert any("no speech" in r.message for r in caplog.records) # --------------------------------------------------------------------------- @@ -173,7 +387,7 @@ async def test_split_then_immediate_restart_keeps_both_slices(recorder): A tgid change closes one recording and opens the next at the same instant — the second must still find its pre-roll in the buffer. """ - fill(recorder, T0, T0 + 10.0) + ingest(recorder, T0, T0 + 10.0) split = T0 + 5.0 await recorder.start_recording("call-1", start_epoch=T0 + 1.0) @@ -182,9 +396,9 @@ async def test_split_then_immediate_restart_keeps_both_slices(recorder): await recorder.start_recording("call-2", start_epoch=split) second = await recorder.stop_recording(end_epoch=T0 + 8.0) - assert first is not None and first.stat().st_size > 0 - assert second is not None and second.stat().st_size > 0 - assert first != second + assert first is not None and first.path.stat().st_size > 0 + assert second is not None and second.path.stat().st_size > 0 + assert first.path != second.path # --------------------------------------------------------------------------- @@ -201,3 +415,10 @@ def test_ffmpeg_command_reads_pulse_and_flushes_packets(recorder): # writing, which would destroy the ring buffer's timestamp resolution. assert "-flush_packets" in cmd assert cmd[-1] == "-" and cmd[-2] == "mp3", "must emit MP3 on stdout for /upload" + + +def test_memory_ceiling_covers_the_longest_allowed_call(): + """The cap must bound RAM without ever being able to truncate a legal call.""" + bytes_per_second = 16_000 // 8 + assert MAX_RECORDING_BYTES >= MAX_RECORDING_SECONDS * bytes_per_second + assert MAX_RECORDING_BYTES <= 8 * 1024 * 1024, "must stay small enough for a Pi" diff --git a/drb-edge-node/tests/test_metadata_watcher.py b/drb-edge-node/tests/test_metadata_watcher.py index 9c5a831..3e6f00a 100644 --- a/drb-edge-node/tests/test_metadata_watcher.py +++ b/drb-edge-node/tests/test_metadata_watcher.py @@ -11,7 +11,6 @@ from unittest.mock import AsyncMock, patch from app.config import settings from app.internal.metadata_watcher import ( MetadataWatcher, - TAIL_PAD_SECONDS, OP25_OFFLINE_GRACE, ) from app.internal.op25_client import TerminalUpdate, parse_terminal_messages @@ -229,10 +228,59 @@ async def test_srcaddr_edge_then_idle_timeout_ends_call(watcher, clock): assert payload["end_reason"] == "idle_timeout" # The audio ends at the last transmission plus a short pad, NOT at "now" — # otherwise every recording carries call_idle_timeout seconds of silence. - assert payload["ended_at_epoch"] == pytest.approx(edge_time + TAIL_PAD_SECONDS) + assert payload["ended_at_epoch"] == pytest.approx(edge_time + settings.call_tail_pad_seconds) assert payload["tgid"] == 1234 +@pytest.mark.asyncio +async def test_tail_pad_is_configurable_and_defaults_to_one_second(watcher, clock, monkeypatch): + """ + 0.5s left only ~0.3s of real trailing margin in field measurement and one + recording ended mid-word, so the default moved to 1.0 — and it has to be a + setting, not a magic number, so it can be tuned per node. + """ + assert settings.call_tail_pad_seconds == 1.0 + + monkeypatch.setattr(settings, "call_tail_pad_seconds", 2.5) + + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=555)], + )) + clock.advance(0.5) + edge_time = clock.now + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + + clock.advance(settings.call_idle_timeout + 1.0) + await tick(watcher, update(channels=[channel()])) + + payload = watcher.on_call_end.call_args[0][0] + assert payload["ended_at_epoch"] == pytest.approx(edge_time + 2.5) + + +@pytest.mark.asyncio +async def test_idle_close_logs_the_measured_control_channel_idle(watcher, clock, caplog): + """ + The correct idle timeout can only be tuned from the CONTROL-CHANNEL idle, not + from silence measured in the audio (which also contains the ~1.9s P25 + grant→speech delay). So the real measured value has to reach the log. + """ + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=555)], + )) + clock.advance(0.5) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + + with caplog.at_level("INFO", logger="drb-edge-node"): + clock.advance(settings.call_idle_timeout + 0.25) + await tick(watcher, update(channels=[channel()])) + + idle_lines = [r.message for r in caplog.records if "measured control-channel idle" in r.message] + assert idle_lines, "idle-timeout closes must log the measured idle for later tuning" + assert "3.25s" in idle_lines[0] + + @pytest.mark.asyncio async def test_ongoing_transmission_never_times_out(watcher, clock): await tick(watcher, update(