mirror of
https://github.com/iperov/DeepFaceLive
synced 2025-08-14 10:47:00 -07:00
+xlib.facemeta and refactoring
This commit is contained in:
iperov
parent
081dde23c7
commit
63adc2995e
28 changed files with 962 additions and 613 deletions
|
|
@ -5,10 +5,32 @@ from typing import List, Union, Tuple
|
|||
import numpy as np
|
||||
from xlib import mp as lib_mp
|
||||
from xlib import time as lib_time
|
||||
from xlib.facemeta import FaceMark
|
||||
from xlib.mp import csw as lib_csw
|
||||
from xlib.python.EventListener import EventListener
|
||||
|
||||
from xlib.facemeta import FRect, FLandmarks2D, FPose
|
||||
|
||||
class BackendFaceSwapInfo:
|
||||
def __init__(self):
|
||||
self.image_name = None
|
||||
self.face_urect : FRect = None
|
||||
self.face_pose : FPose = None
|
||||
self.face_ulmrks : FLandmarks2D = None
|
||||
|
||||
self.face_align_image_name : str = None
|
||||
self.face_align_mask_name : str = None
|
||||
self.face_swap_image_name : str = None
|
||||
self.face_swap_mask_name : str = None
|
||||
|
||||
self.image_to_align_uni_mat = None
|
||||
self.face_align_ulmrks : FLandmarks2D = None
|
||||
|
||||
def __getstate__(self):
|
||||
return self.__dict__.copy()
|
||||
|
||||
def __setstate__(self, d):
|
||||
self.__init__()
|
||||
self.__dict__.update(d)
|
||||
|
||||
class BackendConnectionData:
|
||||
"""
|
||||
|
|
@ -27,13 +49,14 @@ class BackendConnectionData:
|
|||
self._uid = uid
|
||||
self._is_frame_reemitted = None
|
||||
|
||||
self._frame_name = None
|
||||
self._frame_image_name = None
|
||||
self._frame_count = None
|
||||
self._frame_num = None
|
||||
self._frame_fps = None
|
||||
self._frame_timestamp = None
|
||||
self._merged_frame_name = None
|
||||
self._face_mark_list = []
|
||||
self._merged_image_name = None
|
||||
|
||||
self._face_swap_info_list = []
|
||||
|
||||
def __getstate__(self, ):
|
||||
d = self.__dict__.copy()
|
||||
|
|
@ -43,43 +66,43 @@ class BackendConnectionData:
|
|||
def assign_weak_heap(self, weak_heap : lib_mp.MPWeakHeap):
|
||||
self._weak_heap = weak_heap
|
||||
|
||||
def set_file(self, name : str, data : Union[bytes, bytearray, memoryview]):
|
||||
self._weak_heap_refs[name] = self._weak_heap.add_data(data)
|
||||
def set_file(self, key, data : Union[bytes, bytearray, memoryview]):
|
||||
self._weak_heap_refs[key] = self._weak_heap.add_data(data)
|
||||
|
||||
def get_file(self, name : str) -> Union[bytes, None]:
|
||||
ref = self._weak_heap_refs.get(name, None)
|
||||
def get_file(self, key) -> Union[bytes, None]:
|
||||
ref = self._weak_heap_refs.get(key, None)
|
||||
if ref is not None:
|
||||
return self._weak_heap.get_data(ref)
|
||||
return None
|
||||
|
||||
def set_image(self, name : str, image : np.ndarray):
|
||||
def set_image(self, key, image : np.ndarray):
|
||||
"""
|
||||
store image to weak heap
|
||||
|
||||
name str
|
||||
key
|
||||
|
||||
image np.ndarray
|
||||
"""
|
||||
self.set_file(name, image.data)
|
||||
self._weak_heap_image_infos[name] = (image.shape, image.dtype)
|
||||
|
||||
def get_image_shape_dtype(self, name:str) -> Union[None, Tuple[List, np.dtype]]:
|
||||
self.set_file(key, image.data)
|
||||
self._weak_heap_image_infos[key] = (image.shape, image.dtype)
|
||||
|
||||
def get_image_shape_dtype(self, key) -> Union[None, Tuple[List, np.dtype]]:
|
||||
"""
|
||||
returns (image shape, dtype) or (None, None) if file does not exist
|
||||
"""
|
||||
if name is None:
|
||||
return None
|
||||
image_info = self._weak_heap_image_infos.get(name, None)
|
||||
if key is None:
|
||||
return (None, None)
|
||||
image_info = self._weak_heap_image_infos.get(key, None)
|
||||
if image_info is not None:
|
||||
shape, dtype = image_info
|
||||
return shape, dtype
|
||||
return (None, None)
|
||||
|
||||
def get_image(self, name : str) -> Union[np.ndarray, None]:
|
||||
if name is None:
|
||||
|
||||
def get_image(self, key) -> Union[np.ndarray, None]:
|
||||
if key is None:
|
||||
return None
|
||||
image_info = self._weak_heap_image_infos.get(name, None)
|
||||
buffer = self.get_file(name)
|
||||
image_info = self._weak_heap_image_infos.get(key, None)
|
||||
buffer = self.get_file(key)
|
||||
|
||||
if image_info is not None and buffer is not None:
|
||||
shape, dtype = image_info
|
||||
|
|
@ -90,10 +113,6 @@ class BackendConnectionData:
|
|||
|
||||
def get_is_frame_reemitted(self) -> Union[bool, None]: return self._is_frame_reemitted
|
||||
def set_is_frame_reemitted(self, is_frame_reemitted : bool): self._is_frame_reemitted = is_frame_reemitted
|
||||
|
||||
def get_frame_name(self) -> Union[str, None]: return self._frame_name
|
||||
def set_frame_name(self, frame_name : str): self._frame_name = frame_name
|
||||
|
||||
def get_frame_count(self) -> Union[int, None]: return self._frame_count
|
||||
def set_frame_count(self, frame_count : int): self._frame_count = frame_count
|
||||
def get_frame_num(self) -> Union[int, None]: return self._frame_num
|
||||
|
|
@ -103,14 +122,17 @@ class BackendConnectionData:
|
|||
def get_frame_timestamp(self) -> Union[float, None]: return self._frame_timestamp
|
||||
def set_frame_timestamp(self, frame_timestamp : float): self._frame_timestamp = frame_timestamp
|
||||
|
||||
def get_merged_frame_name(self) -> Union[str, None]: return self._merged_frame_name
|
||||
def set_merged_frame_name(self, merged_frame_name : str): self._merged_frame_name = merged_frame_name
|
||||
def get_frame_image_name(self) -> Union[str, None]: return self._frame_image_name
|
||||
def set_frame_image_name(self, frame_image_name : str): self._frame_image_name = frame_image_name
|
||||
def get_merged_image_name(self) -> Union[str, None]: return self._merged_image_name
|
||||
def set_merged_image_name(self, merged_frame_name : str): self._merged_image_name = merged_frame_name
|
||||
|
||||
def get_face_swap_info_list(self) -> List[BackendFaceSwapInfo]: return self._face_swap_info_list
|
||||
def add_face_swap_info(self, fsi : BackendFaceSwapInfo):
|
||||
if not isinstance(fsi, BackendFaceSwapInfo):
|
||||
raise ValueError(f'fsi must be an instance of BackendFaceSwapInfo')
|
||||
self._face_swap_info_list.append(fsi)
|
||||
|
||||
def get_face_mark_list(self) -> List[FaceMark]: return self._face_mark_list
|
||||
def add_face_mark(self, face_mark : FaceMark):
|
||||
if not isinstance(face_mark, FaceMark):
|
||||
raise ValueError(f'face_mark must be an instance of FaceMark')
|
||||
self._face_mark_list.append(face_mark)
|
||||
|
||||
|
||||
class BackendConnection:
|
||||
|
|
@ -144,7 +166,7 @@ class BackendConnection:
|
|||
def is_full_read(self, buffer_size=0) -> bool:
|
||||
"""
|
||||
if fully readed by receiver side minus buffer_size
|
||||
"""
|
||||
"""
|
||||
return self._rd.get_read_id() >= (self._rd.get_write_id() - buffer_size)
|
||||
|
||||
|
||||
|
|
@ -205,3 +227,4 @@ class BackendWorker(lib_csw.Worker):
|
|||
|
||||
def stop_profile_timing(self):
|
||||
self.send_msg('_profile_timing', self._profile_timing_measurer.stop() )
|
||||
|
||||
|
|
|
|||
|
|
@ -253,7 +253,7 @@ class CameraSourceWorker(BackendWorker):
|
|||
|
||||
bcd.assign_weak_heap(self.weak_heap)
|
||||
frame_name = f'Camera_{state.device_idx}_{bcd_uid:06}'
|
||||
bcd.set_frame_name(frame_name)
|
||||
bcd.set_frame_image_name(frame_name)
|
||||
bcd.set_frame_num(bcd_uid)
|
||||
bcd.set_frame_timestamp(timestamp)
|
||||
bcd.set_image(frame_name, img)
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@ import time
|
|||
|
||||
import numpy as np
|
||||
from xlib import os as lib_os
|
||||
from xlib.facemeta import FaceAlign, FaceULandmarks
|
||||
from xlib.mp import csw as lib_csw
|
||||
from xlib.python import all_is_not_None
|
||||
|
||||
|
|
@ -51,7 +50,7 @@ class FaceAlignerWorker(BackendWorker):
|
|||
|
||||
cs.exclude_moving_parts.enable()
|
||||
cs.exclude_moving_parts.set_flag(state.exclude_moving_parts if state.exclude_moving_parts is not None else True)
|
||||
|
||||
|
||||
cs.head_mode.enable()
|
||||
cs.head_mode.set_flag(state.head_mode if state.head_mode is not None else False)
|
||||
|
||||
|
|
@ -91,7 +90,7 @@ class FaceAlignerWorker(BackendWorker):
|
|||
state.head_mode = head_mode
|
||||
self.save_state()
|
||||
self.reemit_frame_signal.send()
|
||||
|
||||
|
||||
def on_cs_x_offset(self, x_offset):
|
||||
state, cs = self.get_state(), self.get_control_sheet()
|
||||
cfg = cs.x_offset.get_config()
|
||||
|
|
@ -118,21 +117,17 @@ class FaceAlignerWorker(BackendWorker):
|
|||
if bcd is not None:
|
||||
bcd.assign_weak_heap(self.weak_heap)
|
||||
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image_name = bcd.get_frame_image_name()
|
||||
frame_image = bcd.get_image(frame_image_name)
|
||||
|
||||
if all_is_not_None(state.face_coverage, state.resolution, frame_name, frame_image):
|
||||
for face_id,face_mark in enumerate( bcd.get_face_mark_list() ):
|
||||
face_ulmrks = face_mark.get_face_ulandmarks_by_type(FaceULandmarks.Type.LANDMARKS_468)
|
||||
if face_ulmrks is None:
|
||||
face_ulmrks = face_mark.get_face_ulandmarks_by_type(FaceULandmarks.Type.LANDMARKS_68)
|
||||
|
||||
if all_is_not_None(state.face_coverage, state.resolution, frame_image):
|
||||
for face_id, fsi in enumerate( bcd.get_face_swap_info_list() ):
|
||||
head_yaw = None
|
||||
if state.head_mode:
|
||||
face_pose = face_mark.get_face_pose()
|
||||
if face_pose is not None:
|
||||
head_yaw = face_pose.as_radians()[1]
|
||||
|
||||
if fsi.face_pose is not None:
|
||||
head_yaw = fsi.face_pose.as_radians()[1]
|
||||
|
||||
face_ulmrks = fsi.face_ulmrks
|
||||
if face_ulmrks is not None:
|
||||
face_image, uni_mat = face_ulmrks.cut(frame_image, state.face_coverage, state.resolution,
|
||||
exclude_moving_parts=state.exclude_moving_parts,
|
||||
|
|
@ -140,19 +135,11 @@ class FaceAlignerWorker(BackendWorker):
|
|||
x_offset=state.x_offset,
|
||||
y_offset=state.y_offset)
|
||||
|
||||
face_align_image_name = f'{frame_name}_{face_id}_aligned'
|
||||
fsi.face_align_image_name = f'{frame_image_name}_{face_id}_aligned'
|
||||
fsi.image_to_align_uni_mat = uni_mat
|
||||
fsi.face_align_ulmrks = face_ulmrks.transform(uni_mat)
|
||||
|
||||
face_align = FaceAlign()
|
||||
face_align.set_image_name(face_align_image_name)
|
||||
face_align.set_coverage(state.face_coverage)
|
||||
face_align.set_source_face_ulandmarks_type(face_ulmrks.get_type())
|
||||
face_align.set_source_to_aligned_uni_mat(uni_mat)
|
||||
|
||||
for face_ulmrks in face_mark.get_face_ulandmarks_list():
|
||||
face_align.add_face_ulandmarks( face_ulmrks.transform(uni_mat) )
|
||||
face_mark.set_face_align(face_align)
|
||||
|
||||
bcd.set_image(face_align_image_name, face_image)
|
||||
bcd.set_image(fsi.face_align_image_name, face_image)
|
||||
|
||||
self.stop_profile_timing()
|
||||
self.pending_bcd = bcd
|
||||
|
|
|
|||
|
|
@ -4,14 +4,14 @@ from enum import IntEnum
|
|||
import numpy as np
|
||||
from modelhub import onnx as onnx_models
|
||||
from xlib import os as lib_os
|
||||
from xlib.facemeta import FaceMark, FaceURect
|
||||
from xlib.facemeta import FRect
|
||||
from xlib.image import ImageProcessor
|
||||
from xlib.mp import csw as lib_csw
|
||||
from xlib.python import all_is_not_None
|
||||
|
||||
from .BackendBase import (BackendConnection, BackendDB, BackendHost,
|
||||
BackendSignal, BackendWeakHeap, BackendWorker,
|
||||
BackendWorkerState)
|
||||
BackendWorkerState, BackendFaceSwapInfo)
|
||||
|
||||
|
||||
class DetectorType(IntEnum):
|
||||
|
|
@ -213,10 +213,10 @@ class FaceDetectorWorker(BackendWorker):
|
|||
|
||||
detector_state = state.get_detector_state()
|
||||
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image_name = bcd.get_frame_image_name()
|
||||
frame_image = bcd.get_image(frame_image_name)
|
||||
|
||||
if all_is_not_None(frame_image, frame_name):
|
||||
if frame_image is not None:
|
||||
_,H,W,_ = ImageProcessor(frame_image).get_dims()
|
||||
|
||||
rects = []
|
||||
|
|
@ -228,13 +228,13 @@ class FaceDetectorWorker(BackendWorker):
|
|||
rects = self.YoloV5Face.extract (frame_image, threshold=detector_state.threshold, fixed_window=detector_state.fixed_window_size)[0]
|
||||
|
||||
# to list of FaceURect
|
||||
rects = [ FaceURect.from_ltrb( (l/W, t/H, r/W, b/H) ) for l,t,r,b in rects ]
|
||||
rects = [ FRect.from_ltrb( (l/W, t/H, r/W, b/H) ) for l,t,r,b in rects ]
|
||||
|
||||
# sort
|
||||
if detector_state.sort_by == FaceSortBy.LARGEST:
|
||||
rects = FaceURect.sort_by_area_size(rects)
|
||||
rects = FRect.sort_by_area_size(rects)
|
||||
elif detector_state.sort_by == FaceSortBy.DIST_FROM_CENTER:
|
||||
rects = FaceURect.sort_by_dist_from_center(rects)
|
||||
rects = FRect.sort_by_dist_from_center(rects)
|
||||
|
||||
if len(rects) != 0:
|
||||
max_faces = detector_state.max_faces
|
||||
|
|
@ -245,20 +245,20 @@ class FaceDetectorWorker(BackendWorker):
|
|||
if len(self.temporal_rects) != len(rects):
|
||||
self.temporal_rects = [ [] for _ in range(len(rects)) ]
|
||||
|
||||
for face_id, face_rect in enumerate(rects):
|
||||
for face_id, face_urect in enumerate(rects):
|
||||
if detector_state.temporal_smoothing != 1:
|
||||
if not is_frame_reemitted or len(self.temporal_rects[face_id]) == 0:
|
||||
self.temporal_rects[face_id].append( face_rect.as_4pts() )
|
||||
self.temporal_rects[face_id].append( face_urect.as_4pts() )
|
||||
|
||||
self.temporal_rects[face_id] = self.temporal_rects[face_id][-detector_state.temporal_smoothing:]
|
||||
|
||||
face_rect = FaceURect.from_4pts ( np.mean(self.temporal_rects[face_id],0 ) )
|
||||
face_urect = FRect.from_4pts ( np.mean(self.temporal_rects[face_id],0 ) )
|
||||
|
||||
if face_rect.get_area() != 0:
|
||||
face_mark = FaceMark()
|
||||
face_mark.set_image_name(frame_name)
|
||||
face_mark.set_face_urect ( face_rect )
|
||||
bcd.add_face_mark(face_mark)
|
||||
if face_urect.get_area() != 0:
|
||||
fsi = BackendFaceSwapInfo()
|
||||
fsi.image_name = frame_image_name
|
||||
fsi.face_urect = face_urect
|
||||
bcd.add_face_swap_info(fsi)
|
||||
|
||||
self.stop_profile_timing()
|
||||
self.pending_bcd = bcd
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ from modelhub import onnx as onnx_models
|
|||
from modelhub import cv as cv_models
|
||||
|
||||
from xlib import os as lib_os
|
||||
from xlib.facemeta import FaceULandmarks, FacePose
|
||||
from xlib.facemeta import ELandmarks2D, FLandmarks2D, FPose
|
||||
from xlib.image import ImageProcessor
|
||||
from xlib.mp import csw as lib_csw
|
||||
from xlib.python import all_is_not_None
|
||||
|
|
@ -153,20 +153,18 @@ class FaceMarkerWorker(BackendWorker):
|
|||
is_google_facemesh = marker_type == MarkerType.GOOGLE_FACEMESH and self.google_facemesh is not None
|
||||
|
||||
if marker_type is not None:
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image = bcd.get_image(bcd.get_frame_image_name())
|
||||
|
||||
if all_is_not_None(frame_image) and (is_opencv_lbf or is_google_facemesh):
|
||||
face_mark_list = bcd.get_face_mark_list()
|
||||
if frame_image is not None and (is_opencv_lbf or is_google_facemesh):
|
||||
fsi_list = bcd.get_face_swap_info_list()
|
||||
if marker_state.temporal_smoothing != 1 and \
|
||||
len(self.temporal_lmrks) != len(face_mark_list):
|
||||
self.temporal_lmrks = [ [] for _ in range(len(face_mark_list)) ]
|
||||
len(self.temporal_lmrks) != len(fsi_list):
|
||||
self.temporal_lmrks = [ [] for _ in range(len(fsi_list)) ]
|
||||
|
||||
for face_id, face_mark in enumerate(face_mark_list):
|
||||
face_mark_rect = face_mark.get_face_urect()
|
||||
if face_mark_rect is not None:
|
||||
for face_id, fsi in enumerate(fsi_list):
|
||||
if fsi.face_urect is not None:
|
||||
# Cut the face to feed to the face marker
|
||||
face_image, face_uni_mat = face_mark_rect.cut(frame_image, marker_state.marker_coverage, 256 if is_opencv_lbf else \
|
||||
face_image, face_uni_mat = fsi.face_urect.cut(frame_image, marker_state.marker_coverage, 256 if is_opencv_lbf else \
|
||||
192 if is_google_facemesh else 0 )
|
||||
_,H,W,_ = ImageProcessor(face_image).get_dims()
|
||||
|
||||
|
|
@ -183,20 +181,17 @@ class FaceMarkerWorker(BackendWorker):
|
|||
lmrks = np.mean(self.temporal_lmrks[face_id],0 )
|
||||
|
||||
if is_google_facemesh:
|
||||
face_mark.set_face_pose(FacePose.from_3D_468_landmarks(lmrks))
|
||||
fsi.face_pose = FPose.from_3D_468_landmarks(lmrks)
|
||||
|
||||
if is_opencv_lbf:
|
||||
lmrks /= (W,H)
|
||||
elif is_google_facemesh:
|
||||
lmrks = lmrks[...,0:2] / (W,H)
|
||||
|
||||
face_ulmrks = FaceULandmarks.create (FaceULandmarks.Type.LANDMARKS_68 if is_opencv_lbf else \
|
||||
FaceULandmarks.Type.LANDMARKS_468 if is_google_facemesh else None, lmrks)
|
||||
|
||||
face_ulmrks = FLandmarks2D.create (ELandmarks2D.L68 if is_opencv_lbf else \
|
||||
ELandmarks2D.L468 if is_google_facemesh else None, lmrks)
|
||||
face_ulmrks = face_ulmrks.transform(face_uni_mat, invert=True)
|
||||
face_mark.add_face_ulandmarks (face_ulmrks)
|
||||
|
||||
|
||||
fsi.face_ulmrks = face_ulmrks
|
||||
|
||||
self.stop_profile_timing()
|
||||
self.pending_bcd = bcd
|
||||
|
|
|
|||
|
|
@ -255,44 +255,35 @@ class FaceMergerWorker(BackendWorker):
|
|||
if bcd is not None:
|
||||
bcd.assign_weak_heap(self.weak_heap)
|
||||
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image_name = bcd.get_frame_image_name()
|
||||
frame_image = bcd.get_image(frame_image_name)
|
||||
|
||||
if frame_image is not None:
|
||||
for face_mark in bcd.get_face_mark_list():
|
||||
face_align = face_mark.get_face_align()
|
||||
if face_align is not None:
|
||||
face_swap = face_align.get_face_swap()
|
||||
face_align_mask = face_align.get_face_mask()
|
||||
for fsi in bcd.get_face_swap_info_list():
|
||||
face_align_img_shape, _ = bcd.get_image_shape_dtype(fsi.face_align_image_name)
|
||||
face_align_mask_img = bcd.get_image(fsi.face_align_mask_name)
|
||||
face_swap_img = bcd.get_image(fsi.face_swap_image_name)
|
||||
face_swap_mask_img = bcd.get_image(fsi.face_swap_mask_name)
|
||||
image_to_align_uni_mat = fsi.image_to_align_uni_mat
|
||||
|
||||
if face_swap is not None:
|
||||
face_swap_mask = face_swap.get_face_mask()
|
||||
if face_swap_mask is not None:
|
||||
if all_is_not_None(face_align_img_shape, face_align_mask_img, face_swap_img, face_swap_mask_img, image_to_align_uni_mat):
|
||||
face_height, face_width = face_align_img_shape[:2]
|
||||
frame_height, frame_width = frame_image.shape[:2]
|
||||
aligned_to_source_uni_mat = image_to_align_uni_mat.invert()
|
||||
aligned_to_source_uni_mat = aligned_to_source_uni_mat.source_translated(-state.face_x_offset, -state.face_y_offset)
|
||||
aligned_to_source_uni_mat = aligned_to_source_uni_mat.source_scaled_around_center(state.face_scale,state.face_scale)
|
||||
aligned_to_source_uni_mat = aligned_to_source_uni_mat.to_exact_mat (face_width, face_height, frame_width, frame_height)
|
||||
|
||||
face_align_img_shape, _ = bcd.get_image_shape_dtype(face_align.get_image_name())
|
||||
face_align_mask_img = bcd.get_image(face_align_mask.get_image_name())
|
||||
face_swap_img = bcd.get_image(face_swap.get_image_name())
|
||||
face_swap_mask_img = bcd.get_image(face_swap_mask.get_image_name())
|
||||
source_to_aligned_uni_mat = face_align.get_source_to_aligned_uni_mat()
|
||||
if state.device == 'CPU':
|
||||
merged_frame = self._merge_on_cpu(frame_image, face_align_mask_img, face_swap_img, face_swap_mask_img, aligned_to_source_uni_mat, frame_width, frame_height )
|
||||
else:
|
||||
merged_frame = self._merge_on_gpu(frame_image, face_align_mask_img, face_swap_img, face_swap_mask_img, aligned_to_source_uni_mat, frame_width, frame_height )
|
||||
# keep image in float32 in order not to extra load FaceMerger
|
||||
|
||||
if all_is_not_None(face_align_img_shape, face_align_mask_img, face_swap_img, face_swap_mask_img):
|
||||
face_height, face_width = face_align_img_shape[:2]
|
||||
frame_height, frame_width = frame_image.shape[:2]
|
||||
aligned_to_source_uni_mat = source_to_aligned_uni_mat.invert()
|
||||
aligned_to_source_uni_mat = aligned_to_source_uni_mat.source_translated(-state.face_x_offset, -state.face_y_offset)
|
||||
aligned_to_source_uni_mat = aligned_to_source_uni_mat.source_scaled_around_center(state.face_scale,state.face_scale)
|
||||
aligned_to_source_uni_mat = aligned_to_source_uni_mat.to_exact_mat (face_width, face_height, frame_width, frame_height)
|
||||
|
||||
if state.device == 'CPU':
|
||||
merged_frame = self._merge_on_cpu(frame_image, face_align_mask_img, face_swap_img, face_swap_mask_img, aligned_to_source_uni_mat, frame_width, frame_height )
|
||||
else:
|
||||
merged_frame = self._merge_on_gpu(frame_image, face_align_mask_img, face_swap_img, face_swap_mask_img, aligned_to_source_uni_mat, frame_width, frame_height )
|
||||
# keep image in float32 in order not to extra load FaceMerger
|
||||
|
||||
merged_frame_name = f'{frame_name}_merged'
|
||||
bcd.set_merged_frame_name(merged_frame_name)
|
||||
bcd.set_image(merged_frame_name, merged_frame)
|
||||
break
|
||||
merged_image_name = f'{frame_image_name}_merged'
|
||||
bcd.set_merged_image_name(merged_image_name)
|
||||
bcd.set_image(merged_image_name, merged_frame)
|
||||
break
|
||||
|
||||
self.stop_profile_timing()
|
||||
self.pending_bcd = bcd
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ from pathlib import Path
|
|||
import numpy as np
|
||||
from modelhub import DFLive
|
||||
from xlib import os as lib_os
|
||||
from xlib.facemeta import FaceMask, FaceSwap
|
||||
from xlib.image.ImageProcessor import ImageProcessor
|
||||
from xlib.mp import csw as lib_csw
|
||||
from xlib.python import all_is_not_None
|
||||
|
|
@ -171,7 +170,7 @@ class FaceSwapperWorker(BackendWorker):
|
|||
cs.model_dl_progress.enable()
|
||||
cs.model_dl_progress.set_config( lib_csw.Progress.Config(title='@FaceSwapper.downloading_model') )
|
||||
cs.model_dl_progress.set_progress(0)
|
||||
|
||||
|
||||
elif events.new_status_initialized:
|
||||
self.dfm_model = events.dfm_model
|
||||
self.dfm_model_initializer = None
|
||||
|
|
@ -239,47 +238,39 @@ class FaceSwapperWorker(BackendWorker):
|
|||
if all_is_not_None(dfm_model, model_state):
|
||||
face_id = model_state.face_id
|
||||
if face_id is not None:
|
||||
for i, face_mark in enumerate(bcd.get_face_mark_list()):
|
||||
if face_id == i:
|
||||
face_align = face_mark.get_face_align()
|
||||
if face_align is not None:
|
||||
face_align_image_name = face_align.get_image_name()
|
||||
face_align_image = bcd.get_image(face_align_image_name)
|
||||
if face_align_image is not None:
|
||||
for i, fsi in enumerate(bcd.get_face_swap_info_list()):
|
||||
if face_id != i:
|
||||
continue
|
||||
|
||||
pre_gamma_red = model_state.pre_gamma_red
|
||||
pre_gamma_green = model_state.pre_gamma_green
|
||||
pre_gamma_blue = model_state.pre_gamma_blue
|
||||
face_align_image = bcd.get_image(fsi.face_align_image_name)
|
||||
if face_align_image is not None:
|
||||
|
||||
fai_ip = ImageProcessor(face_align_image)
|
||||
if model_state.presharpen_amount != 0:
|
||||
fai_ip.sharpen(factor=model_state.presharpen_amount)
|
||||
pre_gamma_red = model_state.pre_gamma_red
|
||||
pre_gamma_green = model_state.pre_gamma_green
|
||||
pre_gamma_blue = model_state.pre_gamma_blue
|
||||
|
||||
if pre_gamma_red != 1.0 or pre_gamma_green != 1.0 or pre_gamma_blue != 1.0:
|
||||
fai_ip.adjust_gamma(pre_gamma_red, pre_gamma_green, pre_gamma_blue)
|
||||
face_align_image = fai_ip.get_image('HWC')
|
||||
fai_ip = ImageProcessor(face_align_image)
|
||||
if model_state.presharpen_amount != 0:
|
||||
fai_ip.sharpen(factor=model_state.presharpen_amount)
|
||||
|
||||
celeb_face, celeb_face_mask_img, face_align_mask_img = dfm_model.convert(face_align_image, morph_factor=model_state.morph_factor)
|
||||
celeb_face, celeb_face_mask_img, face_align_mask_img = celeb_face[0], celeb_face_mask_img[0], face_align_mask_img[0]
|
||||
if pre_gamma_red != 1.0 or pre_gamma_green != 1.0 or pre_gamma_blue != 1.0:
|
||||
fai_ip.adjust_gamma(pre_gamma_red, pre_gamma_green, pre_gamma_blue)
|
||||
face_align_image = fai_ip.get_image('HWC')
|
||||
|
||||
if model_state.two_pass:
|
||||
celeb_face, celeb_face_mask_img, _ = dfm_model.convert(celeb_face, morph_factor=model_state.morph_factor)
|
||||
celeb_face, celeb_face_mask_img = celeb_face[0], celeb_face_mask_img[0]
|
||||
celeb_face, celeb_face_mask_img, face_align_mask_img = dfm_model.convert(face_align_image, morph_factor=model_state.morph_factor)
|
||||
celeb_face, celeb_face_mask_img, face_align_mask_img = celeb_face[0], celeb_face_mask_img[0], face_align_mask_img[0]
|
||||
|
||||
face_align_mask = FaceMask()
|
||||
face_align_mask.set_image_name(f'{face_align_image_name}_mask')
|
||||
face_align.set_face_mask(face_align_mask)
|
||||
bcd.set_image(face_align_mask.get_image_name(), face_align_mask_img)
|
||||
if model_state.two_pass:
|
||||
celeb_face, celeb_face_mask_img, _ = dfm_model.convert(celeb_face, morph_factor=model_state.morph_factor)
|
||||
celeb_face, celeb_face_mask_img = celeb_face[0], celeb_face_mask_img[0]
|
||||
|
||||
face_swap = FaceSwap()
|
||||
face_swap.set_image_name (f"{face_align_image_name}_swapped")
|
||||
face_align.set_face_swap(face_swap)
|
||||
bcd.set_image(face_swap.get_image_name(), celeb_face)
|
||||
fsi.face_align_mask_name = f'{fsi.face_align_image_name}_mask'
|
||||
fsi.face_swap_image_name = f'{fsi.face_align_image_name}_swapped'
|
||||
fsi.face_swap_mask_name = f'{fsi.face_swap_image_name}_mask'
|
||||
|
||||
face_swap_mask = FaceMask()
|
||||
face_swap_mask.set_image_name(f'{face_swap.get_image_name()}_mask')
|
||||
face_swap.set_face_mask(face_swap_mask)
|
||||
bcd.set_image(face_swap_mask.get_image_name(), celeb_face_mask_img)
|
||||
bcd.set_image(fsi.face_align_mask_name, face_align_mask_img)
|
||||
bcd.set_image(fsi.face_swap_image_name, celeb_face)
|
||||
bcd.set_image(fsi.face_swap_mask_name, celeb_face_mask_img)
|
||||
|
||||
self.stop_profile_timing()
|
||||
self.pending_bcd = bcd
|
||||
|
|
|
|||
|
|
@ -287,7 +287,7 @@ class FileSourceWorker(BackendWorker):
|
|||
bcd.set_frame_num(p_frame.frame_num)
|
||||
bcd.set_frame_fps(p_frame.fps)
|
||||
bcd.set_frame_timestamp(p_frame.timestamp)
|
||||
bcd.set_frame_name(p_frame.name)
|
||||
bcd.set_frame_image_name(p_frame.name)
|
||||
|
||||
image = ImageProcessor(p_frame.image).to_uint8().get_image('HWC')
|
||||
bcd.set_image(p_frame.name, image)
|
||||
|
|
|
|||
|
|
@ -72,8 +72,8 @@ class FrameAdjusterWorker(BackendWorker):
|
|||
if bcd is not None:
|
||||
bcd.assign_weak_heap(self.weak_heap)
|
||||
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image_name = bcd.get_frame_image_name()
|
||||
frame_image = bcd.get_image(frame_image_name)
|
||||
|
||||
if frame_image is not None:
|
||||
frame_image_ip = ImageProcessor(frame_image)
|
||||
|
|
@ -81,7 +81,7 @@ class FrameAdjusterWorker(BackendWorker):
|
|||
frame_image_ip.degrade_resize( state.degrade_bicubic_per / 100.0, interpolation=ImageProcessor.Interpolation.CUBIC)
|
||||
|
||||
frame_image = frame_image_ip.get_image('HWC')
|
||||
bcd.set_image(frame_name, frame_image)
|
||||
bcd.set_image(frame_image_name, frame_image)
|
||||
|
||||
self.stop_profile_timing()
|
||||
self.pending_bcd = bcd
|
||||
|
|
|
|||
|
|
@ -202,31 +202,26 @@ class StreamOutputWorker(BackendWorker):
|
|||
view_image = None
|
||||
|
||||
if source_type == SourceType.SOURCE_FRAME:
|
||||
view_image = bcd.get_image(bcd.get_frame_name())
|
||||
view_image = bcd.get_image(bcd.get_frame_image_name())
|
||||
elif source_type == SourceType.MERGED_FRAME:
|
||||
view_image = bcd.get_image(bcd.get_merged_frame_name())
|
||||
view_image = bcd.get_image(bcd.get_merged_image_name())
|
||||
|
||||
elif source_type == SourceType.ALIGNED_FACE:
|
||||
aligned_face_id = state.aligned_face_id
|
||||
for i, face_mark in enumerate(bcd.get_face_mark_list()):
|
||||
for i, fsi in enumerate(bcd.get_face_swap_info_list()):
|
||||
if aligned_face_id == i:
|
||||
face_align = face_mark.get_face_align()
|
||||
if face_align is not None:
|
||||
view_image = bcd.get_image(face_align.get_image_name())
|
||||
view_image = bcd.get_image(fsi.face_align_image_name)
|
||||
break
|
||||
|
||||
elif source_type == SourceType.SWAPPED_FACE:
|
||||
for face_mark in bcd.get_face_mark_list():
|
||||
face_align = face_mark.get_face_align()
|
||||
if face_align is not None:
|
||||
face_swap = face_align.get_face_swap()
|
||||
if face_swap is not None:
|
||||
view_image = bcd.get_image(face_swap.get_image_name())
|
||||
break
|
||||
for fsi in bcd.get_face_swap_info_list():
|
||||
view_image = bcd.get_image(fsi.face_swap_image_name)
|
||||
if view_image is not None:
|
||||
break
|
||||
|
||||
elif source_type == SourceType.SOURCE_N_MERGED_FRAME:
|
||||
source_frame = bcd.get_image(bcd.get_frame_name())
|
||||
merged_frame = bcd.get_image(bcd.get_merged_frame_name())
|
||||
source_frame = bcd.get_image(bcd.get_frame_image_name())
|
||||
merged_frame = bcd.get_image(bcd.get_merged_image_name())
|
||||
if source_frame is not None and merged_frame is not None:
|
||||
source_frame = ImageProcessor(source_frame).to_ufloat32().get_image('HWC')
|
||||
view_image = np.concatenate( (source_frame, merged_frame), 1 )
|
||||
|
|
|
|||
|
|
@ -92,20 +92,18 @@ class QFaceDetector(QBackendPanel):
|
|||
if bcd is not None:
|
||||
bcd.assign_weak_heap(self._weak_heap)
|
||||
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image = bcd.get_image(bcd.get_frame_image_name())
|
||||
frame_image_w_h = None
|
||||
if frame_image is not None:
|
||||
h,w = frame_image.shape[0:2]
|
||||
frame_image_w_h = (w,h)
|
||||
|
||||
info = []
|
||||
for face_num,face_mark in enumerate(bcd.get_face_mark_list()):
|
||||
info_str = f'{face_num}: '
|
||||
|
||||
rect = face_mark.get_face_urect()
|
||||
if rect is not None:
|
||||
l,t,r,b = rect.as_ltrb_bbox(frame_image_w_h).astype(np.int)
|
||||
for face_id, fsi in enumerate(bcd.get_face_swap_info_list()):
|
||||
info_str = f'{face_id}: '
|
||||
|
||||
if fsi.face_urect is not None:
|
||||
l,t,r,b = fsi.face_urect.as_ltrb_bbox(frame_image_w_h).astype(np.int)
|
||||
info_str += f'[{l},{t},{r},{b}]'
|
||||
|
||||
info.append(info_str)
|
||||
|
|
|
|||
|
|
@ -5,8 +5,6 @@ from PyQt6.QtGui import *
|
|||
from PyQt6.QtWidgets import *
|
||||
from resources.fonts import QXFontDB
|
||||
from xlib import qt as lib_qt
|
||||
from xlib.facemeta import FaceULandmarks
|
||||
from xlib.python import all_is_not_None
|
||||
|
||||
from ... import backend
|
||||
|
||||
|
|
@ -47,36 +45,25 @@ class QBCFaceAlignViewer(lib_qt.QXCollapsibleSection):
|
|||
|
||||
self._layered_images.clear_images()
|
||||
|
||||
for face_mark in bcd.get_face_mark_list():
|
||||
face_align = face_mark.get_face_align()
|
||||
if face_align is not None:
|
||||
face_image = bcd.get_image (face_align.get_image_name())
|
||||
if face_image is not None:
|
||||
source_to_aligned_uni_mat = face_align.get_source_to_aligned_uni_mat()
|
||||
for fsi in bcd.get_face_swap_info_list():
|
||||
face_image = bcd.get_image (fsi.face_align_image_name)
|
||||
if face_image is not None:
|
||||
h,w = face_image.shape[:2]
|
||||
self._layered_images.add_image(face_image)
|
||||
|
||||
h,w = face_image.shape[:2]
|
||||
self._layered_images.add_image(face_image)
|
||||
if fsi.face_align_ulmrks is not None:
|
||||
lmrks_layer = np.zeros( (self._preview_width, self._preview_width, 4), dtype=np.uint8)
|
||||
|
||||
face_ulmrks = face_align.get_face_ulandmarks_by_type(FaceULandmarks.Type.LANDMARKS_468)
|
||||
if face_ulmrks is None:
|
||||
face_ulmrks = face_align.get_face_ulandmarks_by_type(FaceULandmarks.Type.LANDMARKS_68)
|
||||
fsi.face_align_ulmrks.draw(lmrks_layer, (0,255,0,255))
|
||||
|
||||
if face_ulmrks is not None:
|
||||
lmrks_layer = np.zeros( (self._preview_width, self._preview_width, 4), dtype=np.uint8)
|
||||
if fsi.face_urect is not None and fsi.image_to_align_uni_mat is not None:
|
||||
aligned_uni_rect = fsi.face_urect.transform(fsi.image_to_align_uni_mat)
|
||||
aligned_uni_rect.draw(lmrks_layer, (0,0,255,255) )
|
||||
|
||||
face_ulmrks.draw(lmrks_layer, (0,255,0,255))
|
||||
self._layered_images.add_image(lmrks_layer)
|
||||
|
||||
face_mark_rect = face_mark.get_face_urect()
|
||||
if face_mark_rect is not None:
|
||||
aligned_uni_rect = face_mark_rect.transform(source_to_aligned_uni_mat)
|
||||
aligned_uni_rect.draw(lmrks_layer, (0,0,255,255) )
|
||||
|
||||
self._layered_images.add_image(lmrks_layer)
|
||||
|
||||
|
||||
self._info_label.setText(f'{w}x{h}')
|
||||
|
||||
return
|
||||
self._info_label.setText(f'{w}x{h}')
|
||||
return
|
||||
|
||||
|
||||
def clear(self):
|
||||
|
|
|
|||
|
|
@ -45,17 +45,13 @@ class QBCFaceSwapViewer(lib_qt.QXCollapsibleSection):
|
|||
|
||||
self._layered_images.clear_images()
|
||||
|
||||
for face_mark in bcd.get_face_mark_list():
|
||||
face_align = face_mark.get_face_align()
|
||||
if face_align is not None:
|
||||
face_swap = face_align.get_face_swap()
|
||||
if face_swap is not None:
|
||||
face_swap_image = bcd.get_image(face_swap.get_image_name())
|
||||
if face_swap_image is not None:
|
||||
self._layered_images.add_image(face_swap_image)
|
||||
h,w = face_swap_image.shape[0:2]
|
||||
self._info_label.setText(f'{w}x{h}')
|
||||
return
|
||||
for fsi in bcd.get_face_swap_info_list():
|
||||
face_swap_image = bcd.get_image(fsi.face_swap_image_name)
|
||||
if face_swap_image is not None:
|
||||
self._layered_images.add_image(face_swap_image)
|
||||
h,w = face_swap_image.shape[0:2]
|
||||
self._info_label.setText(f'{w}x{h}')
|
||||
return
|
||||
|
||||
def clear(self):
|
||||
self._layered_images.clear_images()
|
||||
|
|
|
|||
|
|
@ -43,16 +43,16 @@ class QBCFinalFrameViewer(lib_qt.QXCollapsibleSection):
|
|||
|
||||
self._layered_images.clear_images()
|
||||
|
||||
merged_frame_name = bcd.get_merged_frame_name()
|
||||
merged_frame_image = bcd.get_image(merged_frame_name)
|
||||
merged_image_name = bcd.get_merged_image_name()
|
||||
merged_image = bcd.get_image(merged_image_name)
|
||||
|
||||
if merged_frame_image is not None:
|
||||
if merged_frame_image.dtype != np.uint8:
|
||||
merged_frame_image = lib_image.ImageProcessor(merged_frame_image).to_uint8().get_image('HWC')
|
||||
if merged_image is not None:
|
||||
if merged_image.dtype != np.uint8:
|
||||
merged_image = lib_image.ImageProcessor(merged_image).to_uint8().get_image('HWC')
|
||||
|
||||
self._layered_images.add_image(merged_frame_image)
|
||||
h,w = merged_frame_image.shape[0:2]
|
||||
self._info_label.setText(f'{merged_frame_name} {w}x{h}')
|
||||
self._layered_images.add_image(merged_image)
|
||||
h,w = merged_image.shape[0:2]
|
||||
self._info_label.setText(f'{merged_image_name} {w}x{h}')
|
||||
|
||||
|
||||
def clear(self):
|
||||
|
|
|
|||
|
|
@ -42,16 +42,13 @@ class QBCFrameViewer(lib_qt.QXCollapsibleSection):
|
|||
|
||||
self._layered_images.clear_images()
|
||||
|
||||
frame_name = bcd.get_frame_name()
|
||||
frame_image = bcd.get_image(frame_name)
|
||||
frame_image_name = bcd.get_frame_image_name()
|
||||
frame_image = bcd.get_image(frame_image_name)
|
||||
|
||||
if frame_image is not None:
|
||||
self._layered_images.add_image (frame_image)
|
||||
h,w = frame_image.shape[:2]
|
||||
if frame_name is not None:
|
||||
self._info_label.setText(f'{frame_name} {w}x{h}')
|
||||
else:
|
||||
self._info_label.setText(f'{w}x{h}')
|
||||
self._info_label.setText(f'{frame_image_name} {w}x{h}')
|
||||
|
||||
|
||||
def clear(self):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue