root
95bb7c50ee
Publish and subscribe over wss://livekit.uni-wh.de:7800 and refuse cleartext ws://. Conference room is uwh-telhai. Includes the uncommitted encoded H.264 publish path, Rally hairpin, and KMS wall overlay.
100 lines
2.7 KiB
Python
100 lines
2.7 KiB
Python
"""Logitech Rally Camera V-R0010 + Dante room mic (speaker identity)."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
from pathlib import Path
|
|
|
|
from discovery import Camera
|
|
from usb_ids import (
|
|
RALLY_IDENTITY,
|
|
is_rally_pid,
|
|
looks_like_rally_name,
|
|
usb_vid_pid_of_video_node,
|
|
)
|
|
|
|
log = logging.getLogger("cameras.rally")
|
|
|
|
_CARD_RE = re.compile(r"\s*(\d+)\s+\[([A-Za-z0-9_]+)\s*\]:\s*(.*)")
|
|
|
|
|
|
def dante_capture_card() -> int | None:
|
|
"""ALSA card index for the Audinate Dante USB I/O module, if present."""
|
|
path = Path("/proc/asound/cards")
|
|
if not path.is_file():
|
|
return None
|
|
for line in path.read_text().splitlines():
|
|
m = _CARD_RE.match(line)
|
|
if not m:
|
|
continue
|
|
rest = (m.group(3) or "").lower()
|
|
name = (m.group(2) or "").lower()
|
|
if "dante" in rest or "dante" in name:
|
|
return int(m.group(1))
|
|
return None
|
|
|
|
|
|
def rally_video_node() -> str | None:
|
|
pinned = Path("/dev/rally")
|
|
if pinned.exists():
|
|
return str(pinned)
|
|
sysfs = Path("/sys/class/video4linux")
|
|
if not sysfs.is_dir():
|
|
return None
|
|
for p in sorted(sysfs.glob("video*"), key=lambda x: int(x.name[5:] or 0)):
|
|
name = (p / "name").read_text().strip() if (p / "name").is_file() else ""
|
|
idx = (p / "index").read_text().strip() if (p / "index").is_file() else ""
|
|
node = f"/dev/{p.name}"
|
|
if idx != "0":
|
|
continue
|
|
if looks_like_rally_name(name):
|
|
return node
|
|
ids = usb_vid_pid_of_video_node(node)
|
|
if ids is not None and is_rally_pid(ids[1]):
|
|
return node
|
|
return None
|
|
|
|
|
|
def _serial_of_video(node: str) -> str:
|
|
link = Path(f"/sys/class/video4linux/{Path(node).name}/device")
|
|
try:
|
|
p = link.resolve()
|
|
except OSError:
|
|
return ""
|
|
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:
|
|
break
|
|
return ""
|
|
|
|
|
|
def discover_rally(*, width: int = 1920, height: int = 1080) -> Camera | None:
|
|
"""Speaker camera: Rally UVC + Dante capture. None if the head is missing."""
|
|
video = rally_video_node()
|
|
if not video:
|
|
return None
|
|
card = dante_capture_card()
|
|
serial = _serial_of_video(video)
|
|
cam = Camera(
|
|
index=20,
|
|
identity=RALLY_IDENTITY,
|
|
video_device=video,
|
|
audio_card=card,
|
|
serial=serial,
|
|
width=width,
|
|
height=height,
|
|
)
|
|
log.info(
|
|
"rally video=%s audio=%s serial=%s %dx%d",
|
|
video,
|
|
f"hw:{card},0" if card is not None else "none",
|
|
serial or "-",
|
|
width,
|
|
height,
|
|
)
|
|
return cam
|