"""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 select_outputs( n: int = 3, names: Optional[list[str]] = None, connected_only: bool = False, ) -> list[DrmOutput]: all_outs = list_outputs() if names: by_name = {o.name: o for o in all_outs} picked = [] for name in names: if name not in by_name: raise KeyError(f"DRM connector {name!r} not found. Have: {sorted(by_name)}") picked.append(by_name[name]) return picked connected = [o for o in all_outs if o.connected] rest = [o for o in all_outs if not o.connected] # Prefer the discrete GPU (xe / Arc) HDMI heads for a 3-display wall. def rank(o: DrmOutput) -> tuple: return ( 0 if o.driver == "xe" else 1, 0 if o.name.startswith("HDMI-A") else 1, o.name, ) connected.sort(key=rank) rest.sort(key=rank) ordered = connected + ([] if connected_only else rest) return ordered[:n] 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)