v1

README.md

@@ -1,12 +1,32 @@
 ---
-title: Face Coordinates Extraction
-emoji: 👀
-colorFrom: red
-colorTo: blue
+title: Dub Module Step1-Step3 App
 sdk: gradio
-sdk_version: 6.6.0
 app_file: app.py
+python_version: "3.10"
 pinned: false
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Dub Module Gradio App (Step 1 + Step 3)
+
+This folder creates a new Gradio app based on the workflow described in [`how_to.txt`](../how_to.txt).
+
+Implemented workflow:
+- Step 1 (in app): Upload main video, extract cropped face video, save `face_coords_avg.pkl`, and download both outputs.
+- Step 2 (manual): Not part of app.
+- Step 3 (in app): Upload original video + synced face video + `face_coords_avg.pkl` to generate final output video.
+
+Notes:
+- Audio upload is intentionally removed. The app attempts to use audio from the synced face video.
+- Generated runtime files are stored under `work/`.
+
+## Files
+- `app.py`: Gradio UI for Step 1 and Step 3.
+- `pipeline.py`: Face extraction, coordinate generation, merging, and audio muxing.
+- `requirements.txt`: Package versions known to work.
+
+## Run
+
+```bash
+pip install -r requirements.txt
+python app.py
+```

app.py

@@ -0,0 +1,139 @@
+from pathlib import Path
+
+import gradio as gr
+
+from pipeline import (
+    copy_file_to_dir,
+    extract_face_and_coords,
+    make_run_dir,
+    merge_synced_face,
+)
+
+BASE_DIR = Path(__file__).resolve().parent
+WORK_DIR = BASE_DIR / "work"
+WORK_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _normalize_upload_path(file_obj):
+    if file_obj is None:
+        return None
+    if isinstance(file_obj, str):
+        return file_obj
+    return str(file_obj)
+
+
+def run_step1(main_video):
+    try:
+        main_path = _normalize_upload_path(main_video)
+        if not main_path:
+            raise ValueError("Please upload the main/original video.")
+
+        run_dir = make_run_dir(WORK_DIR, "step1")
+        local_main = copy_file_to_dir(main_path, run_dir, "main_video.mp4")
+
+        coords_path, cropped_face_path, bbox = extract_face_and_coords(
+            video_path=str(local_main),
+            output_dir=str(run_dir),
+            coords_name="face_coords_avg.pkl",
+            cropped_name="cropped_face.mp4",
+        )
+
+        status = f"Step 1 completed. Face bbox saved: {bbox}"
+        return status, cropped_face_path, cropped_face_path, coords_path
+    except Exception as exc:
+        return f"Step 1 failed: {exc}", None, None, None
+
+
+def run_step3(main_video, synced_face_video, face_coords):
+    try:
+        main_path = _normalize_upload_path(main_video)
+        synced_path = _normalize_upload_path(synced_face_video)
+        coords_path = _normalize_upload_path(face_coords)
+
+        if not main_path:
+            raise ValueError("Please upload the original/main video.")
+        if not synced_path:
+            raise ValueError("Please upload the synced face video from manual Step 2.")
+        if not coords_path:
+            raise ValueError("Please upload face coordinates (.pkl) from Step 1.")
+
+        run_dir = make_run_dir(WORK_DIR, "step3")
+        local_main = copy_file_to_dir(main_path, run_dir, "original_video.mp4")
+        local_synced = copy_file_to_dir(synced_path, run_dir, "synced_face_video.mp4")
+        local_coords = copy_file_to_dir(coords_path, run_dir, "face_coords_avg.pkl")
+
+        final_path = run_dir / "final_output_with_audio.mp4"
+        output_path, audio_used = merge_synced_face(
+            original_video_path=str(local_main),
+            synced_face_video_path=str(local_synced),
+            face_coords_path=str(local_coords),
+            final_output_path=str(final_path),
+        )
+
+        if audio_used == "synced_face_video":
+            status = "Step 3 completed. Final video generated with audio from synced face video."
+        else:
+            status = "Step 3 completed. Final video generated without muxed audio (audio track not found)."
+
+        return status, output_path, output_path
+    except Exception as exc:
+        return f"Step 3 failed: {exc}", None, None
+
+
+with gr.Blocks(title="Dub Module - Step 1 and Step 3") as demo:
+    gr.Markdown(
+        """
+# Dub Module Gradio App (Step 1 + Step 3)
+Workflow follows `how_to.txt` in this repo with these app boundaries:
+- Step 1 is in-app: extract cropped face + `face_coords_avg.pkl`.
+- Step 2 is manual and outside the app.
+- Step 3 is in-app: merge synced face video back to original and produce final video.
+- Separate audio upload is skipped because synced face video audio is used.
+        """
+    )
+
+    with gr.Tab("Step 1 - Extract Face + Coordinates"):
+        gr.Markdown("Upload the main video to generate cropped face video and face coordinates.")
+        s1_video = gr.File(label="Main Video", file_types=["video"], type="filepath")
+        s1_run = gr.Button("Run Step 1")
+        s1_status = gr.Textbox(label="Status", interactive=False)
+        s1_preview = gr.Video(label="Cropped Face Preview")
+        s1_face_file = gr.File(label="Download Cropped Face Video")
+        s1_coords_file = gr.File(label="Download Face Coordinates (.pkl)")
+
+        s1_run.click(
+            fn=run_step1,
+            inputs=[s1_video],
+            outputs=[s1_status, s1_preview, s1_face_file, s1_coords_file],
+        )
+
+    with gr.Tab("Step 2 - Manual (Outside App)"):
+        gr.Markdown(
+            """
+Do manual lip-sync generation outside this app using the Step 1 cropped face video.
+Then return to Step 3 tab with:
+1. Original main video
+2. Synced face video (with audio)
+3. `face_coords_avg.pkl`
+            """
+        )
+
+    with gr.Tab("Step 3 - Merge and Final Video"):
+        gr.Markdown("Upload inputs from Step 1 and manual Step 2 to generate final output video.")
+        s3_main_video = gr.File(label="Original Main Video", file_types=["video"], type="filepath")
+        s3_synced_video = gr.File(label="Synced Face Video", file_types=["video"], type="filepath")
+        s3_coords = gr.File(label="Face Coordinates (.pkl)", file_types=[".pkl"], type="filepath")
+        s3_run = gr.Button("Run Step 3")
+        s3_status = gr.Textbox(label="Status", interactive=False)
+        s3_preview = gr.Video(label="Final Output Preview")
+        s3_file = gr.File(label="Download Final Video")
+
+        s3_run.click(
+            fn=run_step3,
+            inputs=[s3_main_video, s3_synced_video, s3_coords],
+            outputs=[s3_status, s3_preview, s3_file],
+        )
+
+
+if __name__ == "__main__":
+    demo.launch()

