root
95bb7c50ee
Publish and subscribe over wss://livekit.uni-wh.de:7800 and refuse cleartext ws://. Conference room is uwh-telhai. Includes the uncommitted encoded H.264 publish path, Rally hairpin, and KMS wall overlay.
1057 lines
38 KiB
Python
1057 lines
38 KiB
Python
"""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()
|