#!/usr/bin/env python3 """JARVIS. Asistente de voz local, sin framework debajo. jarvis.py --di "Buenas noches, señor" prueba la boca jarvis.py --voces las que hay jarvis.py --accion "cuanto espacio queda" prueba las manos, sin voz Esto sustituye a la capa que corria dentro de Newelle. Lo que se trajo tal cual —el emparejador, las tareas, el sudo, el HUD— esta en `manos/` y `cara/`; lo que habia que reescribir porque era de Newelle esta en `oido/`, `boca/` y `cerebro/`. ## Por que fuera del flatpak No es dogma, es que el sandbox costaba mas de lo que daba: escondia `/tmp`, obligaba a `LD_LIBRARY_PATH` a mano para las libs de CUDA de whisper, metia un `flatpak-spawn --host` en cada accion, y cambiar una linea de Python pedia recompilar el paquete entero. Aqui un cambio es editar y reiniciar. """ import argparse import os import shutil import subprocess import sys import time AQUI = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, AQUI) from boca.voz import Boca, VOCES # noqa: E402 from cara.panel import Panel # noqa: E402 from cara.servidor import Cara # noqa: E402 from cerebro.ollama import Cerebro, MODELO # noqa: E402 from manos import Manos # noqa: E402 from manos.diario import anota # noqa: E402 from manos.acciones import Catalogo # noqa: E402 from oido.captura import Captura, fuentes # noqa: E402 from oido.despierta import Centinela, NOMBRE # noqa: E402 from oido.whisper import Oido, prompt_dominio, es_ruido # noqa: E402 CATALOGO = os.path.join(AQUI, "datos", "acciones.json") PROPIO = os.path.expanduser("~/.config/jarvis/acciones.json") def catalogo() -> Catalogo: # El del usuario primero: puede redefinir cualquiera del paquete. Mismo # orden que tenia el panel de Newelle. return Catalogo([PROPIO, CATALOGO]) def atiende(texto, manos: Manos, cerebro: Cerebro | None, traza=print): """La cadena entera para una frase: catalogo primero, cerebro despues. El orden no es un detalle de eficiencia, es lo que hace usable esto. Las 150 acciones se emparejan sin tocar ningun modelo y contestan al instante; el cerebro solo ve lo que no encaja. Por eso un modelo lento se soporta: la mayoria de las ordenes ni pasan por el. """ t0 = time.time() accion, argumento = manos.busca(texto) if accion is not None: traza(f" catalogo: {accion.id}" + (f" <- {argumento}" if argumento else "")) dicho, _ = manos.haz(accion, argumento) anota(texto, "catalogo", accion.id, argumento, time.time() - t0, dicho) return dicho if cerebro is None: traza(" (sin cerebro conectado)") anota(texto, "sin-cerebro", seg=time.time() - t0) return "" traza(" al cerebro...") dicho = cerebro.responde(texto, al_ejecutar=lambda i, a: traza( f" usa {i}" + (f"({a[:52]})" if a else ""))) # La via mas valiosa del diario: lo que el catalogo NO cubre es justo la # lista de acciones que faltan por escribir. anota(texto, "cerebro", seg=time.time() - t0, dicho=dicho) return dicho def manda_accion(frase: str, boca: Boca | None = None, cerebro=None) -> int: manos = Manos(catalogo()) dicho = atiende(frase, manos, cerebro, traza=lambda t: print(t)) if not dicho: print(f' sin accion para "{frase}"') return 1 print(f" {dicho}") if boca: boca.di(dicho) return 0 def escucha(boca: Boca, modelo="small", fuente=None, callado=False, cerebro=None, panel=None, centinela=None) -> int: """El bucle de verdad: oir, entender, hacer, contestar. Es la version desnuda de lo que hacia el panel, sin GTK y sin HUD todavia. Sirve para comprobar que la cadena entera funciona antes de ponerle cara. """ manos = Manos(catalogo()) # Con el prompt del NOMBRE, no con el del catalogo entero: sin el, Whisper # nunca escribe "JARVIS" y el centinela no despierta jamas (ver whisper.py). oreja = Oido(modelo=modelo) if not oreja.disponible(): print(f" no encuentro whisper o el modelo {modelo}") return 1 print(f" arrancando whisper ({modelo}, beam {oreja.beam}, sns, nombre)...") if not oreja.arranca(): print(" whisper no arranco") return 1 aviso = (f"esperando a que le llamen: «{NOMBRE.upper()}»" if centinela and centinela.activo else "atendiendo todo lo que oiga (--palabra para exigir el nombre)") print(f" escuchando por {fuente or 'la entrada por defecto'}, {aviso}.") print(" Ctrl+C para parar.\n") if panel: panel.estado("dormido" if centinela and centinela.activo else "escuchando") # La captura consulta a la boca en cada trozo: mientras JARVIS habla, ese # audio no se mira siquiera. Es lo que evita que se oiga a si mismo. oreja_mic = Captura(boca=boca, fuente=fuente) try: for wav in oreja_mic.frases(): # En pausa (boton PARAR): se descarta el audio sin transcribir. El # microfono sigue abierto pero JARVIS no atiende nada hasta que se # pulse "reanudar". Se comprueba antes de whisper para no gastar GPU. if panel and getattr(panel, "parado", False): try: os.unlink(wav) except OSError: pass continue try: texto = oreja.transcribe(wav) finally: try: os.unlink(wav) except OSError: pass if es_ruido(texto): if texto: print(f" · ruido descartado: {texto!r}") anota(texto, "ruido") if panel: panel.dicho(texto, atendido=False, motivo="ruido") continue # ¿Hay una pregunta en el aire? Entonces esto es la respuesta. # # Va antes que TODO —antes de la palabra de activacion y antes del # emparejador— porque si JARVIS acaba de preguntarte que fichero # buscas, obligarte a decir "JARVIS, informes" seria absurdo: ya # esta hablando contigo. if panel and panel.esperando(): print(f" » {texto} (respondiendo a lo preguntado)") anota(texto, "respuesta") panel.dicho(texto) if panel.responde(texto): continue # La palabra de activacion, antes de nada: sin esto obedece a # cualquiera que pase, y a la tele. if centinela: atender, texto_orden, recien = centinela.filtra(texto) if not atender: if recien: print(f" » {texto} (le han llamado)") if panel: panel.dicho(texto) panel.estado("escuchando") # ya despierto boca.di_si_libre("¿Señor?") else: print(f" · {texto} (no le han llamado)") anota(texto, "sin-nombre") # Se enseña IGUAL, marcado: que se vea que te ha oido # bien y que lo unico que falta es llamarle. if panel: panel.dicho(texto, atendido=False, motivo=f"diga «{NOMBRE.upper()}» primero") continue texto = texto_orden # El candado tapa el panel, pero lo que hay que impedir es que # OBEDEZCA: si no, quien pase por delante da ordenes hablando, que # es justo lo que el candado existe para evitar. if panel and getattr(panel, "candado", False) and not panel.desbloqueado: print(f" · {texto} (bloqueado)") anota(texto, "bloqueado") panel.dicho(texto, atendido=False, motivo="hace falta la contraseña") continue print(f" » {texto}") if panel: panel.dicho(texto) panel.estado("pensando") dicho = atiende(texto, manos, cerebro) if not dicho: if panel: panel.estado("escuchando") continue print(f" {dicho}") if panel: panel.respuesta(dicho) panel.estado("hablando") if not callado: boca.di(dicho) if panel: # Sigue despierto un rato tras atender, para poder encadenar sin # repetir el nombre: el rotulo tiene que decir eso y no "dormido". panel.estado("escuchando" if not centinela or centinela.despierto else "dormido") except KeyboardInterrupt: print("\n parado") finally: oreja_mic.para() oreja.para() return 0 def panel(boca: Boca, cerebro=None, escuchando=False, modelo="small", fuente=None, abre=True, centinela=None, sin_candado=False) -> int: """Levanta el panel y, si se pide, ademas escucha el microfono.""" # Un SOLO Manos, compartido con el cerebro: asi hay una unica sesion de root # y un unico sitio donde pedir la contraseña. Antes el cerebro tenia su Manos # y el panel otro, con roots separados: desbloquear en uno no abria el otro. manos = cerebro.manos if (cerebro and getattr(cerebro, "manos", None)) else Manos(catalogo()) cara = Cara() p = Panel(cara, boca, manos, cerebro, candado=not sin_candado, centinela=centinela) cara.al_pulsar = p.pulsado cara.al_cerrar = p.cerrar # cerrar la ultima pestaña apaga JARVIS url = cara.arranca() if not url: return 1 p.arranca() print(f" panel en {url}") if abre: # Una sola forma de abrir la ventana, en ventana.sh, para que sea igual # se llegue por donde se llegue. Tener aqui su propia lista de # navegadores fue lo que dejo que arranca.sh abriera una pestaña de # Firefox cuando JARVIS ya estaba en marcha. abridor = os.path.join(AQUI, "ventana.sh") if os.access(abridor, os.X_OK): subprocess.Popen([abridor, url], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True) if not escuchando: print(" sin microfono (--escucha para que ademas oiga). Ctrl+C para parar.") try: while True: time.sleep(1) except KeyboardInterrupt: print("\n parado") p.para() return 0 return escucha(boca, modelo, fuente, False, cerebro, p, centinela) def main() -> int: p = argparse.ArgumentParser(description="JARVIS") p.add_argument("--di", metavar="TEXTO", help="decir algo en voz alta") p.add_argument("--voz", default=None, help="con que voz") p.add_argument("--voces", action="store_true", help="listar las voces") p.add_argument("--accion", metavar="FRASE", help="ejecutar una orden del catalogo") p.add_argument("--callado", action="store_true", help="no hablar el resultado") p.add_argument("--escucha", action="store_true", help="escuchar el microfono") p.add_argument("--modelo", default="small", help="modelo de whisper") p.add_argument("--fuente", default=None, help="entrada de audio (ver --fuentes)") p.add_argument("--fuentes", action="store_true", help="listar las entradas") p.add_argument("--cerebro", default=MODELO, help="modelo de ollama; 'no' para trabajar solo con el catalogo") p.add_argument("--ram", action="store_true", help="el modelo entero en RAM, sin usar la grafica") p.add_argument("--panel", action="store_true", help="abrir el panel") p.add_argument("--sin-abrir", action="store_true", help="levantar el panel pero no abrir el navegador") p.add_argument("--palabra", action="store_true", help=f"exigir que le llamen «{NOMBRE}» antes de cada orden") p.add_argument("--candado", action="store_true", help="pedir la contraseña al abrir y bloquear el panel. Por " "defecto NO: no molesta al arrancar, y el root se pide " "solo la primera vez que una accion lo necesita") a = p.parse_args() boca = Boca(a.voz or "davefx") cerebro = None if a.cerebro and a.cerebro != "no": cerebro = Cerebro(catalogo(), modelo=a.cerebro, en_ram=a.ram, manos=Manos(catalogo())) if a.voces: hay = boca.disponibles() for nombre in VOCES: marca = "·" if nombre in hay else "×" actual = " <- puesta" if nombre == boca.voz else "" print(f" {marca} {nombre}{actual}") return 0 if a.di: if a.voz and not boca.cambia(a.voz): print(f" no existe la voz {a.voz}") return 1 print(f" [{boca.voz}] {a.di}") if not boca.di(a.di): print(" no sono. mira que el modelo de la voz este en su sitio") return 1 return 0 if a.fuentes: for f in fuentes(): print(f" {f}") return 0 # La palabra de activacion va APAGADA por defecto. # # Decirla en cada frase es un peaje, y medido no evitaba nada: sobre 412 # frases reales oidas —con todo el eco y el ruido ambiente dentro— solo 3 # habrian disparado una accion, y las tres eran ordenes de verdad. Cero # falsos positivos del ruido, porque el emparejador ya es conservador. # # El candado cubre lo otro, que es quien puede darle ordenes. # # Se queda disponible con --palabra y en el boton del panel, por si algun # dia hay una tele cerca y el ruido empieza a colarse. centinela = Centinela(activo=a.palabra) if a.panel: return panel(boca, cerebro, a.escucha, a.modelo, a.fuente, abre=not a.sin_abrir, centinela=centinela, sin_candado=not a.candado) if a.escucha: return escucha(boca, a.modelo, a.fuente, a.callado, cerebro, centinela=centinela) if a.accion: return manda_accion(a.accion, None if a.callado else boca, cerebro) p.print_help() return 0 if __name__ == "__main__": sys.exit(main())