Add files via upload

CONTRIBUTING.md

@@ -0,0 +1 @@
+Please always push on the experimental to ensure we don't mess with the main branch. All the test will be done on the experimental and will be pushed to the main branch after few days of testing.

models/instructions.txt

@@ -0,0 +1 @@
+just put the models in this folder

modules/capturer.py

@@ -0,0 +1,20 @@
+from typing import Any
+import cv2
+
+
+def get_video_frame(video_path: str, frame_number: int = 0) -> Any:
+    capture = cv2.VideoCapture(video_path)
+    frame_total = capture.get(cv2.CAP_PROP_FRAME_COUNT)
+    capture.set(cv2.CAP_PROP_POS_FRAMES, min(frame_total, frame_number - 1))
+    has_frame, frame = capture.read()
+    capture.release()
+    if has_frame:
+        return frame
+    return None
+
+
+def get_video_frame_total(video_path: str) -> int:
+    capture = cv2.VideoCapture(video_path)
+    video_frame_total = int(capture.get(cv2.CAP_PROP_FRAME_COUNT))
+    capture.release()
+    return video_frame_total

modules/core.py

@@ -0,0 +1,247 @@
+import os
+import sys
+# single thread doubles cuda performance - needs to be set before torch import
+if any(arg.startswith('--execution-provider') for arg in sys.argv):
+    os.environ['OMP_NUM_THREADS'] = '1'
+# reduce tensorflow log level
+os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
+import warnings
+from typing import List
+import platform
+import signal
+import shutil
+import argparse
+import torch
+import onnxruntime
+import tensorflow
+
+import modules.globals
+import modules.metadata
+import modules.ui as ui
+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
+
+if 'ROCMExecutionProvider' in modules.globals.execution_providers:
+    del torch
+
+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('--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('--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.video_encoder = args.video_encoder
+    modules.globals.video_quality = args.video_quality
+    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
+    
+    modules.globals.nsfw = 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]
+
+
+def decode_execution_providers(execution_providers: List[str]) -> List[str]:
+    return [provider for provider, encoded_execution_provider in zip(onnxruntime.get_available_providers(), encode_execution_providers(onnxruntime.get_available_providers()))
+            if any(execution_provider in encoded_execution_provider for execution_provider in execution_providers)]
+
+
+def suggest_max_memory() -> int:
+    if platform.system().lower() == 'darwin':
+        return 4
+    return 16
+
+
+def suggest_execution_providers() -> List[str]:
+    return encode_execution_providers(onnxruntime.get_available_providers())
+
+
+def suggest_execution_threads() -> int:
+    if 'DmlExecutionProvider' in modules.globals.execution_providers:
+        return 1
+    if 'ROCMExecutionProvider' in modules.globals.execution_providers:
+        return 1
+    return 8
+
+
+def limit_resources() -> None:
+    # prevent tensorflow memory leak
+    gpus = tensorflow.config.experimental.list_physical_devices('GPU')
+    for gpu in gpus:
+        tensorflow.config.experimental.set_memory_growth(gpu, True)
+    # limit memory usage
+    if modules.globals.max_memory:
+        memory = modules.globals.max_memory * 1024 ** 3
+        if platform.system().lower() == 'darwin':
+            memory = modules.globals.max_memory * 1024 ** 6
+        if platform.system().lower() == 'windows':
+            import ctypes
+            kernel32 = ctypes.windll.kernel32
+            kernel32.SetProcessWorkingSetSize(-1, ctypes.c_size_t(memory), ctypes.c_size_t(memory))
+        else:
+            import resource
+            resource.setrlimit(resource.RLIMIT_DATA, (memory, memory))
+
+
+def release_resources() -> None:
+    if 'CUDAExecutionProvider' in modules.globals.execution_providers:
+        torch.cuda.empty_cache()
+
+
+def pre_check() -> 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'):
+        update_status('ffmpeg is not installed.')
+        return False
+    return True
+
+
+def update_status(message: str, scope: str = 'DLC.CORE') -> 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
+    # process image to image
+    if has_image_extension(modules.globals.target_path):
+        if modules.globals.nsfw == False:
+            from modules.predicter import predict_image
+            if predict_image(modules.globals.target_path):
+                destroy()
+        shutil.copy2(modules.globals.target_path, modules.globals.output_path)
+        for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
+            update_status('Progressing...', frame_processor.NAME)
+            frame_processor.process_image(modules.globals.source_path, modules.globals.output_path, modules.globals.output_path)
+            release_resources()
+        if is_image(modules.globals.target_path):
+            update_status('Processing to image succeed!')
+        else:
+            update_status('Processing to image failed!')
+        return
+    # process image to videos
+    if modules.globals.nsfw == False:
+        from modules.predicter import predict_video
+        if predict_video(modules.globals.target_path):
+            destroy()
+    update_status('Creating temp resources...')
+    create_temp(modules.globals.target_path)
+    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):
+        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() -> None:
+    if modules.globals.target_path:
+        clean_temp(modules.globals.target_path)
+    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()
+    if modules.globals.headless:
+        start()
+    else:
+        window = ui.init(start, destroy)
+        window.mainloop()

modules/face_analyser.py

