Vibe coded benchmark command part3

facefusion/benchmarker.py

@@ -3,17 +3,17 @@ import os
 import statistics
 import tempfile
 from time import perf_counter
-from typing import Any, Dict, List
+from typing import Any, Dict, Generator, List
 
-from facefusion import state_manager
+from facefusion import core, state_manager
 from facefusion.cli_helper import render_table
 from facefusion.download import conditional_download, resolve_download_url
-from facefusion.filesystem import get_file_extension, is_video
+from facefusion.filesystem import get_file_extension
-from facefusion.memory import limit_system_memory
 from facefusion.vision import count_video_frame_total, detect_video_fps, detect_video_resolution, pack_resolution
 
 
-BENCHMARKS : Dict[str, str] = {
+BENCHMARKS : Dict[str, str] =\
+{
 	'240p': '.assets/examples/target-240p.mp4',
 	'360p': '.assets/examples/target-360p.mp4',
 	'540p': '.assets/examples/target-540p.mp4',
@@ -40,50 +40,7 @@ def pre_check() -> bool:
 	return True
 
 
-def suggest_output_path(target_path : str) -> str:
+def run() -> Generator[List[Any], None, None]:
-	if is_video(target_path):
-		target_file_extension = get_file_extension(target_path)
-		return os.path.join(tempfile.gettempdir(), hashlib.sha1().hexdigest()[:8] + target_file_extension)
-	return ''
-
-
-def pre_process() -> None:
-	system_memory_limit = state_manager.get_item('system_memory_limit')
-	if system_memory_limit and system_memory_limit > 0:
-		limit_system_memory(system_memory_limit)
-
-
-def benchmark_target(benchmark_cycles : int) -> List[Any]:
-	from facefusion.core import conditional_process
-
-	process_times = []
-	video_frame_total = count_video_frame_total(state_manager.get_item('target_path'))
-	output_video_resolution = detect_video_resolution(state_manager.get_item('target_path'))
-	state_manager.set_item('output_video_resolution', pack_resolution(output_video_resolution))
-	state_manager.set_item('output_video_fps', detect_video_fps(state_manager.get_item('target_path')))
-
-	conditional_process()
-	for index in range(benchmark_cycles):
-		start_time = perf_counter()
-		conditional_process()
-		end_time = perf_counter()
-		process_times.append(end_time - start_time)
-	average_run = round(statistics.mean(process_times), 2)
-	fastest_run = round(min(process_times), 2)
-	slowest_run = round(max(process_times), 2)
-	relative_fps = round(video_frame_total * benchmark_cycles / sum(process_times), 2)
-
-	return [
-		state_manager.get_item('target_path'),
-		benchmark_cycles,
-		average_run,
-		fastest_run,
-		slowest_run,
-		relative_fps
-	]
-
-
-def run() -> None:
 	benchmark_resolutions = state_manager.get_item('benchmark_resolutions')
 	benchmark_cycles = state_manager.get_item('benchmark_cycles')
 
@@ -95,21 +52,64 @@ def run() -> None:
 	state_manager.set_item('video_memory_strategy', 'tolerant')
 
 	benchmark_results = []
-	target_paths = [ BENCHMARKS[benchmark_resolution] for benchmark_resolution in benchmark_resolutions if benchmark_resolution in BENCHMARKS ]
+	target_paths = [ BENCHMARKS.get(benchmark_resolution) for benchmark_resolution in benchmark_resolutions if benchmark_resolution in BENCHMARKS ]
 
-	if target_paths:
-		pre_process()
 	for target_path in target_paths:
 		state_manager.set_item('target_path', target_path)
 		state_manager.set_item('output_path', suggest_output_path(state_manager.get_item('target_path')))
-			benchmark_results.append(benchmark_target(benchmark_cycles))
+		benchmark_results.append(cycle(benchmark_cycles))
+		yield benchmark_results
 
-	headers = [
+
-		'Target Path',
+def cycle(benchmark_cycles : int) -> List[Any]:
-		'Cycles',
+	process_times = []
-		'Average (s)',
+	video_frame_total = count_video_frame_total(state_manager.get_item('target_path'))
-		'Fastest (s)',
+	output_video_resolution = detect_video_resolution(state_manager.get_item('target_path'))
-		'Slowest (s)',
+	state_manager.set_item('output_video_resolution', pack_resolution(output_video_resolution))
-		'Relative FPS'
+	state_manager.set_item('output_video_fps', detect_video_fps(state_manager.get_item('target_path')))
+
+	core.conditional_process()
+
+	for index in range(benchmark_cycles):
+		start_time = perf_counter()
+		core.conditional_process()
+		end_time = perf_counter()
+		process_times.append(end_time - start_time)
+
+	average_run = round(statistics.mean(process_times), 2)
+	fastest_run = round(min(process_times), 2)
+	slowest_run = round(max(process_times), 2)
+	relative_fps = round(video_frame_total * benchmark_cycles / sum(process_times), 2)
+
+	return\
+	[
+		state_manager.get_item('target_path'),
+		benchmark_cycles,
+		average_run,
+		fastest_run,
+		slowest_run,
+		relative_fps
 	]
+
+
+def suggest_output_path(target_path : str) -> str:
+	target_file_extension = get_file_extension(target_path)
+	return os.path.join(tempfile.gettempdir(), hashlib.sha1().hexdigest()[:8] + target_file_extension)
+
+
+def render() -> None:
+	benchmark_results = []
+	headers = \
+	[
+		'target_path',
+		'benchmark_cycles',
+		'average_run',
+		'fastest_run',
+		'slowest_run',
+		'relative_fps'
+	]
+
+	for cycle_result in run():
+		benchmark_results = cycle_result
+
 	render_table(headers, benchmark_results)

facefusion/core.py

@@ -6,7 +6,7 @@ from time import time
 
 import numpy
 
-from facefusion import cli_helper, content_analyser, face_classifier, face_detector, face_landmarker, face_masker, face_recognizer, logger, process_manager, state_manager, video_manager, voice_extractor, wording
+from facefusion import benchmarker, cli_helper, content_analyser, face_classifier, face_detector, face_landmarker, face_masker, face_recognizer, logger, process_manager, state_manager, video_manager, voice_extractor, wording
 from facefusion.args import apply_args, collect_job_args, reduce_job_args, reduce_step_args
 from facefusion.common_helper import get_first
 from facefusion.content_analyser import analyse_image, analyse_video
@@ -60,14 +60,9 @@ def route(args : Args) -> None:
 		return hard_exit(error_code)
 
 	if state_manager.get_item('command') == 'benchmark':
-		import facefusion.benchmarker as benchmarker
+		if not common_pre_check() or not processors_pre_check() or not benchmarker.pre_check():
-		
-		if not common_pre_check() or not processors_pre_check():
 			return hard_exit(2)
-		if not benchmarker.pre_check():
+		benchmarker.render()
-			return hard_exit(2)
-		benchmarker.run()
-		return
 
 	if state_manager.get_item('command') in [ 'job-list', 'job-create', 'job-submit', 'job-submit-all', 'job-delete', 'job-delete-all', 'job-add-step', 'job-remix-step', 'job-insert-step', 'job-remove-step' ]:
 		if not job_manager.init_jobs(state_manager.get_item('jobs_path')):

facefusion/uis/components/benchmark.py

@@ -2,8 +2,7 @@ from typing import Any, Generator, List, Optional
 
 import gradio
 
-from facefusion import state_manager, wording
+from facefusion import benchmarker, state_manager, wording
-from facefusion.benchmarker import BENCHMARKS, benchmark_target, pre_process, suggest_output_path
 from facefusion.uis.core import get_ui_component
 
 BENCHMARK_BENCHMARKS_DATAFRAME : Optional[gradio.Dataframe] = None
@@ -51,25 +50,10 @@ def listen() -> None:
 
 
 def start(benchmark_resolutions : List[str], benchmark_cycles : int) -> Generator[List[Any], None, None]:
-	state_manager.set_item('source_paths', [ '.assets/examples/source.jpg', '.assets/examples/source.mp3' ])
+	state_manager.set_item('benchmark_resolutions', benchmark_resolutions)
-	state_manager.set_item('face_landmarker_score', 0)
+	state_manager.set_item('benchmark_cycles', benchmark_cycles)
-	state_manager.set_item('temp_frame_format', 'bmp')
-	state_manager.set_item('output_audio_volume', 0)
-	state_manager.set_item('output_video_preset', 'ultrafast')
-	state_manager.set_item('video_memory_strategy', 'tolerant')
 	state_manager.sync_item('execution_providers')
 	state_manager.sync_item('execution_thread_count')
 	state_manager.sync_item('execution_queue_count')
-	state_manager.sync_item('system_memory_limit')
-	benchmark_results = []
-	target_paths = [ BENCHMARKS[benchmark_resolution] for benchmark_resolution in benchmark_resolutions if benchmark_resolution in BENCHMARKS ]
-
-	if target_paths:
-		pre_process()
-		for target_path in target_paths:
-			state_manager.set_item('target_path', target_path)
-			state_manager.set_item('output_path', suggest_output_path(state_manager.get_item('target_path')))
-			benchmark_results.append(benchmark_target(benchmark_cycles))
-			yield benchmark_results
-
 
+	yield from benchmarker.run()

facefusion/wording.py

@@ -197,6 +197,9 @@ WORDING : Dict[str, Any] =\
 		# download
 		'download_providers': 'download using different providers (choices: {choices}, ...)',
 		'download_scope': 'specify the download scope',
+		# benchmark
+		'benchmark_resolutions': 'choose the resolution for the benchmarks (choices: {choices}, ...)',
+		'benchmark_cycles': 'specify the number of benchmark cycles',
 		# memory
 		'video_memory_strategy': 'balance fast processing and low VRAM usage',
 		'system_memory_limit': 'limit the available RAM that can be used while processing',
@@ -208,10 +211,7 @@ WORDING : Dict[str, Any] =\
 		'headless_run': 'run the program in headless mode',
 		'batch_run': 'run the program in batch mode',
 		'force_download': 'force automate downloads and exit',
-		'benchmark': 'run performance benchmarks and exit',
+		'benchmark': 'run performance benchmarks',
-		# benchmark
-		'benchmark_resolutions': 'choose the resolution for the benchmark runs (choices: {choices}, ...)',
-		'benchmark_cycles': 'specify the number of benchmark cycles',
 		# jobs
 		'job_id': 'specify the job id',
 		'job_status': 'specify the job status',