Refactor application structure.

modules/core.py

@@ -17,7 +17,7 @@ import tensorflow
 
 import modules.globals
 import modules.metadata
-import modules.ui as ui
+from modules.ui import DeepFakeUI
 from modules.processors.frame.core import get_frame_processors_modules
 from modules.utilities import has_image_extension, is_image, is_video, detect_fps, create_video, extract_frames, get_temp_frame_paths, restore_audio, create_temp, move_temp, clean_temp, normalize_output_path
 
@@ -27,84 +27,6 @@ if 'ROCMExecutionProvider' in modules.globals.execution_providers:
 warnings.filterwarnings('ignore', category=FutureWarning, module='insightface')
 warnings.filterwarnings('ignore', category=UserWarning, module='torchvision')
 
-
-def parse_args() -> None:
-    signal.signal(signal.SIGINT, lambda signal_number, frame: destroy())
-    program = argparse.ArgumentParser()
-    program.add_argument('-s', '--source', help='select an source image', dest='source_path')
-    program.add_argument('-t', '--target', help='select an target image or video', dest='target_path')
-    program.add_argument('-o', '--output', help='select output file or directory', dest='output_path')
-    program.add_argument('--frame-processor', help='pipeline of frame processors', dest='frame_processor', default=['face_swapper'], choices=['face_swapper', 'face_enhancer'], nargs='+')
-    program.add_argument('--keep-fps', help='keep original fps', dest='keep_fps', action='store_true', default=False)
-    program.add_argument('--keep-audio', help='keep original audio', dest='keep_audio', action='store_true', default=True)
-    program.add_argument('--keep-frames', help='keep temporary frames', dest='keep_frames', action='store_true', default=False)
-    program.add_argument('--many-faces', help='process every face', dest='many_faces', action='store_true', default=False)
-    program.add_argument('--nsfw-filter', help='filter the NSFW image or video', dest='nsfw_filter', action='store_true', default=False)
-    program.add_argument('--map-faces', help='map source target faces', dest='map_faces', action='store_true', default=False)
-    program.add_argument('--video-encoder', help='adjust output video encoder', dest='video_encoder', default='libx264', choices=['libx264', 'libx265', 'libvpx-vp9'])
-    program.add_argument('--video-quality', help='adjust output video quality', dest='video_quality', type=int, default=18, choices=range(52), metavar='[0-51]')
-    program.add_argument('--live-mirror', help='The live camera display as you see it in the front-facing camera frame', dest='live_mirror', action='store_true', default=False)
-    program.add_argument('--live-resizable', help='The live camera frame is resizable', dest='live_resizable', action='store_true', default=False)
-    program.add_argument('--max-memory', help='maximum amount of RAM in GB', dest='max_memory', type=int, default=suggest_max_memory())
-    program.add_argument('--execution-provider', help='execution provider', dest='execution_provider', default=['cpu'], choices=suggest_execution_providers(), nargs='+')
-    program.add_argument('--execution-threads', help='number of execution threads', dest='execution_threads', type=int, default=suggest_execution_threads())
-    program.add_argument('-v', '--version', action='version', version=f'{modules.metadata.name} {modules.metadata.version}')
-
-    # register deprecated args
-    program.add_argument('-f', '--face', help=argparse.SUPPRESS, dest='source_path_deprecated')
-    program.add_argument('--cpu-cores', help=argparse.SUPPRESS, dest='cpu_cores_deprecated', type=int)
-    program.add_argument('--gpu-vendor', help=argparse.SUPPRESS, dest='gpu_vendor_deprecated')
-    program.add_argument('--gpu-threads', help=argparse.SUPPRESS, dest='gpu_threads_deprecated', type=int)
-
-    args = program.parse_args()
-
-    modules.globals.source_path = args.source_path
-    modules.globals.target_path = args.target_path
-    modules.globals.output_path = normalize_output_path(modules.globals.source_path, modules.globals.target_path, args.output_path)
-    modules.globals.frame_processors = args.frame_processor
-    modules.globals.headless = args.source_path or args.target_path or args.output_path
-    modules.globals.keep_fps = args.keep_fps
-    modules.globals.keep_audio = args.keep_audio
-    modules.globals.keep_frames = args.keep_frames
-    modules.globals.many_faces = args.many_faces
-    modules.globals.nsfw_filter = args.nsfw_filter
-    modules.globals.map_faces = args.map_faces
-    modules.globals.video_encoder = args.video_encoder
-    modules.globals.video_quality = args.video_quality
-    modules.globals.live_mirror = args.live_mirror
-    modules.globals.live_resizable = args.live_resizable
-    modules.globals.max_memory = args.max_memory
-    modules.globals.execution_providers = decode_execution_providers(args.execution_provider)
-    modules.globals.execution_threads = args.execution_threads
-
-    #for ENHANCER tumbler:
-    if 'face_enhancer' in args.frame_processor:
-        modules.globals.fp_ui['face_enhancer'] = True
-    else:
-        modules.globals.fp_ui['face_enhancer'] = False
-
-    # translate deprecated args
-    if args.source_path_deprecated:
-        print('\033[33mArgument -f and --face are deprecated. Use -s and --source instead.\033[0m')
-        modules.globals.source_path = args.source_path_deprecated
-        modules.globals.output_path = normalize_output_path(args.source_path_deprecated, modules.globals.target_path, args.output_path)
-    if args.cpu_cores_deprecated:
-        print('\033[33mArgument --cpu-cores is deprecated. Use --execution-threads instead.\033[0m')
-        modules.globals.execution_threads = args.cpu_cores_deprecated
-    if args.gpu_vendor_deprecated == 'apple':
-        print('\033[33mArgument --gpu-vendor apple is deprecated. Use --execution-provider coreml instead.\033[0m')
-        modules.globals.execution_providers = decode_execution_providers(['coreml'])
-    if args.gpu_vendor_deprecated == 'nvidia':
-        print('\033[33mArgument --gpu-vendor nvidia is deprecated. Use --execution-provider cuda instead.\033[0m')
-        modules.globals.execution_providers = decode_execution_providers(['cuda'])
-    if args.gpu_vendor_deprecated == 'amd':
-        print('\033[33mArgument --gpu-vendor amd is deprecated. Use --execution-provider cuda instead.\033[0m')
-        modules.globals.execution_providers = decode_execution_providers(['rocm'])
-    if args.gpu_threads_deprecated:
-        print('\033[33mArgument --gpu-threads is deprecated. Use --execution-threads instead.\033[0m')
-        modules.globals.execution_threads = args.gpu_threads_deprecated
-
-
 def encode_execution_providers(execution_providers: List[str]) -> List[str]:
     return [execution_provider.replace('ExecutionProvider', '').lower() for execution_provider in execution_providers]
 