@@ -0,0 +1,31 @@
+from typing import Any
+import insightface
+
+import modules.globals
+from modules.typing import Frame
+
+FACE_ANALYSER = None
+
+
+def get_face_analyser() -> Any:
+    global FACE_ANALYSER
+
+    if FACE_ANALYSER is None:
+        FACE_ANALYSER = insightface.app.FaceAnalysis(name='buffalo_l', providers=modules.globals.execution_providers)
+        FACE_ANALYSER.prepare(ctx_id=0, det_size=(640, 640))
+    return FACE_ANALYSER
+
+
+def get_one_face(frame: Frame) -> Any:
+    face = get_face_analyser().get(frame)
+    try:
+        return min(face, key=lambda x: x.bbox[0])
+    except ValueError:
+        return None
+
+
+def get_many_faces(frame: Frame) -> Any:
+    try:
+        return get_face_analyser().get(frame)
+    except IndexError:
+        return None

modules/globals.py

@@ -0,0 +1,30 @@
+import os
+from typing import List, Dict
+
+ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
+WORKFLOW_DIR = os.path.join(ROOT_DIR, 'workflow')
+
+file_types = [
+    ('Image', ('*.png','*.jpg','*.jpeg','*.gif','*.bmp')),
+    ('Video', ('*.mp4','*.mkv'))
+]
+
+source_path = None
+target_path = None
+output_path = None
+frame_processors: List[str] = []
+keep_fps = None
+keep_audio = None
+keep_frames = None
+many_faces = None
+video_encoder = None
+video_quality = None
+max_memory = None
+execution_providers: List[str] = []
+execution_threads = None
+headless = None
+log_level = 'error'
+fp_ui: Dict[str, bool] = {}
+nsfw = None
+camera_input_combobox = None
+webcam_preview_running = False

modules/metadata.py

@@ -0,0 +1,3 @@
+name = 'Deep Live Cam'
+version = '1.3.0'
+edition = 'Portable'

modules/predicter.py

@@ -0,0 +1,25 @@
+import numpy
+import opennsfw2
+from PIL import Image
+
+from modules.typing import Frame
+
+MAX_PROBABILITY = 0.85
+
+
+def predict_frame(target_frame: Frame) -> bool:
+    image = Image.fromarray(target_frame)
+    image = opennsfw2.preprocess_image(image, opennsfw2.Preprocessing.YAHOO)
+    model = opennsfw2.make_open_nsfw_model()
+    views = numpy.expand_dims(image, axis=0)
+    _, probability = model.predict(views)[0]
+    return probability > MAX_PROBABILITY
+
+
+def predict_image(target_path: str) -> bool:
+    return opennsfw2.predict_image(target_path) > MAX_PROBABILITY
+
+
+def predict_video(target_path: str) -> bool:
+    _, probabilities = opennsfw2.predict_video_frames(video_path=target_path, frame_interval=100)
+    return any(probability > MAX_PROBABILITY for probability in probabilities)

modules/processors/frame/core.py

@@ -0,0 +1,73 @@
+import sys
+import importlib
+from concurrent.futures import ThreadPoolExecutor
+from types import ModuleType
+from typing import Any, List, Callable
+from tqdm import tqdm
+
+import modules
+import modules.globals                   
+
+FRAME_PROCESSORS_MODULES: List[ModuleType] = []
+FRAME_PROCESSORS_INTERFACE = [
+    'pre_check',
+    'pre_start',
+    'process_frame',
+    'process_image',
+    'process_video'
+]
+
+
+def load_frame_processor_module(frame_processor: str) -> Any:
+    try:
+        frame_processor_module = importlib.import_module(f'modules.processors.frame.{frame_processor}')
+        for method_name in FRAME_PROCESSORS_INTERFACE:
+            if not hasattr(frame_processor_module, method_name):
+                sys.exit()
+    except ImportError:
+        print(f"Frame processor {frame_processor} not found")
+        sys.exit()
+    return frame_processor_module
+
+
+def get_frame_processors_modules(frame_processors: List[str]) -> List[ModuleType]:
+    global FRAME_PROCESSORS_MODULES
+
+    if not FRAME_PROCESSORS_MODULES:
+        for frame_processor in frame_processors:
+            frame_processor_module = load_frame_processor_module(frame_processor)
+            FRAME_PROCESSORS_MODULES.append(frame_processor_module)
+    set_frame_processors_modules_from_ui(frame_processors)
+    return FRAME_PROCESSORS_MODULES
+
+def set_frame_processors_modules_from_ui(frame_processors: List[str]) -> None:
+    global FRAME_PROCESSORS_MODULES
+    for frame_processor, state in modules.globals.fp_ui.items():
+        if state == True and frame_processor not in frame_processors:
+            frame_processor_module = load_frame_processor_module(frame_processor)
+            FRAME_PROCESSORS_MODULES.append(frame_processor_module)
+            modules.globals.frame_processors.append(frame_processor)
+        if state == False:
+            try:
+                frame_processor_module = load_frame_processor_module(frame_processor)
+                FRAME_PROCESSORS_MODULES.remove(frame_processor_module)
+                modules.globals.frame_processors.remove(frame_processor)
+            except:
+                pass
+
+def multi_process_frame(source_path: str, temp_frame_paths: List[str], process_frames: Callable[[str, List[str], Any], None], progress: Any = None) -> None:
+    with ThreadPoolExecutor(max_workers=modules.globals.execution_threads) as executor:
+        futures = []
+        for path in temp_frame_paths:
+            future = executor.submit(process_frames, source_path, [path], progress)
+            futures.append(future)
+        for future in futures:
+            future.result()
+
+
+def process_video(source_path: str, frame_paths: list[str], process_frames: Callable[[str, List[str], Any], None]) -> None:
+    progress_bar_format = '{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}{postfix}]'
+    total = len(frame_paths)
+    with tqdm(total=total, desc='Processing', unit='frame', dynamic_ncols=True, bar_format=progress_bar_format) as progress:
+        progress.set_postfix({'execution_providers': modules.globals.execution_providers, 'execution_threads': modules.globals.execution_threads, 'max_memory': modules.globals.max_memory})
+        multi_process_frame(source_path, frame_paths, process_frames, progress)

