02-ago: [proyecto] vigilancia - teleprompter en la tele: video mp4 con el texto componiendose palabra a palabra sincronizado con la voz (edge-tts SentenceBoundary + ffmpeg via imageio-ffmpeg)

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

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})")