diff --git a/vigilancia/teleprompter.py b/vigilancia/teleprompter.py new file mode 100644 index 0000000..58adad7 --- /dev/null +++ b/vigilancia/teleprompter.py @@ -0,0 +1,143 @@ +import asyncio +import os +import subprocess + +import imageio_ffmpeg + +import config + +FFMPEG = imageio_ffmpeg.get_ffmpeg_exe() +W, H, FPS = 1280, 720, 20 +FUENTE = None +for p in (r"C:\Windows\Fonts\arialbd.ttf", + r"C:\Windows\Fonts\arial.ttf", + r"C:\Windows\Fonts\segoeui.ttf"): + if os.path.exists(p): + FUENTE = p + break + + +def _duracion_audio(ruta): + try: + r = subprocess.run( + [FFMPEG, "-i", ruta, "-f", "null", "-"], + capture_output=True, text=True, timeout=30, + ) + for linea in (r.stderr or "").splitlines(): + if "Duration:" in linea: + hh, mm, ss = linea.split("Duration:")[1].split(",")[0].strip().split(":") + return int(hh) * 3600 + int(mm) * 60 + float(ss) + except Exception: + pass + return None + + +def sintetizar_con_oraciones(texto, nombre="anuncio.mp3"): + """Genera el audio y devuelve (ruta_audio, oraciones_con_tiempos).""" + from PIL import Image # noqa: F401 (import de dependencia temprano) + import edge_tts + + ruta = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "voz_media", nombre) + os.makedirs(os.path.dirname(ruta), exist_ok=True) + audio = [] + oraciones = [] + + async def _gen(): + comunicador = edge_tts.Communicate( + texto, config.VOZ_EDGE, + rate=config.VOZ_RATE, pitch=config.VOZ_PITCH, + ) + async for ch in comunicador.stream(): + if ch["type"] == "audio": + audio.append(ch["data"]) + elif ch["type"] == "SentenceBoundary": + inicio = ch["offset"] / 1e7 + fin = inicio + ch["duration"] / 1e7 + oraciones.append((ch["text"], inicio, fin)) + + asyncio.run(_gen()) + if not audio: + return None, [] + with open(ruta, "wb") as f: + f.write(b"".join(audio)) + return ruta, oraciones + + +def _palabras(oraciones): + """Distribuye las palabras dentro de cada frase por longitud.""" + palabras = [] + for texto, s, e in oraciones: + tokens = texto.split() + total = sum(len(t) for t in tokens) or 1 + t = s + for tok in tokens: + dur = (e - s) * (len(tok) / total) + palabras.append((tok, t, t + dur)) + t += dur + return palabras + + +def generar_video(texto, ruta_audio, oraciones, nombre="anuncio.mp4"): + from PIL import Image, ImageDraw, ImageFont + + ruta_video = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "voz_media", nombre) + palabras = _palabras(oraciones) + + dur = _duracion_audio(ruta_audio) or ((palabras[-1][2] if palabras else 0) + 2) + total = dur + 0.5 + n_frames = int(total * FPS) + + font = ImageFont.truetype(FUENTE, 52) if FUENTE else ImageFont.load_default() + gap = 18 + + # agrupar en lineas + lineas = [] + linea = [] + ancho_linea = 0 + for tok, s, e in palabras: + ancho = font.getlength(tok) + if ancho_linea + ancho + gap > W - 80 and linea: + lineas.append(linea) + linea = [(tok, s, e)] + ancho_linea = ancho + else: + linea.append((tok, s, e)) + ancho_linea += ancho + gap + if linea: + lineas.append(linea) + + cmd = [FFMPEG, "-y", "-f", "rawvideo", "-vcodec", "rawvideo", + "-s", f"{W}x{H}", "-pix_fmt", "rgb24", "-r", str(FPS), "-i", "-", + "-i", ruta_audio, + "-c:v", "libx264", "-preset", "veryfast", "-pix_fmt", "yuv420p", + "-c:a", "aac", "-b:a", "128k", "-shortest", ruta_video] + proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, + stderr=subprocess.DEVNULL) + + for i in range(n_frames): + t = i / FPS + img = Image.new("RGB", (W, H), (8, 8, 8)) + d = ImageDraw.Draw(img) + y = H // 2 - (len(lineas) // 2) * 80 + 30 + for linea_words in lineas: + x = 40 + for tok, s, e in linea_words: + if s <= t < e: + color = (255, 220, 0) + elif t >= e: + color = (255, 255, 255) + else: + color = (90, 90, 90) + d.text((x, y), tok, fill=color, font=font) + x += font.getlength(tok) + gap + y += 80 + try: + proc.stdin.write(img.tobytes()) + except BrokenPipeError: + break + + proc.stdin.close() + proc.wait() + return ruta_video diff --git a/vigilancia/voz.py b/vigilancia/voz.py index 150caa5..9d256a8 100644 --- a/vigilancia/voz.py +++ b/vigilancia/voz.py @@ -82,6 +82,23 @@ def _descubrir(): 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_audio(texto, nombre="anuncio.mp3"): ruta = sintetizar(texto, nombre=nombre) if ruta is None: @@ -210,7 +227,10 @@ def anuncio_en_teles(texto): mensaje y apaga por ADB.""" if not config.GOOGLE_TV_ACTIVO: return - url_mensaje = _url_audio(texto) + 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 @@ -251,7 +271,7 @@ def anuncio_en_teles(texto): f"Pitido de emergencia en {c.cast_info.friendly_name}") _reproducir_y_esperar( - c, url_mensaje, "audio/mpeg", texto[:80], + c, url_mensaje, tipo_mensaje, texto[:80], f"Anuncio en {c.cast_info.friendly_name} " f"(vol {config.GOOGLE_TV_VOLUMEN})")