"""Cattura dello schermo per ambilight su Wayland/KDE (ScreenCast portal + PipeWire). Restituisce i colori medi di N zone lungo un bordo dello schermo. Nessuna dipendenza da X11: usa xdg-desktop-portal (consenso una volta sola, token persistente) e GStreamer con pipewiresrc. """ import os CACHE_DIR = os.path.expanduser("~/.cache/ambilight") PORTAL = "org.freedesktop.portal.Desktop" PPATH = "/org/freedesktop/portal/desktop" def screencast_stream(cache_dir=CACHE_DIR): """Negozia la condivisione del monitor; ritorna (node_id, width, height).""" import dbus from dbus.mainloop.glib import DBusGMainLoop from gi.repository import GLib DBusGMainLoop(set_as_default=True) bus = dbus.SessionBus() sc = dbus.Interface(bus.get_object(PORTAL, PPATH), "org.freedesktop.portal.ScreenCast") sender = bus.get_unique_name()[1:].replace(".", "_") counter = [0] def request(fn, *args, **opts): counter[0] += 1 token = f"ambi{os.getpid()}{counter[0]}" path = f"{PPATH}/request/{sender}/{token}" iface = dbus.Interface(bus.get_object(PORTAL, path, introspect=False), "org.freedesktop.portal.Request") box = {} loop = GLib.MainLoop() def on_response(code, results): box["code"], box["results"] = code, results loop.quit() iface.connect_to_signal("Response", on_response) fn(*args, dbus.Dictionary({"handle_token": token, **opts}, signature="sv")) GLib.timeout_add_seconds(180, lambda: (loop.quit(), False)[1]) loop.run() if box.get("code") != 0: raise RuntimeError("condivisione schermo annullata o rifiutata") return box["results"] res = request(sc.CreateSession, session_handle_token="ambilight") session = str(res["session_handle"]) token_file = os.path.join(cache_dir, "restore_token") opts = {} if os.path.exists(token_file): tok = open(token_file).read().strip() if tok: opts["restore_token"] = tok request(sc.SelectSources, session, types=dbus.UInt32(1), multiple=False, cursor_mode=dbus.UInt32(1), persist_mode=dbus.UInt32(2), **opts) res = request(sc.Start, session, "") node = int(res["streams"][0][0]) size = [int(v) for v in res["streams"][0][1].get("size", [0, 0])] if "restore_token" in res: os.makedirs(cache_dir, exist_ok=True) with open(token_file, "w") as f: f.write(str(res["restore_token"])) return node, size[0], size[1] class Capture: """Pipeline GStreamer ridotta a (led*sub) x gh pixel RGB, pronta per il calcolo zone.""" def __init__(self, node, leds, sub=4, rows=72, fps=60): import gi gi.require_version("Gst", "1.0") gi.require_version("GstApp", "1.0") from gi.repository import Gst, GstApp # noqa: F401 (registra il tipo AppSink) Gst.init(None) self.gw = leds * sub self.gh = rows desc = (f"pipewiresrc path={node} do-timestamp=true ! " f"videoconvert ! videoscale ! videorate ! " f"video/x-raw,format=RGB,width={self.gw},height={self.gh}," f"framerate={int(fps)}/1 ! " f"appsink name=sink emit-signals=false max-buffers=2 drop=true sync=false") self.pipe = Gst.parse_launch(desc) self.sink = self.pipe.get_by_name("sink") def start(self): from gi.repository import Gst self.pipe.set_state(Gst.State.PLAYING) def stop(self): from gi.repository import Gst self.pipe.set_state(Gst.State.NULL) def frame(self, timeout_ms=200): """Ritorna un array numpy (gh, gw, 3) uint8, oppure None.""" import numpy as np from gi.repository import Gst sample = self.sink.try_pull_sample(timeout_ms * Gst.MSECOND) if sample is None: return None buf = sample.get_buffer() ok, info = buf.map(Gst.MapFlags.READ) if not ok: return None try: need = self.gw * self.gh * 3 data = bytes(info.data)[:need] finally: buf.unmap(info) if len(data) < need: return None return np.frombuffer(data, dtype=np.uint8).reshape(self.gh, self.gw, 3) # ------------------------------------------------------------------ kmsgrab FFMPEG_CAP = "/usr/local/libexec/ambilight/ffmpeg-kmsgrab" FFMPEG_SYS = "/usr/bin/ffmpeg" class KmsgrabCapture: """Cattura DRM diretta (ffmpeg -f kmsgrab) senza portal e senza indicatore. Richiede CAP_SYS_ADMIN sul binario ffmpeg di appoggio (FFMPEG_CAP), oppure root. ffmpeg riduce il frame a (leds*sub) x rows in RGB e lo scrive su stdout come rawvideo: leggiamo esattamente un frame per volta. """ def __init__(self, leds, sub=4, rows=72, fps=60, device="/dev/dri/card1", ffmpeg=None): self.gw = leds * sub self.gh = rows self.fps = max(1, int(fps)) self.device = device ff = ffmpeg or (FFMPEG_CAP if os.path.exists(FFMPEG_CAP) else FFMPEG_SYS) self.binary = ff self.frame_bytes = self.gw * self.gh * 3 self.proc = None self.stderr_path = os.path.join(CACHE_DIR, "kmsgrab.log") # hwdownload porta il frame in RAM; lo scaling avviene poi in ffmpeg. self.vf = (f"hwdownload,format=bgr0," f"scale={self.gw}:{self.gh}:flags=bilinear,format=rgb24") self.cmd = [ff, "-hide_banner", "-loglevel", "error", "-nostdin", "-f", "kmsgrab", "-device", device, "-i", "-", "-vf", self.vf, "-r", str(self.fps), "-f", "rawvideo", "-pix_fmt", "rgb24", "pipe:1"] def start(self): import subprocess os.makedirs(CACHE_DIR, exist_ok=True) self.err = open(self.stderr_path, "wb") self.proc = subprocess.Popen(self.cmd, stdout=subprocess.PIPE, stderr=self.err, bufsize=self.frame_bytes) return self def stop(self): try: if self.proc is not None: self.proc.terminate() try: self.proc.wait(timeout=2) except Exception: self.proc.kill() finally: self.proc = None try: self.err.close() except Exception: pass def alive(self): return self.proc is not None and self.proc.poll() is None def frame(self, timeout_ms=200): """Legge un frame completo; None se non arriva entro il timeout.""" import select import numpy as np if self.proc is None: return None fd = self.proc.stdout buf = bytearray() deadline = timeout_ms / 1000.0 while len(buf) < self.frame_bytes: ready, _, _ = select.select([fd], [], [], deadline) if not ready: return None chunk = fd.read(self.frame_bytes - len(buf)) if not chunk: return None buf += chunk return np.frombuffer(bytes(buf), dtype=np.uint8).reshape(self.gh, self.gw, 3) def error_text(self): try: with open(self.stderr_path, "rb") as f: return f.read()[-400:].decode("utf-8", "replace").strip() except Exception: return ""