#!/usr/bin/env python3 """Bucle de mejora que corre solo, ronda tras ronda, y se protege de empeorar. python3 ~/COFRE/CODERS/JARVIS/config/mejora_nocturna.py # hasta la mañana python3 ~/COFRE/CODERS/JARVIS/config/mejora_nocturna.py --horas 2 # un rato python3 ~/COFRE/CODERS/JARVIS/config/mejora_nocturna.py --ronda # una sola Que hace en cada ronda: 1. Dice en voz alta (Piper) las frases del catalogo, en varias voces, velocidades y con ruido y sala aleatorios, distintos cada ronda. 2. Las transcribe con el mismo Whisper del asistente. 3. Mira que porcentaje llega a su accion. Ese es el marcador de la ronda. 4. De lo que NO llega, se queda con los fallos SISTEMATICOS (repetidos) y SIN AMBIGUEDAD (que no puedan ser dos acciones), y los propone como alias. 5. GUARDA: antes de aplicar nada, comprueba en una copia que a) el acierto sube, y b) la conversacion normal SIGUE yendo al modelo (no se cuela ninguna). Solo si las dos cosas se cumplen, aplica los alias de verdad. 6. Copia de seguridad del catalogo antes de tocarlo, y todo al registro. Por que es seguro dejarlo sin mirar: nunca aplica algo que baje el acierto ni que haga que una frase de charla acabe ejecutando un comando. Si una ronda empeorara, revierte. Y cada cambio queda respaldado, asi que siempre se puede volver atras a mano. Que NO hace: reentrenar el modelo (no cabe en 4 GB de VRAM) ni tocar Whisper o Piper. Mejora la capa de emparejamiento —alias y umbrales—, que es donde estan casi todos los fallos y donde una mejora es inmediata, legible y reversible. En marcha se puede seguir con: tail -f ~/COFRE/CODERS/JARVIS/muestras/mejora/registro.log cat ~/COFRE/CODERS/JARVIS/muestras/mejora/marcador.json """ import argparse import importlib.util import json import os import random import shutil import subprocess import sys import tempfile import time from collections import defaultdict CASA = os.path.expanduser("~") JARVIS = os.path.join(CASA, "COFRE/CODERS/JARVIS") CONFIG = os.path.join(CASA, ".var/app/io.github.qwersyk.Newelle/config") APP = "io.github.qwersyk.Newelle//master" PIPER = os.path.join(JARVIS, "voz/piper/piper") MODELOS_VOZ = os.path.join(JARVIS, "voz/modelos") WHISPER = os.path.join(CONFIG, "models/whisper/whisper.cpp/build/bin/whisper-cli") MODELOS_STT = os.path.join(CONFIG, "models/whisper/whisper.cpp/models") CATALOGO = os.path.expanduser( "~/.local/share/flatpak/app/io.github.qwersyk.Newelle/x86_64/master/" "active/files/share/newelle/acciones/acciones.json") CATALOGO_FUENTE = os.path.join(JARVIS, "newelle/data/acciones/acciones.json") PROPIO = os.path.join(CONFIG, "jarvis-acciones.json") MOTOR = os.path.join(JARVIS, "newelle/src/utility/acciones.py") SALA = os.path.join(JARVIS, "muestras/mejora") REGISTRO = os.path.join(SALA, "registro.log") MARCADOR = os.path.join(SALA, "marcador.json") COPIAS = os.path.join(SALA, "copias") WAVS = os.path.join(CONFIG, "mejora_wavs") VOCES = ["es_ES-sharvard-medium", "es_ES-davefx-medium", "es_ES-carlfm-x_low"] # Cada ronda escoge una rejilla distinta de velocidades y ruidos, para que no # se estanque midiendo siempre lo mismo y siga sacando fallos raros. VELOCIDADES = [0.85, 0.92, 1.0, 1.08, 1.18] RUIDOS = [ "", "aecho=0.8:0.7:22:0.12", "volume=0.72,highpass=f=120", "aecho=0.6:0.5:35:0.2,volume=0.8", "highpass=f=150,lowpass=f=6500", # microfono cutre "volume=0.65", # lejos ] RELLENOS = ["notas", "el informe", "documentos", "la reunion de ayer", "gatos"] # Frases de charla que NUNCA deben acabar en una accion. Es el guardarrail: si # despues de aplicar alias alguna se cuela, la ronda se revierte. CHARLA = [ "cuentame un chiste sobre gatos", "que opinas del proyecto", "hola que tal estas", "gracias por todo", "no se que hacer con esto", "me parece que esto no funciona bien", "buenas tardes que tal", "cuentame algo interesante", "estoy cansado de programar", "que raro todo esto la verdad", "a ver si esto va mejor ahora", "pues no tengo ni idea de que decir", "vale perfecto muchas gracias", "oye una cosa que te queria preguntar", ] TOPE_ALIAS_POR_ACCION = 25 # que una accion no acumule cientos de alias UMBRAL_SISTEMATICO = 2 # veces que hay que fallar igual para creerselo DRY_PARA_PARAR = 6 # rondas seguidas sin nada nuevo -> converge def log(msg): linea = f"[{time.strftime('%H:%M:%S')}] {msg}" print(linea, flush=True) try: with open(REGISTRO, "a") as f: f.write(linea + "\n") except OSError: pass def disco_libre_gb() -> float: """GB libres en la particion del home. Guardarrail: aunque el bucle apenas ocupa 3 MB por ronda y los limpia, si algo mas llenara el disco durante la noche conviene parar en vez de dejar el sistema sin espacio.""" try: s = os.statvfs(CASA) return s.f_bavail * s.f_frsize / 1e9 except OSError: return 999.0 def carga_motor(): # se recarga cada ronda: el catalogo propio cambia y el motor debe verlo spec = importlib.util.spec_from_file_location("acciones", MOTOR) m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) return m def dentro(orden, limite=1200): return subprocess.run( ["flatpak", "run", "--command=bash", APP, "-c", orden], capture_output=True, text=True, timeout=limite) def genera(texto, voz, velocidad, ruido, destino): crudo = tempfile.mktemp(suffix=".wav") try: r = subprocess.run( [PIPER, "--model", os.path.join(MODELOS_VOZ, voz + ".onnx"), "--length_scale", str(1 / velocidad), "--output_file", crudo], input=texto, capture_output=True, text=True, timeout=60) if r.returncode != 0 or not os.path.exists(crudo): return False filtros = "aresample=16000" + ("," + ruido if ruido else "") subprocess.run(["ffmpeg", "-hide_banner", "-loglevel", "error", "-i", crudo, "-ac", "1", "-af", filtros, "-y", destino], check=True, timeout=60) return os.path.exists(destino) except (OSError, subprocess.SubprocessError): return False finally: if os.path.exists(crudo): os.remove(crudo) def transcribe(carpeta, ficheros, hilos=8, por_tanda=40): salida = {} for i in range(0, len(ficheros), por_tanda): tanda = ficheros[i:i + por_tanda] lista = " ".join(f'"{carpeta}/{n}"' for n in tanda) orden = (f'for f in {lista}; do echo "### $(basename "$f")"; ' f'{WHISPER} -m {MODELOS_STT}/ggml-base.bin -l es -nt -t {hilos} ' f'-f "$f" 2>/dev/null | tr "\\n" " "; echo; done') try: r = dentro(orden) except subprocess.SubprocessError: continue actual = None for linea in r.stdout.splitlines(): if linea.startswith("### "): actual = linea[4:].strip() elif actual: salida[actual] = linea.strip() actual = None return salida def catalogo_base(): ruta = CATALOGO if os.path.exists(CATALOGO) else CATALOGO_FUENTE with open(ruta) as f: return json.load(f)["acciones"], ruta def evalua(motor, textos, indice): """(aciertos, total, candidatos{id: {frase_mal: veces}}).""" aciertos = 0 candidatos = defaultdict(lambda: defaultdict(int)) for nombre, (ident, frase) in indice.items(): oido = textos.get(nombre, "").strip() if not oido: continue a, _ = _CAT.busca(oido) if a is not None and a.id == ident: aciertos += 1 else: candidatos[ident][motor.normaliza(oido)] += 1 return aciertos, len([1 for n in indice if textos.get(n)]), candidatos _CAT = None # Palabras de funcion: aparecen en cualquier frase y no distinguen una orden de # una conversacion. Un alias hecho solo de estas ("que me tengo") es una trampa. VACIAS = { "que", "de", "la", "el", "los", "las", "un", "una", "me", "te", "se", "lo", "mi", "tu", "su", "es", "esta", "estan", "hay", "por", "para", "con", "en", "y", "o", "a", "al", "del", "como", "cuanto", "cuanta", "muy", "ya", "he", "ha", "tengo", "tiene", "esto", "eso", "aqui", "ahi", "va", "van", "asi", } def contenido(frase: str) -> set: """Las palabras con significado: las que de verdad identifican la orden.""" return {p for p in frase.split() if len(p) > 3 and p not in VACIAS} def filtra_candidatos(motor, candidatos, acciones): """Solo los alias que valen: sistematicos, sin ambiguedad, y que conservan una palabra distintiva de la orden original. El guardarrail clave es el ultimo: una confusion del reconocedor que aun sirve de alias mantiene al menos una palabra con significado ("kernel", "wikipedia", "bateria"). Si las perdio todas y solo quedan palabras de relleno, es indistinguible de una frase de charla y se descarta, por mucho que se haya repetido. Es lo que separa "que kernel tengo" -> "que kernel me tengo" (vale) de "que me tengo" (no vale). """ porfrase = defaultdict(set) for ident, frases in candidatos.items(): for f in frases: porfrase[f].add(ident) porid = {a["id"]: a for a in acciones} propuestos = defaultdict(list) for ident, frases in candidatos.items(): accion = porid.get(ident) if accion is None: continue # Las de argumento no generan alias: el relleno de prueba ("notas") se # cuela en la transcripcion y quedaria horneado en el alias. Su # emparejamiento ya es por parecido del arranque, que las cubre. if accion.get("captura"): continue orig = motor.normaliza(accion["frases"][0] if accion["frases"] else "") claves = contenido(orig) for f, veces in frases.items(): if veces < UMBRAL_SISTEMATICO: continue if len(porfrase[f]) > 1: # la misma frase, dos acciones continue if len(f.split()) < 2: # una palabra engancha de todo continue if _CAT.busca(f)[0] is not None: # ya lo coge algo continue if motor.parecido(f, orig) < 0.6: # se perdio del todo: ruido continue # conserva una palabra distintiva? exacta o casi (el reconocedor # cambia una letra: "wikipedia" -> "wikipeda") palabras_f = set(f.split()) if claves and not any( c in palabras_f or any(motor.parecido(c, w) > 0.8 for w in palabras_f) for c in claves): continue propuestos[ident].append((f, veces)) return propuestos def aplica(propuestos, acciones_orig): """Escribe los alias en el catalogo propio. Devuelve cuantos.""" try: with open(PROPIO) as f: propio = json.load(f) except (OSError, json.JSONDecodeError): propio = {"version": 1, "acciones": []} porid = {a["id"]: a for a in propio.get("acciones", [])} original = {a["id"]: a for a in acciones_orig} n = 0 for ident, lista in propuestos.items(): entrada = porid.get(ident) if entrada is None: entrada = dict(original[ident]) porid[ident] = entrada propio.setdefault("acciones", []).append(entrada) if len(entrada["frases"]) >= TOPE_ALIAS_POR_ACCION: continue for f, _ in lista: if f not in entrada["frases"]: entrada["frases"].append(f) n += 1 with open(PROPIO, "w") as f: json.dump(propio, f, indent=2, ensure_ascii=False) return n def charla_se_cuela(motor): """True si alguna frase de charla acaba en una accion. El guardarrail.""" for frase in CHARLA: a, _ = _CAT.busca(frase) if a is not None: return frase, a.id return None def marcador_guarda(datos): try: with open(MARCADOR, "w") as f: json.dump(datos, f, indent=2, ensure_ascii=False) except OSError: pass def copia_catalogo(ronda): if not os.path.exists(PROPIO): return os.makedirs(COPIAS, exist_ok=True) shutil.copy2(PROPIO, os.path.join(COPIAS, f"ronda_{ronda:03d}.json")) def una_ronda(motor, ronda, muestras_por_acap): global _CAT acciones, _ = catalogo_base() # muestras de esta ronda: condiciones aleatorias, distintas cada vez trabajos = [] for a in acciones: if not a["frases"]: continue frase = a["frases"][0] if a.get("captura"): frase += " " + random.choice(RELLENOS) for _ in range(muestras_por_acap): trabajos.append((a["id"], frase, random.choice(VOCES), random.choice(VELOCIDADES), random.choice(RUIDOS))) random.shuffle(trabajos) shutil.rmtree(WAVS, ignore_errors=True) os.makedirs(WAVS, exist_ok=True) indice = {} for n, (ident, frase, voz, vel, ruido) in enumerate(trabajos, 1): nombre = f"{n:05d}.wav" if genera(frase, voz, vel, ruido, os.path.join(WAVS, nombre)): indice[nombre] = (ident, frase) log(f"ronda {ronda}: {len(indice)} frases dichas, transcribiendo...") textos = transcribe(WAVS, sorted(indice)) # medir ANTES de tocar nada _CAT = motor.Catalogo([p for p in (PROPIO, CATALOGO) if os.path.exists(p)] or [CATALOGO_FUENTE]) aciertos, total, candidatos = evalua(motor, textos, indice) antes = 100 * aciertos / max(1, total) log(f"ronda {ronda}: acierto {antes:.1f} % ({aciertos}/{total})") propuestos = filtra_candidatos(motor, candidatos, acciones) nuevos = sum(len(v) for v in propuestos.values()) if not nuevos: log(f"ronda {ronda}: sin alias nuevos que valgan") return antes, 0, False # aplicar en el catalogo propio, con copia de seguridad antes copia_catalogo(ronda) n = aplica(propuestos, acciones) # revalidar: recargar el catalogo con los alias puestos _CAT = motor.Catalogo([p for p in (PROPIO, CATALOGO) if os.path.exists(p)] or [CATALOGO_FUENTE]) # GUARDARRAIL 1: la charla no se cuela cuela = charla_se_cuela(motor) if cuela: frase, ident = cuela log(f"ronda {ronda}: REVERTIDA — '{frase}' se colaria en {ident}") shutil.copy2(os.path.join(COPIAS, f"ronda_{ronda:03d}.json"), PROPIO) return antes, 0, False # GUARDARRAIL 2: el acierto sube en las mismas muestras aciertos2, _, _ = evalua(motor, textos, indice) despues = 100 * aciertos2 / max(1, total) if despues < antes - 0.5: log(f"ronda {ronda}: REVERTIDA — el acierto bajaria a {despues:.1f} %") shutil.copy2(os.path.join(COPIAS, f"ronda_{ronda:03d}.json"), PROPIO) return antes, 0, False log(f"ronda {ronda}: +{n} alias · acierto {antes:.1f} -> {despues:.1f} %") for ident, lista in sorted(propuestos.items()): for f, v in lista[:2]: log(f" {ident} <- \"{f}\" ({v}x)") return despues, n, True def main(): ap = argparse.ArgumentParser() ap.add_argument("--horas", type=float, default=0, help="parar tras N horas") ap.add_argument("--ronda", action="store_true", help="una sola ronda") ap.add_argument("--muestras", type=int, default=3, help="muestras por accion y ronda") ap.add_argument("--dry", type=int, default=DRY_PARA_PARAR, help="rondas secas seguidas antes de parar") args = ap.parse_args() for ruta, que in ((PIPER, "piper"), (WHISPER, "whisper")): if not os.path.exists(ruta): print(f"falta {que}: {ruta}") return 1 os.makedirs(SALA, exist_ok=True) limite = time.time() + args.horas * 3600 if args.horas else None motor = carga_motor() log("=" * 56) log(f"arranca la mejora nocturna · {args.muestras} muestras/accion/ronda") if limite: log(f"parara en {args.horas} h") else: log("corre hasta converger o hasta que la pares") historial = [] dry = 0 ronda = 0 try: while True: ronda += 1 libre = disco_libre_gb() if libre < 10: log(f'PARADA: solo quedan {libre:.1f} GB de disco. Fin por seguridad.') break ini = time.time() try: score, nuevos, aplicado = una_ronda(motor, ronda, args.muestras) except Exception as e: log(f"ronda {ronda}: error, se salta — {type(e).__name__}: {e}") time.sleep(5) continue historial.append({"ronda": ronda, "acierto": round(score, 1), "alias_nuevos": nuevos, "minutos": round((time.time() - ini) / 60, 1)}) marcador_guarda({ "actualizado": time.strftime("%Y-%m-%d %H:%M"), "rondas": ronda, "ultimo_acierto": round(score, 1), "alias_totales": alias_totales(), "historial": historial[-40:], }) dry = dry + 1 if nuevos == 0 else 0 if args.ronda: break if dry >= args.dry: log(f"converge: {DRY_PARA_PARAR} rondas sin nada nuevo. Fin.") break if limite and time.time() > limite: log("se acabo el tiempo pedido. Fin.") break except KeyboardInterrupt: log("parada a mano") shutil.rmtree(WAVS, ignore_errors=True) log(f"terminado tras {ronda} rondas · acierto final " f"{historial[-1]['acierto'] if historial else '?'} % · " f"{alias_totales()} alias en el catalogo propio") log("los cambios estan en jarvis-acciones.json; se aplican al abrir JARVIS") return 0 def alias_totales(): try: with open(PROPIO) as f: d = json.load(f) base = {a["id"]: len(a["frases"]) for a in json.load(open(CATALOGO if os.path.exists(CATALOGO) else CATALOGO_FUENTE))["acciones"]} return sum(max(0, len(a["frases"]) - base.get(a["id"], 0)) for a in d.get("acciones", [])) except (OSError, json.JSONDecodeError): return 0 if __name__ == "__main__": sys.exit(main())