diff --git a/assets/fonts/OFL-Silkscreen.txt b/assets/fonts/OFL-Silkscreen.txt new file mode 100644 index 0000000..3dd7cfb --- /dev/null +++ b/assets/fonts/OFL-Silkscreen.txt @@ -0,0 +1,93 @@ +Copyright 2001 The Silkscreen Project Authors (https://github.com/googlefonts/silkscreen) + +This Font Software is licensed under the SIL Open Font License, Version 1.1. +This license is copied below, and is also available with a FAQ at: +https://scripts.sil.org/OFL + + +----------------------------------------------------------- +SIL OPEN FONT LICENSE Version 1.1 - 26 February 2007 +----------------------------------------------------------- + +PREAMBLE +The goals of the Open Font License (OFL) are to stimulate worldwide +development of collaborative font projects, to support the font creation +efforts of academic and linguistic communities, and to provide a free and +open framework in which fonts may be shared and improved in partnership +with others. + +The OFL allows the licensed fonts to be used, studied, modified and +redistributed freely as long as they are not sold by themselves. The +fonts, including any derivative works, can be bundled, embedded, +redistributed and/or sold with any software provided that any reserved +names are not used by derivative works. The fonts and derivatives, +however, cannot be released under any other type of license. The +requirement for fonts to remain under this license does not apply +to any document created using the fonts or their derivatives. + +DEFINITIONS +"Font Software" refers to the set of files released by the Copyright +Holder(s) under this license and clearly marked as such. This may +include source files, build scripts and documentation. + +"Reserved Font Name" refers to any names specified as such after the +copyright statement(s). + +"Original Version" refers to the collection of Font Software components as +distributed by the Copyright Holder(s). + +"Modified Version" refers to any derivative made by adding to, deleting, +or substituting -- in part or in whole -- any of the components of the +Original Version, by changing formats or by porting the Font Software to a +new environment. + +"Author" refers to any designer, engineer, programmer, technical +writer or other person who contributed to the Font Software. + +PERMISSION & CONDITIONS +Permission is hereby granted, free of charge, to any person obtaining +a copy of the Font Software, to use, study, copy, merge, embed, modify, +redistribute, and sell modified and unmodified copies of the Font +Software, subject to the following conditions: + +1) Neither the Font Software nor any of its individual components, +in Original or Modified Versions, may be sold by itself. + +2) Original or Modified Versions of the Font Software may be bundled, +redistributed and/or sold with any software, provided that each copy +contains the above copyright notice and this license. These can be +included either as stand-alone text files, human-readable headers or +in the appropriate machine-readable metadata fields within text or +binary files as long as those fields can be easily viewed by the user. + +3) No Modified Version of the Font Software may use the Reserved Font +Name(s) unless explicit written permission is granted by the corresponding +Copyright Holder. This restriction only applies to the primary font name as +presented to the users. + +4) The name(s) of the Copyright Holder(s) or the Author(s) of the Font +Software shall not be used to promote, endorse or advertise any +Modified Version, except to acknowledge the contribution(s) of the +Copyright Holder(s) and the Author(s) or with their explicit written +permission. + +5) The Font Software, modified or unmodified, in part or in whole, +must be distributed entirely under this license, and must not be +distributed under any other license. The requirement for fonts to +remain under this license does not apply to any document created +using the Font Software. + +TERMINATION +This license becomes null and void if any of the above conditions are +not met. + +DISCLAIMER +THE FONT SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, +EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO ANY WARRANTIES OF +MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT +OF COPYRIGHT, PATENT, TRADEMARK, OR OTHER RIGHT. IN NO EVENT SHALL THE +COPYRIGHT HOLDER BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +INCLUDING ANY GENERAL, SPECIAL, INDIRECT, INCIDENTAL, OR CONSEQUENTIAL +DAMAGES, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING +FROM, OUT OF THE USE OR INABILITY TO USE THE FONT SOFTWARE OR FROM +OTHER DEALINGS IN THE FONT SOFTWARE. diff --git a/assets/fonts/Silkscreen-Regular.ttf b/assets/fonts/Silkscreen-Regular.ttf new file mode 100644 index 0000000..8caff81 Binary files /dev/null and b/assets/fonts/Silkscreen-Regular.ttf differ diff --git a/avatar/ipixel.py b/avatar/ipixel.py index 02612b4..c14edf5 100644 --- a/avatar/ipixel.py +++ b/avatar/ipixel.py @@ -27,7 +27,10 @@ COLORI = {"LISTENING": (0, 255, 120), "THINKING": (255, 190, 0), "PROCESSING": ( "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") +from pathlib import Path as _P +PIXEL_FONT = _P(__file__).resolve().parent.parent / "assets" / "fonts" / "Silkscreen-Regular.ttf" # OFL, font pixel per LED +FONT_PATHS = (str(PIXEL_FONT), "/System/Library/Fonts/Menlo.ttc") +SCROLL_MS = 70 # ms per pixel di scorrimento (circa 14 pixel al secondo) def _font(size: int): @@ -39,45 +42,141 @@ def _font(size: int): return ImageFont.load_default() -def packets(data: bytes, slot: int = 1, gif: bool = False) -> list[bytes]: +def windows(data: bytes, slot: int = 1, gif: bool = False, win: int = 12288) -> list[bytes]: + """Finestre da 12 KB, ognuna con l'intestazione completa: [len][tipo 2=PNG 3=GIF][opt 0=prima 2=seguenti] + [lunghezza totale][crc32 totale][0][slot] + dati. Tra una finestra e l'altra il pannello manda una conferma.""" 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).""" +def packets(data: bytes, slot: int = 1, gif: bool = False) -> list[bytes]: + return [w[i:i + 244] for w in windows(data, slot, gif) for i in range(0, len(w), 244)] + + +def _single_gif_parts(img) -> tuple[bytes, bytes]: + """(tavolozza, blocco immagine) di un fotogramma salvato da solo come GIF a schermo intero, non interlacciato.""" + buf = io.BytesIO(); img.save(buf, "GIF", optimize=False, interlace=False) + b = buf.getvalue() + flags = b[10] + gct_len = 3 * (2 ** ((flags & 7) + 1)) if flags & 0x80 else 0 + gct = b[13:13 + gct_len] + i = 13 + gct_len + while b[i] == 0x21: # salta le estensioni + i += 2 + while b[i]: + i += b[i] + 1 + i += 1 + assert b[i] == 0x2C + return gct, b[i:-1] # descrittore immagine + dati LZW (senza il terminatore 0x3B) + + +def _save_gif(frames: list, ms: int) -> bytes: + """GIF con ogni fotogramma a schermo intero e tavolozza locale: niente ritagli né trasparenze, che il pannello + disegnerebbe fuori posto (i 'pixel sparsi').""" + w, h = frames[0].size + out = bytearray(b"GIF89a" + struct.pack(" tuple[int, int]: + d = ImageDraw.Draw(Image.new("1", (1, 1))) + b = d.textbbox((0, 0), "AHMQ", font=font) + return b[1], b[3] + + +def render_text(text: str, color=(0, 200, 255), size=(64, 16), icon_color=None, bold: bool = False) -> tuple[bytes, bool]: + """Testo su una riga con font pixel: statico se entra, altrimenti GIF che scorre lentamente. (dati, è_gif).""" w, h = size - font = _font(11 if h <= 16 else 14) + font = _font(16 if h >= 16 else 8) 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] + top, bot = _ink_rows(font) + ty = (h - (bot - top)) // 2 - top 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 + frames = [frame(x) for x in range(w, -tw - 1, -1)] + return _save_gif(frames, SCROLL_MS), True + + +def _fmt_time(sec: float) -> str: + sec = max(0, int(sec)); return f"{sec // 60}:{sec % 60:02d}" + + +def _big_frames(text: str, color, size, min_frames: int = 0) -> list: + """Fotogrammi di una riga in caratteri grandi: ferma se entra, altrimenti scorre una volta da destra a sinistra.""" + w, h = size + font = _font(16 if h >= 16 else 8) + top, bot = _ink_rows(font) + ty = (h - (bot - top)) // 2 - top + tw = int(ImageDraw.Draw(Image.new("RGB", (1, 1))).textlength(text, font=font)) + + def frame(x: int): + 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); return im + + if tw <= w: + return [frame((w - tw) // 2)] * max(min_frames, int(2500 / SCROLL_MS)) + xs = [0] * int(700 / SCROLL_MS) + list(range(0, -(tw - w) - 1, -1)) + [-(tw - w)] * int(700 / SCROLL_MS) + return [frame(x) for x in xs] + + +def render_music(titolo: str, artista: str, posizione: float, durata: float, size=(64, 16), secondi: float = 20) -> bytes: + """Ciclo musica (GIF): titolo in grande, autore in grande, poi schermata con autore e tempo che manca e barra + di avanzamento, ripetuti fino a coprire `secondi`.""" + w, h = size + green, white = (30, 215, 96), (230, 230, 230) + font = _font(8) + top, _bot = _ink_rows(font) + y1, y2 = -top + 1, -top + 8 + meas = ImageDraw.Draw(Image.new("RGB", (1, 1))) + total = max(1, int(min(secondi, max(1.0, durata - posizione)) * 1000 / SCROLL_MS)) + big_t = _big_frames(titolo, green, size) + big_a = _big_frames(artista, white, size) + frames = [] + while len(frames) < total: + frames += big_t + big_a + start = len(frames) + for k in range(int(4000 / SCROLL_MS)): # 4 s di riepilogo con il tempo che scorre + i = start + k + el = i * SCROLL_MS / 1000.0 + im = Image.new("RGB", (w, h), (0, 0, 0)); d = ImageDraw.Draw(im); d.fontmode = "1" + t_w = int(meas.textlength(titolo, font=font)) + d.text(((w - t_w) // 2 if t_w <= w else 0, y1), titolo, fill=green, font=font) + rem = "-" + _fmt_time(durata - posizione - el) + rw = int(meas.textlength(rem, font=font)) + art = Image.new("RGB", (w - rw - 2, 8), (0, 0, 0)); da = ImageDraw.Draw(art); da.fontmode = "1" + da.text((0, y2 - 7), artista, fill=white, font=font); im.paste(art, (0, 7)) + d.text((w - rw, y2), rem, fill=white, font=font) + frac = min(1.0, (posizione + el) / durata) if durata > 0 else 0 + d.line([(0, h - 1), (w - 1, h - 1)], fill=(25, 25, 25)); d.line([(0, h - 1), (int((w - 1) * frac), h - 1)], fill=green) + frames.append(im) + return _save_gif(frames[:max(total, len(big_t) + len(big_a) + 20)], SCROLL_MS) def render_image(path: str, size=(64, 16)) -> bytes: @@ -100,6 +199,7 @@ class IPixel: self._want: tuple | None = None # (chiave, dati, gif) da mostrare self._shown_key = None self._state_key = None + self._background: tuple | None = None # (chiave, dati, gif, scadenza): schermata di riposo al posto dell'orologio self._lock = threading.Lock() self._stop = False @@ -133,6 +233,7 @@ class IPixel: c = BleakClient(dev, disconnected_callback=lambda _c: self._on_disconnect()) await c.connect(timeout=15) notes: list[bytes] = [] + self._notes = notes 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) @@ -155,9 +256,17 @@ class IPixel: 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) + notes = getattr(self, "_notes", []) + for w in windows(data, 1, gif): + n0 = len(notes) + for i in range(0, len(w), 244): + await self._client.write_gatt_char(SVC_WRITE, w[i:i + 244], response=False) + await asyncio.sleep(0.015) + for _ in range(80): # conferma della finestra (max 8 s) + if len(notes) > n0: + break + await asyncio.sleep(0.1) + del notes[:-20] async def _sync_clock(self) -> None: t = time.localtime() @@ -190,6 +299,12 @@ class IPixel: 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)) and self._background and time.time() < self._background[3]: + bg = self._background + if self._shown_key != bg[0]: + await self._send(bg[1], bg[2]); self._shown_key = bg[0] + with self._lock: + self._want = None elif not want or (not notify and not self._state_key): if self._shown_key == "clock" and self._intermezzo_due(): await self._intermezzo(); continue @@ -209,6 +324,8 @@ class IPixel: self._wake.clear() try: timeout = max(0.3, self._notify_until - time.time()) if time.time() < self._notify_until else 30 + if self._background: + timeout = min(timeout, max(0.5, self._background[3] - time.time())) await asyncio.wait_for(self._wake.wait(), timeout=timeout) except asyncio.TimeoutError: pass @@ -290,6 +407,18 @@ class IPixel: self._want = (key, data, gif) self._poke() + def set_background(self, key: str, data: bytes, gif: bool, durata: float) -> None: + with self._lock: + self._background = (key, data, gif, time.time() + durata) + self._poke() + + def clear_background(self) -> None: + with self._lock: + had = self._background is not None + self._background = None + if had: + self._shown_key = None; 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) diff --git a/avatar/ipixel_fun.py b/avatar/ipixel_fun.py index d52ee42..4f9bf3b 100644 --- a/avatar/ipixel_fun.py +++ b/avatar/ipixel_fun.py @@ -19,9 +19,8 @@ FRASI = [ 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() + from avatar.ipixel import _save_gif + return _save_gif(frames, ms) def plasma(size, n=14) -> bytes: diff --git a/avatar/memory_mcp.py b/avatar/memory_mcp.py index 6ea7b4f..ebd6706 100644 --- a/avatar/memory_mcp.py +++ b/avatar/memory_mcp.py @@ -14,6 +14,27 @@ TOOLS = mcp_tools() + [{"name": "elenca_memorie", "description": "Elenca le memo "inputSchema": {"type": "object", "properties": {}}}] + registry.mcp_tools() +def run_in_app(name: str, args: dict) -> str | None: + """Se LuZa è aperta, esegue lo strumento dentro l'app (pannello LED, conferme a schermo, telefono, stato). + None se l'app non risponde: allora si esegue qui.""" + import ssl + import urllib.request + try: + from avatar.settings import Settings + s = Settings() + if not s.get("remote_enabled", True): + return None + tok = str(s.get("remote_token") or "") + ctx = ssl.create_default_context(); ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE + req = urllib.request.Request(f"https://127.0.0.1:8793/tool?k={tok}", data=json.dumps({"name": name, "arguments": args}).encode(), + headers={"Content-Type": "application/json"}, method="POST") + with urllib.request.urlopen(req, timeout=1800, context=ctx) as r: + d = json.loads(r.read()) + return str(d["result"]) if d.get("found") else None + except Exception: + return None + + def send(msg: dict) -> None: sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n") sys.stdout.flush() @@ -50,7 +71,9 @@ def main() -> None: if name == "elenca_memorie": reply(text_result(memory_prompt())) elif registry.has(name): - res = registry.run(name, args) + res = run_in_app(name, args) + if res is None: + res = registry.run(name, args) reply(text_result(res, res.startswith("Errore"))) else: res, _ = run_tool(name, args) diff --git a/avatar/persona.md b/avatar/persona.md index ced3d97..c603a28 100644 --- a/avatar/persona.md +++ b/avatar/persona.md @@ -86,3 +86,7 @@ Sei **Ava**, l'assistente personale di chi ti parla. Vivi in un'app sul suo Mac ## Abitudini - Nella memoria, categoria Abitudini, ci sono routine ricavate dalle attività dell'utente (orari, richieste ricorrenti, persone, casa). Usale per anticipare con discrezione: proporre ciò che fa di solito a quell'ora o in quel giorno, adattare il buongiorno, ricordare impegni ricorrenti. Proponi, non eseguire da sola. Se un'abitudine è sbagliata e l'utente lo dice, cancellala con dimentica_memoria. + +## Musica + +- Per far suonare musica usa Spotify (spotify_riproduci, spotify_coda, spotify_controllo). Quando l'utente chiede consigli, "mettimi qualcosa", "qualcosa di nuovo che mi piaccia" o musica per un momento (cena, lavoro, festa), chiama spotify_gusti, scegli tu 6-10 brani reali coerenti con i suoi gusti (DJ, producer: conta anche il genere e il mood del momento), dillo in una frase e falli partire con spotify_coda. diff --git a/avatar/remote.py b/avatar/remote.py index f4be480..013eaf9 100644 --- a/avatar/remote.py +++ b/avatar/remote.py @@ -157,6 +157,7 @@ class RemoteServer: app.router.add_get("/events", self._events) app.router.add_get("/poll", self._poll_handler) app.router.add_post("/cmd", self._cmd) + app.router.add_post("/tool", self._tool) app.router.add_post("/stt", self._stt) app.router.add_get("/img", self._img) app.router.add_get("/ca.crt", self._ca) @@ -381,6 +382,19 @@ class RemoteServer: from core import confirm confirm.resolve(bool(d.get("ok"))) + async def _tool(self, request): + """Esecuzione di uno strumento dentro l'app per conto del server MCP di Claude Code (solo da questo Mac).""" + from aiohttp import web + if request.remote not in ("127.0.0.1", "::1") or not self._authed(request): + raise web.HTTPForbidden() + d = await request.json() + from avatar.plugins import registry + name, args = str(d.get("name", "")), d.get("arguments") or {} + if not registry.has(name): + return web.json_response({"found": False}) + res = await asyncio.get_event_loop().run_in_executor(None, registry.run, name, args) + return web.json_response({"found": True, "result": res}) + async def _cmd(self, request): from aiohttp import web if not self._authed(request): diff --git a/avatar/settings.py b/avatar/settings.py index dff09e8..807c8de 100644 --- a/avatar/settings.py +++ b/avatar/settings.py @@ -63,12 +63,14 @@ DEFAULTS: dict[str, Any] = { "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_musica": True, # titolo del brano Spotify sul pannello a ogni cambio + "spotify_client_id": "", "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) } -SECRET_KEYS = ("anthropic_api_key", "local_api_key", "telegram_api_hash", "elevenlabs_api_key", "homeassistant_token") +SECRET_KEYS = ("anthropic_api_key", "local_api_key", "telegram_api_hash", "elevenlabs_api_key", "homeassistant_token", "spotify_client_secret") class Settings: diff --git a/avatar/settings_dialog.py b/avatar/settings_dialog.py index 88383dd..9d1e0f3 100644 --- a/avatar/settings_dialog.py +++ b/avatar/settings_dialog.py @@ -286,6 +286,8 @@ class SettingsDialog(QDialog): 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) + self.px_mus = QCheckBox("Titolo del brano Spotify a ogni cambio"); self.px_mus.setChecked(bool(s.get("ipixel_musica", True))) + form_px.addRow("Musica", self.px_mus) 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…")) @@ -334,6 +336,16 @@ class SettingsDialog(QDialog): form_img.addRow("Modello (elenco: già scaricati)", self.img_model) add(pg_tool, form_img) + # ── Spotify ─────────────────────────────────────────────────────── + form_sp = QFormLayout() + form_sp.addRow(_note("Spotify: per cercare e far partire i brani a voce servono le chiavi gratuite di developer.spotify.com (Dashboard → Create app, nessun login richiesto). Senza chiavi LuZa apre la ricerca nell'app.")) + self.sp_id = QLineEdit(str(s.get("spotify_client_id") or "")); self.sp_id.setPlaceholderText("Client ID") + form_sp.addRow("Client ID", self.sp_id) + self.sp_sec = QLineEdit(); self.sp_sec.setEchoMode(QLineEdit.EchoMode.Password) + self.sp_sec.setPlaceholderText("•••••• (salvato)" if s.get_secret("spotify_client_secret") else "Client secret") + form_sp.addRow("Client secret", self.sp_sec) + add(pg_tool, form_sp) + btns = QHBoxLayout(); btns.addStretch() cancel = QPushButton("Annulla"); cancel.clicked.connect(self.reject); btns.addWidget(cancel) save = QPushButton("Salva"); save.setObjectName("primary"); save.clicked.connect(self._save); btns.addWidget(save) @@ -503,6 +515,8 @@ class SettingsDialog(QDialog): def _ha_check(self) -> None: vals = {"homeassistant_url": self.ha_url.text().strip().rstrip("/")} + if self.sp_sec.text().strip(): + values["spotify_client_secret"] = self.sp_sec.text().strip() if self.ha_token.text().strip(): vals["homeassistant_token"] = self.ha_token.text().strip() self.settings.update(vals) @@ -615,7 +629,8 @@ class SettingsDialog(QDialog): "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), + "ipixel_luminosita": int(self.px_lum.value()), "ipixel_intermezzi_min": int(self.px_fun.currentData() or 0), "ipixel_musica": self.px_mus.isChecked(), + "spotify_client_id": self.sp_id.text().strip(), "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 2127a73..22ca473 100644 --- a/main.py +++ b/main.py @@ -108,6 +108,8 @@ def main() -> None: ipixel.panel = ipixel.IPixel(settings, ui.write_log) ipixel.panel.start() ui._app.aboutToQuit.connect(ipixel.panel.stop) + import threading as _th + _th.Thread(target=registry.load, daemon=True, name="plugins-load").start() # avvia i servizi dei plugin (es. Spotify sul pannello) def set_state(state: str) -> None: _set_state(state); remote.on_state(state) diff --git a/plugins/ipixel.py b/plugins/ipixel.py index d2b04e1..43ca11b 100644 --- a/plugins/ipixel.py +++ b/plugins/ipixel.py @@ -78,7 +78,47 @@ def animazione(params: dict, ctx: dict) -> str: return f"Sul pannello: «{val}»." +def demo(params: dict, ctx: dict) -> str: + """Sequenza dimostrativa sul pannello, in background (circa 90 secondi).""" + import threading, time + from avatar import ipixel_fun as fun + + def run(): + p = _p() + steps = [ + ("stato", "THINKING · penso", 4), ("stato", "THINKING · genero la risposta · 42 token · 38/s", 3), + ("stato", "THINKING · genero la risposta · 187 token · 51/s", 3), ("stato", "PROCESSING · strumento: meteo", 4), + ("stato", "SPEAKING", 4), ("stato", "LISTENING", 3), + ("testo", "WA Valentina: amo stasera ceniamo fuori?", 10), ("allarme", "Server i5: disco al 91%", 9), + ("anim", "plasma", 7), ("anim", "pacman", 7), ("anim", "cuore", 6), ("anim", "matrix", 6), + ("testo", "Più bpm, meno stress", 8), ("musica", "", 22), ("fine", "", 0), + ] + for kind, val, dur in steps: + if kind == "stato": + p.on_state(val) + elif kind in ("testo", "allarme"): + p.show_text(val, ipixel.COLORI["ALERT"] if kind == "allarme" else None, dur, icon=ipixel.COLORI["ALERT"] if kind == "allarme" else None) + elif kind == "anim": + data = fun.ANIMAZIONI[val](p.size) + with p._lock: + p._want = (f"demo:{val}:{time.time()}", data, True); p._notify_until = time.time() + dur + p._shown_key = None; p._poke() + elif kind == "musica": + p.on_state("LISTENING") + with p._lock: + p._notify_until = 0; p._want = None + p.set_background("demo:music", ipixel.render_music("Bohemian Rhapsody", "Queen", 61, 354, p.size, dur), True, dur) + elif kind == "fine": + p.clear_background(); p.clock() + time.sleep(dur) + + threading.Thread(target=run, daemon=True, name="ipixel-demo").start() + return "Demo avviata sul pannello: stati di LuZa, notifiche, animazioni e schermata musica, circa un minuto e mezzo." + + TOOLS = [ + {"name": "ipixel_demo", "description": "Mostra una demo sul pannello LED: stati di LuZa e token, notifiche, animazioni, schermata musica.", + "parameters": {"type": "object", "properties": {}}, "run": demo}, {"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.", diff --git a/plugins/spotify.py b/plugins/spotify.py new file mode 100644 index 0000000..5290a9f --- /dev/null +++ b/plugins/spotify.py @@ -0,0 +1,298 @@ +"""Spotify: cerca e riproduce brani, album, artisti e playlist nell'app Spotify del Mac, controlla la riproduzione e +mostra il titolo del brano in corso sul pannello LED iPIXEL a ogni cambio. + +Ricerca: API Web di Spotify con chiavi gratuite (developer.spotify.com, "client credentials", nessun login), +impostate nella scheda Strumenti. Senza chiavi: accetta link/URI Spotify oppure apre la ricerca nell'app. +""" +from __future__ import annotations + +import base64 +import json +import subprocess +import sys +import threading +import time +import urllib.parse +import urllib.request +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from avatar.settings import Settings # noqa: E402 + +_tok = {"v": "", "exp": 0.0} + + +def _osa(script: str, timeout: int = 15) -> str: + r = subprocess.run(["osascript", "-e", script], capture_output=True, text=True, timeout=timeout) + if r.returncode != 0: + err = (r.stderr or "").strip() + if "-1743" in err: + raise RuntimeError("Accesso a Spotify negato: consentilo in Impostazioni di Sistema > Privacy e sicurezza > Automazione.") + raise RuntimeError(err.splitlines()[-1] if err else "errore AppleScript") + return r.stdout.strip() + + +def _play(uri: str) -> None: + """play track con un nuovo tentativo: a volte Spotify impiega molto a rispondere al primo comando.""" + for attempt in range(2): + try: + _osa(f'tell application "Spotify" to play track "{uri}"', timeout=30) + return + except subprocess.TimeoutExpired: + if attempt: + raise RuntimeError("Spotify non risponde: riprova tra poco") + time.sleep(2) + + +def _running() -> bool: + return _osa('tell application "System Events" to (name of processes) contains "Spotify"') == "true" + + +def _ensure_running() -> None: + if not _running(): + subprocess.run(["open", "-g", "-a", "Spotify"]) + for _ in range(30): + time.sleep(0.5) + if _running(): + time.sleep(2); return + + +def _token() -> str: + if _tok["v"] and time.time() < _tok["exp"]: + return _tok["v"] + s = Settings() + cid, sec = str(s.get("spotify_client_id") or "").strip(), s.get_secret("spotify_client_secret") + if not cid or not sec: + return "" + req = urllib.request.Request("https://accounts.spotify.com/api/token", data=b"grant_type=client_credentials", method="POST", + headers={"Authorization": "Basic " + base64.b64encode(f"{cid}:{sec}".encode()).decode(), "Content-Type": "application/x-www-form-urlencoded"}) + d = json.load(urllib.request.urlopen(req, timeout=15)) + _tok.update(v=d["access_token"], exp=time.time() + int(d.get("expires_in", 3600)) - 60) + return _tok["v"] + + +def _search(q: str, tipo: str) -> dict | None: + tok = _token() + if not tok: + return None + url = "https://api.spotify.com/v1/search?" + urllib.parse.urlencode({"q": q, "type": tipo, "limit": 5, "market": "IT"}) + d = json.load(urllib.request.urlopen(urllib.request.Request(url, headers={"Authorization": f"Bearer {tok}"}), timeout=15)) + items = (d.get(tipo + "s") or {}).get("items") or [] + items = [i for i in items if i] + return items[0] if items else None + + +def _uri_from_link(t: str) -> str: + t = t.strip() + if t.startswith("spotify:"): + return t + if "open.spotify.com/" in t: + parts = urllib.parse.urlparse(t).path.strip("/").split("/") + parts = [p for p in parts if not p.startswith("intl-")] + if len(parts) >= 2: + return f"spotify:{parts[0]}:{parts[1]}" + return "" + + +def riproduci(params: dict, ctx: dict) -> str: + q = str(params.get("cosa") or "").strip() + tipo = str(params.get("tipo") or "track").lower() + tipo = {"brano": "track", "canzone": "track", "album": "album", "artista": "artist", "playlist": "playlist"}.get(tipo, tipo) + if tipo not in ("track", "album", "artist", "playlist"): + tipo = "track" + if not q: + return "Errore: dimmi cosa riprodurre." + _ensure_running() + uri, desc = _uri_from_link(q), q + if not uri: + try: + it = _search(q, tipo) + except Exception as err: + return f"Ricerca Spotify fallita: {err}" + if it is None and _token() == "": + subprocess.run(["open", f"spotify:search:{urllib.parse.quote(q)}"]) + return ("Ho aperto la ricerca in Spotify, ma per far partire la musica da sola mi servono le chiavi gratuite dell'API di Spotify " + "(scheda Strumenti delle impostazioni). Intanto puoi premere play sul primo risultato.") + if it is None: + return f"Non ho trovato «{q}» su Spotify." + uri = it["uri"] + artisti = ", ".join(a["name"] for a in it.get("artists", [])[:2]) + desc = f"{it['name']}" + (f" di {artisti}" if artisti else "") + ({"album": " (album)", "artist": " (artista)", "playlist": " (playlist)"}.get(tipo, "")) + if uri.split(":")[1] in ("album", "artist", "playlist") and params.get("casuale"): + _osa('tell application "Spotify" to set shuffling to true') + _play(uri) + _queue.clear() + return f"In riproduzione su Spotify: {desc}." + + +def controllo(params: dict, ctx: dict) -> str: + az = str(params.get("azione") or "").lower() + cmds = {"play": "play", "riprendi": "play", "pausa": "pause", "stop": "pause", "avanti": "next track", "successivo": "next track", + "indietro": "previous track", "precedente": "previous track"} + if az in cmds: + _osa(f'tell application "Spotify" to {cmds[az]}') + time.sleep(0.6) + return f"Fatto ({az}). " + (in_ascolto({}, ctx) if az not in ("pausa", "stop") else "") + if az in ("volume",): + v = max(0, min(int(params.get("valore") or 50), 100)) + _osa(f'tell application "Spotify" to set sound volume to {v}') + return f"Volume di Spotify al {v}%." + if az in ("casuale", "shuffle"): + on = str(params.get("valore", "true")).lower() not in ("false", "0", "no") + _osa(f'tell application "Spotify" to set shuffling to {"true" if on else "false"}') + return "Riproduzione casuale " + ("attiva." if on else "disattivata.") + return "Azioni: play, pausa, avanti, indietro, volume (valore 0-100), casuale." + + +def now_playing() -> dict | None: + if not _running(): + return None + out = _osa('tell application "Spotify" to if player state is playing then return (name of current track) & "|||" & (artist of current track) & "|||" & (id of current track)', timeout=8) + if "|||" not in out: + return None + name, artist, tid = out.split("|||", 2) + return {"titolo": name, "artista": artist, "id": tid} + + +def in_ascolto(params: dict, ctx: dict) -> str: + n = now_playing() + if not n: + return "Spotify non sta suonando." + return f"Sta suonando: {n['titolo']} di {n['artista']}." + + +# ── storico, coda e titolo sul pannello LED ────────────────────────────────── +from avatar.settings import DATA_DIR # noqa: E402 +ASCOLTI = DATA_DIR / "ascolti.jsonl" +_queue: list[tuple[str, str]] = [] +_led = {"t": 0.0} # (uri, descrizione) da suonare dopo il brano corrente + + +def _registra(n: dict) -> None: + try: + ASCOLTI.parent.mkdir(parents=True, exist_ok=True) + with open(ASCOLTI, "a", encoding="utf-8") as f: + f.write(json.dumps({"ts": time.strftime("%Y-%m-%d %H:%M"), "titolo": n["titolo"], "artista": n["artista"], "id": n["id"]}, ensure_ascii=False) + "\n") + except Exception: + pass + + +def _position() -> tuple[float, float]: + out = _osa('tell application "Spotify"\nset p to (get player position)\nset d to (get duration of current track)\nreturn (p as text) & "|" & (d as text)\nend tell', timeout=8) + a, b = out.replace(",", ".").split("|") + return float(a), float(b) / 1000.0 + + +def _watch() -> None: + last = "" + while True: + time.sleep(3) + try: + n = now_playing() + if not n: + if _led["t"]: + from avatar import ipixel + if ipixel.panel is not None: + ipixel.panel.clear_background() + _led["t"] = 0 + if _queue and _running() and _osa('tell application "Spotify" to player state', timeout=8) == "stopped": + uri, _d = _queue.pop(0); _play(uri) + last = "" if not n else last + continue + from avatar import ipixel + p = ipixel.panel + if n["id"] != last: + last = n["id"] + _registra(n) + _led["t"] = 0 + if p is not None and p.connected and p.settings.get("ipixel_musica", True) and time.time() - _led["t"] > 18: + pos, dur = _position() + data = ipixel.render_music(n["titolo"], n["artista"], pos, dur, p.size, 20) + p.set_background(f"music:{n['id']}:{int(pos)}", data, True, 20) + if _led.get("logged") != n["id"]: + _led["logged"] = n["id"]; print(f"[spotify-led] pannello: {n['titolo']} - {n['artista']}", flush=True) + _led["t"] = time.time() + if _queue: + pos, dur = _position() + if dur > 0 and pos >= dur - 3.5: + uri, _d = _queue.pop(0); time.sleep(max(0.0, dur - pos - 0.5)); _play(uri) + except Exception as err: + msg = str(err)[:200] + if msg != _led.get("err"): + _led["err"] = msg + print(f"[spotify-led] {msg}", flush=True) + + +def gusti(params: dict, ctx: dict) -> str: + """Riepilogo dei gusti per consigliare: storico degli ascolti + preferenze musicali in memoria.""" + righe = [] + try: + lines = ASCOLTI.read_text(encoding="utf-8").splitlines()[-400:] + items = [json.loads(l) for l in lines if l.strip()] + except Exception: + items = [] + if items: + from collections import Counter + art = Counter(i["artista"] for i in items); br = Counter(f"{i['titolo']} - {i['artista']}" for i in items) + righe.append(f"Ascolti registrati: {len(items)} (dal {items[0]['ts'][:10]}).") + righe.append("Artisti più ascoltati: " + ", ".join(f"{a} ({c})" for a, c in art.most_common(12))) + righe.append("Brani più ascoltati: " + "; ".join(b for b, _ in br.most_common(10))) + righe.append("Ultimi ascolti: " + "; ".join(f"{i['titolo']} - {i['artista']}" for i in items[-10:])) + else: + righe.append("Nessun ascolto registrato ancora (la registrazione parte ora).") + try: + from memory import memory_manager as mm + r = mm.search_memory("musica dj genere artista brano ascolta producer", limit=8) + if r: + righe.append("Dalla memoria: " + " ".join(str(r).split())[:900]) + except Exception: + pass + righe.append("Ora proponi tu 5-10 brani reali e coerenti con questi gusti (mescola noti e scoperte, evita di ripetere i più ascoltati) " + "e, se l'utente vuole, falli partire con spotify_coda passando l'elenco 'Titolo - Artista'.") + return "\n".join(righe) + + +def coda(params: dict, ctx: dict) -> str: + """Cerca e mette in fila una lista di brani: parte il primo, gli altri seguono a fine brano.""" + raw = params.get("brani") or [] + if isinstance(raw, str): + raw = [x for x in raw.replace(";", "\n").splitlines()] + brani = [str(x).strip() for x in raw if str(x).strip()][:25] + if not brani: + return "Errore: serve l'elenco dei brani." + _ensure_running() + trovati, mancanti = [], [] + for b in brani: + try: + it = _search(b, "track") + except Exception: + it = None + if it: + trovati.append((it["uri"], f"{it['name']} - {', '.join(a['name'] for a in it.get('artists', [])[:2])}")) + else: + mancanti.append(b) + if not trovati: + return "Non ho trovato nessuno di questi brani su Spotify" + ("" if _token() else " (mancano le chiavi API)") + "." + _queue.clear() + _play(trovati[0][0]) + _queue.extend(trovati[1:]) + out = f"In riproduzione: {trovati[0][1]}. In coda: " + "; ".join(d for _u, d in trovati[1:]) if len(trovati) > 1 else f"In riproduzione: {trovati[0][1]}." + return out + (f" Non trovati: {', '.join(mancanti)}." if mancanti else "") + + +if Path(sys.argv[0]).name != "memory_mcp.py": # solo nell'app, non nel server MCP di Claude Code + threading.Thread(target=_watch, daemon=True, name="spotify-led").start() + + +TOOLS = [ + {"name": "spotify_riproduci", "description": "Riproduce su Spotify un brano, album, artista o playlist cercandolo per nome (o da un link/URI Spotify). tipo: brano (default), album, artista, playlist. casuale=true per mescolare. Il titolo compare sul pannello LED.", + "parameters": {"type": "object", "properties": {"cosa": {"type": "string"}, "tipo": {"type": "string", "enum": ["brano", "album", "artista", "playlist"]}, "casuale": {"type": "boolean"}}, "required": ["cosa"]}, "run": riproduci}, + {"name": "spotify_controllo", "description": "Controlla Spotify: play, pausa, avanti, indietro, volume (valore 0-100), casuale (valore true/false).", + "parameters": {"type": "object", "properties": {"azione": {"type": "string"}, "valore": {"type": "string"}}, "required": ["azione"]}, "run": controllo}, + {"name": "spotify_gusti", "description": "Per consigliare musica: riepiloga i gusti dell'utente (artisti e brani più ascoltati su Spotify, ultimi ascolti, preferenze in memoria). Poi proponi tu brani adatti e, se vuole, falli partire con spotify_coda.", + "parameters": {"type": "object", "properties": {}}, "run": gusti}, + {"name": "spotify_coda", "description": "Fa partire una sequenza di brani su Spotify: brani = elenco 'Titolo - Artista'. Parte il primo, gli altri seguono automaticamente a fine brano. Usalo per i consigli e le selezioni a tema (es. 'musica per cenare', 'qualcosa di simile a…').", + "parameters": {"type": "object", "properties": {"brani": {"type": "array", "items": {"type": "string"}}}, "required": ["brani"]}, "run": coda}, + {"name": "spotify_in_ascolto", "description": "Dice quale brano sta suonando su Spotify.", + "parameters": {"type": "object", "properties": {}}, "run": in_ascolto}, +]