321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""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"
|
|
|
|
# Sessione ScreenCast attiva: va CHIUSA esplicitamente, altrimenti il portal continua
|
|
# a considerarla viva e lascia nel vassoio l'indicatore "Trasmissione dello schermo".
|
|
_ACTIVE = {"bus": None, "session": None}
|
|
|
|
|
|
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"]))
|
|
|
|
_ACTIVE["bus"] = bus
|
|
_ACTIVE["session"] = session
|
|
return node, size[0], size[1]
|
|
|
|
|
|
def close_stream():
|
|
"""Chiude la sessione ScreenCast e la connessione D-Bus (rimuove l'indicatore)."""
|
|
session, bus = _ACTIVE.get("session"), _ACTIVE.get("bus")
|
|
_ACTIVE["bus"] = _ACTIVE["session"] = None
|
|
if not session:
|
|
return
|
|
try:
|
|
import dbus
|
|
iface = dbus.Interface(bus.get_object(PORTAL, session), "org.freedesktop.portal.Session")
|
|
iface.Close()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
bus.close()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
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 ""
|
|
|
|
|
|
# ------------------------------------------------------------------ processo figlio
|
|
class ChildCapture:
|
|
"""Legge frame RGB grezzi dallo stdout di un processo di cattura separato.
|
|
|
|
Il figlio (ambicatcher.py per il portal, ffmpeg per kmsgrab) produce:
|
|
prima riga -> JSON di intestazione ({"node":..} oppure {"error":..})
|
|
poi -> frame da (leds*sub) x rows pixel RGB, senza separatori
|
|
Tenere la cattura fuori dal demone isola GStreamer/PipeWire e, per il portal,
|
|
fa chiudere la sessione ScreenCast insieme al processo (niente indicatori orfani).
|
|
"""
|
|
|
|
def __init__(self, argv, leds, sub=4, rows=72, expect_header=True):
|
|
self.cmd = list(argv)
|
|
self.gw = leds * sub
|
|
self.gh = rows
|
|
self.frame_bytes = self.gw * self.gh * 3
|
|
self.expect_header = expect_header
|
|
self.header = {}
|
|
self.proc = None
|
|
self.stderr_path = os.path.join(CACHE_DIR, "catcher.log")
|
|
|
|
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, stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE, stderr=self.err,
|
|
bufsize=0)
|
|
if self.expect_header:
|
|
line = self.proc.stdout.readline()
|
|
if not line:
|
|
err = self.error_text()
|
|
raise RuntimeError(f"cattura non avviata: {err or 'nessun output'}")
|
|
try:
|
|
import json
|
|
self.header = json.loads(line.decode())
|
|
except Exception:
|
|
self.header = {"raw": line[:100].decode("utf-8", "replace")}
|
|
if "error" in self.header:
|
|
raise RuntimeError(self.header["error"])
|
|
return self
|
|
|
|
def stop(self):
|
|
try:
|
|
if self.proc is not None:
|
|
try:
|
|
if self.proc.stdin:
|
|
self.proc.stdin.close() # il figlio esce e chiude la sessione
|
|
except Exception:
|
|
pass
|
|
try:
|
|
self.proc.wait(timeout=3)
|
|
except Exception:
|
|
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):
|
|
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 ""
|