mirror of
https://github.com/facefusion/facefusion.git
synced 2026-04-23 18:06:08 +02:00
Harisreedhar
ab24cd3f2e
* implement basic webrtc_stream * add aiortc to requirements.txt * update aiortc version * rename variables with rtc_ prefix * changes * changes * change helper to assert_helper and stream_helper * rename variables with rtc_ prefix * add error handling * return whole connection * remove monkey patch and some cleaning * cleanup * tiny adjustments * tiny adjustments * proper typing and naming for rtc offer set * - remove async from on_video_track method - rename source -> target - add audio * audio always before video --------- Co-authored-by: henryruhs <info@henryruhs.com>
333 lines
11 KiB
Python
333 lines
11 KiB
Python
import os
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
from facefusion.download import conditional_download
|
|
from facefusion.filesystem import copy_file, create_directory, get_file_extension
|
|
from facefusion.jobs.job_manager import add_step, clear_jobs, create_job, init_jobs, move_job_file, submit_job, submit_jobs
|
|
from facefusion.jobs.job_runner import collect_output_set, finalize_steps, retry_job, retry_jobs, run_job, run_jobs, run_steps
|
|
from facefusion.types import Args
|
|
from .assert_helper import get_test_example_file, get_test_examples_directory, get_test_jobs_directory, get_test_output_path, is_test_output_file, is_test_output_sequence, prepare_test_output_directory
|
|
|
|
|
|
@pytest.fixture(scope = 'module', autouse = True)
|
|
def before_all() -> None:
|
|
conditional_download(get_test_examples_directory(),
|
|
[
|
|
'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/source.jpg',
|
|
'https://github.com/facefusion/facefusion-assets/releases/download/examples-3.0.0/target-240p.mp4'
|
|
])
|
|
subprocess.run([ 'ffmpeg', '-i', get_test_example_file('target-240p.mp4'), '-vframes', '1', get_test_example_file('target-240p.jpg') ])
|
|
|
|
|
|
@pytest.fixture(scope = 'function', autouse = True)
|
|
def before_each() -> None:
|
|
clear_jobs(get_test_jobs_directory())
|
|
init_jobs(get_test_jobs_directory())
|
|
prepare_test_output_directory()
|
|
|
|
|
|
def process_step(job_id : str, step_index : int, step_args : Args) -> bool:
|
|
output_path = step_args.get('output_path')
|
|
target_path = step_args.get('target_path')
|
|
|
|
if output_path and not get_file_extension(output_path):
|
|
if create_directory(output_path):
|
|
return copy_file(target_path, os.path.join(output_path, os.path.basename(target_path)))
|
|
return False
|
|
|
|
return copy_file(target_path, output_path)
|
|
|
|
|
|
def test_run_job() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
args_2 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-2.mp4')
|
|
}
|
|
args_3 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-3.jpg')
|
|
}
|
|
args_4 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-4')
|
|
}
|
|
args_5 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-4')
|
|
}
|
|
|
|
assert run_job('job-invalid', process_step) is False
|
|
|
|
create_job('job-test-run-job')
|
|
add_step('job-test-run-job', args_1)
|
|
add_step('job-test-run-job', args_2)
|
|
add_step('job-test-run-job', args_2)
|
|
add_step('job-test-run-job', args_3)
|
|
add_step('job-test-run-job', args_4)
|
|
add_step('job-test-run-job', args_5)
|
|
|
|
assert run_job('job-test-run-job', process_step) is False
|
|
|
|
submit_job('job-test-run-job')
|
|
|
|
assert run_job('job-test-run-job', process_step) is True
|
|
|
|
|
|
def test_run_jobs() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
args_2 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-2.mp4')
|
|
}
|
|
args_3 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-3.jpg')
|
|
}
|
|
halt_on_error = True
|
|
|
|
assert run_jobs(process_step, halt_on_error) is False
|
|
|
|
create_job('job-test-run-jobs-1')
|
|
create_job('job-test-run-jobs-2')
|
|
add_step('job-test-run-jobs-1', args_1)
|
|
add_step('job-test-run-jobs-1', args_1)
|
|
add_step('job-test-run-jobs-2', args_2)
|
|
add_step('job-test-run-jobs-3', args_3)
|
|
|
|
assert run_jobs(process_step, halt_on_error) is False
|
|
|
|
submit_jobs(halt_on_error)
|
|
|
|
assert run_jobs(process_step, halt_on_error) is True
|
|
|
|
|
|
def test_retry_job() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
|
|
assert retry_job('job-invalid', process_step) is False
|
|
|
|
create_job('job-test-retry-job')
|
|
add_step('job-test-retry-job', args_1)
|
|
submit_job('job-test-retry-job')
|
|
|
|
assert retry_job('job-test-retry-job', process_step) is False
|
|
|
|
move_job_file('job-test-retry-job', 'failed')
|
|
|
|
assert retry_job('job-test-retry-job', process_step) is True
|
|
|
|
|
|
def test_retry_jobs() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
args_2 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-2.mp4')
|
|
}
|
|
args_3 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-3.jpg')
|
|
}
|
|
halt_on_error = True
|
|
|
|
assert retry_jobs(process_step, halt_on_error) is False
|
|
|
|
create_job('job-test-retry-jobs-1')
|
|
create_job('job-test-retry-jobs-2')
|
|
add_step('job-test-retry-jobs-1', args_1)
|
|
add_step('job-test-retry-jobs-1', args_1)
|
|
add_step('job-test-retry-jobs-2', args_2)
|
|
add_step('job-test-retry-jobs-3', args_3)
|
|
|
|
assert retry_jobs(process_step, halt_on_error) is False
|
|
|
|
move_job_file('job-test-retry-jobs-1', 'failed')
|
|
move_job_file('job-test-retry-jobs-2', 'failed')
|
|
|
|
assert retry_jobs(process_step, halt_on_error) is True
|
|
|
|
|
|
def test_run_steps() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
args_2 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-2.mp4')
|
|
}
|
|
args_3 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-3.jpg')
|
|
}
|
|
|
|
assert run_steps('job-invalid', process_step) is False
|
|
|
|
create_job('job-test-run-steps')
|
|
add_step('job-test-run-steps', args_1)
|
|
add_step('job-test-run-steps', args_1)
|
|
add_step('job-test-run-steps', args_2)
|
|
add_step('job-test-run-steps', args_3)
|
|
|
|
assert run_steps('job-test-run-steps', process_step) is True
|
|
|
|
|
|
def test_finalize_steps() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
args_2 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-2.mp4')
|
|
}
|
|
args_3 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-3.jpg')
|
|
}
|
|
args_4 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-4')
|
|
}
|
|
args_5 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-4')
|
|
}
|
|
|
|
create_job('job-test-finalize-steps')
|
|
add_step('job-test-finalize-steps', args_1)
|
|
add_step('job-test-finalize-steps', args_1)
|
|
add_step('job-test-finalize-steps', args_2)
|
|
add_step('job-test-finalize-steps', args_3)
|
|
add_step('job-test-finalize-steps', args_4)
|
|
add_step('job-test-finalize-steps', args_5)
|
|
|
|
copy_file(args_1.get('target_path'), get_test_output_path('output-1-job-test-finalize-steps-0.mp4'))
|
|
copy_file(args_1.get('target_path'), get_test_output_path('output-1-job-test-finalize-steps-1.mp4'))
|
|
copy_file(args_2.get('target_path'), get_test_output_path('output-2-job-test-finalize-steps-2.mp4'))
|
|
copy_file(args_3.get('target_path'), get_test_output_path('output-3-job-test-finalize-steps-3.jpg'))
|
|
|
|
temp_directory_1 = get_test_output_path('output-4-job-test-finalize-steps-4')
|
|
temp_directory_2 = get_test_output_path('output-4-job-test-finalize-steps-5')
|
|
create_directory(temp_directory_1)
|
|
create_directory(temp_directory_2)
|
|
copy_file(args_4.get('target_path'), os.path.join(temp_directory_1, '00000001.jpg'))
|
|
copy_file(args_5.get('target_path'), os.path.join(temp_directory_2, '00000002.jpg'))
|
|
|
|
assert finalize_steps('job-test-finalize-steps') is True
|
|
assert is_test_output_file('output-1.mp4') is True
|
|
assert is_test_output_file('output-2.mp4') is True
|
|
assert is_test_output_file('output-3.jpg') is True
|
|
assert is_test_output_sequence(get_test_output_path('output-4')) is True
|
|
|
|
|
|
def test_collect_output_set() -> None:
|
|
args_1 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-1.mp4')
|
|
}
|
|
args_2 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-2.mp4')
|
|
}
|
|
args_3 =\
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.jpg'),
|
|
'output_path': get_test_output_path('output-3.jpg')
|
|
}
|
|
args_4 = \
|
|
{
|
|
'source_path': get_test_example_file('source.jpg'),
|
|
'target_path': get_test_example_file('target-240p.mp4'),
|
|
'output_path': get_test_output_path('output-4')
|
|
}
|
|
|
|
create_job('job-test-collect-output-set')
|
|
add_step('job-test-collect-output-set', args_1)
|
|
add_step('job-test-collect-output-set', args_1)
|
|
add_step('job-test-collect-output-set', args_2)
|
|
add_step('job-test-collect-output-set', args_3)
|
|
add_step('job-test-collect-output-set', args_4)
|
|
|
|
output_set =\
|
|
{
|
|
get_test_output_path('output-1.mp4'):
|
|
[
|
|
get_test_output_path('output-1-job-test-collect-output-set-0.mp4'),
|
|
get_test_output_path('output-1-job-test-collect-output-set-1.mp4')
|
|
],
|
|
get_test_output_path('output-2.mp4'):
|
|
[
|
|
get_test_output_path('output-2-job-test-collect-output-set-2.mp4')
|
|
],
|
|
get_test_output_path('output-3.jpg'):
|
|
[
|
|
get_test_output_path('output-3-job-test-collect-output-set-3.jpg')
|
|
],
|
|
get_test_output_path('output-4'):
|
|
[
|
|
get_test_output_path('output-4-job-test-collect-output-set-4')
|
|
]
|
|
}
|
|
|
|
assert collect_output_set('job-test-collect-output-set') == output_set
|