"""Take a replay: cut it out of today's recording, put it on air, and give the programme back. Runs **on the machine OBS runs on** — the A4000 — because that is where the recording is. The producer is somewhere else entirely (`vr-producer` in the office, or the cloud), and this must keep working when the link to it does not: the cut is a local file operation and the actuation is obs-websocket on loopback. What the producer contributes is *suggestions*, and losing them costs the operator a list, not the button. BROADCAST_ALLOW_CONTROL=1 VR_BROADCAST_CONTROL_TOKEN=... python -m v3.broadcast.live.replay_take \ --anchor D:/trwc/recording_anchor.jsonl --replay-dir D:/vr/replay --feed http://localhost:8765 A take is a sequence and deliberately nothing more: cut (day_cutter) -> verify -> atomic replace -> mute programme -> scene "Replay" -> wait the clip out -> scene "Broadcast 3D" -> unmute [finally] THE MUTE IS THE ONLY PART THAT CAN HURT. OBS mutes an INPUT, not a scene item, so the programme browser source keeps talking underneath the clip unless something silences it — and the clip carries its own commentary, the voice that described the moment, which is the whole reason a replay sounds right. Two commentators over one picture is the failure. So the programme is muted for the take and restored in a `finally`; and because a `finally` does not run when a process is killed, **the module also unmutes unconditionally on start-up**. A broadcast left silent is the most expensive thing that can go wrong here, and it must not need a person to notice it. WHAT THIS DOES NOT DO. It does not decide. Cues from the producer are listed for the operator and never acted on: the picture changes because somebody pressed something. That is a deliberate limit and not a stage on the way to automation — when `obs_bridge.py` lands (`docs/plans/ai-picture-switcher.md` phase 3) the replay take belongs there as one of ITS options, with the lease and receipt machinery that stops two things cutting OBS at once. A second automatic actuator beside it is exactly what that plan exists to prevent, so this one stays manual until it can be folded in. """ from __future__ import annotations import argparse import contextlib import html import http.server import json import os import socketserver import sys import threading import time import urllib.request from dataclasses import asdict, dataclass, field from pathlib import Path _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) import day_cutter # noqa: E402 from broadcast_control import control_enabled # noqa: E402 from day_cutter import DayCutterError # noqa: E402 from director_obs import OBSController # noqa: E402 from recording_anchor import read_samples # noqa: E402 #: The scenes and the source, as `obs/stream_1080.json` names them. Not configurable on a whim: the #: scene file is the layout's authority, and a follower that invented its own names would be the #: second opinion that file exists to prevent. REPLAY_SCENE = "Replay" PROGRAMME_SCENE = "Broadcast 3D" PROGRAMME_INPUT = "programme" #: A replay of "the last N seconds", when the operator has no particular moment in mind. DEFAULT_LOOKBACK_S = 40.0 #: Around a cue: enough before the moment to see it develop, enough after to see what it did. DEFAULT_LEAD_S = 25.0 DEFAULT_TRAIL_S = 15.0 #: How far behind the wall clock the recording's write frontier is assumed to be. The encoder holds a #: little, and asking for the literal current instant returns a clip that ends before it starts being #: a clip. Small, because every second of margin is a second of the moment the operator just saw. WRITE_MARGIN_MS = 2_000 #: Cues kept for the operator. Bounded because this list is fed from the network. MAX_CUES = 40 #: After the clip's own duration, before cutting back. `clear_on_media_end` on the scene already lands #: on the programme bed if we are late, so this is about not cutting away a frame EARLY. RETURN_PAD_S = 0.3 class ReplayTakeError(RuntimeError): """A take that could not be made, said so an operator can act on it.""" @dataclass class Cue: """A moment the producer thinks is worth a replay. A SUGGESTION — nothing here acts on one.""" id: str kind: str label: str utcMs: int race: str = "" @dataclass class TakeState: """What the last (or current) take is doing, for the operator's page.""" status: str = "idle" # idle | rolling | failed detail: str = "" label: str = "" fromUtcMs: int = 0 toUtcMs: int = 0 durationS: float = 0.0 startedUtcMs: int = 0 pieces: list = field(default_factory=list) class CueLog: """The newest cues, oldest dropped. Thread-safe: the feed reader writes, the page reads.""" def __init__(self, *, limit: int = MAX_CUES): self._lock = threading.Lock() self._rows: list[Cue] = [] self._limit = limit def add(self, cue: Cue) -> None: with self._lock: if any(row.id == cue.id for row in self._rows): return self._rows.append(cue) del self._rows[:-self._limit] def recent(self) -> list[Cue]: with self._lock: return list(reversed(self._rows)) def get(self, cue_id: str) -> Cue | None: with self._lock: return next((row for row in self._rows if row.id == cue_id), None) class ReplayTake: """The take itself. OBS and the cutter are injected so the sequence is testable without either.""" def __init__(self, *, anchor: Path, replay_dir: Path, obs: OBSController | None = None, cutter=day_cutter.cut, # noqa: ANN001 now_ms=None, sleep=time.sleep, # noqa: ANN001 ffmpeg: str = "ffmpeg", ffprobe: str = "ffprobe", on_log=print): # noqa: ANN001 self.anchor = Path(anchor) self.output = Path(replay_dir) / "replay.mp4" self._obs = obs self._cutter = cutter self._now_ms = now_ms or (lambda: int(time.time() * 1000)) self._sleep = sleep self._ffmpeg, self._ffprobe = ffmpeg, ffprobe self._log = on_log self._lock = threading.Lock() # one take at a time; a second is refused self.state = TakeState() # --- the audio obligation --------------------------------------------------------------------- def restore_audio(self) -> bool: """Unmute the programme, whatever happened last time. Called at start-up. A `finally` does not run when a process is killed, and the state it would have restored lives in OBS rather than here — so a crash mid-take leaves the broadcast SILENT with nothing to say why. This is the only self-healing in the module and it is unconditional on purpose: unmuting something already unmuted costs nothing, and the alternative failure is a silent championship. """ try: obs = self._connect() if obs.input_muted(PROGRAMME_INPUT): self._log(f"[replay] {PROGRAMME_INPUT} was left muted — restoring the programme audio") obs.set_input_muted(PROGRAMME_INPUT, False) return True except Exception as e: # noqa: BLE001 — OBS not up yet is a normal start self._log(f"[replay] could not check the programme audio ({type(e).__name__}: {e})") return False def _connect(self) -> OBSController: if self._obs is None: self._obs = OBSController(host=os.environ.get("OBS_WS_HOST", "localhost"), port=int(os.environ.get("OBS_WS_PORT", "4455")), password=os.environ.get("OBS_WS_PASSWORD", "")) self._obs.connect() return self._obs # --- the window -------------------------------------------------------------------------------- def window_for_cue(self, cue: Cue, *, lead_s: float = DEFAULT_LEAD_S, trail_s: float = DEFAULT_TRAIL_S) -> tuple[int, int]: """Around the cue's own instant, which is already PRESENTATION time (see `cue_kind`'s callers): the producer adds the broadcast delay before publishing, because the recording holds the picture as it went out, not the water as it happened.""" return int(cue.utcMs - lead_s * 1000), int(cue.utcMs + trail_s * 1000) def window_for_lookback(self, seconds: float) -> tuple[int, int]: """The last `seconds` of what has actually been written to disk.""" end = self._now_ms() - WRITE_MARGIN_MS return int(end - seconds * 1000), int(end) # --- the take ---------------------------------------------------------------------------------- def take(self, *, from_utc_ms: int, to_utc_ms: int, label: str = "") -> TakeState: """Cut, install, roll, and give the programme back. Blocks for the length of the clip.""" if not self._lock.acquire(blocking=False): raise ReplayTakeError("a replay is already rolling") try: self.state = TakeState(status="rolling", label=label, fromUtcMs=from_utc_ms, toUtcMs=to_utc_ms, startedUtcMs=self._now_ms()) try: details = self._cutter(read_samples(self.anchor), from_utc_ms, to_utc_ms, output=self.output, ffmpeg=self._ffmpeg, ffprobe=self._ffprobe) except DayCutterError as e: self.state = TakeState(status="failed", detail=str(e), label=label, fromUtcMs=from_utc_ms, toUtcMs=to_utc_ms) raise ReplayTakeError(str(e)) from e self.state.durationS = float(details.get("durationS") or 0.0) self.state.pieces = list(details.get("pieces") or []) self._roll(self.state.durationS) self.state.status = "idle" self.state.detail = "returned to programme" return self.state finally: self._lock.release() def _roll(self, duration_s: float) -> None: """Put it on air and take it off again. The programme is restored on every path out.""" obs = self._connect() obs.set_input_muted(PROGRAMME_INPUT, True) try: obs.set_scene(REPLAY_SCENE) self._sleep(max(0.0, duration_s) + RETURN_PAD_S) finally: # Order matters: the picture first, then the voice. Unmuting while the clip is still up # would put the live commentary over the replayed picture for as long as the scene change # takes — briefly, but that is precisely the pairing this module exists to avoid. with contextlib.suppress(Exception): obs.set_scene(PROGRAMME_SCENE) obs.set_input_muted(PROGRAMME_INPUT, False) class CueReader(threading.Thread): """Read `replay_cue` off the producer's SSE and file them. Never raises, never acts. The producer may be unreachable for the whole day — the office link, a restart, a cloud producer behind a tunnel — and the operator's button must not depend on it. So every failure is a log line and a retry, and the only thing lost is the list of suggestions. """ def __init__(self, base: str, cues: CueLog, *, token: str = "", on_log=print, # noqa: ANN001 retry_s: float = 10.0, open_url=None): # noqa: ANN001 super().__init__(daemon=True, name="replay-cues") self._url = base.rstrip("/") + "/events" + (f"?token={token}" if token else "") self._cues, self._log, self._retry_s = cues, on_log, retry_s self._open = open_url or (lambda url: urllib.request.urlopen(url, timeout=30)) # noqa: S310 self._stop = threading.Event() def stop(self) -> None: self._stop.set() def run(self) -> None: while not self._stop.is_set(): try: self._consume() except Exception as e: # noqa: BLE001 — a producer that is not there self._log(f"[replay] no cues from the producer ({type(e).__name__}: {e})") self._stop.wait(self._retry_s) def _consume(self) -> None: with self._open(self._url) as stream: event = "" for raw in stream: if self._stop.is_set(): return line = raw.decode("utf-8", "replace").rstrip("\r\n") if line.startswith("event:"): event = line[6:].strip() elif line.startswith("data:") and event == "replay_cue": self._file(line[5:].strip()) elif not line: event = "" def _file(self, payload: str) -> None: try: doc = json.loads(payload) except json.JSONDecodeError: return if not isinstance(doc, dict) or not doc.get("id") or doc.get("utcMs") is None: return self._cues.add(Cue(id=str(doc["id"])[:64], kind=str(doc.get("kind") or "")[:32], label=str(doc.get("label") or "")[:120], utcMs=int(doc["utcMs"]), race=str(doc.get("race") or "")[:64])) # --- the operator's surface ------------------------------------------------------------------------- def _refusal() -> dict | None: """The same two locks the producer's control routes use, refusing the same two mistakes apart. Not the feed token: being allowed to READ the broadcast is not permission to change the picture. """ if not control_enabled(): return {"code": 503, "error": "broadcast control is not enabled on this box", "hint": "BROADCAST_ALLOW_CONTROL"} if not (os.environ.get("VR_BROADCAST_CONTROL_TOKEN") or "").strip(): return {"code": 503, "error": "no control token is configured", "hint": "VR_BROADCAST_CONTROL_TOKEN"} return None def _authorised(headers) -> bool: # noqa: ANN001 want = (os.environ.get("VR_BROADCAST_CONTROL_TOKEN") or "").strip() auth = headers.get("Authorization") or "" got = auth[7:].strip() if auth.lower().startswith("bearer ") else "" return bool(got) and bool(want) and got == want PAGE = """Replay

