Integra calibrazione RGB nel plasmoide e ripristina backend funzionante
This commit is contained in:
@@ -44,6 +44,13 @@ una volta (viene memorizzato in `~/.cache/ambilight/restore_token`).
|
||||
|
||||
Il plasmoide appare nel vassoio come **Ambilight** (icona TV); clic per il popup.
|
||||
|
||||
## Calibrazione del bianco
|
||||
|
||||
Dal popup del plasmoide scegli **Calibra…**: la fascia superiore bianca permette di
|
||||
regolare in tempo reale i gain RGB e la luminosità. Le variazioni vengono salvate
|
||||
nel file di configurazione dal demone; **Esporta profilo JSON…** ne salva una copia
|
||||
separata. Chiudi la finestra con il pulsante **Chiudi**. Richiede PySide6.
|
||||
|
||||
## Uso da terminale
|
||||
|
||||
```sh
|
||||
|
||||
@@ -10,10 +10,6 @@ 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)."""
|
||||
@@ -68,30 +64,9 @@ def screencast_stream(cache_dir=CACHE_DIR):
|
||||
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."""
|
||||
|
||||
@@ -138,183 +113,3 @@ 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 ""
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ 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 ""
|
||||
|
||||
@@ -66,19 +66,15 @@ def main(argv):
|
||||
key, raw = argv[2], argv[3]
|
||||
if key in ("reverse",):
|
||||
value = raw.lower() in ("1", "true", "yes", "on")
|
||||
vtype = "b"
|
||||
elif key in ("leds", "baud"):
|
||||
elif key == "leds":
|
||||
value = int(raw)
|
||||
vtype = "i"
|
||||
elif key in ("port", "side", "capture", "drm_device"):
|
||||
value = str(raw)
|
||||
vtype = "s"
|
||||
else:
|
||||
value = float(raw)
|
||||
vtype = "d"
|
||||
try:
|
||||
from gi.repository import GLib
|
||||
res = call("SetParam", GLib.Variant("(sv)", (key, GLib.Variant(vtype, value))))
|
||||
res = call("SetParam", GLib.Variant("(sv)", (key, GLib.Variant(
|
||||
"b" if isinstance(value, bool) else "d" if isinstance(value, float) else "i",
|
||||
value))))
|
||||
err = res.unpack()[0]
|
||||
if err:
|
||||
print(json.dumps({"ok": False, "error": err}))
|
||||
|
||||
@@ -3,17 +3,15 @@ Description=Ambilight: striscia LED sincronizzata con il bordo dello schermo
|
||||
Documentation=https://git.enne2.net/enne2/ambilight-plasmoid
|
||||
After=graphical-session.target
|
||||
PartOf=graphical-session.target
|
||||
StartLimitIntervalSec=120
|
||||
StartLimitBurst=3
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
# Rete di sicurezza: un crash non deve mai diventare un loop che apre decine di
|
||||
# sessioni ScreenCast (e quindi decine di indicatori nel vassoio).
|
||||
Restart=on-failure
|
||||
RestartSec=10
|
||||
StartLimitIntervalSec=120
|
||||
StartLimitBurst=3
|
||||
ExecStart=/usr/bin/python3 %h/.local/bin/ambilightd.py
|
||||
Environment=PYTHONUNBUFFERED=1
|
||||
Restart=on-failure
|
||||
RestartSec=10
|
||||
|
||||
[Install]
|
||||
WantedBy=graphical-session.target
|
||||
|
||||
+42
-39
@@ -19,7 +19,6 @@ Config: ~/.config/ambilight/config.json
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
|
||||
@@ -36,6 +35,9 @@ DEFAULTS = {
|
||||
"brightness": 1.0, # moltiplicatore 0.05 .. 2.0
|
||||
"contrast": 1.0, # 0.5 .. 2.0 (pivot sul grigio 128)
|
||||
"saturation": 1.0, # 0.0 .. 2.0
|
||||
"gain_r": 1.0, # bilanciamento del bianco per canale
|
||||
"gain_g": 1.0,
|
||||
"gain_b": 1.0,
|
||||
"band": 0.12, # frazione di altezza campionata sul bordo
|
||||
"reverse": True, # LED 0 sul lato destro dello schermo (misurato su frigate)
|
||||
"leds": 47,
|
||||
@@ -43,14 +45,15 @@ DEFAULTS = {
|
||||
"baud": 115200,
|
||||
"fps": 60,
|
||||
"side": "top", # top | bottom
|
||||
"capture": "portal", # portal (ScreenCast) | kmsgrab (DRM diretto, senza indicatore)
|
||||
"drm_device": "/dev/dri/card1",
|
||||
"auto_start": True, # streaming attivo all'avvio del demone
|
||||
}
|
||||
PARAM_RANGES = {
|
||||
"brightness": (0.05, 2.0),
|
||||
"contrast": (0.5, 2.0),
|
||||
"saturation": (0.0, 2.0),
|
||||
"gain_r": (0.0, 2.0),
|
||||
"gain_g": (0.0, 2.0),
|
||||
"gain_b": (0.0, 2.0),
|
||||
"band": (0.02, 0.5),
|
||||
}
|
||||
|
||||
@@ -163,6 +166,7 @@ class State:
|
||||
with self.lock:
|
||||
d = dict(self.cfg)
|
||||
d["running"] = self.running
|
||||
d["target_fps"] = d["fps"]
|
||||
d["fps"] = round(self.fps, 1)
|
||||
d["stream_error"] = self.stream_error
|
||||
d["last_error"] = self.last_error
|
||||
@@ -189,60 +193,40 @@ def apply_params(zones, cfg):
|
||||
b = cfg["brightness"]
|
||||
if b != 1.0:
|
||||
z = z * b
|
||||
z = z * np.array([cfg["gain_r"], cfg["gain_g"], cfg["gain_b"]])
|
||||
return np.clip(z, 0, 255)
|
||||
|
||||
|
||||
def capture_loop(state, strip):
|
||||
"""Ciclo di cattura: sceglie il backend (portal o kmsgrab) e invia i frame."""
|
||||
from ambicapture import ChildCapture, KmsgrabCapture
|
||||
from ambicapture import Capture, screencast_stream
|
||||
|
||||
cap = None
|
||||
key = None
|
||||
leds_cfg = None
|
||||
prev = None
|
||||
sent = 0
|
||||
t_win = time.time()
|
||||
|
||||
def build(cfg):
|
||||
"""Crea e avvia il backend; ritorna (cap, descrizione)."""
|
||||
if cfg["capture"] == "kmsgrab":
|
||||
c = KmsgrabCapture(cfg["leds"], fps=cfg["fps"], device=cfg["drm_device"]).start()
|
||||
return c, f"kmsgrab {c.binary} su {cfg['drm_device']}"
|
||||
helper = os.path.join(os.path.dirname(os.path.abspath(__file__)), "ambicatcher.py")
|
||||
c = ChildCapture([sys.executable, helper,
|
||||
"--leds", str(cfg["leds"]), "--rows", "72", "--sub", "4",
|
||||
"--fps", str(int(cfg["fps"])), "--drain"],
|
||||
cfg["leds"]).start()
|
||||
return c, f"portal (processo separato) node={c.header.get('node')} " \
|
||||
f"sorgente {c.header.get('w')}x{c.header.get('h')}"
|
||||
|
||||
last_write = 0.0
|
||||
t_log = time.time()
|
||||
n_none = 0
|
||||
while True:
|
||||
cfg = state.snapshot()
|
||||
if not cfg["running"]:
|
||||
if cap is not None:
|
||||
cap.stop()
|
||||
cap = None
|
||||
key = None
|
||||
prev = None
|
||||
time.sleep(0.15)
|
||||
continue
|
||||
|
||||
want = (cfg["capture"], cfg["leds"], cfg["drm_device"], cfg["fps"])
|
||||
dead = cap is not None and hasattr(cap, "alive") and not cap.alive()
|
||||
if cap is None or want != key or dead:
|
||||
if cap is None or leds_cfg != cfg["leds"]:
|
||||
if cap is not None:
|
||||
err = cap.error_text() if hasattr(cap, "error_text") else ""
|
||||
cap.stop()
|
||||
cap = None
|
||||
if dead:
|
||||
with state.lock:
|
||||
state.stream_error = f"kmsgrab terminato: {err}"[:300]
|
||||
log("kmsgrab terminato:", err)
|
||||
time.sleep(3)
|
||||
continue
|
||||
try:
|
||||
cap, desc = build(cfg)
|
||||
key = want
|
||||
log(f"cattura avviata ({cfg['capture']}): {desc} led={cfg['leds']}")
|
||||
node, w, h = screencast_stream()
|
||||
cap = Capture(node, cfg["leds"], fps=cfg["target_fps"])
|
||||
cap.start()
|
||||
leds_cfg = cfg["leds"]
|
||||
log(f"cattura avviata: node={node} sorgente {w}x{h} led={cfg['leds']}")
|
||||
with state.lock:
|
||||
state.stream_error = ""
|
||||
state.last_error = ""
|
||||
@@ -255,11 +239,17 @@ def capture_loop(state, strip):
|
||||
|
||||
frame = cap.frame(200)
|
||||
if frame is None:
|
||||
n_none += 1
|
||||
time.sleep(0.005)
|
||||
if time.time() - t_log >= 10:
|
||||
log(f"diagnostica: 10 s senza frame (timeout={n_none}), "
|
||||
f"fps={state.fps} sent={sent}")
|
||||
t_log, n_none = time.time(), 0
|
||||
continue
|
||||
|
||||
rows = max(1, int(round(cap.gh * cfg["band"])))
|
||||
gh, gw = frame.shape[0], frame.shape[1]
|
||||
rows = min(max(1, int(round(gh * cfg["band"]))), gh)
|
||||
rows = min(rows, gh)
|
||||
band = frame[:rows] if cfg["side"] == "top" else frame[gh - rows:]
|
||||
sub = gw // cfg["leds"]
|
||||
if sub < 1:
|
||||
@@ -274,7 +264,18 @@ def capture_loop(state, strip):
|
||||
px.reverse()
|
||||
|
||||
state.strip_dev = strip.resolved
|
||||
if strip.write(adalight_frame(px)):
|
||||
# PACING: mai sopra ~60% della capacità della seriale. Il firmware Adalight fa
|
||||
# SERIAL_FLUSH alla chiusura del frame: se il frame successivo e' già nel buffer
|
||||
# ne perde l'header -> frame scartati e striscia nera, senza errori lato host.
|
||||
frame_bits = (6 + 3 * cfg["leds"]) * 10
|
||||
min_interval = max(1.0 / max(1, cfg["target_fps"]),
|
||||
1.7 * frame_bits / max(9600, cfg["baud"]))
|
||||
wait = min_interval - (time.time() - last_write)
|
||||
if wait > 0:
|
||||
time.sleep(wait)
|
||||
ok = strip.write(adalight_frame(px))
|
||||
if ok:
|
||||
last_write = time.time()
|
||||
sent += 1
|
||||
now = time.time()
|
||||
if now - t_win >= 1.0:
|
||||
@@ -384,12 +385,14 @@ class DBusService:
|
||||
value = bool(value)
|
||||
elif key == "leds":
|
||||
value = max(1, min(256, int(value)))
|
||||
elif key in ("port", "side", "baud", "fps", "capture", "drm_device"):
|
||||
elif key in ("port", "side", "baud", "fps"):
|
||||
value = type(DEFAULTS[key])(value)
|
||||
else:
|
||||
return False, f"parametro sconosciuto: {key}"
|
||||
self.state.update(**{key: value})
|
||||
save_config(self.state.snapshot())
|
||||
with self.state.lock:
|
||||
config = dict(self.state.cfg)
|
||||
save_config(config)
|
||||
return True, ""
|
||||
|
||||
# --- metodi
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Calibrazione Ambilight: fascia bianca e regolazioni live via ambilight-ctl."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtCore import QProcess, QTimer, Qt
|
||||
from PySide6.QtWidgets import (QApplication, QFileDialog, QHBoxLayout, QLabel,
|
||||
QMessageBox, QPushButton, QSlider, QVBoxLayout, QWidget)
|
||||
|
||||
CTL = str(Path.home() / ".local/bin/ambilight-ctl")
|
||||
FIELDS = (("gain_r", "Rosso", 0, 200), ("gain_g", "Verde", 0, 200),
|
||||
("gain_b", "Blu", 0, 200), ("brightness", "Luminosità", 5, 200))
|
||||
|
||||
|
||||
class Calibration(QWidget):
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.setWindowTitle("Calibrazione Ambilight")
|
||||
self.values = {}
|
||||
self.pending = {}
|
||||
self.initialized = False
|
||||
self.process = QProcess(self)
|
||||
self.process.finished.connect(self.finished)
|
||||
self.process.errorOccurred.connect(self.process_error)
|
||||
self.current = None
|
||||
self.debounce = QTimer(self, interval=80, singleShot=True)
|
||||
self.debounce.timeout.connect(self.pump)
|
||||
|
||||
outer = QVBoxLayout(self)
|
||||
outer.setContentsMargins(0, 0, 0, 0)
|
||||
outer.setSpacing(0)
|
||||
self.white = QWidget()
|
||||
self.white.setStyleSheet("background: white;")
|
||||
outer.addWidget(self.white, 30) # 30%, maggiore della banda predefinita del 12%
|
||||
panel = QWidget()
|
||||
panel.setStyleSheet("background: #20242b; color: #eeeeee;")
|
||||
outer.addWidget(panel, 70)
|
||||
layout = QVBoxLayout(panel)
|
||||
layout.setContentsMargins(50, 20, 50, 24)
|
||||
title = QLabel("Calibrazione Ambilight")
|
||||
title.setStyleSheet("font-size: 24px; font-weight: bold;")
|
||||
layout.addWidget(title)
|
||||
hint = QLabel("La fascia superiore è bianca. Le modifiche si applicano e salvano nel demone in tempo reale.")
|
||||
hint.setWordWrap(True)
|
||||
layout.addWidget(hint)
|
||||
self.status = QLabel("Connessione al demone…")
|
||||
layout.addWidget(self.status)
|
||||
self.sliders = {}
|
||||
self.labels = {}
|
||||
for key, name, lo, hi in FIELDS:
|
||||
row = QHBoxLayout()
|
||||
label = QLabel(name)
|
||||
label.setMinimumWidth(120)
|
||||
row.addWidget(label)
|
||||
slider = QSlider(Qt.Horizontal)
|
||||
slider.setRange(lo, hi)
|
||||
slider.setEnabled(False)
|
||||
row.addWidget(slider, 1)
|
||||
value_label = QLabel("—")
|
||||
value_label.setMinimumWidth(55)
|
||||
row.addWidget(value_label)
|
||||
slider.valueChanged.connect(lambda value, k=key: self.changed(k, value))
|
||||
self.sliders[key], self.labels[key] = slider, value_label
|
||||
layout.addLayout(row)
|
||||
layout.addStretch(1)
|
||||
buttons = QHBoxLayout()
|
||||
save = QPushButton("Esporta profilo JSON…")
|
||||
save.clicked.connect(self.export_profile)
|
||||
buttons.addWidget(save)
|
||||
buttons.addStretch(1)
|
||||
close = QPushButton("Chiudi")
|
||||
close.clicked.connect(self.close)
|
||||
buttons.addWidget(close)
|
||||
layout.addLayout(buttons)
|
||||
self.call("state", [])
|
||||
|
||||
def call(self, kind, args):
|
||||
self.current = kind
|
||||
self.process.start(CTL, args or ["state"])
|
||||
|
||||
def changed(self, key, value):
|
||||
self.labels[key].setText(f"{value}%")
|
||||
if not self.initialized:
|
||||
return
|
||||
self.pending[key] = value / 100.0
|
||||
self.status.setText("Applicazione in corso…")
|
||||
self.debounce.start()
|
||||
|
||||
def pump(self):
|
||||
if self.process.state() != QProcess.NotRunning or not self.pending:
|
||||
return
|
||||
key = next(iter(self.pending))
|
||||
value = self.pending.pop(key)
|
||||
self.call(key, ["set", key, f"{value:.2f}"])
|
||||
|
||||
def process_error(self, error):
|
||||
self.status.setText("Errore: impossibile avviare ambilight-ctl; controlla il servizio.")
|
||||
self.current = None
|
||||
|
||||
def finished(self, code, exit_status):
|
||||
kind, self.current = self.current, None
|
||||
raw = bytes(self.process.readAllStandardOutput()).decode("utf-8", "replace")
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
if code != 0 or data.get("ok") is False:
|
||||
raise ValueError(data.get("error", "servizio non raggiungibile"))
|
||||
state = data.get("state", data)
|
||||
if not state.get("daemon"):
|
||||
raise ValueError(state.get("error", "servizio non raggiungibile"))
|
||||
if kind == "state":
|
||||
for key, _, _, _ in FIELDS:
|
||||
value = round(float(state[key]) * 100)
|
||||
self.sliders[key].setValue(value)
|
||||
self.sliders[key].setEnabled(True)
|
||||
self.values[key] = value / 100
|
||||
self.initialized = True
|
||||
else:
|
||||
self.values[kind] = float(state[kind])
|
||||
self.status.setText("Valori applicati e salvati dal demone." if not self.pending
|
||||
else "Applicazione in corso…")
|
||||
except (ValueError, KeyError, TypeError) as exc:
|
||||
self.status.setText(f"Errore: {exc}. Valore non confermato.")
|
||||
self.pump()
|
||||
|
||||
def export_profile(self):
|
||||
if not self.initialized or self.pending or self.process.state() != QProcess.NotRunning:
|
||||
QMessageBox.warning(self, "Profilo", "Attendi la conferma di tutte le regolazioni.")
|
||||
return
|
||||
filename, _ = QFileDialog.getSaveFileName(self, "Esporta profilo", "ambilight-calibrazione.json", "JSON (*.json)")
|
||||
if not filename:
|
||||
return
|
||||
try:
|
||||
with open(filename, "w", encoding="utf-8") as f:
|
||||
json.dump({key: self.values[key] for key, _, _, _ in FIELDS}, f, indent=2)
|
||||
f.write("\n")
|
||||
self.status.setText(f"Profilo esportato: {filename}")
|
||||
except OSError as exc:
|
||||
QMessageBox.critical(self, "Errore esportazione", str(exc))
|
||||
|
||||
def closeEvent(self, event):
|
||||
if self.pending or self.process.state() != QProcess.NotRunning:
|
||||
self.status.setText("Attendi la conferma delle modifiche prima di chiudere.")
|
||||
event.ignore()
|
||||
return
|
||||
event.accept()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app = QApplication(sys.argv)
|
||||
window = Calibration()
|
||||
window.showFullScreen()
|
||||
sys.exit(app.exec())
|
||||
+4
-1
@@ -27,6 +27,7 @@ check "import gi; gi.require_version('Gst','1.0'); gi.require_version('GstApp','
|
||||
check "import dbus" "python3-dbus"
|
||||
check "import serial" "python3-pyserial"
|
||||
check "import numpy" "python3-numpy"
|
||||
check "import PySide6" "python3-pyside6 (calibrazione)"
|
||||
if ! gst-inspect-1.0 pipewiresrc >/dev/null 2>&1; then
|
||||
warn "manca: gstreamer1-plugin-pipewire (pipewiresrc)"
|
||||
missing=1
|
||||
@@ -43,6 +44,7 @@ install -m 755 "$HERE/backend/ambilightd.py" "$BIN/ambilightd.py"
|
||||
install -m 644 "$HERE/backend/ambicapture.py" "$BIN/ambicapture.py"
|
||||
install -m 755 "$HERE/backend/ambicatcher.py" "$BIN/ambicatcher.py"
|
||||
install -m 755 "$HERE/backend/ambilight-ctl" "$BIN/ambilight-ctl"
|
||||
install -m 755 "$HERE/calibration/ambilight_calibrate.py" "$BIN/ambilight-calibrate"
|
||||
|
||||
say "Installo il servizio utente"
|
||||
install -d "$SYSDIR"
|
||||
@@ -56,7 +58,8 @@ say "Installo il plasmoide $APPLET_ID"
|
||||
TMP="$(mktemp -d)"
|
||||
trap 'rm -rf "$TMP"' EXIT
|
||||
cp -r "$HERE/package/$APPLET_ID" "$TMP/"
|
||||
sed -i "s|@CTL@|$BIN/ambilight-ctl|g" "$TMP/$APPLET_ID/contents/ui/main.qml"
|
||||
sed -i -e "s|@CTL@|$BIN/ambilight-ctl|g" \
|
||||
-e "s|@CALIBRATE@|$BIN/ambilight-calibrate|g" "$TMP/$APPLET_ID/contents/ui/main.qml"
|
||||
if kpackagetool6 --type Plasma/Applet --show "$APPLET_ID" >/dev/null 2>&1; then
|
||||
kpackagetool6 --type Plasma/Applet --upgrade "$TMP/$APPLET_ID"
|
||||
else
|
||||
|
||||
@@ -10,6 +10,9 @@ PlasmoidItem {
|
||||
|
||||
// Percorso dell'helper, sostituito da install.sh con il percorso assoluto
|
||||
readonly property string ctl: "@CTL@"
|
||||
readonly property string calibration: "@CALIBRATE@"
|
||||
property bool polling: false
|
||||
property bool calibrating: false
|
||||
|
||||
property var st: ({ daemon: false, running: false, brightness: 1.0,
|
||||
contrast: 1.0, saturation: 1.0, band: 0.12,
|
||||
@@ -28,9 +31,18 @@ PlasmoidItem {
|
||||
}
|
||||
|
||||
function refresh() {
|
||||
if (root.polling) return
|
||||
root.polling = true
|
||||
poll.connectSource(root.ctl + " state")
|
||||
}
|
||||
|
||||
Timer {
|
||||
interval: 2000
|
||||
running: true
|
||||
repeat: true
|
||||
onTriggered: root.refresh()
|
||||
}
|
||||
|
||||
function run(args) {
|
||||
if (root.busy)
|
||||
return
|
||||
@@ -46,6 +58,7 @@ PlasmoidItem {
|
||||
connectedSources: []
|
||||
onNewData: function(source, data) {
|
||||
poll.disconnectSource(source)
|
||||
root.polling = false
|
||||
const out = String(data["stdout"] || "").trim()
|
||||
if (!out)
|
||||
return
|
||||
@@ -248,6 +261,14 @@ PlasmoidItem {
|
||||
RowLayout {
|
||||
Layout.fillWidth: true
|
||||
|
||||
PC3.Button {
|
||||
text: i18n("Calibra…")
|
||||
enabled: root.st.daemon && !root.calibrating
|
||||
onClicked: {
|
||||
root.calibrating = true
|
||||
calibrationSource.connectSource("/usr/bin/python3 " + JSON.stringify(root.calibration))
|
||||
}
|
||||
}
|
||||
PC3.Button {
|
||||
text: i18n("Riavvia demone")
|
||||
font: Kirigami.Theme.smallFont
|
||||
@@ -279,5 +300,18 @@ PlasmoidItem {
|
||||
}
|
||||
}
|
||||
|
||||
P5Support.DataSource {
|
||||
id: calibrationSource
|
||||
engine: "executable"
|
||||
connectedSources: []
|
||||
onNewData: function(source, data) {
|
||||
calibrationSource.disconnectSource(source)
|
||||
root.calibrating = false
|
||||
if (data["exit code"] !== 0)
|
||||
root.lastError = String(data["stderr"] || i18n("Impossibile avviare la calibrazione"))
|
||||
root.refresh()
|
||||
}
|
||||
}
|
||||
|
||||
Component.onCompleted: root.refresh()
|
||||
}
|
||||
|
||||
+1
-1
@@ -18,6 +18,6 @@ say "Rimuovo il plasmoide"
|
||||
kpackagetool6 --type Plasma/Applet --remove "$APPLET_ID" 2>/dev/null || true
|
||||
|
||||
say "Rimuovo il backend"
|
||||
rm -f "$BIN/ambilightd.py" "$BIN/ambicapture.py" "$BIN/ambilight-ctl"
|
||||
rm -f "$BIN/ambilightd.py" "$BIN/ambicapture.py" "$BIN/ambilight-ctl" "$BIN/ambilight-calibrate"
|
||||
|
||||
say "Fatto. Configurazione lasciata in ~/.config/ambilight"
|
||||
|
||||
Reference in New Issue
Block a user