Introduce Commands type

facefusion/download.py

@@ -11,12 +11,11 @@ import facefusion.choices
 from facefusion import logger, process_manager, state_manager, wording
 from facefusion.filesystem import get_file_name, get_file_size, is_file, remove_file
 from facefusion.hash_helper import validate_hash
-from facefusion.typing import DownloadProvider, DownloadSet
+from facefusion.typing import Commands, DownloadProvider, DownloadSet
 
 
-def open_curl(args : List[str]) -> subprocess.Popen[bytes]:
+def open_curl(commands : Commands) -> subprocess.Popen[bytes]:
-	commands = [ shutil.which('curl'), '--silent', '--insecure', '--location' ]
+	commands = [ shutil.which('curl'), '--silent', '--insecure', '--location' ] + commands
-	commands.extend(args)
 	return subprocess.Popen(commands, stdin = subprocess.PIPE, stdout = subprocess.PIPE)
 
 

facefusion/ffmpeg.py

@@ -9,14 +9,13 @@ from tqdm import tqdm
 from facefusion import logger, process_manager, state_manager, wording
 from facefusion.filesystem import get_file_format, remove_file
 from facefusion.temp_helper import get_temp_file_path, get_temp_frames_pattern, resolve_temp_frame_paths
-from facefusion.typing import AudioBuffer, Fps, OutputVideoPreset, UpdateProgress
+from facefusion.typing import AudioBuffer, Commands, Fps, OutputVideoPreset, UpdateProgress
 from facefusion.vision import count_trim_frame_total, detect_video_duration, restrict_video_fps
 
 
-def run_ffmpeg_with_progress(args: List[str], update_progress : UpdateProgress) -> subprocess.Popen[bytes]:
+def run_ffmpeg_with_progress(commands : Commands, update_progress : UpdateProgress) -> subprocess.Popen[bytes]:
 	log_level = state_manager.get_item('log_level')
-	commands = [ shutil.which('ffmpeg'), '-hide_banner', '-nostats', '-loglevel', 'error', '-progress', '-' ]
+	commands = [ shutil.which('ffmpeg'), '-hide_banner', '-nostats', '-loglevel', 'error', '-progress', '-' ] + commands
-	commands.extend(args)
 	process = subprocess.Popen(commands, stderr = subprocess.PIPE, stdout = subprocess.PIPE)
 
 	while process_manager.is_processing():
@@ -39,10 +38,9 @@ def run_ffmpeg_with_progress(args: List[str], update_progress : UpdateProgress)
 	return process
 
 
-def run_ffmpeg(args : List[str]) -> subprocess.Popen[bytes]:
+def run_ffmpeg(commands : Commands) -> subprocess.Popen[bytes]:
 	log_level = state_manager.get_item('log_level')
-	commands = [ shutil.which('ffmpeg'), '-hide_banner', '-nostats', '-loglevel', 'error' ]
+	commands = [ shutil.which('ffmpeg'), '-hide_banner', '-nostats', '-loglevel', 'error' ] + commands
-	commands.extend(args)
 	process = subprocess.Popen(commands, stderr = subprocess.PIPE, stdout = subprocess.PIPE)
 
 	while process_manager.is_processing():
@@ -59,9 +57,8 @@ def run_ffmpeg(args : List[str]) -> subprocess.Popen[bytes]:
 	return process
 
 
-def open_ffmpeg(args : List[str]) -> subprocess.Popen[bytes]:
+def open_ffmpeg(commands : Commands) -> subprocess.Popen[bytes]:
-	commands = [ shutil.which('ffmpeg'), '-loglevel', 'quiet' ]
+	commands = [ shutil.which('ffmpeg'), '-loglevel', 'quiet' ] + commands
-	commands.extend(args)
 	return subprocess.Popen(commands, stdin = subprocess.PIPE, stdout = subprocess.PIPE)
 
 

facefusion/typing.py

@@ -85,6 +85,8 @@ ProcessStep = Callable[[str, int, Args], bool]
 
 Content = Dict[str, Any]
 
+Commands = List[str]
+
 WarpTemplate = Literal['arcface_112_v1', 'arcface_112_v2', 'arcface_128_v2', 'dfl_whole_face', 'ffhq_512', 'mtcnn_512', 'styleganex_384']
 WarpTemplateSet = Dict[WarpTemplate, NDArray[Any]]
 ProcessMode = Literal['output', 'preview', 'stream']