feat: local inference fallback using DeepFakeAI face_swapper

app.py

@@ -2,6 +2,12 @@ import gradio as gr
 import os, uuid, time
 from gradio_client import Client, handle_file
 from moviepy.editor import VideoFileClip
+from typing import Optional
+
+# Local pipeline imports
+import DeepFakeAI.globals as DF_G
+from DeepFakeAI import utilities as DF_U
+from DeepFakeAI.processors.frame.modules import face_swapper as DF_FS
 
 # ── config ───────────────────────────────────────────────────────────
 hf_token      = os.environ.get("TOKEN")
@@ -36,6 +42,46 @@ def preprocess_video(path: str, target_fps: int = 12,
     clip.close()
     return out_path
 
+def _run_local_faceswap(source_image_path: str, target_video_path: str) -> Optional[str]:
+    # Configure defaults for local pipeline
+    DF_G.source_path = source_image_path
+    DF_G.target_path = target_video_path
+    DF_G.output_video_encoder = 'libx264'
+    DF_G.output_video_quality = 20
+    DF_G.temp_frame_format = 'png'
+    DF_G.temp_frame_quality = 95
+    DF_G.keep_temp = False
+    DF_G.skip_audio = False
+    DF_G.execution_thread_count = 2
+    DF_G.execution_queue_count = 2
+    # Prefer CUDA on T4 if available; fallback to CPU
+    DF_G.execution_providers = DF_U.decode_execution_providers(['cuda', 'cpu'])
+
+    # Ensure model exists
+    DF_FS.pre_check()
+
+    # Extract frames
+    fps = DF_U.detect_fps(target_video_path) or 12.0
+    DF_U.create_temp(target_video_path)
+    ok = DF_U.extract_frames(target_video_path, fps)
+    if not ok:
+        return None
+    temp_frames = DF_U.get_temp_frame_paths(target_video_path)
+    if not temp_frames:
+        return None
+
+    # Process frames
+    DF_FS.process_video(source_image_path, temp_frames)
+
+    # Rebuild video and restore audio
+    if not DF_U.create_video(target_video_path, fps):
+        return None
+    out_path = DF_U.normalize_output_path(source_image_path, target_video_path, os.path.join(output_dir, f"out_{uuid.uuid4().hex}.mp4"))
+    DF_U.restore_audio(target_video_path, out_path)
+    DF_U.clear_temp(target_video_path)
+    return out_path
+
+
 # ── main generate ────────────────────────────────────────────────────
 def generate(input_image, input_video):
 
@@ -52,24 +98,26 @@ def generate(input_image, input_video):
 
     try:
         pre_video = preprocess_video(input_video)
-        job = client.submit(
-            input_image=handle_file(input_image),
-            input_video={"video": handle_file(pre_video)},
-            device='cpu',
-            selector='many',
-            gender=None,
-            race=None,
-            order=None,
-            api_name="/predict"
-        )
-        while not job.done():
-            time.sleep(5)
-
-        if not job.status().success:
-            return None
 
-        video_path = job.outputs()[0]["video"]
-        return video_path
+        if client is not None:
+            job = client.submit(
+                input_image=handle_file(input_image),
+                input_video={"video": handle_file(pre_video)},
+                device='cpu',
+                selector='many',
+                gender=None,
+                race=None,
+                order=None,
+                api_name="/predict"
+            )
+            while not job.done():
+                time.sleep(5)
+            if not job.status().success:
+                return None
+            return job.outputs()[0]["video"]
+        else:
+            # Local fallback
+            return _run_local_faceswap(input_image, pre_video)
 
     except Exception as e:
         gr.Error(f"Generation failed: {e}")