@@ -155,101 +77,183 @@ def release_resources() -> None:
     if 'CUDAExecutionProvider' in modules.globals.execution_providers:
         torch.cuda.empty_cache()
 
+class DeepFakeApp:
+    def __init__(self):
+        self.ui = DeepFakeUI(
+            self.start,
+            self.destroy
+        )
 
-def pre_check() -> bool:
+    def update_status(self, message: str, scope: str = 'DLC.CORE') -> None:
-    if sys.version_info < (3, 9):
+        print(f'[{scope}] {message}')
-        update_status('Python version is not supported - please upgrade to 3.9 or higher.')
+        if not modules.globals.headless:
-        return False
+            self.ui.update_status(message)
-    if not shutil.which('ffmpeg'):
-        update_status('ffmpeg is not installed.')
-        return False
-    return True
     
+    def pre_check(self) -> bool:
+        if sys.version_info < (3, 9):
+            update_status('Python version is not supported - please upgrade to 3.9 or higher.')
+            return False
+        if not shutil.which('ffmpeg'):
+            self.update_status('ffmpeg is not installed.')
+            return False
+        return True
 
-def update_status(message: str, scope: str = 'DLC.CORE') -> None:
+    def start(self) -> None:
-    print(f'[{scope}] {message}')
-    if not modules.globals.headless:
-        ui.update_status(message)
-
-def start() -> None:
-    for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
-        if not frame_processor.pre_start():
-            return
-    update_status('Processing...')
-    # process image to image
-    if has_image_extension(modules.globals.target_path):
-        if modules.globals.nsfw_filter and ui.check_and_ignore_nsfw(modules.globals.target_path, destroy):
-            return
-        try:
-            shutil.copy2(modules.globals.target_path, modules.globals.output_path)
-        except Exception as e:
-            print("Error copying file:", str(e))
         for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
