livekit-cameras/generate_udev_rules.py

#!/usr/bin/env python3
"""Generate udev rules that pin each C920 webcam to a fixed slot.

Slot N (cam-01..cam-20) is bound to the USB *serial number* of the camera
that currently occupies it, giving stable:
  - video: /dev/camNN (primary capture node), /dev/camNN.meta (UVC metadata)
  - audio: fixed ALSA card index (hw:INDEX,0)

Run:  python3 generate_udev_rules.py            # prints to stdout
      python3 generate_udev_rules.py --write     # also writes /etc/udev/rules.d/99-c920-pin.rules
"""
import re
import subprocess
import sys

from usb_ids import is_fleet_video, is_rally_pid, usb_vid_pid_from_sysfs

HEADER = """\
# Stable slots for 20x Logitech HD Pro Webcam C920 (046d:08e5).
# Slot = physical camera, identified by USB serial. Survives reboots,
# re-enumeration, and hub reordering. Generated by generate_udev_rules.py.
#
#   /dev/camNN     -> primary video capture node (UVC stream 0)
#   /dev/camNN.meta -> UVC metadata node (stream 1)
#   hw:N,0         -> mic of cam-NN (N from SOUND_CARD_INDEX below)
"""


def sh(cmd):
    return subprocess.run(cmd, capture_output=True, text=True).stdout


def _serial_of(sysfs_path):
    """Walk up from a sysfs path (interface, function, or card device)
    until we reach the USB device that carries the 'serial' attribute."""
    import pathlib
    p = pathlib.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 current_state():
    """Map serial -> (primary_video_node, sound_card_index)."""
    import pathlib
    # sound cards: index -> serial
    cards = {}
    for line in sh(["cat", "/proc/asound/cards"]).splitlines():
        m = re.match(r"\s*(\d+)\s+\[([A-Za-z0-9_]+)\s*\]:\s*(.*)", line)
        if not m:
            continue
        idx = int(m.group(1))
        if "C920" not in m.group(3) and "920" not in m.group(2):
            continue
        dev = f"/sys/class/sound/card{idx}/device"
        real = sh(["readlink", "-f", dev]).strip()
        cards[idx] = _serial_of(real)
    # video: primary node (index 0) per USB serial, via sysfs
    primary_by_serial = {}
    for node in sorted(pathlib.Path("/sys/class/video4linux").glob("video*"),
                       key=lambda p: int(p.name.replace("video", ""))):
        idx_attr = (node / "index").read_text().strip()
        if idx_attr != "0":
            continue
        real = sh(["readlink", "-f", str(node / "device")]).strip()
        ids = usb_vid_pid_from_sysfs(real)
        if ids is not None and not is_fleet_video(ids[0], ids[1]):
            continue
        if ids is not None and is_rally_pid(ids[1]):
            continue
        s = _serial_of(real)
        primary_by_serial[s] = f"/dev/{node.name}"
    return primary_by_serial, cards


def main():
    primary_by_serial, cards = current_state()
    # assign slots: cam-01..cam-20 in current video-node order
    ordered = []
    for serial, node in primary_by_serial.items():
        n = int(node.replace("/dev/video", ""))
        ordered.append((n, serial, node))
    ordered.sort()
    if len(ordered) != 20:
        print(f"warning: expected 20 cameras, got {len(ordered)}", file=sys.stderr)

    lines = [HEADER]
    for i, (vnum, serial, node) in enumerate(ordered, start=1):
        card = next((c for c, s in cards.items() if s == serial), None)
        slot = f"cam{i:02d}"
        lines.append(f"# {slot}: serial {serial}  (was {node}, card {card})")
        lines.append(
            f'SUBSYSTEM=="video4linux", KERNEL=="video*", '
            f'ENV{{ID_VENDOR_ID}}=="046d", ENV{{ID_MODEL_ID}}=="08e5", '
            f'ENV{{ID_USB_SERIAL_SHORT}}=="{serial}", '
            f'ATTR{{index}}=="0", SYMLINK+="{slot}"')
        lines.append(
            f'SUBSYSTEM=="video4linux", KERNEL=="video*", '
            f'ENV{{ID_VENDOR_ID}}=="046d", ENV{{ID_MODEL_ID}}=="08e5", '
            f'ENV{{ID_USB_SERIAL_SHORT}}=="{serial}", '
            f'ATTR{{index}}=="1", SYMLINK+="{slot}.meta"')
        if card is not None:
            lines.append(
                f'SUBSYSTEM=="sound", KERNEL=="card*", '
                f'ENV{{ID_VENDOR_ID}}=="046d", ENV{{ID_MODEL_ID}}=="08e5", '
                f'ENV{{ID_USB_SERIAL_SHORT}}=="{serial}", '
                f'ENV{{SOUND_CARD_INDEX}}="{card}"')
        else:
            print(f"warning: no sound card for serial {serial}", file=sys.stderr)
        lines.append("")
    text = "\n".join(lines)
    print(text)
    if "--write" in sys.argv:
        path = "/etc/udev/rules.d/99-c920-pin.rules"
        with open(path, "w") as f:
            f.write(text)
        print(f"# wrote {path}", file=sys.stderr)
        # durable slot map for the app (discovery.py reads this)
        import json
        import os
        slots = {}
        for i, (vnum, serial, node) in enumerate(ordered, start=1):
            card = next((c for c, s in cards.items() if s == serial), None)
            slots[f"cam-{i:02d}"] = {
                "serial": serial,
                "video": f"/dev/cam{i:02d}",
                "card": card,
            }
        base = os.path.dirname(os.path.abspath(__file__))
        slots_path = os.path.join(base, "slots.json")
        with open(slots_path, "w") as f:
            json.dump(slots, f, indent=2)
        print(f"# wrote {slots_path}", file=sys.stderr)
        # summary table to stderr
        for i, (vnum, serial, node) in enumerate(ordered, start=1):
            card = next((c for c, s in cards.items() if s == serial), "?")
            print(f"#   cam-{i:02d}  serial={serial}  video={node}  card={card}",
                  file=sys.stderr)


if __name__ == "__main__":
    main()