add

.gitignore

@@ -0,0 +1,2 @@
+
+*.pyc

app.py

@@ -0,0 +1,871 @@
+import os
+import cv2
+import glob
+import time
+import torch
+import shutil
+import gfpgan
+import argparse
+import platform
+import datetime
+import subprocess
+import insightface
+import onnxruntime
+import numpy as np
+import gradio as gr
+from moviepy.editor import VideoFileClip, ImageSequenceClip
+
+from face_analyser import detect_conditions, analyse_face
+from utils import trim_video, StreamerThread, ProcessBar, open_directory
+from face_parsing import init_parser, swap_regions, mask_regions, mask_regions_to_list
+from swapper import (
+    swap_face,
+    swap_face_with_condition,
+    swap_specific,
+    swap_options_list,
+)
+
+## ------------------------------ USER ARGS ------------------------------
+
+parser = argparse.ArgumentParser(description="Swap-Mukham Face Swapper")
+parser.add_argument("--out_dir", help="Default Output directory", default=os.getcwd())
+parser.add_argument("--cuda", action="store_true", help="Enable cuda", default=False)
+parser.add_argument(
+    "--colab", action="store_true", help="Enable colab mode", default=False
+)
+user_args = parser.parse_args()
+
+## ------------------------------ DEFAULTS ------------------------------
+
+USE_COLAB = user_args.colab
+USE_CUDA = user_args.cuda
+DEF_OUTPUT_PATH = user_args.out_dir
+WORKSPACE = None
+OUTPUT_FILE = None
+CURRENT_FRAME = None
+STREAMER = None
+DETECT_CONDITION = "left most"
+DETECT_SIZE = 640
+DETECT_THRESH = 0.6
+NUM_OF_SRC_SPECIFIC = 10
+MASK_INCLUDE = [
+    "Skin",
+    "R-Eyebrow",
+    "L-Eyebrow",
+    "L-Eye",
+    "R-Eye",
+    "Nose",
+    "Mouth",
+    "L-Lip",
+    "U-Lip"
+]
+MASK_EXCLUDE = ["R-Ear", "L-Ear", "Hair", "Hat"]
+MASK_BLUR = 25
+
+FACE_SWAPPER = None
+FACE_ANALYSER = None
+FACE_ENHANCER = None
+FACE_PARSER = None
+
+## ------------------------------ SET EXECUTION PROVIDER ------------------------------
+# Note: For AMD,MAC or non CUDA users, change settings here
+
+PROVIDER = ["CPUExecutionProvider"]
+
+if USE_CUDA:
+    available_providers = onnxruntime.get_available_providers()
+    if "CUDAExecutionProvider" in available_providers:
+        print("\n********** Running on CUDA **********\n")
+        PROVIDER = ["CUDAExecutionProvider", "CPUExecutionProvider"]
+    else:
+        USE_CUDA = False
+        print("\n********** CUDA unavailable running on CPU **********\n")
+else:
+    USE_CUDA = False
+    print("\n********** Running on CPU **********\n")
+
+
+## ------------------------------ LOAD MODELS ------------------------------
+
+def load_face_analyser_model(name="buffalo_l"):
+    global FACE_ANALYSER
+    if FACE_ANALYSER is None:
+        FACE_ANALYSER = insightface.app.FaceAnalysis(name=name, providers=PROVIDER)
+        FACE_ANALYSER.prepare(
+            ctx_id=0, det_size=(DETECT_SIZE, DETECT_SIZE), det_thresh=DETECT_THRESH
+        )
+
+
+def load_face_swapper_model(name="./assets/pretrained_models/inswapper_128.onnx"):
+    global FACE_SWAPPER
+    path = os.path.join(os.path.abspath(os.path.dirname(__file__)), name)
+    if FACE_SWAPPER is None:
+        FACE_SWAPPER = insightface.model_zoo.get_model(path, providers=PROVIDER)
+
+
+def load_face_enhancer_model(name="./assets/pretrained_models/GFPGANv1.4.pth"):
+    global FACE_ENHANCER
+    path = os.path.join(os.path.abspath(os.path.dirname(__file__)), name)
+    if FACE_ENHANCER is None:
+        FACE_ENHANCER = gfpgan.GFPGANer(model_path=path, upscale=1)
+
+
+def load_face_parser_model(name="./assets/pretrained_models/79999_iter.pth"):
+    global FACE_PARSER
+    path = os.path.join(os.path.abspath(os.path.dirname(__file__)), name)
+    if FACE_PARSER is None:
+        FACE_PARSER = init_parser(name, use_cuda=USE_CUDA)
+
+
+load_face_analyser_model()
+load_face_swapper_model()
+
+## ------------------------------ MAIN PROCESS ------------------------------
+
+
+def process(
+    input_type,
+    image_path,
+    video_path,
+    directory_path,
+    source_path,
+    output_path,
+    output_name,
+    keep_output_sequence,
+    condition,
+    age,
+    distance,
+    face_enhance,
+    enable_face_parser,
+    mask_include,
+    mask_exclude,
+    mask_blur,
+    *specifics,
+):
+    global WORKSPACE
+    global OUTPUT_FILE
+    global PREVIEW
+    WORKSPACE, OUTPUT_FILE, PREVIEW = None, None, None
+
+    ## ------------------------------ GUI UPDATE FUNC ------------------------------
+
+    def ui_before():
+        return (
+            gr.update(visible=True, value=PREVIEW),
+            gr.update(interactive=False),
+            gr.update(interactive=False),
+            gr.update(visible=False),
+        )
+
+    def ui_after():
+        return (
+            gr.update(visible=True, value=PREVIEW),
+            gr.update(interactive=True),
+            gr.update(interactive=True),
+            gr.update(visible=False),
+        )
+
+    def ui_after_vid():
+        return (
+            gr.update(visible=False),
+            gr.update(interactive=True),
+            gr.update(interactive=True),
+            gr.update(value=OUTPUT_FILE, visible=True),
+        )
+
+    ## ------------------------------ LOAD PENDING MODELS ------------------------------
+    start_time = time.time()
+    specifics = list(specifics)
+    half = len(specifics) // 2
+    sources = specifics[:half]
+    specifics = specifics[half:]
+
+    yield "### \n ⌛ Loading face analyser model...", *ui_before()
+    load_face_analyser_model()
+
+    yield "### \n ⌛ Loading face swapper model...", *ui_before()
+    load_face_swapper_model()
+
+    if face_enhance:
+        yield "### \n ⌛ Loading face enhancer model...", *ui_before()
+        load_face_enhancer_model()
+
+    if enable_face_parser:
+        yield "### \n ⌛ Loading face parsing model...", *ui_before()
+        load_face_parser_model()
+
+    yield "### \n ⌛ Analysing Face...", *ui_before()
+
+    mi = mask_regions_to_list(mask_include)
+    me = mask_regions_to_list(mask_exclude)
+    models = {
+        "swap": FACE_SWAPPER,
+        "enhance": FACE_ENHANCER,
+        "enhance_sett": face_enhance,
+        "face_parser": FACE_PARSER,
+        "face_parser_sett": (enable_face_parser, mi, me, int(mask_blur)),
+    }
+
+    ## ------------------------------ ANALYSE SOURCE & SPECIFIC ------------------------------
+
+    analysed_source_specific = []
+    if condition == "Specific Face":
+        for source, specific in zip(sources, specifics):
+            if source is None or specific is None:
+                continue
+            analysed_source = analyse_face(
+                source,
+                FACE_ANALYSER,
+                return_single_face=True,
+                detect_condition=DETECT_CONDITION,
+            )
+            analysed_specific = analyse_face(
+                specific,
+                FACE_ANALYSER,
+                return_single_face=True,
+                detect_condition=DETECT_CONDITION,
+            )
+            analysed_source_specific.append([analysed_source, analysed_specific])
+    else:
+        source = cv2.imread(source_path)
+        analysed_source = analyse_face(
+            source,
+            FACE_ANALYSER,
+            return_single_face=True,
+            detect_condition=DETECT_CONDITION,
+        )
+
+    ## ------------------------------ IMAGE ------------------------------
+
+    if input_type == "Image":
+        target = cv2.imread(image_path)
+        analysed_target = analyse_face(target, FACE_ANALYSER, return_single_face=False)
+        if condition == "Specific Face":
+            swapped = swap_specific(
+                analysed_source_specific,
+                analysed_target,
+                target,
+                models,
+                threshold=distance,
+            )
+        else:
+            swapped = swap_face_with_condition(
+                target, analysed_target, analysed_source, condition, age, models
+            )
+
+        filename = os.path.join(output_path, output_name + ".png")
+        cv2.imwrite(filename, swapped)
+        OUTPUT_FILE = filename
+        WORKSPACE = output_path
+        PREVIEW = swapped[:, :, ::-1]
+
+        tot_exec_time = time.time() - start_time
+        _min, _sec = divmod(tot_exec_time, 60)
+
+        yield f"Completed in {int(_min)} min {int(_sec)} sec.", *ui_after()
+
+    ## ------------------------------ VIDEO ------------------------------
+
+    elif input_type == "Video":
+        temp_path = os.path.join(output_path, output_name, "sequence")
+        os.makedirs(temp_path, exist_ok=True)
+
+        video_clip = VideoFileClip(video_path)
+        duration = video_clip.duration
+        fps = video_clip.fps
+        total_frames = video_clip.reader.nframes
+
+        analysed_targets = []
+        process_bar = ProcessBar(30, total_frames)
+        yield "### \n ⌛ Analysing...", *ui_before()
+        for i, frame in enumerate(video_clip.iter_frames()):
+            analysed_targets.append(
+                analyse_face(frame, FACE_ANALYSER, return_single_face=False)
+            )
+            info_text = "Analysing Faces || "
+            info_text += process_bar.get(i)
+            print("\033[1A\033[K", end="", flush=True)
+            print(info_text)
+            if i % 10 == 0:
+                yield "### \n" + info_text, *ui_before()
+        video_clip.close()
+
+        image_sequence = []
+        video_clip = VideoFileClip(video_path)
+        audio_clip = video_clip.audio if video_clip.audio is not None else None
+        process_bar = ProcessBar(30, total_frames)
+        yield "### \n ⌛ Swapping...", *ui_before()
+        for i, frame in enumerate(video_clip.iter_frames()):
+            swapped = frame
+            analysed_target = analysed_targets[i]
+
+            if condition == "Specific Face":
+                swapped = swap_specific(
+                    frame,
+                    analysed_target,
+                    analysed_source_specific,
+                    models,
+                    threshold=distance,
+                )
+            else:
+                swapped = swap_face_with_condition(
+                    frame, analysed_target, analysed_source, condition, age, models
+                )
+
+            image_path = os.path.join(temp_path, f"frame_{i}.png")
+            cv2.imwrite(image_path, swapped[:, :, ::-1])
+            image_sequence.append(image_path)
+
+            info_text = "Swapping Faces || "
+            info_text += process_bar.get(i)
+            print("\033[1A\033[K", end="", flush=True)
+            print(info_text)
+            if i % 6 == 0:
+                PREVIEW = swapped
+                yield "### \n" + info_text, *ui_before()
+
+        yield "### \n ⌛ Merging...", *ui_before()
+        edited_video_clip = ImageSequenceClip(image_sequence, fps=fps)
+
+        if audio_clip is not None:
+            edited_video_clip = edited_video_clip.set_audio(audio_clip)
+
+        output_video_path = os.path.join(output_path, output_name + ".mp4")
+        edited_video_clip.set_duration(duration).write_videofile(
+            output_video_path, codec="libx264"
+        )
+        edited_video_clip.close()
+        video_clip.close()
+
+        if os.path.exists(temp_path) and not keep_output_sequence:
+            yield "### \n ⌛ Removing temporary files...", *ui_before()
+            shutil.rmtree(temp_path)
+
+        WORKSPACE = output_path
+        OUTPUT_FILE = output_video_path
+
+        tot_exec_time = time.time() - start_time
+        _min, _sec = divmod(tot_exec_time, 60)
+
+        yield f"✔️ Completed in {int(_min)} min {int(_sec)} sec.", *ui_after_vid()
+
+    ## ------------------------------ DIRECTORY ------------------------------
+
+    elif input_type == "Directory":
+        source = cv2.imread(source_path)
+        source = analyse_face(
+            source,
+            FACE_ANALYSER,
+            return_single_face=True,
+            detect_condition=DETECT_CONDITION,
+        )
+        extensions = ["jpg", "jpeg", "png", "bmp", "tiff", "ico", "webp"]
+        temp_path = os.path.join(output_path, output_name)
+        if os.path.exists(temp_path):
+            shutil.rmtree(temp_path)
+        os.mkdir(temp_path)
+        swapped = None
+
+        files = []
+        for file_path in glob.glob(os.path.join(directory_path, "*")):
+            if any(file_path.lower().endswith(ext) for ext in extensions):
+                files.append(file_path)
+
+        files_length = len(files)
+        filename = None
+        for i, file_path in enumerate(files):
+            target = cv2.imread(file_path)
+            analysed_target = analyse_face(
+                target, FACE_ANALYSER, return_single_face=False
+            )
+
+            if condition == "Specific Face":
+                swapped = swap_specific(
+                    target,
+                    analysed_target,
+                    analysed_source_specific,
+                    models,
+                    threshold=distance,
+                )
+            else:
+                swapped = swap_face_with_condition(
+                    target, analysed_target, analysed_source, condition, age, models
+                )
+
+            filename = os.path.join(temp_path, os.path.basename(file_path))
+            cv2.imwrite(filename, swapped)
+            info_text = f"### \n ⌛ Processing file {i+1} of {files_length}"
+            PREVIEW = swapped[:, :, ::-1]
+            yield info_text, *ui_before()
+
+        WORKSPACE = temp_path
+        OUTPUT_FILE = filename
+
+        tot_exec_time = time.time() - start_time
+        _min, _sec = divmod(tot_exec_time, 60)
+
+        yield f"✔️ Completed in {int(_min)} min {int(_sec)} sec.", *ui_after()
+
+    ## ------------------------------ STREAM ------------------------------
+
+    elif input_type == "Stream":
+        yield "### \n ⌛ Starting...", *ui_before()
+        global STREAMER
+        STREAMER = StreamerThread(src=directory_path)
+        STREAMER.start()
+
+        while True:
+            try:
+                target = STREAMER.frame
+                analysed_target = analyse_face(
+                    target, FACE_ANALYSER, return_single_face=False
+                )
+                if condition == "Specific Face":
+                    swapped = swap_specific(
+                        target,
+                        analysed_target,
+                        analysed_source_specific,
+                        models,
+                        threshold=distance,
+                    )
+                else:
+                    swapped = swap_face_with_condition(
+                        target, analysed_target, analysed_source, condition, age, models
+                    )
+                PREVIEW = swapped[:, :, ::-1]
+                yield f"Streaming...", *ui_before()
+            except AttributeError:
+                yield "Streaming...", *ui_before()
+        STREAMER.stop()
+
+
+## ------------------------------ GRADIO FUNC ------------------------------
+
+
+def update_radio(value):
+    if value == "Image":
+        return (
+            gr.update(visible=True),
+            gr.update(visible=False),
+            gr.update(visible=False),
+        )
+    elif value == "Video":
+        return (
+            gr.update(visible=False),
+            gr.update(visible=True),
+            gr.update(visible=False),
+        )
+    elif value == "Directory":
+        return (
+            gr.update(visible=False),
+            gr.update(visible=False),
+            gr.update(visible=True),
+        )
+    elif value == "Stream":
+        return (
+            gr.update(visible=False),
+            gr.update(visible=False),
+            gr.update(visible=True),
+        )
+
+
+def swap_option_changed(value):
+    if value == swap_options_list[1] or value == swap_options_list[2]:
+        return (
+            gr.update(visible=True),
+            gr.update(visible=False),
+            gr.update(visible=True),
+        )
+    elif value == swap_options_list[5]:
+        return (
+            gr.update(visible=False),
+            gr.update(visible=True),
+            gr.update(visible=False),
+        )
+    return gr.update(visible=False), gr.update(visible=False), gr.update(visible=True)
+
+
+def video_changed(video_path):
+    sliders_update = gr.Slider.update
+    button_update = gr.Button.update
+    number_update = gr.Number.update
+
+    if video_path is None:
+        return (
+            sliders_update(minimum=0, maximum=0, value=0),
+            sliders_update(minimum=1, maximum=1, value=1),
+            number_update(value=1),
+        )
+    try:
+        clip = VideoFileClip(video_path)
+        fps = clip.fps
+        total_frames = clip.reader.nframes
+        clip.close()
+        return (
+            sliders_update(minimum=0, maximum=total_frames, value=0, interactive=True),
+            sliders_update(
+                minimum=0, maximum=total_frames, value=total_frames, interactive=True
+            ),
+            number_update(value=fps),
+        )
+    except:
+        return (
+            sliders_update(value=0),
+            sliders_update(value=0),
+            number_update(value=1),
+        )
+
+
+def analyse_settings_changed(detect_condition, detection_size, detection_threshold):
+    yield "### \n ⌛ Applying new values..."
+    global FACE_ANALYSER
+    global DETECT_CONDITION
+    DETECT_CONDITION = detect_condition
+    FACE_ANALYSER = insightface.app.FaceAnalysis(name="buffalo_l", providers=PROVIDER)
+    FACE_ANALYSER.prepare(
+        ctx_id=0,
+        det_size=(int(detection_size), int(detection_size)),
+        det_thresh=float(detection_threshold),
+    )
+    yield f"### \n ✔️ Applied detect condition:{detect_condition}, detection size: {detection_size}, detection threshold: {detection_threshold}"
+
+
+def stop_running():
+    global STREAMER
+    if hasattr(STREAMER, "stop"):
+        STREAMER.stop()
+        STREAMER = None
+    return "Cancelled"
+
+
+def slider_changed(show_frame, video_path, frame_index):
+    if not show_frame:
+        return None, None
+    if video_path is None:
+        return None, None
+    clip = VideoFileClip(video_path)
+    frame = clip.get_frame(frame_index / clip.fps)
+    frame_array = np.array(frame)
+    clip.close()
+    return gr.Image.update(value=frame_array, visible=True), gr.Video.update(
+        visible=False
+    )
+
+
+def trim_and_reload(video_path, output_path, output_name, start_frame, stop_frame):
+    yield video_path, f"### \n ⌛ Trimming video frame {start_frame} to {stop_frame}..."
+    try:
+        output_path = os.path.join(output_path, output_name)
+        trimmed_video = trim_video(video_path, output_path, start_frame, stop_frame)
+        yield trimmed_video, "### \n ✔️ Video trimmed and reloaded."
+    except Exception as e:
+        print(e)
+        yield video_path, "### \n ❌ Video trimming failed. See console for more info."
+
+
+## ------------------------------ GRADIO GUI ------------------------------
+
+css = """
+footer{display:none !important}
+"""
+
+with gr.Blocks(css=css) as interface:
+    gr.Markdown("# 🗿 Swap Mukham")
+    gr.Markdown("### Face swap app based on insightface inswapper.")
+    with gr.Row():
+        with gr.Row():
+            with gr.Column(scale=0.4):
+                with gr.Tab("📄 Swap Condition"):
+                    swap_option = gr.Radio(
+                        swap_options_list,
+                        show_label=False,
+                        value=swap_options_list[0],
+                        interactive=True,
+                    )
+                    age = gr.Number(
+                        value=25, label="Value", interactive=True, visible=False
+                    )
+
+                with gr.Tab("🎚️ Detection Settings"):
+                    detect_condition_dropdown = gr.Dropdown(
+                        detect_conditions,
+                        label="Condition",
+                        value=DETECT_CONDITION,
+                        interactive=True,
+                        info="This condition is only used when multiple faces are detected on source or specific image.",
+                    )
+                    detection_size = gr.Number(
+                        label="Detection Size", value=DETECT_SIZE, interactive=True
+                    )
+                    detection_threshold = gr.Number(
+                        label="Detection Threshold",
+                        value=DETECT_THRESH,
+                        interactive=True,
+                    )
+                    apply_detection_settings = gr.Button("Apply settings")
+
+                with gr.Tab("📤 Output Settings"):
+                    output_directory = gr.Text(
+                        label="Output Directory",
+                        value=DEF_OUTPUT_PATH,
+                        interactive=True,
+                    )
+                    output_name = gr.Text(
+                        label="Output Name", value="Result", interactive=True
+                    )
+                    keep_output_sequence = gr.Checkbox(
+                        label="Keep output sequence", value=False, interactive=True
+                    )
+
+                with gr.Tab("🪄 Other Settings"):
+                    with gr.Accordion("Enhance Face", open=True):
+                        enable_face_enhance = gr.Checkbox(
+                            label="Enable GFPGAN", value=False, interactive=True
+                        )
+                    with gr.Accordion("Advanced Mask", open=False):
+                        enable_face_parser_mask = gr.Checkbox(
+                            label="Enable Face Parsing",
+                            value=False,
+                            interactive=True,
+                        )
+
+                        mask_include = gr.Dropdown(
+                            mask_regions.keys(),
+                            value=MASK_INCLUDE,
+                            multiselect=True,
+                            label="Include",
+                            interactive=True,
+                        )
+                        mask_exclude = gr.Dropdown(
+                            mask_regions.keys(),
+                            value=MASK_EXCLUDE,
+                            multiselect=True,
+                            label="Exclude",
+                            interactive=True,
+                        )
+                        mask_blur = gr.Number(
+                            label="Blur Mask",
+                            value=MASK_BLUR,
+                            minimum=0,
+                            interactive=True,
+                        )
+
+                source_image_input = gr.Image(
+                    label="Source face", type="filepath", interactive=True
+                )
+
+                with gr.Box(visible=False) as specific_face:
+                    for i in range(NUM_OF_SRC_SPECIFIC):
+                        idx = i + 1
+                        code = "\n"
+                        code += f"with gr.Tab(label='({idx})'):"
+                        code += "\n\twith gr.Row():"
+                        code += f"\n\t\tsrc{idx} = gr.Image(interactive=True, type='numpy', label='Source Face {idx}')"
+                        code += f"\n\t\ttrg{idx} = gr.Image(interactive=True, type='numpy', label='Specific Face {idx}')"
+                        exec(code)
+
+                    distance_slider = gr.Slider(
+                        minimum=0,
+                        maximum=2,
+                        value=0.6,
+                        interactive=True,
+                        label="Distance",
+                        info="Lower distance is more similar and higher distance is less similar to the target face.",
+                    )
+
+                with gr.Group():
+                    input_type = gr.Radio(
+                        ["Image", "Video", "Directory", "Stream"],
+                        label="Target Type",
+                        value="Video",
+                    )
+
+                    with gr.Box(visible=False) as input_image_group:
+                        image_input = gr.Image(
+                            label="Target Image", interactive=True, type="filepath"
+                        )
+
+                    with gr.Box(visible=True) as input_video_group:
+                        vid_widget = gr.Video if USE_COLAB else gr.Text
+                        video_input = vid_widget(
+                            label="Target Video Path", interactive=True
+                        )
+                        with gr.Accordion("✂️ Trim video", open=False):
+                            with gr.Column():
+                                with gr.Row():
+                                    set_slider_range_btn = gr.Button(
+                                        "Set frame range", interactive=True
+                                    )
+                                    show_trim_preview_btn = gr.Checkbox(
+                                        label="Show frame when slider change",
+                                        value=True,
+                                        interactive=True,
+                                    )
+
+                                video_fps = gr.Number(
+                                    value=30,
+                                    interactive=False,
+                                    label="Fps",
+                                    visible=False,
+                                )
+                                start_frame = gr.Slider(
+                                    minimum=0,
+                                    maximum=1,
+                                    value=0,
+                                    step=1,
+                                    interactive=True,
+                                    label="Start Frame",
+                                    info="",
+                                )
+                                end_frame = gr.Slider(
+                                    minimum=0,
+                                    maximum=1,
+                                    value=1,
+                                    step=1,
+                                    interactive=True,
+                                    label="End Frame",
+                                    info="",
+                                )
+                            trim_and_reload_btn = gr.Button(
+                                "Trim and Reload", interactive=True
+                            )
+
+                    with gr.Box(visible=False) as input_directory_group:
+                        direc_input = gr.Text(label="Path", interactive=True)
+
+            with gr.Column(scale=0.6):
+                info = gr.Markdown(value="...")
+
+                with gr.Row():
+                    swap_button = gr.Button("✨ Swap", variant="primary")
+                    cancel_button = gr.Button("⛔ Cancel")
+
+                preview_image = gr.Image(label="Output", interactive=False)
+                preview_video = gr.Video(
+                    label="Output", interactive=False, visible=False
+                )
+
+                with gr.Row():
+                    output_directory_button = gr.Button(
+                        "📂", interactive=False, visible=not USE_COLAB
+                    )
+                    output_video_button = gr.Button(
+                        "🎬", interactive=False, visible=not USE_COLAB
+                    )
+
+                with gr.Column():
+                    gr.Markdown(
+                        '[!["Buy Me A Coffee"](https://www.buymeacoffee.com/assets/img/custom_images/orange_img.png)](https://www.buymeacoffee.com/harisreedhar)'
+                    )
+                    gr.Markdown(
+                        "### [Source code](https://github.com/harisreedhar/Swap-Mukham) . [Disclaimer](https://github.com/harisreedhar/Swap-Mukham#disclaimer) . [Gradio](https://gradio.app/)"
+                    )
+
+    ## ------------------------------ GRADIO EVENTS ------------------------------
+
+    set_slider_range_event = set_slider_range_btn.click(
+        video_changed,
+        inputs=[video_input],
+        outputs=[start_frame, end_frame, video_fps],
+    )
+
+    trim_and_reload_event = trim_and_reload_btn.click(
+        fn=trim_and_reload,
+        inputs=[video_input, output_directory, output_name, start_frame, end_frame],
+        outputs=[video_input, info],
+    )
+
+    start_frame_event = start_frame.release(
+        fn=slider_changed,
+        inputs=[show_trim_preview_btn, video_input, start_frame],
+        outputs=[preview_image, preview_video],
+        show_progress=False,
+    )
+
+    end_frame_event = end_frame.release(
+        fn=slider_changed,
+        inputs=[show_trim_preview_btn, video_input, end_frame],
+        outputs=[preview_image, preview_video],
+        show_progress=False,
+    )
+
+    input_type.change(
+        update_radio,
+        inputs=[input_type],
+        outputs=[input_image_group, input_video_group, input_directory_group],
+    )
+    swap_option.change(
+        swap_option_changed,
+        inputs=[swap_option],
+        outputs=[age, specific_face, source_image_input],
+    )
+
+    apply_detection_settings.click(
+        analyse_settings_changed,
+        inputs=[detect_condition_dropdown, detection_size, detection_threshold],
+        outputs=[info],
+    )
+
+    src_specific_inputs = []
+    gen_variable_txt = ",".join(
+        [f"src{i+1}" for i in range(NUM_OF_SRC_SPECIFIC)]
+        + [f"trg{i+1}" for i in range(NUM_OF_SRC_SPECIFIC)]
+    )
+    exec(f"src_specific_inputs = ({gen_variable_txt})")
+    swap_inputs = [
+        input_type,
+        image_input,
+        video_input,
+        direc_input,
+        source_image_input,
+        output_directory,
+        output_name,
+        keep_output_sequence,
+        swap_option,
+        age,
+        distance_slider,
+        enable_face_enhance,
+        enable_face_parser_mask,
+        mask_include,
+        mask_exclude,
+        mask_blur,
+        *src_specific_inputs,
+    ]
+
+    swap_outputs = [
+        info,
+        preview_image,
+        output_directory_button,
+        output_video_button,
+        preview_video,
+    ]
+
+    swap_event = swap_button.click(
+        fn=process, inputs=swap_inputs, outputs=swap_outputs, show_progress=False
+    )
+
+    cancel_button.click(
+        fn=stop_running,
+        inputs=None,
+        outputs=[info],
+        cancels=[
+            swap_event,
+            trim_and_reload_event,
+            set_slider_range_event,
+            start_frame_event,
+            end_frame_event,
+        ],
+        show_progress=False,
+    )
+    output_directory_button.click(
+        lambda: open_directory(path=WORKSPACE), inputs=None, outputs=None
+    )
+    output_video_button.click(
+        lambda: open_directory(path=OUTPUT_FILE), inputs=None, outputs=None
+    )
+
+if __name__ == "__main__":
+    if USE_COLAB:
+        print("Running in colab mode")
+
+    interface.queue(concurrency_count=2, max_size=20).launch(share=USE_COLAB)

