livekit-cameras/discovery.py

"""Discover C920 cameras: pair each /dev/video* node with its ALSA card.

Identity is stable across reboots and connect order: each camera is bound
to its USB *serial number* by udev rules (see generate_udev_rules.py and
slots.json in this directory), which pin
  - video: /dev/camNN symlinks (primary UVC node per serial)
  - audio: fixed ALSA card index per serial (SOUND_CARD_INDEX)

discover_cameras() therefore assigns cam-01..cam-20 by serial, never by
enumeration order. If slots.json is missing (first run), it falls back to
USB-port-based pairing and enumeration order.
"""

import json
import logging
import re
import subprocess
from dataclasses import dataclass, field
from pathlib import Path

from usb_ids import (
    is_fleet_video,
    is_rally_pid,
    looks_like_rally_name,
    usb_vid_pid_of_video_node,
)

log = logging.getLogger("cameras.discovery")

_SLOTS_FILE = Path(__file__).parent / "slots.json"


@dataclass
class Camera:
    index: int
    identity: str
    video_device: str
    audio_card: int | None
    usb_port: str = ""
    serial: str = ""
    extra_video_devices: list[str] = field(default_factory=list)
    # None = use fleet VIDEO_WIDTH / VIDEO_HEIGHT (C920 360p). Rally is 720p.
    width: int | None = None
    height: int | None = None


def select_cameras(cameras: list[Camera], spec: list[str]) -> list[Camera]:
    """`--cameras 1,5` means identities cam-01 and cam-05, not discovery index."""
    if spec == ["all"]:
        return list(cameras)
    wanted = {f"cam-{int(i):02d}" for i in spec}
    return [c for c in cameras if c.identity in wanted]


def _run(cmd: list[str]) -> str:
    return subprocess.run(cmd, capture_output=True, text=True, check=False).stdout


def _strip_interface(port: str) -> str:
    """Drop the trailing ':config.interface' so the video interface (x:1.0)
    and audio interface (x:1.2) of the same C920 compare equal."""
    if ":" in port:
        port = port.rsplit(":", 1)[0]
    return port


def _serial_of(sysfs_path: str) -> str:
    """Walk up from a sysfs path (interface, function, or card device)
    until we reach the USB device that carries the 'serial' attribute."""
    p = Path(sysfs_path)
    for _ in range(12):
        cand = p / "serial"
        if cand.is_file():
            return cand.read_text().strip()
        if str(p).startswith("/sys/devices"):
            p = p.parent
        else:
            return ""
    return ""


def _load_slots() -> dict[str, dict]:
    """slots.json: identity -> {serial, video, card}. Empty if absent."""
    if _SLOTS_FILE.exists():
        try:
            return json.loads(_SLOTS_FILE.read_text())
        except (OSError, ValueError) as e:
            log.warning("could not read %s (%s); using port-based fallback",
                        _SLOTS_FILE, e)
    return {}


def _usb_port_of_video_node(video_node: str) -> str:
    """Resolve the USB port (e.g. '0c:00.0-1.1.2.1') for a video device node."""
    link = Path(f"/sys/class/video4linux/{Path(video_node).name}")
    if not link.exists():
        return ""
    dev = link / "device"
    try:
        real = dev.resolve()
    except OSError:
        return ""
    # The USB port is the segment after the last '/' of the USB part, e.g.
    # /sys/devices/.../usb9/9-1/9-1.1/9-1.1.2/9-1.1.2.1/...
    parts = real.parts
    for i, p in enumerate(parts):
        if p.startswith("usb"):
            # take everything after the usbX part as the port path
            port_parts = [q for q in parts[i + 1:] if re.match(r"^\d+-", q)]
            if port_parts:
                return "/".join(port_parts)
    return ""


def _usb_port_of_sound_card(card: int) -> str:
    dev = Path(f"/sys/class/sound/card{card}/device")
    if not dev.exists():
        return ""
    try:
        real = dev.resolve()
    except OSError:
        return ""
    parts = real.parts
    for i, p in enumerate(parts):
        if p.startswith("usb"):
            port_parts = [q for q in parts[i + 1:] if re.match(r"^\d+-", q)]
            if port_parts:
                return "/".join(port_parts)
    return ""


def _sound_card_names() -> dict[int, str]:
    cards: dict[int, str] = {}
    out = _run(["cat", "/proc/asound/cards"])
    # format: " 2 [C920            ]: USB-Audio - HD Pro Webcam C920"
    # (the card name is left-justified in a fixed-width bracket field)
    for line in out.splitlines():
        m = re.match(r"\s*(\d+)\s+\[([A-Za-z0-9_]+)\s*\]:\s*(.*)", line)
        if m:
            cards[int(m.group(1))] = m.group(3).strip()
    return cards


