02-ago: [proyecto] vigilancia - anuncio atemorizante en teles Google: voz grave (Alvaro pitch-15Hz), volumen 30, mensaje configurable, apagado automatico al terminar

vigilancia/.env.example

@@ -42,7 +42,15 @@ DISCORD_IDS=
 # Teles Google (Chromecast/Google TV) que hablan el aviso.
 # GOOGLE_TV_ACTIVO=1/0 (siempre por defecto activo).
 # GOOGLE_TV_NOMBRES: nombres de las teles separados por coma (vacio = todas las detectadas).
+# GOOGLE_TV_VOLUMEN: volumen al que se enciende (0-100). GOOGLE_TV_APAGAR=1 apaga tras el anuncio.
+# VOZ_EDGE/VOZ_PITCH/VOZ_RATE: voz (Alvaro=grave), tono y velocidad.
 GOOGLE_TV_ACTIVO=1
 GOOGLE_TV_NOMBRES=
-VOZ_EDGE=es-ES-ElviraNeural
+GOOGLE_TV_VOLUMEN=30
+GOOGLE_TV_APAGAR=1
+VOZ_EDGE=es-ES-AlvaroNeural
+VOZ_PITCH=-15Hz
+VOZ_RATE=-10%
 GOOGLE_TV_PUERTO=8500
+VOZ_MENSAJE_ALERTA=Por favor, os avisamos de que no podéis cruzar este límite. Tenéis que ir a vuestro cuarto.
+VOZ_MENSAJE_CONOCIDO={nombres}, ¿qué estás haciendo? No entres ahí.

vigilancia/config.py

@@ -46,5 +46,18 @@ GOOGLE_TV_ACTIVO = os.getenv("GOOGLE_TV_ACTIVO", "1") == "1"
 GOOGLE_TV_NOMBRES = {
     n.strip() for n in os.getenv("GOOGLE_TV_NOMBRES", "").split(",") if n.strip()
 }
-VOZ_EDGE = os.getenv("VOZ_EDGE", "es-ES-ElviraNeural")
+GOOGLE_TV_VOLUMEN = int(os.getenv("GOOGLE_TV_VOLUMEN", "30"))
+GOOGLE_TV_APAGAR = os.getenv("GOOGLE_TV_APAGAR", "1") == "1"
+VOZ_EDGE = os.getenv("VOZ_EDGE", "es-ES-AlvaroNeural")
+VOZ_PITCH = os.getenv("VOZ_PITCH", "-15Hz")
+VOZ_RATE = os.getenv("VOZ_RATE", "-10%")
 GOOGLE_TV_PUERTO = int(os.getenv("GOOGLE_TV_PUERTO", "8500"))
+VOZ_MENSAJE_ALERTA = os.getenv(
+    "VOZ_MENSAJE_ALERTA",
+    "Por favor, os avisamos de que no podéis cruzar este límite. "
+    "Tenéis que ir a vuestro cuarto.",
+)
+VOZ_MENSAJE_CONOCIDO = os.getenv(
+    "VOZ_MENSAJE_CONOCIDO",
+    "{nombres}, ¿qué estás haciendo? No entres ahí.",
+)

vigilancia/main.py

@@ -39,14 +39,16 @@ def notificar_discord(resultados, cam_nombre):
 def texto_voz(resultados, cam_nombre):
     conocidas = {r["nombre"] for r in resultados if "nombre" in r}
     if conocidas:
-        return f"{', '.join(sorted(conocidas))}, ¿qué estás haciendo? No entres ahí."
-    return f"Presencia detectada en {cam_nombre}."
+        return config.VOZ_MENSAJE_CONOCIDO.format(
+            nombres=", ".join(sorted(conocidas)), camara=cam_nombre
+        )
+    return config.VOZ_MENSAJE_ALERTA.format(camara=cam_nombre)
 
 
 def hablar_en_teles(mensaje):
     try:
-        from voz import reproducir_en_teles
-        reproducir_en_teles(mensaje)
+        from voz import anuncio_en_teles
+        anuncio_en_teles(mensaje)
     except Exception as e:
         print(f"[TV] Error: {e}")
 

vigilancia/voz.py

@@ -2,6 +2,7 @@ import asyncio
 import os
 import socket
 import threading
+import time
 from http.server import HTTPServer, SimpleHTTPRequestHandler
 
 import config
@@ -19,14 +20,19 @@ def _ip_lan():
         s.close()
 
 
-def sintetizar(texto, nombre="alerta.mp3"):
+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)
+            comunicador = edge_tts.Communicate(
+                texto,
+                config.VOZ_EDGE,
+                rate=config.VOZ_RATE,
+                pitch=config.VOZ_PITCH,
+            )
             await comunicador.save(ruta)
 
         asyncio.run(_gen())
@@ -55,17 +61,51 @@ def _arrancar_servidor():
         print(f"[Voz] No se pudo arrancar el servidor HTTP: {e}")
 
 
-def reproducir_en_teles(texto):
-    if not config.GOOGLE_TV_ACTIVO:
-        return
-    ruta = sintetizar(texto)
+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_audio(texto, nombre="anuncio.mp3"):
+    ruta = sintetizar(texto, nombre=nombre)
     if ruta is None:
-        return
+        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_fin(cast, timeout=90):
+    mc = cast.media_controller
+    t0 = time.time()
+    while time.time() - t0 < timeout:
+        try:
+            st = mc.status
+            if st.player_state == "IDLE":
+                return True
+        except Exception:
+            pass
+        time.sleep(1)
+    return False
+
+
+def anuncio_en_teles(texto):
+    """Enciende la tele con el volumen configurado, reproduce el mensaje y la apaga."""
+    if not config.GOOGLE_TV_ACTIVO:
+        return
+    url = _url_audio(texto)
+    if url is None:
         return
 
-    url = f"http://{_ip_lan()}:{config.GOOGLE_TV_PUERTO}/{os.path.basename(ruta)}"
     try:
         import pychromecast
 
@@ -84,12 +124,23 @@ def reproducir_en_teles(texto):
             for c in objetivos:
                 try:
                     c.wait(timeout=10)
+                    c.set_volume(config.GOOGLE_TV_VOLUMEN / 100.0)
                     c.media_controller.play_media(url, "audio/mpeg", title=texto[:80])
                     c.media_controller.block_until_active(timeout=10)
-                    print(f"[Voz] Reproduciendo en {c.cast_info.friendly_name}")
+                    print(f"[Voz] Anuncio en {c.cast_info.friendly_name} "
+                          f"(vol {config.GOOGLE_TV_VOLUMEN})")
+                    if _esperar_fin(c):
+                        print(f"[Voz] Anuncio terminado en {c.cast_info.friendly_name}")
+                    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)