biblioteca_conocimiento_laboratorio/vigilancia/voz.py

import asyncio
import os
import socket
import subprocess
import sys
import time

import config

MEDIA_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "voz_media")
_server = None


def _python_firewall_ok():
    """Python del sistema (con regla de firewall Allow) para servir el mp3."""
    if hasattr(sys, "base_prefix") and sys.base_prefix != sys.prefix:
        ruta = os.path.join(sys.base_prefix, "python.exe")
        if os.path.exists(ruta):
            return ruta
    return sys.executable


def _ip_lan():
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80))
        return s.getsockname()[0]
    finally:
        s.close()


def sintetizar(texto, nombre="anuncio.mp3"):
    os.makedirs(MEDIA_DIR, exist_ok=True)
    ruta = os.path.join(MEDIA_DIR, nombre)
    try:
        import edge_tts

        async def _gen():
            comunicador = edge_tts.Communicate(
                texto,
                config.VOZ_EDGE,
                rate=config.VOZ_RATE,
                pitch=config.VOZ_PITCH,
            )
            await comunicador.save(ruta)

        asyncio.run(_gen())
        return ruta
    except Exception as e:
        print(f"[Voz] Error sintetizando: {e}")
        return None


def _arrancar_servidor():
    global _server
    if _server is not None and _server.poll() is None:
        return
    try:
        py = _python_firewall_ok()
        _server = subprocess.Popen(
            [py, "-m", "http.server", str(config.GOOGLE_TV_PUERTO),
             "--directory", MEDIA_DIR],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
        time.sleep(1)
    except Exception as e:
        print(f"[Voz] No se pudo arrancar el servidor HTTP: {e}")


def _descubrir():
    import pychromecast

    chromecasts, browser = pychromecast.get_chromecasts(timeout=8)
    try:
        if config.GOOGLE_TV_NOMBRES:
            return [c for c in chromecasts
                    if c.cast_info.friendly_name in config.GOOGLE_TV_NOMBRES]
        return chromecasts
    finally:
        pychromecast.discovery.stop_discovery(browser)


def _url_video(texto):
    """Genera el video teleprompter (texto que se compone mientras habla)."""
    try:
        from teleprompter import generar_video, sintetizar_con_oraciones
        ruta_audio, oraciones = sintetizar_con_oraciones(texto)
        if ruta_audio is None or not oraciones:
            return None, None
        ruta_video = generar_video(texto, ruta_audio, oraciones)
        _arrancar_servidor()
        if _server is None:
            return None, None
        return f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta_video)}", "video/mp4"
    except Exception as e:
        print(f"[Voz] No se pudo generar el video: {e}")
        return None, None


def _url_texto_final(texto):
    """Video estatico con el texto completo (al final del bucle)."""
    if config.GOOGLE_TV_TEXTO_FINAL_SG <= 0:
        return None
    try:
        from teleprompter import generar_video_texto_final
        ruta = generar_video_texto_final(texto, config.GOOGLE_TV_TEXTO_FINAL_SG)
        _arrancar_servidor()
        if _server is None:
            return None
        return f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta)}"
    except Exception as e:
        print(f"[Voz] No se pudo generar el texto final: {e}")
        return None


def _url_audio(texto, nombre="anuncio.mp3"):
    ruta = sintetizar(texto, nombre=nombre)
    if ruta is None:
        return None
    _arrancar_servidor()
    if _server is None:
        return None
    return f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta)}"


def _esperar_reproduccion(cast, url, timeout=20):
    """Espera a que NUESTRA media empiece (PLAYING/BUFFERING) por content_id."""
    mc = cast.media_controller
    t0 = time.time()
    while time.time() - t0 < timeout:
        try:
            st = mc.status
            if st.content_id == url and st.player_state in ("PLAYING", "BUFFERING"):
                return True
        except Exception:
            pass
        time.sleep(1)
    return False


def _esperar_fin(cast, url, timeout=120):
    """Espera a que NUESTRA media termine (IDLE + FINISHED) por content_id."""
    mc = cast.media_controller
    t0 = time.time()
    while time.time() - t0 < timeout:
        try:
            st = mc.status
            if st.content_id == url and st.player_state == "IDLE" and st.idle_reason == "FINISHED":
                return True
        except Exception:
            pass
        time.sleep(1)
    return False


def _generar_pitido():
    """Genera una sirena de emergencia (WAV) si no existe."""
    ruta = os.path.join(MEDIA_DIR, "pitido_emergencia.wav")
    if os.path.exists(ruta):
        return ruta
    try:
        import numpy as np
        import wave as _wave

        os.makedirs(MEDIA_DIR, exist_ok=True)
        sr = 44100
        dur = config.GOOGLE_TV_PITIDO_SG
        t = np.linspace(0, dur, int(sr * dur), endpoint=False)
        freq = 500 + 400 * np.abs(np.sin(2 * np.pi * 0.5 * t))
        fase = 2 * np.pi * np.cumsum(freq) / sr
        y = 0.7 * np.sin(fase)
        fade = min(0.2, dur / 4)
        n_fade = int(fade * sr)
        y[:n_fade] *= np.linspace(0, 1, n_fade)
        y[-n_fade:] *= np.linspace(1, 0, n_fade)
        data = (y * 32767).astype(np.int16)
        with _wave.open(ruta, "w") as w:
            w.setnchannels(1)
            w.setsampwidth(2)
            w.setframerate(sr)
            w.writeframes(data.tobytes())
    except Exception as e:
        print(f"[Voz] No se pudo generar el pitido: {e}")
    return ruta


