App web per il telefono (HTTPS, PIN, WebSocket/SSE/interrogazione, voce, immagini, PDF, impostazioni), stati dettagliati, maestre di violino e canto, PDF, sessioni Claude Code, plugin manager rifatto
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
1 parent
a078dff608
commit
706c80cc20
23 files changed
+1944
-50
No files matched your search
@@ -0,0 +1,133 @@
|
||||
"""Maestra di canto: ascolta la voce, misura intonazione assoluta e relativa (intervalli), deriva di tonalità,
|
||||
vibrato e sostegno, e propone l'esercizio "canta questa nota" con il suono di riferimento.
|
||||
Riusa l'analizzatore del plugin violino (librosa/pYIN) con la gamma della voce.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from avatar.settings import DATA_DIR # noqa: E402
|
||||
import violino # noqa: E402
|
||||
|
||||
OUT_DIR = DATA_DIR / "canto"
|
||||
FMIN, FMAX = 90.0, 1500.0
|
||||
|
||||
|
||||
def _nota_midi(nome: str) -> int | None:
|
||||
"""'La4', 'Do5', 'Fa#4' → numero MIDI."""
|
||||
t = (nome or "").strip().lower().replace(" ", "")
|
||||
for k in sorted(violino.NOTE_IDX, key=len, reverse=True):
|
||||
if t.startswith(k):
|
||||
rest = t[len(k):]
|
||||
alt = 0
|
||||
if rest.startswith("#"):
|
||||
alt, rest = 1, rest[1:]
|
||||
elif rest.startswith("b"):
|
||||
alt, rest = -1, rest[1:]
|
||||
try:
|
||||
octave = int(rest) if rest else 4
|
||||
except ValueError:
|
||||
return None
|
||||
return (octave + 1) * 12 + violino.NOTE_IDX[k] + alt
|
||||
return None
|
||||
|
||||
|
||||
def _suona(midi: int, secondi: float = 1.5, a4: float = 440.0) -> None:
|
||||
import sounddevice as sd
|
||||
f = a4 * 2 ** ((midi - 69) / 12)
|
||||
t = np.linspace(0, secondi, int(violino.SR * secondi), endpoint=False)
|
||||
env = np.minimum(1, t / 0.05) * np.minimum(1, (secondi - t) / 0.2)
|
||||
tone = (0.5 * np.sin(2 * np.pi * f * t) + 0.15 * np.sin(4 * np.pi * f * t)) * env * 0.4
|
||||
sd.play(tone.astype(np.float32), violino.SR)
|
||||
sd.wait()
|
||||
|
||||
|
||||
def _testo_canto(rep: dict) -> str:
|
||||
if rep.get("silenzio"):
|
||||
return "Non ho sentito la voce: canta un po' più forte o avvicinati al microfono (mezzo metro va bene)."
|
||||
notes = rep.get("note") or []
|
||||
if not notes:
|
||||
return rep.get("msg", "Nessuna nota stabile: prova con note tenute, una vocale aperta come 'a'.")
|
||||
cents = np.array([x["cents"] for x in notes])
|
||||
seq = " ".join(f"{violino._nome(x['midi'])}({x['cents']:+.0f})" for x in notes[:40])
|
||||
out = [f"Registrazione di {rep['durata']:.0f} s, {rep['n_note']} note (nome e scarto in cent):", seq + (" …" if len(notes) > 40 else "")]
|
||||
# intonazione assoluta e relativa
|
||||
ivl_err = []
|
||||
for a, b in zip(notes, notes[1:]):
|
||||
d = (b["midi_f"] - a["midi_f"]) * 100
|
||||
ivl_err.append(d - round(d / 100) * 100)
|
||||
ivl = np.abs(np.array(ivl_err)) if ivl_err else np.array([])
|
||||
out.append(f"Intonazione assoluta: scarto medio {np.mean(np.abs(cents)):.0f} cent, {np.mean(np.abs(cents) <= 20) * 100:.0f}% delle note entro 20 cent, tendenza {'crescente' if cents.mean() > 8 else 'calante' if cents.mean() < -8 else 'centrata'} ({cents.mean():+.0f} cent).")
|
||||
if len(ivl):
|
||||
out.append(f"Intonazione relativa (intervalli tra note consecutive): errore medio {ivl.mean():.0f} cent, {np.mean(ivl <= 25) * 100:.0f}% degli intervalli puliti; intervallo peggiore {ivl.max():.0f} cent.")
|
||||
# deriva di tonalità: regressione dei cent nel tempo
|
||||
if len(notes) >= 4:
|
||||
t = np.array([x["start"] for x in notes]); k = np.polyfit(t, cents, 1)[0]
|
||||
out.append(f"Deriva della tonalità: {k * 10:+.0f} cent ogni 10 secondi ({'stabile' if abs(k * 10) < 12 else 'scende piano piano' if k < 0 else 'sale piano piano'}).")
|
||||
lung = [x for x in notes if x["dur"] >= 0.6]
|
||||
if lung:
|
||||
vib = np.mean([x["vib"] for x in lung]);
|
||||
out.append(f"Note tenute: {len(lung)}, oscillazione media {vib:.0f} cent ({'ferma' if vib < 10 else 'vibrato leggero' if vib < 25 else 'traballante: sostegno del fiato'}).")
|
||||
rng = rep["dinamica_range_db"]
|
||||
out.append(f"Volume: escursione {rng:.0f} dB {'(uniforme)' if rng < 6 else '(espressivo)' if rng < 18 else '(cali di fiato o note spinte)'}; note troppo brevi {rep['note_corte_pct']:.0f}%.")
|
||||
if rep.get("scala"):
|
||||
fuori = rep.get("fuori_scala") or []
|
||||
out.append(f"Tonalità {rep['scala']}: " + ("tutte le note appartengono alla scala." if not fuori else "note fuori: " + ", ".join(f"{violino._nome(x['midi'])} a {x['start']:.1f} s" for x in fuori[:6]) + "."))
|
||||
out.append("Interpreta per l'allieva: complimento specifico, poi al massimo due consigli (ascolta la nota prima di cantarla, fiato dal diaframma, vocale aperta), e un esercizio breve.")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def ascolta(params: dict, ctx: dict) -> str:
|
||||
secondi = max(5, min(int(params.get("secondi") or 15), 60))
|
||||
a4 = float(params.get("nota_la") or 440)
|
||||
y = violino.registra(secondi, ctx)
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
path = OUT_DIR / f"{time.strftime('%Y%m%d-%H%M%S')}.wav"
|
||||
try:
|
||||
import soundfile as sf
|
||||
sf.write(str(path), y, violino.SR)
|
||||
except Exception:
|
||||
pass
|
||||
rep = violino.analizza(y, violino.SR, a4, str(params.get("tonalita") or ""), FMIN, FMAX)
|
||||
return _testo_canto(rep) + (f"\nRegistrazione salvata: {path.name}" if path.exists() else "")
|
||||
|
||||
|
||||
def nota(params: dict, ctx: dict) -> str:
|
||||
"""Fa sentire una nota e ascolta se viene ricantata intonata."""
|
||||
import librosa
|
||||
nome = str(params.get("nota") or "La4")
|
||||
midi = _nota_midi(nome)
|
||||
if midi is None:
|
||||
return f"Nota '{nome}' non riconosciuta: usa nomi come Do4, Re4, Mi4, Fa#4, La4, Do5."
|
||||
a4 = float(params.get("nota_la") or 440)
|
||||
violino._progress(ctx, f"suono la nota {violino._nome(midi)}")
|
||||
_suona(midi, 1.5, a4)
|
||||
y = violino.registra(max(2, min(int(params.get("secondi") or 3), 8)), ctx, attesa=1)
|
||||
if 20 * math.log10(max(float(np.sqrt(np.mean(y ** 2))), 1e-6)) < -45:
|
||||
return "Non ho sentito la voce. Riproviamo: canta subito dopo il suono, tenendo la nota."
|
||||
f0, voiced, prob = librosa.pyin(y, fmin=FMIN, fmax=FMAX, sr=violino.SR)
|
||||
ok = voiced & (prob >= 0.6)
|
||||
if not np.any(ok):
|
||||
return "Ho sentito qualcosa ma non una nota tenuta: prova con una 'a' aperta e ferma."
|
||||
m = 69 + 12 * np.log2(np.nanmedian(f0[ok]) / a4)
|
||||
diff = (m - midi) * 100
|
||||
ottava = ""
|
||||
if abs(diff) > 700: # cantata all'ottava (voce di bambina o maschile)
|
||||
diff = diff - round(diff / 1200) * 1200; ottava = " (in un'altra ottava, va bene)"
|
||||
giudizio = "perfetta" if abs(diff) <= 15 else "quasi: un pelo " + ("alta" if diff > 0 else "bassa") if abs(diff) <= 40 else ("decisamente alta" if diff > 0 else "decisamente bassa")
|
||||
return f"Nota richiesta {violino._nome(midi)}, cantata a {diff:+.0f} cent{ottava}: {giudizio}. Stabilità {np.nanstd((69 + 12 * np.log2(f0[ok] / a4) - midi) * 100):.0f} cent."
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "canto_ascolta", "description": "Ascolta la voce dal microfono per alcuni secondi (default 15, max 60) e misura intonazione assoluta e relativa (intervalli tra le note), deriva della tonalità, note tenute e vibrato, volume e fiato. Con tonalita (es. 'Do maggiore') segnala le note fuori scala. Annuncia prima che stai per ascoltare: ci sono 3 secondi di preparazione. Restituisce dati da trasformare in consigli.",
|
||||
"parameters": {"type": "object", "properties": {"secondi": {"type": "integer"}, "tonalita": {"type": "string"}, "nota_la": {"type": "number"}}}, "run": ascolta},
|
||||
{"name": "canto_nota", "description": "Esercizio 'canta questa nota': fa sentire una nota di riferimento dalle casse (es. La4, Do5, Mi4) per 1,5 secondi, poi ascolta 3 secondi e dice se è stata cantata intonata, quanto alta o bassa e quanto ferma.",
|
||||
"parameters": {"type": "object", "properties": {"nota": {"type": "string"}, "secondi": {"type": "integer"}, "nota_la": {"type": "number"}}}, "run": nota},
|
||||
]
|
||||
@@ -0,0 +1,315 @@
|
||||
"""Sessioni di Claude Code (come l'app GestioneClaude): progetti e sessioni da ~/.claude*/projects, lettura,
|
||||
ricerca, ripresa nel Terminale e continuazione diretta da LuZa (claude -p --resume) con permessi a scelta.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.macos import confirm_or_param, CONFERMATO # noqa: E402
|
||||
from avatar.engines.claude_code import resolve_claude, claude_env # noqa: E402
|
||||
|
||||
HOME = Path.home()
|
||||
_cache: dict = {"t": 0.0, "sessions": []}
|
||||
LIVELLI = {
|
||||
"lettura": {"tools": "Read,Glob,Grep,WebSearch,WebFetch", "allowed": ["Read", "Glob", "Grep", "WebSearch", "WebFetch"], "mode": None},
|
||||
"modifica": {"tools": "default", "allowed": ["Bash", "Read", "Edit", "Write", "MultiEdit", "NotebookEdit", "Glob", "Grep", "WebSearch", "WebFetch", "Agent"], "mode": "acceptEdits"},
|
||||
}
|
||||
|
||||
|
||||
def _roots() -> list[Path]:
|
||||
out = []
|
||||
for d in sorted(HOME.iterdir()):
|
||||
if d.name.startswith(".claude") and (d / "projects").is_dir():
|
||||
out.append(d)
|
||||
return out
|
||||
|
||||
|
||||
def _decode_dir(name: str) -> str:
|
||||
return name.replace("-", "/") if name.startswith("-") else name
|
||||
|
||||
|
||||
def _text_of(msg) -> str:
|
||||
if not isinstance(msg, dict):
|
||||
return ""
|
||||
c = msg.get("content")
|
||||
if isinstance(c, str):
|
||||
return c
|
||||
if isinstance(c, list):
|
||||
return "\n".join(b.get("text", "") for b in c if isinstance(b, dict) and b.get("type") == "text")
|
||||
return ""
|
||||
|
||||
|
||||
def _head(path: Path) -> dict:
|
||||
"""cwd, primo prompt e data dalle prime righe del file."""
|
||||
cwd = prompt = ts = None
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
||||
for i, line in enumerate(f):
|
||||
if i > 80 or (cwd and prompt and ts):
|
||||
break
|
||||
try:
|
||||
o = json.loads(line)
|
||||
except Exception:
|
||||
continue
|
||||
cwd = cwd or o.get("cwd")
|
||||
ts = ts or o.get("timestamp")
|
||||
if not prompt and o.get("type") == "user" and not o.get("isSidechain"):
|
||||
t = _text_of(o.get("message")).strip()
|
||||
if t and not t.startswith(("<command-", "<local-command", "Caveat:", "[Request interrupted")):
|
||||
prompt = " ".join(t.split())[:160]
|
||||
except Exception:
|
||||
pass
|
||||
return {"cwd": cwd, "prompt": prompt, "ts": ts}
|
||||
|
||||
|
||||
def scan(force: bool = False) -> list[dict]:
|
||||
if not force and time.time() - _cache["t"] < 60 and _cache["sessions"]:
|
||||
return _cache["sessions"]
|
||||
out = []
|
||||
for root in _roots():
|
||||
for pdir in (root / "projects").iterdir():
|
||||
if not pdir.is_dir():
|
||||
continue
|
||||
for f in pdir.glob("*.jsonl"):
|
||||
try:
|
||||
st = f.stat()
|
||||
except OSError:
|
||||
continue
|
||||
if st.st_size < 200:
|
||||
continue
|
||||
h = _head(f)
|
||||
out.append({"id": f.stem, "file": f, "root": root.name, "cwd": h["cwd"] or _decode_dir(pdir.name), "prompt": h["prompt"] or "(senza prompt)",
|
||||
"start": h["ts"], "mtime": st.st_mtime, "size": st.st_size})
|
||||
out.sort(key=lambda s: -s["mtime"])
|
||||
_cache.update(t=time.time(), sessions=out)
|
||||
return out
|
||||
|
||||
|
||||
def _projects(sessions: list[dict]) -> list[dict]:
|
||||
by: dict[str, dict] = {}
|
||||
for s in sessions:
|
||||
p = by.setdefault(s["cwd"], {"path": s["cwd"], "name": Path(s["cwd"]).name or s["cwd"], "n": 0, "last": 0, "size": 0, "roots": set(), "exists": Path(s["cwd"]).is_dir()})
|
||||
p["n"] += 1; p["last"] = max(p["last"], s["mtime"]); p["size"] += s["size"]; p["roots"].add(s["root"])
|
||||
return sorted(by.values(), key=lambda p: -p["last"])
|
||||
|
||||
|
||||
def _norm(t: str) -> str:
|
||||
return re.sub(r"[^a-z0-9]", "", (t or "").lower())
|
||||
|
||||
|
||||
def _fmt_dt(ts: float) -> str:
|
||||
return datetime.fromtimestamp(ts).strftime("%d/%m %H:%M")
|
||||
|
||||
|
||||
def _find_project(nome: str, sessions: list[dict]) -> dict | None:
|
||||
n = _norm(nome)
|
||||
if not n:
|
||||
return None
|
||||
projs = _projects(sessions)
|
||||
for p in projs:
|
||||
if _norm(p["name"]) == n or _norm(p["path"]) == n:
|
||||
return p
|
||||
for p in projs:
|
||||
if n in _norm(p["name"]) or n in _norm(p["path"]):
|
||||
return p
|
||||
return None
|
||||
|
||||
|
||||
def _find_session(rif: str, progetto: str, sessions: list[dict]) -> dict | None:
|
||||
rif = (rif or "").strip().lower()
|
||||
pool = sessions
|
||||
if progetto:
|
||||
p = _find_project(progetto, sessions)
|
||||
if p:
|
||||
pool = [s for s in sessions if s["cwd"] == p["path"]]
|
||||
if not rif or rif in ("ultima", "recente", "last"):
|
||||
return pool[0] if pool else None
|
||||
for s in pool:
|
||||
if s["id"].lower() == rif or s["id"].lower().startswith(rif):
|
||||
return s
|
||||
for s in pool:
|
||||
if rif in s["prompt"].lower():
|
||||
return s
|
||||
return None
|
||||
|
||||
|
||||
def progetti(params: dict, ctx: dict) -> str:
|
||||
sessions = scan(force=True)
|
||||
filtro = _norm(str(params.get("filtro") or ""))
|
||||
projs = [p for p in _projects(sessions) if not filtro or filtro in _norm(p["name"]) or filtro in _norm(p["path"])]
|
||||
if not projs:
|
||||
return "Nessun progetto Claude Code trovato."
|
||||
lim = max(3, min(int(params.get("limite") or 20), 60))
|
||||
lines = [f"- {p['name']} — {p['n']} sessioni, ultima {_fmt_dt(p['last'])}, {p['size'] / 1e6:.1f} MB{'' if p['exists'] else ' (cartella non più presente)'} [{p['path']}]" for p in projs[:lim]]
|
||||
return f"Progetti Claude Code ({len(projs)}, {len(sessions)} sessioni in {len(_roots())} profili):\n" + "\n".join(lines)
|
||||
|
||||
|
||||
def sessioni(params: dict, ctx: dict) -> str:
|
||||
sessions = scan()
|
||||
nome = str(params.get("progetto") or "")
|
||||
p = _find_project(nome, sessions) if nome else None
|
||||
if nome and not p:
|
||||
return f"Progetto '{nome}' non trovato. Usa claude_progetti per l'elenco."
|
||||
pool = [s for s in sessions if not p or s["cwd"] == p["path"]]
|
||||
lim = max(3, min(int(params.get("numero") or 10), 40))
|
||||
lines = [f"- {s['id'][:8]} · {_fmt_dt(s['mtime'])} · {s['size'] / 1e3:.0f} KB · {s['prompt'][:90]}" + ("" if p else f" [{Path(s['cwd']).name}]") for s in pool[:lim]]
|
||||
return (f"Sessioni di {p['name']}" if p else "Sessioni recenti") + f" ({len(pool)}):\n" + "\n".join(lines) + "\nRiferisciti a una sessione con le prime cifre dell'id o con 'ultima'."
|
||||
|
||||
|
||||
def _messages(path: Path):
|
||||
with open(path, "r", encoding="utf-8", errors="replace") as f:
|
||||
for line in f:
|
||||
try:
|
||||
o = json.loads(line)
|
||||
except Exception:
|
||||
continue
|
||||
if o.get("isSidechain") or o.get("type") not in ("user", "assistant"):
|
||||
continue
|
||||
t = _text_of(o.get("message")).strip()
|
||||
if not t or t.startswith(("<command-", "<local-command", "Caveat:")):
|
||||
continue
|
||||
yield o["type"], t, o.get("timestamp", "")
|
||||
|
||||
|
||||
def leggi(params: dict, ctx: dict) -> str:
|
||||
sessions = scan()
|
||||
s = _find_session(str(params.get("sessione") or ""), str(params.get("progetto") or ""), sessions)
|
||||
if not s:
|
||||
return "Sessione non trovata: indica progetto e le prime cifre dell'id (vedi claude_sessioni)."
|
||||
n = max(4, min(int(params.get("numero") or 20), 80))
|
||||
per = max(200, min(int(params.get("caratteri") or 700), 3000))
|
||||
msgs = list(_messages(s["file"]))
|
||||
parte = msgs[:n] if str(params.get("parte") or "fine") == "inizio" else msgs[-n:]
|
||||
out = [f"Sessione {s['id'][:8]} di {Path(s['cwd']).name} ({len(msgs)} messaggi, {'inizio' if parte is msgs[:n] else 'fine'}):"]
|
||||
for role, t, ts in parte:
|
||||
out.append(f"[{'Utente' if role == 'user' else 'Claude'} {ts[11:16] if ts else ''}] {t[:per]}{'…' if len(t) > per else ''}")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def cerca(params: dict, ctx: dict) -> str:
|
||||
q = str(params.get("query") or "").strip().lower()
|
||||
if not q:
|
||||
return "Errore: serve il testo da cercare."
|
||||
sessions = scan()
|
||||
p = _find_project(str(params.get("progetto") or ""), sessions) if params.get("progetto") else None
|
||||
hits, checked = [], 0
|
||||
for s in sessions:
|
||||
if p and s["cwd"] != p["path"]:
|
||||
continue
|
||||
if s["size"] > 60_000_000:
|
||||
continue
|
||||
checked += 1
|
||||
try:
|
||||
for role, t, ts in _messages(s["file"]):
|
||||
if q in t.lower():
|
||||
i = t.lower().find(q)
|
||||
hits.append(f"- {Path(s['cwd']).name} · {s['id'][:8]} · {_fmt_dt(s['mtime'])} [{'Utente' if role == 'user' else 'Claude'}]: …{' '.join(t[max(0, i - 80):i + 120].split())}…")
|
||||
if len([h for h in hits if s['id'][:8] in h]) >= 2:
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
if len(hits) >= 25:
|
||||
break
|
||||
if not hits:
|
||||
return f"Nessun risultato per '{q}' in {checked} sessioni."
|
||||
return f"Risultati per '{q}' ({len(hits)}):\n" + "\n".join(hits)
|
||||
|
||||
|
||||
def _terminal(cmd: str) -> None:
|
||||
script = f'tell application "Terminal"\nactivate\ndo script {json.dumps(cmd)}\nend tell'
|
||||
subprocess.run(["osascript", "-e", script], capture_output=True, timeout=20)
|
||||
|
||||
|
||||
def riprendi(params: dict, ctx: dict) -> str:
|
||||
sessions = scan()
|
||||
s = _find_session(str(params.get("sessione") or ""), str(params.get("progetto") or ""), sessions)
|
||||
if not s:
|
||||
return "Sessione non trovata."
|
||||
env = f"CLAUDE_CONFIG_DIR={json.dumps(str(HOME / s['root']))} " if s["root"] != ".claude" else ""
|
||||
_terminal(f"cd {json.dumps(s['cwd'])} && {env}claude --resume {s['id']}")
|
||||
return f"Aperto il Terminale nel progetto {Path(s['cwd']).name} con la sessione {s['id'][:8]} ripresa."
|
||||
|
||||
|
||||
def nuova(params: dict, ctx: dict) -> str:
|
||||
sessions = scan()
|
||||
p = _find_project(str(params.get("progetto") or ""), sessions)
|
||||
path = p["path"] if p else str(Path(str(params.get("progetto") or "")).expanduser())
|
||||
if not Path(path).is_dir():
|
||||
return f"Cartella non trovata: {path}"
|
||||
root = next(iter(sorted(p["roots"])), ".claude") if p else ".claude"
|
||||
env = f"CLAUDE_CONFIG_DIR={json.dumps(str(HOME / root))} " if root != ".claude" else ""
|
||||
_terminal(f"cd {json.dumps(path)} && {env}claude")
|
||||
return f"Aperto il Terminale in {Path(path).name} con una nuova sessione di Claude Code."
|
||||
|
||||
|
||||
def continua(params: dict, ctx: dict) -> str:
|
||||
"""Manda una richiesta a Claude Code dentro una sessione esistente (o nuova) del progetto, senza Terminale."""
|
||||
sessions = scan()
|
||||
richiesta = str(params.get("richiesta") or "").strip()
|
||||
if not richiesta:
|
||||
return "Errore: serve la richiesta da fare a Claude."
|
||||
s = _find_session(str(params.get("sessione") or ""), str(params.get("progetto") or ""), sessions)
|
||||
p = _find_project(str(params.get("progetto") or ""), sessions) if params.get("progetto") else None
|
||||
if not s and not p:
|
||||
return "Indica la sessione (prime cifre dell'id o 'ultima') o il progetto."
|
||||
cwd = s["cwd"] if s else p["path"]
|
||||
root = s["root"] if s else next(iter(sorted(p["roots"])), ".claude")
|
||||
livello = "modifica" if str(params.get("permessi") or "lettura").lower().startswith("mod") else "lettura"
|
||||
binary = resolve_claude("")
|
||||
if not binary:
|
||||
return "Non trovo il comando claude."
|
||||
|
||||
def do() -> str:
|
||||
flags = LIVELLI[livello]
|
||||
args = [binary, "-p", "--output-format", "json", "--tools", flags["tools"], "--allowedTools", ",".join(flags["allowed"])]
|
||||
if flags["mode"]:
|
||||
args += ["--permission-mode", flags["mode"]]
|
||||
args += ["--resume", s["id"]] if s else []
|
||||
player = ctx.get("player")
|
||||
if player is not None and hasattr(player, "set_state"):
|
||||
player.set_state(f"PROCESSING · Claude Code su {Path(cwd).name}")
|
||||
env = claude_env(str(HOME / root) if root != ".claude" else "")
|
||||
try:
|
||||
r = subprocess.run(args, input=richiesta, cwd=cwd, env=env, capture_output=True, text=True, timeout=900)
|
||||
except subprocess.TimeoutExpired:
|
||||
return "Claude Code non ha risposto entro 15 minuti."
|
||||
try:
|
||||
d = json.loads(r.stdout.strip().splitlines()[-1])
|
||||
text = d.get("result") or ""
|
||||
sid = d.get("session_id", "")
|
||||
if d.get("is_error"):
|
||||
return f"Claude Code ha segnalato un errore: {text or r.stderr[-300:]}"
|
||||
except Exception:
|
||||
text, sid = (r.stdout or r.stderr)[-2000:], ""
|
||||
_cache["t"] = 0
|
||||
return f"Risposta di Claude Code ({Path(cwd).name}, sessione {(sid or (s['id'] if s else ''))[:8]}, permessi: {livello}):\n{text}"
|
||||
|
||||
if livello == "modifica":
|
||||
return confirm_or_param(ctx, params, "claude_continua", f"Far modificare il progetto {Path(cwd).name} a Claude Code?", richiesta[:300], do)
|
||||
return do()
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "claude_progetti", "description": "Elenca i progetti con sessioni di Claude Code (tutti i profili ~/.claude*): numero di sessioni, ultima attività, spazio, percorso. Filtro opzionale per nome.",
|
||||
"parameters": {"type": "object", "properties": {"filtro": {"type": "string"}, "limite": {"type": "integer"}}}, "run": progetti},
|
||||
{"name": "claude_sessioni", "description": "Elenca le sessioni di Claude Code di un progetto (o le più recenti di tutti), con id abbreviato, data, dimensione e prima richiesta.",
|
||||
"parameters": {"type": "object", "properties": {"progetto": {"type": "string"}, "numero": {"type": "integer"}}}, "run": sessioni},
|
||||
{"name": "claude_leggi", "description": "Legge i messaggi di una sessione di Claude Code (utente e Claude), dalla fine o dall'inizio, per riassumerla o riprendere il filo. sessione = prime cifre dell'id oppure 'ultima' (con progetto).",
|
||||
"parameters": {"type": "object", "properties": {"sessione": {"type": "string"}, "progetto": {"type": "string"}, "numero": {"type": "integer"}, "parte": {"type": "string", "enum": ["fine", "inizio"]}, "caratteri": {"type": "integer"}}}, "run": leggi},
|
||||
{"name": "claude_cerca", "description": "Cerca un testo nei messaggi di tutte le sessioni di Claude Code (o di un progetto): utile per ritrovare dove si è parlato di qualcosa.",
|
||||
"parameters": {"type": "object", "properties": {"query": {"type": "string"}, "progetto": {"type": "string"}}, "required": ["query"]}, "run": cerca},
|
||||
{"name": "claude_continua", "description": "Fa lavorare Claude Code su un progetto senza aprire il Terminale: manda la richiesta dentro una sessione esistente (sessione = prime cifre dell'id o 'ultima') o in una nuova sessione del progetto, e riporta la risposta. permessi=lettura (default: legge e cerca nel codice) oppure modifica (può cambiare file ed eseguire comandi: chiede conferma sullo schermo). Può richiedere minuti.",
|
||||
"parameters": {"type": "object", "properties": {"richiesta": {"type": "string"}, "sessione": {"type": "string"}, "progetto": {"type": "string"}, "permessi": {"type": "string", "enum": ["lettura", "modifica"]}, "confermato": CONFERMATO}, "required": ["richiesta"]}, "run": continua},
|
||||
{"name": "claude_riprendi", "description": "Apre il Terminale nel progetto e riprende una sessione di Claude Code (claude --resume) perché l'utente continui a mano.",
|
||||
"parameters": {"type": "object", "properties": {"sessione": {"type": "string"}, "progetto": {"type": "string"}}}, "run": riprendi},
|
||||
{"name": "claude_nuova", "description": "Apre il Terminale in un progetto e avvia una nuova sessione di Claude Code.",
|
||||
"parameters": {"type": "object", "properties": {"progetto": {"type": "string"}}, "required": ["progetto"]}, "run": nuova},
|
||||
]
|
||||
@@ -0,0 +1,204 @@
|
||||
"""Documenti PDF: impagina in A4 un testo in Markdown (titoli, elenchi, grassetto, corsivo, tabelle, immagini),
|
||||
lo salva in data/documenti/ e lo apre in Anteprima. Il file arriva anche al telefono.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.settings import DATA_DIR # noqa: E402
|
||||
|
||||
OUT_DIR = DATA_DIR / "documenti"
|
||||
FONT_DIR = Path("/System/Library/Fonts/Supplemental")
|
||||
|
||||
|
||||
def _fonts():
|
||||
from reportlab.pdfbase import pdfmetrics
|
||||
from reportlab.pdfbase.ttfonts import TTFont
|
||||
from reportlab.lib.fonts import addMapping
|
||||
if "Arial" in pdfmetrics.getRegisteredFontNames():
|
||||
return "Arial"
|
||||
try:
|
||||
pdfmetrics.registerFont(TTFont("Arial", str(FONT_DIR / "Arial.ttf")))
|
||||
pdfmetrics.registerFont(TTFont("Arial-Bold", str(FONT_DIR / "Arial Bold.ttf")))
|
||||
pdfmetrics.registerFont(TTFont("Arial-Italic", str(FONT_DIR / "Arial Italic.ttf")))
|
||||
pdfmetrics.registerFont(TTFont("Arial-BoldItalic", str(FONT_DIR / "Arial Bold Italic.ttf")))
|
||||
addMapping("Arial", 0, 0, "Arial"); addMapping("Arial", 1, 0, "Arial-Bold")
|
||||
addMapping("Arial", 0, 1, "Arial-Italic"); addMapping("Arial", 1, 1, "Arial-BoldItalic")
|
||||
return "Arial"
|
||||
except Exception:
|
||||
return "Helvetica"
|
||||
|
||||
|
||||
def _inline(text: str) -> str:
|
||||
"""Markdown in linea → mini-HTML di reportlab (grassetto, corsivo, codice, link)."""
|
||||
t = html.escape(text, quote=False)
|
||||
t = re.sub(r"\*\*(.+?)\*\*", r"<b>\1</b>", t)
|
||||
t = re.sub(r"(?<!\*)\*(?!\s)(.+?)(?<!\s)\*(?!\*)", r"<i>\1</i>", t)
|
||||
t = re.sub(r"`(.+?)`", r"<font face='Courier'>\1</font>", t)
|
||||
t = re.sub(r"\[(.+?)\]\((https?://[^\s)]+)\)", r"<link href='\2' color='#0645ad'>\1</link>", t)
|
||||
return t
|
||||
|
||||
|
||||
def costruisci(titolo: str, markdown: str, path: Path, autore: str = "", immagini: list[str] | None = None) -> None:
|
||||
from reportlab.lib.pagesizes import A4
|
||||
from reportlab.lib.styles import ParagraphStyle
|
||||
from reportlab.lib.units import mm
|
||||
from reportlab.lib import colors
|
||||
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, ListFlowable, ListItem, Table, TableStyle, Image, PageBreak, HRFlowable
|
||||
font = _fonts()
|
||||
st = {
|
||||
"title": ParagraphStyle("t", fontName=font, fontSize=22, leading=27, spaceAfter=10, textColor=colors.HexColor("#123")),
|
||||
"h1": ParagraphStyle("h1", fontName=font, fontSize=16, leading=20, spaceBefore=12, spaceAfter=6, textColor=colors.HexColor("#123")),
|
||||
"h2": ParagraphStyle("h2", fontName=font, fontSize=13, leading=17, spaceBefore=10, spaceAfter=4),
|
||||
"h3": ParagraphStyle("h3", fontName=font, fontSize=11.5, leading=15, spaceBefore=8, spaceAfter=3),
|
||||
"p": ParagraphStyle("p", fontName=font, fontSize=10.5, leading=15, spaceAfter=6),
|
||||
"quote": ParagraphStyle("q", fontName=font, fontSize=10.5, leading=15, leftIndent=14, textColor=colors.HexColor("#444"), spaceAfter=6),
|
||||
"small": ParagraphStyle("s", fontName=font, fontSize=8.5, leading=11, textColor=colors.HexColor("#777")),
|
||||
"code": ParagraphStyle("c", fontName="Courier", fontSize=9, leading=12, backColor=colors.HexColor("#f4f4f4"), leftIndent=6, spaceAfter=6),
|
||||
}
|
||||
story = []
|
||||
if titolo:
|
||||
story += [Paragraph(_inline(titolo), st["title"]), HRFlowable(width="100%", color=colors.HexColor("#123"), thickness=0.8, spaceAfter=8)]
|
||||
if autore:
|
||||
story.append(Paragraph(html.escape(autore), st["small"])); story.append(Spacer(1, 6))
|
||||
lines = (markdown or "").replace("\r", "").split("\n")
|
||||
i, para, lst, table, code = 0, [], None, None, None
|
||||
|
||||
def flush_para():
|
||||
nonlocal para
|
||||
if para:
|
||||
story.append(Paragraph(_inline(" ".join(para)), st["p"])); para = []
|
||||
|
||||
def flush_list():
|
||||
nonlocal lst
|
||||
if lst:
|
||||
kind, items = lst
|
||||
story.append(ListFlowable([ListItem(Paragraph(_inline(x), st["p"]), leftIndent=12) for x in items],
|
||||
bulletType="1" if kind == "ol" else "bullet", start="1" if kind == "ol" else None, bulletFormat="%s." if kind == "ol" else None, leftIndent=14, bulletFontName=font, bulletFontSize=9))
|
||||
lst = None
|
||||
|
||||
def flush_table():
|
||||
nonlocal table
|
||||
if table:
|
||||
rows = [[Paragraph(_inline(c), st["p"]) for c in r] for r in table]
|
||||
t = Table(rows, hAlign="LEFT", repeatRows=1)
|
||||
t.setStyle(TableStyle([("GRID", (0, 0), (-1, -1), 0.4, colors.HexColor("#999")), ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#e8eef3")),
|
||||
("VALIGN", (0, 0), (-1, -1), "TOP"), ("LEFTPADDING", (0, 0), (-1, -1), 5), ("RIGHTPADDING", (0, 0), (-1, -1), 5)]))
|
||||
story.append(t); story.append(Spacer(1, 8)); table = None
|
||||
|
||||
while i < len(lines):
|
||||
ln = lines[i].rstrip()
|
||||
if code is not None:
|
||||
if ln.strip().startswith("```"):
|
||||
story.append(Paragraph(html.escape("\n".join(code)).replace("\n", "<br/>"), st["code"])); code = None
|
||||
else:
|
||||
code.append(ln)
|
||||
i += 1; continue
|
||||
if ln.strip().startswith("```"):
|
||||
flush_para(); flush_list(); flush_table(); code = []; i += 1; continue
|
||||
if not ln.strip():
|
||||
flush_para(); flush_list(); flush_table(); i += 1; continue
|
||||
if ln.strip() in ("---", "***", "___"):
|
||||
flush_para(); flush_list(); flush_table(); story.append(HRFlowable(width="100%", color=colors.HexColor("#bbb"), spaceBefore=4, spaceAfter=8)); i += 1; continue
|
||||
if ln.strip().lower() in ("\\pagebreak", "<pagebreak>", "[pagina]"):
|
||||
flush_para(); flush_list(); flush_table(); story.append(PageBreak()); i += 1; continue
|
||||
m = re.match(r"^(#{1,3})\s+(.*)$", ln)
|
||||
if m:
|
||||
flush_para(); flush_list(); flush_table()
|
||||
story.append(Paragraph(_inline(m.group(2)), st[f"h{len(m.group(1))}"])); i += 1; continue
|
||||
m = re.match(r"^!\[([^\]]*)\]\(([^)]+)\)", ln.strip())
|
||||
if m:
|
||||
flush_para(); flush_list(); flush_table()
|
||||
p = Path(m.group(2)).expanduser()
|
||||
if p.exists():
|
||||
img = Image(str(p)); ratio = img.imageHeight / img.imageWidth; w = min(160 * mm, img.imageWidth * 0.264 * mm)
|
||||
img.drawWidth, img.drawHeight = w, w * ratio; story.append(img)
|
||||
if m.group(1):
|
||||
story.append(Paragraph(html.escape(m.group(1)), st["small"]))
|
||||
story.append(Spacer(1, 8))
|
||||
i += 1; continue
|
||||
if ln.strip().startswith("|"):
|
||||
flush_para(); flush_list()
|
||||
cells = [c.strip() for c in ln.strip().strip("|").split("|")]
|
||||
if not all(re.fullmatch(r":?-{2,}:?", c) for c in cells):
|
||||
table = (table or []) + [cells]
|
||||
i += 1; continue
|
||||
m = re.match(r"^\s*([-*•]|\d+[.)])\s+(.*)$", ln)
|
||||
if m:
|
||||
flush_para(); flush_table()
|
||||
kind = "ol" if m.group(1)[0].isdigit() else "ul"
|
||||
if lst and lst[0] != kind:
|
||||
flush_list()
|
||||
lst = (kind, (lst[1] if lst else []) + [m.group(2)]); i += 1; continue
|
||||
if ln.startswith(">"):
|
||||
flush_para(); flush_list(); flush_table()
|
||||
story.append(Paragraph(_inline(ln.lstrip("> ")), st["quote"])); i += 1; continue
|
||||
flush_list(); flush_table(); para.append(ln.strip()); i += 1
|
||||
flush_para(); flush_list(); flush_table()
|
||||
for p in immagini or []:
|
||||
pp = Path(p).expanduser()
|
||||
if pp.exists():
|
||||
img = Image(str(pp)); ratio = img.imageHeight / img.imageWidth; w = min(160 * mm, img.imageWidth * 0.264 * mm)
|
||||
img.drawWidth, img.drawHeight = w, w * ratio; story += [Spacer(1, 8), img]
|
||||
|
||||
def footer(canvas, doc):
|
||||
canvas.saveState(); canvas.setFont(font, 8); canvas.setFillColor(colors.HexColor("#888"))
|
||||
canvas.drawRightString(A4[0] - 20 * mm, 12 * mm, f"{doc.page}")
|
||||
if titolo:
|
||||
canvas.drawString(20 * mm, 12 * mm, titolo[:80])
|
||||
canvas.restoreState()
|
||||
|
||||
doc = SimpleDocTemplate(str(path), pagesize=A4, leftMargin=20 * mm, rightMargin=20 * mm, topMargin=20 * mm, bottomMargin=20 * mm,
|
||||
title=titolo or path.stem, author=autore or "LuZa")
|
||||
doc.build(story or [Paragraph("(documento vuoto)", st["p"])], onFirstPage=footer, onLaterPages=footer)
|
||||
|
||||
|
||||
def crea(params: dict, ctx: dict) -> str:
|
||||
titolo = str(params.get("titolo") or "").strip()
|
||||
contenuto = str(params.get("contenuto") or "").strip()
|
||||
if not contenuto and not titolo:
|
||||
return "Errore: serve il contenuto del documento (in Markdown)."
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
nome = re.sub(r"[^a-z0-9]+", "-", str(params.get("nome_file") or titolo or "documento").lower()).strip("-")[:60] or "documento"
|
||||
path = OUT_DIR / f"{time.strftime('%Y%m%d-%H%M%S')}-{nome}.pdf"
|
||||
immagini = params.get("immagini") or []
|
||||
if isinstance(immagini, str):
|
||||
immagini = [immagini]
|
||||
try:
|
||||
costruisci(titolo, contenuto, path, str(params.get("autore") or ""), [str(x) for x in immagini])
|
||||
except Exception as err:
|
||||
return f"Impaginazione fallita: {err}"
|
||||
if params.get("mostra", True) not in (False, "false", "no"):
|
||||
subprocess.Popen(["open", str(path)])
|
||||
pagine = ""
|
||||
try:
|
||||
from pypdf import PdfReader
|
||||
pagine = f", {len(PdfReader(str(path)).pages)} pagine"
|
||||
except Exception:
|
||||
pass
|
||||
return f"PDF creato: {path}{pagine}. Aperto in Anteprima.\nFILE: {path}"
|
||||
|
||||
|
||||
def ultimi(params: dict, ctx: dict) -> str:
|
||||
files = sorted(OUT_DIR.glob("*.pdf"), key=lambda p: p.stat().st_mtime, reverse=True) if OUT_DIR.exists() else []
|
||||
if not files:
|
||||
return "Nessun documento creato finora."
|
||||
if str(params.get("apri", "")).lower() in ("true", "1", "sì", "si"):
|
||||
subprocess.Popen(["open", str(files[0])])
|
||||
n = max(1, min(int(params.get("numero") or 5), 20))
|
||||
return "Ultimi documenti:\n" + "\n".join(f"- {p.name}" for p in files[:n]) + f"\nFILE: {files[0]}"
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "pdf_crea", "description": "Crea un documento PDF in A4 da un testo in Markdown: usa # ## ### per i titoli, - o 1. per gli elenchi, **grassetto**, *corsivo*, tabelle con |,  per le immagini, --- per una riga, [pagina] per il salto pagina. Scrivi tu il contenuto completo e ben strutturato (lettere, relazioni, elenchi, ricette, verbali…). Il PDF viene salvato, aperto in Anteprima e inviato al telefono.",
|
||||
"parameters": {"type": "object", "properties": {"titolo": {"type": "string"}, "contenuto": {"type": "string", "description": "testo in Markdown"}, "nome_file": {"type": "string"}, "autore": {"type": "string"},
|
||||
"immagini": {"type": "array", "items": {"type": "string"}, "description": "percorsi di immagini da accodare"}, "mostra": {"type": "boolean"}}, "required": ["contenuto"]}, "run": crea},
|
||||
{"name": "pdf_ultimi", "description": "Elenca gli ultimi PDF creati (cartella data/documenti); con apri=true apre il più recente.",
|
||||
"parameters": {"type": "object", "properties": {"numero": {"type": "integer"}, "apri": {"type": "boolean"}}}, "run": ultimi},
|
||||
]
|
||||
+1
-1
@@ -148,7 +148,7 @@ def ultima(params: dict, ctx: dict) -> str:
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "immagine_genera", "description": "Genera un'immagine in locale (MLX) da una descrizione. Scrivi la descrizione in INGLESE, dettagliata (soggetto, ambiente, luce, stile, inquadratura), anche se l'utente parla italiano. formato: quadrata (default), orizzontale, verticale, schermo, piccola. L'immagine viene salvata e aperta in Anteprima; richiede 10-60 secondi, il primo uso scarica il modello. Con immagine_base (percorso) e forza (0-1) parte da un'immagine esistente.",
|
||||
{"name": "immagine_genera", "description": "Genera un'immagine in locale (MLX) da una descrizione. Scrivi la descrizione in INGLESE traducendo FEDELMENTE la richiesta dell'utente: stessi soggetti e stessi dettagli, senza attenuare, censurare o cambiare ciò che ha chiesto; aggiungi solo dettagli tecnici (luce, inquadratura, stile) se mancano. Se l'utente ha scritto il prompt in inglese, usalo tale e quale. formato: quadrata (default), orizzontale, verticale, schermo, piccola. L'immagine viene salvata e aperta in Anteprima; richiede 10-60 secondi, il primo uso scarica il modello. Con immagine_base (percorso) e forza (0-1) parte da un'immagine esistente.",
|
||||
"parameters": {"type": "object", "properties": {"descrizione": {"type": "string"}, "formato": {"type": "string", "enum": ["quadrata", "orizzontale", "verticale", "schermo", "piccola"]},
|
||||
"passi": {"type": "integer"}, "seme": {"type": "integer"}, "immagine_base": {"type": "string"}, "forza": {"type": "number"}, "mostra": {"type": "boolean"}},
|
||||
"required": ["descrizione"]}, "run": genera},
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
"""Maestro di violino: ascolta dal microfono, misura intonazione, ritmo, dinamica e pulizia del suono e riferisce
|
||||
i dati al modello, che li trasforma in consigli. Analisi con librosa (pYIN), registrazioni salvate in data/violino/.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
from avatar.settings import DATA_DIR # noqa: E402
|
||||
|
||||
SR = 44100
|
||||
OUT_DIR = DATA_DIR / "violino"
|
||||
CORDE = {"Sol": 196.00, "Re": 293.66, "La": 440.00, "Mi": 659.25}
|
||||
NOTE_IT = ["Do", "Do#", "Re", "Re#", "Mi", "Fa", "Fa#", "Sol", "Sol#", "La", "La#", "Si"]
|
||||
NOTE_IDX = {"do": 0, "re": 2, "mi": 4, "fa": 5, "sol": 7, "la": 9, "si": 11}
|
||||
SCALE_MAJ = [0, 2, 4, 5, 7, 9, 11]
|
||||
SCALE_MIN = [0, 2, 3, 5, 7, 8, 10]
|
||||
|
||||
|
||||
def _device():
|
||||
"""Stesso microfono dell'assistente; vuoto = predefinito di sistema."""
|
||||
try:
|
||||
from memory import config_manager as cm
|
||||
d = cm.get_input_device()
|
||||
return d if d not in ("", None, "default") else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _progress(ctx: dict, text: str) -> None:
|
||||
p = ctx.get("player")
|
||||
if p is not None and hasattr(p, "set_state"):
|
||||
try:
|
||||
p.set_state(f"PROCESSING · {text}")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def registra(secondi: float, ctx: dict, attesa: float = 3.0) -> np.ndarray:
|
||||
"""Registra dal microfono; prima un breve conto alla rovescia (LuZa sta ancora parlando)."""
|
||||
import sounddevice as sd
|
||||
for i in range(int(attesa), 0, -1):
|
||||
_progress(ctx, f"preparati · {i}")
|
||||
time.sleep(1)
|
||||
buf = sd.rec(int(secondi * SR), samplerate=SR, channels=1, dtype="float32", device=_device())
|
||||
t0 = time.time()
|
||||
while time.time() - t0 < secondi:
|
||||
_progress(ctx, f"ascolto il violino · {int(secondi - (time.time() - t0))} s")
|
||||
time.sleep(0.5)
|
||||
sd.wait()
|
||||
_progress(ctx, "analizzo")
|
||||
return buf[:, 0].copy()
|
||||
|
||||
|
||||
def _nome(midi: int) -> str:
|
||||
return f"{NOTE_IT[midi % 12]}{midi // 12 - 1}"
|
||||
|
||||
|
||||
def _scala(testo: str) -> set[int] | None:
|
||||
t = (testo or "").lower().strip()
|
||||
if not t:
|
||||
return None
|
||||
parts = t.replace("#", " #").replace("b", " b").split()
|
||||
if not parts or parts[0] not in NOTE_IDX:
|
||||
return None
|
||||
root = NOTE_IDX[parts[0]]
|
||||
if "#" in parts[1:2]:
|
||||
root += 1
|
||||
elif "b" in parts[1:2]:
|
||||
root -= 1
|
||||
minor = "min" in t
|
||||
return {(root + x) % 12 for x in (SCALE_MIN if minor else SCALE_MAJ)}
|
||||
|
||||
|
||||
def analizza(y: np.ndarray, sr: int = SR, a4: float = 440.0, scala: str = "", fmin: float = 180.0, fmax: float = 2800.0) -> dict:
|
||||
import librosa
|
||||
y = y.astype(np.float32)
|
||||
hop = 512
|
||||
rms = librosa.feature.rms(y=y, frame_length=2048, hop_length=hop)[0]
|
||||
db = 20 * np.log10(np.maximum(rms, 1e-6))
|
||||
if db.max() < -45:
|
||||
return {"silenzio": True}
|
||||
f0, voiced, prob = librosa.pyin(y, fmin=fmin, fmax=fmax, sr=sr, frame_length=2048, hop_length=hop)
|
||||
flat = librosa.feature.spectral_flatness(y=y, n_fft=2048, hop_length=hop)[0]
|
||||
n = min(len(f0), len(db), len(flat))
|
||||
f0, voiced, db, flat = f0[:n], voiced[:n], db[:n], flat[:n]
|
||||
from scipy.signal import medfilt
|
||||
ok = voiced & np.isfinite(f0) & (prob >= 0.6) & (db >= db.max() - 32)
|
||||
midi = np.where(ok, 69 + 12 * np.log2(np.maximum(np.nan_to_num(f0, nan=1.0), 1) / a4), np.nan)
|
||||
# stabilizza il contorno: mediana su 5 frame, poi segmentazione con isteresi (3 frame fuori dal ±60 cent per cambiare nota)
|
||||
sm = midi.copy()
|
||||
fin = np.isfinite(midi)
|
||||
if fin.sum() > 5:
|
||||
filled = np.interp(np.arange(n), np.flatnonzero(fin), midi[fin])
|
||||
sm = np.where(fin, medfilt(filled, 5), np.nan)
|
||||
segs, cur, gap, out_cnt = [], [], 0, 0
|
||||
for i in range(n):
|
||||
if np.isnan(sm[i]):
|
||||
gap += 1
|
||||
if gap >= 5 and cur:
|
||||
segs.append(cur); cur = []
|
||||
continue
|
||||
gap = 0
|
||||
if cur and abs(sm[i] - np.nanmedian(sm[cur[-10:]])) * 100 > 60:
|
||||
out_cnt += 1
|
||||
if out_cnt >= 3:
|
||||
segs.append(cur[:-2] if len(cur) > 2 else cur); cur = cur[-2:] + [i]; out_cnt = 0
|
||||
continue
|
||||
else:
|
||||
out_cnt = 0
|
||||
cur.append(i)
|
||||
if cur:
|
||||
segs.append(cur)
|
||||
# note: scarta i frammenti brevi, unisci segmenti contigui con la stessa nota
|
||||
notes = []
|
||||
for idx in segs:
|
||||
if len(idx) < 8:
|
||||
continue
|
||||
idx = np.array(idx)
|
||||
m = float(np.nanmedian(sm[idx]))
|
||||
cents_arr = (sm[idx] - round(m)) * 100
|
||||
note = {"start": idx[0] * hop / sr, "dur": len(idx) * hop / sr, "midi": int(round(m)), "midi_f": m, "cents": float(np.nanmedian(cents_arr)),
|
||||
"vib": float(np.nanstd(cents_arr)), "db": float(np.mean(db[idx])), "flat": float(np.mean(flat[idx]))}
|
||||
# stessa nota ripetuta: resta separata se tra le due il volume cala (nuovo attacco d'arco o di voce)
|
||||
gap_lo = int((notes[-1]["start"] + notes[-1]["dur"]) * sr / hop) if notes else 0
|
||||
dip = float(notes[-1]["db"] - db[gap_lo:idx[0] + 1].min()) if notes and idx[0] > gap_lo else 0.0
|
||||
if notes and notes[-1]["midi"] == note["midi"] and note["start"] - (notes[-1]["start"] + notes[-1]["dur"]) < 0.12 and dip < 8:
|
||||
prev = notes[-1]; w1, w2 = prev["dur"], note["dur"]
|
||||
prev["cents"] = (prev["cents"] * w1 + note["cents"] * w2) / (w1 + w2)
|
||||
prev["dur"] = note["start"] + note["dur"] - prev["start"]
|
||||
prev["vib"] = max(prev["vib"], note["vib"]); prev["flat"] = (prev["flat"] * w1 + note["flat"] * w2) / (w1 + w2)
|
||||
continue
|
||||
notes.append(note)
|
||||
if not notes:
|
||||
return {"silenzio": False, "note": [], "msg": "Ho sentito suono ma nessuna nota stabile: forse rumore o troppe corde insieme."}
|
||||
cents = np.array([x["cents"] for x in notes])
|
||||
durs = np.array([x["dur"] for x in notes])
|
||||
onsets = np.array([x["start"] for x in notes])
|
||||
ioi = np.diff(onsets) if len(onsets) > 2 else np.array([])
|
||||
sc = _scala(scala)
|
||||
fuori = [x for x in notes if sc is not None and x["midi"] % 12 not in sc]
|
||||
rep = {
|
||||
"silenzio": False, "n_note": len(notes), "durata": float(len(y) / sr),
|
||||
"intonazione_media_abs": float(np.mean(np.abs(cents))), "tendenza": float(np.mean(cents)),
|
||||
"note_intonate_pct": float(np.mean(np.abs(cents) <= 15) * 100),
|
||||
"peggiori": sorted(notes, key=lambda x: -abs(x["cents"]))[:3],
|
||||
"vibrato_note": int(sum(1 for x in notes if x["vib"] > 15 and x["dur"] > 0.35)),
|
||||
"ritmo_cv": float(np.std(ioi) / np.mean(ioi)) if len(ioi) >= 3 else None,
|
||||
"dinamica_range_db": float(np.max([x["db"] for x in notes]) - np.min([x["db"] for x in notes])),
|
||||
"graffio_pct": float(np.mean([x["flat"] > 0.25 for x in notes]) * 100),
|
||||
"note_corte_pct": float(np.mean(durs < 0.15) * 100),
|
||||
"fuori_scala": fuori, "scala": scala, "note": notes,
|
||||
}
|
||||
return rep
|
||||
|
||||
|
||||
def _testo(rep: dict) -> str:
|
||||
if rep.get("silenzio"):
|
||||
return "Non ho sentito il violino: il microfono ha registrato quasi silenzio. Controlla la distanza (ideale 1-2 metri) e riprova."
|
||||
if not rep.get("note"):
|
||||
return rep.get("msg", "Nessuna nota rilevata.")
|
||||
seq = " ".join(f"{_nome(x['midi'])}({x['cents']:+.0f})" for x in rep["note"][:40])
|
||||
out = [f"Registrazione di {rep['durata']:.0f} s, {rep['n_note']} note rilevate (nome e scarto in cent, + = crescente, - = calante):",
|
||||
seq + (" …" if rep["n_note"] > 40 else ""),
|
||||
f"Intonazione: scarto medio {rep['intonazione_media_abs']:.0f} cent, {rep['note_intonate_pct']:.0f}% delle note entro 15 cent, tendenza {'crescente' if rep['tendenza'] > 5 else 'calante' if rep['tendenza'] < -5 else 'centrata'} ({rep['tendenza']:+.0f} cent).",
|
||||
"Note più stonate: " + ", ".join(f"{_nome(x['midi'])} {x['cents']:+.0f} cent a {x['start']:.1f} s" for x in rep["peggiori"]) + "."]
|
||||
if rep["ritmo_cv"] is not None:
|
||||
out.append(f"Ritmo: regolarità {'buona' if rep['ritmo_cv'] < 0.25 else 'discreta' if rep['ritmo_cv'] < 0.45 else 'instabile'} (variazione {rep['ritmo_cv'] * 100:.0f}% tra una nota e l'altra).")
|
||||
rng = rep["dinamica_range_db"]
|
||||
giudizio = "molto uniforme" if rng < 6 else "espressiva" if rng < 18 else "sbalzi forti: pressione dell'arco irregolare?"
|
||||
out.append(f"Dinamica: escursione {rng:.0f} dB ({giudizio}).")
|
||||
out.append(f"Suono: {rep['graffio_pct']:.0f}% delle note con rumore d'arco evidente (graffio o arco troppo leggero/veloce); note troppo brevi {rep['note_corte_pct']:.0f}%; vibrato su {rep['vibrato_note']} note.")
|
||||
if rep["scala"]:
|
||||
if rep["fuori_scala"]:
|
||||
out.append(f"Scala {rep['scala']}: note fuori scala: " + ", ".join(f"{_nome(x['midi'])} a {x['start']:.1f} s" for x in rep["fuori_scala"][:6]) + " (dita in posizione sbagliata?).")
|
||||
else:
|
||||
out.append(f"Scala {rep['scala']}: tutte le note appartengono alla scala.")
|
||||
out.append("Interpreta questi dati per l'allieva: complimenti sinceri prima, poi al massimo due consigli concreti.")
|
||||
return "\n".join(out)
|
||||
|
||||
|
||||
def ascolta(params: dict, ctx: dict) -> str:
|
||||
secondi = max(5, min(int(params.get("secondi") or 15), 60))
|
||||
a4 = float(params.get("nota_la") or 440)
|
||||
y = registra(secondi, ctx)
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
path = OUT_DIR / f"{time.strftime('%Y%m%d-%H%M%S')}.wav"
|
||||
try:
|
||||
import soundfile as sf
|
||||
sf.write(str(path), y, SR)
|
||||
except Exception:
|
||||
pass
|
||||
rep = analizza(y, SR, a4, str(params.get("scala") or ""))
|
||||
return _testo(rep) + (f"\nRegistrazione salvata: {path.name}" if path.exists() else "")
|
||||
|
||||
|
||||
def accorda(params: dict, ctx: dict) -> str:
|
||||
import librosa
|
||||
y = registra(max(3, min(int(params.get("secondi") or 4), 10)), ctx, attesa=2)
|
||||
rms = float(np.sqrt(np.mean(y ** 2)))
|
||||
if 20 * math.log10(max(rms, 1e-6)) < -45:
|
||||
return "Non sento la corda: suonala a vuoto, con l'arco, vicino al microfono."
|
||||
f0, voiced, _ = librosa.pyin(y, fmin=150.0, fmax=1200.0, sr=SR)
|
||||
f = float(np.nanmedian(f0[voiced])) if np.any(voiced) else 0.0
|
||||
if not f:
|
||||
return "Non riesco a isolare la nota: suona una corda sola, a vuoto."
|
||||
a4 = float(params.get("nota_la") or 440)
|
||||
nome, target = min(CORDE.items(), key=lambda kv: abs(math.log2(f / (kv[1] * a4 / 440))))
|
||||
target *= a4 / 440
|
||||
cents = 1200 * math.log2(f / target)
|
||||
verdetto = "intonata" if abs(cents) <= 5 else ("crescente: allenta un po'" if cents > 0 else "calante: tira un po'")
|
||||
return f"Corda {nome}: {f:.1f} Hz, {cents:+.0f} cent rispetto a {target:.1f} Hz → {verdetto}. (Oltre ±30 cent usa i piroli, altrimenti i tiracantini.)"
|
||||
|
||||
|
||||
TOOLS = [
|
||||
{"name": "violino_ascolta", "description": "Ascolta il violino dal microfono per alcuni secondi (default 15, max 60) e misura intonazione di ogni nota in cent, tendenza crescente/calante, regolarità del ritmo, dinamica, rumore d'arco e vibrato. Con scala (es. 'Sol maggiore', 'Re minore') segnala le note fuori scala. Prima di chiamarlo di' all'allieva che stai per ascoltare: ci sono 3 secondi di preparazione. Restituisce dati da trasformare in consigli.",
|
||||
"parameters": {"type": "object", "properties": {"secondi": {"type": "integer"}, "scala": {"type": "string"}, "nota_la": {"type": "number", "description": "frequenza del La di riferimento, 440 o 442"}}}, "run": ascolta},
|
||||
{"name": "violino_accorda", "description": "Accordatore: ascolta una corda a vuoto (Sol, Re, La, Mi) e dice di quanti cent è crescente o calante e come correggere.",
|
||||
"parameters": {"type": "object", "properties": {"secondi": {"type": "integer"}, "nota_la": {"type": "number"}}}, "run": accorda},
|
||||
]
|
||||
Reference in new issue
Block a user