"""Where UTC zero is inside the local recording of the broadcast — and inside the outgoing stream. A race is cut out of a day-long recording by knowing which UTC time corresponds to frame zero of the file. That fact exists only while OBS is recording, and it cannot be reconstructed afterwards from the video: **nothing in the picture is machine-readable**. The stamp on the streamed picture is human-readable text, and so is the clock a device burns into its own footage — the 52-cell machine strip that used to ride along the bottom of a boat-cam clip has been removed, so not even a composited camera tile carries a decodable instant any more. Reading a time back out of a recording now means a person looking at digits. So we write the anchor down while it is still knowable: anchor_utc_ms = now_utc_ms - record_duration_ms Sampled **every 60 s** rather than once at the start, for three reasons, all of which have happened to someone: the recording may be started by hand from the OBS window rather than by us, so there is no "start" for us to stamp; OBS splits the file on a timer, and each new segment is a new zero; and the duration lags the button press by however long the encoder took to open, which a single reading at start-up would bake in as a permanent offset. Each line is one sample, appended, never rewritten — a crashed run leaves everything up to the crash, which is exactly what a recording that also stopped at the crash needs. **The same reading answers it for the outgoing STREAM**, which has a different zero and belongs to somebody else. A partner taking our feed records it at their end, and their frame zero is the moment our stream opened — not the moment our file did. So each sample also carries stream_anchor_utc_ms = now_utc_ms - stream_duration_ms read back by `stream_anchor_for`. Where the recording groups by FILE (a split starts a new zero), the stream has no filename to group by, so a session boundary is read off the elapsed time itself: a duration that goes DOWN means the stream stopped and restarted. This is what makes the `streamOffsetS` in the published metadata (`docs/broadcast-metadata.md` §3) a real number rather than a guess — though UTC stays normative there, because only the recipient knows when their own recorder started. Run it beside the show: python -m v3.broadcast.live.recording_anchor --out D:/trwc/recording_anchor.jsonl Reading it back is `anchor_for(samples, path)` below, used by the day cutter. """ from __future__ import annotations import argparse import json import sys import time from collections.abc import Callable from pathlib import Path _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: # so `director_obs` resolves either way sys.path.insert(0, str(_HERE)) #: One sample a minute. Frequent enough that a file split is noticed inside a minute, rare enough that #: a day's file is a few hundred lines and OBS is never asked anything expensive. DEFAULT_INTERVAL_S = 60.0 def sample(status: dict, *, now_ms: int) -> dict | None: """One anchor row from an `obs_status()` reading, or None when neither output is running. TWO anchors, because there are two artefacts and they do not share a zero. The **recording** is the local file a race is cut out of afterwards; the **stream** is what a recipient of the feed recorded at their end, and their frame zero is when our stream opened, not when our file did. Both are `now − elapsed`, and OBS gives the elapsed of each as a number. `None` rather than a row of nulls: a minute in which OBS was doing neither is not a measurement of anything, and a reader that has to skip null rows will eventually forget to. A minute in which it was doing ONE of them is a measurement of that one, so the other's fields are null and the readers below filter on the field they need.""" row: dict = {"tUtcMs": int(now_ms)} rec_ms = status.get("recordDurationMs") if status.get("recording") else None if rec_ms is not None: row |= {"durationMs": int(rec_ms), "anchorUtcMs": int(now_ms) - int(rec_ms), "path": status.get("recordPath")} stream_ms = status.get("streamDurationMs") if status.get("streaming") else None if stream_ms is not None: row |= {"streamDurationMs": int(stream_ms), "streamAnchorUtcMs": int(now_ms) - int(stream_ms)} return row if len(row) > 1 else None def anchor_for(samples: list[dict], path: str | None = None) -> int | None: """The UTC time of frame zero for `path` (or for the most recent file), or None. The MEDIAN of that file's samples, not the first and not the last. Each sample carries the jitter of one round trip to OBS plus whatever the machine was doing that second; the median throws away the outliers without needing a threshold anyone has to justify. Samples are grouped by file because every split starts a new zero — averaging across a split would place every race in the day somewhere between two files.""" rows = [s for s in samples if s.get("anchorUtcMs") is not None] if path is not None: rows = [s for s in rows if s.get("path") == path] elif rows: rows = [s for s in rows if s.get("path") == rows[-1].get("path")] if not rows: return None anchors = sorted(int(s["anchorUtcMs"]) for s in rows) return anchors[len(anchors) // 2] def stream_anchor_for(samples: list[dict]) -> int | None: """The UTC time our outgoing STREAM opened, for the most recent session, or None. The recording groups by file, because every split starts a new zero. A stream has no filename to group by, so the session boundary is read off the elapsed time itself: **a duration that goes DOWN means the stream stopped and started again**, and samples from before that point belong to a different zero. Averaging across a reconnect would place the whole day between two sessions. Median within the session, for the same reason `anchor_for` uses one: each sample carries the jitter of one round trip to OBS, and the median discards the outliers without a threshold anyone has to justify. """ rows = [s for s in samples if s.get("streamAnchorUtcMs") is not None] if not rows: return None rows.sort(key=lambda s: int(s.get("tUtcMs", 0))) session = [rows[0]] for prev, cur in zip(rows, rows[1:], strict=False): if int(cur.get("streamDurationMs", 0)) < int(prev.get("streamDurationMs", 0)): session = [cur] # the stream restarted — a new zero else: session.append(cur) anchors = sorted(int(s["streamAnchorUtcMs"]) for s in session) return anchors[len(anchors) // 2] def read_samples(out: Path) -> list[dict]: """Every sample written so far. A truncated last line (the run was killed mid-write) is dropped rather than raising — the file is an append log, and its tail is the least valuable part.""" if not out.exists(): return [] rows = [] for line in out.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue return rows def run(out: Path, *, interval_s: float = DEFAULT_INTERVAL_S, status_fn: Callable[[], dict] | None = None, now_ms_fn: Callable[[], int] | None = None, sleep_fn: Callable[[float], None] = time.sleep, max_samples: int | None = None, on_log: Callable[[str], None] = print) -> int: """Sample until interrupted. Returns the number of rows written. A failure to reach OBS is logged and retried on the next tick rather than ending the run: this process outlives an OBS restart, and the samples on either side of one are still good.""" if status_fn is None: # From `director_obs`, NOT from `obs_mcp_server`. The latter imports `mcp.server.fastmcp` at # module scope, so this process — which must run beside the show all day on the render box and # wants exactly one function — could not start unless an agent framework was installed. It # could not start on the render box at all until 2026-08-24 for that reason, which is why # nothing had ever written an anchor there. from director_obs import obs_status # noqa: PLC0415 — optional dep, only when running live status_fn = obs_status now_ms = now_ms_fn or (lambda: int(time.time() * 1000)) out.parent.mkdir(parents=True, exist_ok=True) written, n, last_path, last_stream = 0, 0, None, None while max_samples is None or n < max_samples: n += 1 try: row = sample(status_fn(), now_ms=now_ms()) except Exception as e: # noqa: BLE001 — OBS down is a normal minute on_log(f"[anchor] OBS unreachable ({type(e).__name__}: {e}) — retrying") row = None if row is not None: with out.open("a", encoding="utf-8") as fh: fh.write(json.dumps(row) + "\n") written += 1 # Announce each output's zero once, and ONLY off a row that actually carries it. A row # can now describe the stream but not the recording (or the reverse), so keying the log # on `path` alone would read a missing anchor out of a stream-only row. if row.get("anchorUtcMs") is not None and row.get("path") != last_path: last_path = row.get("path") on_log(f"[anchor] recording {last_path} — frame zero at " f"{row['anchorUtcMs']} (UTC ms)") stream_anchor = row.get("streamAnchorUtcMs") if stream_anchor is not None and ( last_stream is None or int(row["streamDurationMs"]) < last_stream): on_log(f"[anchor] streaming — session zero at {stream_anchor} (UTC ms)") last_stream = row.get("streamDurationMs") if max_samples is None or n < max_samples: sleep_fn(interval_s) return written def main(argv=None) -> int: # noqa: ANN001 p = argparse.ArgumentParser(description="Write down where UTC zero is in the local recording.") p.add_argument("--out", default="recording_anchor.jsonl", help="append-only sample log (default: ./recording_anchor.jsonl)") p.add_argument("--interval", type=float, default=DEFAULT_INTERVAL_S, help=f"seconds between samples (default {DEFAULT_INTERVAL_S:.0f})") a = p.parse_args(argv) out = Path(a.out).expanduser() print(f"[anchor] sampling every {a.interval:.0f}s -> {out} (Ctrl-C to stop)") try: run(out, interval_s=a.interval) except KeyboardInterrupt: print("\n[anchor] stopped") return 0 if __name__ == "__main__": raise SystemExit(main())