SmartestHome/hosts/thin-client/agent/thinclient_agent/capture_control.py

"""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