def _serial_of_video_node(video_node: str) -> str:
    link = Path(f"/sys/class/video4linux/{Path(video_node).name}/device")
    try:
        real = link.resolve()
    except OSError:
        return ""
    return _serial_of(str(real))


def _serial_of_sound_card(card: int) -> str:
    dev = Path(f"/sys/class/sound/card{card}/device")
    try:
        real = dev.resolve()
    except OSError:
        return ""
    return _serial_of(str(real))


def discover_cameras() -> list[Camera]:
    """Return cameras with stable serial-based identities (cam-01..cam-20)."""
    slots = _load_slots()
    slots_by_serial = {s["serial"]: (ident, s)
                       for ident, s in slots.items() if s.get("serial")}

    v4l_out = _run(["v4l2-ctl", "--list-devices"])
    if not v4l_out:
        raise RuntimeError("v4l2-ctl not available; is v4l-utils installed?")

    # group video nodes by device name from v4l2-ctl output
    # v4l2-ctl output: "NAME:\n\t/dev/videoX\n\t/dev/videoY\n\t/dev/mediaZ"
    current: str | None = None
    groups: dict[str, list[str]] = {}
    for raw in v4l_out.splitlines():
        line = raw.strip()
        if not line:
            continue
        if line.endswith(":") and "\t" not in raw:
            current = line[:-1]
            groups.setdefault(current, [])
        elif current and line.startswith("/dev/video"):
            groups[current].append(line)

    # sound cards keyed by USB serial (authoritative) with port fallback
    card_names = _sound_card_names()
    card_by_serial: dict[str, int] = {}
    card_by_port: dict[str, int] = {}
    for card, name in card_names.items():
        if "C920" not in name and "920" not in name:
            continue
        serial = _serial_of_sound_card(card)
        if serial:
            card_by_serial[serial] = card
        else:
            port = _usb_port_of_sound_card(card)
            if port:
                card_by_port[_strip_interface(port)] = card

    cameras: list[Camera] = []
    for name, nodes in groups.items():
        if not nodes:
            continue
        if looks_like_rally_name(name):
            log.info("skipping Rally/speaker device %r (not a cam-NN slot)", name)
            continue
        nodes.sort(key=lambda n: int(Path(n).name.replace("video", "")))
        primary = nodes[0]
        ids = usb_vid_pid_of_video_node(primary)
        if ids is not None and not is_fleet_video(ids[0], ids[1]):
            log.info("skipping non-C920 USB %04x:%04x at %s", ids[0], ids[1], primary)
            continue
        if ids is not None and is_rally_pid(ids[1]):
            log.info("skipping Rally USB %04x:%04x at %s", ids[0], ids[1], primary)
            continue
        serial = _serial_of_video_node(primary)
        port = _usb_port_of_video_node(primary)

        identity = None
        video_device = primary
        audio_card = None
        slot = slots_by_serial.get(serial)
        if slot is not None:
            identity, slot_info = slot
            # prefer the pinned udev symlink if it exists
            pinned = slot_info.get("video", "")
            if pinned and Path(pinned).exists():
                video_device = pinned
            if slot_info.get("card") is not None:
                audio_card = slot_info["card"]
            if serial in card_by_serial:
                audio_card = card_by_serial[serial]  # live serial match wins
        else:
            log.warning("camera %s (%s): serial %s not in %s; "
                        "port-based fallback", name, primary, serial,
                        _SLOTS_FILE.name)
            if serial:
                audio_card = card_by_serial.get(serial)
            if audio_card is None:
                audio_card = card_by_port.get(_strip_interface(port))
            if audio_card is None:
                log.warning("camera %s (%s): no matching sound card "
                            "(port=%s)", name, primary, port)
            identity = None  # assigned below by enumeration order

        cameras.append(Camera(
            index=len(cameras),
            identity=identity or f"cam-{len(cameras) + 1:02d}",
            video_device=video_device,
            audio_card=audio_card,
            usb_port=port,
            serial=serial,
            extra_video_devices=nodes[1:],
        ))

    # stable order: known slots by slot number, unknowns by video node number
    def _video_num(cam: Camera) -> int:
        try:
            # resolve a pinned /dev/camNN symlink to its real /dev/videoN node
            real = Path(cam.video_device).resolve().name
            return int(real.replace("video", ""))
        except (ValueError, OSError):
            return 10_000

    def _slot_num(cam: Camera) -> int:
        m = re.match(r"cam-(\d+)$", cam.identity)
        return int(m.group(1)) if m else 1000
    cameras.sort(key=lambda c: (_slot_num(c), _video_num(c)))
    for i, cam in enumerate(cameras):
        cam.index = i
    return cameras