mirror of
https://github.com/facefusion/facefusion.git
synced 2026-08-23 17:22:30 +02:00
Next (#384)
* feat/yoloface (#334)
* added yolov8 to face_detector (#323)
* added yolov8 to face_detector
* added yolov8 to face_detector
* Initial cleanup and renaming
* Update README
* refactored detect_with_yoloface (#329)
* refactored detect_with_yoloface
* apply review
* Change order again
* Restore working code
* modified code (#330)
* refactored detect_with_yoloface
* apply review
* use temp_frame in detect_with_yoloface
* reorder
* modified
* reorder models
* Tiny cleanup
---------
Co-authored-by: tamoharu <133945583+tamoharu@users.noreply.github.com>
* include audio file functions (#336)
* Add testing for audio handlers
* Change order
* Fix naming
* Use correct typing in choices
* Update help message for arguments, Notation based wording approach (#347)
* Update help message for arguments, Notation based wording approach
* Fix installer
* Audio functions (#345)
* Update ffmpeg.py
* Create audio.py
* Update ffmpeg.py
* Update audio.py
* Update audio.py
* Update typing.py
* Update ffmpeg.py
* Update audio.py
* Rename Frame to VisionFrame (#346)
* Minor tidy up
* Introduce audio testing
* Add more todo for testing
* Add more todo for testing
* Fix indent
* Enable venv on the fly
* Enable venv on the fly
* Revert venv on the fly
* Revert venv on the fly
* Force Gradio to shut up
* Force Gradio to shut up
* Clear temp before processing
* Reduce terminal output
* include audio file functions
* Enforce output resolution on merge video
* Minor cleanups
* Add age and gender to face debugger items (#353)
* Add age and gender to face debugger items
* Rename like suggested in the code review
* Fix the output framerate vs. time
* Lip Sync (#356)
* Cli implementation of wav2lip
* - create get_first_item()
- remove non gan wav2lip model
- implement video memory strategy
- implement get_reference_frame()
- implement process_image()
- rearrange crop_mask_list
- implement test_cli
* Simplify testing
* Rename to lip syncer
* Fix testing
* Fix testing
* Minor cleanup
* Cuda 12 installer (#362)
* Make cuda nightly (12) the default
* Better keep legacy cuda just in case
* Use CUDA and ROCM versions
* Remove MacOS options from installer (CoreML include in default package)
* Add lip-syncer support to source component
* Add lip-syncer support to source component
* Fix the check in the source component
* Add target image check
* Introduce more helpers to suite the lip-syncer needs
* Downgrade onnxruntime as of buggy 1.17.0 release
* Revert "Downgrade onnxruntime as of buggy 1.17.0 release"
This reverts commit f4a7ae6824.
* More testing and add todos
* Fix the frame processor API to at least not throw errors
* Introduce dict based frame processor inputs (#364)
* Introduce dict based frame processor inputs
* Forgot to adjust webcam
* create path payloads (#365)
* create index payload to paths for process_frames
* rename to payload_paths
* This code now is poetry
* Fix the terminal output
* Make lip-syncer work in the preview
* Remove face debugger test for now
* Reoder reference_faces, Fix testing
* Use inswapper_128 on buggy onnxruntime 1.17.0
* Undo inswapper_128_fp16 duo broken onnxruntime 1.17.0
* Undo inswapper_128_fp16 duo broken onnxruntime 1.17.0
* Fix lip_syncer occluder & region mask issue
* Fix preview once in case there was no output video fps
* fix lip_syncer custom fps
* remove unused import
* Add 68 landmark functions (#367)
* Add 68 landmark model
* Add landmark to face object
* Re-arrange and modify typing
* Rename function
* Rearrange
* Rearrange
* ignore type
* ignore type
* change type
* ignore
* name
* Some cleanup
* Some cleanup
* Opps, I broke something
* Feat/face analyser refactoring (#369)
* Restructure face analyser and start TDD
* YoloFace and Yunet testing are passing
* Remove offset from yoloface detection
* Cleanup code
* Tiny fix
* Fix get_many_faces()
* Tiny fix (again)
* Use 320x320 fallback for retinaface
* Fix merging mashup
* Upload wave2lip model
* Upload 2dfan2 model and rename internal to face_predictor
* Downgrade onnxruntime for most cases
* Update for the face debugger to render landmark 68
* Try to make detect_face_landmark_68() and detect_gender_age() more uniform
* Enable retinaface testing for 320x320
* Make detect_face_landmark_68() and detect_gender_age() as uniform as … (#370)
* Make detect_face_landmark_68() and detect_gender_age() as uniform as possible
* Revert landmark scale and translation
* Make box-mask for lip-syncer adjustable
* Add create_bbox_from_landmark()
* Remove currently unused code
* Feat/uniface (#375)
* add uniface (#373)
* Finalize UniFace implementation
---------
Co-authored-by: Harisreedhar <46858047+harisreedhar@users.noreply.github.com>
* My approach how todo it
* edit
* edit
* replace vertical blur with gaussian
* remove region mask
* Rebase against next and restore method
* Minor improvements
* Minor improvements
* rename & add forehead padding
* Adjust and host uniface model
* Use 2dfan4 model
* Rename to face landmarker
* Feat/replace bbox with bounding box (#380)
* Add landmark 68 to 5 convertion
* Add landmark 68 to 5 convertion
* Keep 5, 5/68 and 68 landmarks
* Replace kps with landmark
* Replace bbox with bounding box
* Reshape face_landmark5_list different
* Make yoloface the default
* Move convert_face_landmark_68_to_5 to face_helper
* Minor spacing issue
* Dynamic detector sizes according to model (#382)
* Dynamic detector sizes according to model
* Dynamic detector sizes according to model
* Undo false commited files
* Add lib syncer model to the UI
* fix halo (#383)
* Bump to 2.3.0
* Update README and wording
* Update README and wording
* Fix spacing
* Apply _vision suffix
* Apply _vision suffix
* Apply _vision suffix
* Apply _vision suffix
* Apply _vision suffix
* Apply _vision suffix
* Apply _vision suffix, Move mouth mask to face_masker.py
* Apply _vision suffix
* Apply _vision suffix
* increase forehead padding
---------
Co-authored-by: tamoharu <133945583+tamoharu@users.noreply.github.com>
Co-authored-by: Harisreedhar <46858047+harisreedhar@users.noreply.github.com>
This commit is contained in:
Henry Ruhs
co-authored by
Harisreedhar
tamoharu
Harisreedhar
tamoharu
parent
122da0545b
commit
c77493ff9a
@@ -1,13 +1,13 @@
|
||||
from typing import List
|
||||
|
||||
from facefusion.common_helper import create_int_range
|
||||
from facefusion.processors.frame.typings import FaceSwapperModel, FaceEnhancerModel, FrameEnhancerModel, FaceDebuggerItem
|
||||
from facefusion.processors.frame.typings import FaceDebuggerItem, FaceEnhancerModel, FaceSwapperModel, FrameEnhancerModel, LipSyncerModel
|
||||
|
||||
face_swapper_models : List[FaceSwapperModel] = [ 'blendswap_256', 'inswapper_128', 'inswapper_128_fp16', 'simswap_256', 'simswap_512_unofficial' ]
|
||||
face_debugger_items : List[FaceDebuggerItem] = [ 'bounding-box', 'landmark-5', 'landmark-68', 'face-mask', 'score', 'age', 'gender' ]
|
||||
face_enhancer_models : List[FaceEnhancerModel] = [ 'codeformer', 'gfpgan_1.2', 'gfpgan_1.3', 'gfpgan_1.4', 'gpen_bfr_256', 'gpen_bfr_512', 'restoreformer_plus_plus' ]
|
||||
face_swapper_models : List[FaceSwapperModel] = [ 'blendswap_256', 'inswapper_128', 'inswapper_128_fp16', 'simswap_256', 'simswap_512_unofficial', 'uniface_256' ]
|
||||
frame_enhancer_models : List[FrameEnhancerModel] = [ 'real_esrgan_x2plus', 'real_esrgan_x4plus', 'real_esrnet_x4plus' ]
|
||||
face_debugger_items : List[FaceDebuggerItem] = [ 'bbox', 'kps', 'face-mask', 'score' ]
|
||||
lip_syncer_models : List[LipSyncerModel] = [ 'wav2lip_gan' ]
|
||||
|
||||
face_enhancer_blend_range : List[int] = create_int_range(0, 100, 1)
|
||||
frame_enhancer_blend_range : List[int] = create_int_range(0, 100, 1)
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import os
|
||||
import sys
|
||||
import importlib
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
@@ -7,7 +8,7 @@ from typing import Any, List
|
||||
from tqdm import tqdm
|
||||
|
||||
import facefusion.globals
|
||||
from facefusion.typing import Process_Frames
|
||||
from facefusion.typing import Process_Frames, QueuePayload
|
||||
from facefusion.execution_helper import encode_execution_providers
|
||||
from facefusion import logger, wording
|
||||
|
||||
@@ -67,7 +68,8 @@ def clear_frame_processors_modules() -> None:
|
||||
|
||||
|
||||
def multi_process_frames(source_paths : List[str], temp_frame_paths : List[str], process_frames : Process_Frames) -> None:
|
||||
with tqdm(total = len(temp_frame_paths), desc = wording.get('processing'), unit = 'frame', ascii = ' =', disable = facefusion.globals.log_level in [ 'warn', 'error' ]) as progress:
|
||||
queue_payloads = create_queue_payloads(temp_frame_paths)
|
||||
with tqdm(total = len(queue_payloads), desc = wording.get('processing'), unit = 'frame', ascii = ' =', disable = facefusion.globals.log_level in [ 'warn', 'error' ]) as progress:
|
||||
progress.set_postfix(
|
||||
{
|
||||
'execution_providers': encode_execution_providers(facefusion.globals.execution_providers),
|
||||
@@ -76,26 +78,39 @@ def multi_process_frames(source_paths : List[str], temp_frame_paths : List[str],
|
||||
})
|
||||
with ThreadPoolExecutor(max_workers = facefusion.globals.execution_thread_count) as executor:
|
||||
futures = []
|
||||
queue_frame_paths : Queue[str] = create_queue(temp_frame_paths)
|
||||
queue_per_future = max(len(temp_frame_paths) // facefusion.globals.execution_thread_count * facefusion.globals.execution_queue_count, 1)
|
||||
while not queue_frame_paths.empty():
|
||||
submit_frame_paths = pick_queue(queue_frame_paths, queue_per_future)
|
||||
future = executor.submit(process_frames, source_paths, submit_frame_paths, progress.update)
|
||||
queue : Queue[QueuePayload] = create_queue(queue_payloads)
|
||||
queue_per_future = max(len(queue_payloads) // facefusion.globals.execution_thread_count * facefusion.globals.execution_queue_count, 1)
|
||||
while not queue.empty():
|
||||
future = executor.submit(process_frames, source_paths, pick_queue(queue, queue_per_future), progress.update)
|
||||
futures.append(future)
|
||||
for future_done in as_completed(futures):
|
||||
future_done.result()
|
||||
|
||||
|
||||
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)
|
||||
def create_queue(queue_payloads : List[QueuePayload]) -> Queue[QueuePayload]:
|
||||
queue : Queue[QueuePayload] = Queue()
|
||||
for queue_payload in queue_payloads:
|
||||
queue.put(queue_payload)
|
||||
return queue
|
||||
|
||||
|
||||
def pick_queue(queue : Queue[str], queue_per_future : int) -> List[str]:
|
||||
def pick_queue(queue : Queue[QueuePayload], queue_per_future : int) -> List[QueuePayload]:
|
||||
queues = []
|
||||
for _ in range(queue_per_future):
|
||||
if not queue.empty():
|
||||
queues.append(queue.get())
|
||||
return queues
|
||||
|
||||
|
||||
def create_queue_payloads(temp_frame_paths : List[str]) -> List[QueuePayload]:
|
||||
queue_payloads = []
|
||||
temp_frame_paths = sorted(temp_frame_paths, key = os.path.basename)
|
||||
|
||||
for frame_number, frame_path in enumerate(temp_frame_paths):
|
||||
frame_payload : QueuePayload =\
|
||||
{
|
||||
'frame_number' : frame_number,
|
||||
'frame_path' : frame_path
|
||||
}
|
||||
queue_payloads.append(frame_payload)
|
||||
return queue_payloads
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
from typing import List, Optional
|
||||
|
||||
from facefusion.processors.frame.typings import FaceSwapperModel, FaceEnhancerModel, FrameEnhancerModel, FaceDebuggerItem
|
||||
from facefusion.processors.frame.typings import FaceDebuggerItem, FaceEnhancerModel, FaceSwapperModel, FrameEnhancerModel, LipSyncerModel
|
||||
|
||||
face_swapper_model : Optional[FaceSwapperModel] = None
|
||||
face_debugger_items : Optional[List[FaceDebuggerItem]] = None
|
||||
face_enhancer_model : Optional[FaceEnhancerModel] = None
|
||||
face_enhancer_blend : Optional[int] = None
|
||||
face_swapper_model : Optional[FaceSwapperModel] = None
|
||||
frame_enhancer_model : Optional[FrameEnhancerModel] = None
|
||||
frame_enhancer_blend : Optional[int] = None
|
||||
face_debugger_items : Optional[List[FaceDebuggerItem]] = None
|
||||
lip_syncer_model : Optional[LipSyncerModel] = None
|
||||
|
||||
@@ -6,13 +6,14 @@ import numpy
|
||||
import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import config, wording
|
||||
from facefusion.face_analyser import get_one_face, get_average_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion.face_analyser import get_one_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, create_region_mask, clear_face_occluder, clear_face_parser
|
||||
from facefusion.face_helper import warp_face_by_face_landmark_5, categorize_age, categorize_gender
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, FaceSet, Frame, Update_Process, ProcessMode
|
||||
from facefusion.vision import read_image, read_static_image, read_static_images, write_image
|
||||
from facefusion.face_helper import warp_face_by_kps
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, create_region_mask, clear_face_occluder, clear_face_parser
|
||||
from facefusion.typing import Face, VisionFrame, Update_Process, ProcessMode, QueuePayload
|
||||
from facefusion.vision import read_image, read_static_image, write_image
|
||||
from facefusion.processors.frame.typings import FaceDebuggerInputs
|
||||
from facefusion.processors.frame import globals as frame_processors_globals, choices as frame_processors_choices
|
||||
|
||||
NAME = __name__.upper()
|
||||
@@ -35,7 +36,7 @@ def set_options(key : Literal['model'], value : Any) -> None:
|
||||
|
||||
|
||||
def register_args(program : ArgumentParser) -> None:
|
||||
program.add_argument('--face-debugger-items', help = wording.get('face_debugger_items_help').format(choices = ', '.join(frame_processors_choices.face_debugger_items)), default = config.get_str_list('frame_processors.face_debugger_items', 'kps face-mask'), choices = frame_processors_choices.face_debugger_items, nargs = '+', metavar = 'FACE_DEBUGGER_ITEMS')
|
||||
program.add_argument('--face-debugger-items', help = wording.get('help.face_debugger_items').format(choices = ', '.join(frame_processors_choices.face_debugger_items)), default = config.get_str_list('frame_processors.face_debugger_items', 'landmark-5 face-mask'), choices = frame_processors_choices.face_debugger_items, nargs = '+', metavar = 'FACE_DEBUGGER_ITEMS')
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
@@ -66,82 +67,109 @@ def post_process() -> None:
|
||||
clear_face_parser()
|
||||
|
||||
|
||||
def debug_face(source_face : Face, target_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
def debug_face(target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
primary_color = (0, 0, 255)
|
||||
secondary_color = (0, 255, 0)
|
||||
bounding_box = target_face.bbox.astype(numpy.int32)
|
||||
temp_frame = temp_frame.copy()
|
||||
if 'bbox' in frame_processors_globals.face_debugger_items:
|
||||
cv2.rectangle(temp_frame, (bounding_box[0], bounding_box[1]), (bounding_box[2], bounding_box[3]), secondary_color, 2)
|
||||
bounding_box = target_face.bounding_box.astype(numpy.int32)
|
||||
temp_vision_frame = temp_vision_frame.copy()
|
||||
|
||||
if 'bounding-box' in frame_processors_globals.face_debugger_items:
|
||||
cv2.rectangle(temp_vision_frame, (bounding_box[0], bounding_box[1]), (bounding_box[2], bounding_box[3]), secondary_color, 2)
|
||||
if 'face-mask' in frame_processors_globals.face_debugger_items:
|
||||
crop_frame, affine_matrix = warp_face_by_kps(temp_frame, target_face.kps, 'arcface_128_v2', (512, 512))
|
||||
crop_vision_frame, affine_matrix = warp_face_by_face_landmark_5(temp_vision_frame, target_face.landmark['5/68'], 'arcface_128_v2', (512, 512))
|
||||
inverse_matrix = cv2.invertAffineTransform(affine_matrix)
|
||||
temp_frame_size = temp_frame.shape[:2][::-1]
|
||||
temp_size = temp_vision_frame.shape[:2][::-1]
|
||||
crop_mask_list = []
|
||||
if 'box' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_static_box_mask(crop_frame.shape[:2][::-1], 0, facefusion.globals.face_mask_padding))
|
||||
box_mask = create_static_box_mask(crop_vision_frame.shape[:2][::-1], 0, facefusion.globals.face_mask_padding)
|
||||
crop_mask_list.append(box_mask)
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_occlusion_mask(crop_frame))
|
||||
occlusion_mask = create_occlusion_mask(crop_vision_frame)
|
||||
crop_mask_list.append(occlusion_mask)
|
||||
if 'region' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_region_mask(crop_frame, facefusion.globals.face_mask_regions))
|
||||
region_mask = create_region_mask(crop_vision_frame, facefusion.globals.face_mask_regions)
|
||||
crop_mask_list.append(region_mask)
|
||||
crop_mask = numpy.minimum.reduce(crop_mask_list).clip(0, 1)
|
||||
crop_mask = (crop_mask * 255).astype(numpy.uint8)
|
||||
inverse_mask_frame = cv2.warpAffine(crop_mask, inverse_matrix, temp_frame_size)
|
||||
inverse_mask_frame = cv2.threshold(inverse_mask_frame, 100, 255, cv2.THRESH_BINARY)[1]
|
||||
inverse_mask_frame[inverse_mask_frame > 0] = 255
|
||||
inverse_mask_contours = cv2.findContours(inverse_mask_frame, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)[0]
|
||||
cv2.drawContours(temp_frame, inverse_mask_contours, -1, primary_color, 2)
|
||||
inverse_vision_frame = cv2.warpAffine(crop_mask, inverse_matrix, temp_size)
|
||||
inverse_vision_frame = cv2.threshold(inverse_vision_frame, 100, 255, cv2.THRESH_BINARY)[1]
|
||||
inverse_vision_frame[inverse_vision_frame > 0] = 255
|
||||
inverse_contours = cv2.findContours(inverse_vision_frame, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)[0]
|
||||
cv2.drawContours(temp_vision_frame, inverse_contours, -1, primary_color, 2)
|
||||
if bounding_box[3] - bounding_box[1] > 60 and bounding_box[2] - bounding_box[0] > 60:
|
||||
if 'kps' in frame_processors_globals.face_debugger_items:
|
||||
kps = target_face.kps.astype(numpy.int32)
|
||||
for index in range(kps.shape[0]):
|
||||
cv2.circle(temp_frame, (kps[index][0], kps[index][1]), 3, primary_color, -1)
|
||||
top = bounding_box[1]
|
||||
left = bounding_box[0] + 20
|
||||
if 'landmark-5' in frame_processors_globals.face_debugger_items:
|
||||
face_landmark_5 = target_face.landmark['5/68'].astype(numpy.int32)
|
||||
for index in range(face_landmark_5.shape[0]):
|
||||
cv2.circle(temp_vision_frame, (face_landmark_5[index][0], face_landmark_5[index][1]), 3, primary_color, -1)
|
||||
if 'landmark-68' in frame_processors_globals.face_debugger_items:
|
||||
face_landmark_68 = target_face.landmark['68'].astype(numpy.int32)
|
||||
for index in range(face_landmark_68.shape[0]):
|
||||
cv2.circle(temp_vision_frame, (face_landmark_68[index][0], face_landmark_68[index][1]), 3, secondary_color, -1)
|
||||
if 'score' in frame_processors_globals.face_debugger_items:
|
||||
face_score_text = str(round(target_face.score, 2))
|
||||
face_score_position = (bounding_box[0] + 10, bounding_box[1] + 20)
|
||||
cv2.putText(temp_frame, face_score_text, face_score_position, cv2.FONT_HERSHEY_SIMPLEX, 0.5, secondary_color, 2)
|
||||
return temp_frame
|
||||
top = top + 20
|
||||
cv2.putText(temp_vision_frame, face_score_text, (left, top), cv2.FONT_HERSHEY_SIMPLEX, 0.5, secondary_color, 2)
|
||||
if 'age' in frame_processors_globals.face_debugger_items:
|
||||
face_age_text = categorize_age(target_face.age)
|
||||
top = top + 20
|
||||
cv2.putText(temp_vision_frame, face_age_text, (left, top), cv2.FONT_HERSHEY_SIMPLEX, 0.5, secondary_color, 2)
|
||||
if 'gender' in frame_processors_globals.face_debugger_items:
|
||||
face_gender_text = categorize_gender(target_face.gender)
|
||||
top = top + 20
|
||||
cv2.putText(temp_vision_frame, face_gender_text, (left, top), cv2.FONT_HERSHEY_SIMPLEX, 0.5, secondary_color, 2)
|
||||
return temp_vision_frame
|
||||
|
||||
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
pass
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
def process_frame(inputs : FaceDebuggerInputs) -> VisionFrame:
|
||||
reference_faces = inputs['reference_faces']
|
||||
target_vision_frame = inputs['target_vision_frame']
|
||||
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(temp_frame, reference_faces, facefusion.globals.reference_face_distance)
|
||||
similar_faces = find_similar_faces(reference_faces, target_vision_frame, facefusion.globals.reference_face_distance)
|
||||
if similar_faces:
|
||||
for similar_face in similar_faces:
|
||||
temp_frame = debug_face(source_face, similar_face, reference_faces, temp_frame)
|
||||
target_vision_frame = debug_face(similar_face, target_vision_frame)
|
||||
if 'one' in facefusion.globals.face_selector_mode:
|
||||
target_face = get_one_face(temp_frame)
|
||||
target_face = get_one_face(target_vision_frame)
|
||||
if target_face:
|
||||
temp_frame = debug_face(source_face, target_face, None, temp_frame)
|
||||
target_vision_frame = debug_face(target_face, target_vision_frame)
|
||||
if 'many' in facefusion.globals.face_selector_mode:
|
||||
many_faces = get_many_faces(temp_frame)
|
||||
many_faces = get_many_faces(target_vision_frame)
|
||||
if many_faces:
|
||||
for target_face in many_faces:
|
||||
temp_frame = debug_face(source_face, target_face, None, temp_frame)
|
||||
return temp_frame
|
||||
target_vision_frame = debug_face(target_face, target_vision_frame)
|
||||
return target_vision_frame
|
||||
|
||||
|
||||
def process_frames(source_paths : List[str], temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
source_frames = read_static_images(source_paths)
|
||||
source_face = get_average_face(source_frames)
|
||||
def process_frames(source_paths : List[str], queue_payloads : List[QueuePayload], update_progress : Update_Process) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
for temp_frame_path in temp_frame_paths:
|
||||
temp_frame = read_image(temp_frame_path)
|
||||
result_frame = process_frame(source_face, reference_faces, temp_frame)
|
||||
write_image(temp_frame_path, result_frame)
|
||||
|
||||
for queue_payload in queue_payloads:
|
||||
target_vision_path = queue_payload['frame_path']
|
||||
target_vision_frame = read_image(target_vision_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(target_vision_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_paths : List[str], target_path : str, output_path : str) -> None:
|
||||
source_frames = read_static_images(source_paths)
|
||||
source_face = get_average_face(source_frames)
|
||||
target_frame = read_static_image(target_path)
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
result_frame = process_frame(source_face, reference_faces, target_frame)
|
||||
target_vision_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
|
||||
@@ -9,18 +9,19 @@ import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import config, logger, wording
|
||||
from facefusion.face_analyser import get_many_faces, clear_face_analyser, find_similar_faces, get_one_face
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, clear_face_occluder
|
||||
from facefusion.face_helper import warp_face_by_face_landmark_5, paste_back
|
||||
from facefusion.execution_helper import apply_execution_provider_options
|
||||
from facefusion.face_helper import warp_face_by_kps, paste_back
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.typing import Face, FaceSet, Frame, Update_Process, ProcessMode, ModelSet, OptionsWithModel
|
||||
from facefusion.typing import Face, VisionFrame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, QueuePayload
|
||||
from facefusion.common_helper import create_metavar
|
||||
from facefusion.filesystem import is_file, is_image, is_video, resolve_relative_path
|
||||
from facefusion.download import conditional_download, is_download_done
|
||||
from facefusion.vision import read_image, read_static_image, write_image
|
||||
from facefusion.processors.frame.typings import FaceEnhancerInputs
|
||||
from facefusion.processors.frame import globals as frame_processors_globals
|
||||
from facefusion.processors.frame import choices as frame_processors_choices
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, clear_face_occluder
|
||||
|
||||
FRAME_PROCESSOR = None
|
||||
THREAD_SEMAPHORE : threading.Semaphore = threading.Semaphore()
|
||||
@@ -115,8 +116,8 @@ def set_options(key : Literal['model'], value : Any) -> None:
|
||||
|
||||
|
||||
def register_args(program : ArgumentParser) -> None:
|
||||
program.add_argument('--face-enhancer-model', help = wording.get('frame_processor_model_help'), default = config.get_str_value('frame_processors.face_enhancer_model', 'gfpgan_1.4'), choices = frame_processors_choices.face_enhancer_models)
|
||||
program.add_argument('--face-enhancer-blend', help = wording.get('frame_processor_blend_help'), type = int, default = config.get_int_value('frame_processors.face_enhancer_blend', '80'), choices = frame_processors_choices.face_enhancer_blend_range, metavar = create_metavar(frame_processors_choices.face_enhancer_blend_range))
|
||||
program.add_argument('--face-enhancer-model', help = wording.get('help.face_enhancer_model'), default = config.get_str_value('frame_processors.face_enhancer_model', 'gfpgan_1.4'), choices = frame_processors_choices.face_enhancer_models)
|
||||
program.add_argument('--face-enhancer-blend', help = wording.get('help.face_enhancer_blend'), type = int, default = config.get_int_value('frame_processors.face_enhancer_blend', '80'), choices = frame_processors_choices.face_enhancer_blend_range, metavar = create_metavar(frame_processors_choices.face_enhancer_blend_range))
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
@@ -165,97 +166,113 @@ def post_process() -> None:
|
||||
clear_face_occluder()
|
||||
|
||||
|
||||
def enhance_face(target_face: Face, temp_frame : Frame) -> Frame:
|
||||
def enhance_face(target_face: Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
model_template = get_options('model').get('template')
|
||||
model_size = get_options('model').get('size')
|
||||
crop_frame, affine_matrix = warp_face_by_kps(temp_frame, target_face.kps, model_template, model_size)
|
||||
crop_vision_frame, affine_matrix = warp_face_by_face_landmark_5(temp_vision_frame, target_face.landmark['5/68'], model_template, model_size)
|
||||
box_mask = create_static_box_mask(crop_vision_frame.shape[:2][::-1], facefusion.globals.face_mask_blur, (0, 0, 0, 0))
|
||||
crop_mask_list =\
|
||||
[
|
||||
create_static_box_mask(crop_frame.shape[:2][::-1], facefusion.globals.face_mask_blur, (0, 0, 0, 0))
|
||||
box_mask
|
||||
]
|
||||
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_occlusion_mask(crop_frame))
|
||||
crop_frame = prepare_crop_frame(crop_frame)
|
||||
crop_frame = apply_enhance(crop_frame)
|
||||
crop_frame = normalize_crop_frame(crop_frame)
|
||||
occlusion_mask = create_occlusion_mask(crop_vision_frame)
|
||||
crop_mask_list.append(occlusion_mask)
|
||||
crop_vision_frame = prepare_crop_frame(crop_vision_frame)
|
||||
crop_vision_frame = apply_enhance(crop_vision_frame)
|
||||
crop_vision_frame = normalize_crop_frame(crop_vision_frame)
|
||||
crop_mask = numpy.minimum.reduce(crop_mask_list).clip(0, 1)
|
||||
paste_frame = paste_back(temp_frame, crop_frame, crop_mask, affine_matrix)
|
||||
temp_frame = blend_frame(temp_frame, paste_frame)
|
||||
return temp_frame
|
||||
paste_vision_frame = paste_back(temp_vision_frame, crop_vision_frame, crop_mask, affine_matrix)
|
||||
temp_vision_frame = blend_frame(temp_vision_frame, paste_vision_frame)
|
||||
return temp_vision_frame
|
||||
|
||||
|
||||
def apply_enhance(crop_frame : Frame) -> Frame:
|
||||
def apply_enhance(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
frame_processor = get_frame_processor()
|
||||
frame_processor_inputs = {}
|
||||
|
||||
for frame_processor_input in frame_processor.get_inputs():
|
||||
if frame_processor_input.name == 'input':
|
||||
frame_processor_inputs[frame_processor_input.name] = crop_frame
|
||||
frame_processor_inputs[frame_processor_input.name] = crop_vision_frame
|
||||
if frame_processor_input.name == 'weight':
|
||||
weight = numpy.array([ 1 ], dtype = numpy.double)
|
||||
frame_processor_inputs[frame_processor_input.name] = weight
|
||||
with THREAD_SEMAPHORE:
|
||||
crop_frame = frame_processor.run(None, frame_processor_inputs)[0][0]
|
||||
return crop_frame
|
||||
crop_vision_frame = frame_processor.run(None, frame_processor_inputs)[0][0]
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def prepare_crop_frame(crop_frame : Frame) -> Frame:
|
||||
crop_frame = crop_frame[:, :, ::-1] / 255.0
|
||||
crop_frame = (crop_frame - 0.5) / 0.5
|
||||
crop_frame = numpy.expand_dims(crop_frame.transpose(2, 0, 1), axis = 0).astype(numpy.float32)
|
||||
return crop_frame
|
||||
def prepare_crop_frame(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
crop_vision_frame = crop_vision_frame[:, :, ::-1] / 255.0
|
||||
crop_vision_frame = (crop_vision_frame - 0.5) / 0.5
|
||||
crop_vision_frame = numpy.expand_dims(crop_vision_frame.transpose(2, 0, 1), axis = 0).astype(numpy.float32)
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def normalize_crop_frame(crop_frame : Frame) -> Frame:
|
||||
crop_frame = numpy.clip(crop_frame, -1, 1)
|
||||
crop_frame = (crop_frame + 1) / 2
|
||||
crop_frame = crop_frame.transpose(1, 2, 0)
|
||||
crop_frame = (crop_frame * 255.0).round()
|
||||
crop_frame = crop_frame.astype(numpy.uint8)[:, :, ::-1]
|
||||
return crop_frame
|
||||
def normalize_crop_frame(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
crop_vision_frame = numpy.clip(crop_vision_frame, -1, 1)
|
||||
crop_vision_frame = (crop_vision_frame + 1) / 2
|
||||
crop_vision_frame = crop_vision_frame.transpose(1, 2, 0)
|
||||
crop_vision_frame = (crop_vision_frame * 255.0).round()
|
||||
crop_vision_frame = crop_vision_frame.astype(numpy.uint8)[:, :, ::-1]
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def blend_frame(temp_frame : Frame, paste_frame : Frame) -> Frame:
|
||||
def blend_frame(temp_vision_frame : VisionFrame, paste_vision_frame : VisionFrame) -> VisionFrame:
|
||||
face_enhancer_blend = 1 - (frame_processors_globals.face_enhancer_blend / 100)
|
||||
temp_frame = cv2.addWeighted(temp_frame, face_enhancer_blend, paste_frame, 1 - face_enhancer_blend, 0)
|
||||
return temp_frame
|
||||
temp_vision_frame = cv2.addWeighted(temp_vision_frame, face_enhancer_blend, paste_vision_frame, 1 - face_enhancer_blend, 0)
|
||||
return temp_vision_frame
|
||||
|
||||
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
return enhance_face(target_face, temp_frame)
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
return enhance_face(target_face, temp_vision_frame)
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
def process_frame(inputs : FaceEnhancerInputs) -> VisionFrame:
|
||||
reference_faces = inputs['reference_faces']
|
||||
target_vision_frame = inputs['target_vision_frame']
|
||||
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(temp_frame, reference_faces, facefusion.globals.reference_face_distance)
|
||||
similar_faces = find_similar_faces(reference_faces, target_vision_frame, facefusion.globals.reference_face_distance)
|
||||
if similar_faces:
|
||||
for similar_face in similar_faces:
|
||||
temp_frame = enhance_face(similar_face, temp_frame)
|
||||
target_vision_frame = enhance_face(similar_face, target_vision_frame)
|
||||
if 'one' in facefusion.globals.face_selector_mode:
|
||||
target_face = get_one_face(temp_frame)
|
||||
target_face = get_one_face(target_vision_frame)
|
||||
if target_face:
|
||||
temp_frame = enhance_face(target_face, temp_frame)
|
||||
target_vision_frame = enhance_face(target_face, target_vision_frame)
|
||||
if 'many' in facefusion.globals.face_selector_mode:
|
||||
many_faces = get_many_faces(temp_frame)
|
||||
many_faces = get_many_faces(target_vision_frame)
|
||||
if many_faces:
|
||||
for target_face in many_faces:
|
||||
temp_frame = enhance_face(target_face, temp_frame)
|
||||
return temp_frame
|
||||
target_vision_frame = enhance_face(target_face, target_vision_frame)
|
||||
return target_vision_frame
|
||||
|
||||
|
||||
def process_frames(source_path : List[str], temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
def process_frames(source_path : List[str], queue_payloads : List[QueuePayload], update_progress : Update_Process) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
for temp_frame_path in temp_frame_paths:
|
||||
temp_frame = read_image(temp_frame_path)
|
||||
result_frame = process_frame(None, reference_faces, temp_frame)
|
||||
write_image(temp_frame_path, result_frame)
|
||||
|
||||
for queue_payload in queue_payloads:
|
||||
target_vision_path = queue_payload['frame_path']
|
||||
target_vision_frame = read_image(target_vision_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(target_vision_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_path : str, target_path : str, output_path : str) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
target_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(None, reference_faces, target_frame)
|
||||
target_vision_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
from typing import Any, List, Literal, Optional
|
||||
from argparse import ArgumentParser
|
||||
import platform
|
||||
import threading
|
||||
import numpy
|
||||
import onnx
|
||||
@@ -12,16 +11,17 @@ import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import config, logger, wording
|
||||
from facefusion.execution_helper import apply_execution_provider_options
|
||||
from facefusion.face_analyser import get_one_face, get_average_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion.face_helper import warp_face_by_kps, paste_back
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, create_region_mask, clear_face_occluder, clear_face_parser
|
||||
from facefusion.face_helper import warp_face_by_face_landmark_5, paste_back
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, FaceSet, Frame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, Embedding
|
||||
from facefusion.filesystem import is_file, is_image, are_images, is_video, resolve_relative_path
|
||||
from facefusion.typing import Face, Embedding, VisionFrame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, QueuePayload
|
||||
from facefusion.filesystem import is_file, is_image, has_image, is_video, filter_image_paths, resolve_relative_path
|
||||
from facefusion.download import conditional_download, is_download_done
|
||||
from facefusion.vision import read_image, read_static_image, read_static_images, write_image
|
||||
from facefusion.processors.frame.typings import FaceSwapperInputs
|
||||
from facefusion.processors.frame import globals as frame_processors_globals
|
||||
from facefusion.processors.frame import choices as frame_processors_choices
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, create_region_mask, clear_face_occluder, clear_face_parser
|
||||
|
||||
FRAME_PROCESSOR = None
|
||||
MODEL_MATRIX = None
|
||||
@@ -78,7 +78,17 @@ MODELS : ModelSet =\
|
||||
'size': (512, 512),
|
||||
'mean': [ 0.0, 0.0, 0.0 ],
|
||||
'standard_deviation': [ 1.0, 1.0, 1.0 ]
|
||||
}
|
||||
},
|
||||
'uniface_256':
|
||||
{
|
||||
'type': 'uniface',
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/uniface_256.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/uniface_256.onnx'),
|
||||
'template': 'ffhq_512',
|
||||
'size': (256, 256),
|
||||
'mean': [ 0.0, 0.0, 0.0 ],
|
||||
'standard_deviation': [ 1.0, 1.0, 1.0 ]
|
||||
},
|
||||
}
|
||||
OPTIONS : Optional[OptionsWithModel] = None
|
||||
|
||||
@@ -134,11 +144,11 @@ def set_options(key : Literal['model'], value : Any) -> None:
|
||||
|
||||
|
||||
def register_args(program : ArgumentParser) -> None:
|
||||
if platform.system().lower() == 'darwin':
|
||||
if onnxruntime.__version__ == '1.17.0':
|
||||
face_swapper_model_fallback = 'inswapper_128'
|
||||
else:
|
||||
face_swapper_model_fallback = 'inswapper_128_fp16'
|
||||
program.add_argument('--face-swapper-model', help = wording.get('frame_processor_model_help'), default = config.get_str_value('frame_processors.face_swapper_model', face_swapper_model_fallback), choices = frame_processors_choices.face_swapper_models)
|
||||
program.add_argument('--face-swapper-model', help = wording.get('help.face_swapper_model'), default = config.get_str_value('frame_processors.face_swapper_model', face_swapper_model_fallback), choices = frame_processors_choices.face_swapper_models)
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
@@ -150,6 +160,8 @@ def apply_args(program : ArgumentParser) -> None:
|
||||
facefusion.globals.face_recognizer_model = 'arcface_inswapper'
|
||||
if args.face_swapper_model == 'simswap_256' or args.face_swapper_model == 'simswap_512_unofficial':
|
||||
facefusion.globals.face_recognizer_model = 'arcface_simswap'
|
||||
if args.face_swapper_model == 'uniface_256':
|
||||
facefusion.globals.face_recognizer_model = 'arcface_uniface'
|
||||
|
||||
|
||||
def pre_check() -> bool:
|
||||
@@ -173,10 +185,12 @@ def post_check() -> bool:
|
||||
|
||||
|
||||
def pre_process(mode : ProcessMode) -> bool:
|
||||
if not are_images(facefusion.globals.source_paths):
|
||||
if not has_image(facefusion.globals.source_paths):
|
||||
logger.error(wording.get('select_image_source') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
for source_frame in read_static_images(facefusion.globals.source_paths):
|
||||
source_image_paths = filter_image_paths(facefusion.globals.source_paths)
|
||||
source_frames = read_static_images(source_image_paths)
|
||||
for source_frame in source_frames:
|
||||
if not get_one_face(source_frame):
|
||||
logger.error(wording.get('no_source_face_detected') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
@@ -201,50 +215,57 @@ def post_process() -> None:
|
||||
clear_face_parser()
|
||||
|
||||
|
||||
def swap_face(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
def swap_face(source_face : Face, target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
model_template = get_options('model').get('template')
|
||||
model_size = get_options('model').get('size')
|
||||
crop_frame, affine_matrix = warp_face_by_kps(temp_frame, target_face.kps, model_template, model_size)
|
||||
crop_vision_frame, affine_matrix = warp_face_by_face_landmark_5(temp_vision_frame, target_face.landmark['5/68'], model_template, model_size)
|
||||
crop_mask_list = []
|
||||
|
||||
if 'box' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_static_box_mask(crop_frame.shape[:2][::-1], facefusion.globals.face_mask_blur, facefusion.globals.face_mask_padding))
|
||||
box_mask = create_static_box_mask(crop_vision_frame.shape[:2][::-1], facefusion.globals.face_mask_blur, facefusion.globals.face_mask_padding)
|
||||
crop_mask_list.append(box_mask)
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_occlusion_mask(crop_frame))
|
||||
crop_frame = prepare_crop_frame(crop_frame)
|
||||
crop_frame = apply_swap(source_face, crop_frame)
|
||||
crop_frame = normalize_crop_frame(crop_frame)
|
||||
occlusion_mask = create_occlusion_mask(crop_vision_frame)
|
||||
crop_mask_list.append(occlusion_mask)
|
||||
crop_vision_frame = prepare_crop_frame(crop_vision_frame)
|
||||
crop_vision_frame = apply_swap(source_face, crop_vision_frame)
|
||||
crop_vision_frame = normalize_crop_frame(crop_vision_frame)
|
||||
if 'region' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_region_mask(crop_frame, facefusion.globals.face_mask_regions))
|
||||
region_mask = create_region_mask(crop_vision_frame, facefusion.globals.face_mask_regions)
|
||||
crop_mask_list.append(region_mask)
|
||||
crop_mask = numpy.minimum.reduce(crop_mask_list).clip(0, 1)
|
||||
temp_frame = paste_back(temp_frame, crop_frame, crop_mask, affine_matrix)
|
||||
return temp_frame
|
||||
temp_vision_frame = paste_back(temp_vision_frame, crop_vision_frame, crop_mask, affine_matrix)
|
||||
return temp_vision_frame
|
||||
|
||||
|
||||
def apply_swap(source_face : Face, crop_frame : Frame) -> Frame:
|
||||
def apply_swap(source_face : Face, crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
frame_processor = get_frame_processor()
|
||||
model_type = get_options('model').get('type')
|
||||
frame_processor_inputs = {}
|
||||
|
||||
for frame_processor_input in frame_processor.get_inputs():
|
||||
if frame_processor_input.name == 'source':
|
||||
if model_type == 'blendswap':
|
||||
if model_type == 'blendswap' or model_type == 'uniface':
|
||||
frame_processor_inputs[frame_processor_input.name] = prepare_source_frame(source_face)
|
||||
else:
|
||||
frame_processor_inputs[frame_processor_input.name] = prepare_source_embedding(source_face)
|
||||
if frame_processor_input.name == 'target':
|
||||
frame_processor_inputs[frame_processor_input.name] = crop_frame
|
||||
crop_frame = frame_processor.run(None, frame_processor_inputs)[0][0]
|
||||
return crop_frame
|
||||
frame_processor_inputs[frame_processor_input.name] = crop_vision_frame
|
||||
crop_vision_frame = frame_processor.run(None, frame_processor_inputs)[0][0]
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def prepare_source_frame(source_face : Face) -> Frame:
|
||||
source_frame = read_static_image(facefusion.globals.source_paths[0])
|
||||
source_frame, _ = warp_face_by_kps(source_frame, source_face.kps, 'arcface_112_v2', (112, 112))
|
||||
source_frame = source_frame[:, :, ::-1] / 255.0
|
||||
source_frame = source_frame.transpose(2, 0, 1)
|
||||
source_frame = numpy.expand_dims(source_frame, axis = 0).astype(numpy.float32)
|
||||
return source_frame
|
||||
def prepare_source_frame(source_face : Face) -> VisionFrame:
|
||||
model_type = get_options('model').get('type')
|
||||
source_vision_frame = read_static_image(facefusion.globals.source_paths[0])
|
||||
if model_type == 'blendswap':
|
||||
source_vision_frame, _ = warp_face_by_face_landmark_5(source_vision_frame, source_face.landmark['5/68'], 'arcface_112_v2', (112, 112))
|
||||
if model_type == 'uniface':
|
||||
source_vision_frame, _ = warp_face_by_face_landmark_5(source_vision_frame, source_face.landmark['5/68'], 'ffhq_512', (256, 256))
|
||||
source_vision_frame = source_vision_frame[:, :, ::-1] / 255.0
|
||||
source_vision_frame = source_vision_frame.transpose(2, 0, 1)
|
||||
source_vision_frame = numpy.expand_dims(source_vision_frame, axis = 0).astype(numpy.float32)
|
||||
return source_vision_frame
|
||||
|
||||
|
||||
def prepare_source_embedding(source_face : Face) -> Embedding:
|
||||
@@ -258,62 +279,78 @@ def prepare_source_embedding(source_face : Face) -> Embedding:
|
||||
return source_embedding
|
||||
|
||||
|
||||
def prepare_crop_frame(crop_frame : Frame) -> Frame:
|
||||
def prepare_crop_frame(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
model_mean = get_options('model').get('mean')
|
||||
model_standard_deviation = get_options('model').get('standard_deviation')
|
||||
crop_frame = crop_frame[:, :, ::-1] / 255.0
|
||||
crop_frame = (crop_frame - model_mean) / model_standard_deviation
|
||||
crop_frame = crop_frame.transpose(2, 0, 1)
|
||||
crop_frame = numpy.expand_dims(crop_frame, axis = 0).astype(numpy.float32)
|
||||
return crop_frame
|
||||
crop_vision_frame = crop_vision_frame[:, :, ::-1] / 255.0
|
||||
crop_vision_frame = (crop_vision_frame - model_mean) / model_standard_deviation
|
||||
crop_vision_frame = crop_vision_frame.transpose(2, 0, 1)
|
||||
crop_vision_frame = numpy.expand_dims(crop_vision_frame, axis = 0).astype(numpy.float32)
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def normalize_crop_frame(crop_frame : Frame) -> Frame:
|
||||
crop_frame = crop_frame.transpose(1, 2, 0)
|
||||
crop_frame = (crop_frame * 255.0).round()
|
||||
crop_frame = crop_frame[:, :, ::-1]
|
||||
return crop_frame
|
||||
def normalize_crop_frame(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
crop_vision_frame = crop_vision_frame.transpose(1, 2, 0)
|
||||
crop_vision_frame = (crop_vision_frame * 255.0).round()
|
||||
crop_vision_frame = crop_vision_frame[:, :, ::-1]
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
return swap_face(source_face, target_face, temp_frame)
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
return swap_face(source_face, target_face, temp_vision_frame)
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
def process_frame(inputs : FaceSwapperInputs) -> VisionFrame:
|
||||
reference_faces = inputs['reference_faces']
|
||||
source_face = inputs['source_face']
|
||||
target_vision_frame = inputs['target_vision_frame']
|
||||
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(temp_frame, reference_faces, facefusion.globals.reference_face_distance)
|
||||
similar_faces = find_similar_faces(reference_faces, target_vision_frame, facefusion.globals.reference_face_distance)
|
||||
if similar_faces:
|
||||
for similar_face in similar_faces:
|
||||
temp_frame = swap_face(source_face, similar_face, temp_frame)
|
||||
target_vision_frame = swap_face(source_face, similar_face, target_vision_frame)
|
||||
if 'one' in facefusion.globals.face_selector_mode:
|
||||
target_face = get_one_face(temp_frame)
|
||||
target_face = get_one_face(target_vision_frame)
|
||||
if target_face:
|
||||
temp_frame = swap_face(source_face, target_face, temp_frame)
|
||||
target_vision_frame = swap_face(source_face, target_face, target_vision_frame)
|
||||
if 'many' in facefusion.globals.face_selector_mode:
|
||||
many_faces = get_many_faces(temp_frame)
|
||||
many_faces = get_many_faces(target_vision_frame)
|
||||
if many_faces:
|
||||
for target_face in many_faces:
|
||||
temp_frame = swap_face(source_face, target_face, temp_frame)
|
||||
return temp_frame
|
||||
target_vision_frame = swap_face(source_face, target_face, target_vision_frame)
|
||||
return target_vision_frame
|
||||
|
||||
|
||||
def process_frames(source_paths : List[str], temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
def process_frames(source_paths : List[str], queue_payloads : List[QueuePayload], update_progress : Update_Process) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
source_frames = read_static_images(source_paths)
|
||||
source_face = get_average_face(source_frames)
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
for temp_frame_path in temp_frame_paths:
|
||||
temp_frame = read_image(temp_frame_path)
|
||||
result_frame = process_frame(source_face, reference_faces, temp_frame)
|
||||
write_image(temp_frame_path, result_frame)
|
||||
|
||||
for queue_payload in queue_payloads:
|
||||
target_vision_path = queue_payload['frame_path']
|
||||
target_vision_frame = read_image(target_vision_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'source_face': source_face,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(target_vision_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_paths : List[str], target_path : str, output_path : str) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
source_frames = read_static_images(source_paths)
|
||||
source_face = get_average_face(source_frames)
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
target_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(source_face, reference_faces, target_frame)
|
||||
target_vision_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'source_face': source_face,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
|
||||
@@ -10,12 +10,13 @@ import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import config, logger, wording
|
||||
from facefusion.face_analyser import clear_face_analyser
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, FaceSet, Frame, Update_Process, ProcessMode, ModelSet, OptionsWithModel
|
||||
from facefusion.typing import Face, VisionFrame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, QueuePayload
|
||||
from facefusion.common_helper import create_metavar
|
||||
from facefusion.execution_helper import map_torch_backend
|
||||
from facefusion.filesystem import is_file, resolve_relative_path
|
||||
from facefusion.download import conditional_download, is_download_done
|
||||
from facefusion.vision import read_image, read_static_image, write_image
|
||||
from facefusion.processors.frame.typings import FrameEnhancerInputs
|
||||
from facefusion.processors.frame import globals as frame_processors_globals
|
||||
from facefusion.processors.frame import choices as frame_processors_choices
|
||||
|
||||
@@ -91,8 +92,8 @@ def set_options(key : Literal['model'], value : Any) -> None:
|
||||
|
||||
|
||||
def register_args(program : ArgumentParser) -> None:
|
||||
program.add_argument('--frame-enhancer-model', help = wording.get('frame_processor_model_help'), default = config.get_str_value('frame_processors.frame_enhancer_model', 'real_esrgan_x2plus'), choices = frame_processors_choices.frame_enhancer_models)
|
||||
program.add_argument('--frame-enhancer-blend', help = wording.get('frame_processor_blend_help'), type = int, default = config.get_int_value('frame_processors.frame_enhancer_blend', '80'), choices = frame_processors_choices.frame_enhancer_blend_range, metavar = create_metavar(frame_processors_choices.frame_enhancer_blend_range))
|
||||
program.add_argument('--frame-enhancer-model', help = wording.get('help.frame_enhancer_model'), default = config.get_str_value('frame_processors.frame_enhancer_model', 'real_esrgan_x2plus'), choices = frame_processors_choices.frame_enhancer_models)
|
||||
program.add_argument('--frame-enhancer-blend', help = wording.get('help.frame_enhancer_blend'), type = int, default = config.get_int_value('frame_processors.frame_enhancer_blend', '80'), choices = frame_processors_choices.frame_enhancer_blend_range, metavar = create_metavar(frame_processors_choices.frame_enhancer_blend_range))
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
@@ -137,41 +138,48 @@ def post_process() -> None:
|
||||
clear_content_analyser()
|
||||
|
||||
|
||||
def enhance_frame(temp_frame : Frame) -> Frame:
|
||||
def enhance_frame(temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
with THREAD_SEMAPHORE:
|
||||
paste_frame, _ = get_frame_processor().enhance(temp_frame)
|
||||
temp_frame = blend_frame(temp_frame, paste_frame)
|
||||
return temp_frame
|
||||
paste_vision_frame, _ = get_frame_processor().enhance(temp_vision_frame)
|
||||
temp_vision_frame = blend_frame(temp_vision_frame, paste_vision_frame)
|
||||
return temp_vision_frame
|
||||
|
||||
|
||||
def blend_frame(temp_frame : Frame, paste_frame : Frame) -> Frame:
|
||||
def blend_frame(temp_vision_frame : VisionFrame, paste_vision_frame : VisionFrame) -> VisionFrame:
|
||||
frame_enhancer_blend = 1 - (frame_processors_globals.frame_enhancer_blend / 100)
|
||||
paste_frame_height, paste_frame_width = paste_frame.shape[0:2]
|
||||
temp_frame = cv2.resize(temp_frame, (paste_frame_width, paste_frame_height))
|
||||
temp_frame = cv2.addWeighted(temp_frame, frame_enhancer_blend, paste_frame, 1 - frame_enhancer_blend, 0)
|
||||
return temp_frame
|
||||
temp_vision_frame = cv2.resize(temp_vision_frame, (paste_vision_frame.shape[1], paste_vision_frame.shape[0]))
|
||||
temp_vision_frame = cv2.addWeighted(temp_vision_frame, frame_enhancer_blend, paste_vision_frame, 1 - frame_enhancer_blend, 0)
|
||||
return temp_vision_frame
|
||||
|
||||
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
pass
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
return enhance_frame(temp_frame)
|
||||
def process_frame(inputs : FrameEnhancerInputs) -> VisionFrame:
|
||||
target_vision_frame = inputs['target_vision_frame']
|
||||
return enhance_frame(target_vision_frame)
|
||||
|
||||
|
||||
def process_frames(source_paths : List[str], temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
for temp_frame_path in temp_frame_paths:
|
||||
temp_frame = read_image(temp_frame_path)
|
||||
result_frame = process_frame(None, None, temp_frame)
|
||||
write_image(temp_frame_path, result_frame)
|
||||
def process_frames(source_paths : List[str], queue_payloads : List[QueuePayload], update_progress : Update_Process) -> None:
|
||||
for queue_payload in queue_payloads:
|
||||
target_vision_path = queue_payload['frame_path']
|
||||
target_vision_frame = read_image(target_vision_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(target_vision_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_paths : List[str], target_path : str, output_path : str) -> None:
|
||||
target_frame = read_static_image(target_path)
|
||||
result = process_frame(None, None, target_frame)
|
||||
write_image(output_path, result)
|
||||
target_vision_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
def process_video(source_paths : List[str], temp_frame_paths : List[str]) -> None:
|
||||
|
||||
+248
@@ -0,0 +1,248 @@
|
||||
from typing import Any, List, Literal, Optional
|
||||
from argparse import ArgumentParser
|
||||
import threading
|
||||
import cv2
|
||||
import numpy
|
||||
import onnxruntime
|
||||
|
||||
import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import config, logger, wording
|
||||
from facefusion.execution_helper import apply_execution_provider_options
|
||||
from facefusion.face_analyser import get_one_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, create_mouth_mask, clear_face_occluder, clear_face_parser
|
||||
from facefusion.face_helper import warp_face_by_face_landmark_5, warp_face_by_bounding_box, paste_back, create_bounding_box_from_landmark
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, VisionFrame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, AudioFrame, QueuePayload
|
||||
from facefusion.filesystem import is_file, has_audio, resolve_relative_path
|
||||
from facefusion.download import conditional_download, is_download_done
|
||||
from facefusion.audio import read_static_audio, get_audio_frame
|
||||
from facefusion.filesystem import is_image, is_video, filter_audio_paths
|
||||
from facefusion.common_helper import get_first
|
||||
from facefusion.vision import read_image, write_image, read_static_image
|
||||
from facefusion.processors.frame.typings import LipSyncerInputs
|
||||
from facefusion.processors.frame import globals as frame_processors_globals
|
||||
from facefusion.processors.frame import choices as frame_processors_choices
|
||||
|
||||
FRAME_PROCESSOR = None
|
||||
MODEL_MATRIX = None
|
||||
THREAD_LOCK : threading.Lock = threading.Lock()
|
||||
NAME = __name__.upper()
|
||||
MODELS : ModelSet =\
|
||||
{
|
||||
'wav2lip_gan':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/wav2lip_gan.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/wav2lip_gan.onnx'),
|
||||
}
|
||||
}
|
||||
OPTIONS : Optional[OptionsWithModel] = None
|
||||
|
||||
|
||||
def get_frame_processor() -> Any:
|
||||
global FRAME_PROCESSOR
|
||||
|
||||
with THREAD_LOCK:
|
||||
if FRAME_PROCESSOR is None:
|
||||
model_path = get_options('model').get('path')
|
||||
FRAME_PROCESSOR = onnxruntime.InferenceSession(model_path, providers = apply_execution_provider_options(facefusion.globals.execution_providers))
|
||||
return FRAME_PROCESSOR
|
||||
|
||||
|
||||
def clear_frame_processor() -> None:
|
||||
global FRAME_PROCESSOR
|
||||
|
||||
FRAME_PROCESSOR = None
|
||||
|
||||
|
||||
def get_options(key : Literal['model']) -> Any:
|
||||
global OPTIONS
|
||||
|
||||
if OPTIONS is None:
|
||||
OPTIONS =\
|
||||
{
|
||||
'model': MODELS[frame_processors_globals.lip_syncer_model]
|
||||
}
|
||||
return OPTIONS.get(key)
|
||||
|
||||
|
||||
def set_options(key : Literal['model'], value : Any) -> None:
|
||||
global OPTIONS
|
||||
|
||||
OPTIONS[key] = value
|
||||
|
||||
|
||||
def register_args(program : ArgumentParser) -> None:
|
||||
program.add_argument('--lip-syncer-model', help = wording.get('help.lip_syncer_model'), default = config.get_str_value('frame_processors.lip_syncer_model', 'wav2lip_gan'), choices = frame_processors_choices.lip_syncer_models)
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
args = program.parse_args()
|
||||
frame_processors_globals.lip_syncer_model = args.lip_syncer_model
|
||||
|
||||
|
||||
def pre_check() -> bool:
|
||||
if not facefusion.globals.skip_download:
|
||||
download_directory_path = resolve_relative_path('../.assets/models')
|
||||
model_url = get_options('model').get('url')
|
||||
conditional_download(download_directory_path, [ model_url ])
|
||||
return True
|
||||
|
||||
|
||||
def post_check() -> bool:
|
||||
model_url = get_options('model').get('url')
|
||||
model_path = get_options('model').get('path')
|
||||
if not facefusion.globals.skip_download and not is_download_done(model_url, model_path):
|
||||
logger.error(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
elif not is_file(model_path):
|
||||
logger.error(wording.get('model_file_not_present') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def pre_process(mode : ProcessMode) -> bool:
|
||||
if not has_audio(facefusion.globals.source_paths):
|
||||
logger.error(wording.get('select_audio_source') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
if mode in [ 'output', 'preview' ] and not is_image(facefusion.globals.target_path) and not is_video(facefusion.globals.target_path):
|
||||
logger.error(wording.get('select_image_or_video_target') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
if mode == 'output' and not facefusion.globals.output_path:
|
||||
logger.error(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def post_process() -> None:
|
||||
read_static_image.cache_clear()
|
||||
read_static_audio.cache_clear()
|
||||
if facefusion.globals.video_memory_strategy == 'strict' or facefusion.globals.video_memory_strategy == 'moderate':
|
||||
clear_frame_processor()
|
||||
if facefusion.globals.video_memory_strategy == 'strict':
|
||||
clear_face_analyser()
|
||||
clear_content_analyser()
|
||||
clear_face_occluder()
|
||||
clear_face_parser()
|
||||
|
||||
|
||||
def sync_lip(target_face : Face, temp_audio_frame : AudioFrame, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
frame_processor = get_frame_processor()
|
||||
temp_audio_frame = prepare_audio_frame(temp_audio_frame)
|
||||
crop_vision_frame, affine_matrix = warp_face_by_face_landmark_5(temp_vision_frame, target_face.landmark['5/68'], 'ffhq_512', (512, 512))
|
||||
face_landmark_68 = cv2.transform(target_face.landmark['68'].reshape(1, -1, 2), affine_matrix).reshape(-1, 2)
|
||||
bounding_box = create_bounding_box_from_landmark(face_landmark_68)
|
||||
bounding_box[1] -= numpy.abs(bounding_box[3] - bounding_box[1]) * 0.125
|
||||
mouth_mask = create_mouth_mask(face_landmark_68)
|
||||
box_mask = create_static_box_mask(crop_vision_frame.shape[:2][::-1], facefusion.globals.face_mask_blur, facefusion.globals.face_mask_padding)
|
||||
crop_mask_list =\
|
||||
[
|
||||
mouth_mask,
|
||||
box_mask
|
||||
]
|
||||
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
occlusion_mask = create_occlusion_mask(crop_vision_frame)
|
||||
crop_mask_list.append(occlusion_mask)
|
||||
close_vision_frame, closeup_matrix = warp_face_by_bounding_box(crop_vision_frame, bounding_box, (96, 96))
|
||||
close_vision_frame = prepare_crop_frame(close_vision_frame)
|
||||
close_vision_frame = frame_processor.run(None,
|
||||
{
|
||||
'source': temp_audio_frame,
|
||||
'target': close_vision_frame
|
||||
})[0]
|
||||
crop_vision_frame = normalize_crop_frame(close_vision_frame)
|
||||
crop_vision_frame = cv2.warpAffine(crop_vision_frame, cv2.invertAffineTransform(closeup_matrix), (512, 512), borderMode = cv2.BORDER_REPLICATE)
|
||||
crop_mask = numpy.minimum.reduce(crop_mask_list)
|
||||
paste_vision_frame = paste_back(temp_vision_frame, crop_vision_frame, crop_mask, affine_matrix)
|
||||
return paste_vision_frame
|
||||
|
||||
|
||||
def prepare_audio_frame(temp_audio_frame : AudioFrame) -> AudioFrame:
|
||||
temp_audio_frame = numpy.maximum(numpy.exp(-5 * numpy.log(10)), temp_audio_frame)
|
||||
temp_audio_frame = numpy.log10(temp_audio_frame) * 1.6 + 3.2
|
||||
temp_audio_frame = temp_audio_frame.clip(-4, 4).astype(numpy.float32)
|
||||
temp_audio_frame = numpy.expand_dims(temp_audio_frame, axis = (0, 1))
|
||||
return temp_audio_frame
|
||||
|
||||
|
||||
def prepare_crop_frame(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
crop_vision_frame = numpy.expand_dims(crop_vision_frame, axis = 0)
|
||||
prepare_vision_frame = crop_vision_frame.copy()
|
||||
prepare_vision_frame[:, 48:] = 0
|
||||
crop_vision_frame = numpy.concatenate((prepare_vision_frame, crop_vision_frame), axis = 3)
|
||||
crop_vision_frame = crop_vision_frame.transpose(0, 3, 1, 2).astype('float32') / 255.0
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def normalize_crop_frame(crop_vision_frame : VisionFrame) -> VisionFrame:
|
||||
crop_vision_frame = crop_vision_frame[0].transpose(1, 2, 0)
|
||||
crop_vision_frame = crop_vision_frame.clip(0, 1) * 255
|
||||
crop_vision_frame = crop_vision_frame.astype(numpy.uint8)
|
||||
return crop_vision_frame
|
||||
|
||||
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
pass
|
||||
|
||||
|
||||
def process_frame(inputs : LipSyncerInputs) -> VisionFrame:
|
||||
reference_faces = inputs['reference_faces']
|
||||
source_audio_frame = inputs['source_audio_frame']
|
||||
target_vision_frame = inputs['target_vision_frame']
|
||||
is_source_audio_frame = isinstance(source_audio_frame, numpy.ndarray) and source_audio_frame.any()
|
||||
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(reference_faces, target_vision_frame, facefusion.globals.reference_face_distance)
|
||||
if similar_faces and is_source_audio_frame:
|
||||
for similar_face in similar_faces:
|
||||
target_vision_frame = sync_lip(similar_face, source_audio_frame, target_vision_frame)
|
||||
if 'one' in facefusion.globals.face_selector_mode:
|
||||
target_face = get_one_face(target_vision_frame)
|
||||
if target_face and is_source_audio_frame:
|
||||
target_vision_frame = sync_lip(target_face, source_audio_frame, target_vision_frame)
|
||||
if 'many' in facefusion.globals.face_selector_mode:
|
||||
many_faces = get_many_faces(target_vision_frame)
|
||||
if many_faces and is_source_audio_frame:
|
||||
for target_face in many_faces:
|
||||
target_vision_frame = sync_lip(target_face, source_audio_frame, target_vision_frame)
|
||||
return target_vision_frame
|
||||
|
||||
|
||||
def process_frames(source_paths : List[str], queue_payloads : List[QueuePayload], update_progress : Update_Process) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
source_audio_path = get_first(filter_audio_paths(source_paths))
|
||||
target_video_fps = facefusion.globals.output_video_fps
|
||||
|
||||
for queue_payload in queue_payloads:
|
||||
frame_number = queue_payload['frame_number']
|
||||
target_vision_path = queue_payload['frame_path']
|
||||
source_audio_frame = get_audio_frame(source_audio_path, target_video_fps, frame_number)
|
||||
target_vision_frame = read_image(target_vision_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'source_audio_frame': source_audio_frame,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(target_vision_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_paths : List[str], target_path : str, output_path : str) -> None:
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
source_audio_path = get_first(filter_audio_paths(source_paths))
|
||||
source_audio_frame = get_audio_frame(source_audio_path, 25)
|
||||
target_vision_frame = read_static_image(target_path)
|
||||
result_frame = process_frame(
|
||||
{
|
||||
'reference_faces': reference_faces,
|
||||
'source_audio_frame': source_audio_frame,
|
||||
'target_vision_frame': target_vision_frame
|
||||
})
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
def process_video(source_paths : List[str], temp_frame_paths : List[str]) -> None:
|
||||
frame_processors.multi_process_frames(source_paths, temp_frame_paths, process_frames)
|
||||
@@ -1,6 +1,36 @@
|
||||
from typing import Literal
|
||||
from typing import Literal, TypedDict
|
||||
|
||||
FaceSwapperModel = Literal['blendswap_256', 'inswapper_128', 'inswapper_128_fp16', 'simswap_256', 'simswap_512_unofficial']
|
||||
from facefusion.typing import Face, FaceSet, AudioFrame, VisionFrame
|
||||
|
||||
FaceDebuggerItem = Literal['bounding-box', 'landmark-5', 'landmark-68', 'face-mask', 'score', 'age', 'gender']
|
||||
FaceEnhancerModel = Literal['codeformer', 'gfpgan_1.2', 'gfpgan_1.3', 'gfpgan_1.4', 'gpen_bfr_256', 'gpen_bfr_512', 'restoreformer_plus_plus']
|
||||
FaceSwapperModel = Literal['blendswap_256', 'inswapper_128', 'inswapper_128_fp16', 'simswap_256', 'simswap_512_unofficial', 'uniface_256']
|
||||
FrameEnhancerModel = Literal['real_esrgan_x2plus', 'real_esrgan_x4plus', 'real_esrnet_x4plus']
|
||||
FaceDebuggerItem = Literal['bbox', 'kps', 'face-mask', 'score', 'distance']
|
||||
LipSyncerModel = Literal['wav2lip_gan']
|
||||
|
||||
FaceDebuggerInputs = TypedDict('FaceDebuggerInputs',
|
||||
{
|
||||
'reference_faces' : FaceSet,
|
||||
'target_vision_frame' : VisionFrame
|
||||
})
|
||||
FaceEnhancerInputs = TypedDict('FaceEnhancerInputs',
|
||||
{
|
||||
'reference_faces' : FaceSet,
|
||||
'target_vision_frame' : VisionFrame
|
||||
})
|
||||
FaceSwapperInputs = TypedDict('FaceSwapperInputs',
|
||||
{
|
||||
'reference_faces' : FaceSet,
|
||||
'source_face' : Face,
|
||||
'target_vision_frame' : VisionFrame
|
||||
})
|
||||
FrameEnhancerInputs = TypedDict('FrameEnhancerInputs',
|
||||
{
|
||||
'target_vision_frame' : VisionFrame
|
||||
})
|
||||
LipSyncerInputs = TypedDict('LipSyncerInputs',
|
||||
{
|
||||
'reference_faces' : FaceSet,
|
||||
'source_audio_frame' : AudioFrame,
|
||||
'target_vision_frame' : VisionFrame
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user