image to video as sequence

facefusion/choices.py

@@ -45,7 +45,7 @@ face_mask_regions : List[FaceMaskRegion] = list(get_args(FaceMaskRegion))
 
 voice_extractor_models : List[VoiceExtractorModel] = list(get_args(VoiceExtractorModel))
 
-workflows : List[WorkFlow] = [ 'auto', 'audio-to-image', 'image-to-image', 'image-to-video' ]
+workflows : List[WorkFlow] = [ 'auto', 'audio-to-image', 'image-to-image', 'image-to-video', 'image-to-video-as-sequence' ]
 
 audio_type_set : AudioTypeSet =\
 {

facefusion/core.py

@@ -12,15 +12,15 @@ from facefusion.apis.core import create_api
 from facefusion.args_helper import apply_args
 from facefusion.download import conditional_download_hashes, conditional_download_sources
 from facefusion.exit_helper import hard_exit, signal_exit
-from facefusion.filesystem import get_file_extension, get_file_name, resolve_file_paths, resolve_file_pattern
-from facefusion.filesystem import has_audio, has_image, has_video
+from facefusion.filesystem import get_file_extension, has_audio, has_image, has_video
+from facefusion.filesystem import get_file_name, resolve_file_paths, resolve_file_pattern
 from facefusion.jobs import job_helper, job_manager, job_runner
 from facefusion.jobs.job_list import compose_job_list
 from facefusion.processors.core import get_processors_modules
 from facefusion.program import create_program
 from facefusion.program_helper import validate_args
 from facefusion.types import Args, ErrorCode, WorkFlow
-from facefusion.workflows import audio_to_image, image_to_image, image_to_video
+from facefusion.workflows import audio_to_image, image_to_image, image_to_video, image_to_video_as_sequence
 
 
 def cli() -> None:
@@ -342,13 +342,17 @@ def conditional_process() -> ErrorCode:
 		return image_to_image.process(start_time)
 	if state_manager.get_item('workflow') == 'image-to-video':
 		return image_to_video.process(start_time)
+	if state_manager.get_item('workflow') == 'image-to-video-as-sequence':
+		return image_to_video_as_sequence.process(start_time)
 
 	return 0
 
 
 def detect_workflow() -> WorkFlow:
 	if has_video([ state_manager.get_item('target_path') ]):
-		return 'image-to-video'
+		if get_file_extension(state_manager.get_item('output_path')):
+			return 'image-to-video'
+		return 'image-to-video-as-sequence'
 
 	if has_audio(state_manager.get_item('source_paths')) and has_image([ state_manager.get_item('target_path') ]):
 		return 'audio-to-image'

facefusion/filesystem.py

@@ -170,6 +170,13 @@ def create_directory(directory_path : str) -> bool:
 	return False
 
 
+def move_directory(directory_path : str, move_path : str) -> bool:
+	if is_directory(directory_path):
+		shutil.move(directory_path, move_path)
+		return is_directory(move_path)
+	return False
+
+
 def remove_directory(directory_path : str) -> bool:
 	if is_directory(directory_path):
 		shutil.rmtree(directory_path, ignore_errors = True)

facefusion/jobs/job_helper.py

@@ -13,6 +13,8 @@ def get_step_output_path(job_id : str, step_index : int, output_path : str) -> O
 
 		if output_file_name and output_file_extension:
 			return os.path.join(output_directory_path, output_file_name + '-' + job_id + '-' + str(step_index) + output_file_extension)
+		if output_file_path and output_directory_path:
+			return os.path.join(output_directory_path, output_file_path + '-' + job_id + '-' + str(step_index))
 	return None
 
 

facefusion/jobs/job_runner.py

@@ -1,5 +1,7 @@
+import os
+
 from facefusion.ffmpeg import concat_video
-from facefusion.filesystem import are_images, are_videos, move_file, remove_file
+from facefusion.filesystem import are_images, are_videos, copy_file, create_directory, is_directory, is_file, move_directory, move_file, remove_directory, remove_file, resolve_file_paths
 from facefusion.jobs import job_helper, job_manager
 from facefusion.types import JobOutputSet, JobStep, ProcessStep
 
@@ -59,6 +61,8 @@ def run_step(job_id : str, step_index : int, step : JobStep, process_step : Proc
 		output_path = step_args.get('output_path')
 		step_output_path = job_helper.get_step_output_path(job_id, step_index, output_path)
 
+		if is_directory(output_path):
+			return move_directory(output_path, step_output_path) and job_manager.set_step_status(job_id, step_index, 'completed')
 		return move_file(output_path, step_output_path) and job_manager.set_step_status(job_id, step_index, 'completed')
 	job_manager.set_step_status(job_id, step_index, 'failed')
 	return False
@@ -79,13 +83,26 @@ def finalize_steps(job_id : str) -> bool:
 	output_set = collect_output_set(job_id)
 
 	for output_path, temp_output_paths in output_set.items():
-		if are_videos(temp_output_paths):
+		has_videos = are_videos(temp_output_paths)
+		has_images = are_images(temp_output_paths)
+
+		if has_videos:
 			if not concat_video(output_path, temp_output_paths):
 				return False
-		if are_images(temp_output_paths):
+		if not has_videos and has_images:
 			for temp_output_path in temp_output_paths:
 				if not move_file(temp_output_path, output_path):
 					return False
+		if not has_videos and not has_images:
+			if not create_directory(output_path):
+				return False
+
+			for temp_output_path in temp_output_paths:
+				if is_directory(temp_output_path):
+					temp_frame_paths = resolve_file_paths(temp_output_path)
+					for temp_frame_path in temp_frame_paths:
+						if not copy_file(temp_frame_path, os.path.join(output_path, os.path.basename(temp_frame_path))):
+							return False
 	return True
 
 
@@ -94,8 +111,12 @@ def clean_steps(job_id: str) -> bool:
 
 	for temp_output_paths in output_set.values():
 		for temp_output_path in temp_output_paths:
-			if not remove_file(temp_output_path):
-				return False
+			if is_file(temp_output_path):
+				if not remove_file(temp_output_path):
+					return False
+			if is_directory(temp_output_path):
+				if not remove_directory(temp_output_path):
+					return False
 	return True
 
 

facefusion/locales.py

@@ -40,6 +40,8 @@ LOCALES : Locales =\
 		'processing_stopped': 'processing stopped',
 		'processing_image_succeeded': 'processing to image succeeded in {seconds} seconds',
 		'processing_image_failed': 'processing to image failed',
+		'processing_sequence_succeeded': 'processing to sequence succeeded in {seconds} seconds',
+		'processing_sequence_failed': 'processing to sequence failed',
 		'processing_video_succeeded': 'processing to video succeeded in {seconds} seconds',
 		'processing_video_failed': 'processing to video failed',
 		'choose_image_source': 'choose an image for the source',

facefusion/processors/modules/age_modifier/core.py

@@ -145,9 +145,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/background_remover/core.py

@@ -441,9 +441,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/deep_swapper/core.py

@@ -299,9 +299,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/expression_restorer/core.py

@@ -125,9 +125,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/face_debugger/core.py

@@ -46,9 +46,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/face_editor/core.py

@@ -176,9 +176,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/face_enhancer/core.py

@@ -315,9 +315,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/face_swapper/core.py

@@ -551,9 +551,10 @@ def pre_process(mode : ProcessMode) -> bool:
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
 
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 
 	return True
 

facefusion/processors/modules/frame_colorizer/core.py

@@ -207,9 +207,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/processors/modules/frame_enhancer/core.py

@@ -594,9 +594,10 @@ def pre_process(mode : ProcessMode) -> bool:
 	if mode in [ 'output', 'preview' ] and not is_image(state_manager.get_item('target_path')) and not is_video(state_manager.get_item('target_path')):
 		logger.error(translator.get('choose_image_or_video_target') + translator.get('exclamation_mark'), __name__)
 		return False
-	if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
-		logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
-		return False
+	if state_manager.get_item('workflow') in [ 'audio-to-image', 'image-to-image', 'image-to-video' ]:
+		if mode == 'output' and not in_directory(state_manager.get_item('output_path')):
+			logger.error(translator.get('specify_image_or_video_output') + translator.get('exclamation_mark'), __name__)
+			return False
 	return True
 
 

facefusion/types.py

@@ -55,7 +55,7 @@ Language = Literal['en']
 Locales : TypeAlias = Dict[Language, Dict[str, Any]]
 LocalePoolSet : TypeAlias = Dict[str, Locales]
 
-WorkFlow = Literal['auto', 'audio-to-image', 'image-to-image', 'image-to-video']
+WorkFlow = Literal['auto', 'audio-to-image', 'image-to-image', 'image-to-video', 'image-to-video-as-sequence']
 
 VideoCaptureSet : TypeAlias = Dict[str, cv2.VideoCapture]
 VideoWriterSet : TypeAlias = Dict[str, cv2.VideoWriter]

facefusion/workflows/audio_to_image.py

@@ -6,7 +6,8 @@ from facefusion.common_helper import get_first
 from facefusion.filesystem import filter_audio_paths
 from facefusion.types import ErrorCode
 from facefusion.vision import detect_image_resolution, restrict_image_resolution, scale_resolution
-from facefusion.workflows.core import analyse_image, clear, finalize_video, is_process_stopping, merge_frames, process_video, restore_audio, setup
+from facefusion.workflows.core import analyse_image, clear, is_process_stopping, process_frames, setup
+from facefusion.workflows.to_video import finalize_video, merge_frames, restore_audio
 
 
 def process(start_time : float) -> ErrorCode:
@@ -16,7 +17,7 @@ def process(start_time : float) -> ErrorCode:
 		clear,
 		setup,
 		create_temp_frames,
-		process_video,
+		process_frames,
 		merge_frames,
 		restore_audio,
 		partial(finalize_video, start_time),

facefusion/workflows/core.py

@@ -3,16 +3,14 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 import numpy
 from tqdm import tqdm
 
-from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator, video_manager
+from facefusion import content_analyser, logger, process_manager, state_manager, translator
 from facefusion.audio import create_empty_audio_frame, get_audio_frame, get_voice_frame
 from facefusion.common_helper import get_first
-from facefusion.filesystem import filter_audio_paths, is_video
-from facefusion.media_helper import restrict_trim_frame
+from facefusion.filesystem import filter_audio_paths
 from facefusion.processors.core import get_processors_modules
-from facefusion.temp_helper import clear_temp_directory, create_temp_directory, move_temp_file, resolve_temp_frame_paths
-from facefusion.time_helper import calculate_end_time
-from facefusion.types import AudioFrame, ErrorCode, Fps, Resolution, VisionFrame
-from facefusion.vision import conditional_merge_vision_mask, detect_image_resolution, detect_video_resolution, extract_vision_mask, pack_resolution, read_static_image, read_static_images, read_static_video_frame, restrict_video_fps, scale_resolution, write_image
+from facefusion.temp_helper import clear_temp_directory, create_temp_directory, resolve_temp_frame_paths
+from facefusion.types import AudioFrame, ErrorCode, VisionFrame
+from facefusion.vision import conditional_merge_vision_mask, extract_vision_mask, read_static_image, read_static_images, read_static_video_frame, restrict_video_fps, write_image
 
 
 def is_process_stopping() -> bool:
@@ -71,28 +69,11 @@ def conditional_get_source_voice_frame(frame_number: int) -> AudioFrame:
 
 
 def conditional_get_reference_vision_frame() -> VisionFrame:
-	if state_manager.get_item('workflow') == 'image-to-video':
+	if state_manager.get_item('workflow') in [ 'image-to-video', 'image-to-video-as-sequence' ]:
 		return read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
 	return read_static_image(state_manager.get_item('target_path'))
 
 
-def conditional_scale_resolution() -> Resolution:
-	if state_manager.get_item('workflow') == 'image-to-video':
-		return scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
-	return scale_resolution(detect_image_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
-
-
-def conditional_restrict_video_fps() -> Fps:
-	if state_manager.get_item('workflow') == 'image-to-video':
-		return restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
-	return state_manager.get_item('output_video_fps')
-
-
-def conditional_clear_video_pool() -> None:
-	if state_manager.get_item('workflow') == 'image-to-video':
-		video_manager.clear_video_pool()
-
-
 def process_temp_frame(temp_frame_path : str, frame_number : int) -> bool:
 	reference_vision_frame = conditional_get_reference_vision_frame()
 	source_vision_frames = read_static_images(state_manager.get_item('source_paths'))
@@ -118,7 +99,7 @@ def process_temp_frame(temp_frame_path : str, frame_number : int) -> bool:
 	return write_image(temp_frame_path, temp_vision_frame)
 
 
-def process_video() -> ErrorCode:
+def process_frames() -> ErrorCode:
 	temp_frame_paths = resolve_temp_frame_paths(state_manager.get_temp_path(), state_manager.get_item('output_path'), state_manager.get_item('temp_frame_format'))
 
 	if temp_frame_paths:
@@ -150,61 +131,3 @@ def process_video() -> ErrorCode:
 		logger.error(translator.get('temp_frames_not_found'), __name__)
 		return 1
 	return 0
-
-
-def merge_frames() -> ErrorCode:
-	temp_frame_paths = resolve_temp_frame_paths(state_manager.get_temp_path(), state_manager.get_item('output_path'), state_manager.get_item('temp_frame_format'))
-	trim_frame_start, trim_frame_end = restrict_trim_frame(len(temp_frame_paths), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
-	output_video_resolution = conditional_scale_resolution()
-	temp_video_fps = conditional_restrict_video_fps()
-
-	logger.info(translator.get('merging_video').format(resolution = pack_resolution(output_video_resolution), fps = state_manager.get_item('output_video_fps')), __name__)
-	if ffmpeg.merge_video(state_manager.get_item('target_path'), state_manager.get_item('output_path'), temp_video_fps, output_video_resolution, trim_frame_start, trim_frame_end):
-		logger.debug(translator.get('merging_video_succeeded'), __name__)
-	else:
-		if is_process_stopping():
-			return 4
-		logger.error(translator.get('merging_video_failed'), __name__)
-		return 1
-	return 0
-
-
-def restore_audio() -> ErrorCode:
-	temp_frame_paths = resolve_temp_frame_paths(state_manager.get_temp_path(), state_manager.get_item('output_path'), state_manager.get_item('temp_frame_format'))
-	trim_frame_start, trim_frame_end = restrict_trim_frame(len(temp_frame_paths), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
-
-	if state_manager.get_item('output_audio_volume') == 0:
-		logger.info(translator.get('skipping_audio'), __name__)
-		move_temp_file(state_manager.get_temp_path(), state_manager.get_item('output_path'))
-	else:
-		source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
-		if source_audio_path:
-			if ffmpeg.replace_audio(source_audio_path, state_manager.get_item('output_path')):
-				conditional_clear_video_pool()
-				logger.debug(translator.get('replacing_audio_succeeded'), __name__)
-			else:
-				conditional_clear_video_pool()
-				if is_process_stopping():
-					return 4
-				logger.warn(translator.get('replacing_audio_skipped'), __name__)
-				move_temp_file(state_manager.get_temp_path(), state_manager.get_item('output_path'))
-		else:
-			if ffmpeg.restore_audio(state_manager.get_item('target_path'), state_manager.get_item('output_path'), trim_frame_start, trim_frame_end):
-				conditional_clear_video_pool()
-				logger.debug(translator.get('restoring_audio_succeeded'), __name__)
-			else:
-				conditional_clear_video_pool()
-				if is_process_stopping():
-					return 4
-				logger.warn(translator.get('restoring_audio_skipped'), __name__)
-				move_temp_file(state_manager.get_temp_path(), state_manager.get_item('output_path'))
-	return 0
-
-
-def finalize_video(start_time : float) -> ErrorCode:
-	if is_video(state_manager.get_item('output_path')):
-		logger.info(translator.get('processing_video_succeeded').format(seconds = calculate_end_time(start_time)), __name__)
-	else:
-		logger.error(translator.get('processing_video_failed'), __name__)
-		return 1
-	return 0

facefusion/workflows/image_to_video.py

@@ -1,9 +1,9 @@
 from functools import partial
 
-from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator
+from facefusion import process_manager
 from facefusion.types import ErrorCode
-from facefusion.vision import detect_video_resolution, pack_resolution, restrict_trim_video_frame, restrict_video_fps, restrict_video_resolution, scale_resolution
-from facefusion.workflows.core import clear, finalize_video, is_process_stopping, merge_frames, process_video, restore_audio, setup
+from facefusion.workflows.core import clear, process_frames, setup
+from facefusion.workflows.to_video import analyse_video, create_temp_frames, finalize_video, merge_frames, restore_audio
 
 
 def process(start_time : float) -> ErrorCode:
@@ -13,7 +13,7 @@ def process(start_time : float) -> ErrorCode:
 		clear,
 		setup,
 		create_temp_frames,
-		process_video,
+		process_frames,
 		merge_frames,
 		restore_audio,
 		partial(finalize_video, start_time),
@@ -31,28 +31,3 @@ def process(start_time : float) -> ErrorCode:
 
 	process_manager.end()
 	return 0
-
-
-def analyse_video() -> ErrorCode:
-	trim_frame_start, trim_frame_end = restrict_trim_video_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
-
-	if content_analyser.analyse_video(state_manager.get_item('target_path'), trim_frame_start, trim_frame_end):
-		return 3
-	return 0
-
-
-def create_temp_frames() -> ErrorCode:
-	trim_frame_start, trim_frame_end = restrict_trim_video_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
-	output_video_resolution = scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
-	temp_video_resolution = restrict_video_resolution(state_manager.get_item('target_path'), output_video_resolution)
-	temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
-	logger.info(translator.get('extracting_frames').format(resolution=pack_resolution(temp_video_resolution), fps=temp_video_fps), __name__)
-
-	if ffmpeg.extract_frames(state_manager.get_item('target_path'), state_manager.get_item('output_path'), temp_video_resolution, temp_video_fps, trim_frame_start, trim_frame_end):
-		logger.debug(translator.get('extracting_frames_succeeded'), __name__)
-	else:
-		if is_process_stopping():
-			return 4
-		logger.error(translator.get('extracting_frames_failed'), __name__)
-		return 1
-	return 0

facefusion/workflows/image_to_video_as_sequence.py

@@ -0,0 +1,54 @@
+import os
+from functools import partial
+
+from facefusion import logger, process_manager, state_manager, translator
+from facefusion.filesystem import are_images, copy_file, create_directory, resolve_file_paths
+from facefusion.temp_helper import resolve_temp_frame_paths
+from facefusion.time_helper import calculate_end_time
+from facefusion.types import ErrorCode
+from facefusion.workflows.core import clear, process_frames, setup
+from facefusion.workflows.to_video import analyse_video, create_temp_frames
+
+
+def process(start_time : float) -> ErrorCode:
+	tasks =\
+	[
+		analyse_video,
+		clear,
+		setup,
+		create_temp_frames,
+		process_frames,
+		copy_temp_frames,
+		partial(finalize_sequence, start_time),
+		clear
+	]
+
+	process_manager.start()
+
+	for task in tasks:
+		error_code = task() #type:ignore[operator]
+
+		if error_code > 0:
+			process_manager.end()
+			return error_code
+
+	process_manager.end()
+	return 0
+
+
+def copy_temp_frames() -> ErrorCode:
+	temp_frame_paths = resolve_temp_frame_paths(state_manager.get_temp_path(), state_manager.get_item('output_path'), state_manager.get_item('temp_frame_format'))
+
+	for temp_frame_path in temp_frame_paths:
+		if not create_directory(state_manager.get_item('output_path')) or not copy_file(temp_frame_path, os.path.join(state_manager.get_item('output_path'), os.path.basename(temp_frame_path))):
+			return 1
+	return 0
+
+
+def finalize_sequence(start_time : float) -> ErrorCode:
+	if are_images(resolve_file_paths(state_manager.get_item('output_path'))):
+		logger.info(translator.get('processing_sequence_succeeded').format(seconds = calculate_end_time(start_time)), __name__)
+	else:
+		logger.error(translator.get('processing_sequence_failed'), __name__)
+		return 1
+	return 0

facefusion/workflows/to_video.py

@@ -0,0 +1,109 @@
+from facefusion import content_analyser, ffmpeg, logger, state_manager, translator, video_manager
+from facefusion.common_helper import get_first
+from facefusion.filesystem import filter_audio_paths, is_video
+from facefusion.media_helper import restrict_trim_frame
+from facefusion.temp_helper import move_temp_file, resolve_temp_frame_paths
+from facefusion.time_helper import calculate_end_time
+from facefusion.types import ErrorCode, Fps, Resolution
+from facefusion.vision import detect_image_resolution, detect_video_resolution, pack_resolution, restrict_trim_video_frame, restrict_video_fps, restrict_video_resolution, scale_resolution
+from facefusion.workflows.core import is_process_stopping
+
+
+def analyse_video() -> ErrorCode:
+	trim_frame_start, trim_frame_end = restrict_trim_video_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
+
+	if content_analyser.analyse_video(state_manager.get_item('target_path'), trim_frame_start, trim_frame_end):
+		return 3
+	return 0
+
+
+def create_temp_frames() -> ErrorCode:
+	trim_frame_start, trim_frame_end = restrict_trim_video_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
+	output_video_resolution = scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
+	temp_video_resolution = restrict_video_resolution(state_manager.get_item('target_path'), output_video_resolution)
+	temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
+	logger.info(translator.get('extracting_frames').format(resolution=pack_resolution(temp_video_resolution), fps=temp_video_fps), __name__)
+
+	if ffmpeg.extract_frames(state_manager.get_item('target_path'), state_manager.get_item('output_path'), temp_video_resolution, temp_video_fps, trim_frame_start, trim_frame_end):
+		logger.debug(translator.get('extracting_frames_succeeded'), __name__)
+	else:
+		if is_process_stopping():
+			return 4
+		logger.error(translator.get('extracting_frames_failed'), __name__)
+		return 1
+	return 0
+
+
+def merge_frames() -> ErrorCode:
+	temp_frame_paths = resolve_temp_frame_paths(state_manager.get_temp_path(), state_manager.get_item('output_path'), state_manager.get_item('temp_frame_format'))
+	trim_frame_start, trim_frame_end = restrict_trim_frame(len(temp_frame_paths), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
+	output_video_resolution = conditional_scale_resolution()
+	temp_video_fps = conditional_restrict_video_fps()
+
+	logger.info(translator.get('merging_video').format(resolution = pack_resolution(output_video_resolution), fps = state_manager.get_item('output_video_fps')), __name__)
+	if ffmpeg.merge_video(state_manager.get_item('target_path'), state_manager.get_item('output_path'), temp_video_fps, output_video_resolution, trim_frame_start, trim_frame_end):
+		logger.debug(translator.get('merging_video_succeeded'), __name__)
+	else:
+		if is_process_stopping():
+			return 4
+		logger.error(translator.get('merging_video_failed'), __name__)
+		return 1
+	return 0
+
+
+def restore_audio() -> ErrorCode:
+	temp_frame_paths = resolve_temp_frame_paths(state_manager.get_temp_path(), state_manager.get_item('output_path'), state_manager.get_item('temp_frame_format'))
+	trim_frame_start, trim_frame_end = restrict_trim_frame(len(temp_frame_paths), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
+
+	if state_manager.get_item('output_audio_volume') == 0:
+		logger.info(translator.get('skipping_audio'), __name__)
+		move_temp_file(state_manager.get_temp_path(), state_manager.get_item('output_path'))
+	else:
+		source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
+		if source_audio_path:
+			if ffmpeg.replace_audio(source_audio_path, state_manager.get_item('output_path')):
+				conditional_clear_video_pool()
+				logger.debug(translator.get('replacing_audio_succeeded'), __name__)
+			else:
+				conditional_clear_video_pool()
+				if is_process_stopping():
+					return 4
+				logger.warn(translator.get('replacing_audio_skipped'), __name__)
+				move_temp_file(state_manager.get_temp_path(), state_manager.get_item('output_path'))
+		else:
+			if ffmpeg.restore_audio(state_manager.get_item('target_path'), state_manager.get_item('output_path'), trim_frame_start, trim_frame_end):
+				conditional_clear_video_pool()
+				logger.debug(translator.get('restoring_audio_succeeded'), __name__)
+			else:
+				conditional_clear_video_pool()
+				if is_process_stopping():
+					return 4
+				logger.warn(translator.get('restoring_audio_skipped'), __name__)
+				move_temp_file(state_manager.get_temp_path(), state_manager.get_item('output_path'))
+	return 0
+
+
+def finalize_video(start_time : float) -> ErrorCode:
+	if is_video(state_manager.get_item('output_path')):
+		logger.info(translator.get('processing_video_succeeded').format(seconds = calculate_end_time(start_time)), __name__)
+	else:
+		logger.error(translator.get('processing_video_failed'), __name__)
+		return 1
+	return 0
+
+
+def conditional_clear_video_pool() -> None:
+	if state_manager.get_item('workflow') == 'image-to-video':
+		video_manager.clear_video_pool()
+
+
+def conditional_restrict_video_fps() -> Fps:
+	if state_manager.get_item('workflow') == 'image-to-video':
+		return restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
+	return state_manager.get_item('output_video_fps')
+
+
+def conditional_scale_resolution() -> Resolution:
+	if state_manager.get_item('workflow') == 'image-to-video':
+		return scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
+	return scale_resolution(detect_image_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))