""" OBS output control — §6 seam #4: presentation → output. The overlap/replay **director runs in the browser** (core/web/director.js) — it picks the on-air race, LIVE/REPLAY mode, and camera, all inside the single broadcast web page. So on the server side there is no scene-switching state machine to port; the whole broadcast is **one OBS browser source**, and this module's job is just the **OBS lifecycle**: point OBS at the live page and start/stop the YouTube RTMP stream (and recording) via obs-websocket v5. Typical use (with OBS running, obs-websocket enabled in Tools → WebSocket Server Settings): obs = OBSController(password="...") # host/port default to localhost:4455 obs.connect() obs.set_browser_source_url("Broadcast", "http://localhost:8765/core/web/index.html?feed=live") obs.start_stream() # ... broadcast ... obs.stop_stream(); obs.disconnect() or the one-shot helper `broadcast_to_youtube(...)`. `obsws-python` (the obs-websocket v5 client) is imported lazily in `connect()` so this module imports fine without it; `connect()` raises a clear install hint if it's missing. The request methods are thin wrappers over the client, so they're testable by injecting a fake client (see test_director_obs.py). RTMP/YouTube credentials live in OBS itself (Settings → Stream) — not here. Sub-phase: B-C (director + OBS + RTMP). """ from __future__ import annotations import math from collections.abc import Callable class OBSController: """Thin obs-websocket v5 controller for the broadcast's single browser source + stream lifecycle.""" def __init__(self, host: str = "localhost", port: int = 4455, password: str = "", *, client: object | None = None, client_factory: Callable[..., object] | None = None): self.host, self.port, self.password = host, port, password self._client = client # inject a ready client (tests), else built in connect() self._client_factory = client_factory # override how the real client is built (tests) self._boat_inputs: dict[str, dict] = {} # device_id -> {input, label, scenes:{scene: item_id}} def connect(self): """Open the obs-websocket connection (lazy import of obsws-python). Idempotent.""" if self._client is not None: return self factory = self._client_factory if factory is None: try: import obsws_python as obs except ImportError as e: raise RuntimeError( "obs-websocket client not installed. `pip install obsws-python`, and enable OBS's " "WebSocket server (Tools → WebSocket Server Settings)." ) from e factory = obs.ReqClient self._client = factory(host=self.host, port=self.port, password=self.password) return self @property def client(self): if self._client is None: raise RuntimeError("Not connected — call connect() first.") return self._client # --- broadcast lifecycle --- def set_browser_source_url(self, source_name: str, url: str): """Point an existing OBS Browser source at the live page (SetInputSettings, merged).""" self.client.set_input_settings(source_name, {"url": url}, True) def set_scene(self, scene_name: str): """Switch the program scene (e.g. broadcast vs. a holding card).""" self.client.set_current_program_scene(scene_name) def start_stream(self): """Begin the RTMP stream to whatever OBS has configured (YouTube).""" self.client.start_stream() def stop_stream(self): self.client.stop_stream() def start_record(self): """Record the broadcast to a local file (handy for a no-YouTube dry run).""" self.client.start_record() def stop_record(self): self.client.stop_record() # --- dynamic boat-cam sources (obs-websocket v5 create/remove) --- # The broadcast page is one Browser source; these add EXTRA Browser sources, one per live boat, # each pointing at core/web/boatcam.html?boat=. `reconcile_boat_sources` makes OBS hold # exactly the set the caller (live/obs_boat_sources.py supervisor) says are on the air, in either # or both layouts (a full-screen scene per boat, and/or a single auto-tiled "all cams" grid scene). # NOTE: the exact obsws-python method names/response fields below mirror the obs-websocket v5 # protocol but want a final check against a live OBS (same open caveat as the lifecycle calls). @staticmethod def boat_scene_name(label: str) -> str: """Operator-facing scene name for a single boat's full-screen cam (e.g. 'Boat 7').""" return f"Boat {label}" def _scene_names(self) -> set[str]: resp = self.client.get_scene_list() out: set[str] = set() for s in (getattr(resp, "scenes", None) or []): name = s.get("sceneName") if isinstance(s, dict) else ( getattr(s, "scene_name", None) or getattr(s, "sceneName", None)) if name: out.add(name) return out def ensure_scene(self, name: str): """Create a scene if it doesn't exist yet (idempotent).""" if name not in self._scene_names(): self.client.create_scene(name) def remove_scene(self, name: str): self.client.remove_scene(name) def rename_scene(self, old: str, new: str) -> bool: """SetSceneName. Returns False (no-op) when `old` doesn't exist or `new` is already taken.""" scenes = self._scene_names() if old not in scenes or new in scenes: return False self.client.set_scene_name(old, new) return True def current_program_scene(self) -> str | None: resp = self.client.get_current_program_scene() return getattr(resp, "current_program_scene_name", None) or getattr(resp, "scene_name", None) def _input_names(self) -> set[str]: """All current input names (GetInputList, unfiltered — kind=None can return nothing).""" try: resp = self.client.send("GetInputList") except Exception: resp = self.client.get_input_list() # fallback (and the test fake's path) out: set[str] = set() for inp in (getattr(resp, "inputs", None) or []): name = inp.get("inputName") if isinstance(inp, dict) else getattr(inp, "input_name", None) if name: out.add(name) return out def _scene_item_id(self, scene: str, source: str) -> int | None: """Scene-item id of `source` within `scene` (GetSceneItemList), or None if it isn't in it.""" try: resp = self.client.get_scene_item_list(scene) except Exception: return None for it in (getattr(resp, "scene_items", None) or []): nm = it.get("sourceName") if isinstance(it, dict) else getattr(it, "source_name", None) if nm == source: return it.get("sceneItemId") if isinstance(it, dict) else getattr(it, "scene_item_id", None) return None def create_browser_input(self, scene: str, name: str, url: str, width: int = 1920, height: int = 1080) -> int | None: """Create a Browser input in `scene` (CreateInput) and return its scene-item id.""" settings = {"url": url, "width": int(width), "height": int(height)} resp = self.client.create_input(scene, name, "browser_source", settings, True) return getattr(resp, "scene_item_id", None) def add_scene_item(self, scene: str, source: str) -> int | None: """Add an existing input to another scene (CreateSceneItem) and return its scene-item id.""" resp = self.client.create_scene_item(scene, source, True) return getattr(resp, "scene_item_id", None) def refresh_browser_input(self, name: str): """Reload a browser source's page (its properties' 'Refresh' button). A browser source caches whatever it loaded when created/last-loaded, so if the page server wasn't up yet it stays blank until reloaded — call this after (re)wiring a boat source so it re-fetches the now-live page.""" try: self.client.press_input_properties_button(name, "refreshnocache") except Exception: pass def _ensure_boat_item(self, scene: str, input_name: str, url: str, canvas: tuple[int, int], existing_inputs: set[str]) -> int | None: """Ensure browser input `input_name` (showing `url`) exists and is an item in `scene`; return its scene-item id. REUSES an existing input rather than recreating it: OBS keeps sources across runs and briefly reserves a just-removed name, so delete+recreate fails with CreateInput 601. For a leftover input we just refresh its URL and make sure it's in the scene; only a genuinely new boat is created.""" if input_name in existing_inputs: try: self.client.set_input_settings(input_name, {"url": url}, True) # refresh in case port moved except Exception: pass item = self._scene_item_id(scene, input_name) return item if item is not None else self.add_scene_item(scene, input_name) item = self.create_browser_input(scene, input_name, url, *canvas) existing_inputs.add(input_name) return item def remove_input(self, name: str): """Remove an input (RemoveInput) — this also drops every scene item that referenced it.""" self.client.remove_input(name) def _purge_input(self, name: str): """Reliably delete an input: first remove its scene items in every scene (releasing OBS's references), THEN remove the input. Plain RemoveInput was observed to silently no-op on a source still referenced by a scene item (it returns success but the source stays), so clear the items first. Ignores per-step errors.""" try: scenes = self._scene_names() except Exception: scenes = set() for sc in scenes: try: resp = self.client.get_scene_item_list(sc) except Exception: continue for it in (getattr(resp, "scene_items", None) or []): nm = it.get("sourceName") if isinstance(it, dict) else getattr(it, "source_name", None) if nm == name: iid = it.get("sceneItemId") if isinstance(it, dict) else getattr(it, "scene_item_id", None) try: self.client.remove_scene_item(sc, iid) except Exception: pass try: self.client.remove_input(name) except Exception: pass def remove_stale_boat_inputs(self, keep_devices, *, verbose: bool = False): """Remove `boatcam:*` inputs/scenes left over for devices NOT in this race (leftovers from a different race). The current race's inputs are deliberately LEFT IN PLACE for reuse — recreating a just-removed OBS source name is unreliable (OBS reserves it briefly), so reconcile reuses them. Call once after connecting. `verbose` prints what it sees/removes.""" keep = {f"boatcam:{d}" for d in keep_devices} inputs = sorted(self._input_names()) try: scenes = sorted(self._scene_names()) except Exception: scenes = [] if verbose: print(f" OBS inputs: {inputs}") print(f" OBS scenes: {scenes}") for name in inputs: if name.startswith("boatcam:") and name not in keep: self._purge_input(name) # scene-items-first; RemoveInput alone no-ops if verbose: print(f" removed stale input {name!r}") for name in scenes: # a boatcam:* that is actually a SCENE if name.startswith("boatcam:") and name not in keep: try: self.client.remove_scene(name) if verbose: print(f" removed stale scene {name!r}") except Exception as e: if verbose: print(f" could not remove scene {name!r}: {e}") self._boat_inputs.clear() def set_item_transform(self, scene: str, item_id: int, transform: dict): self.client.set_scene_item_transform(scene, item_id, transform) def tile_scene(self, scene: str, sources: list[str], canvas: tuple[int, int]) -> int: """Lay `sources` out as an even grid inside `scene`, in the order given. Returns how many were placed. Takes SOURCE NAMES rather than reading the controller's own bookkeeping, because a panel of fixed slots is not built from per-boat inputs and so has no bookkeeping to read. Without this the four tiles stayed at full canvas size stacked on top of each other, and the scene showed only whichever happened to be last — one camera pretending to be the panel.""" items = [] for name in sources: try: item = self._scene_item_id(scene, name) except Exception: # noqa: BLE001 — a source that isn't there is simply not tiled continue if item is not None: items.append(item) if not items: return 0 cols = math.ceil(math.sqrt(len(items))) rows = math.ceil(len(items) / cols) cw, ch = canvas[0] / cols, canvas[1] / rows for i, item_id in enumerate(items): row, col = divmod(i, cols) try: self.set_item_transform(scene, item_id, { "positionX": col * cw, "positionY": row * ch, "boundsType": "OBS_BOUNDS_SCALE_INNER", "boundsWidth": cw, "boundsHeight": ch, "boundsAlignment": 0, # 0 = OBS_ALIGN_CENTER }) except Exception: # noqa: BLE001 — one tile failing shouldn't skip the rest pass return len(items) def _relayout_grid(self, grid_scene: str, canvas: tuple[int, int], composite: str = ""): """Tile every boat cam in the grid scene into an even grid that re-flows as boats come/go. `composite` (the 3D broadcast) tiles FIRST when present, so the programme sits top-left and the cameras feeding it follow — the panel reads as one production rather than a wall of parts.""" items = [] if composite: try: cid = self._scene_item_id(grid_scene, composite) if cid is not None: items.append(cid) except Exception: # not in this scene (yet) — the caller adds it, we just tile pass items += [r["scenes"][grid_scene] for r in self._boat_inputs.values() if grid_scene in r["scenes"]] n = len(items) if not n: return cols = math.ceil(math.sqrt(n)) rows = math.ceil(n / cols) cw, ch = canvas[0] / cols, canvas[1] / rows for i, item_id in enumerate(items): row, col = divmod(i, cols) try: self.set_item_transform(grid_scene, item_id, { "positionX": col * cw, "positionY": row * ch, "boundsType": "OBS_BOUNDS_SCALE_INNER", "boundsWidth": cw, "boundsHeight": ch, "boundsAlignment": 0, # 0 = OBS_ALIGN_CENTER }) except Exception: # one tile's transform failing shouldn't skip the rest pass def reconcile_boat_sources(self, active: dict[str, str], base_url: str, *, per_boat_scenes: bool = True, grid: bool = True, grid_scene: str = "Boat Cams", canvas: tuple[int, int] = (1920, 1080), protect=(), query: str = "", composite: str = "") -> dict: """Make OBS hold exactly one boatcam Browser source per active boat. Idempotent: only the diff against the currently-managed set is applied. active : {device_id: label} for boats currently on the air (already grace-filtered upstream). base_url: the bare boatcam.html URL; `?feed=live&boat=` is appended per boat. composite: name of an existing input (the 3D broadcast) to tile in the grid BESIDE the cameras, so the panel shows the programme and its sources together. Absent from the grid ⇒ added; already there ⇒ left alone. query : extra query appended to every boat URL — this is how OBS is put on the LIVE HLS stream (`hls=1&instance=…&readBase=…`) instead of the delayed-clip surface. Kept as an opaque string so the page owns its own contract and this file needs no opinion about it. protect: device_ids that must NOT be removed even if absent from `active` (e.g. a boat whose scene is live on the program output mid-cut — defer its removal until it's off-air). Returns {"added": [...], "removed": [...], "relabelled": [...], "repointed": [...]} — "repointed" are boats whose URL was rewritten because `query` changed (a new heat). """ protect = set(protect) changes: dict[str, list] = {"added": [], "removed": [], "relabelled": [], "repointed": []} if per_boat_scenes: # ROSTER FOLLOW: teams rotate through hulls between heats (the kSail join model), so a # boat we already manage can come back under a NEW label. Its input/URL is device-keyed # and stays; only its full-screen scene is RENAMED. Two passes through temp names so # simultaneous label swaps (teams exchanging boats) can't collide. relabels = [] for did, label in active.items(): rec = self._boat_inputs.get(did) if rec is not None and rec["label"] != label: relabels.append((did, rec, label)) for did, rec, _new in relabels: self.rename_scene(self.boat_scene_name(rec["label"]), f"~relabel~{did}") for did, rec, new_label in relabels: old_sc, tmp = self.boat_scene_name(rec["label"]), f"~relabel~{did}" new_sc = self.boat_scene_name(new_label) if not self.rename_scene(tmp, new_sc): self.rename_scene(tmp, old_sc) # collision → restore the old name continue if old_sc in rec["scenes"]: rec["scenes"][new_sc] = rec["scenes"].pop(old_sc) rec["label"] = new_label changes["relabelled"].append(did) def _url(did: str) -> str: return f"{base_url}?feed=live&boat={did}" + (f"&{query}" if query else "") existing = None # input-name snapshot, fetched only if adding for did, label in active.items(): rec = self._boat_inputs.get(did) if rec is not None: # RE-POINT ACROSS HEATS. `query` names the instance whose playlist to play, and the # producer moves to the next race under us — a source created for the previous heat # would keep playing (or failing to find) that heat's stream. The URL is only ever # written at creation otherwise, so re-apply it whenever it has changed. url = _url(did) if rec.get("url") != url: try: self.set_browser_source_url(rec["input"], url) self.refresh_browser_input(rec["input"]) # reload, or the old page lingers rec["url"] = url changes["repointed"].append(did) except Exception: # a source that won't re-point is better than a lost tile pass continue if existing is None: existing = self._input_names() input_name = f"boatcam:{did}" url = _url(did) targets: list[str] = [] if per_boat_scenes: sc = self.boat_scene_name(label) self.ensure_scene(sc) targets.append(sc) if grid: self.ensure_scene(grid_scene) targets.append(grid_scene) if not targets: # no layout chosen → current scene cur = self.current_program_scene() if cur: targets.append(cur) if not targets: continue scenes = {sc: self._ensure_boat_item(sc, input_name, url, canvas, existing) for sc in targets} self.refresh_browser_input(input_name) # reload the page now the feed server is up self._boat_inputs[did] = {"input": input_name, "label": label, "scenes": scenes, "url": url} changes["added"].append(did) for did in list(self._boat_inputs): if did in active or did in protect: continue rec = self._boat_inputs.pop(did) self._purge_input(rec["input"]) # scene-items-first, then the input (reliable) if per_boat_scenes: self.remove_scene(self.boat_scene_name(rec["label"])) changes["removed"].append(did) if grid and (changes["added"] or changes["removed"]): try: if composite and self._scene_item_id(grid_scene, composite) is None: self.add_scene_item(grid_scene, composite) # the programme joins its own sources self._relayout_grid(grid_scene, canvas, composite) except Exception: # tiling is cosmetic — never fail the add/remove pass return changes def disconnect(self): """Close the connection if the client supports it.""" close = getattr(self._client, "disconnect", None) if callable(close): close() self._client = None def broadcast_to_youtube(page_url: str, *, source_name: str = "Broadcast", host: str = "localhost", port: int = 4455, password: str = "") -> OBSController: """Connect, point the named OBS Browser source at `page_url`, and start streaming. Returns the connected controller so the caller can `stop_stream()` / `disconnect()` when the broadcast ends.""" obs = OBSController(host=host, port=port, password=password).connect() obs.set_browser_source_url(source_name, page_url) obs.start_stream() return obs