modules/processors/frame/face_enhancer.py

@@ -0,0 +1,79 @@
+from typing import Any, List
+import cv2
+import threading
+import gfpgan
+import os
+
+import modules.globals
+import modules.processors.frame.core
+from modules.core import update_status
+from modules.face_analyser import get_one_face
+from modules.typing import Frame, Face
+from modules.utilities import conditional_download, resolve_relative_path, is_image, is_video
+
+FACE_ENHANCER = None
+THREAD_SEMAPHORE = threading.Semaphore()
+THREAD_LOCK = threading.Lock()
+NAME = 'DLC.FACE-ENHANCER'
+
+
+def pre_check() -> bool:
+    download_directory_path = resolve_relative_path('..\models')
+    conditional_download(download_directory_path, ['https://github.com/TencentARC/GFPGAN/releases/download/v1.3.4/GFPGANv1.4.pth'])
+    return True
+
+
+def pre_start() -> bool:
+    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)
+        return False
+    return True
+
+
+def get_face_enhancer() -> Any:
+    global FACE_ENHANCER
+
+    with THREAD_LOCK:
+        if FACE_ENHANCER is None:
+            if os.name == 'nt':
+                model_path = resolve_relative_path('..\models\GFPGANv1.4.pth')
+                # todo: set models path https://github.com/TencentARC/GFPGAN/issues/399
+            else:
+                model_path = resolve_relative_path('../models/GFPGANv1.4.pth')
+            FACE_ENHANCER = gfpgan.GFPGANer(model_path=model_path, upscale=1) # type: ignore[attr-defined]
+    return FACE_ENHANCER
+
+
+def enhance_face(temp_frame: Frame) -> Frame:
+    with THREAD_SEMAPHORE:
+        _, _, temp_frame = get_face_enhancer().enhance(
+            temp_frame,
+            paste_back=True
+        )
+    return temp_frame
+
+
+def process_frame(source_face: Face, temp_frame: Frame) -> Frame:
+    target_face = get_one_face(temp_frame)
+    if target_face:
+        temp_frame = enhance_face(temp_frame)
+    return temp_frame
+
+
+def process_frames(source_path: str, temp_frame_paths: List[str], progress: Any = None) -> None:
+    for temp_frame_path in temp_frame_paths:
+        temp_frame = cv2.imread(temp_frame_path)
+        result = process_frame(None, temp_frame)
+        cv2.imwrite(temp_frame_path, result)
+        if progress:
+            progress.update(1)
+
+
+def process_image(source_path: str, target_path: str, output_path: str) -> None:
+    target_frame = cv2.imread(target_path)
+    result = process_frame(None, target_frame)
+    cv2.imwrite(output_path, result)
+
+
+def process_video(source_path: str, temp_frame_paths: List[str]) -> None:
+    modules.processors.frame.core.process_video(None, temp_frame_paths, process_frames)

modules/processors/frame/face_swapper.py

@@ -0,0 +1,86 @@
+from typing import Any, List
+import cv2
+import insightface
+import threading
+
+import modules.globals
+import modules.processors.frame.core
+from modules.core import update_status
+from modules.face_analyser import get_one_face, get_many_faces
+from modules.typing import Face, Frame
+from modules.utilities import conditional_download, resolve_relative_path, is_image, is_video
+
+FACE_SWAPPER = None
+THREAD_LOCK = threading.Lock()
+NAME = 'DLC.FACE-SWAPPER'
+
+
+def pre_check() -> bool:
+    download_directory_path = resolve_relative_path('../models')
+    conditional_download(download_directory_path, ['https://huggingface.co/hacksider/deep-live-cam/blob/main/inswapper_128_fp16.onnx'])
+    return True
+
+
+def pre_start() -> bool:
+    if not is_image(modules.globals.source_path):
+        update_status('Select an image for source path.', NAME)
+        return False
+    elif not get_one_face(cv2.imread(modules.globals.source_path)):
+        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)
+        return False
+    return True
+
+
+def get_face_swapper() -> Any:
+    global FACE_SWAPPER
+
+    with THREAD_LOCK:
+        if FACE_SWAPPER is None:
+            model_path = resolve_relative_path('../models/inswapper_128_fp16.onnx')
+            FACE_SWAPPER = insightface.model_zoo.get_model(model_path, providers=modules.globals.execution_providers)
+    return FACE_SWAPPER
+
+
+def swap_face(source_face: Face, target_face: Face, temp_frame: Frame) -> Frame:
+    return get_face_swapper().get(temp_frame, target_face, source_face, paste_back=True)
+
+
+def process_frame(source_face: Face, temp_frame: Frame) -> Frame:
+    if modules.globals.many_faces:
+        many_faces = get_many_faces(temp_frame)
+        if many_faces:
+            for target_face in many_faces:
+                temp_frame = swap_face(source_face, target_face, temp_frame)
+    else:
+        target_face = get_one_face(temp_frame)
+        if target_face:
+            temp_frame = swap_face(source_face, target_face, temp_frame)
+    return temp_frame
+
+
+def process_frames(source_path: str, temp_frame_paths: List[str], progress: Any = None) -> None:
+    source_face = get_one_face(cv2.imread(source_path))
+    for temp_frame_path in temp_frame_paths:
+        temp_frame = cv2.imread(temp_frame_path)
+        try:
+            result = process_frame(source_face, temp_frame)
+            cv2.imwrite(temp_frame_path, result)
+        except Exception as exception:
+            print(exception)
+            pass
+        if progress:
+            progress.update(1)
+
+
+def process_image(source_path: str, target_path: str, output_path: str) -> None:
+    source_face = get_one_face(cv2.imread(source_path))
+    target_frame = cv2.imread(target_path)
+    result = process_frame(source_face, target_frame)
+    cv2.imwrite(output_path, result)
+
+
+def process_video(source_path: str, temp_frame_paths: List[str]) -> None:
+    modules.processors.frame.core.process_video(source_path, temp_frame_paths, process_frames)