-            update_status('Progressing...', frame_processor.NAME)
+            if not frame_processor.pre_start():
-            frame_processor.process_image(modules.globals.source_path, modules.globals.output_path, modules.globals.output_path)
+                return
-            release_resources()
+        self.update_status('Processing...')
-        if is_image(modules.globals.target_path):
+        # process image to image
-            update_status('Processing to image succeed!')
+        if has_image_extension(modules.globals.target_path):
-        else:
+            if modules.globals.nsfw_filter and self.ui.check_and_ignore_nsfw(modules.globals.target_path, self.destroy):
-            update_status('Processing to image failed!')
+                return
-        return
+            try:
-    # process image to videos
+                shutil.copy2(modules.globals.target_path, modules.globals.output_path)
-    if modules.globals.nsfw_filter and ui.check_and_ignore_nsfw(modules.globals.target_path, destroy):
+            except Exception as e:
-        return
+                print("Error copying file:", str(e))
-
+            for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
-    if not modules.globals.map_faces:
+                self.update_status('Progressing...', frame_processor.NAME)
-        update_status('Creating temp resources...')
+                frame_processor.process_image(modules.globals.source_path, modules.globals.output_path, modules.globals.output_path)
-        create_temp(modules.globals.target_path)
+                release_resources()
-        update_status('Extracting frames...')
+            if is_image(modules.globals.target_path):
-        extract_frames(modules.globals.target_path)
+                self.update_status('Processing to image succeed!')
-
+            else:
-    temp_frame_paths = get_temp_frame_paths(modules.globals.target_path)
+                self.update_status('Processing to image failed!')
-    for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
-        update_status('Progressing...', frame_processor.NAME)
-        frame_processor.process_video(modules.globals.source_path, temp_frame_paths)
-        release_resources()
-    # handles fps
-    if modules.globals.keep_fps:
-        update_status('Detecting fps...')
-        fps = detect_fps(modules.globals.target_path)
-        update_status(f'Creating video with {fps} fps...')
-        create_video(modules.globals.target_path, fps)
-    else:
-        update_status('Creating video with 30.0 fps...')
-        create_video(modules.globals.target_path)
-    # handle audio
-    if modules.globals.keep_audio:
-        if modules.globals.keep_fps:
-            update_status('Restoring audio...')
-        else:
-            update_status('Restoring audio might cause issues as fps are not kept...')
-        restore_audio(modules.globals.target_path, modules.globals.output_path)
-    else:
-        move_temp(modules.globals.target_path, modules.globals.output_path)
-    # clean and validate
-    clean_temp(modules.globals.target_path)
-    if is_video(modules.globals.target_path):
-        update_status('Processing to video succeed!')
-    else:
-        update_status('Processing to video failed!')
-
-
-def destroy(to_quit=True) -> None:
-    if modules.globals.target_path:
-        clean_temp(modules.globals.target_path)
-    if to_quit: quit()
-
-
-def run() -> None:
-    parse_args()
-    if not pre_check():
-        return
-    for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
-        if not frame_processor.pre_check():
             return
