""" The OBS scene layout as a FILE, and a command that makes OBS match it. Until now every OBS setup this product had was written by a script that hard-coded the shape it wanted (`field_tests/walk/configure_obs.py`, `obs_boat_sources.py`). That is fine for one fixed layout and wrong for a production: the operator cannot see what is configured without opening OBS, cannot diff it, cannot put it in git, and cannot rebuild it on the second on-site box. And what OBS holds in memory is not recoverable — a scene deleted there is gone from `Untitled.json` and from OBS's own `.bak` within a minute, taking every scene item's TRANSFORM with it. That happened here: two scenes vanished mid-session and the picture-in-picture layout of one could not be restored, because nothing had ever written it down. python -m v3.broadcast.live.obs_scenes dump obs/trwc.json # OBS -> file (incl. transforms) python -m v3.broadcast.live.obs_scenes apply obs/trwc.json # file -> OBS, no restart ONE FILE, ANY PRODUCER. Where the broadcast is served from is not a property of the scene layout: the same four scenes are wanted whether the pages come from a laptop on the start vessel or from the gateway in gcloud. So a URL never names a host — it names a VARIABLE, and the host is supplied at apply time: "url": "${producer}/core/web/index.html?feed=live&res=1&adaptive=guard&resmin=1&fpsmin=24&trustedAutoplayAudio=1" ...apply --var producer=http://localhost:8766 # local sim / laptop ...apply --var producer=https://trap-v3-producer-xxxx.run.app # gcloud `vars` in the file are the defaults; `--var` overrides them. `dump` writes placeholders back for every variable it is given, so a round-trip does not silently freeze today's host into the file. `apply` is idempotent and CONSERVATIVE: it only ever touches what the file names. A scene you made by hand and never wrote down is left alone, because the alternative — a tool that deletes whatever it does not recognise — is the failure this module exists to prevent. `--prune` opts into removing scenes the file does not list, and `--dry-run` says exactly what it would do first. Editing the file and re-running `apply` is the "refresh OBS with a command" path: no OBS restart, and OBS never has to be closed (closing it is what would overwrite the file OBS itself wrote). Connection settings come from the same place the rest of the live tooling reads them — `--obs-host/--obs-port/--obs-password`, or `$OBS_WS_PASSWORD`. """ from __future__ import annotations import argparse import json import os import sys from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) from director_obs import OBSController # noqa: E402 #: Transform fields obs-websocket accepts on a WRITE. The rest of what a dump returns #: (`sourceWidth`, `sourceHeight`, `width`, `height`) is computed by OBS and is rejected if sent #: back, so a round-trip has to filter rather than echo. _WRITABLE_TRANSFORM = ( "positionX", "positionY", "rotation", "scaleX", "scaleY", "cropTop", "cropBottom", "cropLeft", "cropRight", "alignment", "boundsType", "boundsAlignment", "boundsWidth", "boundsHeight", ) #: The settings worth carrying, PER INPUT KIND. A kind listed here is one `dump` records and `apply` #: can create; a kind that is not is still listed in a dump and still placed in a scene, but its #: settings are left to whatever the machine already has. Everything OBS holds beyond these keys is #: left alone deliberately — the file states what this layout means, not every default OBS ships. #: #: BROWSER — `url` is the point; the next three are what keep a live broadcast page alive and #: audible, and they are exactly the ones an operator forgets: #: shutdown=False keep the page (and its SSE feed + presentation clock) across cuts #: restart_when_active=False do not reload it every time the scene goes on air #: reroute_audio=True commentary into the OBS mixer, not the room's default output #: #: …and `reroute_audio` alone is not enough, because THE PICTURE AND THE SOUND HAVE SEPARATE GATES. #: A `feed=live` page holds its commentary scheduler shut until a gesture unlocks the voice #: (`app.js` `commentaryReady`), and OBS's embedded Chromium delivers no `pointerdown` — ever. So the #: page ran, drew perfectly, and spoke nothing; the route `reroute_audio` opened simply had no audio #: in it (Ulf 2026-08-11, a silent YouTube stream). The programme URL therefore carries #: `trustedAutoplayAudio=1`, which is the page's own declaration that its renderer was launched with #: autoplay permission — true of OBS's CEF, and the reason that parameter exists rather than a plain #: `?audio=`: an ordinary viewer's browser must never be talked into speaking before its first #: gesture. It belongs ONLY on a source that also routes its audio, or the same commentary is spoken #: twice; `test_obs_export` pins the two together. #: #: MEDIA (`ffmpeg_source`) — `local_file` is the point, and three of the rest are what make a file #: source safe on air rather than merely working: #: close_when_inactive=True release the file handle whenever the scene is off programme, so the #: file can be REPLACED between breaks. Windows will not overwrite a file #: OBS holds open, and a syncer that cannot write is a break that never #: changes. #: clear_on_media_end=True go transparent at the end instead of freezing on the last frame — it #: is what lets a scene fall THROUGH to whatever is beneath it, so a roll #: that runs out lands on a picture rather than on a held still. #: restart_on_activate=True start from the top on every cut, so the second break of the day is not #: the tail of the first. #: #: VLC (`vlc_source`) — a playlist over a directory. Not what this layout uses (see the `Break roll` #: scene's note) and here so that a machine which HAS one can be dumped and rebuilt faithfully. _INPUT_KEYS: dict[str, tuple[str, ...]] = { "browser_source": ("url", "width", "height", "shutdown", "restart_when_active", "reroute_audio", "fps_custom", "fps", "css"), "ffmpeg_source": ("local_file", "is_local_file", "looping", "restart_on_activate", "close_when_inactive", "clear_on_media_end", "hw_decode", "speed_percent"), "vlc_source": ("playlist", "loop", "shuffle", "playback_behavior", "subtitle_enable"), } #: Settings whose value is a PATH — the ones `${var}` expansion and `dump`'s collapse apply to. Kept #: as its own list rather than inferred, because expanding a variable inside, say, a CSS block would #: be a silent rewrite of something that only looks like a path. _PATH_KEYS = ("url", "local_file") #: The browser tuple under its old name, for readers (and any caller) that want just that one. _BROWSER_KEYS = _INPUT_KEYS["browser_source"] def _attr(obj, *names, default=None): """Read a field off an obsws-python response or a plain dict — the responses expose snake_case attributes, the raw protocol uses camelCase, and the test fakes use dicts.""" for n in names: if isinstance(obj, dict) and n in obj: return obj[n] if hasattr(obj, n): return getattr(obj, n) return default def expand(text: str, variables: dict[str, str]) -> str: """`${name}` -> its value. Unknown names are left as they are rather than blanked: a URL that still visibly says `${producer}` is a fixable mistake, one silently pointing at `/core/web/...` on no host at all is a black source nobody can explain.""" out = str(text) for name, value in variables.items(): out = out.replace("${" + name + "}", str(value)) return out def collapse(text: str, variables: dict[str, str]) -> str: """The inverse, for `dump`: put `${name}` back wherever a variable's value appears, so dumping on the laptop and applying in gcloud does not need the file edited by hand. Longest value first, so a var whose value is a prefix of another's cannot win the substitution.""" out = str(text) for name, value in sorted(variables.items(), key=lambda kv: -len(str(kv[1]))): if value: out = out.replace(str(value), "${" + name + "}") return out def _scene_names_in_order(obs: OBSController) -> list[str]: """Scene names, bottom-of-the-OBS-list first, so a dump and the UI read the same way.""" resp = obs.client.get_scene_list() out = [] for s in (_attr(resp, "scenes", default=[]) or []): name = _attr(s, "sceneName", "scene_name") if name: out.append(name) out.reverse() # OBS lists top-first; a file reads better bottom-first return out def dump(obs: OBSController, variables: dict[str, str] | None = None) -> dict: """Everything needed to rebuild this OBS: canvas, scenes in order, and per item the source, its browser settings, whether it is enabled, and its TRANSFORM.""" variables = variables or {} vid = obs.client.get_video_settings() doc = { "profile": _attr(obs.client.get_profile_list(), "current_profile_name", "currentProfileName"), # The collection owns the SCENES the way the profile owns the CANVAS — a dump that recorded # only the profile would restore this layout into whatever collection happened to be loaded. "collection": _attr(obs.client.get_scene_collection_list(), "current_scene_collection_name", "currentSceneCollectionName"), "canvas": [int(_attr(vid, "base_width", "baseWidth", default=1920)), int(_attr(vid, "base_height", "baseHeight", default=1080))], "vars": dict(variables), "program": obs.current_program_scene(), "scenes": [], } for scene in _scene_names_in_order(obs): items = [] resp = obs.client.get_scene_item_list(scene) raw = list(_attr(resp, "scene_items", "sceneItems", default=[]) or []) raw.reverse() # bottom of the stack first for it in raw: source = _attr(it, "sourceName", "source_name") item_id = _attr(it, "sceneItemId", "scene_item_id") entry = { "source": source, "kind": _attr(it, "inputKind", "input_kind") or "scene", "enabled": bool(_attr(it, "sceneItemEnabled", "scene_item_enabled", default=True)), } keys = _INPUT_KEYS.get(entry["kind"]) if keys: try: st = obs.client.get_input_settings(source) settings = _attr(st, "input_settings", "inputSettings", default={}) or {} kept = {k: settings[k] for k in keys if k in settings} for k in _PATH_KEYS: if isinstance(kept.get(k), str): kept[k] = collapse(kept[k], variables) entry["settings"] = kept except Exception as e: # noqa: BLE001 — a source we cannot read is still worth listing entry["settings_error"] = str(e) try: tr = obs.client.get_scene_item_transform(scene, item_id) t = _attr(tr, "scene_item_transform", "sceneItemTransform", default={}) or {} entry["transform"] = {k: t[k] for k in _WRITABLE_TRANSFORM if k in t} except Exception: # noqa: BLE001 — transform is a nicety, not a blocker pass items.append(entry) doc["scenes"].append({"name": scene, "items": items}) return doc def _clean_transform(t: dict) -> dict: """Drop the bounds fields when there are no bounds — OBS rejects a zero-sized bounding box, which is exactly what a dump of an un-bounded item contains.""" out = {k: v for k, v in (t or {}).items() if k in _WRITABLE_TRANSFORM} if out.get("boundsType") in (None, "OBS_BOUNDS_NONE"): for k in ("boundsWidth", "boundsHeight", "boundsAlignment"): out.pop(k, None) return out #: obs-websocket validates `baseWidth`/`baseHeight` at 1..4096 and rejects anything larger, so a #: canvas wide enough for a 3x3 wall of 1080p screens (5760) CANNOT be set over the API at all. OBS #: itself is happy with it — the limit is the protocol's. Above this we go round by way of the #: profile's own config file. _WS_CANVAS_MAX = 4096 def _profile_dir(profile: str) -> Path | None: """Where OBS keeps `profile`'s settings. The directory name is a sanitised form of the profile name (`Video wall` -> `Video_wall`), so find it by reading each `basic.ini`'s own `Name=`.""" roots = [Path(os.environ.get("APPDATA", "")) / "obs-studio" / "basic" / "profiles", Path.home() / "Library" / "Application Support" / "obs-studio" / "basic" / "profiles", Path.home() / ".config" / "obs-studio" / "basic" / "profiles"] for root in roots: if not root.is_dir(): continue for d in root.iterdir(): ini = d / "basic.ini" if ini.is_file() and f"Name={profile}" in ini.read_text(encoding="utf-8", errors="replace"): return d return None def _force_canvas_via_profile(obs: OBSController, profile: str, canvas: tuple[int, int], log) -> bool: # noqa: ANN001 """Set a canvas obs-websocket will not accept, by writing the profile's `basic.ini`. OBS holds the CURRENT profile in memory and writes it out when it switches away, so editing the file underneath the live profile would simply be overwritten. The sequence that works is: park on another profile (which flushes this one to disk), edit it, switch back (which reloads it). """ import configparser # noqa: PLC0415 — only needed on this path d = _profile_dir(profile) if d is None: log(f" cannot find the config directory for profile {profile!r} — set Settings → Video by hand") return False others = [str(p) for p in (_attr(obs.client.get_profile_list(), "profiles", default=[]) or []) if str(p) != profile] if not others: log(f" {profile!r} is the only profile, so it cannot be parked to edit — set Settings → Video by hand") return False obs.client.set_current_profile(others[0]) # flush `profile` to disk ini = d / "basic.ini" cfg = configparser.ConfigParser(strict=False) cfg.optionxform = str # OBS keys are case-sensitive (BaseCX) # OBS writes basic.ini with a UTF-8 BOM; `utf-8` alone leaves it on the first line and # configparser then reports "no section headers" on a file that plainly has one. cfg.read(ini, encoding="utf-8-sig") if not cfg.has_section("Video"): cfg.add_section("Video") cfg["Video"]["BaseCX"], cfg["Video"]["BaseCY"] = str(canvas[0]), str(canvas[1]) cfg["Video"]["OutputCX"], cfg["Video"]["OutputCY"] = str(canvas[0]), str(canvas[1]) with ini.open("w", encoding="utf-8-sig") as fh: # keep the BOM OBS put there cfg.write(fh, space_around_delimiters=False) obs.client.set_current_profile(profile) # reload it, now with the wall canvas vid = obs.client.get_video_settings() got = (int(_attr(vid, "base_width", "baseWidth", default=0)), int(_attr(vid, "base_height", "baseHeight", default=0))) if got != tuple(canvas): log(f" canvas is {got[0]}x{got[1]}, wanted {canvas[0]}x{canvas[1]} — set Settings → Video by hand") return False return True #: What to call a kind in a log line. The browser wording predates the other two and is kept exactly, #: so the output of an apply that touches only browser sources is unchanged. _KIND_LABEL = {"browser_source": "browser source", "ffmpeg_source": "media source", "vlc_source": "VLC playlist"} def _target(settings: dict) -> str: """What a source POINTS AT, for a log line: the URL, the file, or the playlist's first entry. An apply's output is the only record of what it did, and `create 'break:roll'` without the path is unreadable at exactly the moment it matters — when the path is the thing that is wrong.""" for key in _PATH_KEYS: if settings.get(key): return str(settings[key]) entries = settings.get("playlist") or [] if isinstance(entries, list) and entries: first = entries[0] return str(first.get("value", "") if isinstance(first, dict) else first) return "" def _settings_differ(obs: OBSController, source: str, wanted: dict) -> bool: """True when any key we manage is missing or different on the live source. A source we cannot read is treated as different — writing is the safe direction.""" try: st = obs.client.get_input_settings(source) have = _attr(st, "input_settings", "inputSettings", default={}) or {} except Exception: # noqa: BLE001 return True return any(have.get(k) != v for k, v in wanted.items()) def validate_audio_continuity(doc: dict) -> None: """Refuse a layout that can silence or duplicate its declared programme-audio bus. OBS scene changes normally replace both picture and audio. A telemetry-camera scene avoids that by keeping the already-defined programme browser source enabled as its bottom item and covering only its picture. The rule is opt-in because an external interview or break scene is allowed to own different audio; manifests declare the scenes in which continuity is required through ``audio_continuity``. This is a structural check on the manifest, deliberately run before any OBS mutation. It turns a lost commentary/countdown/ambience mix into an apply-time error instead of an on-air discovery. """ policy = doc.get("audio_continuity") if policy is None: return if not isinstance(policy, dict): raise ValueError("audio continuity invariant failed: audio_continuity must be an object") source = str(policy.get("source") or "").strip() protected = policy.get("preserve_in_scenes") errors: list[str] = [] if not source: errors.append("source is required") if not isinstance(protected, list) or not protected: errors.append("preserve_in_scenes must name at least one scene") protected = [] scenes = {str(s.get("name") or ""): s for s in doc.get("scenes", [])} definitions = [ (scene.get("name"), item) for scene in doc.get("scenes", []) for item in scene.get("items", []) if item.get("source") == source and bool(item.get("settings")) ] if source else [] if len(definitions) != 1: errors.append(f"{source or 'the audio source'!r} must have exactly one settings definition") else: definition_scene, definition = definitions[0] settings = definition.get("settings") or {} if definition.get("kind", "browser_source") != "browser_source": errors.append(f"{source!r} must be a browser source") if settings.get("reroute_audio") is not True: errors.append(f"{source!r} must route audio into the OBS mixer") if settings.get("shutdown") is not False: errors.append(f"{source!r} must remain loaded while inactive") if settings.get("restart_when_active") is not False: errors.append(f"{source!r} must not restart on scene activation") if "trustedAutoplayAudio=1" not in str(settings.get("url") or ""): errors.append(f"{source!r} must be allowed to start programme audio in OBS") if not definition.get("enabled", True): errors.append(f"{source!r} is disabled in its defining scene {definition_scene!r}") for scene_name in protected: scene_name = str(scene_name) scene = scenes.get(scene_name) if scene is None: errors.append(f"protected scene {scene_name!r} does not exist") continue items = scene.get("items", []) bus_items = [(index, item) for index, item in enumerate(items) if item.get("source") == source] if len(bus_items) != 1: errors.append(f"{scene_name!r} must contain {source!r} exactly once") continue index, bus = bus_items[0] if index != 0: errors.append(f"{scene_name!r} must keep {source!r} as its bottom scene item") if not bus.get("enabled", True): errors.append(f"{scene_name!r} disables {source!r}") if "settings" in bus: errors.append(f"{scene_name!r} must borrow {source!r}, not redefine it") for item in items: if item is bus or item.get("kind", "browser_source") != "browser_source": continue settings = item.get("settings") or {} if settings.get("reroute_audio") is not False: errors.append( f"{scene_name!r} camera source {item.get('source')!r} must not route duplicate audio") if "trustedAutoplayAudio=1" in str(settings.get("url") or ""): errors.append( f"{scene_name!r} camera source {item.get('source')!r} must not claim autoplay audio") if errors: raise ValueError("audio continuity invariant failed:\n - " + "\n - ".join(errors)) def apply(obs: OBSController, doc: dict, *, variables: dict[str, str] | None = None, prune: bool = False, refresh: bool = False, dry_run: bool = False, log=print) -> list[str]: """Make OBS match `doc`. Returns the actions taken (or that would be taken under `dry_run`).""" validate_audio_continuity(doc) variables = {**(doc.get("vars") or {}), **(variables or {})} acted: list[str] = [] original_program = obs.current_program_scene() original_collection = _attr(obs.client.get_scene_collection_list(), "current_scene_collection_name", "currentSceneCollectionName") def did(msg: str): acted.append(msg) log(("would " if dry_run else "") + msg) canvas = tuple(doc.get("canvas") or (1920, 1080)) # THE SCENE COLLECTION, and it is not the same lever as the profile. # # A profile owns the CANVAS; a collection owns the SCENES. Collections are shared across profiles, # so a 1080p `Race cam` laid out at 1920x1080 still exists — with those same absolute transforms — # when the operator switches to a 5760x3240 wall profile, where it occupies the top-left NINTH and # black fills the rest. Measured on 2026-08-06: the panel looked like four dead cameras and the # cause was a layout that was never wrong, only shown against a canvas it was not built for. # # A file that names its own collection can therefore never leak into another canvas. Switching # collections makes OBS reload its scenes, so this runs FIRST — before the profile/canvas work # below, which must land on the collection we are actually going to edit. if doc.get("collection"): want = str(doc["collection"]) have = _attr(obs.client.get_scene_collection_list(), "scene_collections", "sceneCollections", default=[]) or [] current = _attr(obs.client.get_scene_collection_list(), "current_scene_collection_name", "currentSceneCollectionName") if want not in [str(c) for c in have]: did(f"create scene collection {want!r}") if not dry_run: obs.client.create_scene_collection(want) # OBS switches to it as it creates it elif current != want: did(f"switch to scene collection {want!r} (was {current!r})") if not dry_run: obs.client.set_current_scene_collection(want) # The CANVAS is a property of the OBS PROFILE, not of a scene — so a layout that needs its own # resolution (a 5760x3240 video wall) needs its own profile, or it silently reframes every other # scene the machine has. Only a file that names a profile gets this treatment; a plain dump does # not move anyone's settings. if doc.get("profile"): want = str(doc["profile"]) have = _attr(obs.client.get_profile_list(), "profiles", default=[]) or [] if want not in [str(p) for p in have]: did(f"create OBS profile {want!r}") if not dry_run: obs.client.create_profile(want) current = _attr(obs.client.get_profile_list(), "current_profile_name", "currentProfileName") if current != want: did(f"switch to profile {want!r} (was {current!r})") if not dry_run: obs.client.set_current_profile(want) vid = obs.client.get_video_settings() if (int(_attr(vid, "base_width", "baseWidth", default=0)) != canvas[0] or int(_attr(vid, "base_height", "baseHeight", default=0)) != canvas[1]): big = max(canvas) > _WS_CANVAS_MAX did(f"canvas -> {canvas[0]}x{canvas[1]}" + (" (via the profile's config)" if big else "")) if not dry_run and not big: # SetVideoSettings takes the frame rate too, and it is NOT optional — carry the # profile's existing fps through rather than inventing one, or resizing the canvas # would quietly re-time the output as well. obs.client.set_video_settings( int(_attr(vid, "fps_numerator", "fpsNumerator", default=30)), int(_attr(vid, "fps_denominator", "fpsDenominator", default=1)), canvas[0], canvas[1], canvas[0], canvas[1]) elif not dry_run: _force_canvas_via_profile(obs, want, canvas, log) existing_scenes = set(_scene_names_in_order(obs)) wanted = [s["name"] for s in doc.get("scenes", [])] for scene in doc.get("scenes", []): name = scene["name"] if name not in existing_scenes: did(f"create scene {name!r}") if not dry_run: obs.ensure_scene(name) existing_scenes.add(name) for index, item in enumerate(scene.get("items", [])): source = item["source"] kind = item.get("kind", "browser_source") settings = dict(item.get("settings") or {}) for key in _PATH_KEYS: if isinstance(settings.get(key), str): settings[key] = expand(settings[key], variables) # A kind this module knows how to BUILD. One it does not is still placed in the scene and # still ordered and enabled — it is only its settings that are left to the machine. managed = kind in _INPUT_KEYS # …and building it needs something to point AT. An item with no settings is a BORROWED # one — the video wall's idiom, and the break roll's bed — whose settings live on its # first use in the file, so creating it here would make an empty source pointing nowhere # and then leave the real definition unable to claim the name. Falling through instead # lands on `add_scene_item`, which places it if it exists and says so plainly if it does # not. definable = managed and bool(settings) have_input = source in obs._input_names() # noqa: SLF001 — the controller's own lookup item_id = obs._scene_item_id(name, source) if name in existing_scenes else None # noqa: SLF001 if definable and not have_input: did(f"create {_KIND_LABEL.get(kind, kind)} {source!r} in {name!r} -> {_target(settings)}") if not dry_run: # The canvas is merged in for a BROWSER source only. It is the page's viewport, # and it means nothing to a file: a media source is sized by its own frames, and # handing `ffmpeg_source` a width/height it does not have is how a setting that # OBS silently ignores ends up in a file that claims to describe the layout. create = ({"width": canvas[0], "height": canvas[1], **settings} if kind == "browser_source" else dict(settings)) obs.client.create_input(name, source, kind, create, True) item_id = obs._scene_item_id(name, source) # noqa: SLF001 else: # Only write what actually differs. Writing every time would make `--dry-run` report # a change for every source on every run, which is the same as reporting nothing: # an operator cannot tell a real edit from the tool restating the file. if managed and settings and _settings_differ(obs, source, settings): did(f"settings {source!r} -> {_target(settings)}") if not dry_run: obs.client.set_input_settings(source, settings, True) if item_id is None: did(f"add {source!r} to {name!r}") if not dry_run: try: item_id = obs.add_scene_item(name, source) except Exception as e: # noqa: BLE001 # An item can REFERENCE a source the file does not define — that is how # the wall borrows the boat cams without owning (or overwriting) their # URLs. On a machine that has not got them, say so and carry on: one # missing camera must not abort the other eight cells. log(f" {source!r} not on this machine, cell left empty: {e}") continue if dry_run or item_id is None: continue transform = _clean_transform(item.get("transform") or {}) if transform: try: obs.set_item_transform(name, item_id, transform) except Exception as e: # noqa: BLE001 — a rejected transform must not abort the sync log(f" transform {source!r} in {name!r}: {e}") try: obs.client.set_scene_item_index(name, item_id, index) obs.client.set_scene_item_enabled(name, item_id, bool(item.get("enabled", True))) except Exception as e: # noqa: BLE001 log(f" order/enable {source!r} in {name!r}: {e}") # …and only one this file DEFINES. A borrowed item is another scene's source placed # here, and reloading it reloads it everywhere — on the additive standings manifest that # is `programme`, the source carrying the picture and the voice that are ON AIR. It also # stops the full layout reloading the same page once per scene that borrows it. if refresh and kind == "browser_source" and settings: obs.refresh_browser_input(source) # A scene the file DESCRIBES ends up with exactly the items the file lists. Adding without # removing is not "apply this layout" — it is "accumulate every layout ever applied": when # the wall's cameras were re-pointed, the four sources they replaced stayed behind in the # scene, invisible under the new ones, still fetching their dead pages. Scoped to scenes the # file names; a scene it says nothing about is untouched. keep = {i["source"] for i in scene.get("items", [])} try: live = _attr(obs.client.get_scene_item_list(name), "scene_items", "sceneItems", default=[]) or [] except Exception: # noqa: BLE001 — a scene we just created may not read back yet live = [] for it in live: src = _attr(it, "sourceName", "source_name") if src in keep: continue did(f"remove {src!r} from {name!r} — not in the file") if not dry_run: try: obs.client.remove_scene_item(name, _attr(it, "sceneItemId", "scene_item_id")) except Exception as e: # noqa: BLE001 log(f" could not remove {src!r} from {name!r}: {e}") if prune: for name in sorted(existing_scenes - set(wanted)): did(f"PRUNE scene {name!r}") if not dry_run: obs.remove_scene(name) program = doc.get("program") if program and program in wanted and program != obs.current_program_scene(): did(f"program scene -> {program!r}") if not dry_run: obs.set_scene(program) elif not program and not dry_run: # Live OBS may select a newly created scene even though the manifest did not ask to change # programme. Preserve the operator's picture for an additive/partial layout such as the # emergency-only manifest. Do not cross collection boundaries: a scene name from another # collection is not a safe restoration target. current_collection = _attr(obs.client.get_scene_collection_list(), "current_scene_collection_name", "currentSceneCollectionName") if (current_collection == original_collection and original_program in existing_scenes and original_program != obs.current_program_scene()): did(f"restore program scene -> {original_program!r}") obs.set_scene(original_program) return acted def _vars(pairs: list[str] | None) -> dict[str, str]: out: dict[str, str] = {} for pair in pairs or []: if "=" not in pair: raise SystemExit(f"--var {pair!r} must be name=value") k, v = pair.split("=", 1) out[k.strip()] = v.strip() return out def _local_targets(settings: dict, variables: dict[str, str]) -> list[str]: """Every LOCAL path a source points at — a media source's file, a VLC playlist's entries.""" out: list[str] = [] local = expand(str(settings.get("local_file") or ""), variables) if local: out.append(local) entries = settings.get("playlist") for entry in (entries if isinstance(entries, list) else []): value = entry.get("value") if isinstance(entry, dict) else entry value = expand(str(value or ""), variables) if value: out.append(value) return out def preflight(doc: dict, variables: dict[str, str], *, timeout: float = 4.0, log=print) -> list[str]: """Check everything the layout POINTS AT. Returns the targets that did not answer. A browser source whose origin is not listening renders a black rectangle: no error, no fallback card, nothing in any log, and identical on screen to a camera that simply has no clip. Two of the four panel sources were once left pointing at a producer on a port nothing was serving and two at one returning HTTP 500, and the only way to find out was to read the URLs out of OBS by hand. Asking each origin once, up front, turns an hour of that into a line of output. A FILE fails in exactly the same way, so it is checked in exactly the same place. The mistake is a different one — a path that was right on the machine the layout was written on, or a mistyped `--var breakdir=` — but the picture it makes is the same black rectangle, and an empty file or an empty folder makes it too. With ONE exception, and it is the design rather than a let-off: a missing file whose item has something ENABLED BENEATH IT in the same scene does not go black, it falls through to whatever is under it. That is exactly how the break roll is built — a line-up bed with the roll laid over it — and it is also the ordinary state of that scene before the day's first clip is approved. Refusing to build the whole layout over it would make an empty break folder block the boat cameras, so it is reported and not counted. A source with nothing under it is still fatal, which is the case the typo produces. """ import urllib.error import urllib.request origins: dict[str, str] = {} # origin -> the first full URL that used it paths: dict[str, tuple[str, bool]] = {} # local path -> (source, has a bed beneath it) for scene in doc.get("scenes", []): covered = False # an enabled item earlier in the scene = drawn under for item in scene.get("items", []): settings = item.get("settings") or {} url = expand(str(settings.get("url") or ""), variables) if url.startswith(("http://", "https://")): origins.setdefault(url.split("/", 3)[0] + "//" + url.split("/", 3)[2], url) for target in _local_targets(settings, variables): source = str(item.get("source") or "?") was = paths.get(target) # The same file used twice is only covered if EVERY use is covered — the bare one is # the one that goes black. paths[target] = (was[0] if was else source, (was[1] if was else True) and covered) covered = covered or bool(item.get("enabled", True)) bad: list[str] = [] for origin, url in sorted(origins.items()): try: with urllib.request.urlopen(url, timeout=timeout) as r: # noqa: S310 — operator's own layout code = r.status except urllib.error.HTTPError as e: code = e.code except Exception as e: # noqa: BLE001 — refused, DNS, timeout … all "no" log(f" [preflight] {origin} UNREACHABLE ({type(e).__name__}) — sources on it will be black") bad.append(origin) continue if code >= 400: log(f" [preflight] {origin} HTTP {code} — sources on it will be black") bad.append(origin) else: log(f" [preflight] {origin} ok") for path, (source, covered) in sorted(paths.items()): p = Path(path) try: missing = not p.exists() empty = (not missing) and ((p.is_file() and p.stat().st_size == 0) or (p.is_dir() and not any(p.iterdir()))) except OSError as e: # a path we cannot even stat is not one that plays log(f" [preflight] {source}: cannot read {path} ({e.strerror or type(e).__name__}) — it will be black") bad.append(path) continue if not (missing or empty): log(f" [preflight] {source}: {path} ok") continue what = "nothing at" if missing else "nothing in" if covered: log(f" [preflight] {source}: {what} {path} — the scene will show what is beneath it") else: log(f" [preflight] {source}: {what} {path} — it will be black") bad.append(path) return bad def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description=__doc__.splitlines()[1]) ap.add_argument("action", choices=("dump", "apply")) ap.add_argument("file", type=Path) ap.add_argument("--var", action="append", metavar="NAME=VALUE", help="substitute ${NAME} in URLs — e.g. --var producer=http://localhost:8766 " "locally, or the Cloud Run URL in gcloud. Repeatable; overrides the file's " "own `vars`. On `dump`, values are written back as ${NAME}.") ap.add_argument("--obs-host", default="localhost") ap.add_argument("--obs-port", type=int, default=4455) ap.add_argument("--obs-password", default="") ap.add_argument("--prune", action="store_true", help="apply: also REMOVE scenes the file does not list (destructive; --dry-run first)") ap.add_argument("--refresh", action="store_true", help="apply: reload every browser source afterwards, so a page that was served " "before its server was up comes back") ap.add_argument("--dry-run", action="store_true", help="apply: print what would change, change nothing") ap.add_argument("--no-preflight", action="store_true", help="apply: skip the target check. By default every origin the layout points at " "is fetched and every local file it names is looked for, and the apply is " "REFUSED if one does not answer — a source on a dead origin, or on a file " "that is not there, renders a black rectangle that looks exactly like a " "camera with no clip, and says nothing anywhere") a = ap.parse_args(argv) variables = _vars(a.var) obs = OBSController(host=a.obs_host, port=a.obs_port, password=a.obs_password or os.environ.get("OBS_WS_PASSWORD", "")) obs.connect() try: if a.action == "dump": doc = dump(obs, variables) a.file.parent.mkdir(parents=True, exist_ok=True) a.file.write_text(json.dumps(doc, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") items = sum(len(s["items"]) for s in doc["scenes"]) print(f"wrote {a.file} — {len(doc['scenes'])} scene(s), {items} item(s), " f"canvas {doc['canvas'][0]}x{doc['canvas'][1]}") else: doc = json.loads(a.file.read_text(encoding="utf-8")) merged = {**(doc.get("vars") or {}), **variables} missing = [n for n in (doc.get("vars") or {}) if not merged.get(n)] if missing: print(f"note: no value for {', '.join(missing)} — pass --var {missing[0]}=…") if not a.no_preflight: bad = preflight(doc, {**(doc.get("vars") or {}), **variables}) if bad and not a.dry_run: print(f"refusing to apply: {len(bad)} target(s) would render black — fix them, " f"or pass --no-preflight to build the layout anyway") return 2 acted = apply(obs, doc, variables=variables, prune=a.prune, refresh=a.refresh, dry_run=a.dry_run) print(f"{'would change' if a.dry_run else 'changed'}: {len(acted)} thing(s)") finally: obs.disconnect() return 0 if __name__ == "__main__": raise SystemExit(main())