Add application file and Initial commits

.gitignore

@@ -0,0 +1,19 @@
+venv310/
+
+MultiSrcMultiDst/*/*
+!MultiSrcMultiDst/
+!MultiSrcMultiDst/*/
+
+SingleSrcMultiDst/*/*
+!SingleSrcMultiDst/
+!SingleSrcMultiDst/*/
+
+SinglePhoto/*/*
+!SinglePhoto/
+!SinglePhoto/*/
+
+VideoSwapping/*/*
+!VideoSwapping/
+!VideoSwapping/*/
+
+__pycache__/*

MultiSrcMultiDst.py

@@ -0,0 +1,59 @@
+import cv2
+import os
+from SinglePhoto import FaceSwapper  # Updated import
+
+def main():
+    src_dir = os.path.join("MultiSrcMultiDst", "src")
+    dst_dir = os.path.join("MultiSrcMultiDst", "dst")
+    output_root = os.path.join("MultiSrcMultiDst", "output")
+
+    os.makedirs(output_root, exist_ok=True)
+
+    swapper = FaceSwapper()
+
+    # Get all source and destination image paths
+    source_images = [os.path.join(src_dir, f) for f in os.listdir(src_dir) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+    dest_images = [os.path.join(dst_dir, f) for f in os.listdir(dst_dir) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+
+    # Ask user for target_face_idx for each destination image
+    dest_face_indices = {}
+    for dest_path in dest_images:
+        dest_name = os.path.basename(dest_path)
+        while True:
+            try:
+                idx_str = input(f"Enter target_face_idx for destination image '{dest_name}' (default 1): ")
+                if idx_str.strip() == "":
+                    idx = 1
+                else:
+                    idx = int(idx_str)
+                if idx < 1:
+                    print("Index must be >= 1.")
+                    continue
+                dest_face_indices[dest_path] = idx
+                break
+            except ValueError:
+                print("Please enter a valid integer.")
+
+    for source_path in source_images:
+        source_name = os.path.splitext(os.path.basename(source_path))[0]
+        source_output_dir = os.path.join(output_root, source_name)
+        os.makedirs(source_output_dir, exist_ok=True)
+
+        for dest_path in dest_images:
+            dest_name = os.path.splitext(os.path.basename(dest_path))[0]
+            target_face_idx = dest_face_indices[dest_path]
+            try:
+                result = swapper.swap_faces(
+                    source_path=source_path,
+                    source_face_idx=1,
+                    target_path=dest_path,
+                    target_face_idx=target_face_idx
+                )
+                output_path = os.path.join(source_output_dir, f"{source_name}_to_{dest_name}.jpg")
+                cv2.imwrite(output_path, result)
+                print(f"Swapped {source_name} -> {dest_name}: {output_path}")
+            except Exception as e:
+                print(f"Error swapping {source_name} to {dest_name}: {str(e)}")
+
+if __name__ == "__main__":
+    main()

SinglePhoto.py

@@ -0,0 +1,85 @@
+import cv2
+import insightface
+from insightface.app import FaceAnalysis
+import os
+
+class FaceSwapper:
+    def __init__(self):
+        self.app = FaceAnalysis(name='buffalo_l')
+        self.app.prepare(ctx_id=0, det_size=(640, 640))
+        self.swapper = insightface.model_zoo.get_model(
+            'inswapper_128.onnx', download=True, download_zip=True
+        )
+
+    def swap_faces(self, source_path, source_face_idx, target_path, target_face_idx):
+        source_img = cv2.imread(source_path)
+        target_img = cv2.imread(target_path)
+
+        if source_img is None or target_img is None:
+            raise ValueError("Could not read one or both images")
+
+        source_faces = self.app.get(source_img)
+        target_faces = self.app.get(target_img)
+
+        source_faces = sorted(source_faces, key=lambda x: x.bbox[0])
+        target_faces = sorted(target_faces, key=lambda x: x.bbox[0])
+
+        if len(source_faces) < source_face_idx or source_face_idx < 1:
+            raise ValueError(f"Source image contains {len(source_faces)} faces, but requested face {source_face_idx}")
+        if len(target_faces) < target_face_idx or target_face_idx < 1:
+            raise ValueError(f"Target image contains {len(target_faces)} faces, but requested face {target_face_idx}")
+
+        source_face = source_faces[source_face_idx - 1]
+        target_face = target_faces[target_face_idx - 1]
+
+        result = self.swapper.get(target_img, target_face, source_face, paste_back=True)
+        return result
+
+def main():
+    # Paths relative to root
+    source_path = os.path.join("SinglePhoto", "data_src.jpg")
+    target_path = os.path.join("SinglePhoto", "data_dst.jpg")
+    output_dir = os.path.join("SinglePhoto", "output")
+    if not os.path.exists(output_dir):
+        os.makedirs(output_dir)
+
+    swapper = FaceSwapper()
+
+    try:
+        # Ask user for target_face_idx, default to 1 if no input or invalid input
+        try:
+            user_input = input("Enter the target face index (starting from 1, default is 1): ")
+            target_face_idx = int(user_input) if user_input.strip() else 1
+            if target_face_idx < 1:
+                print("Invalid index. Using default value 1.")
+                target_face_idx = 1
+        except ValueError:
+            print("Invalid input. Using default value 1.")
+            target_face_idx = 1
+
+        try:
+            result = swapper.swap_faces(
+                source_path=source_path,
+                source_face_idx=1,
+                target_path=target_path,
+                target_face_idx=target_face_idx
+            )
+        except ValueError as ve:
+            if "Target image contains" in str(ve):
+                print(f"Target face idx {target_face_idx} not found, trying with idx 1.")
+                result = swapper.swap_faces(
+                    source_path=source_path,
+                    source_face_idx=1,
+                    target_path=target_path,
+                    target_face_idx=1
+                )
+            else:
+                raise ve
+        output_path = os.path.join(output_dir, "swapped_face.jpg")
+        cv2.imwrite(output_path, result)
+        print(f"Face swap completed successfully. Result saved to: {output_path}")
+    except Exception as e:
+        print(f"Error occurred: {str(e)}")
+
+if __name__ == "__main__":
+    main()

SingleSrcMultiDst.py

@@ -0,0 +1,70 @@
+import os
+import cv2
+from SinglePhoto import FaceSwapper  # Use the shared class
+
+def main():
+    base_folder = "SingleSrcMultiDst"
+    dst_folder = os.path.join(base_folder, "dst")
+    output_folder = os.path.join(base_folder, "output")
+    source_img_name = "data_src.jpg"  # Change as needed
+
+    if not os.path.exists(output_folder):
+        os.makedirs(output_folder)
+
+    source_img_path = os.path.join(base_folder, source_img_name)
+    if not os.path.exists(source_img_path):
+        print(f"Could not find source image: {source_img_path}")
+        return
+
+    swapper = FaceSwapper()
+
+    # Ask user for source_face_idx (default 1)
+    try:
+        user_input = input("Enter the source face index (starting from 1, default is 1): ")
+        source_face_idx = int(user_input) if user_input.strip() else 1
+        if source_face_idx < 1:
+            print("Invalid index. Using default value 1.")
+            source_face_idx = 1
+    except ValueError:
+        print("Invalid input. Using default value 1.")
+        source_face_idx = 1
+
+    # Ask user for each target_face_idx for each destination image
+    dest_images = [f for f in os.listdir(dst_folder) if f.lower().endswith(('.jpg', '.jpeg', '.png'))]
+    dest_face_indices = {}
+    for filename in dest_images:
+        while True:
+            try:
+                idx_str = input(f"Enter target_face_idx for destination image '{filename}' (default 1): ")
+                if idx_str.strip() == "":
+                    idx = 1
+                else:
+                    idx = int(idx_str)
+                if idx < 1:
+                    print("Index must be >= 1.")
+                    continue
+                dest_face_indices[filename] = idx
+                break
+            except ValueError:
+                print("Please enter a valid integer.")
+
+    for filename in dest_images:
+        target_img_path = os.path.join(dst_folder, filename)
+        if not os.path.exists(target_img_path):
+            print(f"Could not find target image: {target_img_path}")
+            continue
+        try:
+            result = swapper.swap_faces(
+                source_path=source_img_path,
+                source_face_idx=source_face_idx,
+                target_path=target_img_path,
+                target_face_idx=dest_face_indices[filename]
+            )
+            output_path = os.path.join(output_folder, f"swapped_{filename}")
+            cv2.imwrite(output_path, result)
+            print(f"Swapped face saved to: {output_path}")
+        except Exception as e:
+            print(f"Error processing {filename}: {str(e)}")
+
+if __name__ == "__main__":
+    main()

VideoSwapping.py

@@ -0,0 +1,168 @@
+import cv2
+import os
+import shutil
+import time
+from SinglePhoto import FaceSwapper
+
+def extract_frames(video_path, frames_dir):
+    if not os.path.exists(frames_dir):
+        os.makedirs(frames_dir)
+        last_idx = -1
+    else:
+        # Find the last extracted frame index
+        existing = [f for f in os.listdir(frames_dir) if f.startswith("frame_") and f.endswith(".jpg")]
+        if existing:
+            last_idx = max([int(f.split("_")[1].split(".")[0]) for f in existing])
+        else:
+            last_idx = -1
+
+    cap = cv2.VideoCapture(video_path)
+    frame_paths = []
+    idx = 0
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+        frame_path = os.path.join(frames_dir, f"frame_{idx:05d}.jpg")
+        if idx > last_idx:
+            cv2.imwrite(frame_path, frame)
+        frame_paths.append(frame_path)
+        idx += 1
+    cap.release()
+    return frame_paths
+
+def frames_to_video(frames_dir, output_video_path, fps):
+    frames = sorted([os.path.join(frames_dir, f) for f in os.listdir(frames_dir) if f.endswith('.jpg')])
+    if not frames:
+        print("No frames found in directory.")
+        return
+    first_frame = cv2.imread(frames[0])
+    height, width, layers = first_frame.shape
+    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
+    out = cv2.VideoWriter(output_video_path, fourcc, fps, (width, height))
+    for frame_path in frames:
+        frame = cv2.imread(frame_path)
+        out.write(frame)
+    out.release()
+
+def main():
+    # Use files from VideoSwapping folder
+    video_path = os.path.join("VideoSwapping", "data_dst.mp4")
+    source_image_path = os.path.join("VideoSwapping", "data_src.jpg")
+    frames_dir = os.path.join("VideoSwapping", "video_frames")
+    swapped_dir = os.path.join("VideoSwapping", "swapped_frames")
+    output_video_path = os.path.join("VideoSwapping", "output_swapped_video.mp4")
+    source_face_idx = 1
+
+    # Ask user for target_face_idx
+    while True:
+        try:
+            user_input = input("Enter target_face_idx (default is 1): ").strip()
+            if user_input == "":
+                dest_face_idx = 1
+                break
+            dest_face_idx = int(user_input)
+            break
+        except ValueError:
+            print("Invalid input. Please enter an integer value.")
+
+    print("Choose an option:")
+    print("1. Extract frames only")
+    print("2. Face swap only (requires extracted frames)")
+    print("3. Both extract frames and face swap")
+    choice = input("Enter 1, 2, or 3: ").strip()
+
+    frame_paths = []
+    if choice == "1" or choice == "3":
+        print("Extracting frames from video (resuming if needed)...")
+        frame_paths = extract_frames(video_path, frames_dir)
+        print(f"Extracted {len(frame_paths)} frames to {frames_dir}.")
+        if choice == "1":
+            return
+
+    if choice == "2":
+        # If only face swap, ensure frames are present
+        if not os.path.exists(frames_dir):
+            print("Frames directory does not exist. Please extract frames first.")
+            return
+        frame_paths = sorted([os.path.join(frames_dir, f) for f in os.listdir(frames_dir) if f.endswith('.jpg')])
+
+    if choice == "2" or choice == "3":
+        # Prepare output directory
+        if not os.path.exists(swapped_dir):
+            os.makedirs(swapped_dir)
+
+        # Initialize face swapper
+        swapper = FaceSwapper()
+
+        # Swap faces on each frame
+        print("Swapping faces on frames...")
+        start_time = time.time()
+        for idx, frame_path in enumerate(frame_paths):
+            frame_name = os.path.basename(frame_path)
+            # Replace 'frame_' with 'swapped_' in the filename
+            if frame_name.startswith("frame_"):
+                swapped_name = "swapped_" + frame_name[len("frame_"):]
+            else:
+                swapped_name = "swapped_" + frame_name
+            out_path = os.path.join(swapped_dir, swapped_name)
+            if os.path.exists(out_path):
+                # Skip already swapped frames
+                print(f"Frame {idx+1}/{len(frame_paths)} already swapped, skipping...", end='\r')
+                continue
+            try:
+                try:
+                    swapped = swapper.swap_faces(
+                        source_path=source_image_path,
+                        source_face_idx=source_face_idx,
+                        target_path=frame_path,
+                        target_face_idx=dest_face_idx
+                    )
+                except ValueError as ve:
+                    if "Target image contains" in str(ve):
+                        print(f"\nFrame {idx}: Target face idx {dest_face_idx} not found, trying with idx 1.",end='\r')
+                        swapped = swapper.swap_faces(
+                            source_path=source_image_path,
+                            source_face_idx=source_face_idx,
+                            target_path=frame_path,
+                            target_face_idx=1
+                        )
+                    else:
+                        raise ve
+                cv2.imwrite(out_path, swapped)
+            except Exception as e:
+                print(f"\nFrame {idx}: {e}")
+                # Optionally, copy the original frame if swap fails
+                cv2.imwrite(out_path, cv2.imread(frame_path))
+            # Estimate time left
+            elapsed = time.time() - start_time
+            avg_time = elapsed / (idx + 1)
+            remaining = avg_time * (len(frame_paths) - (idx + 1))
+            mins, secs = divmod(int(remaining), 60)
+            print(f"Swapping frame {idx+1}/{len(frame_paths)} | Est. time left: {mins:02d}:{secs:02d}", end='\r')
+        print()  # Move to the next line after the loop
+
+        # Get FPS from original video
+        cap = cv2.VideoCapture(video_path)
+        fps = cap.get(cv2.CAP_PROP_FPS)
+        cap.release()
+
+        # Combine swapped frames into video
+        print("Combining swapped frames into video...")
+        frames_to_video(swapped_dir, output_video_path, fps)
+        print(f"Done! Output video saved as {output_video_path}")
+
+        # Ask user if they want to keep the extracted frames and swapped images
+        answer = input("Do you want to keep the extracted frames and swapped images? (y/n): ").strip().lower()
+        if answer == 'n':
+            try:
+                shutil.rmtree(frames_dir)
+                shutil.rmtree(swapped_dir)
+                print("Temporary folders deleted.")
+            except Exception as e:
+                print(f"Error deleting folders: {e}")
+        else:
+            print("Temporary folders kept.")
+
+if __name__ == "__main__":
+    main()

app.py

@@ -0,0 +1,130 @@
+import gradio as gr
+import os
+import cv2
+from SinglePhoto import FaceSwapper
+
+wellcomingMessage = """
+    <h1>Face Swapping Suite</h1>
+    <p>All-in-one face swapping: single photo, video, multi-source, and multi-destination!</p>
+"""
+
+swapper = FaceSwapper()
+
+def swap_single_photo(src_img, src_idx, dst_img, dst_idx):
+    try:
+        src_path = "tmp_src.jpg"
+        dst_path = "tmp_dst.jpg"
+        cv2.imwrite(src_path, src_img)
+        cv2.imwrite(dst_path, dst_img)
+        result = swapper.swap_faces(src_path, int(src_idx), dst_path, int(dst_idx))
+        return result
+    except Exception as e:
+        return f"Error: {e}"
+
+def swap_video(src_img, src_idx, video, dst_idx):
+    # Save source image
+    src_path = "tmp_src.jpg"
+    cv2.imwrite(src_path, src_img)
+    # Save video
+    video_path = "tmp_video.mp4"
+    with open(video_path, "wb") as f:
+        f.write(video.read())
+    # Extract frames
+    frames_dir = "tmp_video_frames"
+    swapped_dir = "tmp_swapped_frames"
+    output_video_path = "tmp_output_video.mp4"
+    from VideoSwapping import extract_frames, frames_to_video
+    frame_paths = extract_frames(video_path, frames_dir)
+    os.makedirs(swapped_dir, exist_ok=True)
+    for idx, frame_path in enumerate(frame_paths):
+        out_path = os.path.join(swapped_dir, f"swapped_{idx:05d}.jpg")
+        try:
+            swapped = swapper.swap_faces(src_path, int(src_idx), frame_path, int(dst_idx))
+            cv2.imwrite(out_path, swapped)
+        except Exception:
+            cv2.imwrite(out_path, cv2.imread(frame_path))
+    # Combine frames to video
+    cap = cv2.VideoCapture(video_path)
+    fps = cap.get(cv2.CAP_PROP_FPS)
+    cap.release()
+    frames_to_video(swapped_dir, output_video_path, fps)
+    return output_video_path
+
+def swap_multi_src_single_dst(src_imgs, dst_img, dst_idx):
+    # src_imgs: list of images
+    results = []
+    dst_path = "tmp_dst.jpg"
+    cv2.imwrite(dst_path, dst_img)
+    for i, src_img in enumerate(src_imgs):
+        src_path = f"tmp_src_{i}.jpg"
+        cv2.imwrite(src_path, src_img)
+        try:
+            result = swapper.swap_faces(src_path, 1, dst_path, int(dst_idx))
+            results.append(result)
+        except Exception as e:
+            results.append(f"Error: {e}")
+    return results
+
+def swap_multi_src_multi_dst(src_imgs, dst_imgs, dst_indices):
+    # src_imgs, dst_imgs: lists of images; dst_indices: list of indices
+    results = []
+    for i, src_img in enumerate(src_imgs):
+        src_path = f"tmp_src_{i}.jpg"
+        cv2.imwrite(src_path, src_img)
+        for j, dst_img in enumerate(dst_imgs):
+            dst_path = f"tmp_dst_{j}.jpg"
+            cv2.imwrite(dst_path, dst_img)
+            try:
+                result = swapper.swap_faces(src_path, 1, dst_path, int(dst_indices[j]))
+                results.append(result)
+            except Exception as e:
+                results.append(f"Error: {e}")
+    return results
+
+with gr.Blocks() as demo:
+    gr.Markdown(wellcomingMessage)
+    with gr.Tab("Single Photo Swapping"):
+        gr.Interface(
+            fn=swap_single_photo,
+            inputs=[
+                gr.Image(label="Source Image"),
+                gr.Number(value=1, label="Source Face Index"),
+                gr.Image(label="Destination Image"),
+                gr.Number(value=1, label="Destination Face Index"),
+            ],
+            outputs=gr.Image(label="Swapped Image"),
+        )
+    with gr.Tab("Video Swapping"):
+        gr.Interface(
+            fn=swap_video,
+            inputs=[
+                gr.Image(label="Source Image"),
+                gr.Number(value=1, label="Source Face Index"),
+                gr.Video(label="Target Video"),
+                gr.Number(value=1, label="Destination Face Index"),
+            ],
+            outputs=gr.Video(label="Swapped Video"),
+        )
+    with gr.Tab("MultiSrc SingleDst"):
+        gr.Interface(
+            fn=swap_multi_src_single_dst,
+            inputs=[
+                gr.Gallery(label="Source Images", type="numpy", columns=3),
+                gr.Image(label="Destination Image"),
+                gr.Number(value=1, label="Destination Face Index"),
+            ],
+            outputs=gr.Gallery(label="Swapped Images"),
+        )
+    with gr.Tab("MultiSrc MultiDst"):
+        gr.Interface(
+            fn=swap_multi_src_multi_dst,
+            inputs=[
+                gr.Gallery(label="Source Images", type="numpy", columns=3),
+                gr.Gallery(label="Destination Images", type="numpy", columns=3),
+                gr.Textbox(label="Destination Face Indices (comma-separated, e.g. 1,1,2)"),
+            ],
+            outputs=gr.Gallery(label="Swapped Images"),
+        )
+
+if __name__ == "__main__":
+    demo.launch()

inswapper_128.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e4a3f08c753cb72d04e10aa0f7dbe3deebbf39567d4ead6dce08e98aa49e16af
+size 554253681

requirements.txt

@@ -0,0 +1,8 @@
+numpy
+opencv-python
+onnxruntime-gpu
+scikit-learn
+scikit-image
+tqdm
+mxnet
+insightface