assets/pretrained_models/79999_iter.pth

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

assets/pretrained_models/GFPGANv1.4.pth

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

assets/pretrained_models/inswapper_128.onnx

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

assets/pretrained_models/readme.md

@@ -0,0 +1,4 @@
+## Downolad these models here
+- [inswapper_128.onnx](https://huggingface.co/deepinsight/inswapper/resolve/main/inswapper_128.onnx)
+- [GFPGANv1.4.pth](https://github.com/TencentARC/GFPGAN/releases/download/v1.3.0/GFPGANv1.4.pth)
+- [79999_iter.pth](https://drive.google.com/open?id=154JgKpzCPW82qINcVieuPH3fZ2e0P812)

face_analyser.py

@@ -0,0 +1,32 @@
+detect_conditions = [
+    "left most",
+    "right most",
+    "top most",
+    "bottom most",
+    "most width",
+    "most height",
+]
+
+
+def analyse_face(image, model, return_single_face=True, detect_condition="left most"):
+    faces = model.get(image)
+    if not return_single_face:
+        return faces
+
+    total_faces = len(faces)
+    if total_faces == 1:
+        return faces[0]
+
+    print(f"{total_faces} face detected. Using {detect_condition} face.")
+    if detect_condition == "left most":
+        return sorted(faces, key=lambda face: face["bbox"][0])[0]
+    elif detect_condition == "right most":
+        return sorted(faces, key=lambda face: face["bbox"][0])[-1]
+    elif detect_condition == "top most":
+        return sorted(faces, key=lambda face: face["bbox"][1])[0]
+    elif detect_condition == "bottom most":
+        return sorted(faces, key=lambda face: face["bbox"][1])[-1]
+    elif detect_condition == "most width":
+        return sorted(faces, key=lambda face: face["bbox"][2])[-1]
+    elif detect_condition == "most height":
+        return sorted(faces, key=lambda face: face["bbox"][3])[-1]

face_parsing/__init__.py

@@ -0,0 +1 @@
+from .swap import init_parser, swap_regions, mask_regions, mask_regions_to_list

face_parsing/model.py

@@ -0,0 +1,283 @@
+#!/usr/bin/python
+# -*- encoding: utf-8 -*-
+
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import torchvision
+
+from .resnet import Resnet18
+# from modules.bn import InPlaceABNSync as BatchNorm2d
+
+
+class ConvBNReLU(nn.Module):
+    def __init__(self, in_chan, out_chan, ks=3, stride=1, padding=1, *args, **kwargs):
+        super(ConvBNReLU, self).__init__()
+        self.conv = nn.Conv2d(in_chan,
+                out_chan,
+                kernel_size = ks,
+                stride = stride,
+                padding = padding,
+                bias = False)
+        self.bn = nn.BatchNorm2d(out_chan)
+        self.init_weight()
+
+    def forward(self, x):
+        x = self.conv(x)
+        x = F.relu(self.bn(x))
+        return x
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+class BiSeNetOutput(nn.Module):
+    def __init__(self, in_chan, mid_chan, n_classes, *args, **kwargs):
+        super(BiSeNetOutput, self).__init__()
+        self.conv = ConvBNReLU(in_chan, mid_chan, ks=3, stride=1, padding=1)
+        self.conv_out = nn.Conv2d(mid_chan, n_classes, kernel_size=1, bias=False)
+        self.init_weight()
+
+    def forward(self, x):
+        x = self.conv(x)
+        x = self.conv_out(x)
+        return x
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+    def get_params(self):
+        wd_params, nowd_params = [], []
+        for name, module in self.named_modules():
+            if isinstance(module, nn.Linear) or isinstance(module, nn.Conv2d):
+                wd_params.append(module.weight)
+                if not module.bias is None:
+                    nowd_params.append(module.bias)
+            elif isinstance(module, nn.BatchNorm2d):
+                nowd_params += list(module.parameters())
+        return wd_params, nowd_params
+
+
+class AttentionRefinementModule(nn.Module):
+    def __init__(self, in_chan, out_chan, *args, **kwargs):
+        super(AttentionRefinementModule, self).__init__()
+        self.conv = ConvBNReLU(in_chan, out_chan, ks=3, stride=1, padding=1)
+        self.conv_atten = nn.Conv2d(out_chan, out_chan, kernel_size= 1, bias=False)
+        self.bn_atten = nn.BatchNorm2d(out_chan)
+        self.sigmoid_atten = nn.Sigmoid()
+        self.init_weight()
+
+    def forward(self, x):
+        feat = self.conv(x)
+        atten = F.avg_pool2d(feat, feat.size()[2:])
+        atten = self.conv_atten(atten)
+        atten = self.bn_atten(atten)
+        atten = self.sigmoid_atten(atten)
+        out = torch.mul(feat, atten)
+        return out
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+
+class ContextPath(nn.Module):
+    def __init__(self, *args, **kwargs):
+        super(ContextPath, self).__init__()
+        self.resnet = Resnet18()
+        self.arm16 = AttentionRefinementModule(256, 128)
+        self.arm32 = AttentionRefinementModule(512, 128)
+        self.conv_head32 = ConvBNReLU(128, 128, ks=3, stride=1, padding=1)
+        self.conv_head16 = ConvBNReLU(128, 128, ks=3, stride=1, padding=1)
+        self.conv_avg = ConvBNReLU(512, 128, ks=1, stride=1, padding=0)
+
+        self.init_weight()
+
+    def forward(self, x):
+        H0, W0 = x.size()[2:]
+        feat8, feat16, feat32 = self.resnet(x)
+        H8, W8 = feat8.size()[2:]
+        H16, W16 = feat16.size()[2:]
+        H32, W32 = feat32.size()[2:]
+
+        avg = F.avg_pool2d(feat32, feat32.size()[2:])
+        avg = self.conv_avg(avg)
+        avg_up = F.interpolate(avg, (H32, W32), mode='nearest')
+
+        feat32_arm = self.arm32(feat32)
+        feat32_sum = feat32_arm + avg_up
+        feat32_up = F.interpolate(feat32_sum, (H16, W16), mode='nearest')
+        feat32_up = self.conv_head32(feat32_up)
+
+        feat16_arm = self.arm16(feat16)
+        feat16_sum = feat16_arm + feat32_up
+        feat16_up = F.interpolate(feat16_sum, (H8, W8), mode='nearest')
+        feat16_up = self.conv_head16(feat16_up)
+
+        return feat8, feat16_up, feat32_up  # x8, x8, x16
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+    def get_params(self):
+        wd_params, nowd_params = [], []
+        for name, module in self.named_modules():
+            if isinstance(module, (nn.Linear, nn.Conv2d)):
+                wd_params.append(module.weight)
+                if not module.bias is None:
+                    nowd_params.append(module.bias)
+            elif isinstance(module, nn.BatchNorm2d):
+                nowd_params += list(module.parameters())
+        return wd_params, nowd_params
+
+
+### This is not used, since I replace this with the resnet feature with the same size
+class SpatialPath(nn.Module):
+    def __init__(self, *args, **kwargs):
+        super(SpatialPath, self).__init__()
+        self.conv1 = ConvBNReLU(3, 64, ks=7, stride=2, padding=3)
+        self.conv2 = ConvBNReLU(64, 64, ks=3, stride=2, padding=1)
+        self.conv3 = ConvBNReLU(64, 64, ks=3, stride=2, padding=1)
+        self.conv_out = ConvBNReLU(64, 128, ks=1, stride=1, padding=0)
+        self.init_weight()
+
+    def forward(self, x):
+        feat = self.conv1(x)
+        feat = self.conv2(feat)
+        feat = self.conv3(feat)
+        feat = self.conv_out(feat)
+        return feat
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+    def get_params(self):
+        wd_params, nowd_params = [], []
+        for name, module in self.named_modules():
+            if isinstance(module, nn.Linear) or isinstance(module, nn.Conv2d):
+                wd_params.append(module.weight)
+                if not module.bias is None:
+                    nowd_params.append(module.bias)
+            elif isinstance(module, nn.BatchNorm2d):
+                nowd_params += list(module.parameters())
+        return wd_params, nowd_params
+
+
+class FeatureFusionModule(nn.Module):
+    def __init__(self, in_chan, out_chan, *args, **kwargs):
+        super(FeatureFusionModule, self).__init__()
+        self.convblk = ConvBNReLU(in_chan, out_chan, ks=1, stride=1, padding=0)
+        self.conv1 = nn.Conv2d(out_chan,
+                out_chan//4,
+                kernel_size = 1,
+                stride = 1,
+                padding = 0,
+                bias = False)
+        self.conv2 = nn.Conv2d(out_chan//4,
+                out_chan,
+                kernel_size = 1,
+                stride = 1,
+                padding = 0,
+                bias = False)
+        self.relu = nn.ReLU(inplace=True)
+        self.sigmoid = nn.Sigmoid()
+        self.init_weight()
+
+    def forward(self, fsp, fcp):
+        fcat = torch.cat([fsp, fcp], dim=1)
+        feat = self.convblk(fcat)
+        atten = F.avg_pool2d(feat, feat.size()[2:])
+        atten = self.conv1(atten)
+        atten = self.relu(atten)
+        atten = self.conv2(atten)
+        atten = self.sigmoid(atten)
+        feat_atten = torch.mul(feat, atten)
+        feat_out = feat_atten + feat
+        return feat_out
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+    def get_params(self):
+        wd_params, nowd_params = [], []
+        for name, module in self.named_modules():
+            if isinstance(module, nn.Linear) or isinstance(module, nn.Conv2d):
+                wd_params.append(module.weight)
+                if not module.bias is None:
+                    nowd_params.append(module.bias)
+            elif isinstance(module, nn.BatchNorm2d):
+                nowd_params += list(module.parameters())
+        return wd_params, nowd_params
+
+
+class BiSeNet(nn.Module):
+    def __init__(self, n_classes, *args, **kwargs):
+        super(BiSeNet, self).__init__()
+        self.cp = ContextPath()
+        ## here self.sp is deleted
+        self.ffm = FeatureFusionModule(256, 256)
+        self.conv_out = BiSeNetOutput(256, 256, n_classes)
+        self.conv_out16 = BiSeNetOutput(128, 64, n_classes)
+        self.conv_out32 = BiSeNetOutput(128, 64, n_classes)
+        self.init_weight()
+
+    def forward(self, x):
+        H, W = x.size()[2:]
+        feat_res8, feat_cp8, feat_cp16 = self.cp(x)  # here return res3b1 feature
+        feat_sp = feat_res8  # use res3b1 feature to replace spatial path feature
+        feat_fuse = self.ffm(feat_sp, feat_cp8)
+
+        feat_out = self.conv_out(feat_fuse)
+        feat_out16 = self.conv_out16(feat_cp8)
+        feat_out32 = self.conv_out32(feat_cp16)
+
+        feat_out = F.interpolate(feat_out, (H, W), mode='bilinear', align_corners=True)
+        feat_out16 = F.interpolate(feat_out16, (H, W), mode='bilinear', align_corners=True)
+        feat_out32 = F.interpolate(feat_out32, (H, W), mode='bilinear', align_corners=True)
+        return feat_out, feat_out16, feat_out32
+
+    def init_weight(self):
+        for ly in self.children():
+            if isinstance(ly, nn.Conv2d):
+                nn.init.kaiming_normal_(ly.weight, a=1)
+                if not ly.bias is None: nn.init.constant_(ly.bias, 0)
+
+    def get_params(self):
+        wd_params, nowd_params, lr_mul_wd_params, lr_mul_nowd_params = [], [], [], []
+        for name, child in self.named_children():
+            child_wd_params, child_nowd_params = child.get_params()
+            if isinstance(child, FeatureFusionModule) or isinstance(child, BiSeNetOutput):
+                lr_mul_wd_params += child_wd_params
+                lr_mul_nowd_params += child_nowd_params
+            else:
+                wd_params += child_wd_params
+                nowd_params += child_nowd_params
+        return wd_params, nowd_params, lr_mul_wd_params, lr_mul_nowd_params
+
+
+if __name__ == "__main__":
+    net = BiSeNet(19)
+    net.cuda()
+    net.eval()
+    in_ten = torch.randn(16, 3, 640, 480).cuda()
+    out, out16, out32 = net(in_ten)
+    print(out.shape)
+
+    net.get_params()

face_parsing/resnet.py

@@ -0,0 +1,109 @@
+#!/usr/bin/python
+# -*- encoding: utf-8 -*-
+
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+import torch.utils.model_zoo as modelzoo
+
+# from modules.bn import InPlaceABNSync as BatchNorm2d
+
+resnet18_url = 'https://download.pytorch.org/models/resnet18-5c106cde.pth'
+
+
+def conv3x3(in_planes, out_planes, stride=1):
+    """3x3 convolution with padding"""
+    return nn.Conv2d(in_planes, out_planes, kernel_size=3, stride=stride,
+                     padding=1, bias=False)
+
+
+class BasicBlock(nn.Module):
+    def __init__(self, in_chan, out_chan, stride=1):
+        super(BasicBlock, self).__init__()
+        self.conv1 = conv3x3(in_chan, out_chan, stride)
+        self.bn1 = nn.BatchNorm2d(out_chan)
+        self.conv2 = conv3x3(out_chan, out_chan)
+        self.bn2 = nn.BatchNorm2d(out_chan)
+        self.relu = nn.ReLU(inplace=True)
+        self.downsample = None
+        if in_chan != out_chan or stride != 1:
+            self.downsample = nn.Sequential(
+                nn.Conv2d(in_chan, out_chan,
+                          kernel_size=1, stride=stride, bias=False),
+                nn.BatchNorm2d(out_chan),
+                )
+
+    def forward(self, x):
+        residual = self.conv1(x)
+        residual = F.relu(self.bn1(residual))
+        residual = self.conv2(residual)
+        residual = self.bn2(residual)
+
+        shortcut = x
+        if self.downsample is not None:
+            shortcut = self.downsample(x)
+
+        out = shortcut + residual
+        out = self.relu(out)
+        return out
+
+
+def create_layer_basic(in_chan, out_chan, bnum, stride=1):
+    layers = [BasicBlock(in_chan, out_chan, stride=stride)]
+    for i in range(bnum-1):
+        layers.append(BasicBlock(out_chan, out_chan, stride=1))
+    return nn.Sequential(*layers)
+
+
+class Resnet18(nn.Module):
+    def __init__(self):
+        super(Resnet18, self).__init__()
+        self.conv1 = nn.Conv2d(3, 64, kernel_size=7, stride=2, padding=3,
+                               bias=False)
+        self.bn1 = nn.BatchNorm2d(64)
+        self.maxpool = nn.MaxPool2d(kernel_size=3, stride=2, padding=1)
+        self.layer1 = create_layer_basic(64, 64, bnum=2, stride=1)
+        self.layer2 = create_layer_basic(64, 128, bnum=2, stride=2)
+        self.layer3 = create_layer_basic(128, 256, bnum=2, stride=2)
+        self.layer4 = create_layer_basic(256, 512, bnum=2, stride=2)
+        self.init_weight()
+
+    def forward(self, x):
+        x = self.conv1(x)
+        x = F.relu(self.bn1(x))
+        x = self.maxpool(x)
+
+        x = self.layer1(x)
+        feat8 = self.layer2(x) # 1/8
+        feat16 = self.layer3(feat8) # 1/16
+        feat32 = self.layer4(feat16) # 1/32
+        return feat8, feat16, feat32
+
+    def init_weight(self):
+        state_dict = modelzoo.load_url(resnet18_url)
+        self_state_dict = self.state_dict()
+        for k, v in state_dict.items():
+            if 'fc' in k: continue
+            self_state_dict.update({k: v})
+        self.load_state_dict(self_state_dict)
+
+    def get_params(self):
+        wd_params, nowd_params = [], []
+        for name, module in self.named_modules():
+            if isinstance(module, (nn.Linear, nn.Conv2d)):
+                wd_params.append(module.weight)
+                if not module.bias is None:
+                    nowd_params.append(module.bias)
+            elif isinstance(module,  nn.BatchNorm2d):
+                nowd_params += list(module.parameters())
+        return wd_params, nowd_params
+
+
+if __name__ == "__main__":
+    net = Resnet18()
+    x = torch.randn(16, 3, 224, 224)
+    out = net(x)
+    print(out[0].size())
+    print(out[1].size())
+    print(out[2].size())
+    net.get_params()

face_parsing/swap.py

@@ -0,0 +1,91 @@
+import torch
+import torchvision.transforms as transforms
+import cv2
+import numpy as np
+
+from .model import BiSeNet
+
+mask_regions = {
+    "Background":0,
+    "Skin":1,
+    "L-Eyebrow":2,
+    "R-Eyebrow":3,
+    "L-Eye":4,
+    "R-Eye":5,
+    "Eye-G":6,
+    "L-Ear":7,
+    "R-Ear":8,
+    "Ear-R":9,
+    "Nose":10,
+    "Mouth":11,
+    "U-Lip":12,
+    "L-Lip":13,
+    "Neck":14,
+    "Neck-L":15,
+    "Cloth":16,
+    "Hair":17,
+    "Hat":18
+}
+
+run_with_cuda = False
+
+def init_parser(pth_path, use_cuda=False):
+    global run_with_cuda
+    run_with_cuda = use_cuda
+
+    n_classes = 19
+    net = BiSeNet(n_classes=n_classes)
+    if run_with_cuda:
+        net.cuda()
+        net.load_state_dict(torch.load(pth_path))
+    else:
+        net.load_state_dict(torch.load(pth_path, map_location=torch.device('cpu')))
+    net.eval()
+    return net
+
+
+def image_to_parsing(img, net):
+    img = cv2.resize(img, (512, 512))
+    img = img[:,:,::-1]
+    transform = transforms.Compose([
+        transforms.ToTensor(),
+        transforms.Normalize((0.485, 0.456, 0.406), (0.229, 0.224, 0.225))
+    ])
+    img = transform(img.copy())
+    img = torch.unsqueeze(img, 0)
+
+    with torch.no_grad():
+        if run_with_cuda:
+            img = img.cuda()
+        out = net(img)[0]
+        parsing = out.squeeze(0).cpu().numpy().argmax(0)
+        return parsing
+
+
+def get_mask(parsing, classes):
+    res = parsing == classes[0]
+    for val in classes[1:]:
+        res += parsing == val
+    return res
+
+def swap_regions(source, target, net, includes=[1,2,3,4,5,10,11,12,13], excludes=[7,8], blur_size=25):
+    parsing = image_to_parsing(source, net)
+    if len(includes) == 0:
+        return source, np.zeros_like(source)
+    include_mask = get_mask(parsing, includes)
+    include_mask = np.repeat(np.expand_dims(include_mask.astype('float32'), axis=2), 3, 2)
+    if len(excludes) > 0:
+        exclude_mask = get_mask(parsing, excludes)
+        exclude_mask = np.repeat(np.expand_dims(exclude_mask.astype('float32'), axis=2), 3, 2)
+        include_mask -= exclude_mask
+    mask = 1 - cv2.GaussianBlur(include_mask.clip(0,1), (0, 0), blur_size)
+    result = (1 - mask) * cv2.resize(source, (512, 512)) + mask * cv2.resize(target, (512, 512))
+    result = cv2.resize(result.astype("float32"), (source.shape[1], source.shape[0]))
+    return result, mask.astype('float32')
+
+def mask_regions_to_list(values):
+    out_ids = []
+    for value in values:
+        if value in mask_regions.keys():
+            out_ids.append(mask_regions.get(value))
+    return out_ids

gfpgan/weights/detection_Resnet50_Final.pth

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

gfpgan/weights/parsing_parsenet.pth

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

requirements.txt

@@ -0,0 +1,10 @@
+torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
+gradio>=3.33.1
+insightface==0.7.3
+moviepy>=1.0.3
+numpy
+opencv-python>=4.7.0.72
+opencv-python-headless>=4.7.0.72
+onnx==1.14.0
+onnxruntime==1.15.0
+gfpgan==1.3.8

swapper.py

@@ -0,0 +1,106 @@
+import cv2
+import numpy as np
+from insightface.utils import face_align
+from face_parsing.swap import swap_regions
+from utils import add_logo_to_image
+
+swap_options_list = [
+    "All face",
+    "Age less than",
+    "Age greater than",
+    "All Male",
+    "All Female",
+    "Specific Face",
+]
+
+
+def swap_face(whole_img, target_face, source_face, models):
+    inswapper = models.get("swap")
+    face_enhancer = models.get("enhance", None)
+    face_parser = models.get("face_parser", None)
+    fe_enable = models.get("enhance_sett", False)
+
+    bgr_fake, M = inswapper.get(whole_img, target_face, source_face, paste_back=False)
+    image_size = 128 if not fe_enable else 512
+    aimg, _ = face_align.norm_crop2(whole_img, target_face.kps, image_size=image_size)
+
+    if face_parser is not None:
+        fp_enable, mi, me, mb = models.get("face_parser_sett")
+        if fp_enable:
+            bgr_fake, parsed_mask = swap_regions(
+                bgr_fake, aimg, face_parser, includes=mi, excludes=me, blur_size=mb
+            )
+
+    if fe_enable:
+        _, bgr_fake, _ = face_enhancer.enhance(
+            bgr_fake, paste_back=True, has_aligned=True
+        )
+        bgr_fake = bgr_fake[0]
+        M /= 0.25
+
+    IM = cv2.invertAffineTransform(M)
+
+    img_white = np.full((aimg.shape[0], aimg.shape[1]), 255, dtype=np.float32)
+    bgr_fake = cv2.warpAffine(
+        bgr_fake, IM, (whole_img.shape[1], whole_img.shape[0]), borderValue=0.0
+    )
+    img_white = cv2.warpAffine(
+        img_white, IM, (whole_img.shape[1], whole_img.shape[0]), borderValue=0.0
+    )
+    img_white[img_white > 20] = 255
+    img_mask = img_white
+    mask_h_inds, mask_w_inds = np.where(img_mask == 255)
+    mask_h = np.max(mask_h_inds) - np.min(mask_h_inds)
+    mask_w = np.max(mask_w_inds) - np.min(mask_w_inds)
+    mask_size = int(np.sqrt(mask_h * mask_w))
+
+    k = max(mask_size // 10, 10)
+    img_mask = cv2.erode(img_mask, np.ones((k, k), np.uint8), iterations=1)
+
+    k = max(mask_size // 20, 5)
+    kernel_size = (k, k)
+    blur_size = tuple(2 * i + 1 for i in kernel_size)
+    img_mask = cv2.GaussianBlur(img_mask, blur_size, 0) / 255
+
+    img_mask = np.reshape(img_mask, [img_mask.shape[0], img_mask.shape[1], 1])
+    fake_merged = img_mask * bgr_fake + (1 - img_mask) * whole_img.astype(np.float32)
+    fake_merged = add_logo_to_image(fake_merged.astype("uint8"))
+    return fake_merged
+
+
+def swap_face_with_condition(
+    whole_img, target_faces, source_face, condition, age, models
+):
+    swapped = whole_img.copy()
+
+    for target_face in target_faces:
+        if condition == "All face":
+            swapped = swap_face(swapped, target_face, source_face, models)
+        elif condition == "Age less than" and target_face["age"] < age:
+            swapped = swap_face(swapped, target_face, source_face, models)
+        elif condition == "Age greater than" and target_face["age"] > age:
+            swapped = swap_face(swapped, target_face, source_face, models)
+        elif condition == "All Male" and target_face["gender"] == 1:
+            swapped = swap_face(swapped, target_face, source_face, models)
+        elif condition == "All Female" and target_face["gender"] == 0:
+            swapped = swap_face(swapped, target_face, source_face, models)
+
+    return swapped
+
+
+def swap_specific(source_specifics, target_faces, whole_img, models, threshold=0.6):
+    swapped = whole_img.copy()
+
+    for source_face, specific_face in source_specifics:
+        specific_embed = specific_face["embedding"]
+        specific_embed /= np.linalg.norm(specific_embed)
+
+        for target_face in target_faces:
+            target_embed = target_face["embedding"]
+            target_embed /= np.linalg.norm(target_embed)
+            cosine_distance = 1 - np.dot(specific_embed, target_embed)
+            if cosine_distance > threshold:
+                continue
+            swapped = swap_face(swapped, target_face, source_face, models)
+
+    return swapped

utils.py

@@ -0,0 +1,112 @@
+import os
+import cv2
+import time
+import glob
+import shutil
+import platform
+import datetime
+import subprocess
+from threading import Thread
+from moviepy.editor import VideoFileClip, ImageSequenceClip
+from moviepy.video.io.ffmpeg_tools import ffmpeg_extract_subclip
+
+
+def trim_video(video_path, output_path, start_frame, stop_frame):
+    video_name, _ = os.path.splitext(os.path.basename(video_path))
+    trimmed_video_filename = video_name + "_trimmed" + ".mp4"
+    temp_path = os.path.join(output_path, "trim")
+    os.makedirs(temp_path, exist_ok=True)
+    trimmed_video_file_path = os.path.join(temp_path, trimmed_video_filename)
+
+    video = VideoFileClip(video_path)
+    fps = video.fps
+    start_time = start_frame / fps
+    duration = (stop_frame - start_frame) / fps
+
+    trimmed_video = video.subclip(start_time, start_time + duration)
+    trimmed_video.write_videofile(
+        trimmed_video_file_path, codec="libx264", audio_codec="aac"
+    )
+    trimmed_video.close()
+    video.close()
+
+    return trimmed_video_file_path
+
+
+def open_directory(path=None):
+    if path is None:
+        return
+    try:
+        os.startfile(path)
+    except:
+        subprocess.Popen(["xdg-open", path])
+
+
+class StreamerThread(object):
+    def __init__(self, src=0):
+        self.capture = cv2.VideoCapture(src)
+        self.capture.set(cv2.CAP_PROP_BUFFERSIZE, 2)
+        self.FPS = 1 / 30
+        self.FPS_MS = int(self.FPS * 1000)
+        self.thread = None
+        self.stopped = False
+        self.frame = None
+
+    def start(self):
+        self.thread = Thread(target=self.update, args=())
+        self.thread.daemon = True
+        self.thread.start()
+
+    def stop(self):
+        self.stopped = True
+        self.thread.join()
+        print("stopped")
+
+    def update(self):
+        while not self.stopped:
+            if self.capture.isOpened():
+                (self.status, self.frame) = self.capture.read()
+            time.sleep(self.FPS)
+
+
+class ProcessBar:
+    def __init__(self, bar_length, total, before="⬛", after="🟨"):
+        self.bar_length = bar_length
+        self.total = total
+        self.before = before
+        self.after = after
+        self.bar = [self.before] * bar_length
+        self.start_time = time.time()
+
+    def get(self, index):
+        total = self.total
+        elapsed_time = time.time() - self.start_time
+        average_time_per_iteration = elapsed_time / (index + 1)
+        remaining_iterations = total - (index + 1)
+        estimated_remaining_time = remaining_iterations * average_time_per_iteration
+
+        self.bar[int(index / total * self.bar_length)] = self.after
+        info_text = f"({index+1}/{total}) {''.join(self.bar)} "
+        info_text += f"(ETR: {int(estimated_remaining_time // 60)} min {int(estimated_remaining_time % 60)} sec)"
+        return info_text
+
+
+logo_image = cv2.imread("./assets/images/logo.png", cv2.IMREAD_UNCHANGED)
+
+
+def add_logo_to_image(img, logo=logo_image):
+    logo_size = int(img.shape[1] * 0.1)
+    logo = cv2.resize(logo, (logo_size, logo_size))
+    if logo.shape[2] == 4:
+        alpha = logo[:, :, 3]
+    else:
+        alpha = np.ones_like(logo[:, :, 0]) * 255
+    padding = int(logo_size * 0.1)
+    roi = img.shape[0] - logo_size - padding, img.shape[1] - logo_size - padding
+    for c in range(0, 3):
+        img[roi[0] : roi[0] + logo_size, roi[1] : roi[1] + logo_size, c] = (
+            alpha / 255.0
+        ) * logo[:, :, c] + (1 - alpha / 255.0) * img[
+            roi[0] : roi[0] + logo_size, roi[1] : roi[1] + logo_size, c
+        ]
+    return img