"""Capture-card ("receiver box") video source selection. USB/PCIe HDMI capture cards plugged into this machine are enumerated and exposed as an HA select, mirroring audio_control.AudioControl's shape: dynamic discovery, options()/current_option()/select(), a payload from HA only ever used to look up an already-enumerated device — see the security note in mqtt_discovery.py. The actual /dev/videoN path handed to the capture-view launcher is server-side data resolved from that lookup, never the payload itself. PRIVACY INVARIANT — read before touching this file. hosts/thin-client/README.md documents that the gesture-control camera has no HA entity and never will: "nothing reachable over MQTT can turn the camera on." A naive "list every capture-capable /dev/video*" would break that the moment gesture control is enabled on a unit, so list_devices() excludes gesture-control's configured camera_device whenever gesture-config.json's "enabled" is true — the same gate gesture_pointer.py itself uses to decide whether the camera is ever opened at all. It is deliberately NOT an unconditional exclusion: that file's camera_device defaults to /dev/video0 on every image regardless of whether gesture control was even built in, and excluding it while gesture control is off would just hide a real capture card that happens to enumerate there, for no privacy benefit — there is nothing to protect while gesture_pointer.py itself never opens the camera. """ from __future__ import annotations import glob import logging import os import re import subprocess from dataclasses import dataclass from .runtime_state import ensure_runtime_copy, load_json log = logging.getLogger(__name__) GESTURE_CONFIG_FILENAME = "gesture-config.json" NO_SOURCE = "none" _DEVICE_CAPS_HEADER = re.compile(r"^\s*Device Caps\s*:") _CAPTURE_CAPS = {"Video Capture", "Video Capture Multiplanar"} @dataclass(frozen=True) class CaptureDevice: path: str label: str def _run(args: list[str]) -> str | None: try: result = subprocess.run(args, capture_output=True, text=True, timeout=10, check=False) except (OSError, subprocess.TimeoutExpired) as exc: log.warning("%s failed: %s", " ".join(args), exc) return None if result.returncode != 0: log.warning("%s: %s", " ".join(args), result.stderr.strip()) return None return result.stdout def _parse_list_devices(output: str) -> dict[str, list[str]]: """`v4l2-ctl --list-devices` groups /dev/videoN nodes under a physical-device header line, blank-line separated, e.g.: USB Video: USB Video (usb-0000:00:14.0-3): \t/dev/video0 \t/dev/video1 The header already includes a bus-path suffix, which is what keeps two identical capture dongles distinguishable without AudioControl-style "(2)" suffixing. """ groups: dict[str, list[str]] = {} label: str | None = None for raw_line in output.splitlines(): if not raw_line.strip(): label = None continue if not raw_line[0].isspace(): label = raw_line.rstrip(":").strip() groups.setdefault(label, []) continue if label is not None: path = raw_line.strip() if path: groups[label].append(path) return groups def _has_capture_capability(path: str) -> bool: output = _run(["v4l2-ctl", "-d", path, "--info"]) if output is None: return False lines = output.splitlines() in_device_caps = False for line in lines: if _DEVICE_CAPS_HEADER.match(line): in_device_caps = True continue if not in_device_caps: continue stripped = line.strip() if not line[:1].isspace() or not stripped: break if stripped in _CAPTURE_CAPS: return True return False def _usb_device_dir(sysfs_device_symlink: str) -> str | None: """Resolves a component's sysfs `device` symlink up to the nearest ancestor directory that looks like a USB device (has an `idVendor` file) — the shared physical USB device two different interfaces (video, audio) both hang off. """ try: current = os.path.realpath(sysfs_device_symlink) except OSError: return None if not os.path.exists(current): return None # USB sysfs trees are shallow; a hard cap avoids any chance of looping on a # pathological symlink structure. for _ in range(6): if os.path.exists(os.path.join(current, "idVendor")): return current parent = os.path.dirname(current) if parent == current: return None current = parent return None def find_audio_card(device_path: str) -> str | None: """Best-effort match of a /dev/videoN to its sibling USB Audio Class ALSA card. Many cheap USB HDMI-capture dongles present a UVC video interface and a *separate* USB Audio interface — the embedded HDMI audio does not ride along in the V4L2 stream, so mpv needs a second, explicit audio source. Matched by shared physical USB device (sysfs), not by vendor/product ID. Unverified against real hardware — see hosts/thin-client/README.md. Returns None (video-only playback, not a crash) if nothing matches. """ video_name = os.path.basename(device_path.rstrip("/")) video_usb = _usb_device_dir(f"/sys/class/video4linux/{video_name}/device") if video_usb is None: return None for card_dir in sorted(glob.glob("/sys/class/sound/card[0-9]*")): card_usb = _usb_device_dir(os.path.join(card_dir, "device")) if card_usb is not None and card_usb == video_usb: index = os.path.basename(card_dir).removeprefix("card") return f"hw:{index},0" return None class CaptureControl: def __init__(self) -> None: self._devices: list[CaptureDevice] = [] self._current: str | None = None # a device path, or None for "no source" def _excluded_device_path(self) -> str: # Gated on "enabled", not just present: gesture-config.json's template ships # on every image with camera_device defaulting to /dev/video0 regardless of # whether gesture control was even built into this image (see its own "always # copied in" comment in build-thin-client-iso.sh) — gesture_pointer.py itself # never opens the camera while "enabled" is false, so there is nothing to # protect against on the (default, common) image where it's off, and excluding # /dev/video0 unconditionally would just be silently hiding a real capture card # that happens to enumerate there. The exclusion has to track the same gate # gesture_pointer.py uses to decide whether the camera is ever opened at all. path = ensure_runtime_copy(GESTURE_CONFIG_FILENAME) config = load_json(path) if not config.get("enabled"): return "" return str(config.get("camera_device") or "").strip() def list_devices(self) -> list[CaptureDevice]: output = _run(["v4l2-ctl", "--list-devices"]) if output is None: self._devices = [] return self._devices excluded = self._excluded_device_path() devices: list[CaptureDevice] = [] for label, paths in _parse_list_devices(output).items(): for path in paths: if not path.startswith("/dev/video") or path == excluded: continue if _has_capture_capability(path): devices.append(CaptureDevice(path=path, label=label)) break # one entry per physical device — the first real capture node self._devices = devices return devices # --- entity surface ----------------------------------------------------- def options(self) -> list[str]: return [NO_SOURCE] + [d.label for d in self._devices] def current_option(self) -> str: if self._current is None: return NO_SOURCE for device in self._devices: if device.path == self._current: return device.label # Selected, then unplugged since — say so rather than silently reporting # a source that is no longer there. return NO_SOURCE def _find(self, label: str) -> CaptureDevice | None: for device in self._devices: if device.label == label: return device return None def select(self, payload: str) -> CaptureDevice | None: """Handle the HA select. Returns the resolved device, or None for "no source" (including an unrecognised payload, treated the same as picking "none" rather than acting on anything not currently listed).""" label = payload.strip() self.list_devices() if label == NO_SOURCE: self._current = None return None device = self._find(label) if device is None: log.warning("ignoring unknown capture source %r", label) self._current = None return None self._current = device.path return device