modules/typing.py

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

modules/ui.json

@@ -0,0 +1,158 @@
+{
+  "CTk": {
+    "fg_color": ["gray95", "gray10"]
+  },
+  "CTkToplevel": {
+    "fg_color": ["gray95", "gray10"]
+  },
+  "CTkFrame": {
+    "corner_radius": 0,
+    "border_width": 0,
+    "fg_color": ["gray90", "gray13"],
+    "top_fg_color": ["gray85", "gray16"],
+    "border_color": ["gray65", "gray28"]
+  },
+  "CTkButton": {
+    "corner_radius": 0,
+    "border_width": 0,
+    "fg_color": ["#2aa666", "#1f538d"],
+    "hover_color": ["#3cb666", "#14375e"],
+    "border_color": ["#3e4a40", "#949A9F"],
+    "text_color": ["#f3faf6", "#f3faf6"],
+    "text_color_disabled": ["gray74", "gray60"]
+  },
+  "CTkLabel": {
+    "corner_radius": 0,
+    "fg_color": "transparent",
+    "text_color": ["gray14", "gray84"]
+  },
+  "CTkEntry": {
+    "corner_radius": 0,
+    "border_width": 2,
+    "fg_color": ["#F9F9FA", "#343638"],
+    "border_color": ["#979DA2", "#565B5E"],
+    "text_color": ["gray14", "gray84"],
+    "placeholder_text_color": ["gray52", "gray62"]
+  },
+  "CTkCheckbox": {
+    "corner_radius": 0,
+    "border_width": 3,
+    "fg_color": ["#2aa666", "#1f538d"],
+    "border_color": ["#3e4a40", "#949A9F"],
+    "hover_color": ["#3cb666", "#14375e"],
+    "checkmark_color": ["#f3faf6", "gray90"],
+    "text_color": ["gray14", "gray84"],
+    "text_color_disabled": ["gray60", "gray45"]
+  },
+  "CTkSwitch": {
+    "corner_radius": 1000,
+    "border_width": 3,
+    "button_length": 0,
+    "fg_color": ["#939BA2", "#4A4D50"],
+    "progress_color": ["#2aa666", "#1f538d"],
+    "button_color": ["gray36", "#D5D9DE"],
+    "button_hover_color": ["gray20", "gray100"],
+    "text_color": ["gray14", "gray84"],
+    "text_color_disabled": ["gray60", "gray45"]
+  },
+  "CTkRadiobutton": {
+    "corner_radius": 1000,
+    "border_width_checked": 6,
+    "border_width_unchecked": 3,
+    "fg_color": ["#2aa666", "#1f538d"],
+    "border_color": ["#3e4a40", "#949A9F"],
+    "hover_color": ["#3cb666", "#14375e"],
+    "text_color": ["gray14", "gray84"],
+    "text_color_disabled": ["gray60", "gray45"]
+  },
+  "CTkProgressBar": {
+    "corner_radius": 1000,
+    "border_width": 0,
+    "fg_color": ["#939BA2", "#4A4D50"],
+    "progress_color": ["#2aa666", "#1f538d"],
+    "border_color": ["gray", "gray"]
+  },
+  "CTkSlider": {
+    "corner_radius": 1000,
+    "button_corner_radius": 1000,
+    "border_width": 6,
+    "button_length": 0,
+    "fg_color": ["#939BA2", "#4A4D50"],
+    "progress_color": ["gray40", "#AAB0B5"],
+    "button_color": ["#2aa666", "#1f538d"],
+    "button_hover_color": ["#3cb666", "#14375e"]
+  },
+  "CTkOptionMenu": {
+    "corner_radius": 0,
+    "fg_color": ["#2aa666", "#1f538d"],
+    "button_color": ["#3cb666", "#14375e"],
+    "button_hover_color": ["#234567", "#1e2c40"],
+    "text_color": ["#f3faf6", "#f3faf6"],
+    "text_color_disabled": ["gray74", "gray60"]
+  },
+  "CTkComboBox": {
+    "corner_radius": 0,
+    "border_width": 2,
+    "fg_color": ["#F9F9FA", "#343638"],
+    "border_color": ["#979DA2", "#565B5E"],
+    "button_color": ["#979DA2", "#565B5E"],
+    "button_hover_color": ["#6E7174", "#7A848D"],
+    "text_color": ["gray14", "gray84"],
+    "text_color_disabled": ["gray50", "gray45"]
+  },
+  "CTkScrollbar": {
+    "corner_radius": 1000,
+    "border_spacing": 4,
+    "fg_color": "transparent",
+    "button_color": ["gray55", "gray41"],
+    "button_hover_color": ["gray40", "gray53"]
+  },
+  "CTkSegmentedButton": {
+    "corner_radius": 0,
+    "border_width": 2,
+    "fg_color": ["#979DA2", "gray29"],
+    "selected_color": ["#2aa666", "#1f538d"],
+    "selected_hover_color": ["#3cb666", "#14375e"],
+    "unselected_color": ["#979DA2", "gray29"],
+    "unselected_hover_color": ["gray70", "gray41"],
+    "text_color": ["#f3faf6", "#f3faf6"],
+    "text_color_disabled": ["gray74", "gray60"]
+  },
+  "CTkTextbox": {
+    "corner_radius": 0,
+    "border_width": 0,
+    "fg_color": ["gray100", "gray20"],
+    "border_color": ["#979DA2", "#565B5E"],
+    "text_color": ["gray14", "gray84"],
+    "scrollbar_button_color": ["gray55", "gray41"],
+    "scrollbar_button_hover_color": ["gray40", "gray53"]
+  },
+  "CTkScrollableFrame": {
+    "label_fg_color": ["gray80", "gray21"]
+  },
+  "DropdownMenu": {
+    "fg_color": ["gray90", "gray20"],
+    "hover_color": ["gray75", "gray28"],
+    "text_color": ["gray14", "gray84"]
+  },
+  "CTkFont": {
+    "macOS": {
+      "family": "Avenir",
+      "size": 18,
+      "weight": "normal"
+    },
+    "Windows": {
+      "family": "Corbel",
+      "size": 18,
+      "weight": "normal"
+    },
+    "Linux": {
+      "family": "Montserrat",
+      "size": 18,
+      "weight": "normal"
+    }
+  },
+  "URL": {
+    "text_color": ["gray74", "gray60"]
+  }
+}

