logan/node-26
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix tail truncation, decouple call length from buffer, trim silence
CI / lint (push) Failing after 4s
Details
CI / test (push) Successful in 21s
Details

Browse Source 
Measured six recordings off a live P25 node and found two independent
defects causing lost audio at the end of calls:

- stop_recording() sliced the ring buffer immediately, so if the MP3
  muxer had not yet delivered the tail the file was silently short.
  Now waits (bounded, 2s) until buffered audio covers the end epoch.
- TGID-change closes used the new grant's epoch as the end with no pad
  at all, guaranteeing truncation on every split. Tail pad is now a
  setting, default raised 0.5s -> 1.0s.

The ring buffer also capped maximum call length: a call longer than the
buffer had its front silently clamped. The ring now serves the pre-roll
only, with a per-call accumulator for the rest, bounded at 4.8MB.
Clamping is loudly warned rather than silent.

Uploads averaged 63% silence, which inflates STT cost and is a known
Whisper hallucination trigger. Leading/trailing silence is now trimmed
conservatively (-40dB, 0.25s guard, internal pauses untouched).
started_at/ended_at still describe the call; new audio_* fields carry
the trimmed audio bounds so playback can map back to wall clock.
All-silence recordings are skipped and logged instead of uploaded.

Also: log measured control-channel idle on idle-timeout closes so
CALL_IDLE_TIMEOUT can be tuned from data, and quiet the httpx logger
which emitted ~170k lines/day of poll noise.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
2026-08-04 21:39:36 -04:00
parent efdbe7d803
commit b0a8ed2a5a
10 changed files with 1118 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
drb-edge-node/app/internal/call_recorder.py
+250 -61
View File
@@ -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: