a0b9afe2f9
PARAR no pausaba de verdad: el HUD estaba entero para el estado "parado" pero el backend nunca mandaba el campo "parar" en el evento controles, asi que el boton no alternaba ni pausaba la escucha. Ahora pausa (el bucle de escucha descarta el audio en pausa) y alterna a "reanudar". Portabilidad (para que encaje en otros portatiles y torres, no solo en el equipo del autor): - microfono: pw-record / parec / arecord (PipeWire, PulseAudio o ALSA) - whisper: binario del build local o del PATH; modelos por JARVIS_WHISPER_MODELOS o busqueda en varios sitios; CUDA en LD_LIBRARY_PATH solo si existe - README: la tabla de portabilidad refleja audio y whisper
339 lines
14 KiB
Python
Executable file
339 lines
14 KiB
Python
Executable file
#!/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."""
|
||
manos = Manos(catalogo())
|
||
cara = Cara()
|
||
p = Panel(cara, boca, manos, cerebro, candado=not sin_candado,
|
||
centinela=centinela)
|
||
cara.al_pulsar = p.pulsado
|
||
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("--sin-candado", action="store_true",
|
||
help="no pedir la contraseña al abrir (solo para probar)")
|
||
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=a.sin_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())
|