"""20-camera mosaic + framed tiles with name placeholders for the KMS wall.""" from __future__ import annotations import math import threading from typing import Iterable, Optional import cv2 import numpy as np # Compact mosaic for n live tiles (cols, rows). Landscape-first, max 5x4. _GRID_SHAPES = { 0: (1, 1), 1: (1, 1), 2: (2, 1), 3: (2, 2), 4: (2, 2), 5: (3, 2), 6: (3, 2), 7: (4, 2), 8: (4, 2), 9: (3, 3), 10: (4, 3), 11: (4, 3), 12: (4, 3), 13: (5, 3), 14: (5, 3), 15: (5, 3), 16: (4, 4), 17: (5, 4), 18: (5, 4), 19: (5, 4), 20: (5, 4), } def grid_shape(n: int, max_cols: int = 5, max_rows: int = 4) -> tuple[int, int]: """Cols/rows for `n` live tiles. Empty room still occupies 1x1.""" cap = max(1, int(max_cols) * int(max_rows)) n = max(0, min(int(n), cap)) if max_cols == 5 and max_rows == 4 and n in _GRID_SHAPES: return _GRID_SHAPES[n] if n <= 1: return 1, 1 cols = min(int(max_cols), n) rows = min(int(max_rows), math.ceil(n / cols)) while cols * rows < n and (cols < max_cols or rows < max_rows): if cols < max_cols: cols += 1 else: rows += 1 return cols, rows def describe_livekit_error(exc: BaseException) -> str: """Short on-screen reason for a LiveKit connect/session failure.""" if isinstance(exc, TimeoutError): return "connection timed out" if isinstance(exc, ConnectionRefusedError): return "connection refused" raw = str(exc) or type(exc).__name__ low = raw.lower() if "113" in raw or "no route" in low or "unreachable" in low: return "host unreachable" if "111" in raw or "refused" in low: return "connection refused" if "timed out" in low or "timeout" in low: return "connection timed out" if "name or service not known" in low or "nodename nor servname" in low: return "DNS lookup failed" if "connection reset" in low or "broken pipe" in low: return "connection reset" if "websocket" in low or "ws://" in low or "wss://" in low: return raw[:96] return raw[:96] def format_livekit_status( *, url: str, state: str, detail: str = "", retry_s: float | None = None, room: str = "", ) -> list[str]: """Lines for the KMS status overlay. Empty list means show the mosaic.""" url = (url or "").strip() st = (state or "").strip().lower() room_line = f"room {room.strip()}" if (room or "").strip() else "" if st in ("connected", "ok", ""): return [] if st == "connecting": lines = ["CONNECTING TO LIVEKIT"] if url: lines.append(url) if room_line: lines.append(room_line) return lines lines = ["LIVEKIT DISCONNECTED"] if url: lines.append(url) if room_line: lines.append(room_line) if detail: lines.append(str(detail)[:96]) if retry_s is not None: lines.append(f"retrying in {max(0, int(round(float(retry_s))))}s") return lines # Status-screen palettes are BGRA. Amber = offline, cyan = connecting. _STATUS_AMBER = (36, 158, 255, 255) _STATUS_CYAN = (255, 186, 64, 255) _STATUS_STEEL = (196, 148, 92, 255) _STATUS_INK = (18, 14, 12, 255) _STATUS_PAPER = (245, 242, 238, 255) _STATUS_MUTED = (168, 158, 150, 255) def _status_kind(title: str) -> str: t = (title or "").upper() if "CONNECTING" in t: return "connecting" if "NO LIVE" in t or "NO CAMERA" in t: return "empty" return "disconnected" def _round_rect(img: np.ndarray, x: int, y: int, w: int, h: int, r: int, color, thickness: int = -1) -> None: r = max(0, min(int(r), w // 2, h // 2)) x, y, w, h = int(x), int(y), int(w), int(h) if r <= 0: cv2.rectangle(img, (x, y), (x + w, y + h), color, thickness) return if thickness < 0: cv2.rectangle(img, (x + r, y), (x + w - r, y + h), color, -1) cv2.rectangle(img, (x, y + r), (x + w, y + h - r), color, -1) cv2.circle(img, (x + r, y + r), r, color, -1) cv2.circle(img, (x + w - r, y + r), r, color, -1) cv2.circle(img, (x + r, y + h - r), r, color, -1) cv2.circle(img, (x + w - r, y + h - r), r, color, -1) return cv2.ellipse(img, (x + r, y + r), (r, r), 180, 0, 90, color, thickness, cv2.LINE_AA) cv2.ellipse(img, (x + w - r, y + r), (r, r), 270, 0, 90, color, thickness, cv2.LINE_AA) cv2.ellipse(img, (x + r, y + h - r), (r, r), 90, 0, 90, color, thickness, cv2.LINE_AA) cv2.ellipse(img, (x + w - r, y + h - r), (r, r), 0, 0, 90, color, thickness, cv2.LINE_AA) cv2.line(img, (x + r, y), (x + w - r, y), color, thickness, cv2.LINE_AA) cv2.line(img, (x + r, y + h), (x + w - r, y + h), color, thickness, cv2.LINE_AA) cv2.line(img, (x, y + r), (x, y + h - r), color, thickness, cv2.LINE_AA) cv2.line(img, (x + w, y + r), (x + w, y + h - r), color, thickness, cv2.LINE_AA) def _text_size(text: str, scale: float, thick: int) -> tuple[int, int]: (tw, th), _ = cv2.getTextSize(text, cv2.FONT_HERSHEY_SIMPLEX, scale, thick) return tw, th def _draw_text(img: np.ndarray, text: str, x: int, y: int, scale: float, color, thick: int = 1) -> None: cv2.putText(img, text, (int(x), int(y)), cv2.FONT_HERSHEY_SIMPLEX, scale, color, thick, cv2.LINE_AA) def _draw_signal_mark(img: np.ndarray, cx: int, cy: int, accent, *, ok: bool) -> None: """Wifi-style arcs; a slash when the SFU is down.""" for rad in (18, 30, 42): cv2.ellipse(img, (cx, cy + 10), (rad, rad), 0, 225, 315, accent, 2, cv2.LINE_AA) cv2.circle(img, (cx, cy + 18), 4, accent, -1, cv2.LINE_AA) if not ok: cv2.line(img, (cx - 28, cy - 22), (cx + 28, cy + 28), (40, 40, 220, 255), 3, cv2.LINE_AA) def status_screen(width: int, height: int, lines: list[str] | None) -> np.ndarray: """Full-frame KMS status: gradient, ghost grid, accent rail, info card.""" w, h = max(64, int(width)), max(64, int(height)) raw = [str(x).strip() for x in (lines or []) if str(x).strip()] title = raw[0] if raw else "NO LIVE CAMERAS" url = next((ln for ln in raw[1:] if ln.startswith(("ws://", "wss://", "http://"))), "") retry = next((ln for ln in raw[1:] if "retry" in ln.lower()), "") room = next((ln for ln in raw[1:] if ln.lower().startswith("room ")), "") detail = next( (ln for ln in raw[1:] if ln not in (url, retry, room) and ln != title), "", ) kind = _status_kind(title) accent = {"connecting": _STATUS_CYAN, "empty": _STATUS_STEEL}.get(kind, _STATUS_AMBER) badge = {"connecting": "CONNECTING", "empty": "STANDBY"}.get(kind, "OFFLINE") if kind == "connecting" and not retry: retry = "waiting for signal" yy = np.linspace(0.0, 1.0, h, dtype=np.float32)[:, None] if kind == "connecting": top = np.array([28, 22, 16], dtype=np.float32) bot = np.array([42, 28, 14], dtype=np.float32) elif kind == "empty": top = np.array([24, 20, 18], dtype=np.float32) bot = np.array([32, 26, 22], dtype=np.float32) else: top = np.array([22, 16, 12], dtype=np.float32) bot = np.array([16, 14, 20], dtype=np.float32) img = np.empty((h, w, 4), dtype=np.uint8) rgb = top * (1.0 - yy) + bot * yy img[:, :, :3] = rgb.astype(np.uint8)[:, None, :] img[:, :, 3] = 255 # Ghost 5x4 mosaic — the wall that should be here. cols, rows, gap = 5, 4, max(4, w // 240) gx0, gy0 = int(w * 0.08), int(h * 0.12) gx1, gy1 = int(w * 0.92), int(h * 0.88) tw = max(8, (gx1 - gx0 - gap * (cols + 1)) // cols) th = max(8, (gy1 - gy0 - gap * (rows + 1)) // rows) ghost = (58, 48, 42, 255) if kind != "connecting" else (48, 50, 42, 255) for r in range(rows): for c in range(cols): x = gx0 + gap + c * (tw + gap) y = gy0 + gap + r * (th + gap) cv2.rectangle(img, (x, y), (x + tw, y + th), ghost, 1, cv2.LINE_AA) rail = max(6, w // 240) img[:, :rail] = accent s = max(0.55, min(w / 1920.0, h / 1080.0)) pad = int(48 * s) foot_h = int(58 * s) if retry else 0 body = int(188 * s) if url: body += int(72 * s) if room: body += int(72 * s) if detail: body += int(80 * s) body += foot_h + int(28 * s) cw = min(int(w * 0.62), 1080) ch = int(min(max(body, int(h * 0.34)), int(h * 0.56))) cx0 = (w - cw) // 2 cy0 = (h - ch) // 2 card = (32, 26, 24, 255) if kind != "connecting" else (36, 30, 22, 255) _round_rect(img, cx0, cy0, cw, ch, 18, card, -1) _round_rect(img, cx0, cy0, cw, ch, 18, accent, 2) img[cy0:cy0 + 6, cx0 + 18:cx0 + cw - 18] = accent x = cx0 + pad y = cy0 + int(56 * s) pill_w = int(188 * s) _round_rect(img, x, y - int(28 * s), pill_w, int(36 * s), 8, accent, -1) bw, bh = _text_size(badge, 0.52 * s, 1) _draw_text(img, badge, x + (pill_w - bw) // 2, y - int(28 * s) + bh + int(8 * s), 0.52 * s, _STATUS_INK, 1) _draw_signal_mark(img, cx0 + cw - pad - 20, y - 4, accent, ok=(kind == "connecting")) y += int(70 * s) tscale = 1.35 * s tw, th = _text_size(title, tscale, 2) while tw > cw - 2 * pad and tscale > 0.6: tscale *= 0.92 tw, th = _text_size(title, tscale, 2) _draw_text(img, title, x, y + th, tscale, _STATUS_PAPER, 2) y += th + int(22 * s) cv2.line(img, (x, y), (cx0 + cw - pad, y), (64, 54, 48, 255), 1, cv2.LINE_AA) y += int(36 * s) if url: _draw_text(img, "ENDPOINT", x, y, 0.48 * s, _STATUS_MUTED, 1) y += int(28 * s) uscale = 0.78 * s uw, uh = _text_size(url, uscale, 1) while uw > cw - 2 * pad and uscale > 0.4: uscale *= 0.92 uw, uh = _text_size(url, uscale, 1) _draw_text(img, url, x, y + uh, uscale, _STATUS_PAPER, 1) y += uh + int(28 * s) if room: _draw_text(img, "ROOM", x, y, 0.48 * s, _STATUS_MUTED, 1) y += int(28 * s) rscale = 0.9 * s rw, rh = _text_size(room, rscale, 2) while rw > cw - 2 * pad and rscale > 0.45: rscale *= 0.92 rw, rh = _text_size(room, rscale, 2) _draw_text(img, room, x, y + rh, rscale, _STATUS_PAPER, 2) y += rh + int(28 * s) if detail: _draw_text(img, "REASON", x, y, 0.48 * s, _STATUS_MUTED, 1) y += int(28 * s) _draw_text(img, detail[:72], x, y + int(22 * s), 0.78 * s, accent, 1) if retry: fy = cy0 + ch - foot_h band = (26, 20, 16, 255) if kind != "connecting" else (28, 24, 16, 255) img[fy:cy0 + ch - 8, cx0 + 10:cx0 + cw - 10] = band _draw_text(img, retry.upper(), x, fy + int(38 * s), 0.62 * s, accent, 1) corner = room.upper() if room else "CAMERA WALL" _draw_text(img, corner, int(w * 0.08), int(h * 0.07), 0.48 * s, (140, 132, 126, 255), 1) return img ROLES = ("grid", "speaker", "screenshare") # BGRA CANVAS_BG = (10, 10, 10, 255) CELL_BG = (18, 18, 18, 255) EMPTY_FILL = (36, 36, 36, 255) FRAME_COLOR = (196, 196, 196, 255) FRAME_LIVE = (168, 168, 168, 255) FRAME_RAISE = (0, 220, 255, 255) # BGRA yellow — hand raised FRAME_SPEAK = (0, 210, 40, 255) # BGRA green — LiveKit active speaker LABEL_BG = (12, 12, 12, 255) LABEL_FG = (245, 245, 245, 255) NAME_FG = (220, 220, 220, 255) def camera_ids(n: int = 20, prefix: str = "cam") -> list[str]: return [f"{prefix}-{i:02d}" for i in range(1, n + 1)] def is_camera_identity(identity: str, prefix: str = "cam") -> bool: ident = (identity or "").strip() p = prefix.rstrip("-") return ident.startswith(p + "-") or ident.startswith(p) def active_speaker_ids(speakers, *, want=None) -> list[str]: """LiveKit active_speakers_changed payload -> identity list.""" out: list[str] = [] for p in speakers: ident = (getattr(p, "identity", None) or "").strip() if not ident: continue if want is not None and not want(p): continue out.append(ident) return out def primary_grid_speaker(ids: Iterable[str], identities: Iterable[str]) -> list[str]: """Loudest LiveKit speaker that occupies a grid tile (skip rally).""" tiles = set(identities) for ident in ids: if ident in tiles: return [ident] return [] def select_grid_speakers( ranked: Iterable[tuple[str, float]], *, min_db: float = -40.0, margin_db: float = 6.0, floors: dict[str, float] | None = None, rise_db: float = 6.0, ) -> list[str]: """Keep real talkers; drop idle floor and quieter bleed of the same source. Speech is ``rise_db`` above that mic's noise floor (not a global dBFS cut). Two adjacent people stay within ``margin_db`` of each other. """ rise = max(0.0, float(rise_db)) live: list[tuple[str, float]] = [] for ident, raw in ranked: if not ident: continue db = float(raw) if floors is not None: fl = floors.get(ident) if fl is None or db < float(fl) + rise: continue elif db < float(min_db): continue live.append((ident, db)) if not live: return [] peak = max(db for _, db in live) margin = max(0.0, float(margin_db)) return [i for i, db in live if (peak - db) <= margin] def update_noise_floor( floor: float, db: float, *, speaking: bool = False, rise_db: float = 4.0, alpha: float = 0.2, ) -> float: """Follow typical idle; freeze while this hop looks like speech.""" del speaking fl = float(floor) x = float(db) a = min(1.0, max(0.0, float(alpha))) rise = max(0.0, float(rise_db)) if x >= fl + rise: return fl return (1.0 - a) * fl + a * x def confirm_speakers( counts: dict[str, int], picked: Iterable[str], *, need: int = 2, already: Iterable[str] | None = None, ) -> tuple[list[str], dict[str, int]]: """Drop one-hop idle spikes; already-on seats stay without re-arming.""" picked_l = [i for i in picked if i] already_s = set(already or ()) n_need = max(1, int(need)) new_counts: dict[str, int] = {} out: list[str] = [] for ident in picked_l: n = int(counts.get(ident, 0)) + 1 new_counts[ident] = n if n >= n_need or ident in already_s: out.append(ident) return out, new_counts def waveform_similarity( a: np.ndarray, b: np.ndarray, max_lag: int = 320, ) -> float: """Peak |normalized xcorr| over ±max_lag samples (bleed vs two talkers).""" x = np.asarray(a, dtype=np.float64).reshape(-1) y = np.asarray(b, dtype=np.float64).reshape(-1) n = min(x.size, y.size) if n < 32: return 0.0 x = x[:n] - x[:n].mean() y = y[:n] - y[:n].mean() nx = float(np.linalg.norm(x)) ny = float(np.linalg.norm(y)) if nx < 1e-9 or ny < 1e-9: return 0.0 lag = max(0, min(int(max_lag), n - 1)) corr = np.correlate(x, y, mode="full") mid = n - 1 sl = corr[mid - lag: mid + lag + 1] return float(np.max(np.abs(sl)) / (nx * ny)) def collapse_bleed( ids: Iterable[str], waves: dict[str, np.ndarray], levels: dict[str, float], *, corr_min: float = 0.4, max_lag: int = 320, ) -> list[str]: """If two hops are the same waveform, keep the louder mic only.""" ordered = [i for i in ids if i] if len(ordered) < 2: return ordered keep = set(ordered) thresh = float(corr_min) for i, a in enumerate(ordered): if a not in keep: continue wa = waves.get(a) if wa is None: continue for b in ordered[i + 1:]: if b not in keep: continue wb = waves.get(b) if wb is None: continue if waveform_similarity(wa, wb, max_lag=max_lag) < thresh: continue if float(levels.get(a, -120.0)) >= float(levels.get(b, -120.0)): keep.discard(b) else: keep.discard(a) break return [i for i in ordered if i in keep] def hold_speakers( prev: list[str], new: list[str], *, prev_peak: float, new_peak: float, now: float, hold_until: float, hold_s: float = 0.6, switch_db: float = 3.0, ) -> tuple[list[str], float]: """Stick chrome across SFU jitter; never block a second real talker.""" prev_l = list(prev) new_l = list(new) hold = max(0.0, float(hold_s)) if new_l == prev_l: if new_l: return new_l, max(hold_until, now + hold) return [], hold_until if not prev_l: return new_l, now + hold if set(prev_l) <= set(new_l): return new_l, now + hold if not new_l: if now < hold_until: return prev_l, hold_until return [], hold_until if new_peak >= prev_peak + float(switch_db) or now >= hold_until: return new_l, now + hold return prev_l, hold_until def display_name(identity: str) -> str: """Human label for a camera identity (cam-07 -> CAM-07).""" return (identity or "").strip().upper() or "CAMERA" def letterbox_bgra(src: np.ndarray, tw: int, th: int) -> np.ndarray: """Scale BGRA `src` into tw x th, preserving aspect (black bars).""" if tw < 1 or th < 1: raise ValueError("tile size must be positive") if src.ndim != 3 or src.shape[2] not in (3, 4): raise ValueError(f"expected HxWx3/4, got {src.shape}") h, w = src.shape[:2] if src.shape[2] == 3: bgra = cv2.cvtColor(src, cv2.COLOR_BGR2BGRA) else: bgra = src scale = min(tw / w, th / h) nw = max(1, int(round(w * scale))) nh = max(1, int(round(h * scale))) resized = cv2.resize(bgra, (nw, nh), interpolation=cv2.INTER_AREA) out = np.zeros((th, tw, 4), dtype=np.uint8) y = (th - nh) // 2 x = (tw - nw) // 2 out[y : y + nh, x : x + nw] = resized return out def i420_nbytes(width: int, height: int) -> int: return int(width) * int(height) * 3 // 2 def _even(n: int) -> int: n = int(n) if n <= 0: return 0 return n if n % 2 == 0 else n - 1 def letterbox_i420(src: bytes | memoryview, sw: int, sh: int, tw: int, th: int) -> bytes: """Scale packed I420 into tw x th, even UV, studio-black bars (Y=16). Y=0 is below-black and sparkles on HDMI (kmssink limited range). """ tw, th = max(2, _even(tw)), max(2, _even(th)) sw, sh = int(sw), int(sh) need = i420_nbytes(sw, sh) raw = bytes(src) if not isinstance(src, (bytes, bytearray)) else src if len(raw) < need or sw < 2 or sh < 2: raise ValueError("bad I420 source") arr = np.frombuffer(raw, dtype=np.uint8, count=need) ysz = sw * sh uvw, uvh = sw // 2, sh // 2 uvsz = uvw * uvh y = arr[:ysz].reshape(sh, sw) u = arr[ysz:ysz + uvsz].reshape(uvh, uvw) v = arr[ysz + uvsz:ysz + 2 * uvsz].reshape(uvh, uvw) scale = min(tw / sw, th / sh) nw, nh = _even(round(sw * scale)), _even(round(sh * scale)) y2 = cv2.resize(y, (nw, nh), interpolation=cv2.INTER_AREA) u2 = cv2.resize(u, (nw // 2, nh // 2), interpolation=cv2.INTER_AREA) v2 = cv2.resize(v, (nw // 2, nh // 2), interpolation=cv2.INTER_AREA) out = np.empty(i420_nbytes(tw, th), dtype=np.uint8) oy = tw * th ouv = (tw // 2) * (th // 2) out[:oy] = 16 out[oy:] = 128 yo = out[:oy].reshape(th, tw) uo = out[oy:oy + ouv].reshape(th // 2, tw // 2) vo = out[oy + ouv:].reshape(th // 2, tw // 2) x = _even((tw - nw) // 2) y0 = _even((th - nh) // 2) yo[y0:y0 + nh, x:x + nw] = y2 uo[y0 // 2:y0 // 2 + nh // 2, x // 2:x // 2 + nw // 2] = u2 vo[y0 // 2:y0 // 2 + nh // 2, x // 2:x // 2 + nw // 2] = v2 return out.tobytes() def bgra_to_i420(bgra: np.ndarray) -> bytes: """Packed I420 bytes (h*w*3//2) from HxWx4 BGRA.""" if bgra.ndim != 3 or bgra.shape[2] < 3: raise ValueError(f"expected HxWx3/4, got {bgra.shape}") bgr = bgra[:, :, :3] yuv = cv2.cvtColor(bgr, cv2.COLOR_BGR2YUV_I420) return np.ascontiguousarray(yuv).tobytes() def bgra_to_yuv_pixel(bgra: tuple[int, int, int, int] | tuple[int, int, int]) -> tuple[int, int, int]: """Sample Y,U,V of a solid BGRA color via OpenCV I420.""" img = np.empty((2, 2, 4), dtype=np.uint8) b, g, r = int(bgra[0]), int(bgra[1]), int(bgra[2]) a = int(bgra[3]) if len(bgra) > 3 else 255 img[:] = (b, g, r, a) arr = np.frombuffer(bgra_to_i420(img), dtype=np.uint8) return int(arr[0]), int(arr[4]), int(arr[5]) def bgra_from_livekit(data: memoryview | bytes, width: int, height: int) -> np.ndarray: arr = np.frombuffer(data, dtype=np.uint8) return arr.reshape((height, width, 4)).copy() def placeholder( width: int, height: int, lines: list[str], *, color: tuple[int, int, int, int] = (24, 24, 24, 255), ) -> np.ndarray: img = np.zeros((height, width, 4), dtype=np.uint8) img[:] = color if not lines: return img font = cv2.FONT_HERSHEY_SIMPLEX scale = max(0.5, min(width / 960.0, height / 540.0) * 1.2) thick = 2 if scale >= 0.8 else 1 sizes = [cv2.getTextSize(t, font, scale, thick)[0] for t in lines] total_h = sum(s[1] for s in sizes) + 12 * (len(lines) - 1) y = (height - total_h) // 2 + sizes[0][1] for text, (tw, th) in zip(lines, sizes): x = max(8, (width - tw) // 2) cv2.putText(img, text, (x, y), font, scale, (240, 240, 240, 255), thick, cv2.LINE_AA) y += th + 12 return img def _fit_text(text: str, max_w: int, max_h: int) -> tuple[float, int, tuple[int, int]]: font = cv2.FONT_HERSHEY_SIMPLEX scale = max(0.35, min(max_w / 220.0, max_h / 36.0)) thick = 2 if scale >= 0.7 else 1 for _ in range(8): (tw, th), _ = cv2.getTextSize(text, font, scale, thick) if tw <= max_w and th <= max_h: return scale, thick, (tw, th) scale *= 0.85 if scale < 0.3: break (tw, th), _ = cv2.getTextSize(text, font, scale, thick) return scale, thick, (tw, th) def _draw_centered(img: np.ndarray, text: str, box: tuple[int, int, int, int], color=NAME_FG) -> None: x0, y0, x1, y1 = box w, h = max(1, x1 - x0), max(1, y1 - y0) scale, thick, (tw, th) = _fit_text(text, w - 8, h - 6) x = x0 + max(2, (w - tw) // 2) y = y0 + max(th + 2, (h + th) // 2) cv2.putText(img, text, (x, y), cv2.FONT_HERSHEY_SIMPLEX, scale, color, thick, cv2.LINE_AA) class CameraGrid: """5x4 (default) mosaic of framed camera tiles with name placeholders.""" def __init__( self, width: int = 1920, height: int = 1080, cols: int = 5, rows: int = 4, identities: Optional[Iterable[str]] = None, gap: int = 8, border: int = 3, label_h: int = 28, chrome_out: int = 4, ) -> None: if cols < 1 or rows < 1: raise ValueError("cols/rows must be >= 1") self.width = int(width) self.height = int(height) self.max_cols = int(cols) self.max_rows = int(rows) self.cols = int(cols) self.rows = int(rows) self.gap = max(0, int(gap)) self.border = max(1, int(border)) # Extra speak/raise thickness in the gutter only (even, for I420 UV). out = min(self.gap, max(0, int(chrome_out))) self.chrome_out = out if out % 2 == 0 else out - 1 self.label_h = max(16, int(label_h)) self.identities = list(identities) if identities is not None else camera_ids(cols * rows) self._status_lines: list[str] | None = None self._live: set[str] = set() self._raised: set[str] = set() self._speaking: set[str] = set() self._tiles: dict[str, np.ndarray] = {} self._lock = threading.Lock() self._yuv_raise = bgra_to_yuv_pixel(FRAME_RAISE) self._yuv_speak = bgra_to_yuv_pixel(FRAME_SPEAK) self._canvas = np.zeros((self.height, self.width, 4), dtype=np.uint8) self._apply_layout(self.identities, cols=self.cols, rows=self.rows) def _layout_metrics(self, cols: int, rows: int) -> None: cols = max(1, int(cols)) rows = max(1, int(rows)) self.cols = cols self.rows = rows self.tile_w = max(32, (self.width - self.gap * (self.cols + 1)) // self.cols) self.tile_h = max(32, (self.height - self.gap * (self.rows + 1)) // self.rows) self.inner_w = max(1, self.tile_w - 2 * self.border) self.inner_h = max(1, self.tile_h - 2 * self.border - self.label_h) def _rebuild_chrome(self) -> None: chrome = np.frombuffer(self._render_mosaic(), dtype=np.uint8).reshape( self.height, self.width, 4) self._i420 = np.frombuffer(bgra_to_i420(chrome), dtype=np.uint8).copy() self._i420_chrome = self._i420.copy() def _apply_layout( self, identities: Iterable[str], *, cols: int | None = None, rows: int | None = None, ) -> None: ids = [i for i in identities if i] if cols is None or rows is None: cols, rows = grid_shape(len(ids), self.max_cols, self.max_rows) self.identities = ids keep = set(ids) self._live &= keep self._raised &= keep self._speaking &= keep self._layout_metrics(cols, rows) self._tiles = {ident: self._compose_cell(ident, None) for ident in ids} if self._canvas.shape != (self.height, self.width, 4): self._canvas = np.zeros((self.height, self.width, 4), dtype=np.uint8) self._rebuild_chrome() def set_identities(self, identities: Iterable[str]) -> None: """Replace the tile set and shrink/grow cols x rows to match.""" ids = [i for i in identities if i] cols, rows = grid_shape(len(ids), self.max_cols, self.max_rows) with self._lock: if ids == self.identities and cols == self.cols and rows == self.rows: return self._apply_layout(ids, cols=cols, rows=rows) def add_identity(self, identity: str) -> bool: ident = (identity or "").strip() if not ident: return False if ident in self.identities: return True cap = self.max_cols * self.max_rows if len(self.identities) >= cap: return False ids = sorted(self.identities + [ident]) self.set_identities(ids) return ident in self.identities def remove_identity(self, identity: str) -> None: ident = (identity or "").strip() if ident not in self.identities: return self.set_identities([i for i in self.identities if i != ident]) def set_status(self, lines: list[str] | None) -> None: """Full-frame overlay (LiveKit errors). None restores the mosaic.""" cleaned = [str(x) for x in lines if str(x).strip()] if lines else None with self._lock: self._status_lines = cleaned or None def _status_frame(self) -> np.ndarray: lines = self._status_lines or ["NO LIVE CAMERAS"] return status_screen(self.width, self.height, lines) def tile_origin(self, index: int) -> tuple[int, int]: r, c = divmod(int(index), self.cols) x = self.gap + c * (self.tile_w + self.gap) y = self.gap + r * (self.tile_h + self.gap) return x, y def inner_rect(self, index: int) -> tuple[int, int, int, int]: """Inner video rect (x, y, w, h) for tile index, inside the frame.""" x, y = self.tile_origin(index) return x + self.border, y + self.border, self.inner_w, self.inner_h def ensure_slot(self, identity: str) -> bool: """Bind `identity` to an empty tile so remotes can occupy the grid.""" if identity in self._tiles: return True for i, ident in enumerate(self.identities): if ident not in self._live: del self._tiles[ident] self.identities[i] = identity self._tiles[identity] = self._compose_cell(identity, None) return True return False def _compose_cell(self, identity: str, inner: Optional[np.ndarray]) -> np.ndarray: tw, th = self.tile_w, self.tile_h b = self.border cell = np.empty((th, tw, 4), dtype=np.uint8) cell[:] = CELL_BG frame = self._frame_color(identity, inner is not None) cv2.rectangle(cell, (0, 0), (tw - 1, th - 1), frame, b) # video / placeholder region iy0, ix0 = b, b iy1, ix1 = b + self.inner_h, b + self.inner_w if inner is not None: fitted = letterbox_bgra(inner, self.inner_w, self.inner_h) cell[iy0:iy1, ix0:ix1] = fitted else: cell[iy0:iy1, ix0:ix1] = EMPTY_FILL _draw_centered(cell, display_name(identity), (ix0, iy0, ix1, iy1)) # name bar under the feed, inside the frame ly0 = th - b - self.label_h cell[ly0 : th - b, b : tw - b] = LABEL_BG _draw_centered( cell, display_name(identity), (b + 2, ly0, tw - b - 2, th - b), color=LABEL_FG, ) return cell def _frame_color(self, identity: str, live: bool) -> tuple[int, int, int, int]: if identity in self._speaking: return FRAME_SPEAK if identity in self._raised: return FRAME_RAISE return FRAME_LIVE if live else FRAME_COLOR def highlighted(self, identity: str) -> bool: return identity in self._raised def speaking(self, identity: str) -> bool: return identity in self._speaking def _paint_border(self, identity: str) -> None: """Stamp I420 chrome for the current speak/raise/live winner.""" try: idx = self.identities.index(identity) except ValueError: return ox, oy = self.tile_origin(idx) color = self._frame_color(identity, identity in self._live) if color == FRAME_SPEAK: yv, uv, vv = self._yuv_speak self._fill_i420_border(self._i420, ox, oy, yv, uv, vv) elif color == FRAME_RAISE: yv, uv, vv = self._yuv_raise self._fill_i420_border(self._i420, ox, oy, yv, uv, vv) else: self._copy_i420_border(self._i420_chrome, self._i420, ox, oy) cell = self._tiles.get(identity) if cell is not None: cv2.rectangle( cell, (0, 0), (self.tile_w - 1, self.tile_h - 1), color, self.border) def set_highlight(self, identity: str, raised: bool) -> None: """Yellow tile chrome when `raised`. Live I420 path only blits the inner rect.""" if identity not in self._tiles: return want = bool(raised) with self._lock: had = identity in self._raised if want: self._raised.add(identity) else: self._raised.discard(identity) if had == want: return self._paint_border(identity) def set_speakers(self, identities: Iterable[str]) -> None: """Green tile chrome for LiveKit active speakers (grid identities only).""" want = {i for i in identities if i in self._tiles} with self._lock: previous = set(self._speaking) if want == previous: return self._speaking = want for ident in previous | want: self._paint_border(ident) def set_frame(self, identity: str, bgra: np.ndarray) -> None: if identity not in self._tiles: return self._live.add(identity) self._tiles[identity] = self._compose_cell(identity, bgra) def _chrome_geom(self, x: int, y: int) -> tuple[int, int, int, int, int]: """Outer speak/raise ring: original border plus gutter pad.""" out = self.chrome_out return x - out, y - out, self.tile_w + 2 * out, self.tile_h + 2 * out, self.border + out def _fill_i420_border( self, mosa: np.ndarray, x: int, y: int, yv: int, uv: int, vv: int, ) -> None: W, H = self.width, self.height x, y, tw, th, b = self._chrome_geom(x, y) self._stamp_i420_border(mosa, W, H, x, y, tw, th, b, yv, uv, vv) def _copy_i420_border(self, src: np.ndarray, dst: np.ndarray, x: int, y: int) -> None: W, H = self.width, self.height x, y, tw, th, b = self._chrome_geom(x, y) x0, y0 = max(0, x), max(0, y) x1, y1 = min(W, x + tw), min(H, y + th) if x1 <= x0 or y1 <= y0 or b <= 0: return oy = W * H ouv = (W // 2) * (H // 2) sy = src[:oy].reshape(H, W) su = src[oy:oy + ouv].reshape(H // 2, W // 2) sv = src[oy + ouv:].reshape(H // 2, W // 2) dy = dst[:oy].reshape(H, W) du = dst[oy:oy + ouv].reshape(H // 2, W // 2) dv = dst[oy + ouv:].reshape(H // 2, W // 2) yt1 = min(y0 + b, y1) yb0 = max(y1 - b, y0) xl1 = min(x0 + b, x1) xr0 = max(x1 - b, x0) dy[y0:yt1, x0:x1] = sy[y0:yt1, x0:x1] dy[yb0:y1, x0:x1] = sy[yb0:y1, x0:x1] dy[y0:y1, x0:xl1] = sy[y0:y1, x0:xl1] dy[y0:y1, xr0:x1] = sy[y0:y1, xr0:x1] ux0, uy0 = x0 // 2, y0 // 2 ux1, uy1 = (x1 + 1) // 2, (y1 + 1) // 2 ub = max(1, (b + 1) // 2) du[uy0:uy0 + ub, ux0:ux1] = su[uy0:uy0 + ub, ux0:ux1] du[uy1 - ub:uy1, ux0:ux1] = su[uy1 - ub:uy1, ux0:ux1] du[uy0:uy1, ux0:ux0 + ub] = su[uy0:uy1, ux0:ux0 + ub] du[uy0:uy1, ux1 - ub:ux1] = su[uy0:uy1, ux1 - ub:ux1] dv[uy0:uy0 + ub, ux0:ux1] = sv[uy0:uy0 + ub, ux0:ux1] dv[uy1 - ub:uy1, ux0:ux1] = sv[uy1 - ub:uy1, ux0:ux1] dv[uy0:uy1, ux0:ux0 + ub] = sv[uy0:uy1, ux0:ux0 + ub] dv[uy0:uy1, ux1 - ub:ux1] = sv[uy0:uy1, ux1 - ub:ux1] @staticmethod def _stamp_i420_border( mosa: np.ndarray, W: int, H: int, x: int, y: int, tw: int, th: int, b: int, yv: int, uv: int, vv: int, ) -> None: x0, y0 = max(0, x), max(0, y) x1, y1 = min(W, x + tw), min(H, y + th) if x1 <= x0 or y1 <= y0 or b <= 0: return oy = W * H ouv = (W // 2) * (H // 2) yp = mosa[:oy].reshape(H, W) up = mosa[oy:oy + ouv].reshape(H // 2, W // 2) vp = mosa[oy + ouv:].reshape(H // 2, W // 2) yt1 = min(y0 + b, y1) yb0 = max(y1 - b, y0) xl1 = min(x0 + b, x1) xr0 = max(x1 - b, x0) yp[y0:yt1, x0:x1] = yv yp[yb0:y1, x0:x1] = yv yp[y0:y1, x0:xl1] = yv yp[y0:y1, xr0:x1] = yv ux0, uy0 = x0 // 2, y0 // 2 ux1, uy1 = (x1 + 1) // 2, (y1 + 1) // 2 ub = max(1, (b + 1) // 2) up[uy0:uy0 + ub, ux0:ux1] = uv up[uy1 - ub:uy1, ux0:ux1] = uv up[uy0:uy1, ux0:ux0 + ub] = uv up[uy0:uy1, ux1 - ub:ux1] = uv vp[uy0:uy0 + ub, ux0:ux1] = vv vp[uy1 - ub:uy1, ux0:ux1] = vv vp[uy0:uy1, ux0:ux0 + ub] = vv vp[uy0:uy1, ux1 - ub:ux1] = vv def set_frame_i420(self, identity: str, data: bytes | memoryview, width: int, height: int) -> None: try: idx = self.identities.index(identity) except ValueError: return iw, ih = self.inner_w, self.inner_h fitted = letterbox_i420(data, width, height, iw, ih) with self._lock: self._live.add(identity) self._blit_i420_inner(idx, fitted) def _blit_i420_inner(self, index: int, tile: bytes) -> None: x, y, w, h = self.inner_rect(index) x, y, w, h = _even(x), _even(y), _even(w), _even(h) W, H = self.width, self.height need = i420_nbytes(w, h) if len(tile) < need: return src = np.frombuffer(tile, dtype=np.uint8, count=need) ysz = w * h uvsz = (w // 2) * (h // 2) sy = src[:ysz].reshape(h, w) su = src[ysz:ysz + uvsz].reshape(h // 2, w // 2) sv = src[ysz + uvsz:ysz + 2 * uvsz].reshape(h // 2, w // 2) mosa = self._i420 oy = W * H ouv = (W // 2) * (H // 2) mosa[:oy].reshape(H, W)[y:y + h, x:x + w] = sy mosa[oy:oy + ouv].reshape(H // 2, W // 2)[y // 2:y // 2 + h // 2, x // 2:x // 2 + w // 2] = su mosa[oy + ouv:].reshape(H // 2, W // 2)[y // 2:y // 2 + h // 2, x // 2:x // 2 + w // 2] = sv def clear(self, identity: str) -> None: if identity not in self._tiles: return self._live.discard(identity) self._raised.discard(identity) self._speaking.discard(identity) self._tiles[identity] = self._compose_cell(identity, None) try: idx = self.identities.index(identity) except ValueError: return x, y = self.tile_origin(idx) w, h = self.tile_w, self.tile_h x, y, w, h = _even(x), _even(y), _even(w), _even(h) W, H = self.width, self.height with self._lock: oy = W * H ouv = (W // 2) * (H // 2) chrome, mosa = self._i420_chrome, self._i420 mosa[:oy].reshape(H, W)[y:y + h, x:x + w] = chrome[:oy].reshape(H, W)[y:y + h, x:x + w] mosa[oy:oy + ouv].reshape(H // 2, W // 2)[y // 2:y // 2 + h // 2, x // 2:x // 2 + w // 2] = ( chrome[oy:oy + ouv].reshape(H // 2, W // 2)[y // 2:y // 2 + h // 2, x // 2:x // 2 + w // 2]) mosa[oy + ouv:].reshape(H // 2, W // 2)[y // 2:y // 2 + h // 2, x // 2:x // 2 + w // 2] = ( chrome[oy + ouv:].reshape(H // 2, W // 2)[y // 2:y // 2 + h // 2, x // 2:x // 2 + w // 2]) def render_i420(self) -> bytes: with self._lock: if self._status_lines: return bgra_to_i420(self._status_frame()) return self._i420.tobytes() def _render_mosaic(self) -> bytes: canvas = self._canvas canvas[:] = CANVAS_BG g = self.gap for i, ident in enumerate(self.identities): r, c = divmod(i, self.cols) if r >= self.rows: break y = g + r * (self.tile_h + g) x = g + c * (self.tile_w + g) tile = self._tiles[ident] canvas[y : y + self.tile_h, x : x + self.tile_w] = tile return canvas.tobytes() def render(self) -> bytes: with self._lock: if self._status_lines: return self._status_frame().tobytes() return self._render_mosaic()