API experiment part1

facefusion/_preview.py

@@ -0,0 +1,69 @@
+from time import time
+import cv2
+import numpy
+
+from facefusion import core, state_manager
+from facefusion.audio import create_empty_audio_frame, get_audio_frame
+from facefusion.common_helper import get_first
+from facefusion.content_analyser import analyse_frame
+from facefusion.face_analyser import get_average_face, get_many_faces
+from facefusion.face_selector import sort_faces_by_order
+from facefusion.face_store import get_reference_faces
+from facefusion.filesystem import filter_audio_paths, is_image, is_video
+from facefusion.processors.core import get_processors_modules
+from facefusion.typing import AudioFrame, Face, FaceSet, VisionFrame
+from facefusion.vision import get_video_frame, read_static_image, read_static_images, resize_frame_resolution
+
+
+def process_frame(frame_number : int = 0) -> None:
+	core.conditional_append_reference_faces()
+	reference_faces = get_reference_faces() if 'reference' in state_manager.get_item('face_selector_mode') else None
+	source_frames = read_static_images(state_manager.get_item('source_paths'))
+	source_faces = []
+
+	for source_frame in source_frames:
+		temp_faces = get_many_faces([ source_frame ])
+		temp_faces = sort_faces_by_order(temp_faces, 'large-small')
+		if temp_faces:
+			source_faces.append(get_first(temp_faces))
+	source_face = get_average_face(source_faces)
+	source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
+	source_audio_frame = create_empty_audio_frame()
+
+	if source_audio_path and state_manager.get_item('output_video_fps') and state_manager.get_item('reference_frame_number'):
+		reference_audio_frame_number = state_manager.get_item('reference_frame_number')
+		if state_manager.get_item('trim_frame_start'):
+			reference_audio_frame_number -= state_manager.get_item('trim_frame_start')
+		temp_audio_frame = get_audio_frame(source_audio_path, state_manager.get_item('output_video_fps'), reference_audio_frame_number)
+		if numpy.any(temp_audio_frame):
+			source_audio_frame = temp_audio_frame
+
+	if is_image(state_manager.get_item('target_path')):
+		target_vision_frame = read_static_image(state_manager.get_item('target_path'))
+		preview_vision_frame = process_preview_frame(reference_faces, source_face, source_audio_frame, target_vision_frame)
+		return preview_vision_frame
+
+	if is_video(state_manager.get_item('target_path')):
+		temp_vision_frame = get_video_frame(state_manager.get_item('target_path'), frame_number)
+		preview_vision_frame = process_preview_frame(reference_faces, source_face, source_audio_frame, temp_vision_frame)
+		return preview_vision_frame
+
+
+def process_preview_frame(reference_faces : FaceSet, source_face : Face, source_audio_frame : AudioFrame, target_vision_frame : VisionFrame) -> VisionFrame:
+	target_vision_frame = resize_frame_resolution(target_vision_frame, (1024, 1024))
+	source_vision_frame = target_vision_frame.copy()
+
+	if analyse_frame(target_vision_frame):
+		return cv2.GaussianBlur(target_vision_frame, (99, 99), 0)
+
+	for processor_module in get_processors_modules(state_manager.get_item('processors')):
+		if processor_module.pre_process('preview'):
+			target_vision_frame = processor_module.process_frame(
+			{
+				'reference_faces': reference_faces,
+				'source_face': source_face,
+				'source_audio_frame': source_audio_frame,
+				'source_vision_frame': source_vision_frame,
+				'target_vision_frame': target_vision_frame
+			})
+	return target_vision_frame

facefusion/api.py

@@ -0,0 +1,96 @@
+import asyncio
+import time
+from io import BytesIO
+from typing import Any, List
+
+import cv2
+import uvicorn
+from litestar import Litestar, WebSocket, get as read, websocket as stream
+
+from facefusion import choices, execution, _preview
+from facefusion.processors import choices as processors_choices
+from facefusion.state_manager import get_state
+from facefusion.typing import ExecutionDevice
+
+
+@read('/choices')
+async def read_choices() -> Any:
+	__choices__ = {}
+
+	for key in dir(choices):
+		if not key.startswith('__'):
+			value = getattr(choices, key)
+
+			if isinstance(value, (dict, list)):
+				__choices__[key] = value
+
+	return __choices__
+
+
+@read('/processors/choices')
+async def read_processors_choices() -> Any:
+	__processors_choices__ = {}
+
+	for key in dir(processors_choices):
+		if not key.startswith('__'):
+			value = getattr(processors_choices, key)
+
+			if isinstance(value, (dict, list)):
+				__processors_choices__[key] = value
+
+	return __processors_choices__
+
+
+@read('/execution/providers')
+async def read_execution_providers() -> Any:
+	return execution.get_execution_provider_set()
+
+
+@stream('/execution/devices')
+async def stream_execution_devices(socket : WebSocket) -> None:
+	await socket.accept()
+
+	while True:
+		await socket.send_json(execution.detect_execution_devices())
+		await asyncio.sleep(0.5)
+
+
+@read('/execution/devices')
+async def read_execution_devices() -> List[ExecutionDevice]:
+	return execution.detect_execution_devices()
+
+
+@read('/static_execution/devices')
+async def read_static_execution_devices() -> List[ExecutionDevice]:
+	return execution.detect_static_execution_devices()
+
+
+@stream('/state')
+async def stream_state(socket : WebSocket) -> None:
+	await socket.accept()
+
+	while True:
+		await socket.send_json(get_state())
+		await asyncio.sleep(0.5)
+
+
+@read('/preview', media_type = 'image/png')
+async def read_preview() -> None:
+	_, preview_vision_frame = cv2.imencode('.png', _preview.process_frame())
+	return preview_vision_frame.tobytes()
+
+
+api = Litestar(
+[
+	read_choices,
+	read_processors_choices,
+	stream_execution_devices,
+	read_execution_devices,
+	read_static_execution_devices,
+	stream_state,
+	read_preview
+])
+
+
+def run() -> None:
+	uvicorn.run(api)

facefusion/core.py

@@ -6,7 +6,7 @@ from time import time
 
 import numpy
 
-from facefusion import content_analyser, face_classifier, face_detector, face_landmarker, face_masker, face_recognizer, logger, process_manager, state_manager, voice_extractor, wording
+from facefusion import api, content_analyser, face_classifier, face_detector, face_landmarker, face_masker, face_recognizer, logger, process_manager, state_manager, voice_extractor, wording
 from facefusion.args import apply_args, collect_job_args, reduce_job_args, reduce_step_args
 from facefusion.common_helper import get_first
 from facefusion.content_analyser import analyse_image, analyse_video
@@ -61,15 +61,7 @@ def route(args : Args) -> None:
 	if not pre_check():
 		return conditional_exit(2)
 	if state_manager.get_item('command') == 'run':
-		import facefusion.uis.core as ui
-
-		if not common_pre_check() or not processors_pre_check():
-			return conditional_exit(2)
-		for ui_layout in ui.get_ui_layouts_modules(state_manager.get_item('ui_layouts')):
-			if not ui_layout.pre_check():
-				return conditional_exit(2)
-		ui.init()
-		ui.launch()
+		api.run()
 	if state_manager.get_item('command') == 'headless-run':
 		if not job_manager.init_jobs(state_manager.get_item('jobs_path')):
 			hard_exit(1)