From 094807ffe85dea91e2ffa26517e50224057ef3c0 Mon Sep 17 00:00:00 2001 From: luciano Date: Tue, 29 Sep 2026 17:51:43 +0200 Subject: [PATCH] Abitudini: diario delle richieste e distillazione settimanale in memoria (richieste, calendario, messaggi, casa) Co-Authored-By: Claude Fable 5.1 --- avatar/assistant.py | 18 +++ avatar/monitor.py | 10 ++ avatar/persona.md | 4 + avatar/settings.py | 1 + avatar/settings_dialog.py | 4 +- memory/memory_manager.py | 6 +- plugins/abitudini.py | 230 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 270 insertions(+), 3 deletions(-) create mode 100644 plugins/abitudini.py diff --git a/avatar/assistant.py b/avatar/assistant.py index d421de6..5383db8 100644 --- a/avatar/assistant.py +++ b/avatar/assistant.py @@ -44,6 +44,18 @@ def _identity() -> tuple[str, str]: return "Ava", "" +def _diario(testo: str, strumenti: list[str], origine: str) -> None: + """Riga nel diario delle richieste (data/diario.jsonl): base per l'apprendimento delle abitudini.""" + try: + import json as _j + from avatar.settings import DATA_DIR as _D + _D.mkdir(parents=True, exist_ok=True) + with open(_D / "diario.jsonl", "a", encoding="utf-8") as f: + f.write(_j.dumps({"ts": time.strftime("%Y-%m-%d %H:%M"), "testo": testo[:300], "strumenti": strumenti[:8], "origine": origine}, ensure_ascii=False) + "\n") + except Exception: + pass + + class Assistant: def __init__(self, ui, settings: Settings) -> None: self.ui = ui @@ -335,10 +347,16 @@ class Assistant: self.ui.set_state("THINKING") splitter = SentenceSplitter() + used: list[str] = [] + def emit(ev: dict) -> None: if turn != self._turn: return t = ev.get("type") + if t == "status" and ev.get("status") == "working" and ev.get("detail"): + used.append(str(ev["detail"])) + if t in ("done", "error"): + _diario(text, used, getattr(self, "_origin", "local")) if t == "status": label = STATUS_LABELS.get(ev["status"], "THINKING") detail = str(ev.get("detail") or "").strip() diff --git a/avatar/monitor.py b/avatar/monitor.py index 6022b7d..0e40854 100644 --- a/avatar/monitor.py +++ b/avatar/monitor.py @@ -73,6 +73,16 @@ class Monitor: self.ui.write_log(f"SYS: Umore stimato: {r['tono']} ({r.get('nota', '')[:80]})") except Exception as err: print(f"[umore] {err}") + if self.settings.get("abitudini_enabled", True): + try: + import sys as _s; _s.path.insert(0, str(Path(__file__).resolve().parent.parent / "plugins")) + import importlib + ab = importlib.import_module("abitudini") + if ab.da_aggiornare(): + r = ab.aggiorna() + self.ui.write_log(f"SYS: Abitudini aggiornate: {len(r)} voci.") + except Exception as err: + print(f"[abitudini] {err}") self._stop.wait(int(self.settings.get("monitor_intervallo") or 60)) # ── raccolta ───────────────────────────────────────────────────────── diff --git a/avatar/persona.md b/avatar/persona.md index 2b5c19d..ced3d97 100644 --- a/avatar/persona.md +++ b/avatar/persona.md @@ -82,3 +82,7 @@ Sei **Ava**, l'assistente personale di chi ti parla. Vivi in un'app sul suo Mac - Se l'utente ti chiede di rispondere a qualcuno al posto suo ("rispondi a Valentina come me", "digli che arrivo tardi, come lo direi io"), usa come_me_contesto, scrivi il messaggio in prima persona imitando il suo stile (parole, lunghezza, emoji, tono con quella persona) e leggiglielo prima di inviarlo con lo strumento di invio, che chiede conferma. Mai inviare senza conferma. - L'umore stimato dai suoi messaggi serve solo a calibrare il tuo tono: non commentarlo se non te lo chiede, e se noti un momento pesante chiedi come sta, con tatto. + +## 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. diff --git a/avatar/settings.py b/avatar/settings.py index 5da9c9c..c108dc7 100644 --- a/avatar/settings.py +++ b/avatar/settings.py @@ -54,6 +54,7 @@ DEFAULTS: dict[str, Any] = { "homeassistant_url": "http://homeassistant.local:8123", # Home Assistant (token nel portachiavi) "immagini_famiglia": "z-image-turbo", # z-image-turbo | schnell | dev | qwen (comando mflux e passi predefiniti) "immagini_modello": "", # repo Hugging Face o cartella; vuoto = predefinito della famiglia + "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 "remote_enabled": True, # app web per il telefono (HTTPS sulla rete locale) diff --git a/avatar/settings_dialog.py b/avatar/settings_dialog.py index 18a8312..1bcc623 100644 --- a/avatar/settings_dialog.py +++ b/avatar/settings_dialog.py @@ -255,6 +255,8 @@ class SettingsDialog(QDialog): form5.addRow("Escludi dagli avvisi", self.mon_excl) self.umore_on = QCheckBox("Stima il mio umore ogni ora dai messaggi che scrivo (WhatsApp, Telegram) e adatta il tono; dati solo sul Mac") self.umore_on.setChecked(bool(s.get("umore_enabled"))); form5.addRow("Umore", self.umore_on) + self.abit_on = QCheckBox("Impara le mie abitudini ogni settimana (richieste, calendario, messaggi, casa) e salvale in memoria") + self.abit_on.setChecked(bool(s.get("abitudini_enabled", True))); form5.addRow("Abitudini", self.abit_on) self.io_nomi = QLineEdit(str(s.get("io_nomi") or "Luciano")); self.io_nomi.setPlaceholderText("nomi con cui compaio nelle chat esportate, separati da virgola") form5.addRow("Il mio nome nelle chat", self.io_nomi) add(pg_mon, form5) @@ -587,7 +589,7 @@ class SettingsDialog(QDialog): "monitor_intervallo": int(self.mon_int.currentData() or 60), "monitor_regole": self.mon_rules.text().strip(), "monitor_escludi": self.mon_excl.text().strip(), - "umore_enabled": self.umore_on.isChecked(), "io_nomi": self.io_nomi.text().strip() or "Luciano", + "umore_enabled": self.umore_on.isChecked(), "abitudini_enabled": self.abit_on.isChecked(), "io_nomi": self.io_nomi.text().strip() or "Luciano", "server_assist_ssh": self.srv_ssh.text().strip(), "server_assist_dir": self.srv_dir.text().strip() or "/opt/aiserverassistance", "server_assist_bot": self.srv_bot.text().strip().lstrip("@") or "luzaserver_bot", diff --git a/memory/memory_manager.py b/memory/memory_manager.py index 5c78632..e8063fa 100644 --- a/memory/memory_manager.py +++ b/memory/memory_manager.py @@ -52,6 +52,7 @@ def _empty_memory() -> dict: "relationships": {}, "wishes": {}, "notes": {}, + "habits": {}, } def load_memory() -> dict: @@ -133,7 +134,7 @@ def save_memory(memory: dict) -> None: EXPORT_DIR = get_base_dir() / "data" / "memoria" _EXPORT_LABELS = {"identity": "Identità", "preferences": "Preferenze", "projects": "Progetti", "relationships": "Persone", - "wishes": "Desideri e obiettivi", "notes": "Note"} + "wishes": "Desideri e obiettivi", "notes": "Note", "habits": "Abitudini"} def export_markdown(memory: dict, folder: Path | None = None) -> Path: @@ -219,6 +220,7 @@ _CATEGORY_LABELS = { "relationships": "People in their life", "wishes": "Wishes / plans", "notes": "Notes", + "habits": "Habits / routines (learned from activity)", } _IDENTITY_FIELDS = ["name", "age", "birthday", "city", "job", @@ -436,7 +438,7 @@ def all_entries_for_ui() -> list[dict]: return rows def remember(key: str, value: str, category: str = "notes") -> str: - valid = {"identity", "preferences", "projects", "relationships", "wishes", "notes"} + valid = {"identity", "preferences", "projects", "relationships", "wishes", "notes", "habits"} if category not in valid: category = "notes" update_memory({category: {key: {"value": value}}}) diff --git a/plugins/abitudini.py b/plugins/abitudini.py new file mode 100644 index 0000000..0b17fd0 --- /dev/null +++ b/plugins/abitudini.py @@ -0,0 +1,230 @@ +"""Abitudini: ogni settimana distilla routine e ricorrenze dalle attività (richieste a LuZa, calendario, messaggi, +dispositivi di casa) in poche frasi salvate nella memoria (categoria Abitudini). Dati solo locali. +""" +from __future__ import annotations + +import json +import re +import sys +import time +from collections import Counter, defaultdict +from datetime import datetime, timedelta +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from avatar.settings import DATA_DIR # noqa: E402 +from avatar import quick_llm # noqa: E402 +from memory import memory_manager as mm # noqa: E402 + +STATO = DATA_DIR / "abitudini.json" +DIARIO = DATA_DIR / "diario.jsonl" +GIORNI = ["lunedì", "martedì", "mercoledì", "giovedì", "venerdì", "sabato", "domenica"] +HOME = Path.home() + + +def _fascia(h: int) -> str: + return "mattina presto" if h < 8 else "mattina" if h < 13 else "pomeriggio" if h < 18 else "sera" if h < 23 else "notte" + + +def _stato() -> dict: + try: + return json.loads(STATO.read_text(encoding="utf-8")) + except Exception: + return {} + + +def da_aggiornare() -> bool: + return time.time() - float(_stato().get("ultimo", 0)) > 7 * 86400 + + +# ── sorgenti ───────────────────────────────────────────────────────────────── +def _richieste(giorni: int) -> list[dict]: + """Richieste a LuZa: diario + sessioni Claude Code di LuZa (cartella home) come storico iniziale.""" + since = datetime.now() - timedelta(days=giorni) + out = [] + if DIARIO.exists(): + for line in DIARIO.read_text(encoding="utf-8").splitlines(): + try: + d = json.loads(line); ts = datetime.strptime(d["ts"], "%Y-%m-%d %H:%M") + except Exception: + continue + if ts >= since: + out.append({"ts": ts, "testo": d.get("testo", ""), "strumenti": d.get("strumenti", [])}) + try: + import claude_sessioni as cs + for root in cs._roots(): + pdir = root / "projects" / ("-" + str(HOME).strip("/").replace("/", "-")) + if not pdir.is_dir(): + continue + for f in pdir.glob("*.jsonl"): + if datetime.fromtimestamp(f.stat().st_mtime) < since: + continue + try: # solo le sessioni di LuZa (usano gli strumenti mcp__avatar), non quelle del Terminale nella home + if b"mcp__avatar" not in f.read_bytes(): + continue + except Exception: + continue + for role, text, ts in cs._messages(f): + if role != "user" or not ts or "= since and len(text) < 400: + out.append({"ts": t, "testo": text, "strumenti": []}) + except Exception: + pass + return out + + +def _calendario(settimane: int) -> list[dict]: + try: + import calendario as cal + end = datetime.now(); start = end - timedelta(weeks=settimane) + return [{"titolo": str(e.title()), "start": cal._pydate(e.startDate())} for e in cal._events_between(start, end)] + except Exception: + return [] + + +def _messaggi(giorni: int) -> list[dict]: + try: + import come_me + since = time.time() - giorni * 86400 + rows = come_me._wa_mine("", since=since, limit=2000) + come_me._tg_mine(since) + out = [] + for r in rows: + try: + out.append({"ts": datetime.strptime(r["ts"][:16], "%Y-%m-%d %H:%M"), "chat": r["chat"], "canale": r["canale"]}) + except Exception: + pass + return out + except Exception: + return [] + + +def _casa(giorni: int) -> list[dict]: + try: + import home_assistant as ha + start = (datetime.now() - timedelta(days=giorni)).astimezone().isoformat() + rows = ha.api(f"/api/logbook/{start}") or [] + out = [] + for r in rows: + eid = r.get("entity_id", "") + if eid.split(".")[0] not in ("light", "switch", "media_player", "climate", "cover", "scene", "script") or not r.get("context_user_id"): + continue + try: + out.append({"ts": datetime.fromisoformat(r["when"]).astimezone().replace(tzinfo=None), "nome": r.get("name") or eid, "stato": r.get("state") or r.get("message", "")}) + except Exception: + pass + return out + except Exception: + return [] + + +# ── statistiche compatte ───────────────────────────────────────────────────── +def riassunto_dati(giorni: int = 60) -> str: + parti = [] + req = _richieste(giorni) + if req: + ore = Counter(_fascia(r["ts"].hour) for r in req); gg = Counter(GIORNI[r["ts"].weekday()] for r in req) + strum = Counter(s for r in req for s in r["strumenti"]) + temi = Counter() + for r in req: + for w in re.findall(r"[a-zàèéìòù]{5,}", r["testo"].lower()): + temi[w] += 1 + stop = {"della", "delle", "degli", "questo", "questa", "quello", "quella", "anche", "sempre", "fammi", "dimmi", "puoi", "vorrei", "grazie", "cosa", "come", "quando", "perché"} + temi = [(w, c) for w, c in temi.most_common(40) if w not in stop][:15] + parti.append(f"RICHIESTE A LUZA negli ultimi {giorni} giorni: {len(req)} (prime: {req[0]['ts']:%d/%m} … ultime: {req[-1]['ts']:%d/%m}).\n" + f"Per fascia oraria: {dict(ore.most_common())}. Per giorno: {dict(gg.most_common())}.\n" + f"Strumenti più usati: {dict(strum.most_common(10))}. Parole più frequenti nelle richieste: {', '.join(f'{w} ({c})' for w, c in temi)}.\n" + "Esempi di richieste tipiche: " + " | ".join(r["testo"][:70] for r in req[-40::4])) + ev = _calendario(8) + if ev: + per_t = defaultdict(list) + for e in ev: + per_t[re.sub(r"\d+", "#", e["titolo"].strip().lower())[:40]].append(e["start"]) + ric = [(t, ds) for t, ds in per_t.items() if len(ds) >= 3] + ric.sort(key=lambda x: -len(x[1])) + righe = [f"«{t}» {len(ds)} volte, di solito {Counter(GIORNI[d.weekday()] for d in ds).most_common(1)[0][0]} alle {Counter(d.hour for d in ds).most_common(1)[0][0]:02d}" for t, ds in ric[:12]] + parti.append(f"CALENDARIO (8 settimane, {len(ev)} eventi). Ricorrenti: " + ("; ".join(righe) if righe else "nessuno con 3+ occorrenze") + ". Altri titoli recenti: " + ", ".join(sorted({e['titolo'][:30] for e in ev[-15:]}))) + msg = _messaggi(min(giorni, 45)) + if msg: + ore = Counter(_fascia(m["ts"].hour) for m in msg); chats = Counter(m["chat"] for m in msg) + primo = Counter(m["ts"].hour for m in msg if m["ts"].hour < 12).most_common(1) + ultimo = Counter(m["ts"].hour for m in msg if m["ts"].hour >= 18).most_common(1) + parti.append(f"MESSAGGI SCRITTI DALL'UTENTE (WhatsApp/Telegram, {len(msg)}): per fascia {dict(ore.most_common())}; persone più sentite: {', '.join(f'{c} ({n})' for c, n in chats.most_common(8))}; " + f"prima ora del mattino più frequente: {primo[0][0] if primo else '?'}; ora serale più frequente: {ultimo[0][0] if ultimo else '?'}; per giorno: {dict(Counter(GIORNI[m['ts'].weekday()] for m in msg).most_common())}.") + casa = _casa(14) + if casa: + per_d = defaultdict(list) + for c in casa: + per_d[c["nome"]].append(c["ts"].hour) + righe = [f"{n}: {len(h)} azioni, ore tipiche {sorted(Counter(h).most_common(3))}" for n, h in sorted(per_d.items(), key=lambda kv: -len(kv[1]))[:10]] + parti.append("CASA (14 giorni, azioni manuali su dispositivi): " + "; ".join(righe)) + return "\n\n".join(parti) if parti else "" + + +def aggiorna(giorni: int = 60) -> list[str]: + dati = riassunto_dati(giorni) + if not dati: + return [] + prompt = (f"Dati sulle attività di Luciano:\n\n{dati}\n\n" + "Ricava le sue ABITUDINI e routine ricorrenti (orari tipici della giornata, cosa chiede di solito e quando, giorni particolari, persone sentite regolarmente, " + "impegni ricorrenti, dispositivi di casa usati a certe ore). Solo ciò che è davvero ricorrente e utile a un assistente per anticipare; niente ovvietà né dati sensibili. " + "Rispondi SOLO con un JSON: una lista di oggetti {\"chiave\": breve_identificatore_snake_case, \"abitudine\": frase in italiano di massimo 25 parole}. Da 4 a 12 elementi.") + raw = quick_llm.ask(prompt, system="Analista discreto delle abitudini. Rispondi solo con il JSON richiesto.", max_tokens=900) + m = re.search(r"\[.*\]", raw, re.S) + try: + items = json.loads(m.group(0)) if m else [] + except Exception: + items = [] + items = [it for it in items if isinstance(it, dict) and it.get("abitudine")] + if not items: + return [] + mem = mm.load_memory() + for k in list((mem.get("habits") or {}).keys()): + if k.startswith("auto_"): + mm.forget(k, "habits") + out = [] + for it in items[:12]: + key = "auto_" + re.sub(r"[^a-z0-9]+", "_", str(it.get("chiave", "")).lower()).strip("_")[:40] + mm.remember(key, str(it["abitudine"]).strip()[:220], "habits") + out.append(str(it["abitudine"]).strip()) + STATO.parent.mkdir(parents=True, exist_ok=True) + STATO.write_text(json.dumps({"ultimo": time.time(), "quando": time.strftime("%Y-%m-%d %H:%M"), "abitudini": out, "dati": dati[:4000]}, indent=2, ensure_ascii=False), encoding="utf-8") + return out + + +def t_aggiorna(params: dict, ctx: dict) -> str: + if ctx.get("player") is not None and hasattr(ctx["player"], "set_state"): + ctx["player"].set_state("PROCESSING · studio le tue abitudini") + out = aggiorna(max(14, min(int(params.get("giorni") or 60), 180))) + if not out: + return "Non ho abbastanza dati per ricavare abitudini (servono richieste, calendario o messaggi degli ultimi giorni)." + return "Abitudini aggiornate e salvate in memoria:\n" + "\n".join(f"- {a}" for a in out) + + +def t_elenca(params: dict, ctx: dict) -> str: + mem = mm.load_memory().get("habits") or {} + if not mem: + return "Nessuna abitudine appresa finora: usa abitudini_aggiorna." + st = _stato() + lines = [f"- {mm._entry_value(v)}" for k, v in mem.items()] + return f"Abitudini in memoria ({len(lines)}, aggiornate il {st.get('quando', '?')}):\n" + "\n".join(lines) + + +def t_dati(params: dict, ctx: dict) -> str: + d = riassunto_dati(max(14, min(int(params.get("giorni") or 60), 180))) + return d or "Nessun dato disponibile." + + +TOOLS = [ + {"name": "abitudini_aggiorna", "description": "Ricava adesso le abitudini dell'utente dalle attività recenti (richieste, calendario, messaggi, casa) e le salva in memoria, sostituendo quelle apprese in automatico. Normalmente avviene da sola ogni settimana.", + "parameters": {"type": "object", "properties": {"giorni": {"type": "integer"}}}, "run": t_aggiorna}, + {"name": "abitudini_elenca", "description": "Elenca le abitudini dell'utente salvate in memoria (per 'cosa sai delle mie abitudini?').", + "parameters": {"type": "object", "properties": {}}, "run": t_elenca}, + {"name": "abitudini_dati", "description": "Statistiche grezze delle attività dell'utente (orari, giorni, strumenti, persone, eventi ricorrenti) per rispondere a domande come 'a che ora scrivo di solito?' o 'cosa faccio il sabato?'.", + "parameters": {"type": "object", "properties": {"giorni": {"type": "integer"}}}, "run": t_dati}, +]