Ambilight per Plasma 6: demone Adalight + plasmoide con slider
- backend/ambicapture.py: cattura Wayland via ScreenCast portal + PipeWire - backend/ambilightd.py: demone, 47 zone sul bordo superiore, luminosita/contrasto/ saturazione/banda, invio Adalight su seriale, servizio D-Bus org.enne2.Ambilight - backend/ambilight-ctl: CLI JSON per il plasmoide - backend/ambilight.service: avvio automatico (systemd --user) - package/org.enne2.ambilight: applet vassoio con popup e slider - install.sh / uninstall.sh senza root
This commit is contained in:
enne2
@@ -0,0 +1,115 @@
|
||||
"""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)
|
||||
Executable
+99
@@ -0,0 +1,99 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ambilight-ctl — interfaccia a riga di comando per il demone ambilightd.
|
||||
|
||||
Parla con il demone via D-Bus e stampa JSON, cosi' il plasmoide puo' usarla
|
||||
attraverso il DataSource "executable" di Plasma.
|
||||
|
||||
ambilight-ctl state # JSON con tutto lo stato
|
||||
ambilight-ctl start | stop | toggle
|
||||
ambilight-ctl set brightness 0.8 # brightness|contrast|saturation|band|reverse|leds
|
||||
ambilight-ctl get contrast
|
||||
"""
|
||||
import json
|
||||
import sys
|
||||
|
||||
BUS_NAME = "org.enne2.Ambilight"
|
||||
OBJ_PATH = "/org/enne2/Ambilight"
|
||||
IFACE = "org.enne2.Ambilight"
|
||||
|
||||
FALLBACK = {"daemon": False, "running": False, "brightness": 1.0, "contrast": 1.0,
|
||||
"saturation": 1.0, "band": 0.12, "reverse": True, "leds": 47, "fps": 0.0,
|
||||
"error": "demone non raggiungibile"}
|
||||
|
||||
|
||||
def proxy():
|
||||
from gi.repository import Gio, GLib
|
||||
return Gio.DBusProxy.new_for_bus_sync(
|
||||
Gio.BusType.SESSION, Gio.DBusProxyFlags.NONE, None,
|
||||
BUS_NAME, OBJ_PATH, IFACE, None)
|
||||
|
||||
|
||||
def call(method, args=None):
|
||||
from gi.repository import Gio
|
||||
return proxy().call_sync(method, args, Gio.DBusCallFlags.NONE, 1500, None)
|
||||
|
||||
|
||||
def state():
|
||||
try:
|
||||
res = call("GetState")
|
||||
data = json.loads(res.unpack()[0] if isinstance(res.unpack(), tuple) else res.unpack())
|
||||
data["daemon"] = True
|
||||
return data
|
||||
except Exception as e:
|
||||
d = dict(FALLBACK)
|
||||
d["error"] = f"{type(e).__name__}: {e}".strip()
|
||||
return d
|
||||
|
||||
|
||||
def main(argv):
|
||||
cmd = argv[1] if len(argv) > 1 else "state"
|
||||
|
||||
if cmd == "state":
|
||||
d = state()
|
||||
print(json.dumps(d))
|
||||
return 0 if d.get("daemon") else 1
|
||||
|
||||
if cmd in ("start", "stop", "toggle"):
|
||||
try:
|
||||
call({"start": "Start", "stop": "Stop", "toggle": "Toggle"}[cmd])
|
||||
except Exception as e:
|
||||
print(json.dumps({"ok": False, "error": str(e)}))
|
||||
return 1
|
||||
print(json.dumps({"ok": True, "state": state()}))
|
||||
return 0
|
||||
|
||||
if cmd == "set" and len(argv) >= 4:
|
||||
key, raw = argv[2], argv[3]
|
||||
if key in ("reverse",):
|
||||
value = raw.lower() in ("1", "true", "yes", "on")
|
||||
elif key == "leds":
|
||||
value = int(raw)
|
||||
else:
|
||||
value = float(raw)
|
||||
try:
|
||||
from gi.repository import GLib
|
||||
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}))
|
||||
return 1
|
||||
except Exception as e:
|
||||
print(json.dumps({"ok": False, "error": str(e)}))
|
||||
return 1
|
||||
print(json.dumps({"ok": True, "state": state()}))
|
||||
return 0
|
||||
|
||||
if cmd == "get" and len(argv) >= 3:
|
||||
d = state()
|
||||
key = argv[2]
|
||||
print(d.get(key, ""))
|
||||
return 0 if d.get("daemon") else 1
|
||||
|
||||
print(__doc__)
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv))
|
||||
@@ -0,0 +1,15 @@
|
||||
[Unit]
|
||||
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
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python3 %h/.local/bin/ambilightd.py
|
||||
Environment=PYTHONUNBUFFERED=1
|
||||
Restart=on-failure
|
||||
RestartSec=3
|
||||
|
||||
[Install]
|
||||
WantedBy=graphical-session.target
|
||||
Executable
+421
@@ -0,0 +1,421 @@
|
||||
#!/usr/bin/env python3
|
||||
"""ambilightd — demone dell'ambilight (ESP8266 + striscia WS2812B via Adalight).
|
||||
|
||||
Cattura lo schermo su Wayland (ScreenCast portal + PipeWire), calcola il colore
|
||||
medio delle zone lungo il bordo scelto, applica luminosita'/contrasto/saturazione
|
||||
e invia i frame sulla seriale con il protocollo Adalight.
|
||||
|
||||
Espone un servizio D-Bus di sessione:
|
||||
|
||||
nome: org.enne2.Ambilight
|
||||
oggetto: /org/enne2/Ambilight
|
||||
interfaccia: org.enne2.Ambilight
|
||||
|
||||
metodi: Start(), Stop(), GetState() -> s (JSON), SetParam(s,s), Quit()
|
||||
proprieta': Running, Brightness, Contrast, Saturation, Band, Reverse, Leds, Fps
|
||||
segnale: Changed()
|
||||
|
||||
Config: ~/.config/ambilight/config.json
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
|
||||
import numpy as np
|
||||
|
||||
APP = "ambilightd"
|
||||
BUS_NAME = "org.enne2.Ambilight"
|
||||
OBJ_PATH = "/org/enne2/Ambilight"
|
||||
IFACE = "org.enne2.Ambilight"
|
||||
CONFIG_DIR = os.path.expanduser("~/.config/ambilight")
|
||||
CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json")
|
||||
|
||||
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
|
||||
"band": 0.12, # frazione di altezza campionata sul bordo
|
||||
"reverse": True, # LED 0 sul lato destro dello schermo (misurato su frigate)
|
||||
"leds": 47,
|
||||
"port": "/dev/ttyUSB0",
|
||||
"baud": 115200,
|
||||
"fps": 60,
|
||||
"side": "top", # top | bottom
|
||||
"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),
|
||||
"band": (0.02, 0.5),
|
||||
}
|
||||
|
||||
|
||||
def log(*a):
|
||||
print(time.strftime("%H:%M:%S"), *a, flush=True)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ config
|
||||
def load_config():
|
||||
cfg = dict(DEFAULTS)
|
||||
try:
|
||||
with open(CONFIG_FILE) as f:
|
||||
cfg.update(json.load(f))
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
except Exception as e:
|
||||
log("config illeggibile, uso i default:", e)
|
||||
return cfg
|
||||
|
||||
|
||||
def save_config(cfg):
|
||||
"""Salva solo le chiavi di configurazione (niente fps/stato transitorio)."""
|
||||
os.makedirs(CONFIG_DIR, exist_ok=True)
|
||||
data = {k: cfg[k] for k in DEFAULTS if k in cfg}
|
||||
tmp = CONFIG_FILE + ".tmp"
|
||||
with open(tmp, "w") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
os.replace(tmp, CONFIG_FILE)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ adalight
|
||||
def adalight_frame(pixels):
|
||||
count = len(pixels) - 1
|
||||
hi, lo = (count >> 8) & 0xFF, count & 0xFF
|
||||
return b"Ada" + bytes((hi, lo, hi ^ lo ^ 0x55)) + bytes(int(c) for px in pixels for c in px)
|
||||
|
||||
|
||||
class Strip:
|
||||
"""Porta seriale con apertura pigra e riconnessione."""
|
||||
|
||||
def __init__(self, port, baud):
|
||||
self.port, self.baud = port, baud
|
||||
self.ser = None
|
||||
self.error = ""
|
||||
|
||||
def _open(self):
|
||||
import serial
|
||||
self.ser = serial.Serial(self.port, self.baud, timeout=0.2, write_timeout=0.2)
|
||||
try:
|
||||
self.ser.setDTR(False)
|
||||
self.ser.setRTS(False)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def write(self, frame):
|
||||
try:
|
||||
if self.ser is None:
|
||||
self._open()
|
||||
self.error = ""
|
||||
self.ser.write(frame)
|
||||
return True
|
||||
except Exception as e:
|
||||
self.error = f"{type(e).__name__}: {e}"
|
||||
try:
|
||||
if self.ser is not None:
|
||||
self.ser.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.ser = None
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
if self.ser is not None:
|
||||
self.ser.close()
|
||||
except Exception:
|
||||
pass
|
||||
self.ser = None
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ stato
|
||||
class State:
|
||||
def __init__(self, cfg):
|
||||
self.lock = threading.Lock()
|
||||
self.cfg = cfg
|
||||
self.running = bool(cfg.get("auto_start", True))
|
||||
self.fps = 0.0
|
||||
self.stream_error = ""
|
||||
self.last_error = ""
|
||||
|
||||
def snapshot(self):
|
||||
with self.lock:
|
||||
d = dict(self.cfg)
|
||||
d["running"] = self.running
|
||||
d["fps"] = round(self.fps, 1)
|
||||
d["stream_error"] = self.stream_error
|
||||
d["last_error"] = self.last_error
|
||||
return d
|
||||
|
||||
def update(self, **kw):
|
||||
with self.lock:
|
||||
self.cfg.update(kw)
|
||||
return dict(self.cfg)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ cattura
|
||||
def apply_params(zones, cfg):
|
||||
"""zones: (led, 3) float. Applica saturazione, contrasto, luminosita', gamma."""
|
||||
z = zones
|
||||
sat = cfg["saturation"]
|
||||
if sat != 1.0:
|
||||
m = z.mean(axis=1, keepdims=True)
|
||||
z = m + (z - m) * sat
|
||||
c = cfg["contrast"]
|
||||
if c != 1.0:
|
||||
z = (z - 128.0) * c + 128.0
|
||||
b = cfg["brightness"]
|
||||
if b != 1.0:
|
||||
z = z * b
|
||||
return np.clip(z, 0, 255)
|
||||
|
||||
|
||||
def capture_loop(state, strip):
|
||||
from ambicapture import Capture, screencast_stream
|
||||
|
||||
cap = None
|
||||
leds_cfg = None
|
||||
prev = None
|
||||
sent = 0
|
||||
t_win = time.time()
|
||||
while True:
|
||||
cfg = state.snapshot()
|
||||
if not cfg["running"]:
|
||||
if cap is not None:
|
||||
cap.stop()
|
||||
cap = None
|
||||
prev = None
|
||||
time.sleep(0.15)
|
||||
continue
|
||||
|
||||
if cap is None or leds_cfg != cfg["leds"]:
|
||||
if cap is not None:
|
||||
cap.stop()
|
||||
try:
|
||||
node, w, h = screencast_stream()
|
||||
cap = Capture(node, cfg["leds"], fps=cfg["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 = ""
|
||||
except Exception as e:
|
||||
with state.lock:
|
||||
state.stream_error = f"{type(e).__name__}: {e}"
|
||||
log("cattura non disponibile:", e)
|
||||
time.sleep(5)
|
||||
continue
|
||||
|
||||
frame = cap.frame(200)
|
||||
if frame is None:
|
||||
time.sleep(0.005)
|
||||
continue
|
||||
|
||||
rows = max(1, int(round(cap.gh * cfg["band"])))
|
||||
gh, gw = frame.shape[0], frame.shape[1]
|
||||
rows = min(rows, gh)
|
||||
band = frame[:rows] if cfg["side"] == "top" else frame[gh - rows:]
|
||||
sub = gw // cfg["leds"]
|
||||
if sub < 1:
|
||||
continue
|
||||
zones = band.reshape(rows, cfg["leds"], sub, 3).mean(axis=(0, 2))
|
||||
if prev is not None:
|
||||
zones = zones * 0.65 + prev * 0.35
|
||||
prev = zones
|
||||
colors = apply_params(zones, cfg)
|
||||
px = [tuple(int(v) for v in c) for c in colors]
|
||||
if cfg["reverse"]:
|
||||
px.reverse()
|
||||
|
||||
if strip.write(adalight_frame(px)):
|
||||
sent += 1
|
||||
now = time.time()
|
||||
if now - t_win >= 1.0:
|
||||
with state.lock:
|
||||
state.fps = sent / (now - t_win)
|
||||
sent, t_win = 0, now
|
||||
else:
|
||||
with state.lock:
|
||||
state.last_error = strip.error
|
||||
time.sleep(0.5)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ D-Bus
|
||||
INTROSPECTION = f"""<node>
|
||||
<interface name="{IFACE}">
|
||||
<method name="Start"/>
|
||||
<method name="Stop"/>
|
||||
<method name="Toggle"/>
|
||||
<method name="Quit"/>
|
||||
<method name="GetState"><arg type="s" direction="out"/></method>
|
||||
<method name="SetParam">
|
||||
<arg type="s" direction="in"/><arg type="v" direction="in"/>
|
||||
<arg type="s" direction="out"/>
|
||||
</method>
|
||||
<signal name="Changed"/>
|
||||
<property name="Running" type="b" access="read"/>
|
||||
<property name="Brightness" type="d" access="readwrite"/>
|
||||
<property name="Contrast" type="d" access="readwrite"/>
|
||||
<property name="Saturation" type="d" access="readwrite"/>
|
||||
<property name="Band" type="d" access="readwrite"/>
|
||||
<property name="Reverse" type="b" access="readwrite"/>
|
||||
<property name="Leds" type="i" access="readwrite"/>
|
||||
<property name="Fps" type="d" access="read"/>
|
||||
</interface>
|
||||
</node>"""
|
||||
|
||||
PROPS = ["Running", "Brightness", "Contrast", "Saturation", "Band", "Reverse", "Leds", "Fps"]
|
||||
|
||||
|
||||
class DBusService:
|
||||
def __init__(self, state, strip):
|
||||
from gi.repository import Gio, GLib
|
||||
|
||||
self.Gio, self.GLib = Gio, GLib
|
||||
self.state, self.strip = state, strip
|
||||
node = Gio.DBusNodeInfo.new_for_xml(INTROSPECTION)
|
||||
self.info = node.interfaces[0]
|
||||
self.conn = Gio.bus_get_sync(Gio.BusType.SESSION, None)
|
||||
self.loop = GLib.MainLoop()
|
||||
self.reg_id = self.conn.register_object(OBJ_PATH, self.info,
|
||||
self.on_call, self.on_get, self.on_set)
|
||||
Gio.bus_own_name_on_connection(self.conn, BUS_NAME, Gio.BusNameOwnerFlags.NONE,
|
||||
self.on_acquired, None)
|
||||
|
||||
# --- helpers
|
||||
def _prop_value(self, name):
|
||||
snap = self.state.snapshot()
|
||||
table = {
|
||||
"Running": snap["running"],
|
||||
"Brightness": float(snap["brightness"]),
|
||||
"Contrast": float(snap["contrast"]),
|
||||
"Saturation": float(snap["saturation"]),
|
||||
"Band": float(snap["band"]),
|
||||
"Reverse": bool(snap["reverse"]),
|
||||
"Leds": int(snap["leds"]),
|
||||
"Fps": float(snap["fps"]),
|
||||
}
|
||||
return table[name]
|
||||
|
||||
def _emit_changed(self, props=None):
|
||||
try:
|
||||
self.conn.emit_signal(None, OBJ_PATH, IFACE, "Changed", None)
|
||||
if props:
|
||||
changed = {}
|
||||
for p in props:
|
||||
changed[p] = self._variant(p)
|
||||
self.conn.emit_signal(
|
||||
None, OBJ_PATH, "org.freedesktop.DBus.Properties", "PropertiesChanged",
|
||||
self.GLib.Variant("(sa{sv}as)",
|
||||
(IFACE, changed, [])))
|
||||
except Exception as e:
|
||||
log("emit fallita:", e)
|
||||
|
||||
def _variant(self, name):
|
||||
v = self._prop_value(name)
|
||||
return self.GLib.Variant("d" if isinstance(v, float) else
|
||||
"b" if isinstance(v, bool) else "i", v)
|
||||
|
||||
# --- Properties
|
||||
def on_get(self, conn, sender, path, iface, prop):
|
||||
if prop not in PROPS:
|
||||
return None
|
||||
return self._variant(prop)
|
||||
|
||||
def on_set(self, conn, sender, path, iface, prop, value):
|
||||
ok, msg = self.set_param(prop.lower(), value.unpack())
|
||||
if not ok:
|
||||
return False
|
||||
self._emit_changed([prop])
|
||||
return True
|
||||
|
||||
def set_param(self, key, value):
|
||||
if key in PARAM_RANGES:
|
||||
lo, hi = PARAM_RANGES[key]
|
||||
value = min(hi, max(lo, float(value)))
|
||||
elif key == "reverse":
|
||||
value = bool(value)
|
||||
elif key == "leds":
|
||||
value = max(1, min(256, int(value)))
|
||||
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())
|
||||
return True, ""
|
||||
|
||||
# --- metodi
|
||||
def on_call(self, conn, sender, path, iface, method, params, invocation):
|
||||
try:
|
||||
if method == "Start":
|
||||
self.state.running = True
|
||||
invocation.return_value(None)
|
||||
self._emit_changed(["Running"])
|
||||
elif method == "Stop":
|
||||
self.state.running = False
|
||||
self.strip.write(adalight_frame([(0, 0, 0)] * self.state.cfg["leds"]))
|
||||
invocation.return_value(None)
|
||||
self._emit_changed(["Running"])
|
||||
elif method == "Toggle":
|
||||
self.state.running = not self.state.running
|
||||
if not self.state.running:
|
||||
self.strip.write(adalight_frame([(0, 0, 0)] * self.state.cfg["leds"]))
|
||||
invocation.return_value(None)
|
||||
self._emit_changed(["Running"])
|
||||
elif method == "GetState":
|
||||
invocation.return_value(self.GLib.Variant("(s)", (json.dumps(self.state.snapshot()),)))
|
||||
elif method == "SetParam":
|
||||
key, val = params.unpack()
|
||||
ok, msg = self.set_param(str(key), val)
|
||||
if ok:
|
||||
self._emit_changed()
|
||||
invocation.return_value(self.GLib.Variant("(s)", ("" if ok else msg,)))
|
||||
elif method == "Quit":
|
||||
invocation.return_value(None)
|
||||
self.GLib.timeout_add(100, self.quit)
|
||||
else:
|
||||
invocation.return_dbus_error("org.enne2.Ambilight.UnknownMethod", method)
|
||||
except Exception as e:
|
||||
log("errore metodo", method, e)
|
||||
invocation.return_value(self.GLib.Variant("(s)", (str(e),)))
|
||||
|
||||
def on_acquired(self, conn, name, *a):
|
||||
log(f"servizio D-Bus attivo: {name}")
|
||||
|
||||
def quit(self):
|
||||
self.state.running = False
|
||||
self.strip.close()
|
||||
self.loop.quit()
|
||||
return False
|
||||
|
||||
def run(self):
|
||||
log("in ascolto su", OBJ_PATH)
|
||||
self.loop.run()
|
||||
|
||||
|
||||
def main():
|
||||
cfg = load_config()
|
||||
state = State(cfg)
|
||||
strip = Strip(cfg["port"], cfg["baud"])
|
||||
svc = DBusService(state, strip)
|
||||
th = threading.Thread(target=capture_loop, args=(state, strip), daemon=True)
|
||||
th.start()
|
||||
try:
|
||||
svc.run()
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
finally:
|
||||
state.running = False
|
||||
try:
|
||||
strip.write(adalight_frame([(0, 0, 0)] * cfg["leds"]))
|
||||
except Exception:
|
||||
pass
|
||||
strip.close()
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user