Ii
commited on
Delete refacer.py
Browse files- refacer.py +0 -154
refacer.py
DELETED
|
@@ -1,154 +0,0 @@
|
|
| 1 |
-
import cv2
|
| 2 |
-
import onnxruntime as rt
|
| 3 |
-
import sys
|
| 4 |
-
from insightface.app import FaceAnalysis
|
| 5 |
-
sys.path.insert(1, './recognition')
|
| 6 |
-
from scrfd import SCRFD
|
| 7 |
-
from arcface_onnx import ArcFaceONNX
|
| 8 |
-
import os
|
| 9 |
-
from pathlib import Path
|
| 10 |
-
from tqdm import tqdm
|
| 11 |
-
from insightface.model_zoo.inswapper import INSwapper
|
| 12 |
-
import psutil
|
| 13 |
-
from enum import Enum
|
| 14 |
-
from insightface.app.common import Face
|
| 15 |
-
from insightface.utils.storage import ensure_available
|
| 16 |
-
|
| 17 |
-
class RefacerMode(Enum):
|
| 18 |
-
CPU, CUDA, COREML, TENSORRT = range(1, 5)
|
| 19 |
-
|
| 20 |
-
class Refacer:
|
| 21 |
-
def __init__(self, force_cpu=False, colab_performance=False):
|
| 22 |
-
self.first_face = False
|
| 23 |
-
self.force_cpu = force_cpu
|
| 24 |
-
self.colab_performance = colab_performance
|
| 25 |
-
self.__check_providers()
|
| 26 |
-
self.__init_apps()
|
| 27 |
-
|
| 28 |
-
def __check_providers(self):
|
| 29 |
-
if self.force_cpu:
|
| 30 |
-
self.providers = ['CPUExecutionProvider']
|
| 31 |
-
else:
|
| 32 |
-
self.providers = rt.get_available_providers()
|
| 33 |
-
rt.set_default_logger_severity(4)
|
| 34 |
-
self.sess_options = rt.SessionOptions()
|
| 35 |
-
self.sess_options.execution_mode = rt.ExecutionMode.ORT_SEQUENTIAL
|
| 36 |
-
self.sess_options.graph_optimization_level = rt.GraphOptimizationLevel.ORT_ENABLE_ALL
|
| 37 |
-
|
| 38 |
-
if len(self.providers) == 1 and 'CPUExecutionProvider' in self.providers:
|
| 39 |
-
self.mode = RefacerMode.CPU
|
| 40 |
-
self.use_num_cpus = psutil.cpu_count(logical=False)
|
| 41 |
-
print(f"CPU mode with providers {self.providers}")
|
| 42 |
-
elif self.colab_performance:
|
| 43 |
-
self.mode = RefacerMode.TENSORRT
|
| 44 |
-
print(f"TENSORRT mode with providers {self.providers}")
|
| 45 |
-
elif 'CUDAExecutionProvider' in self.providers:
|
| 46 |
-
self.mode = RefacerMode.CUDA
|
| 47 |
-
print(f"CUDA mode with providers {self.providers}")
|
| 48 |
-
|
| 49 |
-
def __init_apps(self):
|
| 50 |
-
assets_dir = ensure_available('models', 'buffalo_l', root='~/.insightface')
|
| 51 |
-
|
| 52 |
-
model_path = os.path.join(assets_dir, 'det_10g.onnx')
|
| 53 |
-
sess_face = rt.InferenceSession(model_path, self.sess_options, providers=self.providers)
|
| 54 |
-
self.face_detector = SCRFD(model_path, sess_face)
|
| 55 |
-
self.face_detector.prepare(0, input_size=(640, 640))
|
| 56 |
-
|
| 57 |
-
model_path = os.path.join(assets_dir, 'w600k_r50.onnx')
|
| 58 |
-
sess_rec = rt.InferenceSession(model_path, self.sess_options, providers=self.providers)
|
| 59 |
-
self.rec_app = ArcFaceONNX(model_path, sess_rec)
|
| 60 |
-
self.rec_app.prepare(0)
|
| 61 |
-
|
| 62 |
-
model_path = 'inswapper_128.onnx'
|
| 63 |
-
sess_swap = rt.InferenceSession(model_path, self.sess_options, providers=self.providers)
|
| 64 |
-
self.face_swapper = INSwapper(model_path, sess_swap)
|
| 65 |
-
|
| 66 |
-
def prepare_faces(self, faces):
|
| 67 |
-
self.replacement_faces = []
|
| 68 |
-
for face in faces:
|
| 69 |
-
if "origin" in face:
|
| 70 |
-
bboxes1, kpss1 = self.face_detector.autodetect(face['origin'], max_num=1)
|
| 71 |
-
if len(kpss1) < 1:
|
| 72 |
-
raise Exception('No face detected on "Face to replace" image')
|
| 73 |
-
feat_original = self.rec_app.get(face['origin'], kpss1[0])
|
| 74 |
-
else:
|
| 75 |
-
feat_original = None
|
| 76 |
-
self.first_face = True
|
| 77 |
-
print('No origin image: First face change')
|
| 78 |
-
_faces = self.__get_faces(face['destination'], max_num=1)
|
| 79 |
-
if len(_faces) < 1:
|
| 80 |
-
raise Exception('No face detected on "Destination face" image')
|
| 81 |
-
self.replacement_faces.append((feat_original, _faces[0], face.get('threshold', 0)))
|
| 82 |
-
|
| 83 |
-
def __get_faces(self, frame, max_num=0):
|
| 84 |
-
bboxes, kpss = self.face_detector.detect(frame, max_num=max_num, metric='default')
|
| 85 |
-
if bboxes.shape[0] == 0:
|
| 86 |
-
return []
|
| 87 |
-
ret = []
|
| 88 |
-
for i in range(bboxes.shape[0]):
|
| 89 |
-
bbox = bboxes[i, 0:4]
|
| 90 |
-
det_score = bboxes[i, 4]
|
| 91 |
-
kps = None
|
| 92 |
-
if kpss is not None:
|
| 93 |
-
kps = kpss[i]
|
| 94 |
-
face = Face(bbox=bbox, kps=kps, det_score=det_score)
|
| 95 |
-
face.embedding = self.rec_app.get(frame, kps)
|
| 96 |
-
ret.append(face)
|
| 97 |
-
return ret
|
| 98 |
-
|
| 99 |
-
def process_first_face(self, frame):
|
| 100 |
-
faces = self.__get_faces(frame, max_num=1)
|
| 101 |
-
if len(faces) != 0:
|
| 102 |
-
frame = self.face_swapper.get(frame, faces[0], self.replacement_faces[0][1], paste_back=True)
|
| 103 |
-
return frame
|
| 104 |
-
|
| 105 |
-
def process_faces(self, frame):
|
| 106 |
-
faces = self.__get_faces(frame, max_num=0)
|
| 107 |
-
for rep_face in self.replacement_faces:
|
| 108 |
-
for i in range(len(faces) - 1, -1, -1):
|
| 109 |
-
sim = self.rec_app.compute_sim(rep_face[0], faces[i].embedding)
|
| 110 |
-
if sim >= rep_face[2]:
|
| 111 |
-
frame = self.face_swapper.get(frame, faces[i], rep_face[1], paste_back=True)
|
| 112 |
-
del faces[i]
|
| 113 |
-
break
|
| 114 |
-
return frame
|
| 115 |
-
|
| 116 |
-
def reface_group(self, faces, frames, output):
|
| 117 |
-
for frame in frames:
|
| 118 |
-
if self.first_face:
|
| 119 |
-
output.write(self.process_first_face(frame))
|
| 120 |
-
else:
|
| 121 |
-
output.write(self.process_faces(frame))
|
| 122 |
-
|
| 123 |
-
def reface(self, video_path, faces):
|
| 124 |
-
print(f"Refacing video: {video_path}")
|
| 125 |
-
# Output video path is the same as input video, with "_refaced" suffix
|
| 126 |
-
output_video_path = f"{os.path.splitext(video_path)[0]}_refaced.mp4"
|
| 127 |
-
self.prepare_faces(faces)
|
| 128 |
-
|
| 129 |
-
cap = cv2.VideoCapture(video_path)
|
| 130 |
-
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
|
| 131 |
-
fps = cap.get(cv2.CAP_PROP_FPS)
|
| 132 |
-
frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
|
| 133 |
-
frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
|
| 134 |
-
|
| 135 |
-
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
|
| 136 |
-
output = cv2.VideoWriter(output_video_path, fourcc, fps, (frame_width, frame_height))
|
| 137 |
-
|
| 138 |
-
frames = []
|
| 139 |
-
with tqdm(total=total_frames, desc="Processing frames") as pbar:
|
| 140 |
-
while cap.isOpened():
|
| 141 |
-
ret, frame = cap.read()
|
| 142 |
-
if not ret:
|
| 143 |
-
break
|
| 144 |
-
frames.append(frame)
|
| 145 |
-
pbar.update()
|
| 146 |
-
|
| 147 |
-
cap.release()
|
| 148 |
-
|
| 149 |
-
# Process and save refaced video
|
| 150 |
-
self.reface_group(faces, frames, output)
|
| 151 |
-
output.release()
|
| 152 |
-
|
| 153 |
-
print(f"Refaced video saved at {output_video_path}")
|
| 154 |
-
return output_video_path # Return final path of video
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|