JARVIS/nucleo/pruebas/prueba_eco.py

#!/usr/bin/env python3
"""La prueba que Newelle no pasa: que JARVIS no se oiga a si mismo.

En Newelle, mas del 21 % de lo que oye y no entiende es su propia voz. Esta
prueba lo comprueba de verdad, con el microfono y los altavoces, en dos partes:

  A. Suena una frase POR FUERA de JARVIS (un reproductor a pelo, como si
     hablara una persona). Tiene que captarla. Si no, el microfono no llega a
     los altavoces y la parte B no demuestra nada.

  B. La misma frase, pero dicha POR JARVIS. NO tiene que captarla.

Las dos partes juntas son la prueba: A sola no dice nada, B sola podria pasar
simplemente porque el microfono no oye nada.

    ./venv/bin/python pruebas/prueba_eco.py

Necesita altavoces encendidos y a un volumen normal. Si estan en silencio, la
parte A falla y avisa.
"""
import os
import subprocess
import sys
import threading
import time

AQUI = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, AQUI)

from boca.voz import Boca                        # noqa: E402
from oido.captura import Captura                 # noqa: E402
from oido.whisper import Oido, es_ruido          # noqa: E402

FRASE = "El sistema esta operativo y a su disposicion, señor."
ESPERA = 14        # cuanto se escucha en cada parte


def _se_parece(oido: str, dicho: str) -> bool:
    """Si lo transcrito es la frase que se dijo, aunque venga maltratada.

    Se compara por palabras y no por parecido de cadena: un eco recortado
    ("operativo y a su disposicion") no se parece a la frase entera como
    cadena, pero comparte casi todas sus palabras, y eso es inequivoco.
    """
    import unicodedata

    def palabras(t):
        t = unicodedata.normalize("NFD", t.lower())
        t = "".join(c for c in t if unicodedata.category(c) != "Mn")
        return {p.strip(".,¡!¿?") for p in t.split() if len(p) > 3}

    a, b = palabras(oido), palabras(dicho)
    return bool(b) and len(a & b) >= 2


def escucha_un_rato(captura, oreja, segundos):
    """Devuelve lo que se haya oido en ese rato."""
    oido = []
    fin = time.monotonic() + segundos
    hilo_paro = threading.Event()

    def bucle():
        for wav in captura.frases(para_cuando=lambda: hilo_paro.is_set()):
            try:
                t = oreja.transcribe(wav)
            finally:
                try:
                    os.unlink(wav)
                except OSError:
                    pass
            if t and not es_ruido(t):
                oido.append(t)

    h = threading.Thread(target=bucle, daemon=True)
    h.start()
    while time.monotonic() < fin:
        time.sleep(0.2)
    hilo_paro.set()
    captura.para()
    h.join(timeout=5)
    return oido


def main() -> int:
    boca = Boca("davefx")
    oreja = Oido(modelo="small")
    if not oreja.disponible():
        print("  no esta whisper o el modelo; no se puede probar")
        return 2
    print("  arrancando whisper...")
    if not oreja.arranca():
        print("  whisper no arranco")
        return 2

    # Se genera el WAV una vez para poder reproducirlo por los dos caminos con
    # exactamente el mismo audio.
    wav = boca._genera(FRASE)
    if not wav:
        print("  no se pudo generar la voz")
        return 2

    fallos = 0

    # --- A: suena por fuera, tiene que oirse -----------------------------
    print("\n  A · suena por fuera (como si hablara alguien)")
    captura = Captura(boca=boca)
    def suelta_fuera():
        time.sleep(2.5)
        subprocess.run(["paplay", wav], capture_output=True)
    threading.Thread(target=suelta_fuera, daemon=True).start()
    oido_a = escucha_un_rato(captura, oreja, ESPERA)
    for t in oido_a:
        print(f"      oido: {t!r}")
    if oido_a:
        print("    OK  el microfono llega a los altavoces")
    else:
        print("    MAL no se oyo nada: ¿altavoces en silencio? sin esto la")
        print("        parte B no demuestra nada")
        fallos += 1

    time.sleep(1.5)

    # --- B: lo dice JARVIS, NO tiene que oirse ---------------------------
    print("\n  B · lo dice JARVIS (la guarda tiene que taparlo)")
    captura = Captura(boca=boca)
    def suelta_jarvis():
        time.sleep(2.5)
        boca.di(FRASE)
    threading.Thread(target=suelta_jarvis, daemon=True).start()
    oido_b = escucha_un_rato(captura, oreja, ESPERA)

    # Lo que se juzga es si se colo LA FRASE, no si se transcribio algo.
    #
    # Esto costo una conclusion equivocada: probando con colas de 350, 700 y
    # 1100 ms el resultado no era monotono —1100 fallaba y 700 no—, lo cual es
    # imposible si el unico factor fuera el eco. Mirando lo que se colaba, eran
    # "¡Adios!" y un parrafo inventado: Whisper ALUCINANDO sobre el silencio,
    # que es otro problema distinto y da igual lo larga que sea la guarda.
    eco = [t for t in oido_b if _se_parece(t, FRASE)]
    alucinado = [t for t in oido_b if t not in eco]

    for t in eco:
        print(f"      ECO: {t!r}")
    for t in alucinado:
        print(f"      (whisper alucinando sobre el silencio: {t[:52]!r})")

    if not eco:
        print("    OK  JARVIS no se oye a si mismo")
    else:
        print("    MAL se ha oido hablar: la guarda no tapa")
        fallos += 1

    oreja.para()
    print(f"\n  {'todo en orden' if not fallos else str(fallos) + ' fallan'}")
    return 1 if fallos else 0


if __name__ == "__main__":
    sys.exit(main())