-    limit_resources()
+        # process image to videos
-    if modules.globals.headless:
+        if modules.globals.nsfw_filter and self.ui.check_and_ignore_nsfw(modules.globals.target_path, self.destroy):
-        start()
+            return
-    else:
+
-        window = ui.init(start, destroy)
+        if not modules.globals.map_faces:
-        window.mainloop()
+            self.update_status('Creating temp resources...')
+            create_temp(modules.globals.target_path)
+            self.update_status('Extracting frames...')
+            extract_frames(modules.globals.target_path)
+
+        temp_frame_paths = get_temp_frame_paths(modules.globals.target_path)
+        for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
+            self.update_status('Progressing...', frame_processor.NAME)
+            frame_processor.process_video(modules.globals.source_path, temp_frame_paths)
+            release_resources()
+        # handles fps
+        if modules.globals.keep_fps:
+            self.update_status('Detecting fps...')
+            fps = detect_fps(modules.globals.target_path)
+            self.update_status(f'Creating video with {fps} fps...')
+            create_video(modules.globals.target_path, fps)
+        else:
+            self.update_status('Creating video with 30.0 fps...')
+            create_video(modules.globals.target_path)
+        # handle audio
+        if modules.globals.keep_audio:
+            if modules.globals.keep_fps:
+                self.update_status('Restoring audio...')
+            else:
+                self.update_status('Restoring audio might cause issues as fps are not kept...')
+            restore_audio(modules.globals.target_path, modules.globals.output_path)
+        else:
+            move_temp(modules.globals.target_path, modules.globals.output_path)
+        # clean and validate
+        clean_temp(modules.globals.target_path)
+        if is_video(modules.globals.target_path):
+            self.update_status('Processing to video succeed!')
+        else:
+            self.update_status('Processing to video failed!')
+
+
+    def destroy(self, to_quit=True) -> None:
+        if modules.globals.target_path:
+            clean_temp(modules.globals.target_path)
+        if to_quit: 
+            sys.exit(0)
+        
+    def parse_args(self) -> None:
+        signal.signal(signal.SIGINT, lambda _: self.destroy())
+        program = argparse.ArgumentParser()
+        program.add_argument('-s', '--source', help='select an source image', dest='source_path')
+        program.add_argument('-t', '--target', help='select an target image or video', dest='target_path')
+        program.add_argument('-o', '--output', help='select output file or directory', dest='output_path')
+        program.add_argument('--frame-processor', help='pipeline of frame processors', dest='frame_processor', default=['face_swapper'], choices=['face_swapper', 'face_enhancer'], nargs='+')
+        program.add_argument('--keep-fps', help='keep original fps', dest='keep_fps', action='store_true', default=False)
+        program.add_argument('--keep-audio', help='keep original audio', dest='keep_audio', action='store_true', default=True)
+        program.add_argument('--keep-frames', help='keep temporary frames', dest='keep_frames', action='store_true', default=False)
+        program.add_argument('--many-faces', help='process every face', dest='many_faces', action='store_true', default=False)
+        program.add_argument('--nsfw-filter', help='filter the NSFW image or video', dest='nsfw_filter', action='store_true', default=False)
+        program.add_argument('--map-faces', help='map source target faces', dest='map_faces', action='store_true', default=False)
+        program.add_argument('--video-encoder', help='adjust output video encoder', dest='video_encoder', default='libx264', choices=['libx264', 'libx265', 'libvpx-vp9'])
+        program.add_argument('--video-quality', help='adjust output video quality', dest='video_quality', type=int, default=18, choices=range(52), metavar='[0-51]')
+        program.add_argument('--live-mirror', help='The live camera display as you see it in the front-facing camera frame', dest='live_mirror', action='store_true', default=False)
+        program.add_argument('--live-resizable', help='The live camera frame is resizable', dest='live_resizable', action='store_true', default=False)
+        program.add_argument('--max-memory', help='maximum amount of RAM in GB', dest='max_memory', type=int, default=suggest_max_memory())
+        program.add_argument('--execution-provider', help='execution provider', dest='execution_provider', default=['cpu'], choices=suggest_execution_providers(), nargs='+')
+        program.add_argument('--execution-threads', help='number of execution threads', dest='execution_threads', type=int, default=suggest_execution_threads())
+        program.add_argument('-v', '--version', action='version', version=f'{modules.metadata.name} {modules.metadata.version}')
+
+        # register deprecated args
+        program.add_argument('-f', '--face', help=argparse.SUPPRESS, dest='source_path_deprecated')
+        program.add_argument('--cpu-cores', help=argparse.SUPPRESS, dest='cpu_cores_deprecated', type=int)
+        program.add_argument('--gpu-vendor', help=argparse.SUPPRESS, dest='gpu_vendor_deprecated')
+        program.add_argument('--gpu-threads', help=argparse.SUPPRESS, dest='gpu_threads_deprecated', type=int)
+
+        args = program.parse_args()
+
+        modules.globals.source_path = args.source_path
+        modules.globals.target_path = args.target_path
+        modules.globals.output_path = normalize_output_path(modules.globals.source_path, modules.globals.target_path, args.output_path)
+        modules.globals.frame_processors = args.frame_processor
+        modules.globals.headless = args.source_path or args.target_path or args.output_path
+        modules.globals.keep_fps = args.keep_fps
+        modules.globals.keep_audio = args.keep_audio
+        modules.globals.keep_frames = args.keep_frames
+        modules.globals.many_faces = args.many_faces
+        modules.globals.nsfw_filter = args.nsfw_filter
+        modules.globals.map_faces = args.map_faces
+        modules.globals.video_encoder = args.video_encoder
+        modules.globals.video_quality = args.video_quality
+        modules.globals.live_mirror = args.live_mirror
+        modules.globals.live_resizable = args.live_resizable
+        modules.globals.max_memory = args.max_memory
+        modules.globals.execution_providers = decode_execution_providers(args.execution_provider)
+        modules.globals.execution_threads = args.execution_threads
+
+        #for ENHANCER tumbler:
+        if 'face_enhancer' in args.frame_processor:
+            modules.globals.fp_ui['face_enhancer'] = True
+        else:
+            modules.globals.fp_ui['face_enhancer'] = False
+
+        # translate deprecated args
+        if args.source_path_deprecated:
+            print('\033[33mArgument -f and --face are deprecated. Use -s and --source instead.\033[0m')
+            modules.globals.source_path = args.source_path_deprecated
+            modules.globals.output_path = normalize_output_path(args.source_path_deprecated, modules.globals.target_path, args.output_path)
+        if args.cpu_cores_deprecated:
+            print('\033[33mArgument --cpu-cores is deprecated. Use --execution-threads instead.\033[0m')
+            modules.globals.execution_threads = args.cpu_cores_deprecated
+        if args.gpu_vendor_deprecated == 'apple':
+            print('\033[33mArgument --gpu-vendor apple is deprecated. Use --execution-provider coreml instead.\033[0m')
+            modules.globals.execution_providers = decode_execution_providers(['coreml'])
+        if args.gpu_vendor_deprecated == 'nvidia':
+            print('\033[33mArgument --gpu-vendor nvidia is deprecated. Use --execution-provider cuda instead.\033[0m')
+            modules.globals.execution_providers = decode_execution_providers(['cuda'])
+        if args.gpu_vendor_deprecated == 'amd':
+            print('\033[33mArgument --gpu-vendor amd is deprecated. Use --execution-provider cuda instead.\033[0m')
+            modules.globals.execution_providers = decode_execution_providers(['rocm'])
+        if args.gpu_threads_deprecated:
+            print('\033[33mArgument --gpu-threads is deprecated. Use --execution-threads instead.\033[0m')
+            modules.globals.execution_threads = args.gpu_threads_deprecated
+
+
+    def run(self) -> None:
+        self.parse_args()
+        if not self.pre_check():
+            return
+        for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
+            if not frame_processor.pre_check():
+                return
+        limit_resources()
+        if modules.globals.headless:
+            self.start()
+        else:
+            self.ui.root.mainloop()
+

