diff --git a/.env.example b/.env.example index ea4c3ef..6b8337a 100644 --- a/.env.example +++ b/.env.example @@ -12,10 +12,15 @@ PARTICIPANT_PREFIX=cam # Streams # "all" = every discovered camera, "1,3,5" or "0-19" = subset CAMERAS=all -VIDEO_WIDTH=1280 -VIDEO_HEIGHT=720 +VIDEO_WIDTH=640 +VIDEO_HEIGHT=360 VIDEO_FPS=30 VIDEO_MIN_FPS=5 +VIDEO_BITRATE=800000 +VIDEO_CODEC=h264 +VIDEO_ENCODER=vaapi +LIBVA_DRM_DEVICE=/dev/dri/renderD129 +LIBVA_DRIVER_NAME=iHD # Audio AUDIO_SAMPLE_RATE=16000 @@ -44,9 +49,18 @@ PUBLISH_TIMEOUT_S=60 LOG_LEVEL=INFO # Headless display wall (run_displays.py / livekit-displays.service) -# 3 DRM outputs showing LiveKit participants via GStreamer kmssink +# 3 DRM roles: 20-cam grid, designated speaker camera, screenshare STUB DISPLAY_COUNT=3 -DISPLAY_IDENTITIES=cam-01,cam-02,cam-03 +DISPLAY_ROLES=grid,speaker,screenshare # Empty = auto (connected heads first, prefer Arc HDMI-A-*) DISPLAY_CONNECTORS= DISPLAY_IDENTITY=display-wall +# Speaker camera pane: cam-NN or "active" +DISPLAY_SPEAKER_CAMERA=cam-01 +DISPLAY_GRID_COLS=5 +DISPLAY_GRID_ROWS=4 +DISPLAY_GRID_WIDTH=1920 +DISPLAY_GRID_HEIGHT=1080 +DISPLAY_GRID_FPS=30 +# Screen share is a stub until production LiveKit is online. +LIVEKIT_PUBLIC_URL= diff --git a/README.md b/README.md index 9d9681f..c6e67ab 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,8 @@ LiveKit room. Each camera becomes one LiveKit participant (`cam-01` .. `cam-20`) that publishes: -- a camera video track (1280x720 @ 30 fps by default, MJPG from V4L2) +- a camera video track (640x360 @ 30 fps MJPG, H.264 VAAPI — sized + for low-latency wall decode, not max quality) - a microphone audio track (16 kHz mono, cleaned with speechbrain) Identity is the USB serial, pinned by udev (`/dev/camNN` + `slots.json`). @@ -72,28 +73,31 @@ re-run `--write` unless you intend to remap slots. ## Headless displays (LiveKit -> GStreamer kmssink) This host has no X11/Wayland. `run_displays.py` joins the LiveKit room as -`display-wall` (subscribe-only) and pipes each camera's video into a -GStreamer `kmssink` on a DRM connector. +`display-wall` (subscribe-only) and drives three DRM roles with GStreamer +`kmssink`: -Default mapping: `cam-01`, `cam-02`, `cam-03` onto the first 3 DRM heads -(connected first, Arc `xe` HDMI preferred). Today HDMI-A-7 is the live 4K -panel; plug two more HDMI cables into the Arc card (`HDMI-A-5`/`HDMI-A-6`) -and they light up on the same process. +1. **grid** — 5x4 mosaic of cam-01..cam-20 +2. **speaker** — designated speaker camera (`DISPLAY_SPEAKER_CAMERA`, default + `cam-01`; set to `active` to follow LiveKit active speaker) +3. **screenshare** — stub placeholder until the production LiveKit server + is online (no subscribe, no speaker login yet) + +Default connector order: connected Arc HDMI first (today HDMI-A-7 is live; +HDMI-A-5 / HDMI-A-6 light up when cabled). ```bash .venv/bin/python run_displays.py --list -.venv/bin/python run_displays.py --test-pattern # smpte/ball/snow, no LiveKit -.venv/bin/python run_displays.py # subscribe cam-01..03 -.venv/bin/python run_displays.py --connectors HDMI-A-7,HDMI-A-6,HDMI-A-5 +.venv/bin/python run_displays.py --test-pattern +.venv/bin/python run_displays.py +.venv/bin/python run_displays.py --speaker-camera cam-07 -sudo cp deploy/livekit-displays.service /etc/systemd/system/ -sudo systemctl daemon-reload -sudo systemctl enable --now livekit-displays +sudo systemctl restart livekit-displays +journalctl -u livekit-displays -f ``` -`DISPLAY_IDENTITIES` / `DISPLAY_CONNECTORS` / `DISPLAY_COUNT` in `.env`. -kmssink takes the DRM master for those connectors (the framebuffer console -on HDMI-A-7 will be replaced by camera video). +`.env`: `DISPLAY_ROLES`, `DISPLAY_CONNECTORS`, `DISPLAY_SPEAKER_CAMERA`, +`DISPLAY_GRID_*`. kmssink takes the DRM master (framebuffer console on +HDMI-A-7 is replaced by the grid). ## Audio cleanup diff --git a/config.py b/config.py index b6ca56e..9a69a8a 100644 --- a/config.py +++ b/config.py @@ -26,10 +26,14 @@ class Config: room: str = "cameras" participant_prefix: str = "cam" # video - width: int = 1280 - height: int = 720 + width: int = 640 + height: int = 360 fps: int = 30 min_fps: int = 5 + video_bitrate: int = 800_000 + video_codec: str = "h264" + video_encoder: str = "vaapi" # auto|software|hardware|nvenc|vaapi + vaapi_device: str = "/dev/dri/renderD129" # audio audio_rate: int = 16000 audio_channels: int = 1 @@ -46,6 +50,22 @@ class Config: publish_timeout_s: float = 60.0 log_level: str = "INFO" cameras: list[str] = field(default_factory=lambda: ["all"]) + # headless displays (LiveKit subscribe -> GStreamer kmssink) + display_count: int = 3 + display_identities: list[str] = field( + default_factory=lambda: ["cam-01", "cam-02", "cam-03"]) + display_connectors: list[str] = field(default_factory=list) + display_identity: str = "display-wall" + display_roles: list[str] = field( + default_factory=lambda: ["grid", "speaker", "screenshare"]) + display_speaker_camera: str = "cam-01" # cam-NN or "active" + display_speaker_participant: str = "speaker" + display_grid_cols: int = 5 + display_grid_rows: int = 4 + display_grid_width: int = 1920 + display_grid_height: int = 1080 + display_grid_fps: int = 30 + livekit_public_url: str = "" def _parse_cameras(spec: str) -> list[str]: @@ -66,6 +86,14 @@ def _parse_cameras(spec: str) -> list[str]: return out +def _parse_list(spec: str) -> list[str]: + """Comma-separated names, preserving case (HDMI-A-7, cam-01).""" + spec = (spec or "").strip() + if not spec: + return [] + return [p.strip() for p in spec.split(",") if p.strip()] + + def load_config(env_path: Optional[Path] = None) -> Config: _load_dotenv(Path(env_path or Path(__file__).parent / ".env")) @@ -75,10 +103,14 @@ def load_config(env_path: Optional[Path] = None) -> Config: api_secret=os.environ.get("LIVEKIT_API_SECRET", ""), room=os.environ.get("LIVEKIT_ROOM", "cameras"), participant_prefix=os.environ.get("PARTICIPANT_PREFIX", "cam"), - width=int(os.environ.get("VIDEO_WIDTH", "1280")), - height=int(os.environ.get("VIDEO_HEIGHT", "720")), + width=int(os.environ.get("VIDEO_WIDTH", "640")), + height=int(os.environ.get("VIDEO_HEIGHT", "360")), fps=int(os.environ.get("VIDEO_FPS", "30")), min_fps=int(os.environ.get("VIDEO_MIN_FPS", "5")), + video_bitrate=int(os.environ.get("VIDEO_BITRATE", "800000")), + video_codec=os.environ.get("VIDEO_CODEC", "h264").lower(), + video_encoder=os.environ.get("VIDEO_ENCODER", "vaapi").lower(), + vaapi_device=os.environ.get("LIBVA_DRM_DEVICE", "/dev/dri/renderD129"), audio_rate=int(os.environ.get("AUDIO_SAMPLE_RATE", "16000")), audio_channels=int(os.environ.get("AUDIO_CHANNELS", "1")), enhance_mode=os.environ.get("ENHANCE_MODE", "auto").lower(), @@ -93,6 +125,23 @@ def load_config(env_path: Optional[Path] = None) -> Config: publish_timeout_s=float(os.environ.get("PUBLISH_TIMEOUT_S", "60")), log_level=os.environ.get("LOG_LEVEL", "INFO").upper(), cameras=_parse_cameras(os.environ.get("CAMERAS", "all")), + display_count=int(os.environ.get("DISPLAY_COUNT", "3")), + display_identities=_parse_list( + os.environ.get("DISPLAY_IDENTITIES", "cam-01,cam-02,cam-03")), + display_connectors=_parse_list(os.environ.get("DISPLAY_CONNECTORS", "")), + display_identity=os.environ.get("DISPLAY_IDENTITY", "display-wall"), + display_roles=_parse_list( + os.environ.get("DISPLAY_ROLES", "grid,speaker,screenshare")) + or ["grid", "speaker", "screenshare"], + display_speaker_camera=os.environ.get("DISPLAY_SPEAKER_CAMERA", "cam-01"), + display_speaker_participant=os.environ.get( + "DISPLAY_SPEAKER_PARTICIPANT", "speaker"), + display_grid_cols=int(os.environ.get("DISPLAY_GRID_COLS", "5")), + display_grid_rows=int(os.environ.get("DISPLAY_GRID_ROWS", "4")), + display_grid_width=int(os.environ.get("DISPLAY_GRID_WIDTH", "1920")), + display_grid_height=int(os.environ.get("DISPLAY_GRID_HEIGHT", "1080")), + display_grid_fps=int(os.environ.get("DISPLAY_GRID_FPS", "30")), + livekit_public_url=os.environ.get("LIVEKIT_PUBLIC_URL", ""), ) if cfg.enhance_mode not in ("auto", "force", "off"): diff --git a/deploy/livekit-displays.service b/deploy/livekit-displays.service new file mode 100644 index 0000000..dab4a4c --- /dev/null +++ b/deploy/livekit-displays.service @@ -0,0 +1,24 @@ +[Unit] +Description=Drive 3 DRM displays with LiveKit camera output (GStreamer kmssink) +Documentation=file:///home/fdenkena/livekit-cameras/README.md +After=network-online.target livekit-server.service livekit-cameras.service +Wants=network-online.target livekit-cameras.service + +[Service] +Type=simple +WorkingDirectory=/home/fdenkena/livekit-cameras +EnvironmentFile=-/home/fdenkena/livekit-cameras/.env +Environment=PYTHONUNBUFFERED=1 +# Exclusive KMS access — no X/Wayland on these connectors. +ExecStartPre=/home/fdenkena/livekit-cameras/deploy/wait-for-livekit.sh +ExecStart=/home/fdenkena/livekit-cameras/.venv/bin/python /home/fdenkena/livekit-cameras/run_displays.py +ExecStop=/bin/kill -TERM $MAINPID +Restart=always +RestartSec=5 +TimeoutStartSec=60 +TimeoutStopSec=20 +KillMode=control-group +KillSignal=SIGTERM + +[Install] +WantedBy=multi-user.target diff --git a/display_grid.py b/display_grid.py new file mode 100644 index 0000000..fa42075 --- /dev/null +++ b/display_grid.py @@ -0,0 +1,197 @@ +"""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() diff --git a/drm_outputs.py b/drm_outputs.py new file mode 100644 index 0000000..741b415 --- /dev/null +++ b/drm_outputs.py @@ -0,0 +1,229 @@ +"""Discover DRM/KMS connectors for headless kmssink output.""" +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Optional + +DRM_CLASS = Path("/sys/class/drm") + +# xf86drmMode.h DRM_MODE_CONNECTOR_* +_TYPE_SYSFS = { + 10: "DP", + 11: "HDMI-A", + 12: "HDMI-B", + 14: "eDP", + 15: "Virtual", +} +_STATUS = {1: "connected", 2: "disconnected", 3: "unknown"} + + +@dataclass +class DrmOutput: + name: str # e.g. HDMI-A-7 + sysfs: str # card0-HDMI-A-7 + node: str # /dev/dri/card0 + driver: str # xe / i915 + connector_id: int + status: str + width: int = 0 + height: int = 0 + crtc_id: int = 0 + plane_id: int = 0 + bus_id: str = "" # pci:0000:03:00.0 style for kmssink + + @property + def connected(self) -> bool: + return self.status == "connected" + + +def _pci_bus_id(node: str) -> str: + link = Path(f"/sys/class/drm/{Path(node).name}/device") + try: + uevent = (link / "uevent").read_text() + except OSError: + return "" + for line in uevent.splitlines(): + if line.startswith("PCI_SLOT_NAME="): + slot = line.split("=", 1)[1].strip() + return f"pci:{slot}" + return "" + + +def _sysfs_connectors() -> dict[str, dict]: + out: dict[str, dict] = {} + if not DRM_CLASS.is_dir(): + return out + for p in sorted(DRM_CLASS.iterdir()): + name = p.name + if not name.startswith("card") or "-" not in name: + continue + card, conn = name.split("-", 1) + status = (p / "status").read_text().strip() if (p / "status").is_file() else "unknown" + modes: list[str] = [] + if (p / "modes").is_file(): + modes = [m.strip() for m in (p / "modes").read_text().splitlines() if m.strip()] + w = h = 0 + if modes and "x" in modes[0]: + try: + wh, *_ = modes[0].split("@") + w, h = (int(x) for x in wh.split("x")[:2]) + except ValueError: + pass + out[name] = { + "card": card, + "conn": conn, + "status": status, + "modes": modes, + "width": w, + "height": h, + } + return out + + +def _drm_info() -> dict: + try: + raw = subprocess.check_output(["drm_info", "-j"], text=True, timeout=10) + except (FileNotFoundError, subprocess.CalledProcessError, subprocess.TimeoutExpired): + return {} + return json.loads(raw) + + +def _plane_type(plane: dict) -> int: + props = plane.get("properties") or {} + t = props.get("type") or {} + return int(t.get("value") or t.get("raw_value") or 0) + + +def _primary_planes(info: dict) -> list[dict]: + planes = info.get("planes") or [] + return [p for p in planes if _plane_type(p) == 1] + + +def _pair_sysfs(card: str, typ_name: str, drm_conns: list[dict], sysfs: dict) -> list[tuple[dict, str]]: + names = [] + for key, meta in sysfs.items(): + if meta["card"] == card and meta["conn"].startswith(typ_name + "-"): + names.append(key) + names.sort(key=lambda k: int(sysfs[k]["conn"].rsplit("-", 1)[-1])) + drm_conns = sorted(drm_conns, key=lambda c: int(c["id"])) + return list(zip(drm_conns, names)) + + +def list_outputs() -> list[DrmOutput]: + sysfs = _sysfs_connectors() + info = _drm_info() + outputs: list[DrmOutput] = [] + for node, ninfo in info.items(): + driver = ((ninfo.get("driver") or {}).get("name")) or "" + card = Path(node).name # card0 + bus_id = _pci_bus_id(node) + crtcs = ninfo.get("crtcs") or [] + primaries = _primary_planes(ninfo) + by_type: dict[int, list[dict]] = {} + for c in ninfo.get("connectors") or []: + by_type.setdefault(int(c.get("type") or 0), []).append(c) + assigned_crtc: set[int] = set() + for typ, conns in by_type.items(): + typ_name = _TYPE_SYSFS.get(typ) + if not typ_name: + continue + for c, sys_key in _pair_sysfs(card, typ_name, conns, sysfs): + meta = sysfs[sys_key] + status = _STATUS.get(int(c.get("status") or 0), meta["status"]) or "unknown" + modes = c.get("modes") or [] + w = h = 0 + if modes: + m0 = modes[0] + w = int(m0.get("hdisplay") or 0) + h = int(m0.get("vdisplay") or 0) + if not w: + w, h = meta["width"], meta["height"] + crtc_id = 0 + plane_id = 0 + props = c.get("properties") or {} + crtc_prop = props.get("CRTC_ID") or {} + already = int(crtc_prop.get("value") or 0) + if already: + crtc_id = already + assigned_crtc.add(already) + outputs.append( + DrmOutput( + name=meta["conn"], + sysfs=sys_key, + node=node, + driver=driver, + connector_id=int(c["id"]), + status=status, + width=w, + height=h, + crtc_id=crtc_id, + plane_id=plane_id, + bus_id=bus_id, + ) + ) + # assign unused CRTCs + primary planes for disconnected heads + crtc_ids = [int(cr["id"]) for cr in crtcs] + unused = [i for i in crtc_ids if i not in assigned_crtc] + card_outs = [o for o in outputs if Path(o.node).name == card] + + def crtc_rank(o: DrmOutput) -> tuple: + return (0 if o.connected else 1, 0 if "HDMI" in o.name else 1, o.name) + + for out in sorted(card_outs, key=crtc_rank): + if not out.crtc_id and unused: + out.crtc_id = unused.pop(0) + if out.crtc_id: + try: + idx = crtc_ids.index(out.crtc_id) + except ValueError: + idx = 0 + mask = 1 << idx + for p in primaries: + if int(p.get("possible_crtcs") or 0) & mask: + out.plane_id = int(p["id"]) + break + outputs.sort(key=lambda o: (o.node, o.name)) + return outputs + + +def select_outputs( + n: int = 3, + names: Optional[list[str]] = None, + connected_only: bool = False, +) -> list[DrmOutput]: + all_outs = list_outputs() + if names: + by_name = {o.name: o for o in all_outs} + picked = [] + for name in names: + if name not in by_name: + raise KeyError(f"DRM connector {name!r} not found. Have: {sorted(by_name)}") + picked.append(by_name[name]) + return picked + connected = [o for o in all_outs if o.connected] + rest = [o for o in all_outs if not o.connected] + # Prefer the discrete GPU (xe / Arc) HDMI heads for a 3-display wall. + def rank(o: DrmOutput) -> tuple: + return ( + 0 if o.driver == "xe" else 1, + 0 if o.name.startswith("HDMI-A") else 1, + o.name, + ) + connected.sort(key=rank) + rest.sort(key=rank) + ordered = connected + ([] if connected_only else rest) + return ordered[:n] + + +def format_outputs(outs: list[DrmOutput]) -> str: + lines = [] + for o in outs: + mode = f"{o.width}x{o.height}" if o.width else "?" + lines.append( + f"{o.name:12} {o.status:12} id={o.connector_id:<4} " + f"{o.driver:4} {o.node} crtc={o.crtc_id} plane={o.plane_id} {mode}" + ) + return "\n".join(lines) diff --git a/gst_sink.py b/gst_sink.py new file mode 100644 index 0000000..19e48fe --- /dev/null +++ b/gst_sink.py @@ -0,0 +1,161 @@ +"""GStreamer kmssink pipelines for one DRM connector.""" +from __future__ import annotations + +import logging +import os +import shutil +import subprocess +from typing import Optional + +from drm_outputs import DrmOutput + +log = logging.getLogger("cameras.display") + +GST_LAUNCH = shutil.which("gst-launch-1.0") or "gst-launch-1.0" + + +def _kms_props(out: DrmOutput) -> list[str]: + props = [ + "kmssink", + f"connector-id={out.connector_id}", + "force-modesetting=true", + "can-scale=true", + "sync=false", + "restore-crtc=true", + ] + if out.driver: + props.append(f"driver-name={out.driver}") + return props + + +def test_pattern_cmd(out: DrmOutput, pattern: str = "smpte", width: int = 1280, height: int = 720) -> list[str]: + return [ + GST_LAUNCH, "-e", + "videotestsrc", "is-live=true", f"pattern={pattern}", + "!", f"video/x-raw,width={width},height={height},framerate=30/1", + "!", "videoconvert", + "!", "videoscale", + "!", *_kms_props(out), + ] + + +def appsrc_cmd(out: DrmOutput, width: int, height: int, fps: int = 30) -> list[str]: + """Raw BGRA frames on stdin -> kmssink.""" + block = width * height * 4 + return [ + GST_LAUNCH, "-e", + "fdsrc", "fd=0", f"blocksize={block}", + "!", "videoparse", + f"width={width}", f"height={height}", "format=bgra", f"framerate={fps}/1", + "!", "queue", "max-size-buffers=2", "leaky=downstream", + "!", "videoconvert", + "!", "videoscale", "method=0", + "!", *_kms_props(out), + ] + + +class KmsPipeline: + """One gst-launch kmssink process, fed raw BGRA on stdin (or self-running testsrc).""" + + def __init__(self, out: DrmOutput, identity: str): + self.out = out + self.identity = identity + self.proc: Optional[subprocess.Popen] = None + self.width = 0 + self.height = 0 + self.mode = "idle" + + @property + def alive(self) -> bool: + return self.proc is not None and self.proc.poll() is None + + def stop(self) -> None: + if self.proc is None: + return + if self.proc.stdin: + try: + self.proc.stdin.close() + except OSError: + pass + if self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(timeout=3) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait(timeout=2) + self.proc = None + self.mode = "idle" + + def start_test_pattern(self, pattern: str = "smpte") -> None: + self.stop() + if not self.out.connected: + log.warning("%s: %s not connected, skip test pattern", self.identity, self.out.name) + return + cmd = test_pattern_cmd(self.out, pattern=pattern) + log.info("%s: gst test pattern %s on %s id=%s", self.identity, pattern, self.out.name, self.out.connector_id) + self.proc = subprocess.Popen( + cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + env=_gst_env(), + ) + self.mode = "pattern" + + def start_appsrc(self, width: int, height: int, fps: int = 30) -> None: + if self.alive and self.mode == "appsrc" and self.width == width and self.height == height: + return + self.stop() + if not self.out.connected: + log.warning("%s: %s not connected, cannot start kmssink", self.identity, self.out.name) + return + cmd = appsrc_cmd(self.out, width, height, fps) + log.info( + "%s: gst appsrc %dx%d -> %s (connector %s %s)", + self.identity, width, height, self.out.name, self.out.connector_id, self.out.driver, + ) + self.proc = subprocess.Popen( + cmd, + stdin=subprocess.PIPE, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + env=_gst_env(), + bufsize=0, + ) + self.width = width + self.height = height + self.mode = "appsrc" + + def push(self, frame: bytes) -> bool: + if not self.alive or self.proc is None or self.proc.stdin is None: + return False + try: + self.proc.stdin.write(frame) + return True + except BrokenPipeError: + log.warning("%s: gst stdin closed (rc=%s)", self.identity, self.proc.poll()) + self.proc = None + return False + + def gst_errors(self) -> str: + if self.proc is None or self.proc.stderr is None: + return "" + try: + os.set_blocking(self.proc.stderr.fileno(), False) + except OSError: + pass + try: + chunk = self.proc.stderr.read() or b"" + except OSError: + return "" + return chunk.decode("utf-8", "replace") + + +def _gst_env() -> dict: + env = os.environ.copy() + # Never try X11/Wayland; kmssink is the point. + env.pop("DISPLAY", None) + env.pop("WAYLAND_DISPLAY", None) + env.setdefault("GST_GL_PLATFORM", "egl") + return env diff --git a/run.py b/run.py index 3acc038..6a3a30f 100644 --- a/run.py +++ b/run.py @@ -37,6 +37,10 @@ def worker_args_for(cfg, cam) -> list[str]: "--width", str(cfg.width), "--height", str(cfg.height), "--fps", str(cfg.fps), + "--bitrate", str(cfg.video_bitrate), + "--video-codec", cfg.video_codec, + "--video-encoder", cfg.video_encoder, + "--vaapi-device", cfg.vaapi_device, "--audio-rate", str(cfg.audio_rate), "--enhance-mode", cfg.enhance_mode, "--enhance-model", cfg.enhance_model, @@ -59,6 +63,8 @@ class Supervisor: def spawn(self, cam) -> None: args = worker_args_for(self.cfg, cam) env = dict(os.environ) + env.setdefault("LIBVA_DRIVER_NAME", "iHD") + env.setdefault("LIBVA_DRM_DEVICE", self.cfg.vaapi_device) p = subprocess.Popen( [sys.executable, str(Path(__file__).parent / "run_worker.py"), *args], diff --git a/run_displays.py b/run_displays.py new file mode 100644 index 0000000..d301db2 --- /dev/null +++ b/run_displays.py @@ -0,0 +1,348 @@ +#!/usr/bin/env python3 +"""Headless 3-display LiveKit wall via GStreamer kmssink. + + 1. grid — 5x4 mosaic of cam-01..cam-20 + 2. speaker — designated speaker camera (DISPLAY_SPEAKER_CAMERA) + 3. screenshare — STUB until the production LiveKit server is online + + .venv/bin/python run_displays.py --list + .venv/bin/python run_displays.py --test-pattern + .venv/bin/python run_displays.py +""" +from __future__ import annotations + +import argparse +import asyncio +import logging +import signal +import sys +import time +from typing import Optional +from urllib.parse import urlparse + +import numpy as np + +from config import load_config +from display_grid import ( + CameraGrid, + ROLES, + bgra_from_livekit, + camera_ids, + is_camera_identity, + placeholder, +) +from drm_outputs import DrmOutput, format_outputs, list_outputs, select_outputs +from gst_sink import KmsPipeline +from tokens import make_viewer_token + +log = logging.getLogger("cameras.display") + +_PATTERNS = ("smpte", "ball", "snow") +_PLACE_W, _PLACE_H = 1280, 720 + + +def _public_url(cfg) -> str: + if cfg.livekit_public_url: + return cfg.livekit_public_url.rstrip("/") + parsed = urlparse(cfg.url) + host = parsed.hostname or "127.0.0.1" + if host in ("127.0.0.1", "localhost"): + host = "10.200.0.20" + scheme = parsed.scheme or "ws" + port = parsed.port + if port: + return f"{scheme}://{host}:{port}" + return f"{scheme}://{host}" + + +def _bind_roles(roles: list[str], connectors: list[str]) -> list[tuple[str, DrmOutput]]: + roles = [r.strip().lower() for r in roles if r.strip()] + for r in roles: + if r not in ROLES: + raise KeyError(f"unknown display role {r!r}; expected {ROLES}") + names = connectors if connectors else None + outs = select_outputs(n=len(roles), names=names) + if len(outs) < len(roles): + log.warning("wanted %d displays, found %d DRM connectors", len(roles), len(outs)) + slots = list(zip(roles, outs)) + for role, out in slots: + log.info("role %s -> %s (%s connector-id=%s %s)", + role, out.name, out.status, out.connector_id, out.driver) + return slots + + +class Wall: + def __init__(self, cfg, slots: list[tuple[str, DrmOutput]]): + self.cfg = cfg + self.slots = {role: KmsPipeline(out, role) for role, out in slots} + prefix = cfg.participant_prefix or "cam" + n = cfg.display_grid_cols * cfg.display_grid_rows + self.grid = CameraGrid( + width=cfg.display_grid_width, + height=cfg.display_grid_height, + cols=cfg.display_grid_cols, + rows=cfg.display_grid_rows, + identities=camera_ids(n, prefix), + ) + self.speaker_camera = (cfg.display_speaker_camera or "cam-01").strip() + self.active_speaker: Optional[str] = None + self._speaker_frame: Optional[np.ndarray] = None + self._tasks: dict[str, asyncio.Task] = {} + self._stop = asyncio.Event() + # Screenshare is a stub until production LiveKit is online. + self._share_placeholder = placeholder(_PLACE_W, _PLACE_H, [ + "Screen share", + "stub — waiting for production LiveKit", + _public_url(cfg), + ]) + cam = self.speaker_camera + self._speaker_placeholder = placeholder(_PLACE_W, _PLACE_H, [ + "Speaker camera", + cam if cam.lower() != "active" else "(following active speaker)", + ]) + + def speaker_identity(self) -> Optional[str]: + if self.speaker_camera.lower() == "active": + return self.active_speaker + return self.speaker_camera + + def _push_sink(self, role: str, data: bytes, width: int, height: int, fps: int) -> None: + sink = self.slots.get(role) + if sink is None or not sink.out.connected: + return + if not sink.alive or sink.width != width or sink.height != height: + sink.start_appsrc(width, height, fps) + if not sink.push(data): + sink.start_appsrc(width, height, fps) + sink.push(data) + err = sink.gst_errors() + if err: + for line in err.strip().splitlines()[-6:]: + log.warning("%s gst: %s", role, line) + + async def _output_loops(self) -> None: + grid_fps = max(1, int(self.cfg.display_grid_fps)) + grid_period = 1.0 / grid_fps + spk_period = 1.0 / max(1, int(self.cfg.fps)) + next_grid = time.monotonic() + next_spk = time.monotonic() + while not self._stop.is_set(): + now = time.monotonic() + if now >= next_grid: + self._push_sink( + "grid", self.grid.render(), + self.grid.width, self.grid.height, grid_fps, + ) + next_grid = now + grid_period + if now >= next_spk: + spk = self._speaker_frame + if spk is None: + img = self._speaker_placeholder + else: + img = spk + self._push_sink("speaker", img.tobytes(), img.shape[1], img.shape[0], self.cfg.fps) + share = self._share_placeholder + self._push_sink("screenshare", share.tobytes(), share.shape[1], share.shape[0], self.cfg.fps) + next_spk = now + spk_period + await asyncio.sleep(0.005) + + async def _pump_camera(self, track, identity: str) -> None: + import livekit.rtc as rtc + + stream = rtc.VideoStream(track) + try: + async for event in stream: + frame = event.frame + if frame.width <= 0 or frame.height <= 0: + continue + try: + converted = frame.convert(rtc.VideoBufferType.BGRA) + except Exception as exc: + log.warning("%s: convert failed: %s", identity, exc) + continue + bgra = bgra_from_livekit(converted.data, converted.width, converted.height) + self.grid.set_frame(identity, bgra) + if identity == self.speaker_identity(): + self._speaker_frame = bgra + finally: + await stream.aclose() + self.grid.clear(identity) + if identity == self.speaker_identity(): + self._speaker_frame = None + + def _track_key(self, participant_identity: str, track) -> str: + return f"{participant_identity}:{getattr(track, 'sid', id(track))}" + + def attach(self, track, publication, participant, rtc) -> None: + ident = participant.identity + if ident == self.cfg.display_identity: + return + key = self._track_key(ident, track) + old = self._tasks.pop(key, None) + if old: + old.cancel() + # Screenshare subscribe/render is stubbed until production LiveKit. + if is_camera_identity(ident, self.cfg.participant_prefix): + self._tasks[key] = asyncio.create_task( + self._pump_camera(track, ident), name=f"cam-{ident}") + + def subscribe_if_wanted(self, participant, rtc) -> None: + ident = participant.identity + if ident == self.cfg.display_identity: + return + cam = is_camera_identity(ident, self.cfg.participant_prefix) + for pub in participant.track_publications.values(): + if pub.kind != rtc.TrackKind.KIND_VIDEO: + continue + want = cam + if want and not pub.subscribed: + log.info("subscribe %s from %s (source=%s)", pub.sid, ident, pub.source) + pub.set_subscribed(True) + + def detach(self, track, participant) -> None: + key = self._track_key(participant.identity, track) + task = self._tasks.pop(key, None) + if task: + task.cancel() + + async def run(self) -> int: + import livekit.rtc as rtc + + room = rtc.Room() + + @room.on("track_subscribed") + def _on_sub(track, publication, participant): + if track.kind == rtc.TrackKind.KIND_VIDEO: + log.info("subscribed %s from %s", publication.sid, participant.identity) + self.attach(track, publication, participant, rtc) + + @room.on("track_unsubscribed") + def _on_unsub(track, publication, participant): + self.detach(track, participant) + + @room.on("track_published") + def _on_pub(publication, participant): + self.subscribe_if_wanted(participant, rtc) + + @room.on("participant_connected") + def _on_pc(participant): + log.info("participant joined: %s", participant.identity) + self.subscribe_if_wanted(participant, rtc) + + @room.on("active_speakers_changed") + def _on_spk(speakers): + cams = [ + p.identity for p in speakers + if is_camera_identity(p.identity, self.cfg.participant_prefix) + ] + self.active_speaker = cams[0] if cams else None + + token = make_viewer_token( + self.cfg.api_key, self.cfg.api_secret, self.cfg.room, self.cfg.display_identity) + log.info("connecting to %s room=%s as %s", self.cfg.url, self.cfg.room, self.cfg.display_identity) + await room.connect(self.cfg.url, token, rtc.RoomOptions(auto_subscribe=False)) + + for rp in room.remote_participants.values(): + self.subscribe_if_wanted(rp, rtc) + for pub in rp.track_publications.values(): + if pub.track and pub.track.kind == rtc.TrackKind.KIND_VIDEO: + self.attach(pub.track, pub, rp, rtc) + + log.info( + "wall up: grid=%dx%d speaker_camera=%s screenshare=STUB", + self.grid.cols, self.grid.rows, self.speaker_camera, + ) + loops = asyncio.create_task(self._output_loops(), name="output-loops") + loop = asyncio.get_running_loop() + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, self._stop.set) + except NotImplementedError: + pass + try: + await self._stop.wait() + finally: + loops.cancel() + for t in self._tasks.values(): + t.cancel() + for sink in self.slots.values(): + sink.stop() + await room.disconnect() + return 0 + + +def run_test_pattern(slots: list[tuple[str, DrmOutput]]) -> int: + sinks = [] + for i, (role, out) in enumerate(slots): + sink = KmsPipeline(out, role) + sink.start_test_pattern(_PATTERNS[i % len(_PATTERNS)]) + sinks.append(sink) + log.info("test patterns on %d sink(s); Ctrl-C to stop", sum(1 for s in sinks if s.alive)) + stop = False + + def _stop(*_): + nonlocal stop + stop = True + + signal.signal(signal.SIGINT, _stop) + signal.signal(signal.SIGTERM, _stop) + try: + while not stop: + for s in sinks: + err = s.gst_errors() + if err: + for line in err.strip().splitlines()[-8:]: + log.warning("%s gst: %s", s.identity, line) + if any(s.out.connected and not s.alive and s.mode == "dead" for s in sinks): + return 1 + time.sleep(0.5) + finally: + for s in sinks: + s.stop() + return 0 + + +def main(argv: Optional[list[str]] = None) -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--list", action="store_true", help="list DRM connectors and exit") + ap.add_argument("--test-pattern", action="store_true", + help="drive displays with GStreamer videotestsrc (no LiveKit)") + ap.add_argument("--connectors", default="", + help="comma-separated DRM names, e.g. HDMI-A-7,HDMI-A-5,HDMI-A-6") + ap.add_argument("--roles", default="", + help="comma-separated roles (default grid,speaker,screenshare)") + ap.add_argument("--speaker-camera", default="", + help="cam-NN or 'active' (overrides DISPLAY_SPEAKER_CAMERA)") + args = ap.parse_args(argv) + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [display] %(levelname)s %(message)s", + ) + + if args.list: + print(format_outputs(list_outputs()) or "(no DRM connectors)") + return 0 + + cfg = load_config() + logging.getLogger().setLevel(getattr(logging, cfg.log_level, logging.INFO)) + if args.speaker_camera: + cfg.display_speaker_camera = args.speaker_camera + roles = [p.strip() for p in args.roles.split(",") if p.strip()] or cfg.display_roles + connectors = [p.strip() for p in args.connectors.split(",") if p.strip()] or cfg.display_connectors + try: + slots = _bind_roles(roles, connectors) + except KeyError as exc: + log.error("%s", exc) + return 2 + if not slots: + log.error("no DRM outputs to drive") + return 2 + + if args.test_pattern: + return run_test_pattern(slots) + return asyncio.run(Wall(cfg, slots).run()) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/run_worker.py b/run_worker.py index 05e8f97..623ea56 100644 --- a/run_worker.py +++ b/run_worker.py @@ -15,12 +15,19 @@ import signal import subprocess import sys import time +from collections import deque import numpy as np log = logging.getLogger("cameras.worker") +def drain_held_frames(pending: deque, now: float, hold_s: float): + """Yield frame payloads captured at least hold_s seconds ago.""" + while pending and (now - pending[0][0]) >= hold_s: + yield pending.popleft()[1] + + def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--camera-json", required=True, @@ -28,9 +35,13 @@ def main() -> int: ap.add_argument("--room", required=True) ap.add_argument("--url", required=True) ap.add_argument("--token", required=True) - ap.add_argument("--width", type=int, default=1280) - ap.add_argument("--height", type=int, default=720) + ap.add_argument("--width", type=int, default=640) + ap.add_argument("--height", type=int, default=360) ap.add_argument("--fps", type=int, default=30) + ap.add_argument("--bitrate", type=int, default=800_000) + ap.add_argument("--video-codec", default="h264") + ap.add_argument("--video-encoder", default="vaapi") + ap.add_argument("--vaapi-device", default="/dev/dri/renderD129") ap.add_argument("--audio-rate", type=int, default=16000) ap.add_argument("--enhance-mode", default="auto") ap.add_argument("--enhance-model", @@ -56,6 +67,9 @@ def main() -> int: identity = cam["identity"] video_dev = cam["video_device"] audio_card = cam.get("audio_card") + if args.vaapi_device: + os.environ.setdefault("LIBVA_DRM_DEVICE", args.vaapi_device) + os.environ.setdefault("LIBVA_DRIVER_NAME", "iHD") async def run() -> int: cleaner = AudioCleaner( @@ -81,29 +95,38 @@ def main() -> int: vproc = subprocess.Popen( ["ffmpeg", "-hide_banner", "-loglevel", "error", + "-fflags", "nobuffer", "-flags", "low_delay", + "-thread_queue_size", "4", "-f", "v4l2", "-input_format", "mjpeg", "-video_size", f"{args.width}x{args.height}", "-framerate", str(args.fps), "-i", video_dev, - "-f", "rawvideo", "-pix_fmt", "bgra", + "-f", "rawvideo", "-pix_fmt", "yuv420p", "pipe:1"], stdout=subprocess.PIPE, ) - frame_bytes = args.width * args.height * 4 + frame_bytes = args.width * args.height * 3 // 2 video_frames = 0 + # Audio is published in enhance hops (default 200 ms). Hold video by + # that same amount so lipsync at the subscriber: a frame captured at + # T is sent when the audio hop covering T is sent. + video_hold_s = float(args.enhance_hop_s) if audio_card is not None else 0.0 + pending: deque[tuple[float, bytes]] = deque() + max_pending = max(8, int(args.fps * (video_hold_s + 0.15)) + 4) + if video_hold_s > 0: + log.info("av sync: holding video %.0f ms to match audio hop", + video_hold_s * 1000.0) async def video_loop(): nonlocal video_frames loop = asyncio.get_running_loop() - last = time.monotonic() while vproc.poll() is None: data = await loop.run_in_executor( None, vproc.stdout.read, frame_bytes) if not data or len(data) < frame_bytes: if not data: break - # partial read: accumulate (v4l2/ffmpeg shouldn't do this) keep = b"" while len(keep) < frame_bytes: more = await loop.run_in_executor( @@ -114,24 +137,28 @@ def main() -> int: keep += more data = keep now = time.monotonic() - ts_us = int((now - last) * 1_000_000) - try: - video_source.capture_frame( - rtc.VideoFrame( - args.width, args.height, - rtc.VideoBufferType.BGRA, data), - timestamp_us=ts_us, - ) - video_frames += 1 - except Exception: # noqa: BLE001 - log.exception("video capture_frame failed") + pending.append((now, data)) + while len(pending) > max_pending: + pending.popleft() + for payload in drain_held_frames(pending, now, video_hold_s): + try: + video_source.capture_frame( + rtc.VideoFrame( + args.width, args.height, + rtc.VideoBufferType.I420, payload), + timestamp_us=int(time.monotonic() * 1_000_000), + ) + video_frames += 1 + except Exception: # noqa: BLE001 + log.exception("video capture_frame failed") # ---------------- audio ---------------- aproc: subprocess.Popen | None = None atrack: rtc.LocalAudioTrack | None = None if audio_card is not None: audio_source = rtc.AudioSource( - sample_rate=args.audio_rate, num_channels=1) + sample_rate=args.audio_rate, num_channels=1, + queue_size_ms=max(250, int(args.enhance_hop_s * 1000) + 50)) atrack = rtc.LocalAudioTrack.create_audio_track( f"{identity}-audio", audio_source) @@ -180,7 +207,7 @@ def main() -> int: aproc = subprocess.Popen( ["ffmpeg", "-hide_banner", "-loglevel", "error", "-fflags", "nobuffer", "-flags", "low_delay", - "-thread_queue_size", "64", + "-thread_queue_size", "8", "-f", "alsa", "-i", f"hw:{audio_card},0", "-ar", str(args.audio_rate), "-ac", "2", "-f", "s16le", "pipe:1"], @@ -188,9 +215,49 @@ def main() -> int: ) # ---------------- publish ---------------- - from livekit.rtc import TrackSource, TrackPublishOptions + from livekit.rtc import TrackSource, TrackPublishOptions, VideoEncoding, VideoCodec + from livekit.rtc._proto.room_pb2 import ( + ENCODER_BACKEND_AUTO, + ENCODER_BACKEND_SOFTWARE, + ENCODER_BACKEND_HARDWARE, + ENCODER_BACKEND_NVENC, + ENCODER_BACKEND_VAAPI, + ENCODER_BACKEND_VIDEOTOOLBOX, + DEGRADATION_PREFERENCE_MAINTAIN_RESOLUTION, + ) + encoder_map = { + "auto": ENCODER_BACKEND_AUTO, + "software": ENCODER_BACKEND_SOFTWARE, + "hardware": ENCODER_BACKEND_HARDWARE, + "nvenc": ENCODER_BACKEND_NVENC, + "vaapi": ENCODER_BACKEND_VAAPI, + "videotoolbox": ENCODER_BACKEND_VIDEOTOOLBOX, + } + codec_map = { + "vp8": VideoCodec.VP8, + "h264": VideoCodec.H264, + "av1": VideoCodec.AV1, + "vp9": VideoCodec.VP9, + "h265": VideoCodec.H265, + } + encoder = encoder_map.get(args.video_encoder, ENCODER_BACKEND_VAAPI) + codec = codec_map.get(args.video_codec, VideoCodec.H264) pub_opts = TrackPublishOptions( - source=TrackSource.SOURCE_CAMERA, simulcast=False) + source=TrackSource.SOURCE_CAMERA, + simulcast=False, + video_codec=codec, + video_encoding=VideoEncoding( + max_bitrate=int(args.bitrate), + max_framerate=int(args.fps), + ), + video_encoder=encoder, + degradation_preference=DEGRADATION_PREFERENCE_MAINTAIN_RESOLUTION, + ) + log.info( + "publish video %dx%d@%d %s encoder=%s bitrate=%d vaapi=%s", + args.width, args.height, args.fps, args.video_codec, + args.video_encoder, args.bitrate, args.vaapi_device, + ) await room.local_participant.publish_track(video_track, pub_opts) if atrack is not None: await room.local_participant.publish_track( diff --git a/scripts/dump_drm.py b/scripts/dump_drm.py new file mode 100644 index 0000000..3027680 --- /dev/null +++ b/scripts/dump_drm.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +"""Print a compact DRM connector/CRTC/plane map from drm_info -j.""" +from __future__ import annotations + +import json +import subprocess +import sys + + +def main() -> int: + raw = subprocess.check_output(["drm_info", "-j"], text=True) + data = json.loads(raw) + for node, info in data.items(): + driver = info.get("driver", {}).get("name") or info.get("driver_name") + device = info.get("device", {}) + print(f"NODE {node} driver={driver} device={device}") + conns = info.get("connectors") or info.get("Connector") or [] + if isinstance(conns, dict): + conns = list(conns.values()) + crtcs = info.get("crtcs") or [] + planes = info.get("planes") or [] + print(f" crtcs={len(crtcs)} planes={len(planes)} connectors={len(conns)}") + if conns and isinstance(conns[0], dict): + print(" connector keys:", sorted(conns[0].keys())[:20]) + for c in conns: + if not isinstance(c, dict): + print(" conn", c) + continue + cid = c.get("id") or c.get("object_id") + typ = c.get("type") or c.get("connector_type") + name = c.get("name") + st = c.get("status") + modes = c.get("modes") or [] + mode0 = None + if modes: + m = modes[0] + if isinstance(m, dict): + mode0 = f"{m.get('name') or m.get('clock')} {m.get('hdisplay')}x{m.get('vdisplay')}" + else: + mode0 = str(m)[:80] + print(f" connector id={cid} type={typ} name={name} status={st} mode0={mode0}") + if planes and isinstance(planes[0], dict): + print(" plane keys:", sorted(planes[0].keys())[:20]) + for p in planes[:8]: + print(" plane", {k: p.get(k) for k in list(p)[:12]}) + if crtcs and isinstance(crtcs[0], dict): + print(" crtc keys:", sorted(crtcs[0].keys())[:20]) + for cr in crtcs: + print(" crtc", {k: cr.get(k) for k in list(cr)[:12]}) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/speaker_invite.py b/scripts/speaker_invite.py new file mode 100644 index 0000000..c4e101f --- /dev/null +++ b/scripts/speaker_invite.py @@ -0,0 +1,60 @@ +#!/usr/bin/env python3 +"""Print a LiveKit Meet URL + token so a speaker can join and share a screen. + +STUB / future: do not use until the production LiveKit server is online. +The third display currently shows a placeholder only. +""" +from __future__ import annotations + +import argparse +import sys +from pathlib import Path +from urllib.parse import quote, urlparse + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from config import load_config +from tokens import make_speaker_token + + +def public_url(cfg) -> str: + if cfg.livekit_public_url: + return cfg.livekit_public_url.rstrip("/") + parsed = urlparse(cfg.url) + host = parsed.hostname or "127.0.0.1" + if host in ("127.0.0.1", "localhost"): + host = "10.200.0.20" + scheme = parsed.scheme or "ws" + port = parsed.port + if port: + return f"{scheme}://{host}:{port}" + return f"{scheme}://{host}" + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--identity", default="", help="default: DISPLAY_SPEAKER_PARTICIPANT") + ap.add_argument("--name", default="Speaker") + args = ap.parse_args() + cfg = load_config() + identity = args.identity or cfg.display_speaker_participant or "speaker" + token = make_speaker_token(cfg.api_key, cfg.api_secret, cfg.room, identity, args.name) + url = public_url(cfg) + meet = ( + "https://meet.livekit.io/custom" + f"?liveKitUrl={quote(url, safe='')}&token={quote(token, safe='')}" + ) + print(f"room: {cfg.room}") + print(f"identity: {identity}") + print(f"url: {url}") + print(f"token: {token}") + print() + print("Open this on the speaker laptop, then Share screen:") + print(meet) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_av_sync.py b/tests/test_av_sync.py new file mode 100644 index 0000000..ee653ec --- /dev/null +++ b/tests/test_av_sync.py @@ -0,0 +1,23 @@ +"""A/V hold queue: video is delayed to match the audio hop.""" +from collections import deque + +from run_worker import drain_held_frames + + +def test_drain_nothing_until_hold(): + q = deque([(0.0, b"a"), (0.05, b"b")]) + assert list(drain_held_frames(q, now=0.1, hold_s=0.2)) == [] + assert len(q) == 2 + + +def test_drain_ready_frames(): + q = deque([(0.0, b"a"), (0.05, b"b"), (0.25, b"c")]) + got = list(drain_held_frames(q, now=0.26, hold_s=0.2)) + assert got == [b"a", b"b"] + assert list(q) == [(0.25, b"c")] + + +def test_no_hold_drains_all(): + q = deque([(1.0, b"x")]) + assert list(drain_held_frames(q, now=1.0, hold_s=0.0)) == [b"x"] + assert not q diff --git a/tests/test_display_grid.py b/tests/test_display_grid.py new file mode 100644 index 0000000..1e8cab8 --- /dev/null +++ b/tests/test_display_grid.py @@ -0,0 +1,81 @@ +"""5x4 camera mosaic helpers.""" +import numpy as np + +from display_grid import ( + CameraGrid, + camera_ids, + display_name, + is_camera_identity, + letterbox_bgra, + placeholder, +) + + +def test_camera_ids(): + ids = camera_ids(20) + assert ids[0] == "cam-01" and ids[-1] == "cam-20" + assert len(ids) == 20 + + +def test_is_camera_identity(): + assert is_camera_identity("cam-07") + assert is_camera_identity("cam-07", "cam") + assert not is_camera_identity("speaker") + assert not is_camera_identity("display-wall") + + +def test_display_name(): + assert display_name("cam-07") == "CAM-07" + assert display_name("cam-01") == "CAM-01" + + +def test_letterbox_preserves_canvas_size(): + src = np.zeros((720, 1280, 4), dtype=np.uint8) + src[:] = (0, 0, 255, 255) + out = letterbox_bgra(src, 384, 270) + assert out.shape == (270, 384, 4) + assert out.dtype == np.uint8 + + +def test_grid_render_size_and_slots(): + g = CameraGrid(width=1920, height=1080, cols=5, rows=4) + assert g.tile_w == (1920 - g.gap * 6) // 5 + assert g.tile_h == (1080 - g.gap * 5) // 4 + assert g.inner_w < g.tile_w + assert g.inner_h < g.tile_h + assert len(g.identities) == 20 + buf = g.render() + assert len(buf) == 1920 * 1080 * 4 + frame = np.zeros((720, 1280, 4), dtype=np.uint8) + frame[:] = (255, 0, 0, 255) + g.set_frame("cam-07", frame) + buf2 = g.render() + assert buf2 != buf + g.clear("cam-07") + assert len(g.render()) == len(buf) + + +def test_empty_tiles_have_name_placeholder(): + g = CameraGrid(width=1920, height=1080, cols=5, rows=4) + # Name is drawn in the inner region — not a flat fill. + tile = g._tiles["cam-01"] + assert tile.shape == (g.tile_h, g.tile_w, 4) + assert tile.sum() > 0 + inner = tile[g.border : g.border + g.inner_h, g.border : g.border + g.inner_w] + # Center of an empty tile is not uniform (text glyphs). + assert inner.std() > 1.0 + + +def test_gap_separates_tiles(): + g = CameraGrid(width=1920, height=1080, cols=5, rows=4, gap=8) + canvas = np.frombuffer(g.render(), dtype=np.uint8).reshape(1080, 1920, 4) + # Pixel in the gutter between col 0 and col 1 is canvas background, not a tile. + x = g.gap + g.tile_w + g.gap // 2 + y = g.gap + 4 + assert tuple(canvas[y, x]) == (10, 10, 10, 255) + + +def test_placeholder_draws_text(): + img = placeholder(640, 360, ["Screen share", "stub"]) + assert img.shape == (360, 640, 4) + assert img.sum() > 0 diff --git a/tests/test_drm_outputs.py b/tests/test_drm_outputs.py new file mode 100644 index 0000000..91182e2 --- /dev/null +++ b/tests/test_drm_outputs.py @@ -0,0 +1,21 @@ +"""DRM connector discovery for the 3-display LiveKit wall.""" +from drm_outputs import list_outputs, select_outputs + + +def test_lists_arc_hdmi(): + outs = list_outputs() + assert outs, "expected at least one DRM connector" + names = {o.name for o in outs} + assert "HDMI-A-7" in names + hdmi7 = next(o for o in outs if o.name == "HDMI-A-7") + assert hdmi7.connector_id > 0 + assert hdmi7.driver in ("xe", "i915") + assert hdmi7.node.startswith("/dev/dri/card") + + +def test_select_three_prefers_xe_hdmi(): + three = select_outputs(n=3) + assert len(three) == 3 + assert three[0].name == "HDMI-A-7" or three[0].connected + ids = [o.connector_id for o in three] + assert len(set(ids)) == 3 diff --git a/tests/test_run_worker_ffmpeg.py b/tests/test_run_worker_ffmpeg.py index 60afb1e..d97fb1b 100644 --- a/tests/test_run_worker_ffmpeg.py +++ b/tests/test_run_worker_ffmpeg.py @@ -7,4 +7,4 @@ def test_alsa_ffmpeg_uses_low_delay_flags(): src = inspect.getsource(run_worker) assert '"-fflags", "nobuffer"' in src assert '"-flags", "low_delay"' in src - assert '"-thread_queue_size", "64"' in src + assert '"-thread_queue_size"' in src diff --git a/tokens.py b/tokens.py index a1d0484..529f6f0 100644 --- a/tokens.py +++ b/tokens.py @@ -18,3 +18,40 @@ def make_token(url: str, api_key: str, api_secret: str, )) ) return token.to_jwt() + + +def make_speaker_token(api_key: str, api_secret: str, + room: str, identity: str = "speaker", + name: str = "") -> str: + """Publisher token so a speaker can join and share their screen.""" + token = ( + AccessToken(api_key, api_secret) + .with_identity(identity) + .with_name(name or identity) + .with_grants(VideoGrants( + room_join=True, + room=room, + can_publish=True, + can_subscribe=True, + can_publish_data=True, + )) + ) + return token.to_jwt() + + +def make_viewer_token(api_key: str, api_secret: str, + room: str, identity: str) -> str: + """Subscribe-only token for headless display clients.""" + token = ( + AccessToken(api_key, api_secret) + .with_identity(identity) + .with_name(identity) + .with_grants(VideoGrants( + room_join=True, + room=room, + can_publish=False, + can_subscribe=True, + can_publish_data=False, + )) + ) + return token.to_jwt()