detect workflow

facefusion.ini

@@ -1,3 +1,6 @@
+[workflow]
+workflow =
+
 [paths]
 temp_path =
 jobs_path =

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] = [ 'audio-to-image', 'image-to-image', 'image-to-video' ]
+workflows : List[WorkFlow] = [ 'auto', 'audio-to-image', 'image-to-image', 'image-to-video' ]
 
 audio_type_set : AudioTypeSet =\
 {

facefusion/core.py

@@ -13,12 +13,13 @@ 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.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, WorkFlow
 from facefusion.workflows import audio_to_image, image_to_image, image_to_video
 
 
@@ -328,6 +329,9 @@ 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') == 'auto':
+		state_manager.set_item('workflow', detect_workflow())
+
 	for processor_module in get_processors_modules(state_manager.get_item('processors')):
 		if not processor_module.pre_process('output'):
 			return 2
@@ -342,3 +346,11 @@ def conditional_process() -> ErrorCode:
 	return 0
 
 
+def detect_workflow() -> WorkFlow:
+	if has_video([ state_manager.get_item('target_path') ]):
+		return 'image-to-video'
+
+	if has_audio(state_manager.get_item('source_paths')) and has_image([ state_manager.get_item('target_path') ]):
+		return 'audio-to-image'
+
+	return 'image-to-image'

facefusion/locales.py

@@ -101,6 +101,7 @@ LOCALES : Locales =\
 		{
 			'install_dependency': 'choose the variant of {dependency} to install',
 			'skip_conda': 'skip the conda environment check',
+			'workflow': 'choose the workflow',
 			'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

@@ -32,7 +32,7 @@ def create_config_path_program() -> ArgumentParser:
 def create_workflow_program() -> ArgumentParser:
 	program = ArgumentParser(add_help = False)
 	group_paths = program.add_argument_group('paths')
-	group_paths.add_argument('--workflow', help = translator.get('help.workflow'), choices = facefusion.choices.workflows)
+	group_paths.add_argument('--workflow', help = translator.get('help.workflow'), default = config.get_str_value('workflow', 'workflow', 'auto'), choices = facefusion.choices.workflows)
 	args_store.register_args([ 'workflow' ], scopes = [ 'api', 'cli' ])
 	return program
 

facefusion/types.py

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