File size: 2,861 Bytes
3284138 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator
from facefusion.filesystem import is_image
from facefusion.processors.core import get_processors_modules
from facefusion.temp_helper import get_temp_file_path
from facefusion.time_helper import calculate_end_time
from facefusion.types import ErrorCode
from facefusion.vision import detect_image_resolution, pack_resolution, read_static_image, restrict_image_resolution, scale_resolution, write_image
from facefusion.workflows.core import conditional_get_target_vision_frames, is_process_stopping, process_temp_frame
def analyse_image() -> ErrorCode:
if content_analyser.analyse_image(state_manager.get_item('target_path')):
return 3
return 0
def prepare_image() -> ErrorCode:
output_image_resolution = scale_resolution(detect_image_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_image_scale'))
temp_image_resolution = restrict_image_resolution(state_manager.get_item('target_path'), output_image_resolution)
logger.info(translator.get('copying_image').format(resolution = pack_resolution(temp_image_resolution)), __name__)
if ffmpeg.copy_image(state_manager.get_item('target_path'), temp_image_resolution):
logger.debug(translator.get('copying_image_succeeded'), __name__)
else:
logger.error(translator.get('copying_image_failed'), __name__)
process_manager.end()
return 1
return 0
def process_image() -> ErrorCode:
temp_image_path = get_temp_file_path(state_manager.get_item('target_path'))
target_vision_frames = conditional_get_target_vision_frames(0)
temp_vision_frame = read_static_image(temp_image_path, 'rgba')
temp_vision_frame = process_temp_frame(target_vision_frames, temp_vision_frame, 0)
write_image(temp_image_path, temp_vision_frame)
for processor_module in get_processors_modules(state_manager.get_item('processors')):
processor_module.post_process()
if is_process_stopping():
return 4
return 0
def finalize_image(start_time : float) -> ErrorCode:
output_image_resolution = scale_resolution(detect_image_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_image_scale'))
logger.info(translator.get('finalizing_image').format(resolution = pack_resolution(output_image_resolution)), __name__)
if ffmpeg.finalize_image(state_manager.get_item('target_path'), state_manager.get_item('output_path'), output_image_resolution):
logger.debug(translator.get('finalizing_image_succeeded'), __name__)
else:
logger.warn(translator.get('finalizing_image_skipped'), __name__)
if is_image(state_manager.get_item('output_path')):
logger.info(translator.get('processing_image_succeeded').format(seconds = calculate_end_time(start_time)), __name__)
else:
logger.error(translator.get('processing_image_failed'), __name__)
return 1
return 0
|