diff --git a/.env.example b/.env.example index 39e1cbd..196c057 100644 --- a/.env.example +++ b/.env.example @@ -16,12 +16,23 @@ C2_URL=http://localhost:8888 # API key is provisioned automatically via MQTT after admin approves the node # Icecast (local container — usually no need to change) +# Live listening only. Call recording and Discord voice use PulseAudio instead. ICECAST_SOURCE_PASSWORD=hackme ICECAST_ADMIN_PASSWORD=admin ICECAST_HOST=localhost ICECAST_PORT=8000 ICECAST_MOUNT=/radio +# PulseAudio capture (usually no need to change) +# Monitor of the drb_sink null sink that Liquidsoap writes into. +PULSE_SOURCE=drb_sink.monitor +# Seconds to wait for the shared PulseAudio socket before giving up and retrying. +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. +CALL_IDLE_TIMEOUT=3 + # 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 0bcd22e..4c5a1a9 100644 --- a/drb-edge-node/app/config.py +++ b/drb-edge-node/app/config.py @@ -18,12 +18,27 @@ class Settings(BaseSettings): # C2 server (audio upload destination); None disables upload c2_url: Optional[str] = None - # Local Icecast + # Local Icecast — live listening only (frontend / mobile). + # NOT used for call recording or Discord voice: it lags 1s and drifts to 100s+. icecast_host: str = "localhost" icecast_port: int = 8000 icecast_mount: str = "/radio" icecast_source_password: str = "hackme" + # PulseAudio — the low-latency path used for call recording and Discord voice. + # Liquidsoap (op25 container) writes into the `drb_sink` null sink; we capture + # its monitor. Addressed explicitly rather than via "default" because the op25 + # entrypoint starts pulseaudio with -n and never applies system.pa's + # `set-default-source` line. + pulse_source: str = "drb_sink.monitor" + # Bounded wait for the shared PulseAudio socket before launching FFmpeg. + pulse_wait_timeout: float = 30.0 + + # 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. + call_idle_timeout: float = 3.0 + # OP25 container op25_api_url: str = "http://localhost:8001" op25_terminal_url: str = "http://localhost:8081" diff --git a/drb-edge-node/app/internal/call_recorder.py b/drb-edge-node/app/internal/call_recorder.py index 5ba83d4..a199e91 100644 --- a/drb-edge-node/app/internal/call_recorder.py +++ b/drb-edge-node/app/internal/call_recorder.py @@ -1,55 +1,100 @@ +""" +Continuous PulseAudio ring buffer + per-call slicing. + +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. + +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 +stack for frontend/mobile live listening; it is not an accuracy path. + +CLOCK DOMAIN: chunks are stamped with time.time(), the host wall clock. OP25's +call_log timestamps are time.time() from inside the op25 container. All client +containers run network_mode: host and share the host kernel clock, so the two are +the same clock and the pre-roll arithmetic below is a direct subtraction with no +offset mapping. (A wall-clock STEP — e.g. a large NTP correction — would corrupt +at most the calls in flight at that instant; the buffer self-heals within +RING_BUFFER_SECONDS.) +""" import asyncio import time from collections import deque from datetime import datetime, timezone from pathlib import Path -from typing import Optional +from typing import List, Optional, Tuple import httpx from app.config import settings -from app.internal import credentials +from app.internal import credentials, pulse from app.internal.logger import logger -MAX_RECORDING_SECONDS = 600 # safety cap; drop call if it runs this long -PRE_BUFFER_SECONDS = 1.0 # seconds of audio to include before call_start -RING_BUFFER_SECONDS = 60 # how much history to keep when no call is active -READ_CHUNK_BYTES = 4096 # bytes per httpx read +# Safety cap on a single recording; mirrors MAX_SEGMENT_SECONDS in metadata_watcher. +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. +PRE_ROLL_SECONDS = 0.25 + +# Rolling history kept when no call is active. 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. +RING_BUFFER_SECONDS = 30 + +# ~128 ms of audio per chunk at 16 kbps. Chunk size IS the timestamp resolution of +# the ring buffer, so it must stay well under PRE_ROLL_SECONDS — the old 4096-byte +# reads were ~2 s per chunk, which made sub-second slicing meaningless. +READ_CHUNK_BYTES = 256 + +# Encoder settings, matched on purpose to what Liquidsoap already pushes to +# Icecast — %mp3(bitrate=16, samplerate=22050, stereo=false) — so the C2 /upload +# endpoint keeps receiving exactly the kind of MP3 it has always received +# (multipart "audio/mpeg", stored to GCS as .mp3, then fed to Whisper). +# Change both of these together if you ever want higher-fidelity uploads. +MP3_BITRATE = "16k" +MP3_SAMPLE_RATE = "22050" + +# Backoff bounds for restarting a dead capture process. +RESTART_BACKOFF_MIN = 1.0 +RESTART_BACKOFF_MAX = 15.0 class CallRecorder: - """ - Maintains a persistent HTTP connection to the Icecast stream and buffers - the raw MP3 bytes in a ring buffer. When a call starts we note the - monotonic clock; when it ends we slice the buffer and write the file. - - This approach eliminates per-call FFmpeg startup latency, which was - causing empty recordings for calls shorter than ~1–2 s. - """ + """Continuous PulseAudio capture into a ring buffer, sliced per call.""" def __init__(self): self._recordings_dir = Path(settings.recordings_path) - # Ring buffer: deque of (monotonic_time, bytes_chunk) - self._buffer: deque[tuple[float, bytes]] = deque() + # Ring buffer: deque of (wall_clock_epoch_at_arrival, mp3_bytes) + self._buffer: deque[Tuple[float, bytes]] = deque() self._buffer_bytes: int = 0 self._stream_task: Optional[asyncio.Task] = None + self._proc: Optional[asyncio.subprocess.Process] = None + self._capturing: bool = False - # Active call state + # Active recording state self._call_id: Optional[str] = None - self._call_start_mono: Optional[float] = None + self._call_start: Optional[float] = None # OP25 grant epoch + self._slice_start: Optional[float] = None # _call_start - PRE_ROLL_SECONDS # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ async def start(self) -> None: - """Start the persistent stream buffer. Call once from app lifespan.""" - self._stream_task = asyncio.create_task(self._stream_loop()) - logger.info("Stream ring-buffer started.") + """Start the persistent capture. Call once from app lifespan.""" + self._stream_task = asyncio.create_task(self._capture_loop()) + logger.info("PulseAudio ring-buffer starting.") async def stop(self) -> None: - """Cancel the stream reader.""" if self._stream_task: self._stream_task.cancel() try: @@ -57,93 +102,202 @@ class CallRecorder: except asyncio.CancelledError: pass self._stream_task = None + await self._terminate_proc() # ------------------------------------------------------------------ - # Stream reader + # Capture # ------------------------------------------------------------------ - async def _stream_loop(self) -> None: - stream_url = ( - f"http://{settings.icecast_host}:{settings.icecast_port}" - f"{settings.icecast_mount}" - ) - timeout = httpx.Timeout(connect=10.0, read=None, write=10.0, pool=10.0) + def _ffmpeg_command(self) -> List[str]: + return [ + "ffmpeg", + "-hide_banner", "-nostdin", "-nostats", + "-loglevel", "warning", + "-f", "pulse", "-i", settings.pulse_source, + "-ac", "1", + "-ar", MP3_SAMPLE_RATE, + "-b:a", MP3_BITRATE, + # Without this, the MP3 muxer fills its 32 KB AVIO buffer before + # writing anything — 16 s of audio per burst at 16 kbps, which would + # destroy the arrival timestamps the slicing depends on. + "-flush_packets", "1", + "-f", "mp3", "-", + ] + + async def _capture_loop(self) -> None: + backoff = RESTART_BACKOFF_MIN while True: try: - async with httpx.AsyncClient(timeout=timeout) as client: - async with client.stream("GET", stream_url) as response: - response.raise_for_status() - logger.info(f"Stream buffer connected to {stream_url}") - async for chunk in response.aiter_bytes(READ_CHUNK_BYTES): - self._ingest(chunk) + # The original bug this path was abandoned for: FFmpeg was launched + # with -f pulse before the shared socket existed, failed instantly, + # and never recovered. Wait for it, bounded, every time. + if not await pulse.wait_until_ready(): + await asyncio.sleep(backoff) + backoff = min(backoff * 2, RESTART_BACKOFF_MAX) + continue + + await self._run_capture() + logger.warning("PulseAudio capture process exited — restarting.") except asyncio.CancelledError: + await self._terminate_proc() raise except Exception as e: - logger.warning(f"Stream buffer disconnected ({e}) — retrying in 3 s") - await asyncio.sleep(3) + logger.warning(f"PulseAudio capture error ({e}) — restarting.") + + self._capturing = False + await asyncio.sleep(backoff) + backoff = min(backoff * 2, RESTART_BACKOFF_MAX) + + async def _run_capture(self) -> None: + cmd = self._ffmpeg_command() + logger.info(f"Starting capture: ffmpeg -f pulse -i {settings.pulse_source}") + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + self._proc = proc + stderr_task = asyncio.create_task(self._drain_stderr(proc)) + try: + assert proc.stdout is not None + while True: + chunk = await proc.stdout.read(READ_CHUNK_BYTES) + if not chunk: + break # EOF — FFmpeg died or the source went away + if not self._capturing: + self._capturing = True + logger.info("PulseAudio capture is producing audio.") + self._ingest(chunk) + finally: + self._capturing = False + # _drain_stderr swallows its own CancelledError, so it finishes cleanly + # and never needs awaiting here. + stderr_task.cancel() + await self._terminate_proc() + + async def _drain_stderr(self, proc: asyncio.subprocess.Process) -> None: + """Surface FFmpeg's diagnostics instead of letting the pipe fill and block.""" + if proc.stderr is None: + return + try: + while True: + line = await proc.stderr.readline() + if not line: + return + logger.warning(f"ffmpeg(pulse): {line.decode(errors='replace').strip()}") + except asyncio.CancelledError: + return + except Exception: + return + + async def _terminate_proc(self) -> None: + proc, self._proc = self._proc, None + if proc is None or proc.returncode is not None: + return + try: + # Synchronous, so the signal lands even if we are being cancelled and + # the reap below never gets to run. + proc.terminate() + except Exception: + return + try: + await asyncio.wait_for(proc.wait(), timeout=5) + except asyncio.TimeoutError: + try: + proc.kill() + except Exception: + pass + except asyncio.CancelledError: + raise + except Exception: + pass def _ingest(self, chunk: bytes) -> None: - """Append a chunk and trim stale data from the front of the buffer.""" - now = time.monotonic() + """Append a chunk and trim stale audio off the front of the buffer.""" + now = time.time() self._buffer.append((now, chunk)) self._buffer_bytes += len(chunk) - # During a call, never trim data newer than (call_start - pre_buffer). - # Between calls, keep a rolling RING_BUFFER_SECONDS window. - if self._call_start_mono is not None: - keep_from = self._call_start_mono - PRE_BUFFER_SECONDS + # 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 - while self._buffer: - ts, old = self._buffer[0] - if ts >= keep_from: - break - self._buffer.popleft() + while self._buffer and self._buffer[0][0] < keep_from: + _, old = self._buffer.popleft() self._buffer_bytes -= len(old) # ------------------------------------------------------------------ - # Call recording API (same interface as before) + # Recording API # ------------------------------------------------------------------ - async def start_recording(self, call_id: str) -> bool: + async def start_recording(self, call_id: str, start_epoch: Optional[float] = None) -> bool: + """ + Open a recording. `start_epoch` is OP25's call_log timestamp (host wall + 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("Recording already active — ignoring start.") + logger.warning(f"Recording already active ({self._call_id}) — ignoring start for {call_id}.") return False + self._call_id = call_id - self._call_start_mono = time.monotonic() - logger.info(f"Recording started (ring-buffer): {call_id}") + self._call_start = start_epoch if start_epoch else time.time() + self._slice_start = self._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.") + + oldest = self._buffer[0][0] if self._buffer else None + if oldest is not None and self._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. + 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." + ) + + logger.info(f"Recording started: {call_id} (slice from {self._slice_start:.3f})") return True - async def stop_recording(self) -> Optional[Path]: + 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: return None call_id = self._call_id - call_start = self._call_start_mono + call_start = self._call_start + slice_start = self._slice_start self._call_id = None - self._call_start_mono = None + self._call_start = None + self._slice_start = None - # Slice: everything from (call_start - pre_buffer) to now - cutoff = (call_start - PRE_BUFFER_SECONDS) if call_start else 0.0 - chunks = [chunk for ts, chunk in self._buffer if ts >= cutoff] - - # Safety cap: if the call ran very long, truncate to MAX_RECORDING_SECONDS + end = end_epoch if end_epoch else time.time() if call_start is not None: - cap_cutoff = call_start + MAX_RECORDING_SECONDS - now = time.monotonic() - if now > cap_cutoff: - # Approximate: trim chunks that arrived after the cap - cap_keep_until = call_start + MAX_RECORDING_SECONDS - chunks = [ - chunk for ts, chunk in self._buffer - if cutoff <= ts <= cap_keep_until - ] + end = min(end, call_start + MAX_RECORDING_SECONDS) + if slice_start is None: + return None + + chunks: List[bytes] = [] + for ts, chunk in self._buffer: + if ts < slice_start: + continue + chunks.append(chunk) + if ts >= end: + # Include the chunk straddling `end` so the tail is never clipped, + # then stop. + break if not chunks: logger.warning( - f"No buffered audio for call {call_id} — " - "stream may not have been connected yet." + f"No buffered audio for call {call_id} " + f"(window {slice_start:.3f}–{end:.3f}) — PulseAudio capture may be down." ) return None @@ -151,12 +305,11 @@ class CallRecorder: ts_str = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") output_path = self._recordings_dir / f"{ts_str}_{call_id}.mp3" - data = b"".join(chunks) - output_path.write_bytes(data) + 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)") + logger.info(f"Recording saved: {output_path.name} ({size} bytes, {end - slice_start:.2f}s window)") return output_path output_path.unlink(missing_ok=True) @@ -213,9 +366,24 @@ class CallRecorder: except Exception: pass + # ------------------------------------------------------------------ + # State + # ------------------------------------------------------------------ + @property def is_recording(self) -> bool: return self._call_id is not None + @property + def is_capturing(self) -> bool: + """True when FFmpeg is alive and audio is actually arriving.""" + return self._capturing + + @property + def buffered_seconds(self) -> float: + if len(self._buffer) < 2: + return 0.0 + return self._buffer[-1][0] - self._buffer[0][0] + call_recorder = CallRecorder() diff --git a/drb-edge-node/app/internal/discord_radio.py b/drb-edge-node/app/internal/discord_radio.py index a8d58f3..0c87945 100644 --- a/drb-edge-node/app/internal/discord_radio.py +++ b/drb-edge-node/app/internal/discord_radio.py @@ -2,11 +2,14 @@ import asyncio from typing import Optional import discord from discord.ext import commands +from app.config import settings +from app.internal import pulse from app.internal.logger import logger BOT_READY_TIMEOUT = 15 # seconds to wait for Discord bot to become ready WATCHDOG_INTERVAL = 30 # seconds between voice-connection health checks REJOIN_DELAY = 5 # seconds to wait before attempting a rejoin +STREAM_RETRY_DELAY = 5 # seconds to back off before re-arming the audio source class RadioBot: @@ -47,6 +50,10 @@ class RadioBot: # Remember where we are so the watchdog can rejoin if we drop self._guild_id = guild_id self._channel_id = channel_id + # Bounded wait for the shared PulseAudio socket. Historically FFmpeg was + # launched before the op25 container had created it, failed instantly, + # and the bot sat silently connected forever. + await pulse.wait_until_ready() self._play_stream() if system_name: await self._bot.change_presence( @@ -108,19 +115,48 @@ class RadioBot: self._ready_event = None def _play_stream(self): + """ + Feed Discord voice straight from the PulseAudio monitor. + + Icecast is NOT used here: it lags ~1 s at connect and drifts to 100 s+, + which is unusable for live listening. Icecast remains the frontend/mobile + listening path only. + """ if not self._voice_client: return - from app.config import settings - stream_url = f"http://{settings.icecast_host}:{settings.icecast_port}{settings.icecast_mount}" + + if not pulse.is_ready(): + logger.error( + f"PulseAudio socket {pulse.socket_path()} missing — " + f"cannot start Discord audio; retrying in {STREAM_RETRY_DELAY}s." + ) + self._schedule_restart() + return + + # before_options land ahead of -i, so this becomes: + # ffmpeg -f pulse -i drb_sink.monitor … source = discord.FFmpegPCMAudio( - stream_url, - before_options="-reconnect 1 -reconnect_streamed 1 -reconnect_delay_max 5", + settings.pulse_source, + before_options="-f pulse", ) self._voice_client.play( discord.PCMVolumeTransformer(source, volume=1.0), after=self._on_stream_end, ) + def _schedule_restart(self, delay: float = STREAM_RETRY_DELAY): + """Re-arm the audio source after a delay — safe to call from any thread.""" + if not self._loop: + return + + async def _delayed_restart(): + await asyncio.sleep(delay) + vc = self._voice_client + if vc and vc.is_connected() and not vc.is_playing(): + self._play_stream() + + self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart())) + def _on_stream_end(self, error): if error: logger.error(f"Stream ended with error: {error}") @@ -128,12 +164,9 @@ class RadioBot: if not (self._loop and vc and vc.is_connected() and not vc.is_playing()): return if error: - # Back off before retrying — prevents tight loop when PulseAudio is unavailable - async def _delayed_restart(): - await asyncio.sleep(5) - if self._voice_client and self._voice_client.is_connected() and not self._voice_client.is_playing(): - self._play_stream() - self._loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_delayed_restart())) + # Back off before retrying — prevents a tight loop when PulseAudio is + # unavailable (FFmpeg exits immediately in that case). + self._schedule_restart() else: self._loop.call_soon_threadsafe(self._play_stream) @@ -232,6 +265,9 @@ class RadioBot: else: self._voice_client = await vc.connect() self._channel_id = vc.id + # A fresh connect() has no audio source attached yet. + if not self._voice_client.is_playing(): + self._play_stream() await message.reply(f"Joined {vc.name}.") except Exception as e: logger.error(f"joinme failed: {e}") diff --git a/drb-edge-node/app/internal/metadata_watcher.py b/drb-edge-node/app/internal/metadata_watcher.py index f932c9e..dad1cff 100644 --- a/drb-edge-node/app/internal/metadata_watcher.py +++ b/drb-edge-node/app/internal/metadata_watcher.py @@ -1,38 +1,125 @@ +""" +Event-driven call state machine. + +Replaces the old hang-counter inference (which derived call start from "a tgid +appeared in channel_update" and call end from N polls of silence) with the two +authoritative signals OP25 actually exposes: + + START — a `call_log` entry. OP25 appends one at channel-grant time stamped with + its own time.time(). This is an exact start timestamp, not the moment + our poll happened to notice, so recordings can be sliced back to it. + + END — the `srcaddr` != 0 → `srcaddr` == 0 transition in `channel_update`. + OP25 never reports call termination externally: internally it ends a + call on the P25 Terminator Data Unit (duid15) or 3 voice-framing + timeouts, but neither becomes a log entry. What *is* observable is that + `srcaddr`/`svcopts` reset to 0/false the instant the call ends, while + `tgid`/`hold_tgid` keep showing the just-ended talkgroup for + TGID_HOLD_TIME (2 s). So the srcaddr edge is a real state change, not a + timeout heuristic. + +SEGMENTS: one emitted call (= one recording, one Firestore doc) spans a whole +conversation, not a single transmission. It stays open across repeated grants on +the same talkgroup and closes when the talkgroup changes or the radio goes quiet +for settings.call_idle_timeout seconds. + +CLOCKS: `call_log["time"]` is time.time() inside the op25 container. All three +client containers run network_mode: host and share the host kernel clock, so that +value is directly comparable to time.time() here — no offset mapping needed. The +call recorder's ring buffer is stamped with the same clock for the same reason. +""" import asyncio +import time import uuid from datetime import datetime, timezone -from typing import Optional, Callable, Awaitable +from typing import Optional, Callable, Awaitable, Any, List, Dict + +from app.config import settings from app.internal.op25_client import op25_client from app.internal.logger import logger CallbackFn = Callable[[dict], Awaitable[None]] -HANG_THRESHOLD = 2 # polls before declaring a call ended (0.5s poll → 1s hang time) -POLL_INTERVAL = 0.5 # seconds +# 500 ms. Do NOT lower: start precision already comes from OP25's own timestamp, +# and http_server.py's request handler has a ~200 ms blocking floor anyway. +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 _as_int(value: Any) -> Optional[int]: + """Coerce an OP25 field to a positive int, or None. Rejects 0/""/"None".""" + if value is None: + return None + try: + number = int(value) + except (TypeError, ValueError): + return None + return number if number > 0 else None + + +def _as_float(value: Any) -> Optional[float]: + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _iso(epoch: Optional[float]) -> Optional[str]: + if epoch is None: + return None + return datetime.fromtimestamp(epoch, timezone.utc).isoformat() class MetadataWatcher: def __init__(self): self._running = False + + # Open segment state + self._active_call_id: Optional[str] = None self._current_tgid: Optional[int] = None self._current_tgid_name: Optional[str] = None - self._hang_counter: int = 0 - self._active_call_id: Optional[str] = None - self._call_started_at: Optional[datetime] = None + self._current_freq: Any = None + self._current_srcaddr: Optional[int] = None + self._started_at: Optional[float] = None # OP25 epoch of the first grant + self._transmissions: int = 0 + + # Transmission tracking within the open segment + self._tx_active: bool = False # last poll saw srcaddr != 0 + self._last_activity: float = 0.0 # epoch of last evidence of traffic + self._last_tx_end: Optional[float] = None # epoch of the srcaddr 1→0 edge + self._last_ok_poll: float = 0.0 + + # Injectable for tests; production is always the host wall clock. + self._clock: Callable[[], float] = time.time # Set these before calling start() self.on_call_start: Optional[CallbackFn] = None self.on_call_end: Optional[CallbackFn] = None + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + async def start(self): self._running = True + self._last_ok_poll = self._clock() asyncio.create_task(self._poll_loop()) - logger.info("Metadata watcher started.") + logger.info("Metadata watcher started (call_log driven).") async def stop(self): self._running = False if self._active_call_id: - await self._end_call() + await self._close_segment(self._clock(), reason="shutdown") async def _poll_loop(self): while self._running: @@ -42,79 +129,218 @@ class MetadataWatcher: logger.warning(f"Metadata poll error: {e}") await asyncio.sleep(POLL_INTERVAL) - async def _tick(self): - status = await op25_client.get_terminal_status() + # ------------------------------------------------------------------ + # One poll + # ------------------------------------------------------------------ - if not status: - # OP25 not responding — hang-out any active call - if self._active_call_id: - self._hang_counter += 1 - if self._hang_counter >= HANG_THRESHOLD: - await self._end_call() + async def _tick(self): + now = self._clock() + update = await op25_client.poll_terminal() + + if update is None: + # OP25 unreachable. Don't kill an open segment on a single blip. + if self._active_call_id and (now - self._last_ok_poll) >= OP25_OFFLINE_GRACE: + await self._close_segment(now, reason="op25_unreachable") return - # OP25 terminal returns either a list of channels or a single dict - channels = status if isinstance(status, list) else [status] - active_tgid: Optional[int] = None - active_meta: dict = {} + self._last_ok_poll = now - for ch in channels: - tgid = ch.get("tgid") or ch.get("tg_id") - if tgid and str(tgid) not in ("0", "", "None"): - active_tgid = int(tgid) - active_meta = ch - break + # 1. call_log first — these are the authoritative starts, and processing + # them before the channel scan means a same-poll grant+state pair is + # already attributed to the new segment by the time we scan channels. + # Sorted defensively: multi-receiver setups append per receiver. + for entry in sorted(update.call_log, key=lambda e: _as_float(e.get("time")) or 0.0): + await self._handle_call_log(entry, now) - if active_tgid: - self._hang_counter = 0 - if self._current_tgid != active_tgid: - # Talkgroup changed — close previous call and open a new one - if self._active_call_id: - await self._end_call() - self._current_tgid = active_tgid - await self._start_call(active_tgid, active_meta) - else: - # No active talkgroup - if self._active_call_id: - self._hang_counter += 1 - if self._hang_counter >= HANG_THRESHOLD: - await self._end_call() + # 2. channel_update — the only external end signal. + await self._handle_channels(update.channels, now) - async def _start_call(self, tgid: int, meta: dict): + async def _handle_call_log(self, entry: Dict[str, Any], now: float) -> None: + tgid = _as_int(entry.get("tgid")) + if tgid is None: + return # a grant with no talkgroup is nothing we can record or label + + # OP25's own stamp. Fall back to now only if the field is missing/garbage. + started_at = _as_float(entry.get("time")) + if started_at is None: + logger.warning(f"call_log entry for tgid={tgid} has no usable time — using local clock.") + started_at = now + + if self._active_call_id is None: + await self._open_segment(entry, tgid, started_at, now) + return + + if tgid == self._current_tgid: + # CONTINUE: same talkgroup, keep one recording so the back-and-forth + # of a single conversation lands in one file. + self._transmissions += 1 + self._tx_active = True + self._last_tx_end = None + self._last_activity = now + self._refresh_meta_from_log(entry) + return + + # SPLIT: different talkgroup. The new grant's OP25 timestamp is the most + # precise end available for the outgoing segment — the new call's audio + # starts exactly there, so no tail pad. + await self._close_segment(started_at, reason="tgid_change") + await self._open_segment(entry, tgid, started_at, now) + + async def _handle_channels(self, channels: List[Dict[str, Any]], now: float) -> None: + if self._active_call_id is None: + return + + tx_active = False + foreign_active_tgid: Optional[int] = None + + for channel in channels: + srcaddr = _as_int(channel.get("srcaddr")) + chan_tgid = _as_int(channel.get("tgid")) + if srcaddr is None: + continue + if chan_tgid == self._current_tgid: + tx_active = True + self._current_srcaddr = srcaddr + self._refresh_meta_from_channel(channel) + elif chan_tgid is not None: + foreign_active_tgid = chan_tgid + + if tx_active: + self._tx_active = True + self._last_tx_end = None + self._last_activity = now + elif self._tx_active: + # The srcaddr != 0 → 0 edge: OP25 has torn the call down. + self._tx_active = False + self._last_tx_end = now + self._last_activity = now + + # Safety net for a dropped call_log event (deque is capped at 10): the one + # receiver we have is plainly on another talkgroup, so our segment is over + # even though we never saw its grant. Close now rather than record + # call_idle_timeout seconds of the wrong tgid. + # + # Restricted to single-receiver setups on purpose: with several receivers, + # another channel being busy says nothing about ours, and closing on it + # would truncate every call whenever a second receiver is active. + if not tx_active and foreign_active_tgid is not None and len(channels) == 1: + logger.warning( + f"tgid {foreign_active_tgid} active without a call_log entry — " + f"closing segment for tgid {self._current_tgid} (call_log event likely dropped)." + ) + await self._close_segment(now, reason="tgid_change_unlogged") + return + + if (now - self._last_activity) >= settings.call_idle_timeout: + # 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 + await self._close_segment(min(end, now), reason="idle_timeout") + return + + if self._started_at is not None and (now - self._started_at) >= MAX_SEGMENT_SECONDS: + logger.warning(f"Segment for tgid {self._current_tgid} hit the {MAX_SEGMENT_SECONDS}s cap — closing.") + await self._close_segment(now, reason="max_length") + + # ------------------------------------------------------------------ + # Segment open / close + # ------------------------------------------------------------------ + + def _refresh_meta_from_log(self, entry: Dict[str, Any]) -> None: + if not self._current_tgid_name: + self._current_tgid_name = entry.get("tgtag") or "" + if entry.get("freq"): + self._current_freq = entry.get("freq") + rid = _as_int(entry.get("rid")) + if rid is not None: + self._current_srcaddr = rid + + def _refresh_meta_from_channel(self, channel: Dict[str, Any]) -> None: + if not self._current_tgid_name: + self._current_tgid_name = channel.get("tag") or "" + if not self._current_freq and channel.get("freq"): + self._current_freq = channel.get("freq") + + async def _open_segment(self, entry: Dict[str, Any], tgid: int, started_at: float, now: float) -> None: self._active_call_id = str(uuid.uuid4()) - self._call_started_at = datetime.now(timezone.utc) - self._current_tgid_name = meta.get("tag") or meta.get("tgid_tag") or "" + self._current_tgid = tgid + self._current_tgid_name = entry.get("tgtag") or "" + self._current_freq = entry.get("freq") + self._current_srcaddr = _as_int(entry.get("rid")) + self._started_at = started_at + self._transmissions = 1 + + # Assume the transmission is still up: we learn otherwise from the next + # channel scan. A grant whose call already ended before we polled simply + # closes on the very next tick via the idle timeout. + self._tx_active = True + self._last_tx_end = None + self._last_activity = now + payload = { "call_id": self._active_call_id, "tgid": tgid, "tgid_name": self._current_tgid_name, - "freq": meta.get("freq"), - "srcaddr": meta.get("srcaddr"), - "started_at": self._call_started_at.isoformat(), + "freq": self._current_freq, + "srcaddr": self._current_srcaddr, + "started_at": _iso(started_at), + # Raw epoch for the recorder's ring-buffer slice — same clock domain. + "started_at_epoch": started_at, } - logger.info(f"Call start: tgid={tgid} id={self._active_call_id}") + logger.info( + f"Call start: tgid={tgid} id={self._active_call_id} " + f"(op25 t={started_at:.3f}, detected {now - started_at:+.2f}s later)" + ) if self.on_call_start: await self.on_call_start(payload) - async def _end_call(self): + async def _close_segment(self, end_epoch: float, reason: str) -> None: if not self._active_call_id: return + + started_at = self._started_at + if started_at is not None: + end_epoch = max(end_epoch, started_at) + payload = { "call_id": self._active_call_id, "tgid": self._current_tgid, "tgid_name": self._current_tgid_name or "", - "started_at": self._call_started_at.isoformat() if self._call_started_at else None, - "ended_at": datetime.now(timezone.utc).isoformat(), + "freq": self._current_freq, + "srcaddr": self._current_srcaddr, + "started_at": _iso(started_at), + "started_at_epoch": started_at, + "ended_at": _iso(end_epoch), + "ended_at_epoch": end_epoch, + "transmissions": self._transmissions, + "end_reason": reason, } - logger.info(f"Call end: id={self._active_call_id}") + duration = (end_epoch - started_at) if started_at is not None else 0.0 + logger.info( + f"Call end: id={self._active_call_id} tgid={self._current_tgid} " + f"reason={reason} transmissions={self._transmissions} duration={duration:.2f}s" + ) + + # Clear state before awaiting so a re-entrant tick can't see a half-closed + # segment (and so an immediately-following _open_segment is clean). self._active_call_id = None self._current_tgid = None self._current_tgid_name = None - self._hang_counter = 0 - self._call_started_at = None + self._current_freq = None + self._current_srcaddr = None + self._started_at = None + self._transmissions = 0 + self._tx_active = False + self._last_tx_end = None + if self.on_call_end: await self.on_call_end(payload) + # ------------------------------------------------------------------ + # Public state (consumed by routers/api.py, main.py and the dashboards) + # ------------------------------------------------------------------ + @property def active_call_id(self) -> Optional[str]: return self._active_call_id diff --git a/drb-edge-node/app/internal/op25_client.py b/drb-edge-node/app/internal/op25_client.py index 8aae869..68f7026 100644 --- a/drb-edge-node/app/internal/op25_client.py +++ b/drb-edge-node/app/internal/op25_client.py @@ -1,8 +1,34 @@ import httpx -from typing import Optional, Dict, Any +from dataclasses import dataclass, field +from typing import Optional, Dict, Any, List from app.config import settings from app.internal.logger import logger +# The OP25 HTTP terminal answers a single "update" command with a LIST of +# messages, each tagged with a `json_type`. We care about two of them: +# +# channel_update — current receiver state. `channels` holds the channel ids and +# each id is also a top-level key holding that channel's dict +# (freq/tgid/tag/srcaddr/svcopts/hold_tgid/…). +# +# call_log — an EVENT QUEUE, not a snapshot. `log` holds entries appended +# by tk_p25.log_call() at channel-grant time, each stamped with +# OP25's own time.time(). get_call_log() DRAINS the deque, so +# every entry is delivered exactly once and a missed poll loses +# it forever. The deque is capped at CALL_LOG_MAX_LEN = 10, so +# the consumer must keep up. +# +# Everything else (trunk_update, rx_update, terminal_config, …) is ignored. +TERMINAL_UPDATE_COMMAND = [{"command": "update", "arg1": 0, "arg2": 0}] + + +@dataclass +class TerminalUpdate: + """One decoded poll of the OP25 HTTP terminal.""" + + channels: List[Dict[str, Any]] = field(default_factory=list) + call_log: List[Dict[str, Any]] = field(default_factory=list) + class OP25Client: def __init__(self): @@ -49,24 +75,67 @@ class OP25Client: logger.error(f"OP25 generate-config failed: {e}") return False - async def get_terminal_status(self) -> Optional[Any]: - """Poll the OP25 HTTP terminal for current call metadata.""" + async def poll_terminal(self) -> Optional[TerminalUpdate]: + """ + Poll the OP25 HTTP terminal once and decode every message we understand. + + Returns None only when OP25 is unreachable / returned garbage — callers + use that to distinguish "no traffic" from "no OP25". + """ try: async with httpx.AsyncClient(timeout=3) as client: - r = await client.post( - self.terminal_url, - json=[{"command": "update", "arg1": 0, "arg2": 0}], - ) + r = await client.post(self.terminal_url, json=TERMINAL_UPDATE_COMMAND) r.raise_for_status() - messages = r.json() - for msg in messages: - if msg.get("json_type") == "channel_update": - channels = msg.get("channels", []) - if channels: - return msg.get(str(channels[0]), {}) - return None + return parse_terminal_messages(r.json()) except Exception: return None + async def get_terminal_status(self) -> Optional[Dict[str, Any]]: + """ + Compatibility shim: the first channel's state dict, as this used to return. + + Prefer poll_terminal() — this discards the call_log, which is the only + source of exact call-start timestamps. + """ + update = await self.poll_terminal() + if not update or not update.channels: + return None + return update.channels[0] + + +def parse_terminal_messages(messages: Any) -> TerminalUpdate: + """ + Decode an OP25 terminal response into channel state + call-log events. + + Deliberately permissive: the response may be a bare dict instead of a list, + may contain json_type values we have never seen, and individual entries may + be malformed. Anything unrecognised is skipped rather than raising, because + dropping a whole poll would drop call_log events that are never re-sent. + """ + update = TerminalUpdate() + + if isinstance(messages, dict): + messages = [messages] + if not isinstance(messages, list): + return update + + for msg in messages: + if not isinstance(msg, dict): + continue + json_type = msg.get("json_type") + + if json_type == "channel_update": + for chan_id in msg.get("channels") or []: + channel = msg.get(str(chan_id)) + if isinstance(channel, dict): + update.channels.append(channel) + + elif json_type == "call_log": + for entry in msg.get("log") or []: + if isinstance(entry, dict): + update.call_log.append(entry) + + return update + op25_client = OP25Client() diff --git a/drb-edge-node/app/internal/pulse.py b/drb-edge-node/app/internal/pulse.py new file mode 100644 index 0000000..37b3d3d --- /dev/null +++ b/drb-edge-node/app/internal/pulse.py @@ -0,0 +1,76 @@ +""" +PulseAudio readiness helpers. + +The PulseAudio daemon lives in the `op25` container and exposes its native +socket on the shared `pulse_socket` docker volume (mounted at /run/pulse in +both containers, with PULSE_SERVER=unix:/run/pulse/native). + +`op25-container/docker-entrypoint.sh` waits up to ~10 s for that socket before +starting its own app, but the edge-node historically had *no* equivalent wait: +FFmpeg would be launched with `-f pulse` before the socket existed, fail +instantly, and the audio path would stay dead for the lifetime of the process. +This module is the missing wait. + +NOTE on the source name: the op25 entrypoint starts pulseaudio with `-n`, which +skips /etc/pulse/system.pa entirely and loads modules from the command line +instead. That means the `set-default-source drb_sink.monitor` line in system.pa +is NOT applied at runtime, so `-i default` is unreliable. Always address the +monitor explicitly via settings.pulse_source (default "drb_sink.monitor"). +""" +import asyncio +import os +import stat +from typing import Optional + +from app.config import settings +from app.internal.logger import logger + +DEFAULT_SOCKET_PATH = "/run/pulse/native" +POLL_INTERVAL = 0.5 + + +def socket_path() -> str: + """Resolve the PulseAudio socket path from PULSE_SERVER (`unix:/path` form).""" + server = os.environ.get("PULSE_SERVER", "") + if server.startswith("unix:"): + candidate = server[len("unix:"):].strip() + if candidate: + return candidate + return DEFAULT_SOCKET_PATH + + +def is_ready() -> bool: + """True when the PulseAudio native socket exists and really is a socket.""" + try: + return stat.S_ISSOCK(os.stat(socket_path()).st_mode) + except OSError: + return False + + +async def wait_until_ready(timeout: Optional[float] = None) -> bool: + """ + Block until the PulseAudio socket appears, or `timeout` seconds elapse. + + Bounded on purpose — never hang the caller forever. Returns True if the + socket is present, False on timeout (caller decides whether to retry). + """ + limit = settings.pulse_wait_timeout if timeout is None else timeout + path = socket_path() + + if is_ready(): + return True + + logger.info(f"Waiting up to {limit:.0f}s for PulseAudio socket at {path}…") + waited = 0.0 + while waited < limit: + await asyncio.sleep(POLL_INTERVAL) + waited += POLL_INTERVAL + if is_ready(): + logger.info(f"PulseAudio socket ready after {waited:.1f}s.") + return True + + logger.error( + f"PulseAudio socket {path} not present after {limit:.0f}s — " + "is the op25 container running? Audio capture will retry." + ) + return False diff --git a/drb-edge-node/app/main.py b/drb-edge-node/app/main.py index 96310e5..5f4f181 100644 --- a/drb-edge-node/app/main.py +++ b/drb-edge-node/app/main.py @@ -24,12 +24,18 @@ async def on_call_start(data: dict): radio_bot.start_stream() await mqtt_manager.publish_status("recording") await mqtt_manager.publish_metadata("call_start", data) - await call_recorder.start_recording(data["call_id"]) + # started_at_epoch is OP25's own call_log timestamp — the recorder slices the + # ring buffer back to it (minus pre-roll), so however late we detected the + # grant, the audio still starts in the right place. + await call_recorder.start_recording( + data["call_id"], + start_epoch=data.get("started_at_epoch"), + ) async def on_call_end(data: dict): radio_bot.stop_stream() - file_path = await call_recorder.stop_recording() + file_path = await call_recorder.stop_recording(end_epoch=data.get("ended_at_epoch")) if file_path: node_cfg = load_node_config() audio_url = await call_recorder.upload_recording( @@ -46,7 +52,8 @@ async def on_call_end(data: dict): else: logger.warning( f"No recording file generated for call {data['call_id']} " - "— call may have been too short or Icecast unreachable." + "— PulseAudio capture may be down (check the op25 container and " + f"the {settings.pulse_source} source)." ) await mqtt_manager.publish_metadata("call_end", data) await mqtt_manager.publish_status("online") @@ -191,7 +198,7 @@ async def lifespan(app: FastAPI): # Start services (radio_bot starts on-demand when a discord_join command arrives) await mqtt_manager.connect() await metadata_watcher.start() - await call_recorder.start() # persistent Icecast stream buffer + await call_recorder.start() # persistent PulseAudio ring buffer # Start system caching in background from app.internal.system_cacher import fetch_and_cache_systems diff --git a/drb-edge-node/app/routers/api.py b/drb-edge-node/app/routers/api.py index deba5ac..fa9a740 100644 --- a/drb-edge-node/app/routers/api.py +++ b/drb-edge-node/app/routers/api.py @@ -48,6 +48,10 @@ async def get_status(): "assigned_system_id": node_cfg.assigned_system_id, "system_name": system_name, "is_recording": call_recorder.is_recording, + # Health of the PulseAudio capture that feeds every recording — the single + # most useful signal when recordings come back empty. + "audio_capture": call_recorder.is_capturing, + "buffered_seconds": round(call_recorder.buffered_seconds, 1), "active_tgid": active_tgid, "active_tgid_name": active_tgid_name, "active_call_id": metadata_watcher.active_call_id, diff --git a/drb-edge-node/tests/test_call_recorder.py b/drb-edge-node/tests/test_call_recorder.py new file mode 100644 index 0000000..febd2dc --- /dev/null +++ b/drb-edge-node/tests/test_call_recorder.py @@ -0,0 +1,203 @@ +""" +Unit tests for the CallRecorder ring buffer and per-call slicing. + +No FFmpeg and no PulseAudio: the buffer is filled directly with timestamped +chunks, which is exactly what _ingest() produces at runtime. +""" +import time +from unittest.mock import patch + +import pytest + +from app.internal.call_recorder import ( + CallRecorder, + MAX_RECORDING_SECONDS, + PRE_ROLL_SECONDS, + RING_BUFFER_SECONDS, +) + +T0 = 1_700_000_000.0 +CHUNK_INTERVAL = 0.1 # seconds of audio per synthetic chunk + + +@pytest.fixture +def recorder(tmp_path): + 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).""" + ts = start + index = 0 + while ts < end: + recorder._buffer.append((ts, marker + str(index).encode() + b";")) + index += 1 + ts = round(ts + CHUNK_INTERVAL, 6) + + +def timestamps(recorder): + return [ts for ts, _ in recorder._buffer] + + +# --------------------------------------------------------------------------- +# Ring buffer trimming +# --------------------------------------------------------------------------- + +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) + + 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) + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + + # 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 min(timestamps(recorder)) <= (T0 + 1.0) - PRE_ROLL_SECONDS + CHUNK_INTERVAL + + +# --------------------------------------------------------------------------- +# Pre-roll and slicing +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_slice_starts_pre_roll_before_the_op25_timestamp(recorder): + fill(recorder, T0, T0 + 10.0) + 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) + + path = await recorder.stop_recording(end_epoch=grant_time + 2.0) + assert path is not None and 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 + + # 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 + # start — erring early is the safe direction, erring late loses speech. + assert first_ts - CHUNK_INTERVAL <= grant_time - PRE_ROLL_SECONDS + 1e-6 + # ...and no more than one chunk of extra pre-roll is dragged in. + assert first_ts >= grant_time - PRE_ROLL_SECONDS - 1e-6 + + +@pytest.mark.asyncio +async def test_tail_chunk_straddling_the_end_is_included(recorder): + fill(recorder, T0, T0 + 10.0) + 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 + + 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 + + await recorder.start_recording("call-1", start_epoch=T0) # 5 s before the head + path = await recorder.stop_recording(end_epoch=T0 + 8.0) + + 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" + + +@pytest.mark.asyncio +async def test_no_buffered_audio_returns_none(recorder): + await recorder.start_recording("call-1", start_epoch=T0) + assert await recorder.stop_recording(end_epoch=T0 + 2.0) is None + + +@pytest.mark.asyncio +async def test_start_epoch_omitted_falls_back_to_now(recorder): + now = time.time() + 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) + + +@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) + + 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 + + assert last_ts <= T0 + 1.0 + MAX_RECORDING_SECONDS + CHUNK_INTERVAL + + +# --------------------------------------------------------------------------- +# Recording lifecycle +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_second_start_is_rejected_while_recording(recorder): + fill(recorder, T0, T0 + 5.0) + assert await recorder.start_recording("call-1", start_epoch=T0 + 1.0) is True + assert await recorder.start_recording("call-2", start_epoch=T0 + 2.0) is False + assert recorder.is_recording + + +@pytest.mark.asyncio +async def test_stop_without_start_is_a_noop(recorder): + assert await recorder.stop_recording() is None + assert not recorder.is_recording + + +@pytest.mark.asyncio +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) + split = T0 + 5.0 + + await recorder.start_recording("call-1", start_epoch=T0 + 1.0) + first = await recorder.stop_recording(end_epoch=split) + + 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 + + +# --------------------------------------------------------------------------- +# FFmpeg invocation +# --------------------------------------------------------------------------- + +def test_ffmpeg_command_reads_pulse_and_flushes_packets(recorder): + cmd = recorder._ffmpeg_command() + joined = " ".join(cmd) + + assert "-f pulse" in joined + assert "drb_sink.monitor" in joined, "must address the monitor explicitly, not 'default'" + # Without -flush_packets the mp3 muxer buffers 32 KB (~16 s at 16 kbps) before + # 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" diff --git a/drb-edge-node/tests/test_metadata_watcher.py b/drb-edge-node/tests/test_metadata_watcher.py index faff8d4..9c5a831 100644 --- a/drb-edge-node/tests/test_metadata_watcher.py +++ b/drb-edge-node/tests/test_metadata_watcher.py @@ -1,190 +1,487 @@ """ -Unit tests for MetadataWatcher state machine. -All OP25 HTTP calls are mocked — no running services required. +Unit tests for the event-driven MetadataWatcher state machine. + +Call START comes from OP25 `call_log` entries (stamped with OP25's own +time.time()); call END comes from the srcaddr != 0 -> srcaddr == 0 transition in +`channel_update`. All OP25 HTTP calls are mocked — no running services required. """ import pytest from unittest.mock import AsyncMock, patch -from app.internal.metadata_watcher import MetadataWatcher, HANG_THRESHOLD + +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 + + +class FakeClock: + """Manually advanced clock so idle timeouts are testable without sleeping.""" + + def __init__(self, start: float = 1_700_000_000.0): + self.now = start + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> float: + self.now += seconds + return self.now @pytest.fixture -def watcher(): +def clock(): + return FakeClock() + + +@pytest.fixture +def watcher(clock): w = MetadataWatcher() + w._clock = clock w.on_call_start = AsyncMock() w.on_call_end = AsyncMock() return w +def grant(tgid: int, time_: float, tgtag: str = "", rid: int = 101, freq: int = 851_000_000): + """One OP25 call_log entry (see tk_p25.log_call).""" + return { + "time": time_, + "sysid": 1, + "rcvr": 0, + "freq": freq, + "slot": None, + "prio": 0, + "tgid": tgid, + "tgtag": tgtag, + "rid": rid, + "rtag": "", + } + + +def channel(tgid: int = 0, srcaddr: int = 0, tag: str = "", hold_tgid: int = 0): + """One OP25 channel_update channel dict (see tk_p25.get_chan_status).""" + return { + "freq": 851_000_000, + "tgid": tgid or None, + "tag": tag, + "srcaddr": srcaddr, + "svcopts": bool(srcaddr), + "hold_tgid": hold_tgid or None, + "encrypted": 0, + "emergency": 0, + } + + +def update(call_log=None, channels=None) -> TerminalUpdate: + return TerminalUpdate(channels=list(channels or []), call_log=list(call_log or [])) + + +def patched(result): + return patch( + "app.internal.metadata_watcher.op25_client.poll_terminal", + new=AsyncMock(return_value=result), + ) + + +async def tick(watcher, result): + with patched(result): + await watcher._tick() + + # --------------------------------------------------------------------------- -# Call start +# op25_client message parsing +# --------------------------------------------------------------------------- + +def test_parser_extracts_call_log_and_channels(): + messages = [ + {"json_type": "trunk_update", "0": {"whatever": 1}}, + {"json_type": "channel_update", "channels": [0], "0": channel(tgid=1234, srcaddr=555)}, + {"json_type": "call_log", "log": [grant(1234, 100.0)]}, + ] + result = parse_terminal_messages(messages) + + assert len(result.channels) == 1 + assert result.channels[0]["tgid"] == 1234 + assert len(result.call_log) == 1 + assert result.call_log[0]["time"] == 100.0 + + +def test_parser_tolerates_garbage(): + """Unknown json_types, bare dicts and malformed entries must not raise.""" + assert parse_terminal_messages(None).call_log == [] + assert parse_terminal_messages("nope").channels == [] + assert parse_terminal_messages([None, 5, {"json_type": "mystery"}]).channels == [] + + # A bare dict instead of a list. + single = parse_terminal_messages({"json_type": "call_log", "log": [grant(1, 1.0), "junk"]}) + assert len(single.call_log) == 1 + + # channel_update naming a channel that isn't present. + missing = parse_terminal_messages([{"json_type": "channel_update", "channels": [0, 1], "0": channel(9)}]) + assert len(missing.channels) == 1 + + +def test_parser_handles_multiple_channels(): + messages = [{ + "json_type": "channel_update", + "channels": [0, 1], + "0": channel(tgid=1111, srcaddr=1), + "1": channel(tgid=2222, srcaddr=2), + }] + result = parse_terminal_messages(messages) + assert [c["tgid"] for c in result.channels] == [1111, 2222] + + +# --------------------------------------------------------------------------- +# Call start — driven by call_log # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_call_starts_when_tgid_appears(watcher): - status = [{"tgid": 1234, "tag": "Police Dispatch"}] - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)): - await watcher._tick() +async def test_call_log_entry_starts_call(watcher, clock): + grant_time = clock.now - 0.4 # OP25 logged it before our poll noticed + await tick(watcher, update( + call_log=[grant(1234, grant_time, tgtag="Police Dispatch")], + channels=[channel(tgid=1234, srcaddr=555)], + )) assert watcher.is_active assert watcher.current_tgid == 1234 watcher.on_call_start.assert_called_once() + payload = watcher.on_call_start.call_args[0][0] assert payload["tgid"] == 1234 assert payload["tgid_name"] == "Police Dispatch" - assert "call_id" in payload - assert "started_at" in payload + assert payload["call_id"] + # The whole point: the start is OP25's timestamp, not our detection time. + assert payload["started_at_epoch"] == grant_time + assert payload["started_at"].startswith("20") @pytest.mark.asyncio -async def test_tgid_zero_does_not_start_call(watcher): - status = [{"tgid": 0}] - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)): - await watcher._tick() +async def test_channel_activity_alone_does_not_start_a_call(watcher): + """No call_log entry => no call, even with a live tgid on the channel.""" + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)])) assert not watcher.is_active watcher.on_call_start.assert_not_called() @pytest.mark.asyncio -async def test_tgid_none_string_does_not_start_call(watcher): - status = [{"tgid": "None"}] - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)): - await watcher._tick() +async def test_call_log_entry_without_tgid_is_ignored(watcher): + await tick(watcher, update(call_log=[grant(0, 100.0)])) assert not watcher.is_active + watcher.on_call_start.assert_not_called() @pytest.mark.asyncio -async def test_op25_offline_does_not_start_call(watcher): - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=None)): - await watcher._tick() +async def test_call_log_without_time_falls_back_to_local_clock(watcher, clock): + entry = grant(1234, 0.0) + entry.pop("time") + await tick(watcher, update(call_log=[entry])) + + assert watcher.is_active + assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == clock.now + + +@pytest.mark.asyncio +async def test_op25_unreachable_does_not_start_call(watcher): + await tick(watcher, None) assert not watcher.is_active watcher.on_call_start.assert_not_called() # --------------------------------------------------------------------------- -# Hang / call end +# Call end — srcaddr edge + idle timeout # --------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_hang_below_threshold_keeps_call_alive(watcher): - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])): - await watcher._tick() - +async def test_srcaddr_edge_then_idle_timeout_ends_call(watcher, clock): + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=555)], + )) assert watcher.is_active - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 0}])): - for _ in range(HANG_THRESHOLD - 1): - await watcher._tick() + # srcaddr resets to 0 while tgid is still held — OP25's end-of-call signal. + clock.advance(0.5) + edge_time = clock.now + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + assert watcher.is_active, "the srcaddr edge alone must not close the segment" + watcher.on_call_end.assert_not_called() + # Still quiet, but not long enough yet. + clock.advance(settings.call_idle_timeout - 0.5) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) assert watcher.is_active + + # Past the timeout. + clock.advance(1.0) + await tick(watcher, update(channels=[channel()])) + assert not watcher.is_active + + watcher.on_call_end.assert_called_once() + payload = watcher.on_call_end.call_args[0][0] + 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["tgid"] == 1234 + + +@pytest.mark.asyncio +async def test_ongoing_transmission_never_times_out(watcher, clock): + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=555)], + )) + + for _ in range(20): + clock.advance(settings.call_idle_timeout) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=555)])) + assert watcher.is_active + watcher.on_call_end.assert_not_called() @pytest.mark.asyncio -async def test_hang_at_threshold_ends_call(watcher): - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])): - await watcher._tick() +async def test_op25_unreachable_ends_active_call_after_grace(watcher, clock): + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=555)], + )) - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 0}])): - for _ in range(HANG_THRESHOLD): - await watcher._tick() + # A single failed poll must not kill the call. + clock.advance(OP25_OFFLINE_GRACE / 2) + await tick(watcher, None) + assert watcher.is_active + + clock.advance(OP25_OFFLINE_GRACE) + await tick(watcher, None) + assert not watcher.is_active + assert watcher.on_call_end.call_args[0][0]["end_reason"] == "op25_unreachable" + + +@pytest.mark.asyncio +async def test_grant_whose_call_already_ended(watcher, clock): + """call_log delivers a start whose transmission is already over.""" + await tick(watcher, update( + call_log=[grant(1234, clock.now - 1.2)], + channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)], # already idle + )) + assert watcher.is_active + + clock.advance(settings.call_idle_timeout + 0.5) + await tick(watcher, update(channels=[channel()])) assert not watcher.is_active watcher.on_call_end.assert_called_once() + + +@pytest.mark.asyncio +async def test_stop_closes_open_segment(watcher, clock): + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=555)], + )) + await watcher.stop() + + assert not watcher.is_active + assert watcher.on_call_end.call_args[0][0]["end_reason"] == "shutdown" + + +# --------------------------------------------------------------------------- +# Segmentation: same-TGID continuation, different-TGID split +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_same_tgid_grant_continues_one_recording(watcher, clock): + """Back-and-forth on one talkgroup must stay a single call/recording.""" + start = clock.now + await tick(watcher, update( + call_log=[grant(1234, start)], + channels=[channel(tgid=1234, srcaddr=555)], + )) + call_id = watcher.active_call_id + + # Transmission ends... + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + + # ...and the other party keys up on the SAME tgid inside the idle window. + clock.advance(1.0) + await tick(watcher, update( + call_log=[grant(1234, clock.now)], + channels=[channel(tgid=1234, srcaddr=777)], + )) + + assert watcher.is_active + assert watcher.active_call_id == call_id, "same tgid must not open a new call" + watcher.on_call_start.assert_called_once() + watcher.on_call_end.assert_not_called() + + # And the whole exchange closes as one segment. + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=1234, srcaddr=0, hold_tgid=1234)])) + clock.advance(settings.call_idle_timeout + 0.5) + await tick(watcher, update(channels=[channel()])) + + watcher.on_call_end.assert_called_once() payload = watcher.on_call_end.call_args[0][0] - assert "call_id" in payload - assert "ended_at" in payload + assert payload["call_id"] == call_id + assert payload["transmissions"] == 2 + assert payload["started_at_epoch"] == start @pytest.mark.asyncio -async def test_op25_offline_triggers_hang_and_ends_call(watcher): - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])): - await watcher._tick() +async def test_different_tgid_grant_splits_recording(watcher, clock): + await tick(watcher, update( + call_log=[grant(1111, clock.now, tgtag="Fire")], + channels=[channel(tgid=1111, srcaddr=1)], + )) + first_id = watcher.active_call_id - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=None)): - for _ in range(HANG_THRESHOLD): - await watcher._tick() - - assert not watcher.is_active - watcher.on_call_end.assert_called_once() - - -@pytest.mark.asyncio -async def test_hang_counter_resets_when_tgid_returns(watcher): - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])): - await watcher._tick() - - # Partial hang — not enough to end - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 0}])): - for _ in range(HANG_THRESHOLD - 1): - await watcher._tick() - - assert watcher.is_active - - # tgid returns — counter resets - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1234}])): - await watcher._tick() - - assert watcher._hang_counter == 0 - assert watcher.is_active - - -# --------------------------------------------------------------------------- -# Talkgroup changes -# --------------------------------------------------------------------------- - -@pytest.mark.asyncio -async def test_talkgroup_change_closes_old_and_opens_new(watcher): - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 1111}])): - await watcher._tick() - - first_call_id = watcher.active_call_id - - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=[{"tgid": 2222}])): - await watcher._tick() + clock.advance(2.0) + split_time = clock.now + await tick(watcher, update( + call_log=[grant(2222, split_time, tgtag="EMS")], + channels=[channel(tgid=2222, srcaddr=2)], + )) assert watcher.is_active assert watcher.current_tgid == 2222 - assert watcher.active_call_id != first_call_id - watcher.on_call_end.assert_called_once() + assert watcher.active_call_id != first_id assert watcher.on_call_start.call_count == 2 + watcher.on_call_end.assert_called_once() - -# --------------------------------------------------------------------------- -# Status format variations -# --------------------------------------------------------------------------- - -@pytest.mark.asyncio -async def test_single_dict_status_instead_of_list(watcher): - """OP25 terminal may return a bare dict instead of a list.""" - status = {"tgid": 9999, "tag": "Fire"} - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)): - await watcher._tick() - - assert watcher.current_tgid == 9999 + ended = watcher.on_call_end.call_args[0][0] + assert ended["call_id"] == first_id + assert ended["tgid"] == 1111 + assert ended["end_reason"] == "tgid_change" + # The outgoing segment ends exactly where the new one begins — no tail pad, + # or it would swallow the first moments of the new talkgroup. + assert ended["ended_at_epoch"] == split_time + assert watcher.on_call_start.call_args[0][0]["started_at_epoch"] == split_time @pytest.mark.asyncio -async def test_tg_id_key_alias(watcher): - """Some OP25 builds use 'tg_id' instead of 'tgid'.""" - status = [{"tg_id": 5555, "tag": "EMS"}] - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)): - await watcher._tick() +async def test_multiple_call_log_entries_in_one_poll(watcher, clock): + """ + The call_log deque is drained per poll and capped at 10, so one poll can + deliver several grants. Same-tgid ones merge, different-tgid ones split. + """ + t0 = clock.now - 1.5 + await tick(watcher, update( + call_log=[ + grant(1111, t0), + grant(1111, t0 + 0.4), # same tgid -> continuation + grant(2222, t0 + 0.9), # different tgid -> split + ], + channels=[channel(tgid=2222, srcaddr=9)], + )) - assert watcher.current_tgid == 5555 + assert watcher.current_tgid == 2222 + assert watcher.on_call_start.call_count == 2 + watcher.on_call_end.assert_called_once() + + ended = watcher.on_call_end.call_args[0][0] + assert ended["tgid"] == 1111 + assert ended["transmissions"] == 2 + assert ended["started_at_epoch"] == t0 + assert ended["ended_at_epoch"] == t0 + 0.9 @pytest.mark.asyncio -async def test_multichannel_uses_first_active(watcher): - """When multiple channels are returned, first active tgid wins.""" - status = [ - {"tgid": 0}, - {"tgid": 7777, "tag": "Roads"}, - {"tgid": 8888, "tag": "Other"}, - ] - with patch("app.internal.metadata_watcher.op25_client.get_terminal_status", new=AsyncMock(return_value=status)): - await watcher._tick() +async def test_out_of_order_call_log_entries_are_sorted(watcher, clock): + t0 = clock.now - 2.0 + await tick(watcher, update( + call_log=[grant(2222, t0 + 1.0), grant(1111, t0)], + channels=[channel(tgid=2222, srcaddr=9)], + )) - assert watcher.current_tgid == 7777 + ended = watcher.on_call_end.call_args[0][0] + assert ended["tgid"] == 1111 + assert watcher.current_tgid == 2222 + + +@pytest.mark.asyncio +async def test_dropped_call_log_event_still_closes_segment(watcher, clock): + """ + CALL_LOG_MAX_LEN is 10, so a slow consumer loses grants. If another talkgroup + is plainly transmitting we must close immediately rather than record it under + the wrong tgid for call_idle_timeout seconds. + """ + await tick(watcher, update( + call_log=[grant(1111, clock.now)], + channels=[channel(tgid=1111, srcaddr=1)], + )) + first_id = watcher.active_call_id + + clock.advance(1.0) + await tick(watcher, update(channels=[channel(tgid=3333, srcaddr=7)])) # no call_log + + assert not watcher.is_active + ended = watcher.on_call_end.call_args[0][0] + assert ended["call_id"] == first_id + assert ended["end_reason"] == "tgid_change_unlogged" + + +@pytest.mark.asyncio +async def test_other_receiver_on_another_tgid_does_not_split(watcher, clock): + """ + The dropped-call_log fallback must not fire on multi-receiver setups: another + receiver being busy says nothing about ours, and closing on it would truncate + every call. + """ + await tick(watcher, update( + call_log=[grant(1111, clock.now)], + channels=[channel(tgid=1111, srcaddr=1)], + )) + first_id = watcher.active_call_id + + clock.advance(0.5) + await tick(watcher, update(channels=[ + channel(tgid=1111, srcaddr=0, hold_tgid=1111), + channel(tgid=3333, srcaddr=7), + ])) + + assert watcher.is_active + assert watcher.active_call_id == first_id + watcher.on_call_end.assert_not_called() + + +@pytest.mark.asyncio +async def test_second_receiver_activity_does_not_hold_segment_open(watcher, clock): + """Only channels on OUR talkgroup count as our transmission.""" + await tick(watcher, update( + call_log=[grant(1111, clock.now)], + channels=[channel(tgid=1111, srcaddr=1)], + )) + + clock.advance(0.5) + await tick(watcher, update(channels=[ + channel(tgid=1111, srcaddr=0, hold_tgid=1111), + channel(tgid=1111, srcaddr=0), + ])) + assert watcher.is_active + + clock.advance(settings.call_idle_timeout + 0.5) + await tick(watcher, update(channels=[channel(tgid=1111, srcaddr=0, hold_tgid=1111)])) + assert not watcher.is_active + + +@pytest.mark.asyncio +async def test_end_never_precedes_start(watcher, clock): + """A clock skew or odd grant order must never produce a negative duration.""" + await tick(watcher, update( + call_log=[grant(1234, clock.now + 5.0)], # OP25 stamp in the future + channels=[channel(tgid=1234, srcaddr=5)], + )) + await watcher.stop() + + payload = watcher.on_call_end.call_args[0][0] + assert payload["ended_at_epoch"] >= payload["started_at_epoch"]