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 sirena estilo Proteccion Civil (wail). Usa solo math (sin numpy)."""
    import math, struct
    ruta = os.path.join(MEDIA_DIR, "pitido_emergencia.wav")
    if os.path.exists(ruta):
        return ruta
    try:
        os.makedirs(MEDIA_DIR, exist_ok=True)
        sr = 44100
        dur = config.GOOGLE_TV_PITIDO_SG
        n = int(sr * dur)
        periodo = 4.0
        nfade = min(int(0.3 * sr), n // 4)
        phase = 0.0
        samples = []
        for i in range(n):
            t = i / sr
            if config.GOOGLE_TV_PITIDO_ESTILO == "wail":
                f = (t % periodo) / periodo
                freq = 400 + 600 * (1 - abs(2 * f - 1))
            else:
                freq = 500 + 400 * abs(math.sin(2 * math.pi * 0.5 * t))
            phase += 2 * math.pi * freq / sr
            y = 0.8 * math.sin(phase)
            if i < nfade: y *= i / nfade
            elif i > n - nfade: y *= (n - i) / nfade
            samples.append(struct.pack("<h", max(-32768, min(32767, int(y * 32767)))))
        import wave as _wave
        with open(ruta, "wb") as f:
            with _wave.open(f, "w") as w:
                w.setnchannels(1); w.setsampwidth(2); w.setframerate(sr)
                w.writeframes(b"".join(samples))
    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 _desconectar(chromecasts):
    for cc in chromecasts:
        try:
            if cc.is_active and cc.is_connected:
                cc.disconnect()
        except Exception:
            pass


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 apagar_tv_remote():
                        print(f"[Voz] {c.cast_info.friendly_name} apagada")
                    else:
                        print(f"[Voz] {c.cast_info.friendly_name} no se pudo apagar")
                except Exception as e:
                    print(f"[Voz] Error en {c.cast_info.friendly_name}: {e}")
        finally:
            _desconectar(chromecasts)
            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)


def youtube_en_teles(url):
    """Reproduce un video de YouTube en la tele y la apaga al terminar."""
    import re
    match = re.search(r"(?:v=|youtu\.be/)([A-Za-z0-9_-]{11})", url)
    if not match:
        print("[Voz] URL de YouTube no válida")
        return
    video_id = match.group(1)
    try:
        import pychromecast
        from pychromecast.controllers.youtube import YouTubeController

        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
            for c in objetivos:
                try:
                    c.wait(timeout=10)
                    c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)
                    yt = YouTubeController()
                    c.register_handler(yt)
                    print(f"[Voz] Lanzando YouTube en {c.cast_info.friendly_name}: {video_id}")
                    yt.play_video(video_id)
                    # Esperar a que termine el video
                    time.sleep(5)
                    while True:
                        try:
                            state = c.media_controller.status.player_state
                            if state == "IDLE":
                                break
                        except Exception:
                            pass
                        time.sleep(3)
                    apagar_tv_remote()
                    print(f"[Voz] Video terminado, TV apagada")
                except Exception as e:
                    print(f"[Voz] Error en {c.cast_info.friendly_name}: {e}")
        finally:
            _desconectar(chromecasts)
            pychromecast.discovery.stop_discovery(browser)
    except Exception as e:
        print(f"[Voz] Error YouTube: {e}")


def apagar_tv_remote():
    """Apaga la TV via AndroidTVRemote (protocolo oficial Google)."""
    try:
        import asyncio
        from androidtvremote2 import AndroidTVRemote
        CERT = "/home/pi/minipc/tv_cert/cert.pem"
        KEY = "/home/pi/minipc/tv_cert/key.pem"
        TV_IP = "192.168.50.16"
        async def _apagar():
            r = AndroidTVRemote(
                client_name="MiniPC", certfile=CERT, keyfile=KEY, host=TV_IP)
            await r.async_connect()
            if r.is_on:
                r.send_key_command("POWER")
                return True
            return False
        return asyncio.run(_apagar())
    except Exception as e:
        print(f"[Voz] Error apagando TV: {e}")
        return False


def anuncio_simple(texto):
    """Mensaje rapido: enciende, reproduce el mensaje una vez y apaga."""
    if not config.GOOGLE_TV_ACTIVO:
        return
    url_mensaje = _url_audio(texto)
    if url_mensaje is None:
        return
    tipo = "audio/mpeg"
    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
            for c in objetivos:
                try:
                    c.wait(timeout=10)
                    c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)
                    _reproducir_y_esperar(
                        c, url_mensaje, tipo, texto[:80],
                        f"Mensaje simple en {c.cast_info.friendly_name}")
                    apagar_tv_remote()
                except Exception as e:
                    print(f"[Voz] Error en {c.cast_info.friendly_name}: {e}")
        finally:
            _desconectar(chromecasts)
            pychromecast.discovery.stop_discovery(browser)
    except Exception as e:
        print(f"[Voz] Error de casting: {e}")