Replay

Cut from today’s recording. The picture changes when you press something.

""" class _Handler(http.server.BaseHTTPRequestHandler): server_version = "ViewRegattaReplay/1" def log_message(self, fmt, *args): # noqa: ANN001 pass # the take logs; the socket need not def _json(self, status: int, body: dict) -> None: raw = json.dumps(body).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(raw))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(raw) def _gate(self) -> bool: refusal = _refusal() if refusal is not None: self._json(refusal.pop("code"), refusal) return False if not _authorised(self.headers): self._json(401, {"error": "control token required"}) return False return True def do_GET(self) -> None: # noqa: N802 path = self.path.split("?", 1)[0].rstrip("/") or "/" if path == "/": body = PAGE.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return if path == "/cues": if not self._gate(): return self._json(200, {"cues": [asdict(c) for c in self.server.cues.recent()], "take": asdict(self.server.take.state)}) return self._json(404, {"error": "no such route"}) def do_POST(self) -> None: # noqa: N802 if self.path.split("?", 1)[0].rstrip("/") != "/take": return self._json(404, {"error": "no such route"}) if not self._gate(): return None try: length = int(self.headers.get("Content-Length") or 0) body = json.loads(self.rfile.read(length) or b"{}") except (ValueError, json.JSONDecodeError): return self._json(400, {"error": "JSON object required"}) if not isinstance(body, dict): return self._json(400, {"error": "JSON object required"}) taker: ReplayTake = self.server.take label = "" if body.get("cueId"): cue = self.server.cues.get(str(body["cueId"])) if cue is None: return self._json(409, {"error": "that cue is no longer listed", "code": "no-cue"}) window = taker.window_for_cue(cue) label = cue.label or cue.kind else: seconds = float(body.get("seconds") or DEFAULT_LOOKBACK_S) window = taker.window_for_lookback(seconds) label = f"last {seconds:g}s" try: state = taker.take(from_utc_ms=window[0], to_utc_ms=window[1], label=label) except ReplayTakeError as e: return self._json(409, {"error": str(e), "code": "cannot-take"}) except Exception as e: # noqa: BLE001 — OBS refused, etc. return self._json(502, {"error": f"{type(e).__name__}: {e}", "code": "obs"}) return self._json(200, asdict(state)) class _Server(socketserver.ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def __init__(self, addr, handler, *, take: ReplayTake, cues: CueLog): # noqa: ANN001 super().__init__(addr, handler) self.take, self.cues = take, cues def main(argv=None) -> int: # noqa: ANN001 p = argparse.ArgumentParser(description="Take a replay out of today's recording.") p.add_argument("--anchor", required=True, type=Path, help="the recording_anchor.jsonl") p.add_argument("--replay-dir", required=True, type=Path, help="where the clip is installed; must match --var replaydir= in the OBS layout") p.add_argument("--feed", default="", help="producer base URL, for cue SUGGESTIONS only") p.add_argument("--feed-token", default=os.environ.get("VR_FEED_TOKEN", "")) p.add_argument("--port", type=int, default=8791) p.add_argument("--ffmpeg", default="ffmpeg") p.add_argument("--ffprobe", default="ffprobe") a = p.parse_args(argv) refusal = _refusal() if refusal is not None: print(f"[replay] {refusal['error']} (set ${refusal['hint']})", file=sys.stderr) return 2 taker = ReplayTake(anchor=a.anchor, replay_dir=a.replay_dir, ffmpeg=a.ffmpeg, ffprobe=a.ffprobe) taker.restore_audio() # whatever last time left behind cues = CueLog() if a.feed: CueReader(a.feed, cues, token=a.feed_token).start() # Loopback only. This route changes the picture on a live broadcast; it has no business being # reachable from the venue network, and the operator is at this machine. srv = _Server(("127.0.0.1", a.port), _Handler, take=taker, cues=cues) token = html.escape(os.environ.get("VR_BROADCAST_CONTROL_TOKEN", ""), quote=True) print(f"[replay] http://127.0.0.1:{a.port}/?token={token}") print(f"[replay] clip -> {taker.output}") try: srv.serve_forever() except KeyboardInterrupt: print("\n[replay] stopped") finally: srv.server_close() taker.restore_audio() return 0 if __name__ == "__main__": raise SystemExit(main())