introduce workflow-mode and workflow-strategy like next major (#1187)

Claude-Session: https://claude.ai/code/session_01Tbcd6VWCiU4BQP1gywPr2a

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

facefusion.ini

@@ -1,3 +1,7 @@
+[workflow]
+workflow_mode =
+workflow_strategy =
+
 [paths]
 temp_path =
 jobs_path =

facefusion/args.py

@@ -9,6 +9,8 @@ from facefusion.vision import detect_video_fps
 
 def apply_args(args : Args, apply_state_item : ApplyStateItem) -> None:
 	apply_state_item('command', args.get('command'))
+	apply_state_item('workflow_mode', args.get('workflow_mode'))
+	apply_state_item('workflow_strategy', args.get('workflow_strategy'))
 	apply_state_item('temp_path', args.get('temp_path'))
 	apply_state_item('jobs_path', args.get('jobs_path'))
 	apply_state_item('source_paths', args.get('source_paths'))

facefusion/choices.py

@@ -2,7 +2,7 @@ import logging
 from typing import List, Sequence, get_args
 
 from facefusion.common_helper import create_float_range, create_int_range
-from facefusion.types import Angle, AudioEncoder, AudioFormat, AudioTypeSet, BenchmarkMode, BenchmarkResolution, BenchmarkSet, DownloadProvider, DownloadProviderSet, DownloadScope, EncoderSet, ExecutionProvider, ExecutionProviderSet, FaceDetectorModel, FaceDetectorSet, FaceLandmarkerModel, FaceMaskArea, FaceMaskAreaSet, FaceMaskRegion, FaceMaskRegionSet, FaceMaskType, FaceOccluderModel, FaceParserModel, FaceSelectorGender, FaceSelectorMode, FaceSelectorOrder, FaceSelectorRace, Gender, ImageFormat, ImageTypeSet, JobStatus, LogLevel, LogLevelSet, Race, Score, TempFrameFormat, UiWorkflow, VideoEncoder, VideoFormat, VideoMemoryStrategy, VideoPreset, VideoTypeSet, VoiceExtractorModel
+from facefusion.types import Angle, AudioEncoder, AudioFormat, AudioTypeSet, BenchmarkMode, BenchmarkResolution, BenchmarkSet, DownloadProvider, DownloadProviderSet, DownloadScope, EncoderSet, ExecutionProvider, ExecutionProviderSet, FaceDetectorModel, FaceDetectorSet, FaceLandmarkerModel, FaceMaskArea, FaceMaskAreaSet, FaceMaskRegion, FaceMaskRegionSet, FaceMaskType, FaceOccluderModel, FaceParserModel, FaceSelectorGender, FaceSelectorMode, FaceSelectorOrder, FaceSelectorRace, Gender, ImageFormat, ImageTypeSet, JobStatus, LogLevel, LogLevelSet, Race, Score, TempFrameFormat, UiWorkflow, VideoEncoder, VideoFormat, VideoMemoryStrategy, VideoPreset, VideoTypeSet, VoiceExtractorModel, WorkflowMode, WorkflowStrategy
 
 face_detector_set : FaceDetectorSet =\
 {
@@ -76,6 +76,9 @@ video_type_set : VideoTypeSet =\
 	'webm': 'video/webm',
 	'wmv': 'video/x-ms-wmv'
 }
+workflow_modes : List[WorkflowMode] = list(get_args(WorkflowMode))
+workflow_strategies : List[WorkflowStrategy] = list(get_args(WorkflowStrategy))
+
 audio_formats : List[AudioFormat] = list(get_args(AudioFormat))
 image_formats : List[ImageFormat] = list(get_args(ImageFormat))
 video_formats : List[VideoFormat] = list(get_args(VideoFormat))

facefusion/core.py

@@ -9,13 +9,13 @@ from facefusion import benchmarker, cli_helper, content_analyser, hash_helper, l
 from facefusion.args import apply_args, collect_job_args, reduce_job_args, reduce_step_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, is_image, is_video, resolve_file_paths, resolve_file_pattern
+from facefusion.filesystem import get_file_extension, get_file_name, is_video, 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
+from facefusion.types import Args, ErrorCode, WorkflowMode
 from facefusion.workflows import image_to_image, image_to_video
 
 
@@ -312,15 +312,25 @@ def process_step(job_id : str, step_index : int, step_args : Args) -> bool:
 def conditional_process() -> ErrorCode:
 	start_time = time()
 
+	if state_manager.get_item('workflow_mode') == 'auto':
+		state_manager.set_item('workflow_mode', detect_workflow_mode())
+
 	for processor_module in get_processors_modules(state_manager.get_item('processors')):
 		if not processor_module.pre_process('output'):
 			return 2
 
-	if is_image(state_manager.get_item('target_path')):
+	if state_manager.get_item('workflow_mode') == 'image-to-image':
 		return image_to_image.process(start_time)
-	if is_video(state_manager.get_item('target_path')):
+	if state_manager.get_item('workflow_mode') == 'image-to-video':
 		return image_to_video.process(start_time)
 
 	return 0
 
 
+def detect_workflow_mode() -> WorkflowMode:
+	if is_video(state_manager.get_item('target_path')):
+		return 'image-to-video'
+
+	return 'image-to-image'
+
+

facefusion/locales.py

@@ -98,6 +98,8 @@ LOCALES : Locales =\
 		{
 			'install_dependency': 'choose the variant of {dependency} to install',
 			'skip_conda': 'skip the conda environment check',
+			'workflow_mode': 'choose the workflow mode',
+			'workflow_strategy': 'choose the workflow strategy',
 			'config_path': 'choose the config file to override defaults',
 			'temp_path': 'specify the directory for the temporary resources',
 			'jobs_path': 'specify the directory to store jobs',

facefusion/program.py

@@ -21,6 +21,15 @@ def create_help_formatter_large(prog : str) -> HelpFormatter:
 	return HelpFormatter(prog, max_help_position = 300)
 
 
+def create_workflow_program() -> ArgumentParser:
+	program = ArgumentParser(add_help = False)
+	group_workflow = program.add_argument_group('workflow')
+	group_workflow.add_argument('--workflow-mode', help = translator.get('help.workflow_mode'), default = config.get_str_value('workflow', 'workflow_mode', 'auto'), choices = facefusion.choices.workflow_modes)
+	group_workflow.add_argument('--workflow-strategy', help = translator.get('help.workflow_strategy'), default = config.get_str_value('workflow', 'workflow_strategy', 'disk'), choices = facefusion.choices.workflow_strategies)
+	job_store.register_step_keys([ 'workflow_mode', 'workflow_strategy' ])
+	return program
+
+
 def create_config_path_program() -> ArgumentParser:
 	program = ArgumentParser(add_help = False)
 	group_paths = program.add_argument_group('paths')
@@ -299,7 +308,7 @@ def create_step_index_program() -> ArgumentParser:
 
 
 def collect_step_program() -> ArgumentParser:
-	return ArgumentParser(parents = [ create_face_detector_program(), create_face_landmarker_program(), create_face_selector_program(), create_face_tracker_program(), create_face_masker_program(), create_voice_extractor_program(), create_frame_extraction_program(), create_frame_distribution_program(), create_output_creation_program(), create_processors_program() ], add_help = False)
+	return ArgumentParser(parents = [ create_workflow_program(), create_face_detector_program(), create_face_landmarker_program(), create_face_selector_program(), create_face_tracker_program(), create_face_masker_program(), create_voice_extractor_program(), create_frame_extraction_program(), create_frame_distribution_program(), create_output_creation_program(), create_processors_program() ], add_help = False)
 
 
 def collect_job_program() -> ArgumentParser:

facefusion/types.py

@@ -63,6 +63,9 @@ Language = Literal['en']
 Locales : TypeAlias = Dict[Language, Dict[str, Any]]
 LocalePoolSet : TypeAlias = Dict[str, Locales]
 
+WorkflowMode = Literal['auto', 'image-to-image', 'image-to-video']
+WorkflowStrategy = Literal['disk']
+
 VideoCaptureSet : TypeAlias = Dict[str, cv2.VideoCapture]
 VideoWriterSet : TypeAlias = Dict[str, cv2.VideoWriter]
 CameraCaptureSet : TypeAlias = Dict[str, cv2.VideoCapture]
@@ -293,6 +296,8 @@ JobSet : TypeAlias = Dict[str, Job]
 StateKey = Literal\
 [
 	'command',
+	'workflow_mode',
+	'workflow_strategy',
 	'config_path',
 	'temp_path',
 	'jobs_path',
@@ -363,6 +368,8 @@ StateKey = Literal\
 State = TypedDict('State',
 {
 	'command' : str,
+	'workflow_mode' : WorkflowMode,
+	'workflow_strategy' : WorkflowStrategy,
 	'config_path' : str,
 	'temp_path' : str,
 	'jobs_path' : str,

facefusion/workflows/image_to_video.py

@@ -21,14 +21,23 @@ def process(start_time : float) -> ErrorCode:
 	[
 		analyse_video,
 		clear,
-		setup,
-		extract_frames,
-		process_frames,
-		merge_frames,
+		setup
+	]
+
+	if state_manager.get_item('workflow_strategy') == 'disk':
+		tasks.extend(
+		[
+			extract_frames,
+			process_frames,
+			merge_frames
+		])
+
+	tasks.extend(
+	[
 		restore_audio,
 		partial(finalize_video, start_time),
 		clear
-	]
+	])
 	process_manager.start()
 
 	for task in tasks: