diff --git a/facefusion/ffprobe.py b/facefusion/ffprobe.py index 72d605a3..22d05db2 100644 --- a/facefusion/ffprobe.py +++ b/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) ) diff --git a/facefusion/ffprobe_builder.py b/facefusion/ffprobe_builder.py index 7aed6bb5..3c9aa923 100644 --- a/facefusion/ffprobe_builder.py +++ b/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 ] diff --git a/tests/test_ffprobe.py b/tests/test_ffprobe.py new file mode 100644 index 00000000..317240a6 --- /dev/null +++ b/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) diff --git a/tests/test_ffprobe_builder.py b/tests/test_ffprobe_builder.py new file mode 100644 index 00000000..b21aafd5 --- /dev/null +++ b/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' ]