def _generar_silencio():
    """Genera un WAV silencioso corto para encender la tele sin sonido."""
    ruta = os.path.join(MEDIA_DIR, "silencio.wav")
    if os.path.exists(ruta):
        return ruta
    try:
        import wave as _wave

        os.makedirs(MEDIA_DIR, exist_ok=True)
        with _wave.open(ruta, "w") as w:
            w.setnchannels(1)
            w.setsampwidth(2)
            w.setframerate(8000)
            w.writeframes(b"\x00\x00" * (8000 * 2))
    except Exception as e:
        print(f"[Voz] No se pudo generar el silencio: {e}")
    return ruta


def _url_pitido():
    ruta = _generar_pitido()
    if not os.path.exists(ruta):
        return None
    _arrancar_servidor()
    if _server is None:
        return None
    return f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta)}"


def _url_silencio():
    ruta = _generar_silencio()
    if not os.path.exists(ruta):
        return None
    _arrancar_servidor()
    if _server is None:
        return None
    return f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta)}"


def _reproducir_y_esperar(c, url, content_type, titulo, desc):
    c.media_controller.play_media(url, content_type, title=titulo)
    print(f"[Voz] {desc}")
    _esperar_reproduccion(c, url)
    return _esperar_fin(c, url)


def anuncio_en_teles(texto):
    """Enciende la tele (vol configurado), espera, sirena de emergencia,
    mensaje y apaga por ADB."""
    if not config.GOOGLE_TV_ACTIVO:
        return
    url_mensaje, tipo_mensaje = _url_video(texto)
    if url_mensaje is None:
        url_mensaje = _url_audio(texto)
        tipo_mensaje = "audio/mpeg"
    if url_mensaje is None:
        return
    url_pitido = _url_pitido() if config.GOOGLE_TV_PITIDO_ACTIVO else None
    url_silencio = _url_silencio() if url_pitido else None
    url_final = _url_texto_final(texto)

    try:
        import pychromecast

        chromecasts, browser = pychromecast.get_chromecasts(timeout=8)
        try:
            if config.GOOGLE_TV_NOMBRES:
                objetivos = [c for c in chromecasts
                             if c.cast_info.friendly_name in config.GOOGLE_TV_NOMBRES]
            else:
                objetivos = chromecasts

            if not objetivos:
                print("[Voz] No hay teles Google detectadas")
                return

            for c in objetivos:
                try:
                    c.wait(timeout=10)
                    c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)

                    if url_silencio:
                        _reproducir_y_esperar(
                            c, url_silencio, "audio/wav", "inicio",
                            f"Tele encendida ({c.cast_info.friendly_name})")

                    if config.GOOGLE_TV_ESPERA_SG > 0:
                        print(f"[Voz] Esperando {config.GOOGLE_TV_ESPERA_SG}s...")
                        time.sleep(config.GOOGLE_TV_ESPERA_SG)

                    if url_pitido:
                        _reproducir_y_esperar(
                            c, url_pitido, "audio/wav", "pitido",
                            f"Pitido de emergencia en {c.cast_info.friendly_name}")

                    rep = max(1, config.GOOGLE_TV_REPETICIONES)
                    for n in range(rep):
                        _reproducir_y_esperar(
                            c, url_mensaje, tipo_mensaje, texto[:80],
                            f"Anuncio {n + 1}/{rep} en {c.cast_info.friendly_name} "
                            f"(vol {config.GOOGLE_TV_VOLUMEN})")

                    if url_final:
                        _reproducir_y_esperar(
                            c, url_final, "video/mp4", "texto final",
                            f"Texto final {config.GOOGLE_TV_TEXTO_FINAL_SG}s en "
                            f"{c.cast_info.friendly_name}")

                    if config.GOOGLE_TV_APAGAR_VIA_ADB:
                        from tv_adb import apagar_tv
                        apagar_tv()
                    elif config.ALEXA_APAGAR_TV_MONKEY:
                        print("[Voz] Apagando la tele via Alexa...")
                        from notificador import disparar_monkey_alexa
                        disparar_monkey_alexa(config.ALEXA_APAGAR_TV_MONKEY)
                    if config.GOOGLE_TV_APAGAR:
                        c.quit_app()
                        print(f"[Voz] {c.cast_info.friendly_name} apagada (standby)")
                except Exception as e:
                    print(f"[Voz] Error en {c.cast_info.friendly_name}: {e}")
        finally:
            pychromecast.discovery.stop_discovery(browser)
    except Exception as e:
        print(f"[Voz] Error de casting: {e}")


def reproducir_en_teles(texto):
    anuncio_en_teles(texto)