mirror of
https://github.com/facefusion/facefusion.git
synced 2026-04-27 11:56:14 +02:00
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
This commit is contained in:
Harisreedhar
@@ -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)
|
||||
|
||||
+11
-7
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
+227
@@ -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)
|
||||
@@ -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']
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user