"""Our scene layout as a file OBS can IMPORT — for a partner who renders the picture themselves. `obs_scenes.py` applies a layout to a running OBS over obs-websocket. That is the right tool for a machine we operate: it is idempotent, it preflights every origin, and it can re-apply mid-show without closing OBS. It is the wrong thing to hand a partner, because it asks them to enable obs-websocket, install Python and trust a script against their production box before they have seen the picture once. This writes the same layout as a **native OBS scene collection** — the JSON that *Scene Collection → Import* accepts. They double-click it, point OBS at our producer, and see the show. It also writes the same thing as a **setup sheet**, so an operator can build it by hand in a few minutes and, more importantly, can CHECK what they built against what we meant. python -m v3.broadcast.live.obs_export --var producer=https:// \\ --out ViewRegatta.json --sheet ViewRegatta-setup.md Both outputs are generated from `obs/stream_1080.json`, the same file we run ourselves, so a partner cannot be sent a layout that has drifted from the one we use. **Partner defaults differ from ours, on purpose.** Our rehearsal rig serves the feed and the media from one HTTP/1.1 process, so the layout splits `camorigin` and `mediabase` onto a second origin to stay under the ~6-connections-per-origin limit. A partner points at Cloud Run, which is HTTP/2 — requests are multiplexed over one connection and the limit does not apply — and the clips arrive as signed GCS URLs, already another origin. So `camorigin` defaults to `producer` here and `mediabase` is dropped, which is what the layout's own header says production should do. """ from __future__ import annotations import argparse import json import re import sys from pathlib import Path _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) from obs_scenes import expand # noqa: E402 — one substitution rule, shared with `apply` LAYOUT = _HERE / "obs" / "stream_1080.json" PRODUCTION_READ_BASE = "https://apps.viewregatta.com" #: OBS's own enum for a scene item's bounds. `SCALE_INNER` fits the source inside the box keeping its #: aspect — which is what every tile in this layout wants. _BOUNDS = {"OBS_BOUNDS_NONE": 0, "OBS_BOUNDS_STRETCH": 1, "OBS_BOUNDS_SCALE_INNER": 2, "OBS_BOUNDS_SCALE_OUTER": 3, "OBS_BOUNDS_SCALE_TO_WIDTH": 4, "OBS_BOUNDS_SCALE_TO_HEIGHT": 5, "OBS_BOUNDS_MAX_ONLY": 6} #: The browser-source settings we care about. Anything else OBS fills with its own defaults, which is #: what we want — a partner's OBS version knows its defaults better than we do. _BROWSER_DEFAULTS = {"width": 1920, "height": 1080, "fps_custom": False, "fps": 30, "shutdown": False, "restart_when_active": False, "reroute_audio": False, "css": ""} def partner_vars(layout: dict, overrides: dict[str, str]) -> dict[str, str]: """The layout's own defaults, adjusted for a partner pointing at a production deployment.""" v = {k: str(val) for k, val in (layout.get("vars") or {}).items() if not k.startswith("_")} v.update(overrides) producer = v.get("producer", "") # Both of these exist only to dodge an HTTP/1.1 connection limit on our own rehearsal rig. v.setdefault("camorigin", producer) if "camorigin" not in overrides: v["camorigin"] = producer if "mediabase" not in overrides: v["mediabase"] = "" # The partner's browser sources run on the producer origin, but the live HLS playlists are # published by the production read plane. Leaving the rehearsal default here produces an OBS # collection that looks valid yet asks the partner's own computer for every camera stream. # Keep an explicit override for staging and future deployments; otherwise use the public read # plane documented by the production layout. if "readbase" not in overrides: v["readbase"] = PRODUCTION_READ_BASE # The series card is a page of the producer's, like every other one — it carries its own var only # so a rig can serve it from somewhere else while the programme keeps running from a deployed # producer. A partner has one origin and it is theirs, exactly as with `camorigin`. if "standingsbase" not in overrides: v["standingsbase"] = producer return v #: A host that only means anything on the machine that wrote the file. Shipping one to a partner is #: the single most likely way this export goes wrong, and it fails as a blank page rather than as an #: error — so it is refused at generation time instead. _LOCAL_HOSTS = ("localhost", "127.0.0.1", "0.0.0.0", "[::1]") def clean_url(url: str) -> str: """Drop query parameters left empty by an unset variable. `?mediaBase=` is not the same as no `mediaBase` at all: the page reads the parameter, finds an empty string and uses it, instead of falling back to the default it would have chosen. A partner on Cloud Run wants the fallback, so the parameter has to go rather than arrive empty.""" head, sep, query = url.partition("?") if not sep: return url kept = [p for p in query.split("&") if p and not p.endswith("=")] return head + ("?" + "&".join(kept) if kept else "") def local_urls(doc: dict) -> list[tuple[str, str]]: """`[(source, url)]` for every emitted URL that points at the generating machine.""" out = [] for src in doc.get("sources") or []: url = str((src.get("settings") or {}).get("url") or "") if url and any(h in url for h in _LOCAL_HOSTS): out.append((src.get("name", "?"), url)) return out #: The last WORD of a query parameter's name that marks it as carrying a credential. Matched on the #: final word rather than as a substring, so `readToken` and `api_key` are caught while `slots` and #: `mediaBase` are not — a guard that cried wolf on ordinary parameters is a guard somebody disables. _SECRET_WORDS = frozenset({"token", "key", "secret", "password", "passwd", "pwd", "auth", "apikey", "sig", "signature", "credential", "credentials"}) def _looks_secret(param: str) -> bool: words = [w for w in re.split(r"[^A-Za-z0-9]+|(?<=[a-z0-9])(?=[A-Z])", param) if w] return bool(words) and words[-1].lower() in _SECRET_WORDS def secret_urls(doc: dict) -> list[tuple[str, str]]: """`[(source, parameter)]` for every emitted URL carrying something that looks like a secret. This file is written to be e-mailed, zipped and forwarded, and `obs_scenes dump` writes URLs back into a layout meant to be diffed and committed. A token in a query string therefore ends up in an inbox and a repository at once, and neither can be un-sent. Credentials belong in a header or a per-partner account — never in a URL. """ out = [] for src in doc.get("sources") or []: url = str((src.get("settings") or {}).get("url") or "") _, _, query = url.partition("?") for part in query.split("&"): name = part.split("=", 1)[0].strip() if name and _looks_secret(name): out.append((src.get("name", "?"), name)) return out #: The only input kind this export can produce — `_input_source` writes a `browser_source` and the #: sheet tells the reader every source is one. A layout item of any other kind is therefore NOT #: exportable, and the scene that needs it is left out with a line saying so. Emitting it anyway #: would hand a partner a browser source with no URL: a permanently black rectangle wearing the name #: of a source they cannot fix, because what it actually wants is a file on OUR machine. _EXPORTABLE_KIND = "browser_source" def machine_local_scenes(layout: dict) -> dict[str, list[str]]: """`{scene: [source, …]}` for every scene that needs an input this export cannot represent. Today that is a source reading a local FILE — the break roll's media source, whose folder is maintained on the broadcast machine by `break_roll.py` and does not exist on a partner's box.""" out: dict[str, list[str]] = {} for scene in layout.get("scenes") or []: names = [str(it.get("source")) for it in (scene.get("items") or []) if str(it.get("kind", _EXPORTABLE_KIND)) != _EXPORTABLE_KIND] if names: out[str(scene.get("name"))] = names return out def _machine_local_sources(layout: dict) -> set[str]: return {n for names in machine_local_scenes(layout).values() for n in names} def _sources(layout: dict, variables: dict[str, str]) -> dict[str, dict]: """Every distinct input the layout names, keyed by its source name. Sources are SHARED between scenes — the video wall borrows the programme page and each boat cam rather than opening a second copy. A second copy would be a second feed connection for an identical page, which is the thing the layout is arranged to avoid, so the export has to preserve the sharing rather than flatten it.""" out: dict[str, dict] = {} local = _machine_local_sources(layout) for scene in layout.get("scenes") or []: for item in scene.get("items") or []: name = item.get("source") if not name or name in out or name in local: continue settings = dict(item.get("settings") or {}) if not settings.get("url"): continue # a borrowed item: its settings live on the first use settings["url"] = clean_url(expand(str(settings["url"]), variables)) out[name] = {**_BROWSER_DEFAULTS, **settings} # A borrowed item can appear before the one that defines it; a second pass is cheaper than # ordering rules nobody would remember. A machine-local source is NOT given browser defaults # here — that is precisely how one would become a blank browser source in a partner's OBS. for scene in layout.get("scenes") or []: for item in scene.get("items") or []: name = item.get("source") if name and name not in out and name not in local: out[name] = dict(_BROWSER_DEFAULTS) return out def _item(item: dict, item_id: int) -> dict: t = item.get("transform") or {} bounds = _BOUNDS.get(str(t.get("boundsType", "OBS_BOUNDS_NONE")), 0) return { "name": item.get("source"), "visible": bool(item.get("enabled", True)), "locked": False, "rot": 0.0, "pos": {"x": float(t.get("positionX", 0)), "y": float(t.get("positionY", 0))}, "scale": {"x": 1.0, "y": 1.0}, "align": int(t.get("alignment", 5)), "bounds_type": bounds, "bounds_align": int(t.get("boundsAlignment", 0)), "bounds": {"x": float(t.get("boundsWidth", 0)), "y": float(t.get("boundsHeight", 0))}, "crop_left": 0, "crop_top": 0, "crop_right": 0, "crop_bottom": 0, "id": item_id, "group_item_backup": False, "scale_filter": "disable", "blend_method": "default", "blend_type": "normal", "show_transition": {"duration": 0}, "hide_transition": {"duration": 0}, "private_settings": {}, } def _input_source(name: str, settings: dict) -> dict: return {"prev_ver": 0, "name": name, "id": "browser_source", "versioned_id": "browser_source", "settings": settings, "mixers": 255, "sync": 0, "flags": 0, "volume": 1.0, "balance": 0.5, "enabled": True, "muted": False, "push-to-mute": False, "push-to-mute-delay": 0, "push-to-talk": False, "push-to-talk-delay": 0, "hotkeys": {}, "deinterlace_mode": 0, "deinterlace_field_order": 0, "monitoring_type": 0, "private_settings": {}} def _scene_source(scene: dict) -> dict: items = [_item(it, n) for n, it in enumerate(scene.get("items") or [], start=1)] return {"prev_ver": 0, "name": scene["name"], "id": "scene", "versioned_id": "scene", "settings": {"id_counter": len(items), "custom_size": False, "items": items}, "mixers": 0, "sync": 0, "flags": 0, "volume": 1.0, "balance": 0.5, "enabled": True, "muted": False, "push-to-mute": False, "push-to-mute-delay": 0, "push-to-talk": False, "push-to-talk-delay": 0, "hotkeys": {}, "deinterlace_mode": 0, "deinterlace_field_order": 0, "monitoring_type": 0, "private_settings": {}} def scene_collection(layout: dict, variables: dict[str, str], *, name: str | None = None, skip: tuple[str, ...] = ()) -> dict: """The layout as an OBS scene collection, ready for *Scene Collection → Import*.""" skip = tuple(skip) + tuple(machine_local_scenes(layout)) scenes = [s for s in (layout.get("scenes") or []) if s.get("name") not in skip] kept = {s["name"] for s in scenes} used = {it.get("source") for s in scenes for it in (s.get("items") or []) if it.get("source")} sources = [_input_source(n, st) for n, st in _sources(layout, variables).items() if n in used] program = layout.get("program") if program not in kept: program = scenes[0]["name"] if scenes else None return { "name": name or layout.get("collection") or "ViewRegatta", "current_scene": program, "current_program_scene": program, "current_transition": "Fade", "scene_order": [{"name": s["name"]} for s in scenes], "sources": sources + [_scene_source(s) for s in scenes], "transitions": [], "transition_duration": 300, "preview_locked": False, "scaling_enabled": False, "modules": {}, } def setup_sheet(layout: dict, variables: dict[str, str], *, skip: tuple[str, ...] = ()) -> str: """The same layout as instructions — the path that cannot silently import wrong. Deliberately includes the canvas and the exact URL per source, because those are the two things an operator cannot infer and the two that fail quietly when wrong.""" w, h = (layout.get("canvas") or [1920, 1080])[:2] dropped = machine_local_scenes(layout) skip = tuple(skip) + tuple(dropped) scenes = [s for s in (layout.get("scenes") or []) if s.get("name") not in skip] srcs = _sources(layout, variables) out = [ "# ViewRegatta — OBS setup", "", "Build this by hand in a few minutes, or import the accompanying `.json` via", "**Scene Collection → Import** and check it against this sheet.", "", f"**Canvas: {w}×{h}.** Settings → Video → Base *and* Output resolution.", "", "Every source is a **Browser** source. Leave *Shutdown source when not visible* and", "*Refresh browser when scene becomes active* **off** — a page that reloads on every cut", "spends a second black and drops its feed connection.", "", "## Sources", "", "| Source | URL | Size |", "| --- | --- | --- |", ] for name, st in srcs.items(): if not any(it.get("source") == name for s in scenes for it in (s.get("items") or [])): continue url = st.get("url") or "—" out.append(f"| `{name}` | `{url}` | {st.get('width')}×{st.get('height')} |") out += ["", "## Scenes", ""] for s in scenes: out.append(f"### {s['name']}") out.append("") for it in s.get("items") or []: t = it.get("transform") or {} box = (f"{int(t.get('boundsWidth', w))}×{int(t.get('boundsHeight', h))} " f"at ({int(t.get('positionX', 0))}, {int(t.get('positionY', 0))})") note = it.get("_cell") or it.get("_why_one_source") or "" out.append(f"- `{it.get('source')}` — {box}" + (f" \n *{note}*" if note else "")) out.append("") if dropped: out += [ "## Left out of this export", "", "These scenes play a FILE held on our broadcast machine, so there is nothing to hand", "over: the material lives where it is produced, not on your box. They are named here so", "that a scene list one short is a fact you were told rather than one you discover.", "", ] for scene, sources in sorted(dropped.items()): out.append(f"- **{scene}** — needs " + ", ".join(f"`{s}`" for s in sources)) out.append("") out += [ "## Two things that fail silently", "", "1. **Only the scene on programme renders.** OBS draws the active scene and nothing else, so", " a scene that is not live freezes on its last texture. On a multiview that reads as *that", " one is showing older content*, not as *that one is off*.", "2. **Roughly six connections per origin.** Every page holds one permanent connection to the", " feed, and OBS shares a single network stack across all its browser sources. This layout", " is arranged to stay under that limit — the four boat cameras are **one** page in", " `Race cam`, not four — so adding copies of these sources costs more than it looks.", "", ] # Only worth saying where a source really is used twice; with a single-use layout it reads as a # warning about something the reader cannot see. reused = {n for n in srcs if sum(1 for s in scenes for it in (s.get("items") or []) if it.get("source") == n) > 1} if reused: out += [ "Some sources are **shared** between scenes rather than copied — a duplicate would be a", "second feed connection for an identical page. Keep it that way: " + ", ".join(f"`{n}`" for n in sorted(reused)) + ".", "", ] out += [ "## The cameras follow the fleet by themselves", "", "A camera URL asks for a **slot** (`?slot=A`), not a boat. Our producer decides which boats", "occupy which slot and re-points them between heats. Nothing here reloads, changes, or has to", "be told: set the switcher up once and it stays correct all week.", ] return "\n".join(out) + "\n" def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--layout", default=str(LAYOUT), help=f"scene layout to export (default {LAYOUT.name})") ap.add_argument("--var", action="append", metavar="NAME=VALUE", default=[], help="substitute ${NAME} in URLs — at least --var producer=") ap.add_argument("--name", help="collection name as it appears in OBS") ap.add_argument("--skip-scene", action="append", default=[], metavar="NAME", help="leave a scene out (e.g. a wall a partner does not drive)") ap.add_argument("--allow-localhost", action="store_true", help="permit URLs pointing at this machine (a local rig, not a partner)") ap.add_argument("--out", help="write the importable OBS scene collection here") ap.add_argument("--sheet", help="write the human setup sheet here") a = ap.parse_args(argv) variables = {} for pair in a.var: if "=" not in pair: raise SystemExit(f"--var {pair!r} must be name=value") k, v = pair.split("=", 1) variables[k] = v layout = json.loads(Path(a.layout).expanduser().read_text(encoding="utf-8")) variables = partner_vars(layout, variables) skip = tuple(a.skip_scene) # Refuse to hand anyone a URL that only resolves on this machine. The layout's own defaults are # the rehearsal rig's, so forgetting a `--var` is the ordinary mistake rather than an exotic one — # and on the partner's box it fails as a permanently blank browser source with nothing in a log. probe = scene_collection(layout, variables, name=a.name, skip=skip) # A credential in a URL would be written into a file that is e-mailed, zipped and forwarded — # and `obs_scenes dump` writes URLs back into a layout meant to be committed. There is no # `--allow` for this one: it is never right. if leaked := secret_urls(probe): lines = "\n ".join(f"{n}: ?{p}=…" for n, p in leaked) raise SystemExit( "these URLs carry something that looks like a credential, and this file is meant to be " "shared:\n " + lines + "\n\nA token belongs in a header or a per-partner account, never in a query string.") if not a.allow_localhost and (bad := local_urls(probe)): lines = "\n ".join(f"{n}: {u}" for n, u in bad) raise SystemExit( "these sources still point at this machine, so they would arrive blank:\n " + lines + "\n\nSupply the missing host (--var producer=… --var burgee=…) or leave the scene out " "(--skip-scene 'Video wall'). --allow-localhost if this really is for a local rig.") if a.out: doc = probe Path(a.out).expanduser().write_text(json.dumps(doc, indent=4, ensure_ascii=False) + "\n", encoding="utf-8") n_scenes = sum(1 for s in doc["sources"] if s["id"] == "scene") print(f"[obs-export] {a.out} ({n_scenes} scenes, " f"{len(doc['sources']) - n_scenes} sources) — import via Scene Collection → Import") if a.sheet: Path(a.sheet).expanduser().write_text(setup_sheet(layout, variables, skip=skip), encoding="utf-8") print(f"[obs-export] {a.sheet}") if not a.out and not a.sheet: print(json.dumps(probe, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())