chore: snapshot livekit-cameras after serial pin + speechbrain lengths fix
This commit is contained in:
Hermes
+244
@@ -0,0 +1,244 @@
|
||||
"""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
|
||||
|
||||
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)
|
||||
|
||||
|
||||
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
|
||||
nodes.sort(key=lambda n: int(Path(n).name.replace("video", "")))
|
||||
primary = nodes[0]
|
||||
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
|
||||
Reference in New Issue
Block a user