dynamic dependency check

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Tbcd6VWCiU4BQP1gywPr2a

facefusion.ini

@@ -51,7 +51,7 @@ voice_extractor_model =
 trim_frame_start =
 trim_frame_end =
 temp_frame_format =
-keep_temp =
+workflow_mode =
 
 [frame_distribution]
 target_frame_amount =

facefusion/args.py

@@ -45,7 +45,8 @@ def apply_args(args : Args, apply_state_item : ApplyStateItem) -> None:
 	apply_state_item('trim_frame_start', args.get('trim_frame_start'))
 	apply_state_item('trim_frame_end', args.get('trim_frame_end'))
 	apply_state_item('temp_frame_format', args.get('temp_frame_format'))
-	apply_state_item('keep_temp', args.get('keep_temp'))
+	apply_state_item('temp_pixel_format', args.get('temp_pixel_format'))
+	apply_state_item('workflow_mode', args.get('workflow_mode'))
 	apply_state_item('target_frame_amount', args.get('target_frame_amount'))
 	apply_state_item('output_image_quality', args.get('output_image_quality'))
 	apply_state_item('output_image_scale', args.get('output_image_scale'))

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, TempPixelFormat, UiWorkflow, VideoEncoder, VideoFormat, VideoMemoryStrategy, VideoPreset, VideoTypeSet, VoiceExtractorModel, WorkflowMode
 
 face_detector_set : FaceDetectorSet =\
 {
@@ -80,8 +80,12 @@ audio_formats : List[AudioFormat] = list(get_args(AudioFormat))
 image_formats : List[ImageFormat] = list(get_args(ImageFormat))
 video_formats : List[VideoFormat] = list(get_args(VideoFormat))
 temp_frame_formats : List[TempFrameFormat] = list(get_args(TempFrameFormat))
+#todo: needs review - [config] [critical: low] temp pixel format choices for the stream pipe
+temp_pixel_formats : List[TempPixelFormat] = list(get_args(TempPixelFormat))
 
 output_audio_encoders : List[AudioEncoder] = list(get_args(AudioEncoder))
+#todo: needs review - [config] [critical: low] workflow mode choices for disk and stream
+workflow_modes : List[WorkflowMode] = list(get_args(WorkflowMode))
 output_video_encoders : List[VideoEncoder] = list(get_args(VideoEncoder))
 output_encoder_set : EncoderSet =\
 {

facefusion/content_analyser.py

@@ -9,7 +9,7 @@ from facefusion.download import conditional_download_hashes, conditional_downloa
 from facefusion.filesystem import resolve_relative_path
 from facefusion.thread_helper import conditional_thread_semaphore
 from facefusion.types import Detection, DownloadScope, DownloadSet, Fps, InferencePool, ModelSet, VisionFrame
-from facefusion.vision import detect_video_fps, fit_contain_frame, read_image, read_video_frame
+from facefusion.vision import detect_video_fps, fit_contain_frame, read_image, sample_video_frames
 
 STREAM_COUNTER = 0
 
@@ -154,19 +154,23 @@ def analyse_image(image_path : str) -> bool:
 	return analyse_frame(vision_frame)
 
 
+#todo: needs review - [sampling] [critical: high] pre sampled frames replace per frame reads, sample_index must stay aligned with the fps modulo
 @lru_cache()
 def analyse_video(video_path : str, trim_frame_start : int, trim_frame_end : int) -> bool:
 	video_fps = detect_video_fps(video_path)
 	frame_range = range(trim_frame_start, trim_frame_end)
+	vision_frames = sample_video_frames(video_path, trim_frame_start, trim_frame_end, int(video_fps))
 	rate = 0.0
 	total = 0
 	counter = 0
+	sample_index = 0
 
 	with tqdm(total = len(frame_range), desc = translator.get('analysing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
 
 		for frame_number in frame_range:
-			if frame_number % int(video_fps) == 0:
-				vision_frame = read_video_frame(video_path, frame_number)
+			if frame_number % int(video_fps) == 0 and sample_index < len(vision_frames):
+				vision_frame = vision_frames[sample_index]
+				sample_index += 1
 
 				if numpy.any(vision_frame):
 					total += 1

facefusion/core.py

@@ -90,20 +90,19 @@ def pre_check() -> bool:
 		logger.error(translator.get('python_not_supported').format(version = '3.10'), __name__)
 		return False
 
-	if not shutil.which('curl'):
-		logger.error(translator.get('curl_not_installed'), __name__)
-		return False
+	for dependency in [ 'curl', 'ffmpeg', 'ffprobe' ]:
+		if not shutil.which(dependency):
+			logger.error(translator.get('dependency_not_installed').format(dependency = dependency), __name__)
+			return False
 
-	if not shutil.which('ffmpeg'):
-		logger.error(translator.get('ffmpeg_not_installed'), __name__)
-		return False
 	return True
 
 
+#todo: needs review - [security] [critical: high] hash guard updated for the sampling change in content_analyser, stale hash silently exits the app
 def common_pre_check() -> bool:
 	content_analyser_content = inspect.getsource(content_analyser).encode()
 
-	return hash_helper.create_hash(content_analyser_content) == '975d67d6'
+	return hash_helper.create_hash(content_analyser_content) == 'f9a5317c'
 
 
 def processors_pre_check() -> bool:

facefusion/ffmpeg.py

@@ -8,9 +8,10 @@ from tqdm import tqdm
 
 import facefusion.choices
 from facefusion import ffmpeg_builder, logger, process_manager, state_manager, translator
+from facefusion.ffprobe import extract_video_metadata
 from facefusion.filesystem import get_file_format, remove_file
 from facefusion.temp_helper import get_temp_file_path, get_temp_frame_pattern
-from facefusion.types import AudioBuffer, AudioEncoder, Command, EncoderSet, Fps, Resolution, UpdateProgress, VideoEncoder, VideoFormat
+from facefusion.types import AudioBuffer, AudioEncoder, Command, EncoderSet, Fps, Resolution, UpdateProgress, VideoFormat
 from facefusion.vision import detect_video_duration, detect_video_fps, pack_resolution, predict_video_frame_total
 
 
@@ -107,7 +108,9 @@ def get_available_encoder_set() -> EncoderSet:
 	return available_encoder_set
 
 
+#todo: needs review - [color] [critical: high] hdr sources are tonemapped to bt709 during extraction, gated by the probed color_transfer
 def extract_frames(target_path : str, temp_video_resolution : Resolution, temp_video_fps : Fps, trim_frame_start : int, trim_frame_end : int) -> bool:
+	color_transfer = extract_video_metadata(target_path).get('color_transfer')
 	extract_frame_total = predict_video_frame_total(target_path, temp_video_fps, trim_frame_start, trim_frame_end)
 	temp_frame_pattern = get_temp_frame_pattern(target_path, '%08d')
 	commands = ffmpeg_builder.chain(
@@ -115,7 +118,10 @@ def extract_frames(target_path : str, temp_video_resolution : Resolution, temp_v
 		ffmpeg_builder.set_media_resolution(pack_resolution(temp_video_resolution)),
 		ffmpeg_builder.set_frame_quality(0),
 		ffmpeg_builder.enforce_pixel_format('rgb24'),
-		ffmpeg_builder.select_frame_range(trim_frame_start, trim_frame_end, temp_video_fps),
+		ffmpeg_builder.concat(
+			ffmpeg_builder.select_frame_range(trim_frame_start, trim_frame_end, temp_video_fps),
+			ffmpeg_builder.tonemap_video(color_transfer)
+		),
 		ffmpeg_builder.prevent_frame_drop(),
 		ffmpeg_builder.set_start_number(trim_frame_start),
 		ffmpeg_builder.set_output(temp_frame_pattern)
@@ -227,7 +233,7 @@ def merge_video(target_path : str, temp_video_fps : Fps, output_video_resolution
 	temp_video_format = cast(VideoFormat, get_file_format(temp_video_path))
 	temp_frame_pattern = get_temp_frame_pattern(target_path, '%08d')
 
-	output_video_encoder = fix_video_encoder(temp_video_format, output_video_encoder)
+	output_video_encoder = ffmpeg_builder.fix_video_encoder(temp_video_format, output_video_encoder)
 	commands = ffmpeg_builder.chain(
 		ffmpeg_builder.set_input_fps(temp_video_fps),
 		ffmpeg_builder.set_start_number(trim_frame_start),
@@ -237,9 +243,11 @@ def merge_video(target_path : str, temp_video_fps : Fps, output_video_resolution
 		ffmpeg_builder.set_video_tag(output_video_encoder, temp_video_format),
 		ffmpeg_builder.set_video_quality(output_video_encoder, output_video_quality),
 		ffmpeg_builder.set_video_preset(output_video_encoder, output_video_preset),
+		#todo: needs review - [color] [critical: high] merge output always converted and tagged bt709
 		ffmpeg_builder.concat(
 			ffmpeg_builder.set_video_fps(output_video_fps),
-			ffmpeg_builder.keep_video_alpha(output_video_encoder)
+			ffmpeg_builder.keep_video_alpha(output_video_encoder),
+			ffmpeg_builder.convert_colorspace('bt709')
 		),
 		ffmpeg_builder.set_pixel_format(output_video_encoder),
 		ffmpeg_builder.force_output(temp_video_path)
@@ -290,13 +298,3 @@ def fix_audio_encoder(video_format : VideoFormat, audio_encoder : AudioEncoder)
 	return audio_encoder
 
 
-def fix_video_encoder(video_format : VideoFormat, video_encoder : VideoEncoder) -> VideoEncoder:
-	if video_format in [ 'm4v', 'mpeg', 'mxf', 'wmv' ]:
-		return 'libx264'
-	if video_format in [ 'mkv', 'mp4' ] and video_encoder == 'rawvideo':
-		return 'libx264'
-	if video_format == 'mov' and video_encoder == 'libvpx-vp9':
-		return 'libx264'
-	if video_format == 'webm':
-		return 'libvpx-vp9'
-	return video_encoder

facefusion/ffmpeg_builder.py

@@ -5,7 +5,7 @@ from typing import List, Optional
 import numpy
 
 from facefusion.filesystem import get_file_format
-from facefusion.types import AudioEncoder, Command, CommandSet, Duration, Fps, StreamMode, VideoEncoder, VideoFormat, VideoPreset
+from facefusion.types import AudioEncoder, ColorTransfer, Command, CommandSet, Duration, Fps, StreamMode, VideoEncoder, VideoFormat, VideoPreset
 
 
 def run(commands : List[Command]) -> List[Command]:
@@ -83,6 +83,16 @@ def unsafe_concat() -> List[Command]:
 	return [ '-f', 'concat', '-safe', '0' ]
 
 
+#todo: needs review - [seeking] [critical: low] seek based start for the pipe reader
+def set_input_seek(seek_time : float) -> List[Command]:
+	return [ '-ss', str(seek_time) ]
+
+
+#todo: needs review - [decoding] [critical: low] raw output format for pipe io
+def set_output_format(output_format : str) -> List[Command]:
+	return [ '-f', output_format ]
+
+
 def enforce_pixel_format(pixel_format : str) -> List[Command]:
 	return [ '-pix_fmt', pixel_format ]
 
@@ -99,6 +109,16 @@ def set_frame_quality(frame_quality : int) -> List[Command]:
 	return [ '-q:v', str(frame_quality) ]
 
 
+#todo: needs review - [sampling] [critical: medium] select expression decodes every frame of the range to pick samples
+def select_frame_samples(frame_start : int, frame_end : int, frame_stride : int) -> List[Command]:
+	return [ '-vf', 'select=between(n\\,' + str(frame_start) + '\\,' + str(frame_end - 1) + ')*not(mod(n\\,' + str(frame_stride) + '))' ]
+
+
+#todo: needs review - [sampling] [critical: low] caps the sampler output frames
+def set_frame_total(frame_total : int) -> List[Command]:
+	return [ '-frames:v', str(frame_total) ]
+
+
 def select_frame_range(frame_start : int, frame_end : int, video_fps : Fps) -> List[Command]:
 	if isinstance(frame_start, int) and isinstance(frame_end, int):
 		return [ '-vf', 'trim=start_frame=' + str(frame_start) + ':end_frame=' + str(frame_end) + ',fps=' + str(video_fps) ]
@@ -113,6 +133,18 @@ def prevent_frame_drop() -> List[Command]:
 	return [ '-vsync', '0' ]
 
 
+#todo: needs review - [color] [critical: high] gated hdr to sdr tonemap, relies on ffmpeg 8 swscale intent support
+def tonemap_video(color_transfer : ColorTransfer) -> List[Command]:
+	if color_transfer in [ 'smpte2084', 'arib-std-b67' ]:
+		return [ '-vf', 'scale=out_primaries=bt709:out_transfer=bt709:intent=perceptual' ]
+	return []
+
+
+#todo: needs review - [color] [critical: high] converts and tags encoder output, prevents player colorimetry guessing
+def convert_colorspace(color_space : str) -> List[Command]:
+	return [ '-vf', 'scale=out_color_matrix=' + color_space + ':out_range=tv:out_primaries=' + color_space + ':out_transfer=' + color_space ]
+
+
 def select_media_range(frame_start : int, frame_end : int, media_fps : Fps) -> List[Command]:
 	commands = []
 
@@ -183,6 +215,11 @@ def set_audio_volume(audio_volume : int) -> List[Command]:
 	return [ '-filter:a', 'volume=' + str(audio_volume / 100) ]
 
 
+#todo: needs review - [encoding] [critical: low] explicit ffmpeg thread cap
+def set_thread_count(thread_count : int) -> List[Command]:
+	return [ '-threads', str(thread_count) ]
+
+
 def set_video_encoder(video_encoder : str) -> List[Command]:
 	return [ '-c:v', video_encoder ]
 
@@ -191,6 +228,19 @@ def copy_video_encoder() -> List[Command]:
 	return set_video_encoder('copy')
 
 
+#todo: needs review - [encoding] [critical: medium] moved from ffmpeg, adds rawvideo to libx264 remap for mkv and mp4
+def fix_video_encoder(video_format : VideoFormat, video_encoder : VideoEncoder) -> VideoEncoder:
+	if video_format in [ 'm4v', 'mpeg', 'mxf', 'wmv' ]:
+		return 'libx264'
+	if video_format in [ 'mkv', 'mp4' ] and video_encoder == 'rawvideo':
+		return 'libx264'
+	if video_format == 'mov' and video_encoder == 'libvpx-vp9':
+		return 'libx264'
+	if video_format == 'webm':
+		return 'libvpx-vp9'
+	return video_encoder
+
+
 def set_faststart(video_format : VideoFormat) -> List[Command]:
 	if video_format in [ 'm4v', 'mov', 'mp4' ]:
 		return [ '-movflags', '+faststart' ]

facefusion/ffprobe.py

@@ -0,0 +1,95 @@
+import subprocess
+from functools import lru_cache
+from typing import Dict, List
+
+from facefusion import ffprobe_builder
+from facefusion.types import Buffer, Command, Fps, VideoMetadata
+
+
+#todo: no review needed - [probing] copy of v4, return type uses the Buffer alias
+def run_ffprobe(commands : List[Command]) -> subprocess.Popen[Buffer]:
+	commands = ffprobe_builder.run(commands)
+	return subprocess.Popen(commands, stderr = subprocess.PIPE, stdout = subprocess.PIPE)
+
+
+#todo: needs review - [probing] [critical: low] extracted from v4 probe_entries, parsing logic unchanged
+def parse_entries(output : Buffer) -> Dict[str, str]:
+	media_entries = {}
+
+	if output:
+		lines = output.decode().strip().splitlines()
+
+		for line in lines:
+			if '=' in line:
+				key, value = line.split('=', 1)
+				media_entries[key] = value
+
+	return media_entries
+
+
+#todo: needs review - [probing] [critical: low] new against v4, probes the first video stream, replaces the v4 probe_entries which had no consumer on this branch
+def probe_video_entries(video_path : str, entries : List[str]) -> Dict[str, str]:
+	commands = ffprobe_builder.chain(
+		ffprobe_builder.select_stream('v:0'),
+		ffprobe_builder.show_stream_entries(entries),
+		ffprobe_builder.format_to_key_value(),
+		ffprobe_builder.set_input(video_path)
+	)
+	output, _ = run_ffprobe(commands).communicate()
+
+	return parse_entries(output)
+
+
+#todo: needs review - [probing] [critical: low] new against v4, container level fallback for duration and bit_rate
+def probe_format_entries(video_path : str, entries : List[str]) -> Dict[str, str]:
+	commands = ffprobe_builder.chain(
+		ffprobe_builder.show_format_entries(entries),
+		ffprobe_builder.format_to_key_value(),
+		ffprobe_builder.set_input(video_path)
+	)
+	output, _ = run_ffprobe(commands).communicate()
+
+	return parse_entries(output)
+
+
+#todo: needs review - [probing] [critical: high] extends v4 with lru_cache, na fallbacks and color_transfer, cache serves stale metadata when files are overwritten
+@lru_cache(maxsize = 128)
+def extract_video_metadata(video_path : str) -> VideoMetadata:
+	video_entries = probe_video_entries(video_path, [ 'duration', 'width', 'height', 'r_frame_rate', 'bit_rate', 'color_transfer' ])
+
+	if video_entries.get('duration') == 'N/A':
+		video_entries['duration'] = probe_format_entries(video_path, [ 'duration' ]).get('duration')
+
+	if video_entries.get('bit_rate') == 'N/A':
+		video_entries['bit_rate'] = probe_format_entries(video_path, [ 'bit_rate' ]).get('bit_rate')
+
+	duration = float(video_entries.get('duration'))
+	fps = extract_video_fps(video_entries.get('r_frame_rate'))
+	frame_total = int(duration * fps)
+	width = int(video_entries.get('width'))
+	height = int(video_entries.get('height'))
+	bit_rate = int(video_entries.get('bit_rate'))
+	color_transfer = video_entries.get('color_transfer', 'unknown')
+
+	video_metadata : VideoMetadata =\
+	{
+		'duration' : duration,
+		'frame_total' : frame_total,
+		'fps' : fps,
+		'resolution' : (width, height),
+		'bit_rate' : bit_rate,
+		'color_transfer' : color_transfer
+	}
+
+	return video_metadata
+
+
+#todo: no review needed - [probing] 100% copy of v4
+def extract_video_fps(frame_rate : str) -> Fps:
+	if frame_rate and '/' in frame_rate:
+		numerator, denominator = frame_rate.split('/')
+
+		if int(numerator) and int(denominator):
+			return int(numerator) / int(denominator)
+
+	return 0.0

facefusion/ffprobe_builder.py

@@ -0,0 +1,40 @@
+import itertools
+import shutil
+from typing import List
+
+from facefusion.types import Command
+
+
+#todo: no review needed - [probing] 100% copy of v4
+def run(commands : List[Command]) -> List[Command]:
+	return [ shutil.which('ffprobe'), '-loglevel', 'error' ] + commands
+
+
+#todo: no review needed - [probing] 100% copy of v4
+def chain(*commands : List[Command]) -> List[Command]:
+	return list(itertools.chain(*commands))
+
+
+#todo: needs review - [probing] [critical: low] new against v4, callers must pass indexed specifiers like v:0 as parse_entries can only represent one stream
+def select_stream(stream : str) -> List[Command]:
+	return [ '-select_streams', stream ]
+
+
+#todo: needs review - [probing] [critical: low] renamed against v4 show_entries, stream level entries
+def show_stream_entries(entries : List[str]) -> List[Command]:
+	return [ '-show_entries', 'stream=' + ','.join(entries) ]
+
+
+#todo: needs review - [probing] [critical: low] new against v4, container level entries
+def show_format_entries(entries : List[str]) -> List[Command]:
+	return [ '-show_entries', 'format=' + ','.join(entries) ]
+
+
+#todo: no review needed - [probing] 100% copy of v4
+def format_to_key_value() -> List[Command]:
+	return [ '-of', 'default=noprint_wrappers=1' ]
+
+
+#todo: no review needed - [probing] 100% copy of v4
+def set_input(input_path : str) -> List[Command]:
+	return [ '-i', input_path ]

facefusion/locales.py

@@ -6,8 +6,7 @@ LOCALES : Locales =\
 	{
 		'conda_not_activated': 'conda is not activated',
 		'python_not_supported': 'python version is not supported, upgrade to {version} or higher',
-		'curl_not_installed': 'curl is not installed',
-		'ffmpeg_not_installed': 'ffmpeg is not installed',
+		'dependency_not_installed': '{dependency} is not installed',
 		'creating_temp': 'creating temporary resources',
 		'extracting_frames': 'extracting frames with a resolution of {resolution} and {fps} frames per second',
 		'extracting_frames_succeeded': 'extracting frames succeeded',
@@ -136,7 +135,9 @@ LOCALES : Locales =\
 			'trim_frame_start': 'specify the starting frame of the target video',
 			'trim_frame_end': 'specify the ending frame of the target video',
 			'temp_frame_format': 'specify the temporary resources format',
-			'keep_temp': 'keep the temporary resources after processing',
+			'temp_pixel_format': 'specify the pixel format for the stream pipe (rgba keeps the alpha channel)',
+			#todo: needs review - [ui] [critical: low] workflow_mode help text replaces keep_temp
+			'workflow_mode': 'specify how frames flow through processing (disk extracts and merges temp frames, stream pipes frames through ffmpeg)',
 			'target_frame_amount': 'specify the amount of target frames forwarded to the processor',
 			'output_image_quality': 'specify the image quality which translates to the image compression',
 			'output_image_scale': 'specify the image scale based on the target image',
@@ -201,9 +202,10 @@ LOCALES : Locales =\
 			'benchmark_cycle_count_slider': 'BENCHMARK CYCLE COUNT',
 			'benchmark_resolutions_checkbox_group': 'BENCHMARK RESOLUTIONS',
 			'clear_button': 'CLEAR',
-			'common_options_checkbox_group': 'OPTIONS',
 			'download_providers_checkbox_group': 'DOWNLOAD PROVIDERS',
 			'execution_providers_checkbox_group': 'EXECUTION PROVIDERS',
+			#todo: needs review - [ui] [critical: low] workflow mode dropdown label replaces common options
+			'workflow_mode_dropdown': 'WORKFLOW MODE',
 			'execution_thread_count_slider': 'EXECUTION THREAD COUNT',
 			'face_detector_angles_checkbox_group': 'FACE DETECTOR ANGLES',
 			'face_detector_model_dropdown': 'FACE DETECTOR MODEL',

facefusion/metadata.py

@@ -4,7 +4,8 @@ METADATA =\
 {
 	'name': 'FaceFusion',
 	'description': 'Industry leading face manipulation platform',
-	'version': '3.7.1',
+	#todo: needs review - [config] [critical: low] version bump to 3.8.0
+	'version': '3.8.0',
 	'license': 'OpenRAIL-AS',
 	'author': 'Henry Ruhs',
 	'url': 'https://facefusion.io'

facefusion/program.py

@@ -169,8 +169,11 @@ def create_frame_extraction_program() -> ArgumentParser:
 	group_frame_extraction.add_argument('--trim-frame-start', help = translator.get('help.trim_frame_start'), type = int, default = facefusion.config.get_int_value('frame_extraction', 'trim_frame_start'))
 	group_frame_extraction.add_argument('--trim-frame-end', help = translator.get('help.trim_frame_end'), type = int, default = facefusion.config.get_int_value('frame_extraction', 'trim_frame_end'))
 	group_frame_extraction.add_argument('--temp-frame-format', help = translator.get('help.temp_frame_format'), default = config.get_str_value('frame_extraction', 'temp_frame_format', 'png'), choices = facefusion.choices.temp_frame_formats)
-	group_frame_extraction.add_argument('--keep-temp', help = translator.get('help.keep_temp'), action = 'store_true', default = config.get_bool_value('frame_extraction', 'keep_temp'))
-	job_store.register_step_keys([ 'trim_frame_start', 'trim_frame_end', 'temp_frame_format', 'keep_temp' ])
+	#todo: needs review - [streaming] [critical: low] new temp pixel format option, rgba enables alpha through the stream pipe
+	group_frame_extraction.add_argument('--temp-pixel-format', help = translator.get('help.temp_pixel_format'), default = config.get_str_value('frame_extraction', 'temp_pixel_format', 'rgb'), choices = facefusion.choices.temp_pixel_formats)
+	#todo: needs review - [workflow] [critical: medium] keep_temp cli option replaced by workflow_mode, breaks existing configs and jobs
+	group_frame_extraction.add_argument('--workflow-mode', help = translator.get('help.workflow_mode'), default = config.get_str_value('frame_extraction', 'workflow_mode', 'disk'), choices = facefusion.choices.workflow_modes)
+	job_store.register_step_keys([ 'trim_frame_start', 'trim_frame_end', 'temp_frame_format', 'temp_pixel_format', 'workflow_mode' ])
 	return program
 
 

facefusion/temp_helper.py

@@ -42,8 +42,7 @@ def create_temp_directory(file_path : str) -> bool:
 	return create_directory(temp_directory_path)
 
 
+#todo: needs review - [workflow] [critical: medium] keep_temp support removed, the temp directory is always deleted
 def clear_temp_directory(file_path : str) -> bool:
-	if not state_manager.get_item('keep_temp'):
-		temp_directory_path = get_temp_directory_path(file_path)
-		return remove_directory(temp_directory_path)
-	return True
+	temp_directory_path = get_temp_directory_path(file_path)
+	return remove_directory(temp_directory_path)

facefusion/types.py

@@ -1,3 +1,4 @@
+import subprocess
 from collections import namedtuple
 from threading import Lock
 from typing import Any, Callable, Dict, List, Literal, NotRequired, Optional, Tuple, TypeAlias, TypedDict
@@ -63,14 +64,9 @@ Language = Literal['en']
 Locales : TypeAlias = Dict[Language, Dict[str, Any]]
 LocalePoolSet : TypeAlias = Dict[str, Locales]
 
-VideoCaptureSet : TypeAlias = Dict[str, cv2.VideoCapture]
-VideoWriterSet : TypeAlias = Dict[str, cv2.VideoWriter]
+#todo: needs review - [types] [critical: low] writer pool holds ffmpeg processes instead of cv2 writers
+VideoWriterSet : TypeAlias = Dict[str, subprocess.Popen[bytes]]
 CameraCaptureSet : TypeAlias = Dict[str, cv2.VideoCapture]
-VideoPoolSet = TypedDict('VideoPoolSet',
-{
-	'capture' : VideoCaptureSet,
-	'writer' : VideoWriterSet
-})
 CameraPoolSet = TypedDict('CameraPoolSet',
 {
 	'capture' : CameraCaptureSet
@@ -97,11 +93,44 @@ VoiceChunk : TypeAlias = NDArray[Any]
 
 Fps : TypeAlias = float
 Duration : TypeAlias = float
+
+Buffer : TypeAlias = bytes
+BitRate : TypeAlias = int
+#todo: needs review - [types] [critical: low] new buffer, bitrate and reader state types for the pipe based video io
+VideoReaderBuffer : TypeAlias = Dict[int, VisionFrame]
+VideoReader = TypedDict('VideoReader',
+{
+	'process' : subprocess.Popen[bytes],
+	'video_path' : str,
+	'width' : int,
+	'height' : int,
+	'fps' : Fps,
+	'frame_total' : int,
+	'position' : int,
+	'frame_buffer' : VideoReaderBuffer
+})
+VideoReaderSet : TypeAlias = Dict[str, VideoReader]
+VideoPoolSet = TypedDict('VideoPoolSet',
+{
+	'reader' : VideoReaderSet,
+	'writer' : VideoWriterSet
+})
 Color : TypeAlias = Tuple[int, int, int, int]
 Padding : TypeAlias = Tuple[int, int, int, int]
 Margin : TypeAlias = Tuple[int, int, int, int]
 Orientation = Literal['landscape', 'portrait']
 Resolution : TypeAlias = Tuple[int, int]
+#todo: needs review - [types] [critical: low] probed video metadata including color_transfer
+ColorTransfer : TypeAlias = str
+VideoMetadata = TypedDict('VideoMetadata',
+{
+	'duration' : Duration,
+	'frame_total' : int,
+	'fps' : Fps,
+	'resolution' : Resolution,
+	'bit_rate' : BitRate,
+	'color_transfer' : ColorTransfer
+})
 
 ProcessState = Literal['checking', 'processing', 'stopping', 'pending']
 Args : TypeAlias = Dict[str, Any]
@@ -143,6 +172,10 @@ AudioFormat = Literal['flac', 'm4a', 'mp3', 'ogg', 'opus', 'wav']
 ImageFormat = Literal['bmp', 'jpeg', 'png', 'tiff', 'webp']
 VideoFormat = Literal['avi', 'm4v', 'mkv', 'mov', 'mp4', 'mpeg', 'mxf', 'webm', 'wmv']
 TempFrameFormat = Literal['bmp', 'jpeg', 'png', 'tiff']
+#todo: needs review - [streaming] [critical: low] new temp pixel format controls the stream pipe channels
+TempPixelFormat = Literal['rgb', 'rgba']
+#todo: needs review - [workflow] [critical: medium] new workflow mode replaces keep_temp in state and jobs
+WorkflowMode = Literal['disk', 'stream']
 AudioTypeSet : TypeAlias = Dict[AudioFormat, str]
 ImageTypeSet : TypeAlias = Dict[ImageFormat, str]
 VideoTypeSet : TypeAlias = Dict[VideoFormat, str]
@@ -313,7 +346,8 @@ StateKey = Literal\
 	'trim_frame_start',
 	'trim_frame_end',
 	'temp_frame_format',
-	'keep_temp',
+	'temp_pixel_format',
+	'workflow_mode',
 	'target_frame_amount',
 	'output_image_quality',
 	'output_image_scale',
@@ -384,7 +418,8 @@ State = TypedDict('State',
 	'trim_frame_start' : int,
 	'trim_frame_end' : int,
 	'temp_frame_format' : TempFrameFormat,
-	'keep_temp' : bool,
+	'temp_pixel_format' : TempPixelFormat,
+	'workflow_mode' : WorkflowMode,
 	'target_frame_amount' : int,
 	'output_image_quality' : int,
 	'output_image_scale' : Scale,

facefusion/uis/choices.py

@@ -5,8 +5,8 @@ from facefusion.uis.types import JobManagerAction, JobRunnerAction, PreviewMode
 
 job_manager_actions : List[JobManagerAction] = [ 'job-create', 'job-submit', 'job-delete', 'job-add-step', 'job-remix-step', 'job-insert-step', 'job-remove-step' ]
 job_runner_actions : List[JobRunnerAction] = [ 'job-run', 'job-run-all', 'job-retry', 'job-retry-all' ]
+#todo: needs review - [ui] [critical: low] common_options removed together with keep_temp
 
-common_options : List[str] = [ 'keep-temp' ]
 
 preview_modes : List[PreviewMode] = [ 'default', 'frame-by-frame', 'face-by-face' ]
 preview_resolutions : List[str] = [ '512x512', '768x768', '1024x1024' ]

facefusion/uis/components/common_options.py

@@ -1,32 +0,0 @@
-from typing import List, Optional
-
-import gradio
-
-from facefusion import state_manager, translator
-from facefusion.uis import choices as uis_choices
-
-COMMON_OPTIONS_CHECKBOX_GROUP : Optional[gradio.Checkboxgroup] = None
-
-
-def render() -> None:
-	global COMMON_OPTIONS_CHECKBOX_GROUP
-
-	common_options = []
-
-	if state_manager.get_item('keep_temp'):
-		common_options.append('keep-temp')
-
-	COMMON_OPTIONS_CHECKBOX_GROUP = gradio.Checkboxgroup(
-		label = translator.get('uis.common_options_checkbox_group'),
-		choices = uis_choices.common_options,
-		value = common_options
-	)
-
-
-def listen() -> None:
-	COMMON_OPTIONS_CHECKBOX_GROUP.change(update, inputs = COMMON_OPTIONS_CHECKBOX_GROUP)
-
-
-def update(common_options : List[str]) -> None:
-	keep_temp = 'keep-temp' in common_options
-	state_manager.set_item('keep_temp', keep_temp)

facefusion/uis/components/workflow.py

@@ -0,0 +1,28 @@
+from typing import Optional
+
+import gradio
+
+import facefusion.choices
+from facefusion import state_manager, translator
+from facefusion.types import WorkflowMode
+
+#todo: needs review - [ui] [critical: low] new workflow mode dropdown component replacing common_options
+WORKFLOW_MODE_DROPDOWN : Optional[gradio.Dropdown] = None
+
+
+def render() -> None:
+	global WORKFLOW_MODE_DROPDOWN
+
+	WORKFLOW_MODE_DROPDOWN = gradio.Dropdown(
+		label = translator.get('uis.workflow_mode_dropdown'),
+		choices = facefusion.choices.workflow_modes,
+		value = state_manager.get_item('workflow_mode')
+	)
+
+
+def listen() -> None:
+	WORKFLOW_MODE_DROPDOWN.change(update_workflow_mode, inputs = WORKFLOW_MODE_DROPDOWN)
+
+
+def update_workflow_mode(workflow_mode : WorkflowMode) -> None:
+	state_manager.set_item('workflow_mode', workflow_mode)

facefusion/uis/layouts/default.py

@@ -1,7 +1,7 @@
 import gradio
 
 from facefusion import state_manager
-from facefusion.uis.components import about, age_modifier_options, background_remover_options, common_options, deep_swapper_options, download, execution, execution_thread_count, expression_restorer_options, face_debugger_options, face_detector, face_editor_options, face_enhancer_options, face_landmarker, face_masker, face_selector, face_swapper_options, face_tracker, frame_colorizer_options, frame_enhancer_options, instant_runner, job_manager, job_runner, lip_syncer_options, memory, output, output_options, preview, preview_options, processors, source, target, temp_frame, terminal, trim_frame, ui_workflow, voice_extractor
+from facefusion.uis.components import about, age_modifier_options, background_remover_options, deep_swapper_options, download, execution, execution_thread_count, expression_restorer_options, face_debugger_options, face_detector, face_editor_options, face_enhancer_options, face_landmarker, face_masker, face_selector, face_swapper_options, face_tracker, frame_colorizer_options, frame_enhancer_options, instant_runner, job_manager, job_runner, lip_syncer_options, memory, output, output_options, preview, preview_options, processors, source, target, temp_frame, terminal, trim_frame, ui_workflow, voice_extractor, workflow
 
 
 def pre_check() -> bool:
@@ -40,6 +40,9 @@ def render() -> gradio.Blocks:
 					lip_syncer_options.render()
 				with gradio.Blocks():
 					voice_extractor.render()
+				#todo: needs review - [ui] [critical: low] workflow component added, common_options removed
+				with gradio.Blocks():
+					workflow.render()
 				with gradio.Blocks():
 					execution.render()
 					execution_thread_count.render()
@@ -81,8 +84,6 @@ def render() -> gradio.Blocks:
 					face_detector.render()
 				with gradio.Blocks():
 					face_landmarker.render()
-				with gradio.Blocks():
-					common_options.render()
 	return layout
 
 
@@ -99,6 +100,8 @@ def listen() -> None:
 	frame_colorizer_options.listen()
 	frame_enhancer_options.listen()
 	lip_syncer_options.listen()
+	#todo: needs review - [ui] [critical: low] workflow listener added, common_options listener removed
+	workflow.listen()
 	execution.listen()
 	execution_thread_count.listen()
 	download.listen()
@@ -121,7 +124,6 @@ def listen() -> None:
 	face_detector.listen()
 	face_landmarker.listen()
 	voice_extractor.listen()
-	common_options.listen()
 
 
 def run(ui : gradio.Blocks) -> None:

facefusion/video_manager.py

@@ -1,46 +1,228 @@
-import cv2
+import os
+import subprocess
+from typing import Optional, Tuple, cast
 
-from facefusion.types import VideoPoolSet
+import numpy
+
+from facefusion import ffmpeg_builder, state_manager
+from facefusion.common_helper import is_windows
+from facefusion.ffprobe import extract_video_metadata
+from facefusion.filesystem import get_file_format
+from facefusion.temp_helper import get_temp_file_path
+from facefusion.types import Fps, Resolution, VideoFormat, VideoPoolSet, VideoReader, VideoReaderBuffer, VisionFrame
 
 VIDEO_POOL_SET : VideoPoolSet =\
 {
-	'capture': {},
+	'reader': {},
 	'writer': {}
 }
 
 
-def get_video_capture(video_path : str) -> cv2.VideoCapture:
-	if video_path not in VIDEO_POOL_SET.get('capture'):
-		video_capture = cv2.VideoCapture(video_path)
-
-		if video_capture.isOpened():
-			VIDEO_POOL_SET['capture'][video_path] = video_capture
-
-	return VIDEO_POOL_SET.get('capture').get(video_path)
+#todo: needs review - [decoding] [critical: medium] new ffmpeg raw pipe reader with seek based start, stderr devnull hides decode errors
+def create_video_reader_process(video_path : str, frame_position : int, video_fps : Fps) -> subprocess.Popen[bytes]:
+	color_transfer = extract_video_metadata(video_path).get('color_transfer')
+	commands = ffmpeg_builder.chain(
+		ffmpeg_builder.set_input_seek(frame_position / video_fps),
+		ffmpeg_builder.set_input(video_path),
+		ffmpeg_builder.tonemap_video(color_transfer),
+		ffmpeg_builder.prevent_frame_drop(),
+		ffmpeg_builder.enforce_pixel_format('bgr24'),
+		ffmpeg_builder.set_output_format('rawvideo'),
+		ffmpeg_builder.cast_stream()
+	)
+	commands = ffmpeg_builder.run(commands)
+	return subprocess.Popen(commands, stdout = subprocess.PIPE, stderr = subprocess.DEVNULL)
 
 
-def conditional_set_video_frame_position(video_capture : cv2.VideoCapture, frame_position : int) -> bool:
-	if not video_capture.get(cv2.CAP_PROP_POS_FRAMES) == frame_position:
-		return video_capture.set(cv2.CAP_PROP_POS_FRAMES, frame_position)
-	return True
+#todo: needs review - [sampling] [critical: medium] select filter decodes the entire range to sample frames, runs at lowered process priority
+def create_video_sampler_process(video_path : str, frame_start : int, frame_end : int, frame_stride : int) -> subprocess.Popen[bytes]:
+	color_transfer = extract_video_metadata(video_path).get('color_transfer')
+	sample_total = len([ frame_number for frame_number in range(frame_start, frame_end) if frame_number % frame_stride == 0 ])
+	sampler_thread_count = 4
+	commands = ffmpeg_builder.chain(
+		ffmpeg_builder.set_thread_count(sampler_thread_count),
+		ffmpeg_builder.set_input(video_path),
+		ffmpeg_builder.concat(
+			ffmpeg_builder.select_frame_samples(frame_start, frame_end, frame_stride),
+			ffmpeg_builder.tonemap_video(color_transfer)
+		),
+		ffmpeg_builder.prevent_frame_drop(),
+		ffmpeg_builder.set_frame_total(sample_total),
+		ffmpeg_builder.enforce_pixel_format('bgr24'),
+		ffmpeg_builder.set_output_format('rawvideo'),
+		ffmpeg_builder.cast_stream()
+	)
+	commands = ffmpeg_builder.run(commands)
+
+	if is_windows():
+		return subprocess.Popen(commands, stdout = subprocess.PIPE, stderr = subprocess.DEVNULL, creationflags = subprocess.IDLE_PRIORITY_CLASS) #type:ignore[attr-defined]
+	return subprocess.Popen(commands, stdout = subprocess.PIPE, stderr = subprocess.DEVNULL, preexec_fn = demote_process_priority)
 
 
-def get_video_writer(video_path : str) -> cv2.VideoWriter:
-	if video_path not in VIDEO_POOL_SET.get('writer'):
-		video_writer = cv2.VideoWriter()
-
-		if video_writer.isOpened():
-			VIDEO_POOL_SET['writer'][video_path] = video_writer
-
-	return VIDEO_POOL_SET.get('writer').get(video_path)
+#todo: needs review - [sampling] [critical: low] posix only nice fallback for the sampler priority
+def demote_process_priority() -> None:
+	os.nice(10)
 
 
+#todo: needs review - [lifecycle] [critical: medium] pooled reader per path is created without lock and only reaped via clear_video_pool
+def get_video_reader(video_path : str) -> VideoReader:
+	if video_path not in VIDEO_POOL_SET.get('reader'):
+		video_metadata = extract_video_metadata(video_path)
+		width, height = video_metadata.get('resolution')
+
+		VIDEO_POOL_SET['reader'][video_path] =\
+		{
+			'process': create_video_reader_process(video_path, 0, video_metadata.get('fps')),
+			'video_path': video_path,
+			'width': width,
+			'height': height,
+			'fps': video_metadata.get('fps'),
+			'frame_total': video_metadata.get('frame_total'),
+			'position': 0,
+			'frame_buffer': {}
+		}
+
+	return VIDEO_POOL_SET.get('reader').get(video_path)
+
+
+#todo: needs review - [seeking] [critical: high] forward skip up to 128 frames by draining the pipe, everything else restarts the ffmpeg process
+def conditional_set_video_reader_position(video_reader : VideoReader, frame_position : int) -> None:
+	skip_margin = 128
+	skipping = video_reader.get('position') < frame_position and frame_position - video_reader.get('position') <= skip_margin
+
+	while skipping:
+		has_vision_frame, vision_frame = read_video_reader_frame(video_reader)
+		skipping = has_vision_frame and video_reader.get('position') < frame_position
+
+	if not video_reader.get('position') == frame_position:
+		restart_video_reader(video_reader, frame_position)
+
+
+#todo: needs review - [seeking] [critical: high] kill and respawn per out of order seek, frequent restarts stack up on random access
+def restart_video_reader(video_reader : VideoReader, frame_position : int) -> None:
+	video_reader.get('process').kill()
+	video_reader.get('process').wait()
+	video_reader['process'] = create_video_reader_process(video_reader.get('video_path'), frame_position, video_reader.get('fps'))
+	video_reader['position'] = frame_position
+	video_reader['frame_buffer'].clear()
+
+
+#todo: needs review - [decoding] [critical: high] partial pipe read returns none and desyncs position from the actual stream
+def read_video_reader_frame(video_reader : VideoReader) -> Tuple[bool, Optional[VisionFrame]]:
+	frame_size = video_reader.get('width') * video_reader.get('height') * 3
+	frame_buffer = video_reader.get('process').stdout.read(frame_size)
+
+	if len(frame_buffer) == frame_size:
+		video_reader['position'] = video_reader.get('position') + 1
+		return True, numpy.frombuffer(frame_buffer, numpy.uint8).reshape(video_reader.get('height'), video_reader.get('width'), 3)
+	return False, None
+
+
+#todo: needs review - [memory] [critical: medium] eviction only trims below frame_start, look ahead growth is bounded by the window callers
+def evict_video_reader_buffer(video_reader : VideoReader, frame_start : int, buffer_margin : int) -> None:
+	frame_buffer = video_reader.get('frame_buffer')
+
+	for frame_number in [ key for key in frame_buffer if key < frame_start - buffer_margin ]:
+		del frame_buffer[frame_number]
+
+
+#todo: needs review - [memory] [critical: high] frame_buffer keeps decoded frames in ram, restart heuristic depends on buffer_margin 16
+def read_video_reader_window(video_reader : VideoReader, frame_start : int, frame_end : int) -> VideoReaderBuffer:
+	frame_buffer = video_reader.get('frame_buffer')
+	buffer_margin = 16
+	read_start = max(frame_start, 0)
+	read_end = frame_end
+
+	if video_reader.get('frame_total') > 0:
+		read_end = min(read_end, video_reader.get('frame_total') - 1)
+
+	if read_start not in frame_buffer and (read_start < video_reader.get('position') or read_start > video_reader.get('position') + buffer_margin):
+		restart_video_reader(video_reader, read_start)
+
+	reading = video_reader.get('position') <= read_end
+
+	while reading:
+		has_vision_frame, vision_frame = read_video_reader_frame(video_reader)
+
+		if has_vision_frame:
+			frame_buffer[video_reader.get('position') - 1] = vision_frame
+		reading = has_vision_frame and video_reader.get('position') <= read_end
+
+	evict_video_reader_buffer(video_reader, read_start, buffer_margin)
+	return frame_buffer
+
+
+#todo: needs review - [encoding] [critical: low] rounds odd dimensions to even for yuv420p compatibility
+def pack_video_resolution(resolution : Resolution) -> str:
+	width, height = resolution
+	return str(round(width / 2) * 2) + 'x' + str(round(height / 2) * 2)
+
+
+#todo: needs review - [color] [critical: high] output is always converted and tagged bt709, encoder_thread_count hardcoded to 16
+def create_video_writer_process(target_path : str, temp_video_fps : Fps, temp_video_resolution : Resolution, output_video_resolution : Resolution, output_video_fps : Fps) -> subprocess.Popen[bytes]:
+	output_video_encoder = state_manager.get_item('output_video_encoder')
+	output_video_quality = state_manager.get_item('output_video_quality')
+	output_video_preset = state_manager.get_item('output_video_preset')
+	temp_video_path = get_temp_file_path(target_path)
+	temp_video_format = cast(VideoFormat, get_file_format(temp_video_path))
+	output_video_encoder = ffmpeg_builder.fix_video_encoder(temp_video_format, output_video_encoder)
+	encoder_thread_count = 16
+	writer_pixel_format = 'bgr24'
+
+	if state_manager.get_item('temp_pixel_format') == 'rgba':
+		writer_pixel_format = 'bgra'
+
+	commands = ffmpeg_builder.chain(
+		ffmpeg_builder.set_output_format('rawvideo'),
+		ffmpeg_builder.enforce_pixel_format(writer_pixel_format),
+		ffmpeg_builder.set_media_resolution(pack_video_resolution(temp_video_resolution)),
+		ffmpeg_builder.set_input_fps(temp_video_fps),
+		ffmpeg_builder.set_input('pipe:0'),
+		ffmpeg_builder.set_media_resolution(pack_video_resolution(output_video_resolution)),
+		ffmpeg_builder.set_video_encoder(output_video_encoder),
+		ffmpeg_builder.set_thread_count(encoder_thread_count),
+		ffmpeg_builder.set_video_tag(output_video_encoder, temp_video_format),
+		ffmpeg_builder.set_video_quality(output_video_encoder, output_video_quality),
+		ffmpeg_builder.set_video_preset(output_video_encoder, output_video_preset),
+		ffmpeg_builder.concat(
+			ffmpeg_builder.set_video_fps(output_video_fps),
+			ffmpeg_builder.convert_colorspace('bt709')
+		),
+		ffmpeg_builder.set_pixel_format(output_video_encoder),
+		ffmpeg_builder.force_output(temp_video_path)
+	)
+	commands = ffmpeg_builder.run(commands)
+	return subprocess.Popen(commands, stdin = subprocess.PIPE, stdout = subprocess.DEVNULL)
+
+
+#todo: needs review - [lifecycle] [critical: low] pooled writer keyed by target_path
+def get_video_writer(target_path : str, temp_video_fps : Fps, temp_video_resolution : Resolution, output_video_resolution : Resolution, output_video_fps : Fps) -> subprocess.Popen[bytes]:
+	if target_path not in VIDEO_POOL_SET.get('writer'):
+		VIDEO_POOL_SET['writer'][target_path] = create_video_writer_process(target_path, temp_video_fps, temp_video_resolution, output_video_resolution, output_video_fps)
+
+	return VIDEO_POOL_SET.get('writer').get(target_path)
+
+
+#todo: needs review - [encoding] [critical: medium] blocking stdin write without backpressure or broken pipe handling
+def write_video_writer_frame(video_writer : subprocess.Popen[bytes], vision_frame : VisionFrame) -> None:
+	video_writer.stdin.write(vision_frame.tobytes())
+
+
+#todo: needs review - [encoding] [critical: medium] encoder failures only surface via returncode at close time
+def close_video_writer(video_writer : subprocess.Popen[bytes]) -> bool:
+	video_writer.stdin.close()
+	video_writer.wait()
+	return video_writer.returncode == 0
+
+
+#todo: needs review - [lifecycle] [critical: medium] terminates writers mid encode, readers are terminated without reaping stdout
 def clear_video_pool() -> None:
-	for video_capture in VIDEO_POOL_SET.get('capture').values():
-		video_capture.release()
-
 	for video_writer in VIDEO_POOL_SET.get('writer').values():
-		video_writer.release()
+		video_writer.terminate()
+		video_writer.wait()
 
-	VIDEO_POOL_SET['capture'].clear()
+	for video_reader in VIDEO_POOL_SET.get('reader').values():
+		video_reader.get('process').terminate()
+
+	VIDEO_POOL_SET['reader'].clear()
 	VIDEO_POOL_SET['writer'].clear()

facefusion/vision.py

@@ -7,10 +7,11 @@ import numpy
 from cv2.typing import Size
 
 from facefusion.common_helper import is_windows
+from facefusion.ffprobe import extract_video_metadata
 from facefusion.filesystem import get_file_extension, is_image, is_video
 from facefusion.thread_helper import thread_lock, thread_semaphore
 from facefusion.types import ColorMode, Duration, Fps, Mask, Orientation, Resolution, Scale, VisionFrame
-from facefusion.video_manager import conditional_set_video_frame_position, get_video_capture
+from facefusion.video_manager import conditional_set_video_reader_position, create_video_sampler_process, get_video_reader, read_video_reader_frame, read_video_reader_window
 
 
 def read_static_images(image_paths : List[str], color_mode : ColorMode = 'rgb') -> List[VisionFrame]:
@@ -75,17 +76,17 @@ def read_static_video_frame(video_path : str, frame_number : int = 0) -> Optiona
 	return read_video_frame(video_path, frame_number)
 
 
+#todo: needs review - [decoding] [critical: medium] cv2 capture replaced by pooled ffmpeg reader, position clamped to the frame_total estimate
 def read_video_frame(video_path : str, frame_number : int = 0) -> Optional[VisionFrame]:
 	if is_video(video_path):
-		video_capture = get_video_capture(video_path)
+		video_reader = get_video_reader(video_path)
 
-		if video_capture and video_capture.isOpened():
-			video_frame_total = int(video_capture.get(cv2.CAP_PROP_FRAME_COUNT))
-			video_frame_position = min(video_frame_total, frame_number)
+		if video_reader:
+			video_frame_position = min(video_reader.get('frame_total'), frame_number)
 
 			with thread_semaphore():
-				conditional_set_video_frame_position(video_capture, video_frame_position)
-				has_vision_frame, vision_frame = video_capture.read()
+				conditional_set_video_reader_position(video_reader, video_frame_position)
+				has_vision_frame, vision_frame = read_video_reader_frame(video_reader)
 
 			if has_vision_frame:
 				return vision_frame
@@ -93,26 +94,52 @@ def read_video_frame(video_path : str, frame_number : int = 0) -> Optional[Visio
 	return None
 
 
+#todo: needs review - [memory] [critical: high] all sampled frames are held in ram, long high resolution videos can exhaust memory
+def sample_video_frames(video_path : str, frame_start : int, frame_end : int, frame_stride : int) -> List[VisionFrame]:
+	vision_frames = []
+
+	if is_video(video_path):
+		width, height = extract_video_metadata(video_path).get('resolution')
+		frame_size = width * height * 3
+		sample_total = len([ frame_number for frame_number in range(frame_start, frame_end) if frame_number % frame_stride == 0 ])
+		video_sampler = create_video_sampler_process(video_path, frame_start, frame_end, frame_stride)
+
+		for _ in range(sample_total):
+			frame_buffer = video_sampler.stdout.read(frame_size)
+
+			if len(frame_buffer) == frame_size:
+				vision_frames.append(numpy.frombuffer(frame_buffer, dtype = numpy.uint8).reshape(height, width, 3))
+
+		video_sampler.stdout.close()
+		video_sampler.wait()
+
+	return vision_frames
+
+
 @lru_cache(maxsize = 2)
 def read_static_video_chunk(video_path : str, chunk_number : int, chunk_size : int) -> Dict[int, VisionFrame]:
 	return read_video_chunk(video_path, chunk_number, chunk_size)
 
 
+#todo: needs review - [decoding] [critical: medium] chunk end clamped by the ffprobe frame_total estimate instead of the exact cv2 count
 def read_video_chunk(video_path : str, chunk_number : int, chunk_size : int) -> Dict[int, VisionFrame]:
 	video_frame_chunk = {}
 
 	if is_video(video_path) and chunk_number > -1:
-		video_capture = get_video_capture(video_path)
+		video_reader = get_video_reader(video_path)
 
-		if video_capture and video_capture.isOpened():
-			video_frame_total = int(video_capture.get(cv2.CAP_PROP_FRAME_COUNT))
+		if video_reader:
 			video_frame_position = chunk_number * chunk_size
+			video_frame_end = video_frame_position + chunk_size
+
+			if video_reader.get('frame_total') > 0:
+				video_frame_end = min(video_frame_end, video_reader.get('frame_total'))
 
 			with thread_semaphore():
-				conditional_set_video_frame_position(video_capture, video_frame_position)
+				conditional_set_video_reader_position(video_reader, video_frame_position)
 
-				for frame_number in range(video_frame_position, min(video_frame_position + chunk_size, video_frame_total)):
-					has_vision_frame, vision_frame = video_capture.read()
+				for frame_number in range(video_frame_position, video_frame_end):
+					has_vision_frame, vision_frame = read_video_reader_frame(video_reader)
 
 					if has_vision_frame:
 						video_frame_chunk[frame_number] = vision_frame
@@ -120,32 +147,32 @@ def read_video_chunk(video_path : str, chunk_number : int, chunk_size : int) ->
 	return video_frame_chunk
 
 
+#todo: needs review - [decoding] [critical: high] window read replaces the chunk cache, out of range frames fall back to empty vision frames
 def select_video_frames(video_path : str, frame_number : int = 0, frame_offset : int = 2) -> List[VisionFrame]:
 	vision_frames = []
-	chunk_size = frame_offset * 2 + 1
+	frame_start = frame_number - frame_offset
+	frame_end = frame_number + frame_offset
 
 	if is_video(video_path):
 		with thread_lock():
-			for current_number in range(frame_number - frame_offset, frame_number + frame_offset + 1):
-				video_frame_chunk = read_static_video_chunk(video_path, current_number // chunk_size, chunk_size)
+			video_reader = get_video_reader(video_path)
+			frame_buffer = read_video_reader_window(video_reader, frame_start, frame_end)
+
+			for current_number in range(frame_start, frame_end + 1):
 				vision_frame = create_empty_vision_frame()
 
-				if current_number in video_frame_chunk:
-					vision_frame = video_frame_chunk.get(current_number)
+				if current_number in frame_buffer:
+					vision_frame = frame_buffer.get(current_number)
 
 				vision_frames.append(vision_frame)
 
 	return vision_frames
 
 
+#todo: needs review - [probing] [critical: medium] frame total is duration times fps estimate, diverges from the real count on vfr sources
 def count_video_frame_total(video_path : str) -> int:
 	if is_video(video_path):
-		video_capture = get_video_capture(video_path)
-
-		if video_capture and video_capture.isOpened():
-			with thread_semaphore():
-				video_frame_total = int(video_capture.get(cv2.CAP_PROP_FRAME_COUNT))
-				return video_frame_total
+		return extract_video_metadata(video_path).get('frame_total')
 
 	return 0
 
@@ -158,14 +185,10 @@ def predict_video_frame_total(video_path : str, fps : Fps, trim_frame_start : in
 	return 0
 
 
+#todo: needs review - [probing] [critical: low] fps now from cached ffprobe metadata
 def detect_video_fps(video_path : str) -> Optional[float]:
 	if is_video(video_path):
-		video_capture = get_video_capture(video_path)
-
-		if video_capture and video_capture.isOpened():
-			with thread_semaphore():
-				video_fps = video_capture.get(cv2.CAP_PROP_FPS)
-				return video_fps
+		return extract_video_metadata(video_path).get('fps')
 
 	return None
 
@@ -211,15 +234,10 @@ def restrict_trim_frame(video_path : str, trim_frame_start : Optional[int], trim
 	return 0, video_frame_total
 
 
+#todo: needs review - [probing] [critical: low] resolution now from cached ffprobe metadata
 def detect_video_resolution(video_path : str) -> Optional[Resolution]:
 	if is_video(video_path):
-		video_capture = get_video_capture(video_path)
-
-		if video_capture and video_capture.isOpened():
-			with thread_semaphore():
-				width = video_capture.get(cv2.CAP_PROP_FRAME_WIDTH)
-				height = video_capture.get(cv2.CAP_PROP_FRAME_HEIGHT)
-				return int(width), int(height)
+		return extract_video_metadata(video_path).get('resolution')
 
 	return None
 

facefusion/workflows/core.py

@@ -1,4 +1,6 @@
-from facefusion import logger, process_manager, translator
+from facefusion import logger, process_manager, state_manager, translator
+from facefusion.temp_helper import clear_temp_directory, create_temp_directory
+from facefusion.types import ErrorCode
 
 
 def is_process_stopping() -> bool:
@@ -6,3 +8,17 @@ def is_process_stopping() -> bool:
 		process_manager.end()
 		logger.info(translator.get('processing_stopped'), __name__)
 	return process_manager.is_pending()
+
+
+#todo: no review needed - [workflow] copy of v4 setup task using the v3 target_path temp helpers
+def setup() -> ErrorCode:
+	if create_temp_directory(state_manager.get_item('target_path')):
+		logger.debug(translator.get('creating_temp'), __name__)
+	return 0
+
+
+#todo: no review needed - [workflow] copy of v4 clear task using the v3 target_path temp helpers
+def clear() -> ErrorCode:
+	if clear_temp_directory(state_manager.get_item('target_path')):
+		logger.debug(translator.get('clearing_temp'), __name__)
+	return 0

facefusion/workflows/image_to_image.py

@@ -1,25 +1,27 @@
 from functools import partial
 
-from facefusion import ffmpeg
-from facefusion import logger, process_manager, state_manager, translator
+from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator
 from facefusion.audio import create_empty_audio_frame
-from facefusion.content_analyser import analyse_image
 from facefusion.filesystem import is_image
 from facefusion.processors.core import get_processors_modules
-from facefusion.temp_helper import clear_temp_directory, create_temp_directory, get_temp_file_path
+from facefusion.temp_helper import get_temp_file_path
 from facefusion.time_helper import calculate_end_time
 from facefusion.types import ErrorCode
 from facefusion.vision import conditional_merge_vision_mask, detect_image_resolution, extract_vision_mask, pack_resolution, read_static_image, read_static_images, restrict_image_resolution, scale_resolution, write_image
-from facefusion.workflows.core import is_process_stopping
+from facefusion.workflows.core import clear, is_process_stopping, setup
 
 
+#todo: no review needed - [workflow] copy of v4 task list, analyse_image gates first and clear runs before setup and after finalize_image
 def process(start_time : float) -> ErrorCode:
 	tasks =\
 	[
+		analyse_image,
+		clear,
 		setup,
 		prepare_image,
 		process_image,
-		partial(finalize_image, start_time)
+		partial(finalize_image, start_time),
+		clear
 	]
 	process_manager.start()
 
@@ -34,16 +36,10 @@ def process(start_time : float) -> ErrorCode:
 	return 0
 
 
-def setup() -> ErrorCode:
-	if analyse_image(state_manager.get_item('target_path')):
+#todo: no review needed - [security] copy of v4 analyse_image task, gates ahead of setup so no temp cleanup is needed on a hit
+def analyse_image() -> ErrorCode:
+	if content_analyser.analyse_image(state_manager.get_item('target_path')):
 		return 3
-
-	if clear_temp_directory(state_manager.get_item('target_path')):
-		logger.debug(translator.get('clearing_temp'), __name__)
-
-	if create_temp_directory(state_manager.get_item('target_path')):
-		logger.debug(translator.get('creating_temp'), __name__)
-
 	return 0
 
 
@@ -95,6 +91,7 @@ def process_image() -> ErrorCode:
 	return 0
 
 
+#todo: no review needed - [workflow] copy of v4 finalize_image task, temp clearing moved out into the clear task
 def finalize_image(start_time : float) -> ErrorCode:
 	output_image_resolution = scale_resolution(detect_image_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_image_scale'))
 
@@ -104,9 +101,6 @@ def finalize_image(start_time : float) -> ErrorCode:
 	else:
 		logger.warn(translator.get('finalizing_image_skipped'), __name__)
 
-	logger.debug(translator.get('clearing_temp'), __name__)
-	clear_temp_directory(state_manager.get_item('target_path'))
-
 	if is_image(state_manager.get_item('output_path')):
 		logger.info(translator.get('processing_image_succeeded').format(seconds = calculate_end_time(start_time)), __name__)
 	else:

facefusion/workflows/image_to_video.py

@@ -1,33 +1,50 @@
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from functools import partial
+from typing import List, Tuple
 
+import cv2
 import numpy
 from tqdm import tqdm
 
-from facefusion import ffmpeg
-from facefusion import logger, process_manager, state_manager, translator, video_manager
+from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator, video_manager
 from facefusion.audio import create_empty_audio_frame, get_audio_frame, get_voice_frame
-from facefusion.common_helper import get_first
-from facefusion.content_analyser import analyse_video
+from facefusion.common_helper import get_first, get_middle
 from facefusion.filesystem import filter_audio_paths, is_video
 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_set
+from facefusion.temp_helper import move_temp_file, resolve_temp_frame_set
 from facefusion.time_helper import calculate_end_time
-from facefusion.types import ErrorCode
+from facefusion.types import ErrorCode, FrameSet, Resolution, VisionFrame
 from facefusion.vision import conditional_merge_vision_mask, detect_video_resolution, extract_vision_mask, pack_resolution, read_static_image, read_static_images, read_static_video_frame, restrict_trim_frame, restrict_video_fps, restrict_video_resolution, scale_resolution, select_video_frames, write_image
-from facefusion.workflows.core import is_process_stopping
+from facefusion.workflows.core import clear, is_process_stopping, setup
 
 
+#todo: needs review - [workflow] [critical: medium] task list starts with the shared tasks, workflow_mode appends the disk or stream specific tasks
 def process(start_time : float) -> ErrorCode:
 	tasks =\
 	[
-		setup,
-		extract_frames,
-		process_video,
-		merge_frames,
-		restore_audio,
-		partial(finalize_video, start_time)
+		analyse_video,
+		clear,
+		setup
 	]
+
+	if state_manager.get_item('workflow_mode') == 'disk':
+		tasks.extend(
+		[
+			extract_frames,
+			process_disk_frames,
+			merge_frames
+		])
+
+	if state_manager.get_item('workflow_mode') == 'stream':
+		tasks.append(process_stream_frames)
+
+	tasks.extend(
+	[
+		restore_audio,
+		partial(finalize_video, start_time),
+		clear
+	])
+
 	process_manager.start()
 
 	for task in tasks:
@@ -41,18 +58,12 @@ def process(start_time : float) -> ErrorCode:
 	return 0
 
 
-def setup() -> ErrorCode:
+#todo: no review needed - [security] copy of v4 analyse_video task, gates both workflows ahead of setup so no temp cleanup is needed on a hit
+def analyse_video() -> ErrorCode:
 	trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
 
-	if analyse_video(state_manager.get_item('target_path'), trim_frame_start, trim_frame_end):
+	if content_analyser.analyse_video(state_manager.get_item('target_path'), trim_frame_start, trim_frame_end):
 		return 3
-
-	if clear_temp_directory(state_manager.get_item('target_path')):
-		logger.debug(translator.get('clearing_temp'), __name__)
-
-	if create_temp_directory(state_manager.get_item('target_path')):
-		logger.debug(translator.get('creating_temp'), __name__)
-
 	return 0
 
 
@@ -73,34 +84,53 @@ def extract_frames() -> ErrorCode:
 	return 0
 
 
-def process_video() -> ErrorCode:
-	temp_frame_set = resolve_temp_frame_set(state_manager.get_item('target_path'))
+#todo: needs review - [streaming] [critical: high] ordered submit and pop keeps writes in order, stopping path drains remaining futures without cancel
+def process_stream_frames() -> ErrorCode:
+	trim_frame_start, trim_frame_end = restrict_trim_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'))
+	frame_range = range(trim_frame_start, trim_frame_end)
 
-	if temp_frame_set:
-		with tqdm(total = len(temp_frame_set), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
+	if frame_range:
+		video_writer = video_manager.get_video_writer(state_manager.get_item('target_path'), temp_video_fps, temp_video_resolution, output_video_resolution, state_manager.get_item('output_video_fps'))
+
+		with tqdm(total = len(frame_range), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
 			progress.set_postfix(execution_providers = state_manager.get_item('execution_providers'))
 
+			read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
+
 			with ThreadPoolExecutor(max_workers = state_manager.get_item('execution_thread_count')) as executor:
 				futures = []
+				#todo: needs review - [memory] [critical: high] frame look ahead bound by hardcoded 3gb budget, extract into a dedicated method
+				width, height = temp_video_resolution
+				frame_memory_budget = 3 * 1024 ** 3
+				frame_memory_usage = width * height * 4 * 6
+				frame_look_ahead = min(state_manager.get_item('execution_thread_count') * 2, max(2, frame_memory_budget // frame_memory_usage))
 
-				for frame_number, temp_frame_path in temp_frame_set.items():
-					future = executor.submit(process_temp_frame, temp_frame_path, frame_number)
-					futures.append(future)
+				for frame_number in frame_range:
+					futures.append(executor.submit(process_stream_frame, frame_number, temp_video_resolution))
 
-				for future in as_completed(futures):
-					if is_process_stopping():
-						for __future__ in futures:
-							__future__.cancel()
-
-					if not future.cancelled():
-						future.result()
+					if len(futures) >= frame_look_ahead and is_process_stopping() is False:
+						_, vision_frame = futures.pop(0).result()
+						video_manager.write_video_writer_frame(video_writer, vision_frame)
 						progress.update()
 
+				for future in futures:
+					if is_process_stopping() is False:
+						_, vision_frame = future.result()
+						video_manager.write_video_writer_frame(video_writer, vision_frame)
+						progress.update()
+
+		close_succeeded = video_manager.close_video_writer(video_writer)
+
 		for processor_module in get_processors_modules(state_manager.get_item('processors')):
 			processor_module.post_process()
 
 		if is_process_stopping():
 			return 4
+		if not close_succeeded:
+			return 1
 	else:
 		logger.error(translator.get('temp_frames_not_found'), __name__)
 		return 1
@@ -154,14 +184,13 @@ def restore_audio() -> ErrorCode:
 	return 0
 
 
-def process_temp_frame(temp_frame_path : str, frame_number : int) -> bool:
+#todo: needs review - [workflow] [critical: medium] shared processing path for disk and stream, receives vision frames instead of paths
+def process_target_frame(frame_number : int, target_vision_frames : List[VisionFrame], temp_vision_frame : VisionFrame) -> VisionFrame:
 	trim_frame_start, _ = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
 	reference_vision_frame = read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
 	source_vision_frames = read_static_images(state_manager.get_item('source_paths'))
 	source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
-	target_vision_frames = select_video_frames(state_manager.get_item('target_path'), frame_number, state_manager.get_item('target_frame_amount'))
 	temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
-	temp_vision_frame = read_static_image(temp_frame_path, 'rgba')
 	temp_vision_mask = extract_vision_mask(temp_vision_frame)
 
 	source_audio_frame = get_audio_frame(source_audio_path, temp_video_fps, frame_number - trim_frame_start)
@@ -184,14 +213,83 @@ def process_temp_frame(temp_frame_path : str, frame_number : int) -> bool:
 			'temp_vision_mask': temp_vision_mask
 		})
 
-	temp_vision_frame = conditional_merge_vision_mask(temp_vision_frame, temp_vision_mask)
+	return conditional_merge_vision_mask(temp_vision_frame, temp_vision_mask)
+
+
+#todo: needs review - [correctness] [critical: high] missing neighbor frames resolve to none paths and rely on read_static_image returning none
+def resolve_disk_target_frames(frame_number : int, frame_amount : int, temp_frame_set : FrameSet) -> List[VisionFrame]:
+	target_vision_frames = []
+
+	for current_number in range(frame_number - frame_amount, frame_number + frame_amount + 1):
+		target_vision_frames.append(read_static_image(temp_frame_set.get(current_number)))
+
+	return target_vision_frames
+
+
+#todo: needs review - [workflow] [critical: low] disk variant reads temp frame and neighbors from disk and writes back in place
+def process_disk_frame(temp_frame_path : str, frame_number : int, temp_frame_set : FrameSet) -> bool:
+	target_vision_frames = resolve_disk_target_frames(frame_number, state_manager.get_item('target_frame_amount'), temp_frame_set)
+	temp_vision_frame = read_static_image(temp_frame_path, 'rgba')
+	temp_vision_frame = process_target_frame(frame_number, target_vision_frames, temp_vision_frame)
 	return write_image(temp_frame_path, temp_vision_frame)
 
 
-def finalize_video(start_time : float) -> ErrorCode:
-	logger.debug(translator.get('clearing_temp'), __name__)
-	clear_temp_directory(state_manager.get_item('target_path'))
+#todo: needs review - [workflow] [critical: medium] renamed from process_video, warms the reference frame before the executor
+def process_disk_frames() -> ErrorCode:
+	temp_frame_set = resolve_temp_frame_set(state_manager.get_item('target_path'))
 
+	if temp_frame_set:
+		with tqdm(total = len(temp_frame_set), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
+			progress.set_postfix(execution_providers = state_manager.get_item('execution_providers'))
+
+			read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
+
+			with ThreadPoolExecutor(max_workers = state_manager.get_item('execution_thread_count')) as executor:
+				futures = []
+
+				for frame_number, temp_frame_path in temp_frame_set.items():
+					futures.append(executor.submit(process_disk_frame, temp_frame_path, frame_number, temp_frame_set))
+
+				for future in as_completed(futures):
+					if is_process_stopping():
+						for __future__ in futures:
+							__future__.cancel()
+
+					if not future.cancelled():
+						future.result()
+						progress.update()
+
+		for processor_module in get_processors_modules(state_manager.get_item('processors')):
+			processor_module.post_process()
+
+		if is_process_stopping():
+			return 4
+	else:
+		logger.error(translator.get('temp_frames_not_found'), __name__)
+		return 1
+	return 0
+
+
+#todo: needs review - [streaming] [critical: high] middle frame resized to temp resolution before processing, channels follow temp_pixel_format to match the writer pipe
+def process_stream_frame(frame_number : int, temp_video_resolution : Resolution) -> Tuple[int, VisionFrame]:
+	target_vision_frames = select_video_frames(state_manager.get_item('target_path'), frame_number, state_manager.get_item('target_frame_amount'))
+	target_vision_frame = get_middle(target_vision_frames)
+	temp_vision_frame = target_vision_frame.copy()
+
+	if not (target_vision_frame.shape[1], target_vision_frame.shape[0]) == temp_video_resolution:
+		temp_vision_frame = cv2.resize(target_vision_frame, temp_video_resolution)
+	temp_vision_frame = process_target_frame(frame_number, target_vision_frames, temp_vision_frame)
+
+	if state_manager.get_item('temp_pixel_format') == 'rgba':
+		temp_vision_frame = cv2.cvtColor(temp_vision_frame, cv2.COLOR_BGR2BGRA)
+	if state_manager.get_item('temp_pixel_format') == 'rgb':
+		temp_vision_frame = temp_vision_frame[:, :, :3]
+
+	return frame_number, numpy.ascontiguousarray(temp_vision_frame)
+
+
+#todo: no review needed - [workflow] copy of v4 finalize_video task, temp clearing moved out into the clear task
+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:

tests/test_ffprobe.py

@@ -0,0 +1,38 @@
+import subprocess
+
+import pytest
+
+from facefusion import process_manager
+from facefusion.download import conditional_download
+from facefusion.ffprobe import extract_video_metadata
+from .helper import get_test_example_file, get_test_examples_directory
+
+
+@pytest.fixture(scope = 'module', autouse = True)
+def before_all() -> None:
+	process_manager.start()
+
+	conditional_download(get_test_examples_directory(),
+	[
+		'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/target-240p.mp4'
+	])
+
+	subprocess.run([ 'ffmpeg', '-i', get_test_example_file('target-240p.mp4'), '-t', '1', get_test_example_file('target-240p-1s.mov') ])
+
+
+def test_extract_video_metadata() -> None:
+	video_metadata = extract_video_metadata(get_test_example_file('target-240p.mp4'))
+
+	assert video_metadata.get('fps') == 25.0
+	assert video_metadata.get('duration') == 10.8
+	assert video_metadata.get('frame_total') == 270
+	assert video_metadata.get('resolution') == (426, 226)
+	assert video_metadata.get('bit_rate') == 138754
+	assert video_metadata.get('color_transfer') == 'smpte170m'
+
+	video_metadata = extract_video_metadata(get_test_example_file('target-240p-1s.mov'))
+
+	assert video_metadata.get('fps') == 25.0
+	assert video_metadata.get('duration') == 1.0
+	assert video_metadata.get('frame_total') == 25
+	assert video_metadata.get('resolution') == (426, 226)

tests/test_ffprobe_builder.py

@@ -0,0 +1,41 @@
+from shutil import which
+
+from facefusion import ffprobe_builder
+from facefusion.ffprobe_builder import chain, format_to_key_value, run, select_stream, set_input, show_format_entries, show_stream_entries
+
+
+def test_run() -> None:
+	assert run([ '-v', 'error' ]) == [ which('ffprobe'), '-loglevel', 'error', '-v', 'error' ]
+
+
+def test_chain() -> None:
+	assert chain(
+		ffprobe_builder.select_stream('v:0'),
+		ffprobe_builder.show_stream_entries([ 'duration' ]),
+		ffprobe_builder.format_to_key_value(),
+		ffprobe_builder.set_input('target.mp4')
+	) == [ '-select_streams', 'v:0', '-show_entries', 'stream=duration', '-of', 'default=noprint_wrappers=1', '-i', 'target.mp4' ]
+
+
+def test_select_stream() -> None:
+	assert select_stream('v:0') == [ '-select_streams', 'v:0' ]
+	assert select_stream('a:0') == [ '-select_streams', 'a:0' ]
+
+
+def test_show_stream_entries() -> None:
+	assert show_stream_entries([ 'duration' ]) == [ '-show_entries', 'stream=duration' ]
+	assert show_stream_entries([ 'duration', 'width' ]) == [ '-show_entries', 'stream=duration,width' ]
+
+
+def test_show_format_entries() -> None:
+	assert show_format_entries([ 'duration' ]) == [ '-show_entries', 'format=duration' ]
+	assert show_format_entries([ 'duration', 'bit_rate' ]) == [ '-show_entries', 'format=duration,bit_rate' ]
+
+
+def test_format_to_key_value() -> None:
+	assert format_to_key_value() == [ '-of', 'default=noprint_wrappers=1' ]
+
+
+def test_set_input() -> None:
+	assert set_input('input.mp3') == [ '-i', 'input.mp3' ]
+	assert set_input('input.wav') == [ '-i', 'input.wav' ]