"""Maintain a labelled last-good video that survives the live application going away. The source URL is read from ``EMERGENCY_VIDEO_SOURCE`` so signed URLs and tokens do not enter shell history. Each successful capture is decoded, normalized to the emergency programme's exact H.264/AAC shape, visibly labelled ``EARLIER PICTURES``, verified, and atomically installed. A failed source, partial write, broken decoder, or application outage leaves the previous known-good file untouched. This is deliberately a cache, not a live claim. It is suitable behind the technical card in the OBS ``Emergency last video`` scene. Repeat it with separate output paths for additional priority channels; the priority set belongs in event configuration, never in this module. """ from __future__ import annotations import argparse import json import os import subprocess import sys import time import urllib.request from collections.abc import Callable from datetime import UTC, datetime from pathlib import Path from emergency_stream import ( BITRATE_K, FPS, HEIGHT, WIDTH, EmergencyStreamError, _filter_path, default_font, verify_media, ) DEFAULT_CAPTURE_S = 30 DEFAULT_INTERVAL_S = 60 DEFAULT_EVENTS_URL = "http://127.0.0.1:8766/events" MAX_SSE_LINE_BYTES = 6 * 1024 * 1024 def source_from_env(env: dict[str, str] | os._Environ[str] = os.environ) -> str: source = (env.get("EMERGENCY_VIDEO_SOURCE") or "").strip() if not source: raise EmergencyStreamError("set EMERGENCY_VIDEO_SOURCE to the priority channel's HLS/MP4 URL") if not source.lower().startswith("https://"): raise EmergencyStreamError("the emergency video source must use https://") return source def _slot_index(slot: str) -> int: value = slot.strip().upper() if len(value) == 1 and "A" <= value <= "Z": return ord(value) - ord("A") try: index = int(value) - 1 except ValueError as e: raise EmergencyStreamError("priority slot must be A-Z or a positive number") from e if index < 0: raise EmergencyStreamError("priority slot must be A-Z or a positive number") return index def priority_source_from_init(doc: dict, slot: str) -> str: """Resolve a stable camera position to the newest signed video in a producer init header.""" meta = doc.get("meta") or {} if meta.get("idle"): raise EmergencyStreamError("producer is waiting for a race; no priority video is available") slots = meta.get("cameraSlots") or [] index = _slot_index(slot) if index >= len(slots) or not slots[index]: raise EmergencyStreamError(f"priority camera slot {slot.upper()} is not assigned in this race") boat = str(slots[index]) rows = (doc.get("media") or {}).get(boat) or [] candidates = [row for row in rows if row.get("video") and str(row.get("file") or "").startswith("https://")] if not candidates: raise EmergencyStreamError(f"priority camera slot {slot.upper()} has no complete video yet") newest = max(candidates, key=lambda row: ( int(row.get("capturedAtMs") or 0), float(row.get("tSec") or 0), str(row.get("file") or "") )) return str(newest["file"]) def priority_source_from_events(events_url: str, slot: str, *, timeout_s: int = 20, opener: Callable = urllib.request.urlopen) -> str: """Read one private loopback SSE init event and resolve the requested priority position. The returned signed URL is kept only in memory and passed directly to FFmpeg. It is never written to the health file or logs. A race handover naturally produces a new init header and mapping. """ if not events_url.lower().startswith("https://"): raise EmergencyStreamError("the producer events URL must use https://") try: with opener(events_url, timeout=timeout_s) as response: event = "" for raw in response: if len(raw) > MAX_SSE_LINE_BYTES: raise EmergencyStreamError("producer init header exceeds the safety limit") line = raw.decode("utf-8").rstrip("\r\n") if line.startswith("event: "): event = line[7:] elif event == "init" and line.startswith("data: "): return priority_source_from_init(json.loads(line[6:]), slot) except EmergencyStreamError: raise except (OSError, UnicodeError, json.JSONDecodeError) as e: raise EmergencyStreamError(f"cannot read the priority channel from the producer: {e}") from e raise EmergencyStreamError("producer closed before publishing an init header") def capture_cmd(*, source: str, output: Path, label: str, captured: datetime, font: Path, ffmpeg: str = "ffmpeg", duration_s: int = DEFAULT_CAPTURE_S, bitrate_k: int = BITRATE_K) -> list[str]: """Build one finite capture command with silent audio and an inseparable stale-picture label.""" if duration_s < 4: raise ValueError("capture must be at least four seconds") safe_label = " ".join(label.replace("'", "").replace(":", " ").split())[:40] or "VIDEO" stamp = captured.astimezone(UTC).strftime("%Y-%m-%d %H\\:%M UTC") face = _filter_path(font) video_filter = ( f"scale={WIDTH}:{HEIGHT}:force_original_aspect_ratio=decrease," f"pad={WIDTH}:{HEIGHT}:(ow-iw)/2:(oh-ih)/2:color=0x071522,fps={FPS}," f"drawbox=x=0:y=0:w=iw:h=92:color=0x071522@0.88:t=fill," f"drawtext=fontfile='{face}':text='EARLIER PICTURES | {safe_label}':" "fontcolor=white:fontsize=36:x=42:y=24," f"drawtext=fontfile='{face}':text='{stamp}':fontcolor=white@0.82:fontsize=28:" "x=w-text_w-42:y=30,format=yuv420p" ) return [ ffmpeg, "-hide_banner", "-y", "-nostdin", "-loglevel", "warning", "-rw_timeout", "15000000", "-i", source, "-f", "lavfi", "-i", "anullsrc=r=48000:cl=stereo", "-map", "0:v:0", "-map", "1:a:0", "-t", str(duration_s), "-vf", video_filter, "-c:v", "libx264", "-preset", "veryfast", "-profile:v", "high", "-level:v", "4.1", "-r", str(FPS), "-g", str(FPS * 2), "-keyint_min", str(FPS * 2), "-sc_threshold", "0", "-b:v", f"{bitrate_k}k", "-maxrate", f"{bitrate_k}k", "-bufsize", f"{bitrate_k * 2}k", "-c:a", "aac", "-b:a", "128k", "-ar", "48000", "-ac", "2", "-movflags", "+faststart", str(output), ] def capture_once(*, output: Path, label: str, ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe", duration_s: int = DEFAULT_CAPTURE_S, now: datetime | None = None, source: str | None = None) -> dict: """Install only a complete, decodable capture; preserve the old file on every failure.""" source = source or source_from_env() output = output.resolve() output.parent.mkdir(parents=True, exist_ok=True) candidate = output.with_name(f".{output.stem}.candidate{output.suffix}") captured = now or datetime.now(UTC) try: subprocess.run( capture_cmd(source=source, output=candidate, label=label, captured=captured, font=default_font(), ffmpeg=ffmpeg, duration_s=duration_s), check=True, ) details = verify_media(candidate, ffprobe=ffprobe) os.replace(candidate, output) return { **details, "path": str(output), "bytes": output.stat().st_size, "capturedUtc": captured.astimezone(UTC).isoformat(), "label": label, } except subprocess.CalledProcessError as e: raise EmergencyStreamError(f"could not capture a complete priority video: {e}") from e finally: candidate.unlink(missing_ok=True) def _write_health(path: Path | None, doc: dict) -> None: if path is None: return path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(doc, sort_keys=True) + "\n", encoding="utf-8") os.replace(temporary, path) def run(*, output: Path, label: str, ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe", duration_s: int = DEFAULT_CAPTURE_S, interval_s: int = DEFAULT_INTERVAL_S, once: bool = False, source_resolver: Callable[[], str] | None = None, health_path: Path | None = None) -> int: """Best-effort refresh loop: failure is reported but never removes the last successful clip.""" while True: started = time.monotonic() try: source = source_resolver() if source_resolver else source_from_env() details = capture_once(output=output, label=label, ffmpeg=ffmpeg, ffprobe=ffprobe, duration_s=duration_s, source=source) _write_health(health_path, { "status": "cached", "updatedUtcMs": int(time.time() * 1000), "capturedUtc": details["capturedUtc"], "bytes": details["bytes"], "label": label, }) print(f"[video-cache] installed {details['capturedUtc']} {details['bytes']} bytes", flush=True) if once: return 0 except EmergencyStreamError as e: _write_health(health_path, { "status": "waiting", "updatedUtcMs": int(time.time() * 1000), "reason": str(e), "lastGoodPresent": output.is_file(), "label": label, }) print(f"[video-cache] refresh failed; previous file preserved: {e}", file=sys.stderr, flush=True) if once: return 2 wait = max(1.0, interval_s - (time.monotonic() - started)) time.sleep(wait) def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--label", required=True, help="stable camera/channel label, not a team assignment") parser.add_argument("--ffmpeg", default="ffmpeg") parser.add_argument("--ffprobe", default="ffprobe") parser.add_argument("--events", help="HTTPS producer SSE endpoint for a dynamic priority slot") parser.add_argument("--slot", help="stable priority camera position (A-Z or 1-based number)") parser.add_argument("--health", type=Path, help="write redacted cache status atomically here") parser.add_argument("--duration", type=int, default=DEFAULT_CAPTURE_S) parser.add_argument("--interval", type=int, default=DEFAULT_INTERVAL_S) parser.add_argument("--once", action="store_true") args = parser.parse_args() try: if bool(args.events) != bool(args.slot): raise EmergencyStreamError("--events and --slot must be supplied together") resolver = ((lambda: priority_source_from_events(args.events, args.slot)) if args.events and args.slot else None) return run(output=args.output, label=args.label, ffmpeg=args.ffmpeg, ffprobe=args.ffprobe, duration_s=args.duration, interval_s=args.interval, once=args.once, source_resolver=resolver, health_path=args.health) except (ValueError, EmergencyStreamError) as e: print(f"emergency video cache: {e}", file=sys.stderr) return 2 if __name__ == "__main__": raise SystemExit(main())