Upload 18 files

roop/FaceSet.py

@@ -0,0 +1,20 @@
+import numpy as np
+
+class FaceSet:
+    faces = []
+    ref_images = []
+    embedding_average = 'None'
+    embeddings_backup = None
+
+    def __init__(self):
+        self.faces = []
+        self.ref_images = []
+        self.embeddings_backup = None
+
+    def AverageEmbeddings(self):
+        if len(self.faces) > 1 and self.embeddings_backup is None:
+            self.embeddings_backup = self.faces[0]['embedding']
+            embeddings = [face.embedding for face in self.faces]
+
+            self.faces[0]['embedding'] = np.mean(embeddings, axis=0)
+            # try median too?

roop/ProcessEntry.py

@@ -0,0 +1,7 @@
+class ProcessEntry:
+    def __init__(self, filename: str, start: int, end: int, fps: float):
+        self.filename = filename
+        self.finalname = None
+        self.startframe = start
+        self.endframe = end
+        self.fps = fps

roop/ProcessMgr.py

@@ -0,0 +1,663 @@
+import os
+import cv2 
+import numpy as np
+import psutil
+
+from roop.ProcessOptions import ProcessOptions
+
+from roop.face_util import get_first_face, get_all_faces, rotate_image_180, rotate_anticlockwise, rotate_clockwise, clamp_cut_values
+from roop.utilities import compute_cosine_distance, get_device, str_to_class
+import roop.vr_util as vr
+
+from typing import Any, List, Callable
+from roop.typing import Frame, Face
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from threading import Thread, Lock
+from queue import Queue
+from tqdm import tqdm
+from roop.ffmpeg_writer import FFMPEG_VideoWriter
+import roop.globals
+
+
+def create_queue(temp_frame_paths: List[str]) -> Queue[str]:
+    queue: Queue[str] = Queue()
+    for frame_path in temp_frame_paths:
+        queue.put(frame_path)
+    return queue
+
+
+def pick_queue(queue: Queue[str], queue_per_future: int) -> List[str]:
+    queues = []
+    for _ in range(queue_per_future):
+        if not queue.empty():
+            queues.append(queue.get())
+    return queues
+
+
+class ProcessMgr():
+    input_face_datas = []
+    target_face_datas = []
+
+    imagemask = None
+
+    processors = []
+    options : ProcessOptions = None
+    
+    num_threads = 1
+    current_index = 0
+    processing_threads = 1
+    buffer_wait_time = 0.1
+
+    lock = Lock()
+
+    frames_queue = None
+    processed_queue = None
+
+    videowriter= None
+
+    progress_gradio = None
+    total_frames = 0
+
+    
+
+
+    plugins =  { 
+    'faceswap'          : 'FaceSwapInsightFace',
+    'mask_clip2seg'     : 'Mask_Clip2Seg',
+    'mask_xseg'         : 'Mask_XSeg',
+    'codeformer'        : 'Enhance_CodeFormer',
+    'gfpgan'            : 'Enhance_GFPGAN',
+    'dmdnet'            : 'Enhance_DMDNet',
+    'gpen'              : 'Enhance_GPEN',
+    'restoreformer++'   : 'Enhance_RestoreFormerPPlus',
+    }
+
+    def __init__(self, progress):
+        if progress is not None:
+            self.progress_gradio = progress
+
+
+    def initialize(self, input_faces, target_faces, options):
+        self.input_face_datas = input_faces
+        self.target_face_datas = target_faces
+        self.options = options
+
+        roop.globals.g_desired_face_analysis=["landmark_3d_68", "landmark_2d_106","detection","recognition"]
+        if options.swap_mode == "all_female" or options.swap_mode == "all_male":
+            roop.globals.g_desired_face_analysis.append("genderage")
+
+        processornames = options.processors.split(",")
+        devicename = get_device()
+        if len(self.processors) < 1:
+            for pn in processornames:
+                classname = self.plugins[pn]
+                module = 'roop.processors.' + classname
+                p = str_to_class(module, classname)
+                if p is not None:
+                    p.Initialize(devicename)
+                    self.processors.append(p)
+                else:
+                    print(f"Not using {module}")
+        else:
+            for i in range(len(self.processors) -1, -1, -1):
+                if not self.processors[i].processorname in processornames:
+                    self.processors[i].Release()
+                    del self.processors[i]
+
+            for i,pn in enumerate(processornames):
+                if i >= len(self.processors) or self.processors[i].processorname != pn:
+                    classname = self.plugins[pn]
+                    module = 'roop.processors.' + classname
+                    p = str_to_class(module, classname)
+                    if p is not None:
+                        p.Initialize(devicename)
+                        self.processors.insert(i, p)
+                    else:
+                        print(f"Not using {module}")
+
+
+        if isinstance(self.options.imagemask, dict) and self.options.imagemask.get("layers") and len(self.options.imagemask["layers"]) > 0:
+            self.options.imagemask  = self.options.imagemask.get("layers")[0]
+            # Get rid of alpha
+            self.options.imagemask = cv2.cvtColor(self.options.imagemask, cv2.COLOR_RGBA2GRAY)
+            if np.any(self.options.imagemask):
+                mo = self.input_face_datas[0].faces[0].mask_offsets
+                self.options.imagemask = self.blur_area(self.options.imagemask, mo[4], mo[5])
+                self.options.imagemask = self.options.imagemask.astype(np.float32) / 255
+                self.options.imagemask = cv2.cvtColor(self.options.imagemask, cv2.COLOR_GRAY2RGB)
+            else:
+                self.options.imagemask = None
+ 
+
+
+
+    def run_batch(self, source_files, target_files, threads:int = 1):
+        progress_bar_format = '{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}{postfix}]'
+        self.total_frames = len(source_files)
+        self.num_threads = threads
+        with tqdm(total=self.total_frames, desc='Processing', unit='frame', dynamic_ncols=True, bar_format=progress_bar_format) as progress:
+            with ThreadPoolExecutor(max_workers=threads) as executor:
+                futures = []
+                queue = create_queue(source_files)
+                queue_per_future = max(len(source_files) // threads, 1)
+                while not queue.empty():
+                    future = executor.submit(self.process_frames, source_files, target_files, pick_queue(queue, queue_per_future), lambda: self.update_progress(progress))
+                    futures.append(future)
+                for future in as_completed(futures):
+                    future.result()
+
+
+    def process_frames(self, source_files: List[str], target_files: List[str], current_files, update: Callable[[], None]) -> None:
+        for f in current_files:
+            if not roop.globals.processing:
+                return
+            
+            # Decode the byte array into an OpenCV image
+            temp_frame = cv2.imdecode(np.fromfile(f, dtype=np.uint8), cv2.IMREAD_COLOR)
+            if temp_frame is not None:
+                resimg = self.process_frame(temp_frame)
+                if resimg is not None:
+                    i = source_files.index(f)
+                    cv2.imwrite(target_files[i], resimg)
+            if update:
+                update()
+
+
+
+    def read_frames_thread(self, cap, frame_start, frame_end, num_threads):
+        num_frame = 0
+        total_num = frame_end - frame_start
+        if frame_start > 0:
+            cap.set(cv2.CAP_PROP_POS_FRAMES,frame_start)
+
+        while True and roop.globals.processing:
+            ret, frame = cap.read()
+            if not ret:
+                break
+                
+            self.frames_queue[num_frame % num_threads].put(frame, block=True)
+            num_frame += 1
+            if num_frame == total_num:
+                break
+
+        for i in range(num_threads):
+            self.frames_queue[i].put(None)
+
+
+
+    def process_videoframes(self, threadindex, progress) -> None:
+        while True:
+            frame = self.frames_queue[threadindex].get()
+            if frame is None:
+                self.processing_threads -= 1
+                self.processed_queue[threadindex].put((False, None))
+                return
+            else:
+                resimg = self.process_frame(frame)
+                self.processed_queue[threadindex].put((True, resimg))
+                del frame
+                progress()
+
+
+    def write_frames_thread(self):
+        nextindex = 0
+        num_producers = self.num_threads
+        
+        while True:
+            process, frame = self.processed_queue[nextindex % self.num_threads].get()
+            nextindex += 1
+            if frame is not None:
+                self.videowriter.write_frame(frame)
+                del frame
+            elif process == False:
+                num_producers -= 1
+                if num_producers < 1:
+                    return
+            
+
+
+    def run_batch_inmem(self, source_video, target_video, frame_start, frame_end, fps, threads:int = 1, skip_audio=False):
+        cap = cv2.VideoCapture(source_video)
+        # frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
+        frame_count = (frame_end - frame_start) + 1
+        width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
+        height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
+
+        self.total_frames = frame_count
+        self.num_threads = threads
+
+        self.processing_threads = self.num_threads
+        self.frames_queue = []
+        self.processed_queue = []
+        for _ in range(threads):
+            self.frames_queue.append(Queue(1))
+            self.processed_queue.append(Queue(1))
+
+        self.videowriter =  FFMPEG_VideoWriter(target_video, (width, height), fps, codec=roop.globals.video_encoder, crf=roop.globals.video_quality, audiofile=None)
+
+        readthread = Thread(target=self.read_frames_thread, args=(cap, frame_start, frame_end, threads))
+        readthread.start()
+
+        writethread = Thread(target=self.write_frames_thread)
+        writethread.start()
+
+        progress_bar_format = '{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}{postfix}]'
+        with tqdm(total=self.total_frames, desc='Processing', unit='frames', dynamic_ncols=True, bar_format=progress_bar_format) as progress:
+            with ThreadPoolExecutor(thread_name_prefix='swap_proc', max_workers=self.num_threads) as executor:
+                futures = []
+                
+                for threadindex in range(threads):
+                    future = executor.submit(self.process_videoframes, threadindex, lambda: self.update_progress(progress))
+                    futures.append(future)
+                
+                for future in as_completed(futures):
+                    future.result()
+        # wait for the task to complete
+        readthread.join()
+        writethread.join()
+        cap.release()
+        self.videowriter.close()
+        self.frames_queue.clear()
+        self.processed_queue.clear()
+
+
+
+
+    def update_progress(self, progress: Any = None) -> None:
+        process = psutil.Process(os.getpid())
+        memory_usage = process.memory_info().rss / 1024 / 1024 / 1024
+        progress.set_postfix({
+            'memory_usage': '{:.2f}'.format(memory_usage).zfill(5) + 'GB',
+            'execution_threads': self.num_threads
+        })
+        progress.update(1)
+        self.progress_gradio((progress.n, self.total_frames), desc='Processing', total=self.total_frames, unit='frames')
+
+
+    def on_no_face_action(self, frame:Frame):
+        if roop.globals.no_face_action == 0:
+            return None, frame
+        elif roop.globals.no_face_action == 2:
+            return None, None
+
+        
+        faces = get_all_faces(frame)
+        if faces is not None:
+            return faces, frame
+        return None, frame
+      
+# https://github.com/deepinsight/insightface#third-party-re-implementation-of-arcface
+# https://github.com/deepinsight/insightface/blob/master/alignment/coordinate_reg/image_infer.py
+# https://github.com/deepinsight/insightface/issues/1350
+# https://github.com/linghu8812/tensorrt_inference
+
+
+    def process_frame(self, frame:Frame):
+        use_original_frame = 0
+        skip_frame = 2
+
+        if len(self.input_face_datas) < 1 and not self.options.show_face_masking:
+            return frame
+        temp_frame = frame.copy()
+        num_swapped, temp_frame = self.swap_faces(frame, temp_frame)
+        if num_swapped > 0:
+            return temp_frame
+        if roop.globals.no_face_action == use_original_frame:
+            return frame
+        if roop.globals.no_face_action == skip_frame:
+            #This only works with in-mem processing, as it simply skips the frame.
+            #For 'extract frames' it simply leaves the unprocessed frame unprocessed and it gets used in the final output by ffmpeg.
+            #If we could delete that frame here, that'd work but that might cause ffmpeg to fail unless the frames are renamed, and I don't think we have the info on what frame it actually is?????
+            #alternatively, it could mark all the necessary frames for deletion, delete them at the end, then rename the remaining frames that might work?
+            return None
+        else:
+            copyframe = frame.copy()
+            copyframe = rotate_image_180(copyframe)
+            temp_frame = copyframe.copy()
+            num_swapped, temp_frame = self.swap_faces(copyframe, temp_frame)
+            if num_swapped == 0:
+                return frame
+            temp_frame = rotate_image_180(temp_frame)
+            return temp_frame
+
+
+    def swap_faces(self, frame, temp_frame):
+        num_faces_found = 0
+
+        if self.options.swap_mode == "first":
+            face = get_first_face(frame)
+
+            if face is None:
+                return num_faces_found, frame
+            
+            num_faces_found += 1
+            temp_frame = self.process_face(self.options.selected_index, face, temp_frame)
+        else:
+            faces = get_all_faces(frame)
+            if faces is None:
+                return num_faces_found, frame
+            
+            if self.options.swap_mode == "all":
+                for face in faces:
+                    num_faces_found += 1
+                    temp_frame = self.process_face(self.options.selected_index, face, temp_frame)
+                    del face
+            
+            elif self.options.swap_mode == "selected":
+                num_targetfaces = len(self.target_face_datas) 
+                use_index = num_targetfaces == 1
+                for i,tf in enumerate(self.target_face_datas):
+                    for face in faces:
+                        if compute_cosine_distance(tf.embedding, face.embedding) <= self.options.face_distance_threshold:
+                            if i < len(self.input_face_datas):
+                                if use_index:
+                                    temp_frame = self.process_face(self.options.selected_index, face, temp_frame)
+                                else:
+                                    temp_frame = self.process_face(i, face, temp_frame)
+                                num_faces_found += 1
+                            del face
+                            if not roop.globals.vr_mode and num_faces_found == num_targetfaces:
+                                break
+            elif self.options.swap_mode == "all_female" or self.options.swap_mode == "all_male":
+                gender = 'F' if self.options.swap_mode == "all_female" else 'M'
+                for face in faces:
+                    if face.sex == gender:
+                        num_faces_found += 1
+                        temp_frame = self.process_face(self.options.selected_index, face, temp_frame)
+                    del face
+
+        if roop.globals.vr_mode and num_faces_found % 2 > 0:
+            # stereo image, there has to be an even number of faces
+            num_faces_found = 0
+            return num_faces_found, frame
+        if num_faces_found == 0:
+            return num_faces_found, frame
+
+        #maskprocessor = next((x for x in self.processors if x.type == 'mask'), None)
+
+        if self.options.imagemask is not None and self.options.imagemask.shape == frame.shape:
+            temp_frame = self.simple_blend_with_mask(temp_frame, frame, self.options.imagemask)
+
+        #if maskprocessor is not None:
+        #    temp_frame = self.process_mask(maskprocessor, frame, temp_frame)
+        return num_faces_found, temp_frame
+
+
+    def rotation_action(self, original_face:Face, frame:Frame):
+        (height, width) = frame.shape[:2]
+
+        bounding_box_width = original_face.bbox[2] - original_face.bbox[0]
+        bounding_box_height = original_face.bbox[3] - original_face.bbox[1]
+        horizontal_face = bounding_box_width > bounding_box_height
+
+        center_x = width // 2.0
+        start_x = original_face.bbox[0]
+        end_x = original_face.bbox[2]
+        bbox_center_x = start_x + (bounding_box_width // 2.0)
+
+        # need to leverage the array of landmarks as decribed here:
+        # https://github.com/deepinsight/insightface/tree/master/alignment/coordinate_reg
+        # basically, we should be able to check for the relative position of eyes and nose
+        # then use that to determine which way the face is actually facing when in a horizontal position
+        # and use that to determine the correct rotation_action
+
+        forehead_x = original_face.landmark_2d_106[72][0]
+        chin_x = original_face.landmark_2d_106[0][0]
+
+        if horizontal_face:
+            if chin_x < forehead_x:
+                # this is someone lying down with their face like this (:
+                return "rotate_anticlockwise"
+            elif forehead_x < chin_x:
+                # this is someone lying down with their face like this :)
+                return "rotate_clockwise"
+            if bbox_center_x >= center_x:
+                # this is someone lying down with their face in the right hand side of the frame
+                return "rotate_anticlockwise"
+            if bbox_center_x < center_x:
+                # this is someone lying down with their face in the left hand side of the frame
+                return "rotate_clockwise"
+
+        return None
+
+
+    def auto_rotate_frame(self, original_face, frame:Frame):
+        target_face = original_face
+        original_frame = frame
+
+        rotation_action = self.rotation_action(original_face, frame)
+
+        if rotation_action == "rotate_anticlockwise":
+            #face is horizontal, rotating frame anti-clockwise and getting face bounding box from rotated frame
+            frame = rotate_anticlockwise(frame)
+        elif rotation_action == "rotate_clockwise":
+            #face is horizontal, rotating frame clockwise and getting face bounding box from rotated frame
+            frame = rotate_clockwise(frame)
+
+        return target_face, frame, rotation_action
+    
+
+    def auto_unrotate_frame(self, frame:Frame, rotation_action):
+        if rotation_action == "rotate_anticlockwise":
+            return rotate_clockwise(frame)
+        elif rotation_action == "rotate_clockwise":
+            return rotate_anticlockwise(frame)
+        
+        return frame
+
+
+
+    def process_face(self,face_index, target_face:Face, frame:Frame):
+        from roop.face_util import align_crop
+
+        enhanced_frame = None
+        if(len(self.input_face_datas) > 0):
+            inputface = self.input_face_datas[face_index].faces[0]
+        else:
+            inputface = None
+
+        rotation_action = None
+        if roop.globals.autorotate_faces:
+            # check for sideways rotation of face
+            rotation_action = self.rotation_action(target_face, frame)
+            if rotation_action is not None:
+                (startX, startY, endX, endY) = target_face["bbox"].astype("int")
+                width = endX - startX
+                height = endY - startY
+                offs = int(max(width,height) * 0.25)
+                rotcutframe,startX, startY, endX, endY = self.cutout(frame, startX - offs, startY - offs, endX + offs, endY + offs)
+                if rotation_action == "rotate_anticlockwise":
+                    rotcutframe = rotate_anticlockwise(rotcutframe)
+                elif rotation_action == "rotate_clockwise":
+                    rotcutframe = rotate_clockwise(rotcutframe)
+                # rotate image and re-detect face to correct wonky landmarks
+                rotface = get_first_face(rotcutframe)
+                if rotface is None:
+                    rotation_action = None
+                else:
+                    saved_frame = frame.copy()
+                    frame = rotcutframe
+                    target_face = rotface
+
+
+
+        # if roop.globals.vr_mode:
+            # bbox = target_face.bbox
+            # [orig_width, orig_height, _] = frame.shape
+
+            # # Convert bounding box to ints
+            # x1, y1, x2, y2 = map(int, bbox)
+
+            # # Determine the center of the bounding box
+            # x_center = (x1 + x2) / 2
+            # y_center = (y1 + y2) / 2
+
+            # # Normalize coordinates to range [-1, 1]
+            # x_center_normalized = x_center / (orig_width / 2) - 1
+            # y_center_normalized = y_center / (orig_width / 2) - 1
+
+            # # Convert normalized coordinates to spherical (theta, phi)
+            # theta = x_center_normalized * 180  # Theta ranges from -180 to 180 degrees
+            # phi = -y_center_normalized * 90  # Phi ranges from -90 to 90 degrees
+
+            # img = vr.GetPerspective(frame, 90, theta, phi, 1280, 1280)  # Generate perspective image
+
+        fake_frame = None
+        aligned_img, M = align_crop(frame, target_face.kps, 128)
+        fake_frame = aligned_img
+        swap_frame = aligned_img
+        target_face.matrix = M
+        for p in self.processors:
+            if p.type == 'swap':
+                if inputface is not None:
+                    for _ in range(0,self.options.num_swap_steps):
+                        swap_frame = p.Run(inputface, target_face, swap_frame)
+                fake_frame = swap_frame
+                scale_factor = 0.0
+            elif p.type == 'mask':
+                fake_frame = self.process_mask(p, aligned_img, fake_frame)
+            else:
+                enhanced_frame, scale_factor = p.Run(self.input_face_datas[face_index], target_face, fake_frame)
+
+        upscale = 512
+        orig_width = fake_frame.shape[1]
+
+        fake_frame = cv2.resize(fake_frame, (upscale, upscale), cv2.INTER_CUBIC)
+        mask_offsets = (0,0,0,0,1,20) if inputface is None else inputface.mask_offsets
+
+        
+        if enhanced_frame is None:
+            scale_factor = int(upscale / orig_width)
+            result = self.paste_upscale(fake_frame, fake_frame, target_face.matrix, frame, scale_factor, mask_offsets)
+        else:
+            result = self.paste_upscale(fake_frame, enhanced_frame, target_face.matrix, frame, scale_factor, mask_offsets)
+
+        if rotation_action is not None:
+            fake_frame = self.auto_unrotate_frame(result, rotation_action)
+            return self.paste_simple(fake_frame, saved_frame, startX, startY)
+        
+        return result
+
+        
+
+
+    def cutout(self, frame:Frame, start_x, start_y, end_x, end_y):
+        if start_x < 0:
+            start_x = 0
+        if start_y < 0:
+            start_y = 0
+        if end_x > frame.shape[1]:
+            end_x = frame.shape[1]
+        if end_y > frame.shape[0]:
+            end_y = frame.shape[0]
+        return frame[start_y:end_y, start_x:end_x], start_x, start_y, end_x, end_y
+
+    def paste_simple(self, src:Frame, dest:Frame, start_x, start_y):
+        end_x = start_x + src.shape[1]
+        end_y = start_y + src.shape[0]
+
+        start_x, end_x, start_y, end_y = clamp_cut_values(start_x, end_x, start_y, end_y, dest)
+        dest[start_y:end_y, start_x:end_x] = src
+        return dest
+        
+    def simple_blend_with_mask(self, image1, image2, mask):
+        # Blend the images
+        blended_image = image1.astype(np.float32) * (1.0 - mask) + image2.astype(np.float32) * mask
+        return blended_image.astype(np.uint8)
+
+
+    def paste_upscale(self, fake_face, upsk_face, M, target_img, scale_factor, mask_offsets):
+        M_scale = M * scale_factor
+        IM = cv2.invertAffineTransform(M_scale)
+
+        face_matte = np.full((target_img.shape[0],target_img.shape[1]), 255, dtype=np.uint8)
+        # Generate white square sized as a upsk_face
+        img_matte = np.zeros((upsk_face.shape[0],upsk_face.shape[1]), dtype=np.uint8)
+
+        w = img_matte.shape[1]
+        h = img_matte.shape[0]
+
+        top = int(mask_offsets[0] * h)
+        bottom = int(h - (mask_offsets[1] * h))
+        left = int(mask_offsets[2] * w)
+        right = int(w - (mask_offsets[3] * w))
+        img_matte[top:bottom,left:right] = 255
+
+        # Transform white square back to target_img
+        img_matte = cv2.warpAffine(img_matte, IM, (target_img.shape[1], target_img.shape[0]), flags=cv2.INTER_NEAREST, borderValue=0.0) 
+        ##Blacken the edges of face_matte by 1 pixels (so the mask in not expanded on the image edges)
+        img_matte[:1,:] = img_matte[-1:,:] = img_matte[:,:1] = img_matte[:,-1:] = 0
+
+        img_matte = self.blur_area(img_matte, mask_offsets[4], mask_offsets[5])
+        #Normalize images to float values and reshape
+        img_matte = img_matte.astype(np.float32)/255
+        face_matte = face_matte.astype(np.float32)/255
+        img_matte = np.minimum(face_matte, img_matte)
+        if self.options.show_face_area_overlay:
+            # Additional steps for green overlay
+            green_overlay = np.zeros_like(target_img)
+            green_color = [0, 255, 0]  # RGB for green
+            for i in range(3):  # Apply green color where img_matte is not zero
+                green_overlay[:, :, i] = np.where(img_matte > 0, green_color[i], 0)        ##Transform upcaled face back to target_img
+        img_matte = np.reshape(img_matte, [img_matte.shape[0],img_matte.shape[1],1]) 
+        paste_face = cv2.warpAffine(upsk_face, IM, (target_img.shape[1], target_img.shape[0]), borderMode=cv2.BORDER_REPLICATE)
+        if upsk_face is not fake_face:
+            fake_face = cv2.warpAffine(fake_face, IM, (target_img.shape[1], target_img.shape[0]), borderMode=cv2.BORDER_REPLICATE)
+            paste_face = cv2.addWeighted(paste_face, self.options.blend_ratio, fake_face, 1.0 - self.options.blend_ratio, 0)
+
+        # Re-assemble image
+        paste_face = img_matte * paste_face
+        paste_face = paste_face + (1-img_matte) * target_img.astype(np.float32)
+        if self.options.show_face_area_overlay:
+            # Overlay the green overlay on the final image
+            paste_face = cv2.addWeighted(paste_face.astype(np.uint8), 1 - 0.5, green_overlay, 0.5, 0)
+        return paste_face.astype(np.uint8)
+
+
+    def blur_area(self, img_matte, num_erosion_iterations, blur_amount):
+        # Detect the affine transformed white area
+        mask_h_inds, mask_w_inds = np.where(img_matte==255) 
+        # Calculate the size (and diagonal size) of transformed white area width and height boundaries
+        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))
+        # Calculate the kernel size for eroding img_matte by kernel (insightface empirical guess for best size was max(mask_size//10,10))
+        # k = max(mask_size//12, 8)
+        k = max(mask_size//(blur_amount // 2) , blur_amount // 2)
+        kernel = np.ones((k,k),np.uint8)
+        img_matte = cv2.erode(img_matte,kernel,iterations = num_erosion_iterations)
+        #Calculate the kernel size for blurring img_matte by blur_size (insightface empirical guess for best size was max(mask_size//20, 5))
+        # k = max(mask_size//24, 4) 
+        k = max(mask_size//blur_amount, blur_amount//5) 
+        kernel_size = (k, k)
+        blur_size = tuple(2*i+1 for i in kernel_size)
+        return cv2.GaussianBlur(img_matte, blur_size, 0)
+
+
+    def process_mask(self, processor, frame:Frame, target:Frame):
+        img_mask = processor.Run(frame, self.options.masking_text)
+        img_mask = cv2.resize(img_mask, (target.shape[1], target.shape[0]))
+        img_mask = np.reshape(img_mask, [img_mask.shape[0],img_mask.shape[1],1])
+
+        if self.options.show_face_masking:
+            result = (1 - img_mask) * frame.astype(np.float32)
+            return np.uint8(result)
+
+
+        target = target.astype(np.float32)
+        result = (1-img_mask) * target
+        result += img_mask * frame.astype(np.float32)
+        return np.uint8(result)
+
+            
+
+
+    def unload_models():
+        pass
+
+
+    def release_resources(self):
+        for p in self.processors:
+            p.Release()
+        self.processors.clear()
+

roop/ProcessOptions.py

@@ -0,0 +1,13 @@
+class ProcessOptions:
+
+    def __init__(self,processors, face_distance,  blend_ratio, swap_mode, selected_index, masking_text, imagemask, num_steps, show_face_area, show_mask=False):
+        self.processors = processors
+        self.face_distance_threshold = face_distance
+        self.blend_ratio = blend_ratio
+        self.swap_mode = swap_mode
+        self.selected_index = selected_index
+        self.masking_text = masking_text
+        self.imagemask = imagemask
+        self.num_swap_steps = num_steps
+        self.show_face_area_overlay = show_face_area
+        self.show_face_masking = show_mask

roop/capturer.py

@@ -0,0 +1,30 @@
+from typing import Optional
+import cv2
+import numpy as np
+
+from roop.typing import Frame
+
+def get_image_frame(filename: str):
+    try:
+        return cv2.imdecode(np.fromfile(filename, dtype=np.uint8), cv2.IMREAD_COLOR)
+    except:
+        print(f"Exception reading {filename}")
+    return None
+
+    
+def get_video_frame(video_path: str, frame_number: int = 0) -> Optional[Frame]:
+    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

roop/core.py

@@ -0,0 +1,352 @@
+#!/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://huggingface.co/countfloyd/deepfake/resolve/main/GPEN-BFR-512.onnx'])
+    util.conditional_download(download_directory_path, ['https://huggingface.co/countfloyd/deepfake/resolve/main/restoreformer_plus_plus.onnx'])
+    util.conditional_download(download_directory_path, ['https://huggingface.co/countfloyd/deepfake/resolve/main/xseg.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(masking_engine):
+    processors = "faceswap"
+    if masking_engine is not None:
+        processors += f",{masking_engine}"
+    
+    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"
+    elif roop.globals.selected_enhancer == 'Restoreformer++':
+        processors += ",restoreformer++"
+    return processors
+
+
+def live_swap(frame, options):
+    global process_mgr
+
+    if frame is None:
+        return frame
+
+    if process_mgr is None:
+        process_mgr = ProcessMgr(None)
+    
+#    if len(roop.globals.INPUT_FACESETS) <= selected_index:
+#        selected_index = 0
+    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 batch_process(files:list[ProcessEntry], masking_engine:str, new_clip_text:str, use_new_method, imagemask, num_swap_steps, progress, selected_index = 0) -> 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)
+    mask = imagemask["layers"][0] if imagemask is not None else None
+    if len(roop.globals.INPUT_FACESETS) <= selected_index:
+        selected_index = 0
+    options = ProcessOptions(get_processing_plugins(masking_engine), roop.globals.distance_threshold, roop.globals.blend_ratio, roop.globals.face_swap_mode, selected_index, new_clip_text, mask, num_swap_steps, False)
+    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, v.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()

roop/face_util.py

@@ -0,0 +1,306 @@
+import threading
+from typing import Any
+import insightface
+
+import roop.globals
+from roop.typing import Frame, Face
+
+import cv2
+import numpy as np
+from skimage import transform as trans
+from roop.capturer import get_video_frame
+from roop.utilities import resolve_relative_path, conditional_download
+
+FACE_ANALYSER = None
+THREAD_LOCK_ANALYSER = threading.Lock()
+THREAD_LOCK_SWAPPER = threading.Lock()
+FACE_SWAPPER = None
+
+
+def get_face_analyser() -> Any:
+    global FACE_ANALYSER
+
+    with THREAD_LOCK_ANALYSER:
+        if FACE_ANALYSER is None or roop.globals.g_current_face_analysis != roop.globals.g_desired_face_analysis:
+            model_path = resolve_relative_path('..')
+            # removed genderage
+            allowed_modules = roop.globals.g_desired_face_analysis
+            roop.globals.g_current_face_analysis = roop.globals.g_desired_face_analysis
+            if roop.globals.CFG.force_cpu:
+                print("Forcing CPU for Face Analysis")
+                FACE_ANALYSER = insightface.app.FaceAnalysis(
+                    name="buffalo_l",
+                    root=model_path, providers=["CPUExecutionProvider"],allowed_modules=allowed_modules
+                )
+            else:
+                FACE_ANALYSER = insightface.app.FaceAnalysis(
+                    name="buffalo_l", root=model_path, providers=roop.globals.execution_providers,allowed_modules=allowed_modules
+                )
+            FACE_ANALYSER.prepare(
+                ctx_id=0,
+                det_size=(640, 640) if roop.globals.default_det_size else (320, 320),
+            )
+    return FACE_ANALYSER
+
+
+def get_first_face(frame: Frame) -> Any:
+    try:
+        faces = get_face_analyser().get(frame)
+        return min(faces, key=lambda x: x.bbox[0])
+    #   return sorted(faces, reverse=True, key=lambda x: (x.bbox[2] - x.bbox[0]) * (x.bbox[3] - x.bbox[1]))[0]
+    except:
+        return None
+
+
+def get_all_faces(frame: Frame) -> Any:
+    try:
+        faces = get_face_analyser().get(frame)
+        return sorted(faces, key=lambda x: x.bbox[0])
+    except:
+        return None
+
+
+def extract_face_images(source_filename, video_info, extra_padding=-1.0):
+    face_data = []
+    source_image = None
+
+    if video_info[0]:
+        frame = get_video_frame(source_filename, video_info[1])
+        if frame is not None:
+            source_image = frame
+        else:
+            return face_data
+    else:
+        source_image = cv2.imdecode(np.fromfile(source_filename, dtype=np.uint8), cv2.IMREAD_COLOR)
+
+    faces = get_all_faces(source_image)
+    if faces is None:
+        return face_data
+
+    i = 0
+    for face in faces:
+        (startX, startY, endX, endY) = face["bbox"].astype("int")
+        startX, endX, startY, endY = clamp_cut_values(startX, endX, startY, endY, source_image)
+        if extra_padding > 0.0:
+            if source_image.shape[:2] == (512, 512):
+                i += 1
+                face_data.append([face, source_image])
+                continue
+
+            found = False
+            for i in range(1, 3):
+                (startX, startY, endX, endY) = face["bbox"].astype("int")
+                startX, endX, startY, endY = clamp_cut_values(startX, endX, startY, endY, source_image)
+                cutout_padding = extra_padding
+                # top needs extra room for detection
+                padding = int((endY - startY) * cutout_padding)
+                oldY = startY
+                startY -= padding
+
+                factor = 0.25 if i == 1 else 0.5
+                cutout_padding = factor
+                padding = int((endY - oldY) * cutout_padding)
+                endY += padding
+                padding = int((endX - startX) * cutout_padding)
+                startX -= padding
+                endX += padding
+                startX, endX, startY, endY = clamp_cut_values(
+                    startX, endX, startY, endY, source_image
+                )
+                face_temp = source_image[startY:endY, startX:endX]
+                face_temp = resize_image_keep_content(face_temp)
+                testfaces = get_all_faces(face_temp)
+                if testfaces is not None and len(testfaces) > 0:
+                    i += 1
+                    face_data.append([testfaces[0], face_temp])
+                    found = True
+                    break
+
+            if not found:
+                print("No face found after resizing, this shouldn't happen!")
+            continue
+
+        face_temp = source_image[startY:endY, startX:endX]
+        if face_temp.size < 1:
+            continue
+
+        i += 1
+        face_data.append([face, face_temp])
+    return face_data
+
+
+def clamp_cut_values(startX, endX, startY, endY, image):
+    if startX < 0:
+        startX = 0
+    if endX > image.shape[1]:
+        endX = image.shape[1]
+    if startY < 0:
+        startY = 0
+    if endY > image.shape[0]:
+        endY = image.shape[0]
+    return startX, endX, startY, endY
+
+
+
+def face_offset_top(face: Face, offset):
+    face["bbox"][1] += offset
+    face["bbox"][3] += offset
+    lm106 = face.landmark_2d_106
+    add = np.full_like(lm106, [0, offset])
+    face["landmark_2d_106"] = lm106 + add
+    return face
+
+
+def resize_image_keep_content(image, new_width=512, new_height=512):
+    dim = None
+    (h, w) = image.shape[:2]
+    if h > w:
+        r = new_height / float(h)
+        dim = (int(w * r), new_height)
+    else:
+        # Calculate the ratio of the width and construct the dimensions
+        r = new_width / float(w)
+        dim = (new_width, int(h * r))
+    image = cv2.resize(image, dim, interpolation=cv2.INTER_AREA)
+    (h, w) = image.shape[:2]
+    if h == new_height and w == new_width:
+        return image
+    resize_img = np.zeros(shape=(new_height, new_width, 3), dtype=image.dtype)
+    offs = (new_width - w) if h == new_height else (new_height - h)
+    startoffs = int(offs // 2) if offs % 2 == 0 else int(offs // 2) + 1
+    offs = int(offs // 2)
+
+    if h == new_height:
+        resize_img[0:new_height, startoffs : new_width - offs] = image
+    else:
+        resize_img[startoffs : new_height - offs, 0:new_width] = image
+    return resize_img
+
+
+def rotate_image_90(image, rotate=True):
+    if rotate:
+        return np.rot90(image)
+    else:
+        return np.rot90(image, 1, (1, 0))
+
+
+def rotate_anticlockwise(frame):
+    return rotate_image_90(frame)
+
+
+def rotate_clockwise(frame):
+    return rotate_image_90(frame, False)
+
+
+def rotate_image_180(image):
+    return np.flip(image, 0)
+
+
+# alignment code from insightface https://github.com/deepinsight/insightface/blob/master/python-package/insightface/utils/face_align.py
+
+arcface_dst = np.array(
+    [
+        [38.2946, 51.6963],
+        [73.5318, 51.5014],
+        [56.0252, 71.7366],
+        [41.5493, 92.3655],
+        [70.7299, 92.2041],
+    ],
+    dtype=np.float32,
+)
+
+
+def estimate_norm(lmk, image_size=112, mode="arcface"):
+    assert lmk.shape == (5, 2)
+    assert image_size % 112 == 0 or image_size % 128 == 0
+    if image_size % 112 == 0:
+        ratio = float(image_size) / 112.0
+        diff_x = 0
+    else:
+        ratio = float(image_size) / 128.0
+        diff_x = 8.0 * ratio
+    dst = arcface_dst * ratio
+    dst[:, 0] += diff_x
+    tform = trans.SimilarityTransform()
+    tform.estimate(lmk, dst)
+    M = tform.params[0:2, :]
+    return M
+
+
+
+# aligned, M = norm_crop2(f[1], face.kps, 512)
+def align_crop(img, landmark, image_size=112, mode="arcface"):
+    M = estimate_norm(landmark, image_size, mode)
+    warped = cv2.warpAffine(img, M, (image_size, image_size), borderValue=0.0)
+    return warped, M
+
+
+def square_crop(im, S):
+    if im.shape[0] > im.shape[1]:
+        height = S
+        width = int(float(im.shape[1]) / im.shape[0] * S)
+        scale = float(S) / im.shape[0]
+    else:
+        width = S
+        height = int(float(im.shape[0]) / im.shape[1] * S)
+        scale = float(S) / im.shape[1]
+    resized_im = cv2.resize(im, (width, height))
+    det_im = np.zeros((S, S, 3), dtype=np.uint8)
+    det_im[: resized_im.shape[0], : resized_im.shape[1], :] = resized_im
+    return det_im, scale
+
+
+def transform(data, center, output_size, scale, rotation):
+    scale_ratio = scale
+    rot = float(rotation) * np.pi / 180.0
+    # translation = (output_size/2-center[0]*scale_ratio, output_size/2-center[1]*scale_ratio)
+    t1 = trans.SimilarityTransform(scale=scale_ratio)
+    cx = center[0] * scale_ratio
+    cy = center[1] * scale_ratio
+    t2 = trans.SimilarityTransform(translation=(-1 * cx, -1 * cy))
+    t3 = trans.SimilarityTransform(rotation=rot)
+    t4 = trans.SimilarityTransform(translation=(output_size / 2, output_size / 2))
+    t = t1 + t2 + t3 + t4
+    M = t.params[0:2]
+    cropped = cv2.warpAffine(data, M, (output_size, output_size), borderValue=0.0)
+    return cropped, M
+
+
+def trans_points2d(pts, M):
+    new_pts = np.zeros(shape=pts.shape, dtype=np.float32)
+    for i in range(pts.shape[0]):
+        pt = pts[i]
+        new_pt = np.array([pt[0], pt[1], 1.0], dtype=np.float32)
+        new_pt = np.dot(M, new_pt)
+        # print('new_pt', new_pt.shape, new_pt)
+        new_pts[i] = new_pt[0:2]
+
+    return new_pts
+
+
+def trans_points3d(pts, M):
+    scale = np.sqrt(M[0][0] * M[0][0] + M[0][1] * M[0][1])
+    # print(scale)
+    new_pts = np.zeros(shape=pts.shape, dtype=np.float32)
+    for i in range(pts.shape[0]):
+        pt = pts[i]
+        new_pt = np.array([pt[0], pt[1], 1.0], dtype=np.float32)
+        new_pt = np.dot(M, new_pt)
+        # print('new_pt', new_pt.shape, new_pt)
+        new_pts[i][0:2] = new_pt[0:2]
+        new_pts[i][2] = pts[i][2] * scale
+
+    return new_pts
+
+
+def trans_points(pts, M):
+    if pts.shape[1] == 2:
+        return trans_points2d(pts, M)
+    else:
+        return trans_points3d(pts, M)
+    
+def create_blank_image(width, height):
+    img = np.zeros((height, width, 4), dtype=np.uint8)
+    img[:] = [0,0,0,0]
+    return img
+

roop/ffmpeg_writer.py

@@ -0,0 +1,218 @@
+"""
+FFMPEG_Writer - write set of frames to video file
+
+original from
+https://github.com/Zulko/moviepy/blob/master/moviepy/video/io/ffmpeg_writer.py
+
+removed unnecessary dependencies
+
+The MIT License (MIT)
+
+Copyright (c) 2015 Zulko
+Copyright (c) 2023 Janvarev Vladislav
+"""
+
+import os
+import subprocess as sp
+
+PIPE = -1
+STDOUT = -2
+DEVNULL = -3
+
+FFMPEG_BINARY = "ffmpeg"
+
+class FFMPEG_VideoWriter:
+    """ A class for FFMPEG-based video writing.
+
+    A class to write videos using ffmpeg. ffmpeg will write in a large
+    choice of formats.
+
+    Parameters
+    -----------
+
+    filename
+      Any filename like 'video.mp4' etc. but if you want to avoid
+      complications it is recommended to use the generic extension
+      '.avi' for all your videos.
+
+    size
+      Size (width,height) of the output video in pixels.
+
+    fps
+      Frames per second in the output video file.
+
+    codec
+      FFMPEG codec. It seems that in terms of quality the hierarchy is
+      'rawvideo' = 'png' > 'mpeg4' > 'libx264'
+      'png' manages the same lossless quality as 'rawvideo' but yields
+      smaller files. Type ``ffmpeg -codecs`` in a terminal to get a list
+      of accepted codecs.
+
+      Note for default 'libx264': by default the pixel format yuv420p
+      is used. If the video dimensions are not both even (e.g. 720x405)
+      another pixel format is used, and this can cause problem in some
+      video readers.
+
+    audiofile
+      Optional: The name of an audio file that will be incorporated
+      to the video.
+
+    preset
+      Sets the time that FFMPEG will take to compress the video. The slower,
+      the better the compression rate. Possibilities are: ultrafast,superfast,
+      veryfast, faster, fast, medium (default), slow, slower, veryslow,
+      placebo.
+
+    bitrate
+      Only relevant for codecs which accept a bitrate. "5000k" offers
+      nice results in general.
+
+    """
+
+    def __init__(self, filename, size, fps, codec="libx265", crf=14, audiofile=None,
+                 preset="medium", bitrate=None,
+                 logfile=None, threads=None, ffmpeg_params=None):
+
+        if logfile is None:
+            logfile = sp.PIPE
+
+        self.filename = filename
+        self.codec = codec
+        self.ext = self.filename.split(".")[-1]
+        w = size[0] - 1 if size[0] % 2 != 0 else size[0]
+        h = size[1] - 1 if size[1] % 2 != 0 else size[1]
+
+
+        # order is important
+        cmd = [
+            FFMPEG_BINARY,
+            '-hide_banner',
+            '-hwaccel', 'auto',
+            '-y',
+            '-loglevel', 'error' if logfile == sp.PIPE else 'info',
+            '-f', 'rawvideo',
+            '-vcodec', 'rawvideo',
+            '-s', '%dx%d' % (size[0], size[1]),
+            #'-pix_fmt', 'rgba' if withmask else 'rgb24',
+            '-pix_fmt', 'bgr24',
+            '-r', str(fps),
+            '-an', '-i', '-' 
+        ]
+
+        if audiofile is not None:
+            cmd.extend([
+                '-i', audiofile,
+                '-acodec', 'copy'
+            ])
+
+        cmd.extend([
+            '-vcodec', codec,
+            '-crf', str(crf)
+            #'-preset', preset,
+        ])
+        if ffmpeg_params is not None:
+            cmd.extend(ffmpeg_params)
+        if bitrate is not None:
+            cmd.extend([
+                '-b', bitrate
+            ])
+
+        # scale to a resolution divisible by 2 if not even
+        cmd.extend(['-vf', f'scale={w}:{h}' if w != size[0] or h != size[1] else 'colorspace=bt709:iall=bt601-6-625:fast=1'])
+
+        if threads is not None:
+            cmd.extend(["-threads", str(threads)])
+
+        cmd.extend([
+            '-pix_fmt', 'yuv420p',
+
+        ])
+        cmd.extend([
+            filename
+        ])
+
+        test = str(cmd)
+        print(test)
+
+        popen_params = {"stdout": DEVNULL,
+                        "stderr": logfile,
+                        "stdin": sp.PIPE}
+
+        # This was added so that no extra unwanted window opens on windows
+        # when the child process is created
+        if os.name == "nt":
+            popen_params["creationflags"] = 0x08000000  # CREATE_NO_WINDOW
+        
+        self.proc = sp.Popen(cmd, **popen_params)
+
+
+    def write_frame(self, img_array):
+        """ Writes one frame in the file."""
+        try:
+            #if PY3:
+            self.proc.stdin.write(img_array.tobytes())
+            # else:
+            #    self.proc.stdin.write(img_array.tostring())
+        except IOError as err:
+            _, ffmpeg_error = self.proc.communicate()
+            error = (str(err) + ("\n\nroop unleashed error: FFMPEG encountered "
+                                 "the following error while writing file %s:"
+                                 "\n\n %s" % (self.filename, str(ffmpeg_error))))
+
+            if b"Unknown encoder" in ffmpeg_error:
+
+                error = error+("\n\nThe video export "
+                  "failed because FFMPEG didn't find the specified "
+                  "codec for video encoding (%s). Please install "
+                  "this codec or change the codec when calling "
+                  "write_videofile. For instance:\n"
+                  "  >>> clip.write_videofile('myvid.webm', codec='libvpx')")%(self.codec)
+
+            elif b"incorrect codec parameters ?" in ffmpeg_error:
+
+                 error = error+("\n\nThe video export "
+                  "failed, possibly because the codec specified for "
+                  "the video (%s) is not compatible with the given "
+                  "extension (%s). Please specify a valid 'codec' "
+                  "argument in write_videofile. This would be 'libx264' "
+                  "or 'mpeg4' for mp4, 'libtheora' for ogv, 'libvpx for webm. "
+                  "Another possible reason is that the audio codec was not "
+                  "compatible with the video codec. For instance the video "
+                  "extensions 'ogv' and 'webm' only allow 'libvorbis' (default) as a"
+                  "video codec."
+                  )%(self.codec, self.ext)
+
+            elif  b"encoder setup failed" in ffmpeg_error:
+
+                error = error+("\n\nThe video export "
+                  "failed, possibly because the bitrate you specified "
+                  "was too high or too low for the video codec.")
+
+            elif b"Invalid encoder type" in ffmpeg_error:
+
+                error = error + ("\n\nThe video export failed because the codec "
+                  "or file extension you provided is not a video")
+
+
+            raise IOError(error)
+
+    def close(self):
+        if self.proc:
+            self.proc.stdin.close()
+            if self.proc.stderr is not None:
+                self.proc.stderr.close()
+            self.proc.wait()
+
+        self.proc = None
+
+    # Support the Context Manager protocol, to ensure that resources are cleaned up.
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.close()
+
+
+
+    

roop/filters.py

@@ -0,0 +1,59 @@
+import numpy as np
+import cv2
+
+c64_palette = np.array([
+    [0, 0, 0],
+    [255, 255, 255],
+    [0x81, 0x33, 0x38],
+    [0x75, 0xce, 0xc8],
+    [0x8e, 0x3c, 0x97],
+    [0x56, 0xac, 0x4d],
+    [0x2e, 0x2c, 0x9b],
+    [0xed, 0xf1, 0x71],
+    [0x8e, 0x50, 0x29],
+    [0x55, 0x38, 0x00],
+    [0xc4, 0x6c, 0x71],
+    [0x4a, 0x4a, 0x4a],
+    [0x7b, 0x7b, 0x7b],
+    [0xa9, 0xff, 0x9f],
+    [0x70, 0x6d, 0xeb],
+    [0xb2, 0xb2, 0xb2]
+])
+
+def fast_quantize_to_palette(image):
+    # Simply round the color values to the nearest color in the palette
+    palette = c64_palette / 255.0  # Normalize palette
+    img_normalized = image / 255.0  # Normalize image
+
+    # Calculate the index in the palette that is closest to each pixel in the image
+    indices = np.sqrt(((img_normalized[:, :, None, :] - palette[None, None, :, :]) ** 2).sum(axis=3)).argmin(axis=2)
+    # Map the image to the palette colors
+    mapped_image = palette[indices]
+
+    return (mapped_image * 255).astype(np.uint8)  # Denormalize and return the image
+
+
+'''
+knn = None
+
+def quantize_to_palette(image, palette):
+    global knn
+
+    NumColors = 16
+    quantized_image = None
+    cv2.pyrMeanShiftFiltering(image, NumColors / 4, NumColors / 2, quantized_image, 1, cv2.TERM_CRITERIA_EPS | cv2.TERM_CRITERIA_MAX_ITER, 5, 1)
+
+    palette = c64_palette
+    X_query = image.reshape(-1, 3).astype(np.float32)
+
+    if(knn == None):
+        X_index = palette.astype(np.float32)
+        knn = cv2.ml.KNearest_create()
+        knn.train(X_index, cv2.ml.ROW_SAMPLE, np.arange(len(palette)))
+    
+    ret, results, neighbours, dist = knn.findNearest(X_query, 1)
+
+    quantized_image = np.array([palette[idx] for idx in neighbours.astype(int)])
+    quantized_image = quantized_image.reshape(image.shape)
+    return quantized_image.astype(np.uint8)
+'''

roop/globals.py

@@ -0,0 +1,53 @@
+from settings import Settings
+from typing import List
+
+source_path = None
+target_path = None
+output_path = None
+target_folder_path = None
+
+frame_processors: List[str] = []
+keep_fps = None
+keep_frames = None
+autorotate_faces = None
+vr_mode = None
+skip_audio = None
+wait_after_extraction = None
+many_faces = None
+use_batch = None
+source_face_index = 0
+target_face_index = 0
+face_position = None
+video_encoder = None
+video_quality = None
+max_memory = None
+execution_providers: List[str] = []
+execution_threads = None
+headless = None
+log_level = 'error'
+selected_enhancer = None
+face_swap_mode = None
+blend_ratio = 0.5
+distance_threshold = 0.65
+default_det_size = True
+
+no_face_action = 0
+
+processing = False
+
+g_current_face_analysis = None
+g_desired_face_analysis = None
+
+FACE_ENHANCER = None
+
+INPUT_FACESETS = []
+TARGET_FACES = []
+
+
+IMAGE_CHAIN_PROCESSOR = None
+VIDEO_CHAIN_PROCESSOR = None
+BATCH_IMAGE_CHAIN_PROCESSOR = None
+
+CFG: Settings = None
+
+

roop/metadata.py

@@ -0,0 +1,2 @@
+name = 'roop unleashed'
+version = '3.9.0'

roop/template_parser.py

@@ -0,0 +1,23 @@
+import re
+from datetime import datetime
+
+template_functions = {
+    "timestamp": lambda data: str(int(datetime.now().timestamp())),
+    "i": lambda data: data.get("index", False),
+    "file": lambda data: data.get("file", False),
+    "date": lambda data: datetime.now().strftime("%Y-%m-%d"),
+    "time": lambda data: datetime.now().strftime("%H-%M-%S"),
+}
+
+
+def parse(text: str, data: dict):
+    pattern = r"\{([^}]+)\}"
+
+    matches = re.findall(pattern, text)
+
+    for match in matches:
+        replacement = template_functions[match](data)
+        if replacement is not False:
+            text = text.replace(f"{{{match}}}", replacement)
+
+    return text

roop/typing.py

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

roop/util_ffmpeg.py

@@ -0,0 +1,110 @@
+
+import os
+import subprocess
+import roop.globals
+import roop.utilities as util
+
+from typing import List, Any
+
+def run_ffmpeg(args: List[str]) -> bool:
+    commands = ['ffmpeg', '-hide_banner', '-hwaccel', 'auto', '-y', '-loglevel', roop.globals.log_level]
+    commands.extend(args)
+    print ("Running ffmpeg")
+    try:
+        subprocess.check_output(commands, stderr=subprocess.STDOUT)
+        return True
+    except Exception as e:
+        print("Running ffmpeg failed! Commandline:")
+        print (" ".join(commands))
+    return False
+
+
+
+def cut_video(original_video: str, cut_video: str, start_frame: int, end_frame: int, reencode: bool):
+    fps = util.detect_fps(original_video)
+    start_time = start_frame / fps
+    num_frames = end_frame - start_frame
+
+    if reencode:
+        run_ffmpeg(['-ss',  format(start_time, ".2f"), '-i', original_video, '-c:v', roop.globals.video_encoder, '-c:a', 'aac', '-frames:v', str(num_frames), cut_video])
+    else:
+        run_ffmpeg(['-ss',  format(start_time, ".2f"), '-i', original_video,  '-frames:v', str(num_frames), '-c:v' ,'copy','-c:a' ,'copy', cut_video])
+
+def join_videos(videos: List[str], dest_filename: str, simple: bool):
+    if simple:
+        txtfilename = util.resolve_relative_path('../temp')
+        txtfilename = os.path.join(txtfilename, 'joinvids.txt')
+        with open(txtfilename, "w", encoding="utf-8") as f:
+            for v in videos:
+                 v = v.replace('\\', '/')
+                 f.write(f"file {v}\n")
+        commands = ['-f', 'concat', '-safe', '0', '-i', f'{txtfilename}', '-vcodec', 'copy', f'{dest_filename}']
+        run_ffmpeg(commands)
+
+    else:
+        inputs = []
+        filter = ''
+        for i,v in enumerate(videos):
+            inputs.append('-i')
+            inputs.append(v)
+            filter += f'[{i}:v:0][{i}:a:0]'
+        run_ffmpeg([" ".join(inputs), '-filter_complex', f'"{filter}concat=n={len(videos)}:v=1:a=1[outv][outa]"', '-map', '"[outv]"', '-map', '"[outa]"', dest_filename])    
+
+        #     filter += f'[{i}:v:0][{i}:a:0]'
+        # run_ffmpeg([" ".join(inputs), '-filter_complex', f'"{filter}concat=n={len(videos)}:v=1:a=1[outv][outa]"', '-map', '"[outv]"', '-map', '"[outa]"', dest_filename])    
+
+
+
+def extract_frames(target_path : str, trim_frame_start, trim_frame_end, fps : float) -> bool:
+    util.create_temp(target_path)
+    temp_directory_path = util.get_temp_directory_path(target_path)
+    commands = ['-i', target_path, '-q:v', '1', '-pix_fmt', 'rgb24', ]
+    if trim_frame_start is not None and trim_frame_end is not None:
+        commands.extend([ '-vf', 'trim=start_frame=' + str(trim_frame_start) + ':end_frame=' + str(trim_frame_end) + ',fps=' + str(fps) ])
+    commands.extend(['-vsync', '0', os.path.join(temp_directory_path, '%06d.' + roop.globals.CFG.output_image_format)])
+    return run_ffmpeg(commands)
+
+
+def create_video(target_path: str, dest_filename: str, fps: float = 24.0, temp_directory_path: str = None) -> None:
+    if temp_directory_path is None:
+        temp_directory_path = util.get_temp_directory_path(target_path)
+    run_ffmpeg(['-r', str(fps), '-i', os.path.join(temp_directory_path, f'%06d.{roop.globals.CFG.output_image_format}'), '-c:v', roop.globals.video_encoder, '-crf', str(roop.globals.video_quality), '-pix_fmt', 'yuv420p', '-vf', 'colorspace=bt709:iall=bt601-6-625:fast=1', '-y', dest_filename])
+    return dest_filename
+
+
+def create_gif_from_video(video_path: str, gif_path):
+    from roop.capturer import get_video_frame
+
+    fps = util.detect_fps(video_path)
+    frame = get_video_frame(video_path)
+
+    run_ffmpeg(['-i', video_path, '-vf', f'fps={fps},scale={frame.shape[0]}:-1:flags=lanczos,split[s0][s1];[s0]palettegen[p];[s1][p]paletteuse', '-loop', '0', gif_path])
+
+
+def restore_audio(intermediate_video: str, original_video: str, trim_frame_start, trim_frame_end, final_video : str) -> None:
+	fps = util.detect_fps(original_video)
+	commands = [ '-i', intermediate_video ]
+	if trim_frame_start is None and trim_frame_end is None:
+		commands.extend([ '-c:a', 'copy' ])
+	else:
+		# if trim_frame_start is not None:
+		# 	start_time = trim_frame_start / fps
+		# 	commands.extend([ '-ss', format(start_time, ".2f")])
+		# else:
+		# 	commands.extend([ '-ss', '0' ])
+		# if trim_frame_end is not None:
+		# 	end_time = trim_frame_end / fps
+		# 	commands.extend([ '-to', format(end_time, ".2f")])
+		# commands.extend([ '-c:a', 'aac' ])
+		if trim_frame_start is not None:
+			start_time = trim_frame_start / fps
+			commands.extend([ '-ss', format(start_time, ".2f")])
+		else:
+			commands.extend([ '-ss', '0' ])
+		if trim_frame_end is not None:
+			end_time = trim_frame_end / fps
+			commands.extend([ '-to', format(end_time, ".2f")])
+		commands.extend([ '-i', original_video, "-c",  "copy" ])
+
+	commands.extend([ '-map', '0:v:0', '-map', '1:a:0?', '-shortest', final_video ])
+	run_ffmpeg(commands)

roop/utilities.py

@@ -0,0 +1,339 @@
+import glob
+import mimetypes
+import os
+import platform
+import shutil
+import ssl
+import subprocess
+import sys
+import urllib
+import torch
+import gradio
+import tempfile
+import cv2
+import zipfile
+import traceback
+
+from pathlib import Path
+from typing import List, Any
+from tqdm import tqdm
+from scipy.spatial import distance
+
+import roop.template_parser as template_parser
+
+import roop.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
+
+
+# https://github.com/facefusion/facefusion/blob/master/facefusion
+def detect_fps(target_path: str) -> float:
+    fps = 24.0
+    cap = cv2.VideoCapture(target_path)
+    if cap.isOpened():
+        fps = cap.get(cv2.CAP_PROP_FPS)
+    cap.release()
+    return fps
+
+
+# Gradio wants Images in RGB
+def convert_to_gradio(image):
+    if image is None:
+        return None
+    return cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
+
+
+def sort_filenames_ignore_path(filenames):
+    """Sorts a list of filenames containing a complete path by their filename,
+    while retaining their original path.
+
+    Args:
+      filenames: A list of filenames containing a complete path.
+
+    Returns:
+      A sorted list of filenames containing a complete path.
+    """
+    filename_path_tuples = [
+        (os.path.split(filename)[1], filename) for filename in filenames
+    ]
+    sorted_filename_path_tuples = sorted(filename_path_tuples, key=lambda x: x[0])
+    return [
+        filename_path_tuple[1] for filename_path_tuple in sorted_filename_path_tuples
+    ]
+
+
+def sort_rename_frames(path: str):
+    filenames = os.listdir(path)
+    filenames.sort()
+    for i in range(len(filenames)):
+        of = os.path.join(path, filenames[i])
+        newidx = i + 1
+        new_filename = os.path.join(
+            path, f"{newidx:06d}." + roop.globals.CFG.output_image_format
+        )
+        os.rename(of, new_filename)
+
+
+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),
+                f"*.{roop.globals.CFG.output_image_format}",
+            )
+        )
+    )
+
+
+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 get_destfilename_from_path(
+    srcfilepath: str, destfilepath: str, extension: str
+) -> str:
+    fn, ext = os.path.splitext(os.path.basename(srcfilepath))
+    if "." in extension:
+        return os.path.join(destfilepath, f"{fn}{extension}")
+    return os.path.join(destfilepath, f"{fn}{extension}{ext}")
+
+
+def replace_template(file_path: str, index: int = 0) -> str:
+    fn, ext = os.path.splitext(os.path.basename(file_path))
+
+    # Remove the "__temp" placeholder that was used as a temporary filename
+    fn = fn.replace("__temp", "")
+
+    template = roop.globals.CFG.output_template
+    replaced_filename = template_parser.parse(
+        template, {"index": str(index), "file": fn}
+    )
+
+    return os.path.join(roop.globals.output_path, f"{replaced_filename}{ext}")
+
+
+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 roop.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 delete_temp_frames(filename: str) -> None:
+    dir = os.path.dirname(os.path.dirname(filename))
+    shutil.rmtree(dir)
+
+
+def has_image_extension(image_path: str) -> bool:
+    return image_path.lower().endswith(("png", "jpg", "jpeg", "webp"))
+
+
+def has_extension(filepath: str, extensions: List[str]) -> bool:
+    return filepath.lower().endswith(tuple(extensions))
+
+
+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=f"Downloading {url}",
+                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 get_local_files_from_folder(folder: str) -> List[str]:
+    if not os.path.exists(folder) or not os.path.isdir(folder):
+        return None
+    files = [
+        os.path.join(folder, f)
+        for f in os.listdir(folder)
+        if os.path.isfile(os.path.join(folder, f))
+    ]
+    return files
+
+
+def resolve_relative_path(path: str) -> str:
+    return os.path.abspath(os.path.join(os.path.dirname(__file__), path))
+
+
+def get_device() -> str:
+    if len(roop.globals.execution_providers) < 1:
+        roop.globals.execution_providers = ["CPUExecutionProvider"]
+
+    prov = roop.globals.execution_providers[0]
+    if "CoreMLExecutionProvider" in prov:
+        return "mps"
+    if "CUDAExecutionProvider" in prov or "ROCMExecutionProvider" in prov:
+        return "cuda"
+    if "OpenVINOExecutionProvider" in prov:
+        return "mkl"
+    return "cpu"
+
+
+def str_to_class(module_name, class_name) -> Any:
+    from importlib import import_module
+
+    class_ = None
+    try:
+        module_ = import_module(module_name)
+        try:
+            class_ = getattr(module_, class_name)()
+        except AttributeError:
+            print(f"Class {class_name} does not exist")
+    except ImportError:
+        print(f"Module {module_name} does not exist")
+    return class_
+
+def is_installed(name:str) -> bool:
+    return shutil.which(name);
+
+# Taken from https://stackoverflow.com/a/68842705
+def get_platform() -> str:
+    if sys.platform == "linux":
+        try:
+            proc_version = open("/proc/version").read()
+            if "Microsoft" in proc_version:
+                return "wsl"
+        except:
+            pass
+    return sys.platform
+
+def open_with_default_app(filename:str):
+    if filename == None:
+        return
+    platform = get_platform()
+    if platform == "darwin":
+        subprocess.call(("open", filename))
+    elif platform in ["win64", "win32"]:        os.startfile(filename.replace("/", "\\"))
+    elif platform == "wsl":
+        subprocess.call("cmd.exe /C start".split() + [filename])
+    else:  # linux variants
+        subprocess.call("xdg-open", filename)
+
+
+def prepare_for_batch(target_files) -> str:
+    print("Preparing temp files")
+    tempfolder = os.path.join(tempfile.gettempdir(), "rooptmp")
+    if os.path.exists(tempfolder):
+        shutil.rmtree(tempfolder)
+    Path(tempfolder).mkdir(parents=True, exist_ok=True)
+    for f in target_files:
+        newname = os.path.basename(f.name)
+        shutil.move(f.name, os.path.join(tempfolder, newname))
+    return tempfolder
+
+
+def zip(files, zipname):
+    with zipfile.ZipFile(zipname, "w") as zip_file:
+        for f in files:
+            zip_file.write(f, os.path.basename(f))
+
+
+def unzip(zipfilename: str, target_path: str):
+    with zipfile.ZipFile(zipfilename, "r") as zip_file:
+        zip_file.extractall(target_path)
+
+
+def mkdir_with_umask(directory):
+    oldmask = os.umask(0)
+    # mode needs octal
+    os.makedirs(directory, mode=0o775, exist_ok=True)
+    os.umask(oldmask)
+
+
+def open_folder(path: str):
+    platform = get_platform()
+    try:
+        if platform == "darwin":
+            subprocess.call(("open", path))
+        elif platform in ["win64", "win32"]:
+            open_with_default_app(path)
+        elif platform == "wsl":
+            subprocess.call("cmd.exe /C start".split() + [path])
+        else:  # linux variants
+            subprocess.Popen(["xdg-open", path])
+    except Exception as e:
+        traceback.print_exc()
+        pass
+        # import webbrowser
+        # webbrowser.open(url)
+
+
+def create_version_html() -> str:
+    python_version = ".".join([str(x) for x in sys.version_info[0:3]])
+    versions_html = f"""
+python: <span title="{sys.version}">{python_version}</span>
+•
+torch: {getattr(torch, '__long_version__',torch.__version__)}
+•
+gradio: {gradio.__version__}
+"""
+    return versions_html
+
+
+def compute_cosine_distance(emb1, emb2) -> float:
+    return distance.cosine(emb1, emb2)

roop/virtualcam.py

@@ -0,0 +1,88 @@
+import cv2
+import roop.globals
+import ui.globals
+import pyvirtualcam
+import threading
+import time
+
+
+cam_active = False
+cam_thread = None
+vcam = None
+
+def virtualcamera(streamobs, cam_num,width,height):
+    from roop.ProcessOptions import ProcessOptions
+    from roop.core import live_swap, get_processing_plugins
+    from roop.filters import fast_quantize_to_palette
+
+    global cam_active
+
+    #time.sleep(2)
+    print('Starting capture')
+    cap = cv2.VideoCapture(cam_num, cv2.CAP_DSHOW)
+    if not cap.isOpened():
+        print("Cannot open camera")
+        cap.release()
+        del cap
+        return
+
+    pref_width = width
+    pref_height = height
+    pref_fps_in = 30
+    cap.set(cv2.CAP_PROP_FRAME_WIDTH, pref_width)
+    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, pref_height)
+    cap.set(cv2.CAP_PROP_FPS, pref_fps_in)
+    cam_active = True
+
+    # native format UYVY
+
+    cam = None
+    if streamobs:
+        print('Detecting virtual cam devices')
+        cam = pyvirtualcam.Camera(width=pref_width, height=pref_height, fps=pref_fps_in, fmt=pyvirtualcam.PixelFormat.BGR, print_fps=False)
+    if cam:
+        print(f'Using virtual camera: {cam.device}')
+        print(f'Using {cam.native_fmt}')
+    else:
+        print(f'Not streaming to virtual camera!')
+
+    while cam_active:
+        ret, frame = cap.read()
+        if not ret:
+            break
+
+        if len(roop.globals.INPUT_FACESETS) > 0:
+            options = ProcessOptions(get_processing_plugins(None), roop.globals.distance_threshold, roop.globals.blend_ratio,
+                              "all", 0, None, None, 1, False)
+            frame = live_swap(frame, options)
+        #frame = fast_quantize_to_palette(frame)
+        if cam:
+            cam.send(frame)
+            cam.sleep_until_next_frame()
+        ui.globals.ui_camera_frame = frame
+
+    if cam:
+        cam.close()
+    cap.release()
+    print('Camera stopped')
+
+
+
+def start_virtual_cam(streamobs, cam_number, resolution):
+    global cam_thread, cam_active
+
+    if not cam_active:
+        width, height = map(int, resolution.split('x'))
+        cam_thread = threading.Thread(target=virtualcamera, args=[streamobs, cam_number, width, height])
+        cam_thread.start()
+
+
+
+def stop_virtual_cam():
+    global cam_active, cam_thread
+
+    if cam_active:
+        cam_active = False
+        cam_thread.join()
+    
+

