enne2
db47640846
- udev/99-ambilight.rules: symlink /dev/ambilight ancorato alla porta USB fisica (il CH340 clone non ha iSerial, quindi by-id non e' univoco), ID_MM_DEVICE_IGNORE, TAG+=uaccess - setup-udev.sh: rileva VID/PID/porta, installa la regola, verifica il symlink - backend: fallback di porta (config -> /dev/ambilight -> /dev/ttyUSB0) e campo 'dev' nello stato D-Bus - README: tabella dei percorsi, prove sull'indicatore di condivisione schermo e le tre alternative (portal, KWin ScreenShot2, kmsgrab)
442 lines
14 KiB
Python
Executable File
442 lines
14 KiB
Python
Executable File
#!/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, nome stabile e riconnessione.
|
|
|
|
Il percorso configurato ha la precedenza; se non esiste si prova il symlink
|
|
udev /dev/ambilight e infine /dev/ttyUSB0 (primo disponibile).
|
|
"""
|
|
|
|
def __init__(self, port, baud):
|
|
self.port, self.baud = port, baud
|
|
self.ser = None
|
|
self.resolved = port
|
|
self.error = ""
|
|
|
|
def _candidates(self):
|
|
seen = []
|
|
for p in (self.port, "/dev/ambilight", "/dev/ttyUSB0"):
|
|
if p and p not in seen:
|
|
seen.append(p)
|
|
return [p for p in seen if os.path.exists(p)]
|
|
|
|
def _open(self):
|
|
import serial
|
|
cands = self._candidates()
|
|
if not cands:
|
|
raise RuntimeError(f"nessun dispositivo seriale fra {self.port}, /dev/ambilight, /dev/ttyUSB0")
|
|
self.resolved = cands[0]
|
|
self.ser = serial.Serial(self.resolved, 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.strip_dev = cfg.get("port", "")
|
|
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
|
|
d["dev"] = getattr(self, "strip_dev", "")
|
|
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()
|
|
|
|
state.strip_dev = strip.resolved
|
|
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"])
|
|
state.strip_dev = strip.resolved
|
|
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())
|