#!/usr/bin/env python3 import os import sys import shutil # 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' import warnings from typing import List import platform import signal import torch import onnxruntime import pathlib from time import time import roop.globals import roop.metadata import roop.utilities as util import roop.util_ffmpeg as ffmpeg import ui.main as main from settings import Settings from roop.face_util import extract_face_images from roop.ProcessEntry import ProcessEntry from roop.ProcessMgr import ProcessMgr from roop.ProcessOptions import ProcessOptions from roop.capturer import get_video_frame_total clip_text = None call_display_ui = None process_mgr = None if 'ROCMExecutionProvider' in roop.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()) roop.globals.headless = False # Always enable all processors when using GUI if len(sys.argv) > 1: print('No CLI args supported - use Settings Tab instead') roop.globals.frame_processors = ['face_swapper', 'face_enhancer'] 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 roop.globals.execution_providers: return 1 if 'ROCMExecutionProvider' in roop.globals.execution_providers: return 1 return 8 def limit_resources() -> None: # limit memory usage if roop.globals.max_memory: memory = roop.globals.max_memory * 1024 ** 3 if platform.system().lower() == 'darwin': memory = roop.globals.max_memory * 1024 ** 6 if platform.system().lower() == 'windows': import ctypes kernel32 = ctypes.windll.kernel32 # type: ignore[attr-defined] 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: import gc global process_mgr if process_mgr is not None: process_mgr.release_resources() process_mgr = None gc.collect() # if 'CUDAExecutionProvider' in roop.globals.execution_providers and torch.cuda.is_available(): # with torch.cuda.device('cuda'): # torch.cuda.empty_cache() # torch.cuda.ipc_collect() 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 download_directory_path = util.resolve_relative_path('../models') util.conditional_download(download_directory_path, ['https://huggingface.co/countfloyd/deepfake/resolve/main/inswapper_128.onnx']) util.conditional_download(download_directory_path, ['https://huggingface.co/countfloyd/deepfake/resolve/main/GFPGANv1.4.onnx']) util.conditional_download(download_directory_path, ['https://github.com/csxmli2016/DMDNet/releases/download/v1/DMDNet.pth']) util.conditional_download(download_directory_path, ['https://github.com/facefusion/facefusion-assets/releases/download/models/GPEN-BFR-512.onnx']) download_directory_path = util.resolve_relative_path('../models/CLIP') util.conditional_download(download_directory_path, ['https://huggingface.co/countfloyd/deepfake/resolve/main/rd64-uni-refined.pth']) download_directory_path = util.resolve_relative_path('../models/CodeFormer') util.conditional_download(download_directory_path, ['https://huggingface.co/countfloyd/deepfake/resolve/main/CodeFormerv0.1.onnx']) if not shutil.which('ffmpeg'): update_status('ffmpeg is not installed.') return True def set_display_ui(function): global call_display_ui call_display_ui = function def update_status(message: str) -> None: global call_display_ui print(message) if call_display_ui is not None: call_display_ui(message) def start() -> None: if roop.globals.headless: print('Headless mode currently unsupported - starting UI!') # faces = extract_face_images(roop.globals.source_path, (False, 0)) # roop.globals.INPUT_FACES.append(faces[roop.globals.source_face_index]) # faces = extract_face_images(roop.globals.target_path, (False, util.has_image_extension(roop.globals.target_path))) # roop.globals.TARGET_FACES.append(faces[roop.globals.target_face_index]) # if 'face_enhancer' in roop.globals.frame_processors: # roop.globals.selected_enhancer = 'GFPGAN' batch_process(None, False, None) def get_processing_plugins(use_clip): processors = "faceswap" if use_clip: processors += ",mask_clip2seg" if roop.globals.selected_enhancer == 'GFPGAN': processors += ",gfpgan" elif roop.globals.selected_enhancer == 'Codeformer': processors += ",codeformer" elif roop.globals.selected_enhancer == 'DMDNet': processors += ",dmdnet" elif roop.globals.selected_enhancer == 'GPEN': processors += ",gpen" return processors def live_swap(frame, swap_mode, use_clip, clip_text, selected_index = 0): global process_mgr if frame is None: return frame if process_mgr is None: process_mgr = ProcessMgr(None) options = ProcessOptions(get_processing_plugins(use_clip), roop.globals.distance_threshold, roop.globals.blend_ratio, swap_mode, selected_index, clip_text) process_mgr.initialize(roop.globals.INPUT_FACESETS, roop.globals.TARGET_FACES, options) newframe = process_mgr.process_frame(frame) if newframe is None: return frame return newframe def preview_mask(frame, clip_text): import numpy as np global process_mgr maskimage = np.zeros((frame.shape), np.uint8) if process_mgr is None: process_mgr = ProcessMgr(None) options = ProcessOptions("mask_clip2seg", roop.globals.distance_threshold, roop.globals.blend_ratio, "None", 0, clip_text) process_mgr.initialize(roop.globals.INPUT_FACESETS, roop.globals.TARGET_FACES, options) maskprocessor = next((x for x in process_mgr.processors if x.processorname == 'clip2seg'), None) return process_mgr.process_mask(maskprocessor, frame, maskimage) def batch_process(files:list[ProcessEntry], use_clip, new_clip_text, use_new_method, progress) -> None: global clip_text, process_mgr roop.globals.processing = True release_resources() limit_resources() # limit threads for some providers max_threads = suggest_execution_threads() if max_threads == 1: roop.globals.execution_threads = 1 imagefiles:list[ProcessEntry] = [] videofiles:list[ProcessEntry] = [] update_status('Sorting videos/images') for index, f in enumerate(files): fullname = f.filename if util.has_image_extension(fullname): destination = util.get_destfilename_from_path(fullname, roop.globals.output_path, f'.{roop.globals.CFG.output_image_format}') destination = util.replace_template(destination, index=index) pathlib.Path(os.path.dirname(destination)).mkdir(parents=True, exist_ok=True) f.finalname = destination imagefiles.append(f) elif util.is_video(fullname) or util.has_extension(fullname, ['gif']): destination = util.get_destfilename_from_path(fullname, roop.globals.output_path, f'__temp.{roop.globals.CFG.output_video_format}') f.finalname = destination videofiles.append(f) if process_mgr is None: process_mgr = ProcessMgr(progress) options = ProcessOptions(get_processing_plugins(use_clip), roop.globals.distance_threshold, roop.globals.blend_ratio, roop.globals.face_swap_mode, 0, new_clip_text) process_mgr.initialize(roop.globals.INPUT_FACESETS, roop.globals.TARGET_FACES, options) if(len(imagefiles) > 0): update_status('Processing image(s)') origimages = [] fakeimages = [] for f in imagefiles: origimages.append(f.filename) fakeimages.append(f.finalname) process_mgr.run_batch(origimages, fakeimages, roop.globals.execution_threads) origimages.clear() fakeimages.clear() if(len(videofiles) > 0): for index,v in enumerate(videofiles): if not roop.globals.processing: end_processing('Processing stopped!') return fps = v.fps if v.fps > 0 else util.detect_fps(v.filename) if v.endframe == 0: v.endframe = get_video_frame_total(v.filename) update_status(f'Creating {os.path.basename(v.finalname)} with {fps} FPS...') start_processing = time() if roop.globals.keep_frames or not use_new_method: util.create_temp(v.filename) update_status('Extracting frames...') ffmpeg.extract_frames(v.filename,v.startframe,v.endframe, fps) if not roop.globals.processing: end_processing('Processing stopped!') return temp_frame_paths = util.get_temp_frame_paths(v.filename) process_mgr.run_batch(temp_frame_paths, temp_frame_paths, roop.globals.execution_threads) if not roop.globals.processing: end_processing('Processing stopped!') return if roop.globals.wait_after_extraction: extract_path = os.path.dirname(temp_frame_paths[0]) util.open_folder(extract_path) input("Press any key to continue...") print("Resorting frames to create video") util.sort_rename_frames(extract_path) ffmpeg.create_video(v.filename, f.finalname, fps) if not roop.globals.keep_frames: util.delete_temp_frames(temp_frame_paths[0]) else: if util.has_extension(v.filename, ['gif']): skip_audio = True else: skip_audio = roop.globals.skip_audio process_mgr.run_batch_inmem(v.filename, v.finalname, v.startframe, v.endframe, fps,roop.globals.execution_threads, skip_audio) if not roop.globals.processing: end_processing('Processing stopped!') return video_file_name = v.finalname if os.path.isfile(video_file_name): destination = '' if util.has_extension(v.filename, ['gif']): gifname = util.get_destfilename_from_path(v.filename, roop.globals.output_path, '.gif') destination = util.replace_template(gifname, index=index) pathlib.Path(os.path.dirname(destination)).mkdir(parents=True, exist_ok=True) update_status('Creating final GIF') ffmpeg.create_gif_from_video(video_file_name, destination) if os.path.isfile(destination): os.remove(video_file_name) else: skip_audio = roop.globals.skip_audio destination = util.replace_template(video_file_name, index=index) pathlib.Path(os.path.dirname(destination)).mkdir(parents=True, exist_ok=True) if not skip_audio: ffmpeg.restore_audio(video_file_name, v.filename, v.startframe, v.endframe, destination) if os.path.isfile(destination): os.remove(video_file_name) else: shutil.move(video_file_name, destination) update_status(f'\nProcessing {os.path.basename(destination)} took {time() - start_processing} secs') else: update_status(f'Failed processing {os.path.basename(v.finalname)}!') end_processing('Finished') def end_processing(msg:str): update_status(msg) roop.globals.target_folder_path = None release_resources() def destroy() -> None: if roop.globals.target_path: util.clean_temp(roop.globals.target_path) release_resources() sys.exit() def run() -> None: parse_args() if not pre_check(): return roop.globals.CFG = Settings('config.yaml') roop.globals.execution_threads = roop.globals.CFG.max_threads roop.globals.video_encoder = roop.globals.CFG.output_video_codec roop.globals.video_quality = roop.globals.CFG.video_quality roop.globals.max_memory = roop.globals.CFG.memory_limit if roop.globals.CFG.memory_limit > 0 else None main.run()