roop/vr_util.py

@@ -0,0 +1,57 @@
+import cv2
+import numpy as np
+
+# VR Lense Distortion
+# Taken from https://github.com/g0kuvonlange/vrswap
+
+
+def get_perspective(img, FOV, THETA, PHI, height, width):
+    #
+    # THETA is left/right angle, PHI is up/down angle, both in degree
+    #
+    [orig_width, orig_height, _] = img.shape
+    equ_h = orig_height
+    equ_w = orig_width
+    equ_cx = (equ_w - 1) / 2.0
+    equ_cy = (equ_h - 1) / 2.0
+
+    wFOV = FOV
+    hFOV = float(height) / width * wFOV
+
+    w_len = np.tan(np.radians(wFOV / 2.0))
+    h_len = np.tan(np.radians(hFOV / 2.0))
+
+    x_map = np.ones([height, width], np.float32)
+    y_map = np.tile(np.linspace(-w_len, w_len, width), [height, 1])
+    z_map = -np.tile(np.linspace(-h_len, h_len, height), [width, 1]).T
+
+    D = np.sqrt(x_map**2 + y_map**2 + z_map**2)
+    xyz = np.stack((x_map, y_map, z_map), axis=2) / np.repeat(
+        D[:, :, np.newaxis], 3, axis=2
+    )
+
+    y_axis = np.array([0.0, 1.0, 0.0], np.float32)
+    z_axis = np.array([0.0, 0.0, 1.0], np.float32)
+    [R1, _] = cv2.Rodrigues(z_axis * np.radians(THETA))
+    [R2, _] = cv2.Rodrigues(np.dot(R1, y_axis) * np.radians(-PHI))
+
+    xyz = xyz.reshape([height * width, 3]).T
+    xyz = np.dot(R1, xyz)
+    xyz = np.dot(R2, xyz).T
+    lat = np.arcsin(xyz[:, 2])
+    lon = np.arctan2(xyz[:, 1], xyz[:, 0])
+
+    lon = lon.reshape([height, width]) / np.pi * 180
+    lat = -lat.reshape([height, width]) / np.pi * 180
+
+    lon = lon / 180 * equ_cx + equ_cx
+    lat = lat / 90 * equ_cy + equ_cy
+
+    persp = cv2.remap(
+        img,
+        lon.astype(np.float32),
+        lat.astype(np.float32),
+        cv2.INTER_CUBIC,
+        borderMode=cv2.BORDER_WRAP,
+    )
+    return persp