fastswap.py

"""
Fast face swap via InsightFace inswapper_128 — deterministic, no prompts, no diffusion.

This is the same pipeline used by roop / FaceSwapKahraman: detect the source face
once, then for every frame detect faces and paste the swapped face back. Runs on
CPU (onnxruntime); GFPGAN enhancement is applied by the caller if requested.
"""

from __future__ import annotations

from typing import Callable

import numpy as np
from huggingface_hub import hf_hub_download

_swapper = None
_analyzer = None

INSWAPPER_REPO = "ezioruan/inswapper_128.onnx"
INSWAPPER_FILE = "inswapper_128.onnx"


def _make_models(providers, ctx_id):
    import insightface
    from insightface.app import FaceAnalysis
    # Fast swap runs InsightFace on CPU on ZeroGPU (onnxruntime-gpu doesn't win over the
    # CPU onnxruntime insightface pulls in). To stay inside the GPU time window we cut the
    # per-frame cost hard: load ONLY detection + recognition (inswapper needs the target
    # face's keypoints from detection + the source embedding from recognition — the
    # genderage and 2D/3D landmark models aren't used), and detect at 320px not 640px.
    # Together that's a ~5-8x speedup vs the old 5-model / 640px setup.
    analyzer = FaceAnalysis(
        name="buffalo_l",
        allowed_modules=["detection", "recognition"],
        providers=providers,
    )
    analyzer.prepare(ctx_id=ctx_id, det_size=(320, 320))
    model_path = hf_hub_download(INSWAPPER_REPO, INSWAPPER_FILE)
    swapper = insightface.model_zoo.get_model(model_path, providers=providers)
    return analyzer, swapper


def _get_models():
    """Prefer the GPU (CUDAExecutionProvider) — CPU inswapper is ~10-50x slower and
    blows the ZeroGPU time window on anything but the shortest clip. Falls back to CPU
    if onnxruntime-gpu / CUDA isn't actually usable, so there's no regression."""
    global _swapper, _analyzer
    if _swapper is not None and _analyzer is not None:
        return _analyzer, _swapper

    cuda = False
    try:
        import onnxruntime as ort
        cuda = "CUDAExecutionProvider" in ort.get_available_providers()
    except Exception:
        cuda = False

    if cuda:
        try:
            _analyzer, _swapper = _make_models(
                ["CUDAExecutionProvider", "CPUExecutionProvider"], 0
            )
            print("[fastswap] InsightFace running on CUDA")
        except Exception as e:  # CUDA listed but not actually usable → fall back
            print(f"[fastswap] CUDA init failed ({e}); falling back to CPU")
            _analyzer = _swapper = None

    if _swapper is None or _analyzer is None:
        _analyzer, _swapper = _make_models(["CPUExecutionProvider"], -1)
        print("[fastswap] InsightFace running on CPU")

    return _analyzer, _swapper


def fast_swap_video(
    frames: np.ndarray,
    face_image,
    progress_cb: Callable[[float, str], None] | None = None,
) -> tuple[np.ndarray, str]:
    """
    Swap the reference face onto every frame.

    Args:
        frames:     uint8 [N, H, W, 3] RGB guide frames
        face_image: PIL image of the reference face

    Returns:
        (uint8 [N, H, W, 3] RGB swapped frames, status message)
    """
    analyzer, swapper = _get_models()

    src_bgr = np.array(face_image.convert("RGB"))[:, :, ::-1].copy()
    src_faces = analyzer.get(src_bgr)
    if not src_faces:
        return frames, "No face detected in the reference image — returning guide video unchanged."
    source_face = max(src_faces, key=lambda f: (f.bbox[2] - f.bbox[0]) * (f.bbox[3] - f.bbox[1]))

    out = np.empty_like(frames)
    swapped_count = 0
    N = len(frames)
    for i in range(N):
        frame_bgr = frames[i][:, :, ::-1].copy()
        faces = analyzer.get(frame_bgr)
        if faces:
            for tgt in faces:
                frame_bgr = swapper.get(frame_bgr, tgt, source_face, paste_back=True)
            swapped_count += 1
        out[i] = frame_bgr[:, :, ::-1]
        if progress_cb and (i % 8 == 0 or i == N - 1):
            progress_cb(i / max(N - 1, 1), f"Swapping faces… {i + 1}/{N}")

    if swapped_count == 0:
        return frames, "No faces detected in the guide video — returning it unchanged."
    return out, f"Swapped {swapped_count}/{N} frames."