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.
264 lines
8.7 KiB
Python
264 lines
8.7 KiB
Python
"""Discover DRM/KMS connectors for headless kmssink output."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
DRM_CLASS = Path("/sys/class/drm")
|
|
|
|
# xf86drmMode.h DRM_MODE_CONNECTOR_*
|
|
_TYPE_SYSFS = {
|
|
10: "DP",
|
|
11: "HDMI-A",
|
|
12: "HDMI-B",
|
|
14: "eDP",
|
|
15: "Virtual",
|
|
}
|
|
_STATUS = {1: "connected", 2: "disconnected", 3: "unknown"}
|
|
|
|
|
|
@dataclass
|
|
class DrmOutput:
|
|
name: str # e.g. HDMI-A-7
|
|
sysfs: str # card0-HDMI-A-7
|
|
node: str # /dev/dri/card0
|
|
driver: str # xe / i915
|
|
connector_id: int
|
|
status: str
|
|
width: int = 0
|
|
height: int = 0
|
|
crtc_id: int = 0
|
|
plane_id: int = 0
|
|
bus_id: str = "" # pci:0000:03:00.0 style for kmssink
|
|
|
|
@property
|
|
def connected(self) -> bool:
|
|
return self.status == "connected"
|
|
|
|
|
|
def _pci_bus_id(node: str) -> str:
|
|
link = Path(f"/sys/class/drm/{Path(node).name}/device")
|
|
try:
|
|
uevent = (link / "uevent").read_text()
|
|
except OSError:
|
|
return ""
|
|
for line in uevent.splitlines():
|
|
if line.startswith("PCI_SLOT_NAME="):
|
|
slot = line.split("=", 1)[1].strip()
|
|
return f"pci:{slot}"
|
|
return ""
|
|
|
|
|
|
def _sysfs_connectors() -> dict[str, dict]:
|
|
out: dict[str, dict] = {}
|
|
if not DRM_CLASS.is_dir():
|
|
return out
|
|
for p in sorted(DRM_CLASS.iterdir()):
|
|
name = p.name
|
|
if not name.startswith("card") or "-" not in name:
|
|
continue
|
|
card, conn = name.split("-", 1)
|
|
status = (p / "status").read_text().strip() if (p / "status").is_file() else "unknown"
|
|
modes: list[str] = []
|
|
if (p / "modes").is_file():
|
|
modes = [m.strip() for m in (p / "modes").read_text().splitlines() if m.strip()]
|
|
w = h = 0
|
|
if modes and "x" in modes[0]:
|
|
try:
|
|
wh, *_ = modes[0].split("@")
|
|
w, h = (int(x) for x in wh.split("x")[:2])
|
|
except ValueError:
|
|
pass
|
|
out[name] = {
|
|
"card": card,
|
|
"conn": conn,
|
|
"status": status,
|
|
"modes": modes,
|
|
"width": w,
|
|
"height": h,
|
|
}
|
|
return out
|
|
|
|
|
|
def _drm_info() -> dict:
|
|
try:
|
|
raw = subprocess.check_output(["drm_info", "-j"], text=True, timeout=10)
|
|
except (FileNotFoundError, subprocess.CalledProcessError, subprocess.TimeoutExpired):
|
|
return {}
|
|
return json.loads(raw)
|
|
|
|
|
|
def _plane_type(plane: dict) -> int:
|
|
props = plane.get("properties") or {}
|
|
t = props.get("type") or {}
|
|
return int(t.get("value") or t.get("raw_value") or 0)
|
|
|
|
|
|
def _primary_planes(info: dict) -> list[dict]:
|
|
planes = info.get("planes") or []
|
|
return [p for p in planes if _plane_type(p) == 1]
|
|
|
|
|
|
def _pair_sysfs(card: str, typ_name: str, drm_conns: list[dict], sysfs: dict) -> list[tuple[dict, str]]:
|
|
names = []
|
|
for key, meta in sysfs.items():
|
|
if meta["card"] == card and meta["conn"].startswith(typ_name + "-"):
|
|
names.append(key)
|
|
names.sort(key=lambda k: int(sysfs[k]["conn"].rsplit("-", 1)[-1]))
|
|
drm_conns = sorted(drm_conns, key=lambda c: int(c["id"]))
|
|
return list(zip(drm_conns, names))
|
|
|
|
|
|
def list_outputs() -> list[DrmOutput]:
|
|
sysfs = _sysfs_connectors()
|
|
info = _drm_info()
|
|
outputs: list[DrmOutput] = []
|
|
for node, ninfo in info.items():
|
|
driver = ((ninfo.get("driver") or {}).get("name")) or ""
|
|
card = Path(node).name # card0
|
|
bus_id = _pci_bus_id(node)
|
|
crtcs = ninfo.get("crtcs") or []
|
|
primaries = _primary_planes(ninfo)
|
|
by_type: dict[int, list[dict]] = {}
|
|
for c in ninfo.get("connectors") or []:
|
|
by_type.setdefault(int(c.get("type") or 0), []).append(c)
|
|
assigned_crtc: set[int] = set()
|
|
for typ, conns in by_type.items():
|
|
typ_name = _TYPE_SYSFS.get(typ)
|
|
if not typ_name:
|
|
continue
|
|
for c, sys_key in _pair_sysfs(card, typ_name, conns, sysfs):
|
|
meta = sysfs[sys_key]
|
|
status = _STATUS.get(int(c.get("status") or 0), meta["status"]) or "unknown"
|
|
modes = c.get("modes") or []
|
|
w = h = 0
|
|
if modes:
|
|
m0 = modes[0]
|
|
w = int(m0.get("hdisplay") or 0)
|
|
h = int(m0.get("vdisplay") or 0)
|
|
if not w:
|
|
w, h = meta["width"], meta["height"]
|
|
crtc_id = 0
|
|
plane_id = 0
|
|
props = c.get("properties") or {}
|
|
crtc_prop = props.get("CRTC_ID") or {}
|
|
already = int(crtc_prop.get("value") or 0)
|
|
if already:
|
|
crtc_id = already
|
|
assigned_crtc.add(already)
|
|
outputs.append(
|
|
DrmOutput(
|
|
name=meta["conn"],
|
|
sysfs=sys_key,
|
|
node=node,
|
|
driver=driver,
|
|
connector_id=int(c["id"]),
|
|
status=status,
|
|
width=w,
|
|
height=h,
|
|
crtc_id=crtc_id,
|
|
plane_id=plane_id,
|
|
bus_id=bus_id,
|
|
)
|
|
)
|
|
# assign unused CRTCs + primary planes for disconnected heads
|
|
crtc_ids = [int(cr["id"]) for cr in crtcs]
|
|
unused = [i for i in crtc_ids if i not in assigned_crtc]
|
|
card_outs = [o for o in outputs if Path(o.node).name == card]
|
|
|
|
def crtc_rank(o: DrmOutput) -> tuple:
|
|
return (0 if o.connected else 1, 0 if "HDMI" in o.name else 1, o.name)
|
|
|
|
for out in sorted(card_outs, key=crtc_rank):
|
|
if not out.crtc_id and unused:
|
|
out.crtc_id = unused.pop(0)
|
|
if out.crtc_id:
|
|
try:
|
|
idx = crtc_ids.index(out.crtc_id)
|
|
except ValueError:
|
|
idx = 0
|
|
mask = 1 << idx
|
|
for p in primaries:
|
|
if int(p.get("possible_crtcs") or 0) & mask:
|
|
out.plane_id = int(p["id"])
|
|
break
|
|
outputs.sort(key=lambda o: (o.node, o.name))
|
|
return outputs
|
|
|
|
|
|
def _rank_wall(o: DrmOutput) -> tuple:
|
|
"""Arc HDMI first, then any connected HDMI. Names shuffle across boots."""
|
|
return (
|
|
0 if o.connected else 1,
|
|
0 if o.driver == "xe" else 1,
|
|
0 if o.name.startswith("HDMI-A") else 1,
|
|
o.name,
|
|
)
|
|
|
|
|
|
def pick_outputs(
|
|
all_outs: list[DrmOutput],
|
|
n: int = 3,
|
|
names: Optional[list[str]] = None,
|
|
connected_only: bool = False,
|
|
) -> list[DrmOutput]:
|
|
"""Choose DRM heads. Stale DISPLAY_CONNECTORS names must not bind a dead port
|
|
when a connected Arc HDMI is available — the wall has to stay lit."""
|
|
if names:
|
|
by_name: dict[str, list[DrmOutput]] = {}
|
|
for o in all_outs:
|
|
by_name.setdefault(o.name, []).append(o)
|
|
missing = [name for name in names if name not in by_name]
|
|
if missing:
|
|
have = sorted({o.name for o in all_outs})
|
|
raise KeyError(f"DRM connector {missing[0]!r} not found. Have: {have}")
|
|
picked: list[DrmOutput | None] = []
|
|
used: set[str] = set()
|
|
for name in names:
|
|
cands = sorted(by_name[name], key=_rank_wall)
|
|
chosen = next((o for o in cands if o.connected and o.sysfs not in used), None)
|
|
if chosen is None:
|
|
picked.append(None)
|
|
else:
|
|
used.add(chosen.sysfs)
|
|
picked.append(chosen)
|
|
fallback = sorted(
|
|
(o for o in all_outs if o.connected and o.sysfs not in used),
|
|
key=_rank_wall,
|
|
)
|
|
fi = 0
|
|
for i, o in enumerate(picked):
|
|
if o is None and fi < len(fallback):
|
|
picked[i] = fallback[fi]
|
|
used.add(fallback[fi].sysfs)
|
|
fi += 1
|
|
return [o for o in picked if o is not None]
|
|
connected = [o for o in all_outs if o.connected]
|
|
rest = [o for o in all_outs if not o.connected]
|
|
connected.sort(key=_rank_wall)
|
|
rest.sort(key=_rank_wall)
|
|
ordered = connected + ([] if connected_only else rest)
|
|
return ordered[:n]
|
|
|
|
|
|
def select_outputs(
|
|
n: int = 3,
|
|
names: Optional[list[str]] = None,
|
|
connected_only: bool = False,
|
|
) -> list[DrmOutput]:
|
|
return pick_outputs(list_outputs(), n=n, names=names, connected_only=connected_only)
|
|
|
|
|
|
def format_outputs(outs: list[DrmOutput]) -> str:
|
|
lines = []
|
|
for o in outs:
|
|
mode = f"{o.width}x{o.height}" if o.width else "?"
|
|
lines.append(
|
|
f"{o.name:12} {o.status:12} id={o.connector_id:<4} "
|
|
f"{o.driver:4} {o.node} crtc={o.crtc_id} plane={o.plane_id} {mode}"
|
|
)
|
|
return "\n".join(lines)
|