refine ffprobe methods and adopt tests from v4

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

facefusion/ffprobe.py

@@ -27,23 +27,11 @@ def parse_entries(output : Buffer) -> Dict[str, str]:
 	return media_entries
 
 
-#todo: no review needed - [probing] copy of v4 with parsing moved to parse_entries
-def probe_entries(media_path : str, entries : List[str]) -> Dict[str, str]:
-	commands = ffprobe_builder.chain(
-		ffprobe_builder.show_entries(entries),
-		ffprobe_builder.format_to_key_value(),
-		ffprobe_builder.set_input(media_path)
-	)
-	output, _ = run_ffprobe(commands).communicate()
-
-	return parse_entries(output)
-
-
-#todo: needs review - [probing] [critical: low] new against v4, probes the first video stream
+#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_video_stream(),
-		ffprobe_builder.show_entries(entries),
+		ffprobe_builder.select_stream('v:0'),
+		ffprobe_builder.show_stream_entries(entries),
 		ffprobe_builder.format_to_key_value(),
 		ffprobe_builder.set_input(video_path)
 	)

facefusion/ffprobe_builder.py

@@ -5,36 +5,36 @@ from typing import List
 from facefusion.types import Command
 
 
-#todo: no review needed - [probing] 100% copy of v4 ffprobe_builder except select_video_stream and show_format_entries
+#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, selects the first video stream
-def select_video_stream() -> List[Command]:
-	return [ '-select_streams', 'v:0' ]
+#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 ]
 
 
-def show_entries(entries : List[str]) -> List[Command]:
+#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 format 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) ]
 
 
-def format_to_value() -> List[Command]:
-	return [ '-of', 'default=noprint_wrappers=1:nokey=1' ]
-
-
+#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 ]

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' ]