Upload pipeline/talking_head.py with huggingface_hub

pipeline/talking_head.py

@@ -0,0 +1,58 @@
+"""
+SadTalker client — sends portrait + audio to the MI300X server, gets MP4 back.
+
+Server (droplet port 8003):
+  POST /generate  multipart: audio file + profile_id form field → MP4 bytes
+  GET  /health    → {"status":"ok"}
+"""
+
+import os
+import tempfile
+import httpx
+
+TALKING_HEAD_URL = os.getenv("TALKING_HEAD_URL", "")
+TIMEOUT          = float(os.getenv("TALKING_HEAD_TIMEOUT", "120"))
+
+
+def generate_talking_video(audio_path: str, profile_id: str) -> str | None:
+    """
+    Animate the profile portrait to match the given audio.
+    Returns path to a temp MP4 file, or None if unavailable/failed.
+    """
+    if not TALKING_HEAD_URL or not audio_path:
+        return None
+    if not os.path.exists(audio_path):
+        return None
+    try:
+        with open(audio_path, "rb") as f:
+            audio_bytes = f.read()
+
+        suffix  = os.path.splitext(audio_path)[1] or ".mp3"
+        mime    = "audio/mpeg" if suffix == ".mp3" else "audio/wav"
+
+        r = httpx.post(
+            f"{TALKING_HEAD_URL.rstrip('/')}/generate",
+            files={"audio": (f"audio{suffix}", audio_bytes, mime)},
+            data={"profile_id": profile_id},
+            timeout=TIMEOUT,
+        )
+        r.raise_for_status()
+
+        tmp = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False)
+        tmp.write(r.content)
+        tmp.close()
+        return tmp.name
+
+    except Exception as e:
+        print(f"[TalkingHead] {e}")
+        return None
+
+
+def is_available() -> bool:
+    if not TALKING_HEAD_URL:
+        return False
+    try:
+        r = httpx.get(f"{TALKING_HEAD_URL.rstrip('/')}/health", timeout=5.0)
+        return r.status_code == 200
+    except Exception:
+        return False