diff --git a/avatar/assistant.py b/avatar/assistant.py index 5383db8..f61e19b 100644 --- a/avatar/assistant.py +++ b/avatar/assistant.py @@ -118,6 +118,9 @@ class Assistant: for ts, chat, sender, text in rows: who = chat if sender == chat else f"{sender} nel gruppo {chat}" self.ui.write_log(f"WhatsApp: {who}: {text[:200]}") + if sender == chat: # sul pannello LED solo le chat private, i gruppi sarebbero troppi + from avatar.monitor import _panel_notify + _panel_notify(f"WA {chat}: {text[:80]}") if self.settings.get("whatsapp_annuncia") and not self._busy and not self._speaking: self.say(f"Messaggio WhatsApp da {who}: {text[:220]}") @@ -148,6 +151,8 @@ class Assistant: else: self.ui.write_log(f"SYS: Timer: {testo}") self.say(f"Promemoria: {testo}") + from avatar.monitor import _panel_notify + _panel_notify(f"Promemoria: {testo}", alert=True) except Exception as err: print(f"[scheduler] {err}") diff --git a/avatar/ipixel.py b/avatar/ipixel.py new file mode 100644 index 0000000..02612b4 --- /dev/null +++ b/avatar/ipixel.py @@ -0,0 +1,306 @@ +"""Pannello LED iPIXEL (BLE, 64×16): connessione persistente in un thread asyncio, coda di schermate con priorità. + +Protocollo (dal progetto Flutter ~/Downloads/Progetti2026/iPixel): servizio fa01, scrittura fa02, notifiche fa03; +comandi brevi (luminosità, accensione, orologio) e immagini PNG/GIF inviate a pacchetti da 244 byte con intestazione +[len][tipo 2=PNG 3=GIF][opt][lunghezza dati][crc32][0][slot]. + +Priorità: notifica (temporanea) > stato dell'assistente (mentre pensa/parla) > riposo (orologio del pannello). +""" +from __future__ import annotations + +import asyncio +import io +import random +import struct +import threading +import time +import zlib + +from PIL import Image, ImageDraw, ImageFont + +SVC_WRITE = "0000fa02-0000-1000-8000-00805f9b34fb" +SVC_NOTIFY = "0000fa03-0000-1000-8000-00805f9b34fb" +SIZES = {128: (64, 64), 129: (32, 32), 130: (32, 16), 131: (64, 16), 132: (96, 16), 133: (64, 20), 134: (128, 16), + 135: (144, 16), 136: (192, 16), 137: (48, 24), 138: (16, 16), 139: (160, 16), 140: (32, 8), 141: (96, 32), + 142: (64, 32), 143: (128, 32), 144: (256, 16), 145: (320, 16), 146: (384, 16), 147: (448, 32)} +COLORI = {"LISTENING": (0, 255, 120), "THINKING": (255, 190, 0), "PROCESSING": (255, 190, 0), "SPEAKING": (255, 90, 0), + "SLEEPING": (40, 120, 140), "MUTED": (255, 40, 90), "INFO": (0, 200, 255), "ALERT": (255, 40, 60)} +NOMI_COLORI = {"rosso": (255, 30, 30), "verde": (0, 230, 80), "blu": (40, 90, 255), "azzurro": (0, 200, 255), "giallo": (255, 210, 0), + "arancione": (255, 120, 0), "viola": (170, 60, 255), "rosa": (255, 80, 170), "bianco": (255, 255, 255), "ciano": (0, 255, 230)} +FONT_PATHS = ("/System/Library/Fonts/Menlo.ttc", "/System/Library/Fonts/Supplemental/Arial Bold.ttf") + + +def _font(size: int): + for p in FONT_PATHS: + try: + return ImageFont.truetype(p, size) + except Exception: + continue + return ImageFont.load_default() + + +def packets(data: bytes, slot: int = 1, gif: bool = False) -> list[bytes]: + typ = 3 if gif else 2 + hdr = struct.pack(" tuple[bytes, bool]: + """Testo su un pannello: statico se entra, altrimenti GIF scorrevole. Restituisce (dati, è_gif).""" + w, h = size + font = _font(11 if h <= 16 else 14) + pad = 6 if icon_color else 0 + meas = ImageDraw.Draw(Image.new("RGB", (1, 1))) + tw = int(meas.textlength(text, font=font)) + bbox = meas.textbbox((0, 0), text, font=font) + ty = (h - (bbox[3] - bbox[1])) // 2 - bbox[1] + + def frame(x: int) -> Image.Image: + im = Image.new("RGB", (w, h), (0, 0, 0)); d = ImageDraw.Draw(im); d.fontmode = "1" + d.text((x, ty), text, fill=color, font=font) + if bold: + d.text((x + 1, ty), text, fill=color, font=font) + if icon_color: + d.rectangle([0, 0, pad - 2, h - 1], fill=(0, 0, 0)); d.ellipse([0, h // 2 - 2, 4, h // 2 + 2], fill=icon_color) + return im + + if tw + pad <= w: + buf = io.BytesIO(); frame(pad + (w - pad - tw) // 2).save(buf, "PNG"); return buf.getvalue(), False + frames = [frame(x) for x in range(w, -tw - 2, -2)] + buf = io.BytesIO() + frames[0].save(buf, "GIF", save_all=True, append_images=frames[1:], duration=45, loop=0, disposal=1, optimize=False) + return buf.getvalue(), True + + +def render_image(path: str, size=(64, 16)) -> bytes: + im = Image.open(path).convert("RGB") + im.thumbnail(size, Image.LANCZOS) + bg = Image.new("RGB", size, (0, 0, 0)); bg.paste(im, ((size[0] - im.width) // 2, (size[1] - im.height) // 2)) + buf = io.BytesIO(); bg.save(buf, "PNG"); return buf.getvalue() + + +class IPixel: + def __init__(self, settings, log=print) -> None: + self.settings, self.log = settings, log + self.size = (64, 16) + self.connected = False + self.error = "" + self._loop: asyncio.AbstractEventLoop | None = None + self._client = None + self._wake: asyncio.Event | None = None + self._notify_until = 0.0 + self._want: tuple | None = None # (chiave, dati, gif) da mostrare + self._shown_key = None + self._state_key = None + self._lock = threading.Lock() + self._stop = False + + # ── ciclo BLE ────────────────────────────────────────────────────────── + def start(self) -> None: + threading.Thread(target=self._run, daemon=True, name="ipixel").start() + + def stop(self) -> None: + self._stop = True + if self._loop: + self._loop.call_soon_threadsafe(lambda: None) + + def _run(self) -> None: + self._loop = asyncio.new_event_loop() + asyncio.set_event_loop(self._loop) + self._wake = asyncio.Event() + self._loop.run_until_complete(self._main()) + + async def _connect(self): + from bleak import BleakClient, BleakScanner + addr = str(self.settings.get("ipixel_address") or "") + dev = None + if addr: + dev = await BleakScanner.find_device_by_address(addr, timeout=10) + if dev is None: + dev = await BleakScanner.find_device_by_filter(lambda d, a: (d.name or a.local_name or "").startswith("LED_BLE_"), timeout=12) + if dev is None: + raise RuntimeError("pannello iPIXEL non trovato (acceso e vicino?)") + if dev.address != addr: + self.settings.set("ipixel_address", dev.address); self.settings.save() + c = BleakClient(dev, disconnected_callback=lambda _c: self._on_disconnect()) + await c.connect(timeout=15) + notes: list[bytes] = [] + await c.start_notify(SVC_NOTIFY, lambda _h, d: notes.append(bytes(d))) + await c.write_gatt_char(SVC_WRITE, bytes([8, 0, 1, 0x80, 0, 0, 0, 0]), response=False) + await asyncio.sleep(0.8) + for n in notes: + if len(n) >= 5 and n[2:4] == b"\x01\x80": + self.size = SIZES.get(n[4], self.size) + self._client = c + self.connected, self.error = True, "" + await self._cmd([5, 0, 4, 0x80, int(self.settings.get("ipixel_luminosita") or 40)]) + await self._sync_clock() + self.log(f"SYS: Pannello iPIXEL collegato ({self.size[0]}×{self.size[1]}).") + + def _on_disconnect(self) -> None: + self.connected = False + self._client = None + self._shown_key = None + + async def _cmd(self, frame: list[int]) -> None: + if self._client: + await self._client.write_gatt_char(SVC_WRITE, bytes(frame), response=False) + + async def _send(self, data: bytes, gif: bool) -> None: + for p in packets(data, 1, gif): + await self._client.write_gatt_char(SVC_WRITE, p, response=False) + await asyncio.sleep(0.015) + + async def _sync_clock(self) -> None: + t = time.localtime() + await self._cmd([8, 0, 1, 0x80, t.tm_hour, t.tm_min, t.tm_sec, 0]) + + async def _show_clock(self) -> None: + # Prima la modalità orologio, poi l'ora: inviata prima, il pannello la azzera entrando in modalità orologio. + t = time.localtime() + await self._cmd([11, 0, 6, 1, int(self.settings.get("ipixel_orologio_stile") or 1), 1, 0, t.tm_year % 100, t.tm_mon, t.tm_mday, t.tm_wday + 1]) + await asyncio.sleep(0.4) + await self._sync_clock() + await asyncio.sleep(0.3) + await self._sync_clock() + self._last_sync = time.time() + + async def _main(self) -> None: + backoff = 5 + while not self._stop: + if not self.settings.get("ipixel_enabled"): + await asyncio.sleep(3); continue + if not self.connected: + try: + await self._connect(); backoff = 5 + except Exception as err: + self.error = str(err)[:160] + await asyncio.sleep(backoff); backoff = min(backoff * 2, 120); continue + try: + with self._lock: + want = self._want + notify = time.time() < self._notify_until + if want and (notify or self._state_key) and want[0] != self._shown_key: + await self._send(want[1], want[2]); self._shown_key = want[0] + elif not want or (not notify and not self._state_key): + if self._shown_key == "clock" and self._intermezzo_due(): + await self._intermezzo(); continue + if self._shown_key == "clock" and time.time() - getattr(self, "_last_sync", 0) > 600: + await self._sync_clock(); self._last_sync = time.time() + if self._shown_key != "clock": + await self._show_clock(); self._shown_key = "clock" + with self._lock: + self._want = None + except Exception as err: + self.error = str(err)[:160]; self.connected = False + try: + await self._client.disconnect() + except Exception: + pass + continue + self._wake.clear() + try: + timeout = max(0.3, self._notify_until - time.time()) if time.time() < self._notify_until else 30 + await asyncio.wait_for(self._wake.wait(), timeout=timeout) + except asyncio.TimeoutError: + pass + + def _intermezzo_due(self) -> bool: + every = int(self.settings.get("ipixel_intermezzi_min") or 0) + if every <= 0: + return False + h = time.localtime().tm_hour + if h >= 23 or h < 7: # di notte niente sorprese + return False + last = getattr(self, "_last_fun", 0.0) or time.time() - every * 60 + 90 # il primo dopo un minuto e mezzo + if not getattr(self, "_last_fun", 0.0): + self._last_fun = last + return time.time() - last > every * 60 * random.uniform(0.8, 1.2) + + async def _intermezzo(self) -> None: + from avatar import ipixel_fun as fun + self._last_fun = time.time() + kind, val = await asyncio.get_event_loop().run_in_executor(None, fun.scegli, self.size) + if kind == "gif": + data, gif = val, True + else: + col = random.choice([(0, 200, 255), (255, 120, 0), (255, 60, 170), (120, 255, 60), (255, 220, 0), (170, 90, 255)]) + data, gif = render_text(str(val), col, self.size) + with self._lock: + self._want = (f"fun:{time.time()}", data, gif) + self._notify_until = time.time() + 12 + self._shown_key = None + + def _poke(self) -> None: + if self._loop and self._wake: + self._loop.call_soon_threadsafe(self._wake.set) + + # ── API (da qualunque thread) ────────────────────────────────────────── + def show_text(self, text: str, color=None, durata: float = 10.0, key: str | None = None, icon=None) -> None: + data, gif = render_text(text, color or COLORI["INFO"], self.size, icon_color=icon) + with self._lock: + self._want = (key or f"n:{text}:{time.time()}", data, gif) + self._notify_until = time.time() + durata + self._poke() + + def show_image(self, path: str, durata: float = 15.0) -> None: + data = render_image(path, self.size) + with self._lock: + self._want = (f"img:{path}:{time.time()}", data, False); self._notify_until = time.time() + durata + self._poke() + + def on_state(self, state: str) -> None: + """Stato dell'assistente: mostrato mentre lavora, orologio a riposo.""" + if not self.settings.get("ipixel_stati", True): + return + base, _, detail = state.partition(" · ") + base = base.split(" ")[0] + if base in ("LISTENING", "SLEEPING", "MUTED", "INITIALISING", ""): + with self._lock: + self._state_key = None + self._poke(); return + if time.time() < self._notify_until: + return + m = None + import re + if "token" in detail or "parole" in detail: + m = re.search(r"(\d+)\s+(token|parole)(?:\s*·\s*(\d+)/s)?", detail) + if m: + text = f"{m.group(1)} tok" + (f" {m.group(3)}/s" if m.group(3) else "") + elif base == "SPEAKING": + text = "parlo" + elif detail: + text = detail.replace("strumento: ", "").replace("genero la risposta", "scrivo")[:40] + else: + text = {"THINKING": "penso…", "PROCESSING": "lavoro…"}.get(base, base.lower()) + key = f"s:{base}:{text}" + if key == self._state_key: + return + data, gif = render_text(text, COLORI.get(base, COLORI["INFO"]), self.size, icon_color=COLORI.get(base)) + with self._lock: + self._state_key = key + self._want = (key, data, gif) + self._poke() + + def run_cmd(self, frame: list[int]) -> None: + if self._loop and self.connected: + asyncio.run_coroutine_threadsafe(self._cmd(frame), self._loop).result(timeout=10) + else: + raise RuntimeError(self.error or "pannello non collegato") + + def clock(self) -> None: + with self._lock: + self._want, self._state_key, self._notify_until = None, None, 0 + self._shown_key = None + self._poke() + + +panel: IPixel | None = None diff --git a/avatar/ipixel_fun.py b/avatar/ipixel_fun.py new file mode 100644 index 0000000..d52ee42 --- /dev/null +++ b/avatar/ipixel_fun.py @@ -0,0 +1,147 @@ +"""Intermezzi per il pannello LED: frasi spiritose, animazioni procedurali (GIF) e piccole informazioni.""" +from __future__ import annotations + +import io +import math +import random +import time + +from PIL import Image, ImageDraw + +FRASI = [ + "Ciao Luciano!", "Respira. Bevi acqua.", "Il caffè è pronto?", "Sto pensando... scherzo", + "LuZa ti osserva (con affetto)", "Oggi spacchi!", "Pausa? Te la meriti", "Error 418: sono una teiera", + "Beat drop in 3, 2, 1...", "Ctrl+Z sulla giornata?", "Ricordati di sorridere", "Più bpm, meno stress", + "Ho bevuto 0 caffè. Invidia", "I server stanno bene, credo", "Se mi spegni piango in binario", + "01001100 01110101 01011010 01100001", "Modalità DJ: attiva", "Valentina ti pensa (probabile)", + "Chiedimi qualcosa!", "Batteria sociale: 87%", "Che si mangia stasera?", "Alzati e sgranchisciti!", +] + + +def _gif(frames: list[Image.Image], ms: int = 60) -> bytes: + buf = io.BytesIO() + frames[0].save(buf, "GIF", save_all=True, append_images=frames[1:], duration=ms, loop=0, disposal=1, optimize=False) + return buf.getvalue() + + +def plasma(size, n=14) -> bytes: + w, h = size; out = [] + for f in range(n): + im = Image.new("RGB", size); px = im.load(); t = f / n * 2 * math.pi + for y in range(h): + for x in range(w): + v = math.sin(x / 6 + t) + math.sin(y / 3 + t * 1.3) + math.sin((x + y) / 8 + t * 0.7) + px[x, y] = (int(128 + 127 * math.sin(v)), int(128 + 127 * math.sin(v + 2.1)), int(128 + 127 * math.sin(v + 4.2))) + out.append(im) + return _gif(out, 70) + + +def matrix(size, n=36) -> bytes: + w, h = size; drops = [random.randint(-h, h) for _ in range(w)]; trail = [[0] * h for _ in range(w)]; out = [] + for _ in range(n): + im = Image.new("RGB", size); px = im.load() + for x in range(w): + for y in range(h): + trail[x][y] = max(0, trail[x][y] - 40) + if random.random() < 0.6: + drops[x] += 1 + if drops[x] >= h + random.randint(0, 8): + drops[x] = random.randint(-6, 0) + if 0 <= drops[x] < h: + trail[x][drops[x]] = 255 + for y in range(h): + v = trail[x][y] + if v: + px[x, y] = (v // 4, v, v // 3) if v < 255 else (200, 255, 200) + out.append(im) + return _gif(out, 60) + + +def stelle(size, n=36) -> bytes: + w, h = size; st = [[random.uniform(-w / 2, w / 2), random.uniform(-h / 2, h / 2), random.uniform(1, 8)] for _ in range(40)]; out = [] + for _ in range(n): + im = Image.new("RGB", size); px = im.load() + for s in st: + s[2] -= 0.25 + if s[2] <= 0.3: + s[:] = [random.uniform(-w / 2, w / 2), random.uniform(-h / 2, h / 2), 8] + x, y = int(w / 2 + s[0] / s[2] * 2), int(h / 2 + s[1] / s[2] * 2) + if 0 <= x < w and 0 <= y < h: + b = int(255 * (1 - s[2] / 8)); px[x, y] = (b, b, min(255, b + 40)) + out.append(im) + return _gif(out, 50) + + +CUORE = ["0110110", "1111111", "1111111", "0111110", "0011100", "0001000"] + + +def cuore(size, n=24) -> bytes: + w, h = size; out = [] + for f in range(n): + im = Image.new("RGB", size); d = ImageDraw.Draw(im) + s = 2 if f % 8 < 4 else 1 + cx = int((w - 7 * s) * (0.5 + 0.45 * math.sin(f / n * 2 * math.pi))) + cy = (h - 6 * s) // 2 + for r, row in enumerate(CUORE): + for c, ch in enumerate(row): + if ch == "1": + d.rectangle([cx + c * s, cy + r * s, cx + c * s + s - 1, cy + r * s + s - 1], fill=(255, 30, 80)) + out.append(im) + return _gif(out, 90) + + +def equalizzatore(size, n=30) -> bytes: + w, h = size; bars = w // 4; lv = [random.uniform(2, h) for _ in range(bars)]; out = [] + for _ in range(n): + im = Image.new("RGB", size); d = ImageDraw.Draw(im) + for i in range(bars): + lv[i] = max(1, min(h, lv[i] + random.uniform(-4, 4))) + for y in range(int(lv[i])): + c = (0, 255, 80) if y < h * 0.5 else (255, 200, 0) if y < h * 0.8 else (255, 40, 40) + d.rectangle([i * 4, h - 1 - y, i * 4 + 2, h - 1 - y], fill=c) + out.append(im) + return _gif(out, 70) + + +def pacman(size, n=40) -> bytes: + w, h = size; out = [] + for f in range(n): + im = Image.new("RGB", size); d = ImageDraw.Draw(im) + x = int(-12 + (w + 24) * f / n); cy = h // 2 + for dx in range(x + 10, w, 8): + d.rectangle([dx, cy - 1, dx + 1, cy], fill=(255, 200, 150)) + a = 35 if f % 4 < 2 else 5 + d.pieslice([x - 6, cy - 6, x + 6, cy + 6], a, 360 - a, fill=(255, 220, 0)) + gx = x - 20 + d.rounded_rectangle([gx - 5, cy - 6, gx + 5, cy + 6], 4, fill=(255, 60, 60)) + d.point([(gx - 2, cy - 2), (gx + 2, cy - 2)], fill=(255, 255, 255)) + out.append(im) + return _gif(out, 60) + + +ANIMAZIONI = {"plasma": plasma, "matrix": matrix, "stelle": stelle, "cuore": cuore, "equalizzatore": equalizzatore, "pacman": pacman} + + +def info_breve() -> str: + t = time.localtime() + giorni = ["lunedì", "martedì", "mercoledì", "giovedì", "venerdì", "sabato", "domenica"] + mesi = ["gen", "feb", "mar", "apr", "mag", "giu", "lug", "ago", "set", "ott", "nov", "dic"] + base = f"{giorni[t.tm_wday]} {t.tm_mday} {mesi[t.tm_mon - 1]}" + try: + import json, urllib.request + u = "https://api.open-meteo.com/v1/forecast?latitude=42.35&longitude=14.40¤t=temperature_2m,weather_code&timezone=Europe/Rome" + d = json.load(urllib.request.urlopen(u, timeout=5))["current"] + return f"{base} · Ortona {round(d['temperature_2m'])}°" + except Exception: + return base + + +def scegli(size) -> tuple[str, object]: + """('testo', frase) oppure ('gif', dati).""" + r = random.random() + if r < 0.45: + return "testo", random.choice(FRASI) + if r < 0.6: + return "info", info_breve() + nome = random.choice(list(ANIMAZIONI)) + return "gif", ANIMAZIONI[nome](size) diff --git a/avatar/monitor.py b/avatar/monitor.py index 0e40854..1844611 100644 --- a/avatar/monitor.py +++ b/avatar/monitor.py @@ -24,6 +24,17 @@ KEYWORDS = re.compile(r"\?|urgent|aiut|assistenz|problem|non funziona|non riesco _lock = threading.Lock() +def _panel_notify(text: str, alert: bool = False) -> None: + """Notifica sul pannello LED iPIXEL, se attivo e abilitato.""" + try: + from avatar import ipixel + p = ipixel.panel + if p is not None and p.settings.get("ipixel_enabled") and p.settings.get("ipixel_notifiche", True): + p.show_text(text, ipixel.COLORI["ALERT"] if alert else ipixel.COLORI["INFO"], 14 if alert else 10, icon=ipixel.COLORI["ALERT"] if alert else None) + except Exception as err: + print(f"[ipixel] {err}") + + class Monitor: def __init__(self, ui, say, settings: Settings) -> None: self.ui, self.say, self.settings = ui, say, settings @@ -252,6 +263,7 @@ class Monitor: def _notify(self, a: dict) -> None: riga = f"{a['fonte']} · {a['chi']}: {a['motivo']}" self.ui.write_log(f"ERR: ATTENZIONE — {riga}") + _panel_notify(f"{a['fonte']} {a['chi']}: {a['motivo']}", alert=True) try: title = "Ava: richiede attenzione" if a["priorita"] != "alta" else "Ava: URGENTE" safe = lambda s: s.replace('"', "'").replace("\\", "") diff --git a/avatar/settings.py b/avatar/settings.py index c108dc7..dff09e8 100644 --- a/avatar/settings.py +++ b/avatar/settings.py @@ -57,6 +57,13 @@ DEFAULTS: dict[str, Any] = { "abitudini_enabled": True, # distillazione settimanale delle abitudini nella memoria "umore_enabled": False, # stima oraria dell'umore dai messaggi scritti dall'utente "io_nomi": "Luciano", # come compare l'utente come mittente nelle chat esportate + "ipixel_enabled": False, # pannello LED iPIXEL via Bluetooth + "ipixel_address": "", # identificativo BLE (rilevato da solo) + "ipixel_luminosita": 40, + "ipixel_stati": True, # mostra lo stato di LuZa (penso, token, strumenti) mentre lavora + "ipixel_notifiche": True, # mostra avvisi e messaggi in arrivo + "ipixel_orologio_stile": 1, + "ipixel_intermezzi_min": 15, # ogni quanti minuti una frase o animazione (0 = mai); non di notte "remote_enabled": True, # app web per il telefono (HTTPS sulla rete locale) "remote_token": "", # chiave di abbinamento (generata al primo avvio) } diff --git a/avatar/settings_dialog.py b/avatar/settings_dialog.py index 1bcc623..88383dd 100644 --- a/avatar/settings_dialog.py +++ b/avatar/settings_dialog.py @@ -8,7 +8,7 @@ from pathlib import Path from PyQt6.QtCore import Qt, pyqtSignal from PyQt6.QtWidgets import (QApplication, QCheckBox, QComboBox, QDialog, QFormLayout, QFrame, QHBoxLayout, QLabel, QLineEdit, - QPushButton, QScrollArea, QStackedWidget, QTabWidget, QVBoxLayout, QWidget, QDoubleSpinBox) + QPushButton, QScrollArea, QStackedWidget, QTabWidget, QVBoxLayout, QWidget, QDoubleSpinBox, QSpinBox) from .engines.claude_code import check_claude_code from .engines.openai_compat import list_models @@ -274,6 +274,26 @@ class SettingsDialog(QDialog): self.ha_hint = QLabel(""); self.ha_hint.setWordWrap(True); form_ha.addRow("", self.ha_hint) add(pg_casa, form_ha) + # ── Pannello LED iPIXEL ─────────────────────────────────────────── + form_px = QFormLayout() + form_px.addRow(_note("Pannello LED iPIXEL via Bluetooth: a riposo mostra l'orologio, mentre LuZa lavora lo stato (penso, token al secondo, strumento in uso), e gli avvisi in arrivo.")) + row = QHBoxLayout() + self.px_on = QCheckBox("Attivo"); self.px_on.setChecked(bool(s.get("ipixel_enabled"))); row.addWidget(self.px_on) + self.px_stati = QCheckBox("Stato di LuZa"); self.px_stati.setChecked(bool(s.get("ipixel_stati", True))); row.addWidget(self.px_stati) + self.px_notif = QCheckBox("Notifiche"); self.px_notif.setChecked(bool(s.get("ipixel_notifiche", True))); row.addWidget(self.px_notif) + row.addStretch(); form_px.addRow("Pannello", row) + self.px_lum = QSpinBox(); self.px_lum.setRange(1, 100); self.px_lum.setValue(int(s.get("ipixel_luminosita") or 40)) + form_px.addRow("Luminosità", self.px_lum) + self.px_fun = _combo([(0, "mai"), (5, "ogni 5 minuti"), (15, "ogni 15 minuti"), (30, "ogni 30 minuti"), (60, "ogni ora")], int(s.get("ipixel_intermezzi_min") if s.get("ipixel_intermezzi_min") is not None else 15)) + form_px.addRow("Frasi e animazioni", self.px_fun) + try: + from . import ipixel as _px + st = "non attivo" if _px.panel is None else (f"collegato ({_px.panel.size[0]}×{_px.panel.size[1]})" if _px.panel.connected else (_px.panel.error or "in collegamento…")) + except Exception: + st = "?" + form_px.addRow("Stato", QLabel(st)) + add(pg_casa, form_px) + # ── Strumenti (server MCP esterni) ──────────────────────────────── from PyQt6.QtWidgets import QPlainTextEdit from .mcp_client import CONFIG_FILE as MCP_FILE, manager as mcp_manager @@ -594,6 +614,8 @@ class SettingsDialog(QDialog): "server_assist_dir": self.srv_dir.text().strip() or "/opt/aiserverassistance", "server_assist_bot": self.srv_bot.text().strip().lstrip("@") or "luzaserver_bot", "homeassistant_url": self.ha_url.text().strip().rstrip("/"), + "ipixel_enabled": self.px_on.isChecked(), "ipixel_stati": self.px_stati.isChecked(), "ipixel_notifiche": self.px_notif.isChecked(), + "ipixel_luminosita": int(self.px_lum.value()), "ipixel_intermezzi_min": int(self.px_fun.currentData() or 0), "immagini_famiglia": self.img_fam.currentData() or "z-image-turbo", "immagini_modello": self.img_model.currentText().strip(), } if self.ha_token.text().strip(): diff --git a/main.py b/main.py index 65664c5..2127a73 100644 --- a/main.py +++ b/main.py @@ -104,8 +104,17 @@ def main() -> None: assistant.remote = remote _set_state, _write_log, _show_confirm, _hide_confirm = ui.set_state, ui.write_log, ui.show_confirm, ui.hide_confirm + from avatar import ipixel + ipixel.panel = ipixel.IPixel(settings, ui.write_log) + ipixel.panel.start() + ui._app.aboutToQuit.connect(ipixel.panel.stop) + def set_state(state: str) -> None: _set_state(state); remote.on_state(state) + try: + ipixel.panel.on_state(state) + except Exception: + pass def write_log(text: str) -> None: _write_log(text); remote.on_log(text) diff --git a/plugins/ipixel.py b/plugins/ipixel.py new file mode 100644 index 0000000..d2b04e1 --- /dev/null +++ b/plugins/ipixel.py @@ -0,0 +1,96 @@ +"""Pannello LED iPIXEL: scritte, immagini, luminosità, orologio, accensione. Lo stato di LuZa e le notifiche +compaiono da soli se l'opzione è attiva nelle impostazioni (scheda Casa).""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from avatar import ipixel # noqa: E402 + + +def _p(): + if ipixel.panel is None: + raise RuntimeError("pannello iPIXEL non attivo: abilitalo nelle impostazioni (scheda Casa)") + return ipixel.panel + + +def scrivi(params: dict, ctx: dict) -> str: + testo = str(params.get("testo") or "").strip() + if not testo: + return "Errore: serve il testo." + col = ipixel.NOMI_COLORI.get(str(params.get("colore") or "").lower(), None) + durata = float(params.get("secondi") or 30) + _p().show_text(testo, col, durata) + return f"Sul pannello: «{testo}» per {int(durata)} secondi." + + +def immagine(params: dict, ctx: dict) -> str: + path = str(params.get("percorso") or "").strip() + if not Path(path).expanduser().is_file(): + return "Errore: file immagine non trovato." + _p().show_image(str(Path(path).expanduser()), float(params.get("secondi") or 30)) + return "Immagine inviata al pannello." + + +def luminosita(params: dict, ctx: dict) -> str: + v = max(1, min(int(params.get("livello") or 40), 100)) + _p().run_cmd([5, 0, 4, 0x80, v]) + _p().settings.set("ipixel_luminosita", v); _p().settings.save() + return f"Luminosità del pannello al {v}%." + + +def accendi(params: dict, ctx: dict) -> str: + on = str(params.get("acceso", True)).lower() not in ("false", "0", "no") + _p().run_cmd([5, 0, 7, 1, 1 if on else 0]) + return "Pannello acceso." if on else "Pannello spento." + + +def orologio(params: dict, ctx: dict) -> str: + _p().clock() + return "Il pannello mostra l'orologio." + + +def stato(params: dict, ctx: dict) -> str: + p = ipixel.panel + if p is None: + return "Pannello iPIXEL disattivato nelle impostazioni." + return (f"Pannello {'collegato' if p.connected else 'non collegato'} ({p.size[0]}×{p.size[1]})" + (f", errore: {p.error}" if p.error and not p.connected else "") + ".") + + +def animazione(params: dict, ctx: dict) -> str: + from avatar import ipixel_fun as fun + import time + nome = str(params.get("nome") or "").lower() + p = _p() + if nome in fun.ANIMAZIONI: + data = fun.ANIMAZIONI[nome](p.size) + with p._lock: + p._want = (f"fun:{time.time()}", data, True); p._notify_until = time.time() + float(params.get("secondi") or 15) + p._shown_key = None; p._poke() + return f"Animazione «{nome}» sul pannello." + kind, val = fun.scegli(p.size) + if kind == "gif": + with p._lock: + p._want = (f"fun:{time.time()}", val, True); p._notify_until = time.time() + 15 + p._shown_key = None; p._poke(); return "Sorpresa animata sul pannello." + p.show_text(str(val), None, 12) + return f"Sul pannello: «{val}»." + + +TOOLS = [ + {"name": "ipixel_animazione", "description": "Mostra un'animazione sul pannello LED: plasma, matrix, stelle, cuore, equalizzatore, pacman; senza nome una sorpresa a caso (frase spiritosa, info o animazione).", + "parameters": {"type": "object", "properties": {"nome": {"type": "string"}, "secondi": {"type": "number"}}}, "run": animazione}, + {"name": "ipixel_scrivi", "description": "Scrive un testo sul pannello LED iPIXEL (scorre se è lungo) per alcuni secondi, poi torna all'orologio. colore: rosso, verde, blu, azzurro, giallo, arancione, viola, rosa, bianco, ciano.", + "parameters": {"type": "object", "properties": {"testo": {"type": "string"}, "colore": {"type": "string"}, "secondi": {"type": "number"}}, "required": ["testo"]}, "run": scrivi}, + {"name": "ipixel_immagine", "description": "Mostra un'immagine (percorso di un file) ridimensionata sul pannello LED.", + "parameters": {"type": "object", "properties": {"percorso": {"type": "string"}, "secondi": {"type": "number"}}, "required": ["percorso"]}, "run": immagine}, + {"name": "ipixel_luminosita", "description": "Imposta la luminosità del pannello LED (1-100).", + "parameters": {"type": "object", "properties": {"livello": {"type": "integer"}}, "required": ["livello"]}, "run": luminosita}, + {"name": "ipixel_accendi", "description": "Accende o spegne il pannello LED (acceso=true/false).", + "parameters": {"type": "object", "properties": {"acceso": {"type": "boolean"}}}, "run": accendi}, + {"name": "ipixel_orologio", "description": "Riporta il pannello LED all'orologio.", + "parameters": {"type": "object", "properties": {}}, "run": orologio}, + {"name": "ipixel_stato", "description": "Dice se il pannello LED è collegato e la sua risoluzione.", + "parameters": {"type": "object", "properties": {}}, "run": stato}, +] diff --git a/scripts/crea_app.py b/scripts/crea_app.py index 8ef4160..fa0a090 100644 --- a/scripts/crea_app.py +++ b/scripts/crea_app.py @@ -140,6 +140,7 @@ exec "{python}" main.py >> "$HOME/Library/Logs/{APP_NAME}.log" 2>&1 "NSContactsUsageDescription": "Per trovare numeri ed email dei contatti.", "NSAppleEventsUsageDescription": "Per controllare Mail, Note, Musica, Messaggi e le altre app.", "NSSpeechRecognitionUsageDescription": "Per il riconoscimento vocale.", + "NSBluetoothAlwaysUsageDescription": "Per comandare il pannello LED iPIXEL.", } with open(app / "Contents" / "Info.plist", "wb") as f: plistlib.dump(info, f)