kmsgrab e ScreenShot2: esiti sperimentali documentati

- kmsgrab non praticabile: piano primario AR30 con modifier AMD DCC/tiled, ffmpeg
  rifiuta il framebuffer prima di applicare format/modifier
- KWin ScreenShot2: NoAuthorized anche con desktop entry system-wide
- indicatore del vassoio identificato come SNI di xdg-desktop-portal-kde
  (titolo 'Trasmissione dello schermo')
- backend kmsgrab comunque implementato e selezionabile (capture=kmsgrab)
- setup-kmsgrab.sh: installa via pkexec una copia di ffmpeg con cap_sys_admin+ep
This commit is contained in:
enne2
2026-11-01 13:22:03 +01:00
parent db47640846
commit 47b5c09fa1
4 changed files with 188 additions and 15 deletions
+86
View File
@@ -113,3 +113,89 @@ class Capture:
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 ""