root
736eccb932
Working snapshot before the OpenVINO MetricGAN GPU spike: thread-capped SpeechBrain DelaySum+MetricGAN on CPU, hop 200 ms, video hold for A/V sync, Arc VAAPI H.264, headless kmssink wall. 32 tests passing. Do not treat this as a GPU audio path.
198 lines
6.9 KiB
Python
198 lines
6.9 KiB
Python
"""20-camera mosaic + framed tiles with name placeholders for the KMS wall."""
|
|
from __future__ import annotations
|
|
|
|
from typing import Iterable, Optional
|
|
|
|
import cv2
|
|
import numpy as np
|
|
|
|
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)
|
|
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 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 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,
|
|
) -> None:
|
|
if cols < 1 or rows < 1:
|
|
raise ValueError("cols/rows must be >= 1")
|
|
self.width = int(width)
|
|
self.height = int(height)
|
|
self.cols = int(cols)
|
|
self.rows = int(rows)
|
|
self.gap = max(0, int(gap))
|
|
self.border = max(1, int(border))
|
|
self.label_h = max(16, int(label_h))
|
|
self.identities = list(identities) if identities is not None else camera_ids(cols * 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)
|
|
self._live: set[str] = set()
|
|
self._tiles: dict[str, np.ndarray] = {}
|
|
for ident in self.identities:
|
|
self._tiles[ident] = self._compose_cell(ident, None)
|
|
self._canvas = np.zeros((self.height, self.width, 4), dtype=np.uint8)
|
|
|
|
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 = FRAME_LIVE if inner is not None else FRAME_COLOR
|
|
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 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 clear(self, identity: str) -> None:
|
|
if identity not in self._tiles:
|
|
return
|
|
self._live.discard(identity)
|
|
self._tiles[identity] = self._compose_cell(identity, None)
|
|
|
|
def render(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()
|