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.
223 lines
7.5 KiB
Python
223 lines
7.5 KiB
Python
"""VISCA-IP façade for Logitech Rally (UVC + XU), not native VISCA.
|
|
|
|
Listens on TCP (default 127.0.0.1:5678), accepts raw VISCA frames ending in
|
|
0xFF, and drives pan/tilt/zoom through v4l2. AutoPTZ should use backend
|
|
visca_ip, address 127.0.0.1:5678, raw framing. visca_usb is pyserial — do
|
|
not point it at this camera.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import logging
|
|
import select
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
from typing import Callable
|
|
|
|
log = logging.getLogger("cameras.visca_xu")
|
|
|
|
# Sony/PTZOptics ACK + completion for a command addressed as 81.
|
|
_ACK = bytes([0x90, 0x41, 0xFF])
|
|
_COMP = bytes([0x90, 0x51, 0xFF])
|
|
_SYNTAX = bytes([0x90, 0x60, 0x02, 0xFF])
|
|
|
|
|
|
def visca_frames(buf: bytes) -> tuple[list[bytes], bytes]:
|
|
"""Split a byte stream into 0xFF-terminated frames; return (frames, rest)."""
|
|
frames: list[bytes] = []
|
|
start = 0
|
|
for i, b in enumerate(buf):
|
|
if b == 0xFF:
|
|
frames.append(buf[start:i + 1])
|
|
start = i + 1
|
|
return frames, buf[start:]
|
|
|
|
|
|
def parse_visca(frame: bytes) -> dict:
|
|
"""Parse one VISCA command. Unknown frames get action='unknown'."""
|
|
f = bytes(frame)
|
|
if len(f) < 5 or f[-1] != 0xFF:
|
|
return {"action": "invalid"}
|
|
# 81 01 06 01 VV WW PP QQ FF pan/tilt drive
|
|
if len(f) >= 9 and f[1:4] == bytes([0x01, 0x06, 0x01]):
|
|
pan_dir, tilt_dir = f[6], f[7]
|
|
pan = 0.0
|
|
tilt = 0.0
|
|
if pan_dir == 0x01:
|
|
pan = -1.0
|
|
elif pan_dir == 0x02:
|
|
pan = 1.0
|
|
if tilt_dir == 0x01:
|
|
tilt = 1.0
|
|
elif tilt_dir == 0x02:
|
|
tilt = -1.0
|
|
if pan_dir == 0x03 and tilt_dir == 0x03:
|
|
return {"action": "stop"}
|
|
return {"action": "move", "pan": pan, "tilt": tilt, "zoom": 0.0}
|
|
# 81 01 04 07 ZZ FF zoom
|
|
if len(f) >= 6 and f[1:4] == bytes([0x01, 0x04, 0x07]):
|
|
z = f[4]
|
|
if z == 0x00:
|
|
return {"action": "zoom_stop"}
|
|
if (z & 0xF0) == 0x20:
|
|
return {"action": "zoom", "zoom": 1.0}
|
|
if (z & 0xF0) == 0x30:
|
|
return {"action": "zoom", "zoom": -1.0}
|
|
return {"action": "zoom_stop"}
|
|
# 81 01 06 04 FF home
|
|
if f[1:4] == bytes([0x01, 0x06, 0x04]):
|
|
return {"action": "home"}
|
|
# 81 01 04 3F 01/02 MM FF preset set/recall
|
|
if len(f) >= 7 and f[1:4] == bytes([0x01, 0x04, 0x3F]):
|
|
if f[4] == 0x02:
|
|
return {"action": "preset_recall", "index": int(f[5])}
|
|
if f[4] == 0x01:
|
|
return {"action": "preset_set", "index": int(f[5])}
|
|
# inquiries: answer none (get_position returns None on AutoPTZ)
|
|
if len(f) >= 5 and f[1] == 0x09:
|
|
return {"action": "inquiry"}
|
|
return {"action": "unknown"}
|
|
|
|
|
|
class RallyV4L2:
|
|
"""Thin v4l2-ctl wrapper. Rally exposes generic pan/tilt CIDs on this host."""
|
|
|
|
def __init__(self, device: str = "/dev/rally"):
|
|
self.device = device
|
|
|
|
def _ctl(self, *args: str) -> str:
|
|
r = subprocess.run(
|
|
["v4l2-ctl", "-d", self.device, *args],
|
|
capture_output=True, text=True, check=False)
|
|
if r.returncode != 0:
|
|
log.warning("v4l2-ctl rc=%s %s", r.returncode, r.stderr.strip())
|
|
return r.stdout
|
|
|
|
def get(self) -> dict[str, int]:
|
|
out = self._ctl("--get-ctrl=pan_absolute,tilt_absolute,zoom_absolute")
|
|
vals: dict[str, int] = {}
|
|
for line in out.splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
k, v = line.split(":", 1)
|
|
try:
|
|
vals[k.strip()] = int(v.strip())
|
|
except ValueError:
|
|
pass
|
|
return vals
|
|
|
|
def set_speed(self, pan: int, tilt: int) -> None:
|
|
pan = max(-1, min(1, int(pan)))
|
|
tilt = max(-1, min(1, int(tilt)))
|
|
self._ctl(f"--set-ctrl=pan_speed={pan},tilt_speed={tilt}")
|
|
|
|
def stop(self) -> None:
|
|
self.set_speed(0, 0)
|
|
|
|
def nudge_zoom(self, direction: int, step: int = 20) -> None:
|
|
cur = self.get().get("zoom_absolute", 100)
|
|
nxt = max(100, min(1500, cur + (step if direction > 0 else -step)))
|
|
self._ctl(f"--set-ctrl=zoom_absolute={nxt}")
|
|
|
|
def home(self) -> None:
|
|
self.stop()
|
|
self._ctl("--set-ctrl=pan_absolute=0,tilt_absolute=0,zoom_absolute=100")
|
|
|
|
def apply(self, cmd: dict) -> None:
|
|
action = cmd.get("action")
|
|
if action in ("stop", "zoom_stop"):
|
|
self.stop()
|
|
elif action == "move":
|
|
pan = cmd.get("pan") or 0.0
|
|
tilt = cmd.get("tilt") or 0.0
|
|
self.set_speed(
|
|
-1 if pan < -0.01 else (1 if pan > 0.01 else 0),
|
|
-1 if tilt < -0.01 else (1 if tilt > 0.01 else 0),
|
|
)
|
|
elif action == "zoom":
|
|
z = cmd.get("zoom") or 0.0
|
|
if abs(z) < 0.01:
|
|
return
|
|
self.nudge_zoom(1 if z > 0 else -1)
|
|
elif action == "home":
|
|
self.home()
|
|
|
|
|
|
def serve(
|
|
host: str,
|
|
port: int,
|
|
apply: Callable[[dict], None],
|
|
stop: threading.Event | None = None,
|
|
) -> None:
|
|
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
srv.bind((host, port))
|
|
srv.listen(4)
|
|
srv.settimeout(0.5)
|
|
log.info("VISCA-IP façade %s:%d (raw) -> Rally v4l2", host, port)
|
|
clients: dict[socket.socket, bytearray] = {}
|
|
try:
|
|
while stop is None or not stop.is_set():
|
|
rlist = [srv, *clients]
|
|
try:
|
|
ready, _, _ = select.select(rlist, [], [], 0.5)
|
|
except InterruptedError:
|
|
continue
|
|
for s in ready:
|
|
if s is srv:
|
|
conn, addr = srv.accept()
|
|
conn.setblocking(False)
|
|
clients[conn] = bytearray()
|
|
log.info("visca client %s", addr)
|
|
continue
|
|
try:
|
|
data = s.recv(4096)
|
|
except BlockingIOError:
|
|
continue
|
|
except OSError:
|
|
data = b""
|
|
if not data:
|
|
clients.pop(s, None)
|
|
s.close()
|
|
continue
|
|
clients[s].extend(data)
|
|
frames, rest = visca_frames(bytes(clients[s]))
|
|
clients[s] = bytearray(rest)
|
|
for fr in frames:
|
|
cmd = parse_visca(fr)
|
|
if cmd.get("action") in ("invalid", "inquiry", "unknown"):
|
|
if cmd.get("action") == "invalid":
|
|
s.sendall(_SYNTAX)
|
|
continue
|
|
try:
|
|
apply(cmd)
|
|
except Exception: # noqa: BLE001
|
|
log.exception("apply failed %s", cmd)
|
|
try:
|
|
s.sendall(_ACK + _COMP)
|
|
except OSError:
|
|
pass
|
|
finally:
|
|
for c in list(clients):
|
|
c.close()
|
|
srv.close()
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__)
|
|
ap.add_argument("--device", default="/dev/rally")
|
|
ap.add_argument("--host", default="127.0.0.1")
|
|
ap.add_argument("--port", type=int, default=5678)
|
|
args = ap.parse_args(argv)
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [visca-xu] %(levelname)s %(message)s")
|
|
rally = RallyV4L2(args.device)
|
|
serve(args.host, args.port, rally.apply)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|