mirror of
https://github.com/facefusion/facefusion.git
synced 2026-07-12 13:16:34 +02:00
2.1.0 (#253)
* Operating system specific installer options * Update dependencies * Sorting before NMS according to the standard * Minor typing fix * Change the wording * Update preview.py (#222) Added a release listener to the preview frame slider, this will update the frame preview with the latest frame * Combine preview slider listener * Remove change listener * Introduce multi source (#223) * Implement multi source * Adjust face enhancer and face debugger to multi source * Implement multi source to UI * Implement multi source to UI part2 * Implement multi source to UI part3 * Implement multi source to UI part4 * Some cleanup * Add face occluder (#225) (#226) * Add face occluder (#225) * add face-occluder (commandline only) * review 1 * Update face_masker.py * Update face_masker.py * Add gui & fix typing * Minor naming cleanup * Minor naming cleanup part2 --------- Co-authored-by: Harisreedhar <46858047+harisreedhar@users.noreply.github.com> * Update usage information * Fix averaged normed_embedding * Remove blur from face occluder, enable accelerators * Switch to RANSAC with 100 threshold * Update face_enhancer.py (#229) * Update face_debugger.py (#230) * Split utilities (#232) * Split utilities * Split utilities part2 * Split utilities part3 * Split utilities part4 * Some cleanup * Implement log level support (#233) * Implement log level support * Fix testing * Implement debug logger * Implement debug logger * Fix alignment offset (#235) * Update face_helper.py * fix 2 * Enforce virtual environment via installer * Enforce virtual environment via installer * Enforce virtual environment via installer * Enforce virtual environment via installer * Feat/multi process reference faces (#239) * Multi processing aware reference faces * First clean up and joining of files * Finalize the face store * Reduce similar face detection to one set, use __name__ for scopes in logger * Rename to face_occluder * Introduce ModelSet type * Improve webcam error handling * Prevent null pointer on is_image() and is_video() * Prevent null pointer on is_image() and is_video() * Fix find similar faces * Fix find similar faces * Fix process_images for face enhancer * Bunch of minor improvements * onnxruntime for ROCM under linux * Improve mask related naming * Fix falsy import * Fix typo * Feat/face parser refactoring (#247) * Face parser update (#244) * face-parser * Update face_masker.py * update debugger * Update globals.py * Update face_masker.py * Refactor code to split occlusion from region * fix (#246) * fix * fix debugger resolution * flip input to horizontal * Clean up UI * Reduce the regions to inside face only * Reduce the regions to inside face only --------- Co-authored-by: Harisreedhar <46858047+harisreedhar@users.noreply.github.com> * Fix enhancer, remove useless dest in add_argument() * Prevent unselect of the face_mask_regions via UI * Prepare next release * Shorten arguments that have choices and nargs * Add missing clear to face debugger --------- Co-authored-by: Mathias <github@feroc.de> Co-authored-by: Harisreedhar <46858047+harisreedhar@users.noreply.github.com>
This commit is contained in:
Henry Ruhs
@@ -8,8 +8,8 @@ from tqdm import tqdm
|
||||
|
||||
import facefusion.globals
|
||||
from facefusion.typing import Process_Frames
|
||||
from facefusion import wording
|
||||
from facefusion.utilities import encode_execution_providers
|
||||
from facefusion.execution_helper import encode_execution_providers
|
||||
from facefusion import logger, wording
|
||||
|
||||
FRAME_PROCESSORS_MODULES : List[ModuleType] = []
|
||||
FRAME_PROCESSORS_METHODS =\
|
||||
@@ -22,6 +22,7 @@ FRAME_PROCESSORS_METHODS =\
|
||||
'apply_args',
|
||||
'pre_check',
|
||||
'pre_process',
|
||||
'get_reference_frame',
|
||||
'process_frame',
|
||||
'process_frames',
|
||||
'process_image',
|
||||
@@ -36,7 +37,8 @@ def load_frame_processor_module(frame_processor : str) -> Any:
|
||||
for method_name in FRAME_PROCESSORS_METHODS:
|
||||
if not hasattr(frame_processor_module, method_name):
|
||||
raise NotImplementedError
|
||||
except ModuleNotFoundError:
|
||||
except ModuleNotFoundError as exception:
|
||||
logger.debug(exception.msg, __name__.upper())
|
||||
sys.exit(wording.get('frame_processor_not_loaded').format(frame_processor = frame_processor))
|
||||
except NotImplementedError:
|
||||
sys.exit(wording.get('frame_processor_not_implemented').format(frame_processor = frame_processor))
|
||||
@@ -61,8 +63,8 @@ def clear_frame_processors_modules() -> None:
|
||||
FRAME_PROCESSORS_MODULES = []
|
||||
|
||||
|
||||
def multi_process_frames(source_path : 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 = ' =') as progress:
|
||||
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:
|
||||
progress.set_postfix(
|
||||
{
|
||||
'execution_providers': encode_execution_providers(facefusion.globals.execution_providers),
|
||||
@@ -75,7 +77,7 @@ def multi_process_frames(source_path : str, temp_frame_paths : List[str], proces
|
||||
queue_per_future = max(len(temp_frame_paths) // facefusion.globals.execution_thread_count * facefusion.globals.execution_queue_count, 1)
|
||||
while not queue_temp_frame_paths.empty():
|
||||
payload_temp_frame_paths = pick_queue(queue_temp_frame_paths, queue_per_future)
|
||||
future = executor.submit(process_frames, source_path, payload_temp_frame_paths, progress.update)
|
||||
future = executor.submit(process_frames, source_paths, payload_temp_frame_paths, progress.update)
|
||||
futures.append(future)
|
||||
for future_done in as_completed(futures):
|
||||
future_done.result()
|
||||
|
||||
@@ -6,15 +6,16 @@ import numpy
|
||||
import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import wording
|
||||
from facefusion.face_analyser import get_one_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion.face_reference import get_face_reference
|
||||
from facefusion.face_analyser import get_one_face, get_average_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, Frame, Update_Process, ProcessMode
|
||||
from facefusion.vision import read_image, read_static_image, write_image
|
||||
from facefusion.face_helper import warp_face, create_static_mask_frame
|
||||
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
|
||||
from facefusion.face_masker import create_static_box_mask, create_occlusion_mask, create_region_mask, clear_face_occluder, clear_face_parser
|
||||
from facefusion.processors.frame import globals as frame_processors_globals, choices as frame_processors_choices
|
||||
|
||||
NAME = 'FACEFUSION.FRAME_PROCESSOR.FACE_DEBUGGER'
|
||||
NAME = __name__.upper()
|
||||
|
||||
|
||||
def get_frame_processor() -> None:
|
||||
@@ -34,7 +35,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'), dest = 'face_debugger_items', default = [ 'kps', 'face-mask' ], choices = frame_processors_choices.face_debugger_items, nargs = '+')
|
||||
program.add_argument('--face-debugger-items', help = wording.get('face_debugger_items_help').format(choices = ', '.join(frame_processors_choices.face_debugger_items)), default = [ 'kps', 'face-mask' ], choices = frame_processors_choices.face_debugger_items, nargs = '+', metavar = 'FACE_DEBUGGER_ITEMS')
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
@@ -54,6 +55,9 @@ def post_process() -> None:
|
||||
clear_frame_processor()
|
||||
clear_face_analyser()
|
||||
clear_content_analyser()
|
||||
clear_face_occluder()
|
||||
clear_face_parser()
|
||||
read_static_image.cache_clear()
|
||||
|
||||
|
||||
def debug_face(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
@@ -63,14 +67,23 @@ def debug_face(source_face : Face, target_face : Face, temp_frame : Frame) -> Fr
|
||||
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)
|
||||
if 'face-mask' in frame_processors_globals.face_debugger_items:
|
||||
crop_frame, affine_matrix = warp_face(temp_frame, target_face.kps, 'arcface_v2', (128, 128))
|
||||
crop_frame, affine_matrix = warp_face(temp_frame, target_face.kps, 'arcface_128_v2', (128, 512))
|
||||
inverse_matrix = cv2.invertAffineTransform(affine_matrix)
|
||||
temp_frame_size = temp_frame.shape[:2][::-1]
|
||||
mask_frame = create_static_mask_frame(crop_frame.shape[:2], 0, facefusion.globals.face_mask_padding)
|
||||
mask_frame[mask_frame > 0] = 255
|
||||
inverse_mask_frame = cv2.warpAffine(mask_frame.astype(numpy.uint8), inverse_matrix, temp_frame_size)
|
||||
inverse_mask_contours = cv2.findContours(inverse_mask_frame, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)[0]
|
||||
cv2.drawContours(temp_frame, inverse_mask_contours, 0, primary_color, 2)
|
||||
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))
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_occlusion_mask(crop_frame))
|
||||
if 'region' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_region_mask(crop_frame, facefusion.globals.face_mask_regions))
|
||||
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_edges = cv2.threshold(inverse_mask_frame, 100, 255, cv2.THRESH_BINARY)[1]
|
||||
inverse_mask_frame_edges[inverse_mask_frame_edges > 0] = 255
|
||||
inverse_mask_contours = cv2.findContours(inverse_mask_frame_edges, cv2.RETR_LIST, cv2.CHAIN_APPROX_NONE)[0]
|
||||
cv2.drawContours(temp_frame, inverse_mask_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)
|
||||
@@ -83,9 +96,13 @@ def debug_face(source_face : Face, target_face : Face, temp_frame : Frame) -> Fr
|
||||
return temp_frame
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_face : Face, temp_frame : Frame) -> Frame:
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
pass
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(temp_frame, reference_face, facefusion.globals.reference_face_distance)
|
||||
similar_faces = find_similar_faces(temp_frame, reference_faces, facefusion.globals.reference_face_distance)
|
||||
if similar_faces:
|
||||
for similar_face in similar_faces:
|
||||
temp_frame = debug_face(source_face, similar_face, temp_frame)
|
||||
@@ -101,23 +118,25 @@ def process_frame(source_face : Face, reference_face : Face, temp_frame : Frame)
|
||||
return temp_frame
|
||||
|
||||
|
||||
def process_frames(source_path : str, temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
source_face = get_one_face(read_static_image(source_path))
|
||||
reference_face = get_face_reference() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
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)
|
||||
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_face, temp_frame)
|
||||
result_frame = process_frame(source_face, reference_faces, temp_frame)
|
||||
write_image(temp_frame_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_path : str, target_path : str, output_path : str) -> None:
|
||||
source_face = get_one_face(read_static_image(source_path))
|
||||
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_face = get_one_face(target_frame, facefusion.globals.reference_face_position) if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
result_frame = process_frame(source_face, reference_face, target_frame)
|
||||
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)
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
def process_video(source_path : str, temp_frame_paths : List[str]) -> None:
|
||||
frame_processors.multi_process_frames(source_path, temp_frame_paths, process_frames)
|
||||
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,4 +1,4 @@
|
||||
from typing import Any, List, Dict, Literal, Optional
|
||||
from typing import Any, List, Literal, Optional
|
||||
from argparse import ArgumentParser
|
||||
import cv2
|
||||
import threading
|
||||
@@ -7,69 +7,73 @@ import onnxruntime
|
||||
|
||||
import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import wording
|
||||
from facefusion.face_analyser import get_many_faces, clear_face_analyser
|
||||
from facefusion import logger, wording
|
||||
from facefusion.face_analyser import get_many_faces, clear_face_analyser, find_similar_faces, get_one_face
|
||||
from facefusion.face_helper import warp_face, paste_back
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, Frame, Update_Process, ProcessMode, ModelValue, OptionsWithModel
|
||||
from facefusion.utilities import conditional_download, resolve_relative_path, is_image, is_video, is_file, is_download_done, create_metavar, update_status
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.typing import Face, FaceSet, Frame, Update_Process, ProcessMode, ModelSet, OptionsWithModel
|
||||
from facefusion.cli_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 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()
|
||||
THREAD_LOCK : threading.Lock = threading.Lock()
|
||||
NAME = 'FACEFUSION.FRAME_PROCESSOR.FACE_ENHANCER'
|
||||
MODELS : Dict[str, ModelValue] =\
|
||||
NAME = __name__.upper()
|
||||
MODELS : ModelSet =\
|
||||
{
|
||||
'codeformer':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/codeformer.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/codeformer.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 512)
|
||||
},
|
||||
'gfpgan_1.2':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/gfpgan_1.2.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/gfpgan_1.2.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 512)
|
||||
},
|
||||
'gfpgan_1.3':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/gfpgan_1.3.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/gfpgan_1.3.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 512)
|
||||
},
|
||||
'gfpgan_1.4':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/gfpgan_1.4.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/gfpgan_1.4.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 512)
|
||||
},
|
||||
'gpen_bfr_256':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/gpen_bfr_256.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/gpen_bfr_256.onnx'),
|
||||
'template': 'arcface_v2',
|
||||
'template': 'arcface_128_v2',
|
||||
'size': (128, 256)
|
||||
},
|
||||
'gpen_bfr_512':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/gpen_bfr_512.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/gpen_bfr_512.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 512)
|
||||
},
|
||||
'restoreformer':
|
||||
{
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/restoreformer.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/restoreformer.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 512)
|
||||
}
|
||||
}
|
||||
@@ -110,8 +114,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'), dest = 'face_enhancer_model', default = 'gfpgan_1.4', choices = frame_processors_choices.face_enhancer_models)
|
||||
program.add_argument('--face-enhancer-blend', help = wording.get('frame_processor_blend_help'), dest = 'face_enhancer_blend', type = int, default = 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('frame_processor_model_help'), default = '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 = 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:
|
||||
@@ -132,16 +136,16 @@ def pre_process(mode : ProcessMode) -> 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):
|
||||
update_status(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
elif not is_file(model_path):
|
||||
update_status(wording.get('model_file_not_present') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('model_file_not_present') + 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):
|
||||
update_status(wording.get('select_image_or_video_target') + wording.get('exclamation_mark'), NAME)
|
||||
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:
|
||||
update_status(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
return True
|
||||
|
||||
@@ -150,6 +154,7 @@ def post_process() -> None:
|
||||
clear_frame_processor()
|
||||
clear_face_analyser()
|
||||
clear_content_analyser()
|
||||
clear_face_occluder()
|
||||
read_static_image.cache_clear()
|
||||
|
||||
|
||||
@@ -158,6 +163,12 @@ def enhance_face(target_face: Face, temp_frame: Frame) -> Frame:
|
||||
model_template = get_options('model').get('template')
|
||||
model_size = get_options('model').get('size')
|
||||
crop_frame, affine_matrix = warp_face(temp_frame, target_face.kps, model_template, model_size)
|
||||
crop_mask_list =\
|
||||
[
|
||||
create_static_box_mask(crop_frame.shape[:2][::-1], facefusion.globals.face_mask_blur, (0, 0, 0, 0))
|
||||
]
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_occlusion_mask(crop_frame))
|
||||
crop_frame = prepare_crop_frame(crop_frame)
|
||||
frame_processor_inputs = {}
|
||||
for frame_processor_input in frame_processor.get_inputs():
|
||||
@@ -168,7 +179,8 @@ def enhance_face(target_face: Face, temp_frame: Frame) -> Frame:
|
||||
with THREAD_SEMAPHORE:
|
||||
crop_frame = frame_processor.run(None, frame_processor_inputs)[0][0]
|
||||
crop_frame = normalize_crop_frame(crop_frame)
|
||||
paste_frame = paste_back(temp_frame, crop_frame, affine_matrix, facefusion.globals.face_mask_blur, (0, 0, 0, 0))
|
||||
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
|
||||
|
||||
@@ -195,27 +207,43 @@ def blend_frame(temp_frame : Frame, paste_frame : Frame) -> Frame:
|
||||
return temp_frame
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_face : Face, temp_frame : Frame) -> Frame:
|
||||
many_faces = get_many_faces(temp_frame)
|
||||
if many_faces:
|
||||
for target_face in many_faces:
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Optional[Frame]:
|
||||
return enhance_face(target_face, temp_frame)
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(temp_frame, reference_faces, facefusion.globals.reference_face_distance)
|
||||
if similar_faces:
|
||||
for similar_face in similar_faces:
|
||||
temp_frame = enhance_face(similar_face, temp_frame)
|
||||
if 'one' in facefusion.globals.face_selector_mode:
|
||||
target_face = get_one_face(temp_frame)
|
||||
if target_face:
|
||||
temp_frame = enhance_face(target_face, temp_frame)
|
||||
if 'many' in facefusion.globals.face_selector_mode:
|
||||
many_faces = get_many_faces(temp_frame)
|
||||
if many_faces:
|
||||
for target_face in many_faces:
|
||||
temp_frame = enhance_face(target_face, temp_frame)
|
||||
return temp_frame
|
||||
|
||||
|
||||
def process_frames(source_path : str, temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
def process_frames(source_path : List[str], temp_frame_paths : List[str], 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, None, temp_frame)
|
||||
result_frame = process_frame(None, reference_faces, temp_frame)
|
||||
write_image(temp_frame_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, None, target_frame)
|
||||
result_frame = process_frame(None, reference_faces, target_frame)
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
def process_video(source_path : str, temp_frame_paths : List[str]) -> None:
|
||||
def process_video(source_paths : List[str], temp_frame_paths : List[str]) -> None:
|
||||
frame_processors.multi_process_frames(None, temp_frame_paths, process_frames)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from typing import Any, List, Dict, Literal, Optional
|
||||
from typing import Any, List, Literal, Optional
|
||||
from argparse import ArgumentParser
|
||||
import threading
|
||||
import numpy
|
||||
@@ -8,29 +8,31 @@ from onnx import numpy_helper
|
||||
|
||||
import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import wording
|
||||
from facefusion.face_analyser import get_one_face, get_many_faces, find_similar_faces, clear_face_analyser
|
||||
from facefusion import logger, wording
|
||||
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, paste_back
|
||||
from facefusion.face_reference import get_face_reference
|
||||
from facefusion.face_store import get_reference_faces
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Face, Frame, Update_Process, ProcessMode, ModelValue, OptionsWithModel, Embedding
|
||||
from facefusion.utilities import conditional_download, resolve_relative_path, is_image, is_video, is_file, is_download_done, update_status
|
||||
from facefusion.vision import read_image, read_static_image, write_image
|
||||
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.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 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
|
||||
THREAD_LOCK : threading.Lock = threading.Lock()
|
||||
NAME = 'FACEFUSION.FRAME_PROCESSOR.FACE_SWAPPER'
|
||||
MODELS : Dict[str, ModelValue] =\
|
||||
NAME = __name__.upper()
|
||||
MODELS : ModelSet =\
|
||||
{
|
||||
'blendswap_256':
|
||||
{
|
||||
'type': 'blendswap',
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/blendswap_256.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/blendswap_256.onnx'),
|
||||
'template': 'ffhq',
|
||||
'template': 'ffhq_512',
|
||||
'size': (512, 256),
|
||||
'mean': [ 0.0, 0.0, 0.0 ],
|
||||
'standard_deviation': [ 1.0, 1.0, 1.0 ]
|
||||
@@ -40,7 +42,7 @@ MODELS : Dict[str, ModelValue] =\
|
||||
'type': 'inswapper',
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/inswapper_128.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/inswapper_128.onnx'),
|
||||
'template': 'arcface_v2',
|
||||
'template': 'arcface_128_v2',
|
||||
'size': (128, 128),
|
||||
'mean': [ 0.0, 0.0, 0.0 ],
|
||||
'standard_deviation': [ 1.0, 1.0, 1.0 ]
|
||||
@@ -50,7 +52,7 @@ MODELS : Dict[str, ModelValue] =\
|
||||
'type': 'inswapper',
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/inswapper_128_fp16.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/inswapper_128_fp16.onnx'),
|
||||
'template': 'arcface_v2',
|
||||
'template': 'arcface_128_v2',
|
||||
'size': (128, 128),
|
||||
'mean': [ 0.0, 0.0, 0.0 ],
|
||||
'standard_deviation': [ 1.0, 1.0, 1.0 ]
|
||||
@@ -60,7 +62,7 @@ MODELS : Dict[str, ModelValue] =\
|
||||
'type': 'simswap',
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/simswap_256.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/simswap_256.onnx'),
|
||||
'template': 'arcface_v1',
|
||||
'template': 'arcface_112_v1',
|
||||
'size': (112, 256),
|
||||
'mean': [ 0.485, 0.456, 0.406 ],
|
||||
'standard_deviation': [ 0.229, 0.224, 0.225 ]
|
||||
@@ -70,7 +72,7 @@ MODELS : Dict[str, ModelValue] =\
|
||||
'type': 'simswap',
|
||||
'url': 'https://github.com/facefusion/facefusion-assets/releases/download/models/simswap_512_unofficial.onnx',
|
||||
'path': resolve_relative_path('../.assets/models/simswap_512_unofficial.onnx'),
|
||||
'template': 'arcface_v1',
|
||||
'template': 'arcface_112_v1',
|
||||
'size': (112, 512),
|
||||
'mean': [ 0.0, 0.0, 0.0 ],
|
||||
'standard_deviation': [ 1.0, 1.0, 1.0 ]
|
||||
@@ -130,7 +132,7 @@ def set_options(key : Literal['model'], value : Any) -> None:
|
||||
|
||||
|
||||
def register_args(program : ArgumentParser) -> None:
|
||||
program.add_argument('--face-swapper-model', help = wording.get('frame_processor_model_help'), dest = 'face_swapper_model', default = 'inswapper_128', choices = frame_processors_choices.face_swapper_models)
|
||||
program.add_argument('--face-swapper-model', help = wording.get('frame_processor_model_help'), default = 'inswapper_128', choices = frame_processors_choices.face_swapper_models)
|
||||
|
||||
|
||||
def apply_args(program : ArgumentParser) -> None:
|
||||
@@ -156,22 +158,23 @@ def pre_process(mode : ProcessMode) -> 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):
|
||||
update_status(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
elif not is_file(model_path):
|
||||
update_status(wording.get('model_file_not_present') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('model_file_not_present') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
if not is_image(facefusion.globals.source_path):
|
||||
update_status(wording.get('select_image_source') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
elif not get_one_face(read_static_image(facefusion.globals.source_path)):
|
||||
update_status(wording.get('no_source_face_detected') + wording.get('exclamation_mark'), NAME)
|
||||
if not are_images(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):
|
||||
if not get_one_face(source_frame):
|
||||
logger.error(wording.get('no_source_face_detected') + 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):
|
||||
update_status(wording.get('select_image_or_video_target') + wording.get('exclamation_mark'), NAME)
|
||||
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:
|
||||
update_status(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
return True
|
||||
|
||||
@@ -181,6 +184,8 @@ def post_process() -> None:
|
||||
clear_model_matrix()
|
||||
clear_face_analyser()
|
||||
clear_content_analyser()
|
||||
clear_face_occluder()
|
||||
clear_face_parser()
|
||||
read_static_image.cache_clear()
|
||||
|
||||
|
||||
@@ -190,6 +195,11 @@ def swap_face(source_face : Face, target_face : Face, temp_frame : Frame) -> Fra
|
||||
model_size = get_options('model').get('size')
|
||||
model_type = get_options('model').get('type')
|
||||
crop_frame, affine_matrix = warp_face(temp_frame, target_face.kps, 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))
|
||||
if 'occlusion' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_occlusion_mask(crop_frame))
|
||||
crop_frame = prepare_crop_frame(crop_frame)
|
||||
frame_processor_inputs = {}
|
||||
for frame_processor_input in frame_processor.get_inputs():
|
||||
@@ -202,13 +212,16 @@ def swap_face(source_face : Face, target_face : Face, temp_frame : Frame) -> Fra
|
||||
frame_processor_inputs[frame_processor_input.name] = crop_frame
|
||||
crop_frame = frame_processor.run(None, frame_processor_inputs)[0][0]
|
||||
crop_frame = normalize_crop_frame(crop_frame)
|
||||
temp_frame = paste_back(temp_frame, crop_frame, affine_matrix, facefusion.globals.face_mask_blur, facefusion.globals.face_mask_padding)
|
||||
if 'region' in facefusion.globals.face_mask_types:
|
||||
crop_mask_list.append(create_region_mask(crop_frame, facefusion.globals.face_mask_regions))
|
||||
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
|
||||
|
||||
|
||||
def prepare_source_frame(source_face : Face) -> numpy.ndarray[Any, Any]:
|
||||
source_frame = read_static_image(facefusion.globals.source_path)
|
||||
source_frame, _ = warp_face(source_frame, source_face.kps, 'arcface_v2', (112, 112))
|
||||
def prepare_source_frame(source_face : Face) -> Frame:
|
||||
source_frame = read_static_image(facefusion.globals.source_paths[0])
|
||||
source_frame, _ = warp_face(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)
|
||||
@@ -243,9 +256,13 @@ def normalize_crop_frame(crop_frame : Frame) -> Frame:
|
||||
return crop_frame
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_face : Face, temp_frame : Frame) -> 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 process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
if 'reference' in facefusion.globals.face_selector_mode:
|
||||
similar_faces = find_similar_faces(temp_frame, reference_face, facefusion.globals.reference_face_distance)
|
||||
similar_faces = find_similar_faces(temp_frame, reference_faces, facefusion.globals.reference_face_distance)
|
||||
if similar_faces:
|
||||
for similar_face in similar_faces:
|
||||
temp_frame = swap_face(source_face, similar_face, temp_frame)
|
||||
@@ -261,23 +278,25 @@ def process_frame(source_face : Face, reference_face : Face, temp_frame : Frame)
|
||||
return temp_frame
|
||||
|
||||
|
||||
def process_frames(source_path : str, temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
source_face = get_one_face(read_static_image(source_path))
|
||||
reference_face = get_face_reference() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
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)
|
||||
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_face, temp_frame)
|
||||
result_frame = process_frame(source_face, reference_faces, temp_frame)
|
||||
write_image(temp_frame_path, result_frame)
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_path : str, target_path : str, output_path : str) -> None:
|
||||
source_face = get_one_face(read_static_image(source_path))
|
||||
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)
|
||||
reference_faces = get_reference_faces() if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
target_frame = read_static_image(target_path)
|
||||
reference_face = get_one_face(target_frame, facefusion.globals.reference_face_position) if 'reference' in facefusion.globals.face_selector_mode else None
|
||||
result_frame = process_frame(source_face, reference_face, target_frame)
|
||||
result_frame = process_frame(source_face, reference_faces, target_frame)
|
||||
write_image(output_path, result_frame)
|
||||
|
||||
|
||||
def process_video(source_path : str, temp_frame_paths : List[str]) -> None:
|
||||
frame_processors.multi_process_frames(source_path, temp_frame_paths, process_frames)
|
||||
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,4 +1,4 @@
|
||||
from typing import Any, List, Dict, Literal, Optional
|
||||
from typing import Any, List, Literal, Optional
|
||||
from argparse import ArgumentParser
|
||||
import threading
|
||||
import cv2
|
||||
@@ -7,11 +7,14 @@ from realesrgan import RealESRGANer
|
||||
|
||||
import facefusion.globals
|
||||
import facefusion.processors.frame.core as frame_processors
|
||||
from facefusion import wording
|
||||
from facefusion import logger, wording
|
||||
from facefusion.face_analyser import clear_face_analyser
|
||||
from facefusion.content_analyser import clear_content_analyser
|
||||
from facefusion.typing import Frame, Face, Update_Process, ProcessMode, ModelValue, OptionsWithModel
|
||||
from facefusion.utilities import conditional_download, resolve_relative_path, is_file, is_download_done, map_device, create_metavar, update_status
|
||||
from facefusion.typing import Face, FaceSet, Frame, Update_Process, ProcessMode, ModelSet, OptionsWithModel
|
||||
from facefusion.cli_helper import create_metavar
|
||||
from facefusion.execution_helper import map_device
|
||||
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 import globals as frame_processors_globals
|
||||
from facefusion.processors.frame import choices as frame_processors_choices
|
||||
@@ -19,8 +22,8 @@ from facefusion.processors.frame import choices as frame_processors_choices
|
||||
FRAME_PROCESSOR = None
|
||||
THREAD_SEMAPHORE : threading.Semaphore = threading.Semaphore()
|
||||
THREAD_LOCK : threading.Lock = threading.Lock()
|
||||
NAME = 'FACEFUSION.FRAME_PROCESSOR.FRAME_ENHANCER'
|
||||
MODELS: Dict[str, ModelValue] =\
|
||||
NAME = __name__.upper()
|
||||
MODELS : ModelSet =\
|
||||
{
|
||||
'real_esrgan_x2plus':
|
||||
{
|
||||
@@ -88,8 +91,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'), dest = 'frame_enhancer_model', default = 'real_esrgan_x2plus', choices = frame_processors_choices.frame_enhancer_models)
|
||||
program.add_argument('--frame-enhancer-blend', help = wording.get('frame_processor_blend_help'), dest = 'frame_enhancer_blend', type = int, default = 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('frame_processor_model_help'), default = '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 = 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:
|
||||
@@ -110,13 +113,13 @@ def pre_process(mode : ProcessMode) -> 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):
|
||||
update_status(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('model_download_not_done') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
elif not is_file(model_path):
|
||||
update_status(wording.get('model_file_not_present') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('model_file_not_present') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
if mode == 'output' and not facefusion.globals.output_path:
|
||||
update_status(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
logger.error(wording.get('select_file_or_directory_output') + wording.get('exclamation_mark'), NAME)
|
||||
return False
|
||||
return True
|
||||
|
||||
@@ -143,11 +146,15 @@ def blend_frame(temp_frame : Frame, paste_frame : Frame) -> Frame:
|
||||
return temp_frame
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_face : Face, temp_frame : Frame) -> Frame:
|
||||
def get_reference_frame(source_face : Face, target_face : Face, temp_frame : Frame) -> Frame:
|
||||
pass
|
||||
|
||||
|
||||
def process_frame(source_face : Face, reference_faces : FaceSet, temp_frame : Frame) -> Frame:
|
||||
return enhance_frame(temp_frame)
|
||||
|
||||
|
||||
def process_frames(source_path : str, temp_frame_paths : List[str], update_progress : Update_Process) -> None:
|
||||
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)
|
||||
@@ -155,11 +162,11 @@ def process_frames(source_path : str, temp_frame_paths : List[str], update_progr
|
||||
update_progress()
|
||||
|
||||
|
||||
def process_image(source_path : str, target_path : str, output_path : str) -> None:
|
||||
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)
|
||||
|
||||
|
||||
def process_video(source_path : str, temp_frame_paths : List[str]) -> None:
|
||||
def process_video(source_paths : List[str], temp_frame_paths : List[str]) -> None:
|
||||
frame_processors.multi_process_frames(None, temp_frame_paths, process_frames)
|
||||
|
||||
Reference in New Issue
Block a user