modules/ui.py

@@ -0,0 +1,296 @@
+import os
+import webbrowser
+import customtkinter as ctk
+from typing import Callable, Tuple
+import cv2
+from PIL import Image, ImageOps
+
+import modules.globals
+import modules.metadata
+from modules.face_analyser import get_one_face
+from modules.capturer import get_video_frame, get_video_frame_total
+from modules.processors.frame.core import get_frame_processors_modules
+from modules.utilities import is_image, is_video, resolve_relative_path
+
+ROOT = None
+ROOT_HEIGHT = 700
+ROOT_WIDTH = 600
+
+PREVIEW = None
+PREVIEW_MAX_HEIGHT = 700
+PREVIEW_MAX_WIDTH = 1200
+
+RECENT_DIRECTORY_SOURCE = None
+RECENT_DIRECTORY_TARGET = None
+RECENT_DIRECTORY_OUTPUT = None
+
+preview_label = None
+preview_slider = None
+source_label = None
+target_label = None
+status_label = None
+
+img_ft, vid_ft = modules.globals.file_types
+
+
+def init(start: Callable[[], None], destroy: Callable[[], None]) -> ctk.CTk:
+    global ROOT, PREVIEW
+
+    ROOT = create_root(start, destroy)
+    PREVIEW = create_preview(ROOT)
+
+    return ROOT
+
+
+def create_root(start: Callable[[], None], destroy: Callable[[], None]) -> ctk.CTk:
+    global source_label, target_label, status_label
+
+    ctk.deactivate_automatic_dpi_awareness()
+    ctk.set_appearance_mode('system')
+    ctk.set_default_color_theme(resolve_relative_path('ui.json'))
+
+    root = ctk.CTk()
+    root.minsize(ROOT_WIDTH, ROOT_HEIGHT)
+    root.title(f'{modules.metadata.name} {modules.metadata.version} {modules.metadata.edition}')
+    root.configure()
+    root.protocol('WM_DELETE_WINDOW', lambda: destroy())
+
+    source_label = ctk.CTkLabel(root, text=None)
+    source_label.place(relx=0.1, rely=0.1, relwidth=0.3, relheight=0.25)
+
+    target_label = ctk.CTkLabel(root, text=None)
+    target_label.place(relx=0.6, rely=0.1, relwidth=0.3, relheight=0.25)
+
+    source_button = ctk.CTkButton(root, text='Select a face', cursor='hand2', command=lambda: select_source_path())
+    source_button.place(relx=0.1, rely=0.4, relwidth=0.3, relheight=0.1)
+
+    target_button = ctk.CTkButton(root, text='Select a target', cursor='hand2', command=lambda: select_target_path())
+    target_button.place(relx=0.6, rely=0.4, relwidth=0.3, relheight=0.1)
+
+    keep_fps_value = ctk.BooleanVar(value=modules.globals.keep_fps)
+    keep_fps_checkbox = ctk.CTkSwitch(root, text='Keep fps', variable=keep_fps_value, cursor='hand2', command=lambda: setattr(modules.globals, 'keep_fps', not modules.globals.keep_fps))
+    keep_fps_checkbox.place(relx=0.1, rely=0.6)
+
+    keep_frames_value = ctk.BooleanVar(value=modules.globals.keep_frames)
+    keep_frames_switch = ctk.CTkSwitch(root, text='Keep frames', variable=keep_frames_value, cursor='hand2', command=lambda: setattr(modules.globals, 'keep_frames', keep_frames_value.get()))
+    keep_frames_switch.place(relx=0.1, rely=0.65)
+
+    # for FRAME PROCESSOR ENHANCER tumbler:
+    enhancer_value = ctk.BooleanVar(value=modules.globals.fp_ui['face_enhancer'])
+    enhancer_switch = ctk.CTkSwitch(root, text='Face Enhancer', variable=enhancer_value, cursor='hand2', command=lambda: update_tumbler('face_enhancer',enhancer_value.get()))
+    enhancer_switch.place(relx=0.1, rely=0.7)
+
+    keep_audio_value = ctk.BooleanVar(value=modules.globals.keep_audio)
+    keep_audio_switch = ctk.CTkSwitch(root, text='Keep audio', variable=keep_audio_value, cursor='hand2', command=lambda: setattr(modules.globals, 'keep_audio', keep_audio_value.get()))
+    keep_audio_switch.place(relx=0.6, rely=0.6)
+
+    many_faces_value = ctk.BooleanVar(value=modules.globals.many_faces)
+    many_faces_switch = ctk.CTkSwitch(root, text='Many faces', variable=many_faces_value, cursor='hand2', command=lambda: setattr(modules.globals, 'many_faces', many_faces_value.get()))
+    many_faces_switch.place(relx=0.6, rely=0.65)
+
+    nsfw_value = ctk.BooleanVar(value=modules.globals.nsfw)
+    nsfw_switch = ctk.CTkSwitch(root, text='NSFW', variable=nsfw_value, cursor='hand2', command=lambda: setattr(modules.globals, 'nsfw', nsfw_value.get()))
+    nsfw_switch.place(relx=0.6, rely=0.7)
+
+    start_button = ctk.CTkButton(root, text='Start', cursor='hand2', command=lambda: select_output_path(start))
+    start_button.place(relx=0.15, rely=0.80, relwidth=0.2, relheight=0.05)
+
+    stop_button = ctk.CTkButton(root, text='Destroy', cursor='hand2', command=lambda: destroy())
+    stop_button.place(relx=0.4, rely=0.80, relwidth=0.2, relheight=0.05)
+
+    preview_button = ctk.CTkButton(root, text='Preview', cursor='hand2', command=lambda: toggle_preview())
+    preview_button.place(relx=0.65, rely=0.80, relwidth=0.2, relheight=0.05)
+
+    live_button = ctk.CTkButton(root, text='Live', cursor='hand2', command=lambda: webcam_preview())
+    live_button.place(relx=0.40, rely=0.86, relwidth=0.2, relheight=0.05)
+
+    status_label = ctk.CTkLabel(root, text=None, justify='center')
+    status_label.place(relx=0.1, rely=0.9, relwidth=0.8)
+
+    donate_label = ctk.CTkLabel(root, text='Deep Live Cam', justify='center', cursor='hand2')
+    donate_label.place(relx=0.1, rely=0.95, relwidth=0.8)
+    donate_label.configure(text_color=ctk.ThemeManager.theme.get('URL').get('text_color'))
+    donate_label.bind('<Button>', lambda event: webbrowser.open('https://paypal.me/hacksider'))
+
+    return root
+
+
+def create_preview(parent: ctk.CTkToplevel) -> ctk.CTkToplevel:
+    global preview_label, preview_slider
+
+    preview = ctk.CTkToplevel(parent)
+    preview.withdraw()
+    preview.title('Preview')
+    preview.configure()
+    preview.protocol('WM_DELETE_WINDOW', lambda: toggle_preview())
+    preview.resizable(width=False, height=False)
+
+    preview_label = ctk.CTkLabel(preview, text=None)
+    preview_label.pack(fill='both', expand=True)
+
+    preview_slider = ctk.CTkSlider(preview, from_=0, to=0, command=lambda frame_value: update_preview(frame_value))
+
+    return preview
+
+
+def update_status(text: str) -> None:
+    status_label.configure(text=text)
+    ROOT.update()
+
+
+def update_tumbler(var: str, value: bool) -> None:
+    modules.globals.fp_ui[var] = value
+
+
+def select_source_path() -> None:
+    global RECENT_DIRECTORY_SOURCE, img_ft, vid_ft
+
+    PREVIEW.withdraw()
+    source_path = ctk.filedialog.askopenfilename(title='select an source image', initialdir=RECENT_DIRECTORY_SOURCE, filetypes=[img_ft])
+    if is_image(source_path):
+        modules.globals.source_path = source_path
+        RECENT_DIRECTORY_SOURCE = os.path.dirname(modules.globals.source_path)
+        image = render_image_preview(modules.globals.source_path, (200, 200))
+        source_label.configure(image=image)
+    else:
+        modules.globals.source_path = None
+        source_label.configure(image=None)
+
+
+def select_target_path() -> None:
+    global RECENT_DIRECTORY_TARGET, img_ft, vid_ft
+
+    PREVIEW.withdraw()
+    target_path = ctk.filedialog.askopenfilename(title='select an target image or video', initialdir=RECENT_DIRECTORY_TARGET, filetypes=[img_ft, vid_ft])
+    if is_image(target_path):
+        modules.globals.target_path = target_path
+        RECENT_DIRECTORY_TARGET = os.path.dirname(modules.globals.target_path)
+        image = render_image_preview(modules.globals.target_path, (200, 200))
+        target_label.configure(image=image)
+    elif is_video(target_path):
+        modules.globals.target_path = target_path
+        RECENT_DIRECTORY_TARGET = os.path.dirname(modules.globals.target_path)
+        video_frame = render_video_preview(target_path, (200, 200))
+        target_label.configure(image=video_frame)
+    else:
+        modules.globals.target_path = None
+        target_label.configure(image=None)
+
+
+def select_output_path(start: Callable[[], None]) -> None:
+    global RECENT_DIRECTORY_OUTPUT, img_ft, vid_ft
+
+    if is_image(modules.globals.target_path):
+        output_path = ctk.filedialog.asksaveasfilename(title='save image output file', filetypes=[img_ft], defaultextension='.png', initialfile='output.png', initialdir=RECENT_DIRECTORY_OUTPUT)
+    elif is_video(modules.globals.target_path):
+        output_path = ctk.filedialog.asksaveasfilename(title='save video output file', filetypes=[vid_ft], defaultextension='.mp4', initialfile='output.mp4', initialdir=RECENT_DIRECTORY_OUTPUT)
+    else:
+        output_path = None
+    if output_path:
+        modules.globals.output_path = output_path
+        RECENT_DIRECTORY_OUTPUT = os.path.dirname(modules.globals.output_path)
+        start()
+
+
+def render_image_preview(image_path: str, size: Tuple[int, int]) -> ctk.CTkImage:
+    image = Image.open(image_path)
+    if size:
+        image = ImageOps.fit(image, size, Image.LANCZOS)
+    return ctk.CTkImage(image, size=image.size)
+
+
+def render_video_preview(video_path: str, size: Tuple[int, int], frame_number: int = 0) -> ctk.CTkImage:
+    capture = cv2.VideoCapture(video_path)
+    if frame_number:
+        capture.set(cv2.CAP_PROP_POS_FRAMES, frame_number)
+    has_frame, frame = capture.read()
+    if has_frame:
+        image = Image.fromarray(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB))
+        if size:
+            image = ImageOps.fit(image, size, Image.LANCZOS)
+        return ctk.CTkImage(image, size=image.size)
+    capture.release()
+    cv2.destroyAllWindows()
+
+
+def toggle_preview() -> None:
+    if PREVIEW.state() == 'normal':
+        PREVIEW.withdraw()
+    elif modules.globals.source_path and modules.globals.target_path:
+        init_preview()
+        update_preview()
+        PREVIEW.deiconify()
+
+
+def init_preview() -> None:
+    if is_image(modules.globals.target_path):
+        preview_slider.pack_forget()
+    if is_video(modules.globals.target_path):
+        video_frame_total = get_video_frame_total(modules.globals.target_path)
+        preview_slider.configure(to=video_frame_total)
+        preview_slider.pack(fill='x')
+        preview_slider.set(0)
+
+
+def update_preview(frame_number: int = 0) -> None:
+    if modules.globals.source_path and modules.globals.target_path:
+        temp_frame = get_video_frame(modules.globals.target_path, frame_number)
+        if modules.globals.nsfw == False:
+            from modules.predicter import predict_frame
+            if predict_frame(temp_frame):
+                quit()
+        for frame_processor in get_frame_processors_modules(modules.globals.frame_processors):
+            temp_frame = frame_processor.process_frame(
+                get_one_face(cv2.imread(modules.globals.source_path)),
+                temp_frame
+            )
+        image = Image.fromarray(cv2.cvtColor(temp_frame, cv2.COLOR_BGR2RGB))
+        image = ImageOps.contain(image, (PREVIEW_MAX_WIDTH, PREVIEW_MAX_HEIGHT), Image.LANCZOS)
+        image = ctk.CTkImage(image, size=image.size)
+        preview_label.configure(image=image)
+
+def webcam_preview():
+    if modules.globals.source_path is None:
+        # No image selected
+        return
+    
+    global preview_label, PREVIEW
+
+    cap = cv2.VideoCapture(0)  # Use index for the webcam (adjust the index accordingly if necessary)    
+    cap.set(cv2.CAP_PROP_FRAME_WIDTH, 960)  # Set the width of the resolution
+    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 540)  # Set the height of the resolution
+    cap.set(cv2.CAP_PROP_FPS, 60)  # Set the frame rate of the webcam
+    PREVIEW_MAX_WIDTH = 960
+    PREVIEW_MAX_HEIGHT = 540
+
+    preview_label.configure(image=None)  # Reset the preview image before startup
+
+    PREVIEW.deiconify()  # Open preview window
+
+    frame_processors = get_frame_processors_modules(modules.globals.frame_processors)
+
+    source_image = None  # Initialize variable for the selected face image
+
+    while True:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        # Select and save face image only once
+        if source_image is None and modules.globals.source_path:
+            source_image = get_one_face(cv2.imread(modules.globals.source_path))
+
+        temp_frame = frame.copy()  #Create a copy of the frame
+
+        for frame_processor in frame_processors:
+            temp_frame = frame_processor.process_frame(source_image, temp_frame)
+
+        image = cv2.cvtColor(temp_frame, cv2.COLOR_BGR2RGB)  # Convert the image to RGB format to display it with Tkinter
+        image = Image.fromarray(image)
+        image = ImageOps.contain(image, (PREVIEW_MAX_WIDTH, PREVIEW_MAX_HEIGHT), Image.LANCZOS)
+        image = ctk.CTkImage(image, size=image.size)
+        preview_label.configure(image=image)
+        ROOT.update()
+
+    cap.release()
+    PREVIEW.withdraw()  # Close preview window when loop is finished

