278 lines
8.6 KiB
Python
278 lines
8.6 KiB
Python
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_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, timeout=15):
|
|
"""Espera a que la reproduccion realmente empiece (PLAYING/BUFFERING)."""
|
|
mc = cast.media_controller
|
|
t0 = time.time()
|
|
while time.time() - t0 < timeout:
|
|
try:
|
|
st = mc.status
|
|
if st.player_state in ("PLAYING", "BUFFERING"):
|
|
return True
|
|
if st.player_state == "IDLE" and st.idle_reason == "FINISHED":
|
|
return True
|
|
except Exception:
|
|
pass
|
|
time.sleep(1)
|
|
return False
|
|
|
|
|
|
def _esperar_fin(cast, timeout=120):
|
|
"""Espera a que el audio termine de verdad (IDLE + idle_reason FINISHED)."""
|
|
mc = cast.media_controller
|
|
t0 = time.time()
|
|
visto_playing = False
|
|
while time.time() - t0 < timeout:
|
|
try:
|
|
st = mc.status
|
|
if st.player_state == "PLAYING":
|
|
visto_playing = True
|
|
if st.player_state == "IDLE" and st.idle_reason == "FINISHED":
|
|
return True
|
|
except Exception:
|
|
pass
|
|
time.sleep(1)
|
|
return visto_playing
|
|
|
|
|
|
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)
|
|
c.media_controller.block_until_active(timeout=10)
|
|
print(f"[Voz] {desc}")
|
|
_esperar_reproduccion(c)
|
|
return _esperar_fin(c)
|
|
|
|
|
|
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 = _url_audio(texto)
|
|
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
|
|
|
|
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}")
|
|
|
|
_reproducir_y_esperar(
|
|
c, url_mensaje, "audio/mpeg", texto[:80],
|
|
f"Anuncio en {c.cast_info.friendly_name} "
|
|
f"(vol {config.GOOGLE_TV_VOLUMEN})")
|
|
|
|
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)
|