From 808460ae09025ffac5e709f67e82a39e62543e0e Mon Sep 17 00:00:00 2001 From: Harisreedhar <46858047+harisreedhar@users.noreply.github.com> Date: Wed, 31 Jan 2024 22:07:04 +0530 Subject: [PATCH] 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 --- facefusion/common_helper.py | 4 + facefusion/core.py | 18 +- facefusion/processors/frame/choices.py | 3 +- facefusion/processors/frame/globals.py | 3 +- .../processors/frame/modules/face_swapper.py | 7 +- .../processors/frame/modules/lip_sync.py | 227 ++++++++++++++++++ facefusion/processors/frame/typings.py | 1 + facefusion/wording.py | 3 + tests/test_cli.py | 17 ++ 9 files changed, 271 insertions(+), 12 deletions(-) create mode 100755 facefusion/processors/frame/modules/lip_sync.py diff --git a/facefusion/common_helper.py b/facefusion/common_helper.py index 5e258511..ea6a028e 100644 --- a/facefusion/common_helper.py +++ b/facefusion/common_helper.py @@ -12,3 +12,7 @@ def create_int_range(start : int, stop : int, step : int) -> List[int]: def create_float_range(start : float, stop : float, step : float) -> List[float]: return (numpy.around(numpy.arange(start, stop + step, step), decimals = 2)).tolist() + + +def get_first_item(__list__ : Any) -> Any: + return next(iter(__list__), None) diff --git a/facefusion/core.py b/facefusion/core.py index 4f4bc0cb..ec8c6aa9 100755 --- a/facefusion/core.py +++ b/facefusion/core.py @@ -19,12 +19,12 @@ from facefusion.face_store import get_reference_faces, append_reference_face from facefusion import face_analyser, face_masker, content_analyser, config, metadata, logger, wording from facefusion.content_analyser import analyse_image, analyse_video from facefusion.processors.frame.core import get_frame_processors_modules, load_frame_processor_module -from facefusion.common_helper import create_metavar +from facefusion.common_helper import create_metavar, get_first_item from facefusion.execution_helper import encode_execution_providers, decode_execution_providers from facefusion.normalizer import normalize_output_path, normalize_padding, normalize_fps from facefusion.memory import limit_system_memory -from facefusion.filesystem import list_directory, get_temp_frame_paths, create_temp, move_temp, clear_temp, is_image, is_video -from facefusion.ffmpeg import extract_frames, compress_image, merge_video, restore_audio +from facefusion.filesystem import list_directory, get_temp_frame_paths, create_temp, move_temp, clear_temp, is_image, is_video, filter_audio_paths +from facefusion.ffmpeg import extract_frames, compress_image, merge_video, restore_audio, replace_audio from facefusion.vision import get_video_frame, read_image, read_static_images, pack_resolution, detect_video_resolution, detect_video_fps, create_video_resolutions onnxruntime.set_default_logger_severity(3) @@ -296,11 +296,15 @@ def process_video(start_time : float) -> None: logger.info(wording.get('skipping_audio'), __name__.upper()) move_temp(facefusion.globals.target_path, facefusion.globals.output_path) else: - if restore_audio(facefusion.globals.target_path, facefusion.globals.output_path, facefusion.globals.output_video_fps): - logger.info(wording.get('restoring_audio_succeed'), __name__.upper()) + if 'lip_sync' in facefusion.globals.frame_processors: + audio_path = get_first_item(filter_audio_paths(facefusion.globals.source_paths)) + if not audio_path or not replace_audio(facefusion.globals.target_path, audio_path, facefusion.globals.output_path): + logger.warn(wording.get('restoring_audio_skipped'), __name__.upper()) + move_temp(facefusion.globals.target_path, facefusion.globals.output_path) else: - logger.warn(wording.get('restoring_audio_skipped'), __name__.upper()) - move_temp(facefusion.globals.target_path, facefusion.globals.output_path) + if not restore_audio(facefusion.globals.target_path, facefusion.globals.output_path, facefusion.globals.output_video_fps): + logger.warn(wording.get('restoring_audio_skipped'), __name__.upper()) + move_temp(facefusion.globals.target_path, facefusion.globals.output_path) # clear temp logger.debug(wording.get('clearing_temp'), __name__.upper()) clear_temp(facefusion.globals.target_path) diff --git a/facefusion/processors/frame/choices.py b/facefusion/processors/frame/choices.py index 50261ec3..cdd3d7b4 100755 --- a/facefusion/processors/frame/choices.py +++ b/facefusion/processors/frame/choices.py @@ -1,12 +1,13 @@ from typing import List from facefusion.common_helper import create_int_range -from facefusion.processors.frame.typings import FaceDebuggerItem, FaceEnhancerModel, FaceSwapperModel, FrameEnhancerModel +from facefusion.processors.frame.typings import FaceDebuggerItem, FaceEnhancerModel, FaceSwapperModel, FrameEnhancerModel, LipSyncModel face_debugger_items : List[FaceDebuggerItem] = [ 'bbox', 'kps', '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' ] frame_enhancer_models : List[FrameEnhancerModel] = [ 'real_esrgan_x2plus', 'real_esrgan_x4plus', 'real_esrnet_x4plus' ] +lip_sync_models : List[LipSyncModel] = [ 'wav2lip' ] face_enhancer_blend_range : List[int] = create_int_range(0, 100, 1) frame_enhancer_blend_range : List[int] = create_int_range(0, 100, 1) diff --git a/facefusion/processors/frame/globals.py b/facefusion/processors/frame/globals.py index 526b8573..4290bd3b 100755 --- a/facefusion/processors/frame/globals.py +++ b/facefusion/processors/frame/globals.py @@ -1,9 +1,10 @@ from typing import List, Optional -from facefusion.processors.frame.typings import FaceSwapperModel, FaceEnhancerModel, FrameEnhancerModel, FaceDebuggerItem +from facefusion.processors.frame.typings import FaceSwapperModel, FaceEnhancerModel, FrameEnhancerModel, FaceDebuggerItem, LipSyncModel face_swapper_model : Optional[FaceSwapperModel] = None face_enhancer_model : Optional[FaceEnhancerModel] = None +lip_sync_model : Optional[LipSyncModel] = None face_enhancer_blend : Optional[int] = None frame_enhancer_model : Optional[FrameEnhancerModel] = None frame_enhancer_blend : Optional[int] = None diff --git a/facefusion/processors/frame/modules/face_swapper.py b/facefusion/processors/frame/modules/face_swapper.py index 73e69c7b..6d1a2e01 100755 --- a/facefusion/processors/frame/modules/face_swapper.py +++ b/facefusion/processors/frame/modules/face_swapper.py @@ -16,7 +16,7 @@ from facefusion.face_helper import warp_face_by_kps, paste_back from facefusion.face_store import get_reference_faces from facefusion.content_analyser import clear_content_analyser from facefusion.typing import Face, FaceSet, VisionFrame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, Embedding -from facefusion.filesystem import is_file, is_image, are_images, is_video, resolve_relative_path +from facefusion.filesystem import is_file, is_image, is_video, resolve_relative_path, filter_image_paths 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 @@ -173,10 +173,11 @@ def post_check() -> bool: def pre_process(mode : ProcessMode) -> bool: - if not are_images(facefusion.globals.source_paths): + source_images = filter_image_paths(facefusion.globals.source_paths) + if not source_images: 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): + for source_frame in read_static_images(source_images): if not get_one_face(source_frame): logger.error(wording.get('no_source_face_detected') + wording.get('exclamation_mark'), NAME) return False diff --git a/facefusion/processors/frame/modules/lip_sync.py b/facefusion/processors/frame/modules/lip_sync.py new file mode 100755 index 00000000..1cf83cf0 --- /dev/null +++ b/facefusion/processors/frame/modules/lip_sync.py @@ -0,0 +1,227 @@ +import os +from typing import Any, List, Literal, Optional +from argparse import ArgumentParser +import threading +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_helper import paste_back, warp_face_by_kps, warp_face_by_bbox +from facefusion.face_store import get_reference_faces +from facefusion.content_analyser import clear_content_analyser +from facefusion.typing import Face, FaceSet, VisionFrame, Update_Process, ProcessMode, ModelSet, OptionsWithModel, AudioFrame +from facefusion.filesystem import is_file, 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_video, filter_audio_paths +from facefusion.vision import read_image, write_image, detect_video_fps, read_static_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, create_region_mask, clear_face_parser +from facefusion.common_helper import get_first_item + +FRAME_PROCESSOR = None +MODEL_MATRIX = None +THREAD_LOCK : threading.Lock = threading.Lock() +NAME = __name__.upper() +MODELS : ModelSet =\ +{ + 'wav2lip': + { + 'url': 'https://huggingface.co/bluefoxcreation/Wav2lip-Onnx/resolve/main/wav2lip_gan.onnx?download=true', + '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_sync_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-sync-model', help = wording.get('help.lip_sync_model_help'), default = config.get_str_value('frame_processors.lip_sync_model', 'wav2lip'), choices = frame_processors_choices.lip_sync_models) + + +def apply_args(program : ArgumentParser) -> None: + args = program.parse_args() + frame_processors_globals.lip_sync_model = args.lip_sync_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: + audio_path = get_first_item(filter_audio_paths(facefusion.globals.source_paths)) + if not audio_path: + logger.error(wording.get('select_audio_source') + wording.get('exclamation_mark'), NAME) + return False + if mode in [ 'output', 'preview' ] and not is_video(facefusion.globals.target_path): + logger.error(wording.get('select_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 lip_sync(audio_frame : AudioFrame, target_face : Face, temp_frame : VisionFrame) -> VisionFrame: + frame_processor = get_frame_processor() + crop_frame, affine_matrix = warp_face_by_bbox(temp_frame, target_face.bbox, (96, 96)) + audio_frame = prepare_audio_frame(audio_frame) + crop_frame = prepare_crop_frame(crop_frame) + crop_frame = frame_processor.run(None, {'vid' : crop_frame, 'mel' : audio_frame})[0] + crop_frame = normalize_crop_frame(crop_frame) + crop_mask = create_static_box_mask(crop_frame.shape[:2][::-1], 0.1, (50, 0, 0, 0)) + paste_frame = paste_back(temp_frame, crop_frame, crop_mask, affine_matrix) + crop_mask_list = [] + if 'occlusion' in facefusion.globals.face_mask_types: + temp_crop_frame, affine_matrix = warp_face_by_kps(temp_frame, target_face.kps, "ffhq_512", (512, 512)) + crop_mask_list.append(create_occlusion_mask(temp_crop_frame)) + if 'region' in facefusion.globals.face_mask_types: + paste_crop_frame, affine_matrix = warp_face_by_kps(paste_frame, target_face.kps, "ffhq_512", (512, 512)) + crop_mask_list.append(create_region_mask(paste_crop_frame, facefusion.globals.face_mask_regions)) + if crop_mask_list: + crop_mask = numpy.minimum.reduce(crop_mask_list) + paste_frame = paste_back(temp_frame, crop_frame, crop_mask, affine_matrix) + return paste_frame + + +def prepare_audio_frame(audio_frame : AudioFrame) -> AudioFrame: + audio_frame = numpy.maximum(numpy.exp(-5 * numpy.log(10)), audio_frame) + audio_frame = numpy.log10(audio_frame) * 1.6 + 3.2 + audio_frame = audio_frame.clip(-4, 4).astype(numpy.float32) + audio_frame = numpy.expand_dims(audio_frame, axis = (0, 1)) + return audio_frame + + +def prepare_crop_frame(crop_frame : VisionFrame) -> VisionFrame: + crop_frame = numpy.expand_dims(crop_frame, axis = 0) + crop_frame_masked = crop_frame.copy() + crop_frame_masked[:, 48:] = 0 + crop_frame_stack = numpy.concatenate((crop_frame_masked, crop_frame), axis = 3) + crop_frame_stack = crop_frame_stack.transpose(0, 3, 1, 2).astype('float32') / 255.0 + return crop_frame_stack + + +def normalize_crop_frame(crop_frame : VisionFrame) -> VisionFrame: + crop_frame = crop_frame[0].transpose(1, 2, 0) + crop_frame = crop_frame.clip(0, 1) * 255 + crop_frame = crop_frame.astype(numpy.uint8) + return crop_frame + + +def get_reference_frame(source_face : Face, target_face : Face, temp_frame : VisionFrame) -> VisionFrame: + audio_path = get_first_item(filter_audio_paths(facefusion.globals.source_paths)) + audio_frame = get_audio_frame(audio_path, detect_video_fps(facefusion.globals.target_path), facefusion.globals.reference_frame_number) + if audio_frame is not None: + return lip_sync(audio_frame, target_face, temp_frame) + return temp_frame + + +def process_frame(audio_frame : AudioFrame, reference_faces : FaceSet, temp_frame : VisionFrame) -> VisionFrame: + 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 = lip_sync(audio_frame, similar_face, temp_frame) + if 'one' in facefusion.globals.face_selector_mode: + target_face = get_one_face(temp_frame) + if target_face: + temp_frame = lip_sync(audio_frame, 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 = lip_sync(audio_frame, target_face, temp_frame) + return temp_frame + + +def process_frames(source_paths : 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 + source_audio_path = get_first_item(filter_audio_paths(source_paths)) + video_fps = detect_video_fps(facefusion.globals.target_path) + for temp_frame_path in temp_frame_paths: + frame_number = int(os.path.basename(temp_frame_path).split(".")[0]) + audio_frame = get_audio_frame(source_audio_path, video_fps, frame_number) + if audio_frame is not None: + temp_frame = read_image(temp_frame_path) + result_frame = process_frame(audio_frame, reference_faces, temp_frame) + write_image(temp_frame_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_item(filter_audio_paths(source_paths)) + audio_frame = get_audio_frame(source_audio_path, 25, 0) + if audio_frame is not None: + target_frame = read_static_image(target_path) + result_frame = process_frame(audio_frame, reference_faces, target_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) diff --git a/facefusion/processors/frame/typings.py b/facefusion/processors/frame/typings.py index 95bdd2bf..90074c12 100644 --- a/facefusion/processors/frame/typings.py +++ b/facefusion/processors/frame/typings.py @@ -4,3 +4,4 @@ FaceDebuggerItem = Literal['bbox', 'kps', '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'] FrameEnhancerModel = Literal['real_esrgan_x2plus', 'real_esrgan_x4plus', 'real_esrnet_x4plus'] +LipSyncModel = Literal['wav2lip'] diff --git a/facefusion/wording.py b/facefusion/wording.py index 561aa0da..bf80300a 100755 --- a/facefusion/wording.py +++ b/facefusion/wording.py @@ -25,6 +25,8 @@ WORDING : Dict[str, Any] =\ 'model_download_not_done': 'Download of the model is not done', 'model_file_not_present': 'File of the model is not present', 'select_image_source': 'Select an image for source path', + 'select_audio_source': 'Select an audio for source path', + 'select_video_target': 'Select a video for target path', 'select_image_or_video_target': 'Select an image or video for target path', 'select_file_or_directory_output': 'Select an file or directory for output path', 'no_source_face_detected': 'No source face detected', @@ -95,6 +97,7 @@ WORDING : Dict[str, Any] =\ 'face_enhancer_model': 'choose the model responsible for enhancing the face', 'face_enhancer_blend': 'blend the enhanced into the previous face', 'face_swapper_model': 'choose the model responsible for swapping the face', + 'lip_sync_model': 'choose the model responsible for lip syncing', 'frame_enhancer_model': 'choose the model responsible for enhancing the frame', 'frame_enhancer_blend': 'blend the enhanced into the previous frame', # uis diff --git a/tests/test_cli.py b/tests/test_cli.py index cad4ffbb..d56f63eb 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -10,6 +10,7 @@ def before_all() -> None: conditional_download('.assets/examples', [ 'https://github.com/facefusion/facefusion-assets/releases/download/examples/source.jpg', + 'https://github.com/facefusion/facefusion-assets/releases/download/examples/source.mp3', 'https://github.com/facefusion/facefusion-assets/releases/download/examples/target-1080p.mp4' ]) subprocess.run([ 'ffmpeg', '-i', '.assets/examples/target-1080p.mp4', '-vframes', '1', '.assets/examples/target-1080p.jpg' ]) @@ -29,3 +30,19 @@ def test_image_to_video() -> None: assert run.returncode == 0 assert 'video succeed' in run.stdout.decode() + + +def test_audio_to_video() -> None: + commands = [ sys.executable, 'run.py', '--frame-processors', 'lip_sync', '-s', '.assets/examples/source.mp3', '-t', '.assets/examples/target-1080p.mp4', '-o', '.assets/examples', '--trim-frame-end', '10', '--headless' ] + run = subprocess.run(commands, stdout = subprocess.PIPE, stderr = subprocess.STDOUT) + + assert run.returncode == 0 + assert 'video succeed' in run.stdout.decode() + + +def test_image_and_audio_to_video() -> None: + commands = [ sys.executable, 'run.py', '--frame-processors', 'lip_sync', 'face_swapper', '-s', '.assets/examples/source.mp3', '-s', '.assets/examples/source.jpg', '-t', '.assets/examples/target-1080p.mp4', '-o', '.assets/examples', '--trim-frame-end', '10', '--headless' ] + run = subprocess.run(commands, stdout = subprocess.PIPE, stderr = subprocess.STDOUT) + + assert run.returncode == 0 + assert 'video succeed' in run.stdout.decode()