"""The break roll: approved phone clips, made playable, on the machine that runs OBS. A broadcast day is mostly pauses. `Break line-up` fills one with the fleet roll-call and can never run out; this fills one with video our own people filmed on their phones. The scene that plays it is in `obs/stream_1080.json` and it plays exactly ONE FILE, `/break_roll.mp4`, which this module builds and rebuilds. python -m v3.broadcast.live.break_roll build --dir D:/vr/break_roll python -m v3.broadcast.live.break_roll sync --event trwc-2026 --read-base https://… \ --read-token "$VR_READ_TOKEN" --interval 60 python -m v3.broadcast.live.break_roll check --dir D:/vr/break_roll `build` is the local half and needs no server at all: whatever sits in `/source/` is normalised and joined. `sync` is `build` with the source folder kept in step with what an operator has APPROVED in the cloud — and the approval is what fetches a clip, so an undecided one is never in this folder waiting to be cut to by mistake. /source/. the originals, byte-for-byte as they were filmed /clips/.mp4 normalised: 1920x1080, yuv420p, 30 fps, -14 LUFS /break_roll.mp4 the joined roll — the ONE file OBS opens /status.json what the last pass found, for a human and for Mission Control THE ORIGINAL IS NEVER DELETED. `clips/` holds derivatives with a fixed frame rate and a fixed loudness; `source/` holds the footage, which cannot be filmed again. ORDER IS THE KEY, and the key never changes. `sync` names a clip by the instant it was APPROVED, so a new approval sorts to the end and nothing is ever renumbered — "I want this one first" is served by approving it first. `build` falls back to the source filename, which sorts the same way for the same reason. A PASS THAT COULD NOT ASK CHANGES NOTHING. `sync` reports `unknown` and leaves the folder exactly as it was when the read plane is unreachable: "I could not ask" is not "there is nothing approved", and emptying a break because 4G blipped is the one failure that shows up on air. """ from __future__ import annotations import argparse import datetime as _dt import json import os import shutil import subprocess import sys import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path from urllib.parse import urljoin #: Where the roll collects when the deployment has not said, mirroring `replay_library.library_dir`. DEFAULT_SUBDIR = "break_roll" DIR_ENV = "VR_BREAK_DIR" #: The produced picture is 1920x1080 and a break is part of it. Not the 1280x720 capture default — #: that is what a PHONE records at, and this is what the programme goes out at. FRAME = (1920, 1080) #: The programme's own loudness. Matching it is the whole point of letting the clips keep their sound: #: the same figure `render_gsys_mp4.py` and `demo/record/build_audio.py` already produce, so cutting #: to a break is a change of picture and not a change of volume. LOUDNORM = "loudnorm=I=-14:TP=-1.5:LRA=11" #: The input is a phone held on a boat, and below about 90 Hz the dominant signal is wind across a #: microphone. Removing it before the loudness pass also stops that rumble from setting the gain. HIGHPASS = "highpass=f=90" #: A clip longer than this is truncated. A break clip is not a film, and one four-minute upload must #: not own the whole pause. The event's own figure overrides it (`presentation.breakClips.maxSeconds`). DEFAULT_MAX_SECONDS = 45 #: Files we will attempt. Anything else in `source/` is reported and skipped rather than fed to #: ffmpeg, so a stray `.DS_Store` or a half-finished download is not an error on air. SOURCE_SUFFIXES = (".mp4", ".mov", ".m4v", ".webm", ".mkv", ".avi") def break_dir(root: Path, explicit: str | None = None) -> Path: """`--dir` > `$VR_BREAK_DIR` > `/_cap/break_roll`, expanded for `~` and other variables — the three-step `replay_library.library_dir` already established, for the same reason: where the material lives is a deployment fact and must not reach the code.""" raw = explicit or os.environ.get(DIR_ENV) if raw: return Path(os.path.expandvars(str(raw))).expanduser() return root / "_cap" / DEFAULT_SUBDIR def utc_iso(ms: float | None = None) -> str: """UTC with its offset, ISO 8601 — what everything this product stores looks like.""" t = _dt.datetime.now(_dt.UTC) if ms is None else _dt.datetime.fromtimestamp(ms / 1000, _dt.UTC) return t.isoformat(timespec="milliseconds") # --- the ffmpeg commands, built as data so they can be read and tested without ffmpeg ------------- def probe_audio_cmd(src: Path) -> list[str]: return ["ffprobe", "-v", "error", "-select_streams", "a", "-show_entries", "stream=index", "-of", "csv=p=0", str(src)] def normalize_cmd(src: Path, dst: Path, *, fps: int = 30, max_seconds: int | None = DEFAULT_MAX_SECONDS, frame: tuple[int, int] = FRAME, encoder: str = "libx264", has_audio: bool = True, mute: bool = False) -> list[str]: """One pass from a phone clip to something the programme can play. ONE FILTER GRAPH FOR BOTH ORIENTATIONS, so there is no branch to get wrong. Half the material will be portrait — a phone held the way people hold phones — and ffmpeg applies the container's rotation on decode, so the clip arrives portrait and `force_original_aspect_ratio` does the rest. The sides are filled with a blurred, darkened, cover-scaled copy of the same clip rather than black bars. For a landscape clip the foreground covers the backdrop entirely, so the identical command is correct for both and nothing has to detect which one it has — a detection that would have to read the rotation matrix rather than the encoded dimensions, since a portrait clip has LANDSCAPE dimensions and expresses the quarter-turn separately. EVERY OUTPUT GETS AN AUDIO TRACK, even a silent clip, and that is what keeps the join a stream copy: a track that appears and disappears part-way through the roll is a re-encode at best and a mixer noticing mid-break at worst. `-shortest` is right HERE — the null source is infinite and the picture is finite — and is the opposite of the case `record_demo.py` documents, where a finite audio track would have cut the picture short. """ w, h = frame silent = mute or not has_audio # `setsar=1` goes LAST, on the overlay's output. Set on the foreground instead, the overlay # inherits the BACKDROP's sample aspect and the file comes out at 10239:10240 — visually nothing, # but the roll is joined with a stream copy, so segments that disagree about SAR are segments a # player is entitled to treat as a new stream part-way through a break. graph = ( f"[0:v]scale={w}:{h}:force_original_aspect_ratio=increase,crop={w}:{h}," f"gblur=sigma=28,eq=brightness=-0.12:saturation=0.75[bg];" f"[0:v]scale={w}:{h}:force_original_aspect_ratio=decrease:flags=lanczos[fg];" f"[bg][fg]overlay=(W-w)/2:(H-h)/2,fps={fps},setsar=1,format=yuv420p[v]" ) cmd = ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-i", str(src)] if silent: cmd += ["-f", "lavfi", "-i", "anullsrc=channel_layout=stereo:sample_rate=48000"] cmd += ["-filter_complex", graph, "-map", "[v]"] if silent: cmd += ["-map", "1:a", "-shortest"] else: cmd += ["-map", "0:a:0", "-af", f"{HIGHPASS},{LOUDNORM}"] if encoder == "h264_nvenc": cmd += ["-c:v", "h264_nvenc", "-preset", "p5", "-rc", "vbr", "-cq", "21", "-b:v", "0"] else: cmd += ["-c:v", "libx264", "-crf", "20", "-preset", "medium"] cmd += ["-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-ar", "48000", "-ac", "2"] if max_seconds: cmd += ["-t", str(int(max_seconds))] cmd += ["-movflags", "+faststart", str(dst)] return cmd def concat_cmd(listing: Path, dst: Path) -> list[str]: """A stream copy, because every part is byte-identical in format by the time it gets here.""" return ["ffmpeg", "-y", "-hide_banner", "-loglevel", "error", "-f", "concat", "-safe", "0", "-i", str(listing), "-c", "copy", "-movflags", "+faststart", str(dst)] def concat_listing(clips: list[Path]) -> str: """ffmpeg's concat demuxer input. Single quotes are escaped the way the demuxer wants, and paths are written with forward slashes — which ffmpeg accepts on Windows and which keeps a backslash from being read as an escape.""" out = [] for clip in clips: path = clip.as_posix().replace("'", r"'\''") out.append(f"file '{path}'") return "\n".join(out) + "\n" # --- running them -------------------------------------------------------------------------------- def _run(cmd: list[str], *, timeout: float = 900.0) -> subprocess.CompletedProcess: return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False) def have_tool(name: str) -> bool: return shutil.which(name) is not None def pick_encoder(run=_run) -> str: """`h264_nvenc` where the box has it — the broadcast machine does — else libx264. Asked once and remembered by the caller. A wrong guess is not a slow render, it is an ffmpeg that exits immediately, which on a poll loop is a roll that silently never updates.""" try: res = run(["ffmpeg", "-hide_banner", "-encoders"], timeout=20) except Exception: # noqa: BLE001 — no ffmpeg is answered below return "libx264" return "h264_nvenc" if "h264_nvenc" in (res.stdout or "") else "libx264" def has_audio(src: Path, run=_run) -> bool: """True unless ffprobe can say for certain there is no audio stream. Defaulting to TRUE on a failed probe is deliberate: mapping an audio track that exists is harmless, while adding a null source to a clip that already has one produces a file with two tracks and a break whose sound depends on which the player picks.""" try: res = run(probe_audio_cmd(src), timeout=30) except Exception: # noqa: BLE001 return True if res.returncode != 0: return True return bool((res.stdout or "").strip()) # --- the local half ------------------------------------------------------------------------------ @dataclass(frozen=True) class Pass: """What one build did, for `status.json` and for the operator's eye.""" state: str # "ok" | "unknown" reason: str = "" normalized: tuple[str, ...] = () removed: tuple[str, ...] = () failed: tuple[str, ...] = () skipped: tuple[str, ...] = () clips: int = 0 bytes_len: int = 0 changed: bool = False def _sources(source_dir: Path) -> tuple[list[Path], list[str]]: """The files we will try, in key order, and the names we are ignoring. A half-finished download (`.part`) is ignored SILENTLY — it is a normal transient, and reporting it every pass would train an operator to skim the one list that is supposed to mean something.""" keep: list[Path] = [] skip: list[str] = [] for p in sorted(source_dir.iterdir()) if source_dir.is_dir() else []: if not p.is_file() or p.suffix.lower() == ".part": continue if p.suffix.lower() in SOURCE_SUFFIXES: keep.append(p) else: skip.append(p.name) return keep, skip def build(directory: Path, *, fps: int = 30, max_seconds: int | None = DEFAULT_MAX_SECONDS, mute: bool = False, encoder: str | None = None, run=_run, log=print) -> Pass: """Normalise anything new in `source/`, drop derivatives whose original is gone, join the rest. Idempotent: a second call with nothing new re-encodes nothing and rewrites nothing. """ source_dir, clips_dir = directory / "source", directory / "clips" source_dir.mkdir(parents=True, exist_ok=True) clips_dir.mkdir(parents=True, exist_ok=True) roll = directory / "break_roll.mp4" if not have_tool("ffmpeg"): # Nothing is touched. A box without ffmpeg cannot build a roll, and pretending otherwise # would leave yesterday's roll in place while reporting success. return Pass(state="unknown", reason="ffmpeg is not on PATH") encoder = encoder or pick_encoder(run) sources, skipped = _sources(source_dir) wanted = {p.stem: p for p in sources} removed = [] for existing in sorted(clips_dir.glob("*.mp4")): if existing.stem not in wanted: existing.unlink(missing_ok=True) removed.append(existing.stem) log(f" dropped {existing.stem} — its original is gone") normalized, failed = [], [] for key, src in sorted(wanted.items()): dst = clips_dir / f"{key}.mp4" if dst.exists() and dst.stat().st_size > 0 and dst.stat().st_mtime >= src.stat().st_mtime: continue cmd = normalize_cmd(src, dst, fps=fps, max_seconds=max_seconds, encoder=encoder, has_audio=has_audio(src, run), mute=mute) res = run(cmd) if res.returncode != 0 or not dst.exists() or dst.stat().st_size == 0: # One unreadable clip must not cost the whole roll. It is named, skipped, and tried again # next pass — a phone that uploaded half a file is the ordinary reason. dst.unlink(missing_ok=True) failed.append(key) log(f" FAILED {key}: {(res.stderr or '').strip().splitlines()[-1:] or ['ffmpeg said nothing']}") continue normalized.append(key) log(f" normalised {key}") ready = sorted(p for p in clips_dir.glob("*.mp4") if p.stat().st_size > 0) changed = bool(normalized or removed) or not roll.exists() if not ready: # An empty roll is REMOVED rather than left as a stale file. The scene falls through to the # line-up when there is nothing to play, and a roll from yesterday is not nothing. if roll.exists(): roll.unlink() changed = True log(" nothing approved — removed the roll; the scene will show the line-up") return Pass(state="ok", normalized=tuple(normalized), removed=tuple(removed), failed=tuple(failed), skipped=tuple(skipped), clips=0, changed=changed) if changed: listing = directory / "clips.txt" listing.write_text(concat_listing(ready), encoding="utf-8") staged = directory / "break_roll.next.mp4" res = run(concat_cmd(listing, staged)) if res.returncode != 0 or not staged.exists(): staged.unlink(missing_ok=True) return Pass(state="unknown", reason="the join failed; the previous roll is untouched", normalized=tuple(normalized), removed=tuple(removed), failed=tuple(failed), skipped=tuple(skipped), clips=len(ready)) try: staged.replace(roll) except OSError as e: # Windows will not replace a file OBS holds open, which is precisely what happens if the # break scene is on air. Keep the staged file for the next pass and say why. return Pass(state="unknown", reason=f"could not replace the roll ({e.strerror or e}) — is the break scene on air?", normalized=tuple(normalized), removed=tuple(removed), failed=tuple(failed), skipped=tuple(skipped), clips=len(ready)) log(f" joined {len(ready)} clip(s) -> {roll.name}") return Pass(state="ok", normalized=tuple(normalized), removed=tuple(removed), failed=tuple(failed), skipped=tuple(skipped), clips=len(ready), bytes_len=roll.stat().st_size if roll.exists() else 0, changed=changed) def write_status(directory: Path, result: Pass, *, approved: int | None = None, fps: int = 30, extra: dict | None = None) -> Path: """`status.json` — what the last pass found. A missing reading is stated, never zeroed: `approved` is null when we could not ask, because a board that draws "0 approved" over a failed fetch is the same lie as a green probe that never ran.""" path = directory / "status.json" prior = {} if path.exists(): try: prior = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): prior = {} now = utc_iso() doc = { "checkedUtc": now, "state": result.state, "reason": result.reason, "approved": approved, "clips": result.clips, "bytes": result.bytes_len, "fps": fps, "normalized": list(result.normalized), "removed": list(result.removed), "failed": list(result.failed), "skipped": list(result.skipped), "lastChangeUtc": now if result.changed else prior.get("lastChangeUtc"), **(extra or {}), } path.write_text(json.dumps(doc, indent=2) + "\n", encoding="utf-8") return path # --- the cloud half ------------------------------------------------------------------------------ def approved_url(read_base: str, event: str) -> str: return f"{read_base.rstrip('/')}/v3/break/{event}/approved.json" def fetch_approved(read_base: str, event: str, token: str, *, timeout: float = 20.0) -> list[dict]: """The allow-list. Raises on anything that is not a clean answer — the caller turns that into `unknown`, which is the whole point: a failed read must not read as an empty one.""" req = urllib.request.Request(approved_url(read_base, event)) # noqa: S310 — operator's own base if token: req.add_header("Authorization", f"Bearer {token}") with urllib.request.urlopen(req, timeout=timeout) as r: # noqa: S310 doc = json.loads(r.read().decode("utf-8")) clips = doc.get("clips") if not isinstance(clips, list): raise ValueError("approved.json carried no clip list") return clips def clip_key(clip: dict) -> str: """A name that sorts into APPROVAL order and never changes once written. The approval instant leads, so a new approval sorts to the end and no existing file is ever renumbered; the id follows so two approvals in the same millisecond stay apart.""" return f"{int(clip.get('approvedUtcMs') or 0):015d}_{str(clip.get('id') or 'clip')}" def _suffix(clip: dict) -> str: url = str(clip.get("url") or "") stem = url.split("?", 1)[0].rsplit("/", 1)[-1] suffix = Path(stem).suffix.lower() return suffix if suffix in SOURCE_SUFFIXES else ".mp4" def reconcile(directory: Path, clips: list[dict], *, read_base: str = "", timeout: float = 120.0, log=print) -> tuple[int, list[str]]: """Make `source/` hold exactly the approved set. Returns (downloaded, failures). A clip's `url` may be ABSOLUTE (a signed GCS URL, which is the production case and points at another host entirely) or RELATIVE to the read plane (a local store, which has no signed URL and is served by a route of ours). Resolving it against the base we asked handles both, and is what a browser does for the review page without being told.""" source_dir = directory / "source" source_dir.mkdir(parents=True, exist_ok=True) want = {clip_key(c): c for c in clips} for existing in sorted(source_dir.iterdir()): if existing.is_file() and existing.stem not in want: # A revoked approval reaches the AIR here, not just the database. Note the other edge of # the same rule: under `sync` the approved set is the whole truth about this folder, so a # file dropped in by hand is removed too. Hand-placed material is what `build` is for. existing.unlink(missing_ok=True) log(f" withdrew {existing.stem}") have = {p.stem for p in source_dir.iterdir() if p.is_file()} downloaded, failures = 0, [] for key, clip in sorted(want.items()): if key in have: continue url = urljoin(read_base or "", str(clip.get("url") or "")) if not url: failures.append(key) continue dst = source_dir / f"{key}{_suffix(clip)}" part = dst.with_suffix(dst.suffix + ".part") try: with urllib.request.urlopen(url, timeout=timeout) as r, part.open("wb") as fh: # noqa: S310 shutil.copyfileobj(r, fh) # Only now does it get its real name: a half-written file must never look complete, or # the next pass normalises a truncated clip and calls it done. part.replace(dst) downloaded += 1 log(f" fetched {key}") except Exception as e: # noqa: BLE001 — one clip must not stop the pass part.unlink(missing_ok=True) failures.append(key) log(f" could not fetch {key}: {e}") return downloaded, failures def sync_once(directory: Path, *, event: str, read_base: str, token: str, fps: int = 30, max_seconds: int | None = DEFAULT_MAX_SECONDS, mute: bool = False, encoder: str | None = None, run=_run, fetch=fetch_approved, log=print) -> Pass: """One pass: ask, reconcile, build, write the status.""" try: clips = fetch(read_base, event, token) except Exception as e: # noqa: BLE001 — every failure is the same answer # THE FOLDER IS NOT TOUCHED. "I could not ask" is not "there is nothing approved". result = Pass(state="unknown", reason=f"could not read approved.json: {e}") write_status(directory, result, approved=None, fps=fps) log(f"[break-roll] unknown — {result.reason}") return result downloaded, failures = reconcile(directory, clips, read_base=read_base, log=log) result = build(directory, fps=fps, max_seconds=max_seconds, mute=mute, encoder=encoder, run=run, log=log) if failures and result.state == "ok": result = Pass(state="unknown", reason=f"{len(failures)} clip(s) could not be fetched", normalized=result.normalized, removed=result.removed, failed=tuple([*result.failed, *failures]), skipped=result.skipped, clips=result.clips, bytes_len=result.bytes_len, changed=result.changed) write_status(directory, result, approved=len(clips), fps=fps, extra={"downloaded": downloaded}) log(f"[break-roll] {result.state} — {len(clips)} approved, {result.clips} in the roll" + (f" ({result.reason})" if result.reason else "")) return result # --- CLI ----------------------------------------------------------------------------------------- def _repo_root() -> Path: return Path(__file__).resolve().parents[3] def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[0], formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("action", choices=("build", "sync", "check")) ap.add_argument("--dir", help=f"where the roll lives (else ${DIR_ENV}, else /_cap/{DEFAULT_SUBDIR})") ap.add_argument("--event", help="sync: the event whose approved clips to follow") ap.add_argument("--read-base", help="sync: the read plane, e.g. https://apps.viewregatta.com") ap.add_argument("--read-token", default="", help="sync: the read token. $VR_READ_TOKEN is a FALLBACK only — a fetch that " "silently ran anonymous once cost a race, so the credential belongs on the " "command line where it can be seen") ap.add_argument("--interval", type=float, default=0, help="sync: keep going, this many seconds apart (0 = one pass and exit)") ap.add_argument("--fps", type=int, default=30, help="must match the OBS profile's frame rate") ap.add_argument("--max-seconds", type=int, default=DEFAULT_MAX_SECONDS, help="truncate a clip longer than this (0 = never)") ap.add_argument("--mute", action="store_true", help="drop the clips' own sound and lay a silent track instead") a = ap.parse_args(argv) directory = break_dir(_repo_root(), a.dir) directory.mkdir(parents=True, exist_ok=True) max_seconds = a.max_seconds or None if a.action == "check": path = directory / "status.json" if not path.exists(): print(f"{directory}: no pass has run yet") return 1 print(path.read_text(encoding="utf-8"), end="") return 0 if a.action == "build": result = build(directory, fps=a.fps, max_seconds=max_seconds, mute=a.mute) write_status(directory, result, fps=a.fps) print(f"[break-roll] {result.state} — {result.clips} clip(s)" + (f" ({result.reason})" if result.reason else "")) return 0 if result.state == "ok" else 1 if not a.event or not a.read_base: raise SystemExit("sync needs --event and --read-base") token = a.read_token or os.environ.get("VR_READ_TOKEN", "") if not token: print("note: no --read-token and no $VR_READ_TOKEN — the fetch will run anonymous") while True: result = sync_once(directory, event=a.event, read_base=a.read_base, token=token, fps=a.fps, max_seconds=max_seconds, mute=a.mute) if a.interval <= 0: return 0 if result.state == "ok" else 1 time.sleep(a.interval) if __name__ == "__main__": sys.exit(main())