""" Audit what OBS is actually PUTTING ON SCREEN, not what it is configured to put there. Every boat-cam failure this product has had looked identical from the operator's chair — a black tile — and none of them was visible in the configuration: * the panel held boats from a heat that had finished, so the cameras it showed were not racing; * a source was still running the page it had loaded hours earlier, from before a fix; * the producer restarted, every page's feed died, and nothing brought them back. In all three the scene list, the input list and the URLs looked perfect. The only way to tell is to look at the pixels, which is what obs-websocket's `GetSourceScreenshot` gives us, and to compare what is on screen against what the FEED says is on air. python -m v3.broadcast.live.obs_audit --feed http://127.0.0.1:8766/events python -m v3.broadcast.live.obs_audit --feed … --watch 30 # a pass every 30 s Exit code is the number of findings (0 = clean), so it can gate a rehearsal checklist. """ from __future__ import annotations import argparse import base64 import io import sys import time from dataclasses import dataclass from pathlib import Path _HERE = Path(__file__).resolve().parent sys.path.insert(0, str(_HERE)) #: Below this per-channel standard deviation a tile carries no picture at all — a flat colour, which #: is what a dead page and OBS's own "no source" render as. BLANK_STDDEV = 4.0 #: A tile showing the page's "no clip for this window" CARD rather than footage. Calibrated by #: measurement, not guessed: on this product's boat-cam card a real picture reads ~47, the fallback #: card (a line of text on a dark field) reads 10-14. It is a heuristic — a genuinely dark night shot #: could land here — so it is reported as a warning that names what to look at, never as a verdict. NO_CLIP_STDDEV = 20.0 #: Two screenshots this far apart. The boat-cam card always has a walking figure and a clock with #: TENTHS, and real footage always has water, so nothing legitimately holds still this long. MOTION_GAP_S = 1.5 #: Mean absolute pixel difference below this counts as "no change" (JPEG-ish noise floor). MOTION_EPSILON = 0.5 @dataclass(frozen=True) class Finding: """One thing wrong, in the operator's language. `where` is the OBS source or scene.""" level: str # 'error' | 'warn' where: str what: str def __str__(self) -> str: return f" [{self.level:5s}] {self.where}: {self.what}" def _image_stats(png_b64: str): # noqa: ANN202 — (stddev, pixels) """(per-channel stddev, greyscale pixel list) for a base64 data URL from GetSourceScreenshot.""" from PIL import Image, ImageStat raw = png_b64.split(",", 1)[-1] im = Image.open(io.BytesIO(base64.b64decode(raw))).convert("L") data = im.get_flattened_data() if hasattr(im, "get_flattened_data") else im.getdata() return ImageStat.Stat(im).stddev[0], list(data) def _mean_abs_diff(a: list, b: list) -> float: """Average absolute difference between two equally sized greyscale frames.""" if len(a) != len(b) or not a: return 255.0 # different size ⇒ definitely changed return sum(abs(x - y) for x, y in zip(a, b, strict=True)) / len(a) def on_air_boats(feed_url: str, *, timeout: float = 20) -> tuple[str, dict]: """`(instanceId, {deviceId: label})` for the race the feed is presenting right now.""" from obs_boatcams import _boats_from_init, _instance_of, iter_live_frames base = feed_url[: -len("/events")] if feed_url.endswith("/events") else feed_url for kind, payload in iter_live_frames(base, timeout=timeout): if kind == "init": return _instance_of(payload) or "", _boats_from_init(payload) return "", {} def audit(obs, feed_url: str, *, grid_scene_prefix: str = "Boat ", composite: str = "Broadcast", motion_gap_s: float = MOTION_GAP_S, sleep=time.sleep) -> list[Finding]: """One pass. Reads the feed, then interrogates OBS — configuration first, pixels second.""" out: list[Finding] = [] try: instance, roster = on_air_boats(feed_url) except Exception as e: # noqa: BLE001 — report, don't crash return [Finding("error", "feed", f"unreachable ({type(e).__name__}: {e})")] if not roster: return [Finding("warn", "feed", "no boats on air — nothing to audit " f"(instance {instance or 'unknown'})")] client = obs._client inputs = {i["inputName"] for i in client.get_input_list().inputs} cams = {n for n in inputs if n.startswith("boatcam:")} want = {f"boatcam:{d}" for d in roster} # --- 1. does the panel hold the boats that are racing? --- for extra in sorted(cams - want): out.append(Finding("error", extra, "not in the race on air — a camera nobody is watching, " "and it is taking a tile from one that is")) for missing in sorted(want - cams): out.append(Finding("error", missing, "racing, but has no source in OBS")) # --- 2. is each source pointed at the race that is on air? --- for name in sorted(cams & want): try: url = client.get_input_settings(name).input_settings.get("url", "") except Exception: # noqa: BLE001 continue if "hls=1" in url and instance and f"instance={instance}" not in url: out.append(Finding("error", name, f"still pointed at a previous race (on air: {instance})")) # --- 3. the pixels: is there a picture, and is it moving? --- first: dict[str, list] = {} for name in sorted(cams & want): try: shot = client.get_source_screenshot(name, "png", 320, 180, -1).image_data except Exception as e: # noqa: BLE001 out.append(Finding("error", name, f"cannot be captured ({type(e).__name__})")) continue stddev, pixels = _image_stats(shot) if stddev < BLANK_STDDEV: out.append(Finding("error", name, f"blank — a flat picture (stddev {stddev:.1f}). " "A dead page and a missing source both look like this")) continue if stddev < NO_CLIP_STDDEV: out.append(Finding("warn", name, f"looks like the 'no clip for this window' card " f"(stddev {stddev:.1f}; footage reads ~47). The boat is " "racing, so either its feed is dead or its clips are not " "reaching this page")) continue # a static card is not news — don't also call it frozen first[name] = pixels if first: sleep(motion_gap_s) for name, before in first.items(): try: shot = client.get_source_screenshot(name, "png", 320, 180, -1).image_data except Exception: # noqa: BLE001 continue _sd, after = _image_stats(shot) diff = _mean_abs_diff(before, after) if diff < MOTION_EPSILON: out.append(Finding("error", name, f"FROZEN — no pixel changed in {motion_gap_s:.1f}s " f"(mean diff {diff:.2f})")) # --- 4. the programme itself --- scenes = [s["sceneName"] for s in client.get_scene_list().scenes] grids = [s for s in scenes if s.startswith(grid_scene_prefix) and "&" in s] if composite and grids: items = [i["sourceName"] for i in client.get_scene_item_list(grids[0]).scene_items] if composite not in items: out.append(Finding("warn", grids[0], f"the programme ({composite}) is not tiled in the " "panel — the operator sees the parts, not the output")) if len(grids) > 1: out.append(Finding("warn", "OBS", f"{len(grids)} team grids exist {grids} — leftovers from " "earlier heats confuse which panel is live")) return out def main() -> int: ap = argparse.ArgumentParser(description="Audit what OBS is actually showing") ap.add_argument("--feed", default="http://127.0.0.1:8766/events", help="SSE race feed URL") 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("--composite", default="Broadcast", help="programme source expected in the grid") ap.add_argument("--watch", type=float, metavar="SECONDS", help="repeat forever, this many seconds between passes") a = ap.parse_args() from director_obs import OBSController obs = OBSController(host=a.obs_host, port=a.obs_port, password=a.obs_password).connect() while True: findings = audit(obs, a.feed, composite=a.composite) stamp = time.strftime("%Y-%m-%d %H:%M:%S") if findings: print(f"{stamp} {len(findings)} finding(s)", flush=True) for f in findings: print(str(f), flush=True) else: print(f"{stamp} OBS clean — every racing boat has a moving picture", flush=True) if not a.watch: return len(findings) time.sleep(a.watch) if __name__ == "__main__": raise SystemExit(main())