move workflow mode

facefusion/core.py

@@ -12,15 +12,16 @@ from facefusion import args_helper, benchmarker, cli_helper, content_analyser, h
 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, has_audio, has_image, has_video
+from facefusion.filesystem import get_file_extension
 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, WorkflowMode
+from facefusion.types import Args, ErrorCode
 from facefusion.workflows import audio_to_image, audio_to_image_as_frames, image_to_image, image_to_video, image_to_video_as_frames
+from facefusion.workflows.core import detect_workflow_mode
 
 
 def cli() -> None:
@@ -317,33 +318,24 @@ def conditional_process() -> ErrorCode:
 	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
+	workflow_mode = state_manager.get_item('workflow_mode')
 
-	if state_manager.get_item('workflow_mode') == 'audio-to-image:video':
-		return audio_to_image.process(start_time)
-	if state_manager.get_item('workflow_mode') == 'audio-to-image:frames':
-		return audio_to_image_as_frames.process(start_time)
-	if state_manager.get_item('workflow_mode') == 'image-to-image':
-		return image_to_image.process(start_time)
-	if state_manager.get_item('workflow_mode') == 'image-to-video':
-		return image_to_video.process(start_time)
-	if state_manager.get_item('workflow_mode') == 'image-to-video:frames':
-		return image_to_video_as_frames.process(start_time)
+	if 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
 
-	return 0
+		if workflow_mode == 'audio-to-image:video':
+			return audio_to_image.process(start_time)
+		if workflow_mode == 'audio-to-image:frames':
+			return audio_to_image_as_frames.process(start_time)
+		if workflow_mode == 'image-to-image':
+			return image_to_image.process(start_time)
+		if workflow_mode == 'image-to-video':
+			return image_to_video.process(start_time)
+		if workflow_mode == 'image-to-video:frames':
+			return image_to_video_as_frames.process(start_time)
 
+		return 0
 
-def detect_workflow_mode() -> WorkflowMode:
-	if has_video([ state_manager.get_item('target_path') ]):
-		if get_file_extension(state_manager.get_item('output_path')):
-			return 'image-to-video'
-		return 'image-to-video:frames'
-
-	if has_audio(state_manager.get_item('source_paths')) and has_image([ state_manager.get_item('target_path') ]):
-		if get_file_extension(state_manager.get_item('output_path')):
-			return 'audio-to-image:video'
-		return 'audio-to-image:frames'
-
-	return 'image-to-image'
+	return 2

facefusion/workflows/core.py

@@ -8,13 +8,30 @@ from tqdm import tqdm
 from facefusion import 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
+from facefusion.filesystem import filter_audio_paths, get_file_extension, has_audio, has_image, has_video
 from facefusion.processors.core import get_processors_modules
 from facefusion.temp_helper import clear_temp_directory, create_temp_directory, resolve_temp_frame_set
-from facefusion.types import AudioFrame, ErrorCode, VisionFrame
+from facefusion.types import AudioFrame, ErrorCode, VisionFrame, WorkflowMode
 from facefusion.vision import conditional_merge_vision_mask, extract_vision_mask, read_static_image, read_static_images, read_static_video_frame, restrict_trim_video_frame, restrict_video_fps, select_video_frames, write_image
 
 
+def detect_workflow_mode() -> WorkflowMode:
+	target_path = state_manager.get_item('target_path')
+	output_path = state_manager.get_item('output_path')
+
+	if has_video([ target_path ]):
+		if get_file_extension(output_path):
+			return 'image-to-video'
+		return 'image-to-video:frames'
+
+	if has_audio(state_manager.get_item('source_paths')) and has_image([ target_path ]):
+		if get_file_extension(output_path):
+			return 'audio-to-image:video'
+		return 'audio-to-image:frames'
+
+	return 'image-to-image'
+
+
 def is_process_stopping() -> bool:
 	if process_manager.is_stopping():
 		process_manager.end()

tests/test_workflow.py

@@ -0,0 +1,43 @@
+import pytest
+
+from facefusion import state_manager
+from facefusion.download import conditional_download
+from facefusion.workflows.core import detect_workflow_mode
+from .assert_helper import get_test_example_file, get_test_examples_directory
+
+
+@pytest.fixture(scope = 'module', autouse = True)
+def before_all() -> None:
+	conditional_download(get_test_examples_directory(),
+	[
+		'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/source.jpg',
+		'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/source.mp3',
+		'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/target-240p.mp4'
+	])
+
+
+def test_detect_workflow_mode() -> None:
+	state_manager.init_item('source_paths', [ get_test_example_file('source.jpg') ])
+	state_manager.init_item('target_path', get_test_example_file('target-240p.mp4'))
+	state_manager.init_item('output_path', 'output.mp4')
+
+	assert detect_workflow_mode() == 'image-to-video'
+
+	state_manager.init_item('output_path', 'output')
+
+	assert detect_workflow_mode() == 'image-to-video:frames'
+
+	state_manager.init_item('source_paths', [ get_test_example_file('source.mp3') ])
+	state_manager.init_item('target_path', get_test_example_file('source.jpg'))
+	state_manager.init_item('output_path', 'output.jpg')
+
+	assert detect_workflow_mode() == 'audio-to-image:video'
+
+	state_manager.init_item('output_path', 'output')
+
+	assert detect_workflow_mode() == 'audio-to-image:frames'
+
+	state_manager.init_item('source_paths', [ get_test_example_file('source.jpg') ])
+	state_manager.init_item('target_path', get_test_example_file('source.jpg'))
+
+	assert detect_workflow_mode() == 'image-to-image'