pipeline.py

@@ -0,0 +1,263 @@
+import pickle
+import shutil
+import subprocess
+import uuid
+from pathlib import Path
+from typing import Optional, Sequence, Tuple
+
+import cv2
+import imageio_ffmpeg
+import numpy as np
+
+
+def _create_face_mesh():
+    try:
+        from mediapipe.python.solutions.face_mesh import FaceMesh
+    except Exception:
+        import mediapipe as mp
+
+        FaceMesh = mp.solutions.face_mesh.FaceMesh
+
+    return FaceMesh(
+        static_image_mode=False,
+        max_num_faces=1,
+        refine_landmarks=True,
+        min_detection_confidence=0.8,
+    )
+
+
+def ensure_dir(path: Path) -> Path:
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def make_run_dir(base_dir: Path, prefix: str) -> Path:
+    run_dir = ensure_dir(base_dir) / f"{prefix}_{uuid.uuid4().hex}"
+    return ensure_dir(run_dir)
+
+
+def copy_file_to_dir(source_path: str, target_dir: Path, target_name: Optional[str] = None) -> Path:
+    source = Path(source_path)
+    if not source.exists():
+        raise FileNotFoundError(f"Input file not found: {source_path}")
+
+    if target_name is None:
+        target_name = source.name
+
+    target_path = target_dir / target_name
+    shutil.copy2(source, target_path)
+    return target_path
+
+
+def get_bbox(
+    landmarks,
+    indices: Sequence[int],
+    iw: int,
+    ih: int,
+    scale_w: float = 1.2,
+    scale_h: float = 1.2,
+) -> Tuple[int, int, int, int]:
+    coords = [(landmarks[i].x * iw, landmarks[i].y * ih) for i in indices]
+    x_min, y_min = np.min(coords, axis=0)
+    x_max, y_max = np.max(coords, axis=0)
+
+    w = x_max - x_min
+    h = y_max - y_min
+    new_w = int(w * scale_w)
+    new_h = int(h * scale_h)
+
+    x = max(0, int(x_min - (new_w - w) // 2))
+    y = max(0, int(y_min - (new_h - h) // 2))
+    new_w = min(new_w, iw - x)
+    new_h = min(new_h, ih - y)
+
+    return (x, y, new_w, new_h)
+
+
+def _load_coords(coords_path: str) -> Tuple[int, int, int, int]:
+    with open(coords_path, "rb") as handle:
+        coords = pickle.load(handle)
+
+    if len(coords) != 4:
+        raise ValueError(f"Invalid coordinates in {coords_path}: expected 4 values, got {len(coords)}")
+
+    return tuple(int(v) for v in coords)
+
+
+def extract_face_and_coords(
+    video_path: str,
+    output_dir: str,
+    coords_name: str = "face_coords_avg.pkl",
+    cropped_name: str = "cropped_face.mp4",
+) -> Tuple[str, str, Tuple[int, int, int, int]]:
+    output_root = ensure_dir(Path(output_dir))
+    coords_out = output_root / coords_name
+    cropped_out = output_root / cropped_name
+
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise ValueError(f"Could not open video: {video_path}")
+
+    face_mesh = _create_face_mesh()
+    face_bbox_list = []
+
+    frame_w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    frame_h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    if fps <= 0:
+        fps = 25.0
+
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
+        results = face_mesh.process(image_rgb)
+
+        if results.multi_face_landmarks:
+            for face_landmarks in results.multi_face_landmarks:
+                ih, iw, _ = frame.shape
+                face_bbox = get_bbox(
+                    face_landmarks.landmark,
+                    range(len(face_landmarks.landmark)),
+                    iw,
+                    ih,
+                    scale_w=1.2,
+                    scale_h=1.2,
+                )
+                face_bbox_list.append(face_bbox)
+
+    cap.release()
+    face_mesh.close()
+
+    if not face_bbox_list:
+        raise ValueError("No faces detected in the video. Check framing and quality.")
+
+    avg_face_bbox = np.mean(np.array(face_bbox_list), axis=0).astype(int)
+    x, y, w, h = (int(v) for v in avg_face_bbox)
+
+    x = max(0, min(x, frame_w - 1))
+    y = max(0, min(y, frame_h - 1))
+    w = max(1, min(w, frame_w - x))
+    h = max(1, min(h, frame_h - y))
+    final_bbox = (x, y, w, h)
+
+    with open(coords_out, "wb") as handle:
+        pickle.dump(final_bbox, handle)
+
+    cap = cv2.VideoCapture(video_path)
+    if not cap.isOpened():
+        raise ValueError(f"Could not reopen video for cropping: {video_path}")
+
+    out = cv2.VideoWriter(
+        str(cropped_out),
+        cv2.VideoWriter_fourcc(*"mp4v"),
+        fps,
+        (w, h),
+    )
+
+    frames_written = 0
+    while cap.isOpened():
+        ret, frame = cap.read()
+        if not ret:
+            break
+        face_img = frame[y:y + h, x:x + w]
+        out.write(face_img)
+        frames_written += 1
+
+    cap.release()
+    out.release()
+
+    if frames_written == 0:
+        raise ValueError("No frames were written for cropped face output.")
+
+    return str(coords_out), str(cropped_out), final_bbox
+
+
+def _mux_audio(video_no_audio: str, audio_source: str, output_path: str) -> bool:
+    ffmpeg_exe = imageio_ffmpeg.get_ffmpeg_exe()
+    cmd = [
+        ffmpeg_exe,
+        "-y",
+        "-i",
+        video_no_audio,
+        "-i",
+        audio_source,
+        "-map",
+        "0:v:0",
+        "-map",
+        "1:a:0",
+        "-c:v",
+        "copy",
+        "-c:a",
+        "aac",
+        "-shortest",
+        output_path,
+    ]
+    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    return result.returncode == 0 and Path(output_path).exists()
+
+
+def merge_synced_face(
+    original_video_path: str,
+    synced_face_video_path: str,
+    face_coords_path: str,
+    final_output_path: str,
+) -> Tuple[str, str]:
+    x, y, w, h = _load_coords(face_coords_path)
+
+    original_cap = cv2.VideoCapture(original_video_path)
+    synced_cap = cv2.VideoCapture(synced_face_video_path)
+
+    if not original_cap.isOpened():
+        raise ValueError(f"Could not open original video: {original_video_path}")
+    if not synced_cap.isOpened():
+        raise ValueError(f"Could not open synced face video: {synced_face_video_path}")
+
+    fps = original_cap.get(cv2.CAP_PROP_FPS)
+    if fps <= 0:
+        fps = 25.0
+
+    frame_w = int(original_cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+    frame_h = int(original_cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+
+    x = max(0, min(x, frame_w - 1))
+    y = max(0, min(y, frame_h - 1))
+    w = max(1, min(w, frame_w - x))
+    h = max(1, min(h, frame_h - y))
+
+    intermediate_path = str(Path(final_output_path).with_name("merged_no_audio.mp4"))
+    out = cv2.VideoWriter(
+        intermediate_path,
+        cv2.VideoWriter_fourcc(*"mp4v"),
+        fps,
+        (frame_w, frame_h),
+    )
+
+    frames_written = 0
+    while original_cap.isOpened():
+        ret_o, original_frame = original_cap.read()
+        if not ret_o:
+            break
+
+        ret_s, synced_frame = synced_cap.read()
+        if ret_s:
+            synced_resized = cv2.resize(synced_frame, (w, h))
+            original_frame[y:y + h, x:x + w] = synced_resized
+
+        out.write(original_frame)
+        frames_written += 1
+
+    original_cap.release()
+    synced_cap.release()
+    out.release()
+
+    if frames_written == 0:
+        raise ValueError("No frames written while creating final video.")
+
+    if _mux_audio(intermediate_path, synced_face_video_path, final_output_path):
+        return final_output_path, "synced_face_video"
+
+    shutil.copy2(intermediate_path, final_output_path)
+    return final_output_path, "none"

requirements.txt

@@ -0,0 +1,6 @@
+gradio>=5.0.0
+opencv-python-headless==4.10.0.84
+mediapipe==0.10.14
+numpy>=1.24.0,<2.1
+imageio-ffmpeg>=0.4.9
+protobuf==4.25.3