modules/utilities.py

@@ -0,0 +1,141 @@
+import glob
+import mimetypes
+import os
+import platform
+import shutil
+import ssl
+import subprocess
+import urllib
+from pathlib import Path
+from typing import List, Any
+from tqdm import tqdm
+
+import modules.globals
+
+TEMP_FILE = 'temp.mp4'
+TEMP_DIRECTORY = 'temp'
+
+# monkey patch ssl for mac
+if platform.system().lower() == 'darwin':
+    ssl._create_default_https_context = ssl._create_unverified_context
+
+
+def run_ffmpeg(args: List[str]) -> bool:
+    commands = ['ffmpeg', '-hide_banner', '-hwaccel', 'auto', '-loglevel', modules.globals.log_level]
+    commands.extend(args)
+    try:
+        subprocess.check_output(commands, stderr=subprocess.STDOUT)
+        return True
+    except Exception:
+        pass
+    return False
+
+
+def detect_fps(target_path: str) -> float:
+    command = ['ffprobe', '-v', 'error', '-select_streams', 'v:0', '-show_entries', 'stream=r_frame_rate', '-of', 'default=noprint_wrappers=1:nokey=1', target_path]
+    output = subprocess.check_output(command).decode().strip().split('/')
+    try:
+        numerator, denominator = map(int, output)
+        return numerator / denominator
+    except Exception:
+        pass
+    return 30.0
+
+
+def extract_frames(target_path: str) -> None:
+    temp_directory_path = get_temp_directory_path(target_path)
+    run_ffmpeg(['-i', target_path, '-pix_fmt', 'rgb24', os.path.join(temp_directory_path, '%04d.png')])
+
+
+def create_video(target_path: str, fps: float = 30.0) -> None:
+    temp_output_path = get_temp_output_path(target_path)
+    temp_directory_path = get_temp_directory_path(target_path)
+    run_ffmpeg(['-r', str(fps), '-i', os.path.join(temp_directory_path, '%04d.png'), '-c:v', modules.globals.video_encoder, '-crf', str(modules.globals.video_quality), '-pix_fmt', 'yuv420p', '-vf', 'colorspace=bt709:iall=bt601-6-625:fast=1', '-y', temp_output_path])
+
+
+def restore_audio(target_path: str, output_path: str) -> None:
+    temp_output_path = get_temp_output_path(target_path)
+    done = run_ffmpeg(['-i', temp_output_path, '-i', target_path, '-c:v', 'copy', '-map', '0:v:0', '-map', '1:a:0', '-y', output_path])
+    if not done:
+        move_temp(target_path, output_path)
+
+
+def get_temp_frame_paths(target_path: str) -> List[str]:
+    temp_directory_path = get_temp_directory_path(target_path)
+    return glob.glob((os.path.join(glob.escape(temp_directory_path), '*.png')))
+
+
+def get_temp_directory_path(target_path: str) -> str:
+    target_name, _ = os.path.splitext(os.path.basename(target_path))
+    target_directory_path = os.path.dirname(target_path)
+    return os.path.join(target_directory_path, TEMP_DIRECTORY, target_name)
+
+
+def get_temp_output_path(target_path: str) -> str:
+    temp_directory_path = get_temp_directory_path(target_path)
+    return os.path.join(temp_directory_path, TEMP_FILE)
+
+
+def normalize_output_path(source_path: str, target_path: str, output_path: str) -> Any:
+    if source_path and target_path:
+        source_name, _ = os.path.splitext(os.path.basename(source_path))
+        target_name, target_extension = os.path.splitext(os.path.basename(target_path))
+        if os.path.isdir(output_path):
+            return os.path.join(output_path, source_name + '-' + target_name + target_extension)
+    return output_path
+
+
+def create_temp(target_path: str) -> None:
+    temp_directory_path = get_temp_directory_path(target_path)
+    Path(temp_directory_path).mkdir(parents=True, exist_ok=True)
+
+
+def move_temp(target_path: str, output_path: str) -> None:
+    temp_output_path = get_temp_output_path(target_path)
+    if os.path.isfile(temp_output_path):
+        if os.path.isfile(output_path):
+            os.remove(output_path)
+        shutil.move(temp_output_path, output_path)
+
+
+def clean_temp(target_path: str) -> None:
+    temp_directory_path = get_temp_directory_path(target_path)
+    parent_directory_path = os.path.dirname(temp_directory_path)
+    if not modules.globals.keep_frames and os.path.isdir(temp_directory_path):
+        shutil.rmtree(temp_directory_path)
+    if os.path.exists(parent_directory_path) and not os.listdir(parent_directory_path):
+        os.rmdir(parent_directory_path)
+
+
+def has_image_extension(image_path: str) -> bool:
+    return image_path.lower().endswith(('png', 'jpg', 'jpeg'))
+
+
+def is_image(image_path: str) -> bool:
+    if image_path and os.path.isfile(image_path):
+        mimetype, _ = mimetypes.guess_type(image_path)
+        return bool(mimetype and mimetype.startswith('image/'))
+    return False
+
+
+def is_video(video_path: str) -> bool:
+    if video_path and os.path.isfile(video_path):
+        mimetype, _ = mimetypes.guess_type(video_path)
+        return bool(mimetype and mimetype.startswith('video/'))
+    return False
+
+
+def conditional_download(download_directory_path: str, urls: List[str]) -> None:
+    if not os.path.exists(download_directory_path):
+        os.makedirs(download_directory_path)
+    for url in urls:
+        download_file_path = os.path.join(download_directory_path, os.path.basename(url))
+        if not os.path.exists(download_file_path):
+            request = urllib.request.urlopen(url) # type: ignore[attr-defined]
+            total = int(request.headers.get('Content-Length', 0))
+            with tqdm(total=total, desc='Downloading', unit='B', unit_scale=True, unit_divisor=1024) as progress:
+                urllib.request.urlretrieve(url, download_file_path, reporthook=lambda count, block_size, total_size: progress.update(block_size)) # type: ignore[attr-defined]
+
+
+def resolve_relative_path(path: str) -> str:
+    return os.path.abspath(os.path.join(os.path.dirname(__file__), path))