modules/processors/frame/core.py

@@ -19,14 +19,10 @@ FRAME_PROCESSORS_INTERFACE = [
 
 
 def load_frame_processor_module(frame_processor: str) -> Any:
-    try:
+    frame_processor_module = importlib.import_module(f'modules.processors.frame.{frame_processor}')
-        frame_processor_module = importlib.import_module(f'modules.processors.frame.{frame_processor}')
+    for method_name in FRAME_PROCESSORS_INTERFACE:
-        for method_name in FRAME_PROCESSORS_INTERFACE:
+        if not hasattr(frame_processor_module, method_name):
-            if not hasattr(frame_processor_module, method_name):
+            sys.exit()
-                sys.exit()
-    except ImportError:
-        print(f"Frame processor {frame_processor} not found")
-        sys.exit()
     return frame_processor_module
 
 

modules/processors/frame/face_swapper.py

@@ -5,7 +5,7 @@ import threading
 import numpy as np
 import modules.globals
 import modules.processors.frame.core
-from modules.core import update_status
+from run import app
 from modules.face_analyser import get_one_face, get_many_faces, default_source_face
 from modules.typing import Face, Frame
 from modules.utilities import (
@@ -36,17 +36,17 @@ def pre_check() -> bool:
 
 def pre_start() -> bool:
     if not modules.globals.map_faces and not is_image(modules.globals.source_path):
-        update_status("Select an image for source path.", NAME)
+        app.update_status("Select an image for source path.", NAME)
         return False
     elif not modules.globals.map_faces and not get_one_face(
         cv2.imread(modules.globals.source_path)
     ):
-        update_status("No face in source path detected.", NAME)
+        app.update_status("No face in source path detected.", NAME)
         return False
     if not is_image(modules.globals.target_path) and not is_video(
         modules.globals.target_path
     ):
-        update_status("Select an image or video for target path.", NAME)
+        app.update_status("Select an image or video for target path.", NAME)
         return False
     return True
 
@@ -236,7 +236,7 @@ def process_image(source_path: str, target_path: str, output_path: str) -> None:
         cv2.imwrite(output_path, result)
     else:
         if modules.globals.many_faces:
-            update_status(
+            app.update_status(
                 "Many faces enabled. Using first source image. Progressing...", NAME
             )
         target_frame = cv2.imread(output_path)
@@ -246,7 +246,7 @@ def process_image(source_path: str, target_path: str, output_path: str) -> None:
 
 def process_video(source_path: str, temp_frame_paths: List[str]) -> None:
     if modules.globals.map_faces and modules.globals.many_faces:
-        update_status(
+        app.update_status(
             "Many faces enabled. Using first source image. Progressing...", NAME
         )
     modules.processors.frame.core.process_video(
@@ -256,7 +256,7 @@ def process_video(source_path: str, temp_frame_paths: List[str]) -> None:
 
 def create_lower_mouth_mask(
     face: Face, frame: Frame
-) -> (np.ndarray, np.ndarray, tuple, np.ndarray):
+) -> tuple[np.ndarray, np.ndarray, tuple, np.ndarray]:
     mask = np.zeros(frame.shape[:2], dtype=np.uint8)
     mouth_cutout = None
     landmarks = face.landmark_2d_106

modules/typing.py

@@ -1,7 +1,6 @@
-from typing import Any
+from typing import Any, TypeAlias
 
 from insightface.app.common import Face
 import numpy
 
-Face = Face
+Frame: TypeAlias = numpy.ndarray[Any, Any]
-Frame = numpy.ndarray[Any, Any]

run.py

@@ -1,6 +1,9 @@
 #!/usr/bin/env python3
 
-from modules import core
+from modules.core import DeepFakeApp
+
+
+app = DeepFakeApp()
 
 if __name__ == '__main__':
-    core.run()
+    app.run()