mirror of
https://github.com/facefusion/facefusion.git
synced 2026-07-27 04:21:03 +02:00
go v4 style for workflow (#1197)
* go v4 style for workflow * remove some todos
This commit is contained in:
Henry Ruhs
@@ -1,6 +1,18 @@
|
||||
from facefusion import logger, process_manager, state_manager, translator
|
||||
from facefusion.temp_helper import clear_temp_directory, create_temp_directory
|
||||
from facefusion.types import ErrorCode
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from typing import List
|
||||
|
||||
import numpy
|
||||
from tqdm import tqdm
|
||||
|
||||
import facefusion.workflows.image_to_video as image_to_video
|
||||
from facefusion import ffprobe, logger, process_manager, state_manager, translator, video_manager
|
||||
from facefusion.audio import create_empty_audio_frame, get_audio_frame, get_voice_frame
|
||||
from facefusion.common_helper import get_first
|
||||
from facefusion.filesystem import filter_audio_paths
|
||||
from facefusion.processors.core import get_processors_modules
|
||||
from facefusion.temp_helper import clear_temp_directory, create_temp_directory, resolve_temp_frame_set
|
||||
from facefusion.types import AudioFrame, ErrorCode, VisionFrame
|
||||
from facefusion.vision import conditional_merge_vision_mask, detect_video_resolution, extract_vision_mask, read_static_image, read_static_images, read_static_video_frame, restrict_trim_frame, restrict_video_fps, restrict_video_resolution, scale_resolution
|
||||
|
||||
|
||||
def is_process_stopping() -> bool:
|
||||
@@ -20,3 +32,143 @@ def clear() -> ErrorCode:
|
||||
if clear_temp_directory(state_manager.get_item('target_path')):
|
||||
logger.debug(translator.get('clearing_temp'), __name__)
|
||||
return 0
|
||||
|
||||
|
||||
#todo: copy - renamed workflow to workflow_mode
|
||||
def conditional_get_reference_vision_frame() -> VisionFrame:
|
||||
if state_manager.get_item('workflow_mode') == 'image-to-video':
|
||||
return read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
|
||||
return read_static_image(state_manager.get_item('target_path'))
|
||||
|
||||
|
||||
#todo: copy - renamed workflow to workflow_mode, video only, trim offset
|
||||
def conditional_get_source_audio_frame(frame_number : int) -> AudioFrame:
|
||||
if state_manager.get_item('workflow_mode') == 'image-to-video':
|
||||
trim_frame_start, _ = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
|
||||
source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
|
||||
source_audio_frame = get_audio_frame(source_audio_path, temp_video_fps, frame_number - trim_frame_start)
|
||||
|
||||
if numpy.any(source_audio_frame):
|
||||
return source_audio_frame
|
||||
|
||||
return create_empty_audio_frame()
|
||||
|
||||
|
||||
#todo: copy - renamed workflow to workflow_mode, video only, trim offset
|
||||
def conditional_get_source_voice_frame(frame_number : int) -> AudioFrame:
|
||||
if state_manager.get_item('workflow_mode') == 'image-to-video':
|
||||
trim_frame_start, _ = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
|
||||
source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
|
||||
source_voice_frame = get_voice_frame(source_audio_path, temp_video_fps, frame_number - trim_frame_start)
|
||||
|
||||
if numpy.any(source_voice_frame):
|
||||
return source_voice_frame
|
||||
|
||||
return create_empty_audio_frame()
|
||||
|
||||
|
||||
#todo: needs review - [workflow] [critical: medium] copy of process_temp_frame, receives vision frames instead of paths
|
||||
def process_target_frame(frame_number : int, target_vision_frames : List[VisionFrame], temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
reference_vision_frame = conditional_get_reference_vision_frame()
|
||||
source_vision_frames = read_static_images(state_manager.get_item('source_paths'))
|
||||
source_audio_frame = conditional_get_source_audio_frame(frame_number)
|
||||
source_voice_frame = conditional_get_source_voice_frame(frame_number)
|
||||
temp_vision_mask = extract_vision_mask(temp_vision_frame)
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
temp_vision_frame, temp_vision_mask = processor_module.process_frame(
|
||||
{
|
||||
'reference_vision_frame': reference_vision_frame,
|
||||
'source_vision_frames': source_vision_frames,
|
||||
'source_audio_frame': source_audio_frame,
|
||||
'source_voice_frame': source_voice_frame,
|
||||
'target_vision_frames': target_vision_frames,
|
||||
'temp_vision_frame': temp_vision_frame[:, :, :3],
|
||||
'temp_vision_mask': temp_vision_mask
|
||||
})
|
||||
|
||||
return conditional_merge_vision_mask(temp_vision_frame, temp_vision_mask)
|
||||
|
||||
|
||||
#todo: needs review - [workflow] [critical: medium] copy of process_frames, uses temp_frame_set and warms the reference frame
|
||||
def process_disk_frames() -> ErrorCode:
|
||||
temp_frame_set = resolve_temp_frame_set(state_manager.get_item('target_path'))
|
||||
|
||||
if temp_frame_set:
|
||||
with tqdm(total = len(temp_frame_set), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
|
||||
progress.set_postfix(execution_providers = state_manager.get_item('execution_providers'))
|
||||
|
||||
read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
|
||||
|
||||
with ThreadPoolExecutor(max_workers = state_manager.get_item('execution_thread_count')) as executor:
|
||||
futures = []
|
||||
|
||||
for frame_number, temp_frame_path in temp_frame_set.items():
|
||||
future = executor.submit(image_to_video.process_disk_frame, temp_frame_path, frame_number, temp_frame_set)
|
||||
futures.append(future)
|
||||
|
||||
for future in as_completed(futures):
|
||||
if is_process_stopping():
|
||||
for pending_future in futures:
|
||||
pending_future.cancel()
|
||||
|
||||
if not future.cancelled():
|
||||
future.result()
|
||||
progress.update()
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
processor_module.post_process()
|
||||
|
||||
if is_process_stopping():
|
||||
return 4
|
||||
else:
|
||||
logger.error(translator.get('temp_frames_not_found'), __name__)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
#todo: needs review - [streaming] [critical: high] ordered submit and drain keeps writes in order, failed close stops the process manager
|
||||
def process_stream_frames() -> ErrorCode:
|
||||
trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
output_video_resolution = scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
|
||||
temp_video_resolution = restrict_video_resolution(state_manager.get_item('target_path'), output_video_resolution)
|
||||
temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
|
||||
frame_range = range(trim_frame_start, trim_frame_end)
|
||||
|
||||
if frame_range:
|
||||
video_metadata = ffprobe.extract_static_video_metadata(state_manager.get_item('target_path'))
|
||||
video_writer = video_manager.get_writer(state_manager.get_item('target_path'), video_metadata, temp_video_fps, temp_video_resolution, output_video_resolution, state_manager.get_item('output_video_fps'))
|
||||
frame_look_ahead = image_to_video.calculate_frame_look_ahead(temp_video_resolution)
|
||||
|
||||
with tqdm(total = len(frame_range), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
|
||||
progress.set_postfix(execution_providers = state_manager.get_item('execution_providers'))
|
||||
|
||||
read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
|
||||
|
||||
with ThreadPoolExecutor(max_workers = state_manager.get_item('execution_thread_count')) as executor:
|
||||
futures = []
|
||||
|
||||
for frame_number in frame_range:
|
||||
future = executor.submit(image_to_video.process_stream_frame, frame_number, temp_video_resolution)
|
||||
futures.append(future)
|
||||
|
||||
if len(futures) > frame_look_ahead:
|
||||
image_to_video.write_stream_frame(video_writer, futures, progress)
|
||||
|
||||
for _ in list(futures):
|
||||
image_to_video.write_stream_frame(video_writer, futures, progress)
|
||||
|
||||
if not video_manager.close_video_writer(video_writer):
|
||||
process_manager.stop()
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
processor_module.post_process()
|
||||
|
||||
if is_process_stopping():
|
||||
return 4
|
||||
else:
|
||||
logger.error(translator.get('temp_frames_not_found'), __name__)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
@@ -1,14 +1,13 @@
|
||||
from functools import partial
|
||||
|
||||
from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator
|
||||
from facefusion.audio import create_empty_audio_frame
|
||||
from facefusion.filesystem import is_image
|
||||
from facefusion.processors.core import get_processors_modules
|
||||
from facefusion.temp_helper import get_temp_file_path
|
||||
from facefusion.time_helper import calculate_end_time
|
||||
from facefusion.types import ErrorCode
|
||||
from facefusion.vision import conditional_merge_vision_mask, detect_image_resolution, extract_vision_mask, pack_resolution, read_static_image, read_static_images, restrict_image_resolution, scale_resolution, write_image
|
||||
from facefusion.workflows.core import clear, is_process_stopping, setup
|
||||
from facefusion.vision import detect_image_resolution, pack_resolution, read_static_image, restrict_image_resolution, scale_resolution, write_image
|
||||
from facefusion.workflows.core import clear, is_process_stopping, process_target_frame, setup
|
||||
|
||||
|
||||
def process(start_time : float) -> ErrorCode:
|
||||
@@ -57,33 +56,14 @@ def prepare_image() -> ErrorCode:
|
||||
|
||||
def process_image() -> ErrorCode:
|
||||
temp_image_path = get_temp_file_path(state_manager.get_item('target_path'))
|
||||
reference_vision_frame = read_static_image(state_manager.get_item('target_path'))
|
||||
source_vision_frames = read_static_images(state_manager.get_item('source_paths'))
|
||||
source_audio_frame = create_empty_audio_frame()
|
||||
source_voice_frame = create_empty_audio_frame()
|
||||
target_vision_frame = read_static_image(state_manager.get_item('target_path'))
|
||||
temp_vision_frame = read_static_image(temp_image_path, 'rgba')
|
||||
temp_vision_mask = extract_vision_mask(temp_vision_frame)
|
||||
temp_vision_frame = process_target_frame(0, [ target_vision_frame ], temp_vision_frame)
|
||||
write_image(temp_image_path, temp_vision_frame)
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
logger.info(translator.get('processing'), processor_module.__name__)
|
||||
|
||||
temp_vision_frame, temp_vision_mask = processor_module.process_frame(
|
||||
{
|
||||
'reference_vision_frame': reference_vision_frame,
|
||||
'source_vision_frames': source_vision_frames,
|
||||
'source_audio_frame': source_audio_frame,
|
||||
'source_voice_frame': source_voice_frame,
|
||||
'target_vision_frames': [ target_vision_frame ],
|
||||
'temp_vision_frame': temp_vision_frame[:, :, :3],
|
||||
'temp_vision_mask': temp_vision_mask
|
||||
})
|
||||
|
||||
processor_module.post_process()
|
||||
|
||||
temp_vision_frame = conditional_merge_vision_mask(temp_vision_frame, temp_vision_mask)
|
||||
write_image(temp_image_path, temp_vision_frame)
|
||||
|
||||
if is_process_stopping():
|
||||
return 4
|
||||
return 0
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
||||
from concurrent.futures import Future
|
||||
from functools import partial
|
||||
from typing import List, Tuple
|
||||
|
||||
@@ -6,42 +6,41 @@ import cv2
|
||||
import numpy
|
||||
from tqdm import tqdm
|
||||
|
||||
from facefusion import content_analyser, ffmpeg, ffprobe, logger, process_manager, state_manager, translator, video_manager
|
||||
from facefusion.audio import create_empty_audio_frame, get_audio_frame, get_voice_frame
|
||||
import facefusion.workflows.core as core
|
||||
from facefusion import content_analyser, ffmpeg, logger, process_manager, state_manager, translator, video_manager
|
||||
from facefusion.common_helper import get_first, get_middle
|
||||
from facefusion.filesystem import filter_audio_paths, is_video
|
||||
from facefusion.processors.core import get_processors_modules
|
||||
from facefusion.temp_helper import move_temp_file, resolve_temp_frame_set
|
||||
from facefusion.temp_helper import move_temp_file
|
||||
from facefusion.time_helper import calculate_end_time
|
||||
from facefusion.types import ErrorCode, FrameSet, Resolution, VideoWriter, VisionFrame
|
||||
from facefusion.vision import conditional_merge_vision_mask, detect_video_resolution, extract_vision_mask, pack_resolution, read_static_image, read_static_images, read_static_video_frame, restrict_trim_frame, restrict_video_fps, restrict_video_resolution, scale_resolution, select_video_frames, write_image
|
||||
from facefusion.workflows.core import clear, is_process_stopping, setup
|
||||
from facefusion.vision import detect_video_resolution, pack_resolution, read_static_image, restrict_trim_frame, restrict_video_fps, restrict_video_resolution, scale_resolution, select_video_frames, write_image
|
||||
|
||||
|
||||
#todo: copy - task list adds workflow_strategy disk and stream branching
|
||||
def process(start_time : float) -> ErrorCode:
|
||||
tasks =\
|
||||
[
|
||||
analyse_video,
|
||||
clear,
|
||||
setup
|
||||
core.clear,
|
||||
core.setup
|
||||
]
|
||||
|
||||
if state_manager.get_item('workflow_strategy') == 'disk':
|
||||
tasks.extend(
|
||||
[
|
||||
extract_frames,
|
||||
process_frames,
|
||||
core.process_disk_frames,
|
||||
merge_frames
|
||||
])
|
||||
|
||||
if state_manager.get_item('workflow_strategy') == 'stream':
|
||||
tasks.append(process_stream_frames)
|
||||
tasks.append(core.process_stream_frames)
|
||||
|
||||
tasks.extend(
|
||||
[
|
||||
restore_audio,
|
||||
partial(finalize_video, start_time),
|
||||
clear
|
||||
core.clear
|
||||
])
|
||||
process_manager.start()
|
||||
|
||||
@@ -56,6 +55,7 @@ def process(start_time : float) -> ErrorCode:
|
||||
return 0
|
||||
|
||||
|
||||
#todo: copy - restrict_trim_video_frame renamed to restrict_trim_frame
|
||||
def analyse_video() -> ErrorCode:
|
||||
trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
|
||||
@@ -64,6 +64,7 @@ def analyse_video() -> ErrorCode:
|
||||
return 0
|
||||
|
||||
|
||||
#todo: copy - renamed from create_temp_frames, ffmpeg.extract_frames without output_path
|
||||
def extract_frames() -> ErrorCode:
|
||||
trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
output_video_resolution = scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
|
||||
@@ -74,50 +75,14 @@ def extract_frames() -> ErrorCode:
|
||||
if ffmpeg.extract_frames(state_manager.get_item('target_path'), temp_video_resolution, temp_video_fps, trim_frame_start, trim_frame_end):
|
||||
logger.debug(translator.get('extracting_frames_succeeded'), __name__)
|
||||
else:
|
||||
if is_process_stopping():
|
||||
if core.is_process_stopping():
|
||||
return 4
|
||||
logger.error(translator.get('extracting_frames_failed'), __name__)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
#todo: needs review - [workflow] [critical: medium] warms the reference frame before the executor, neighbors come from temp frames
|
||||
def process_frames() -> ErrorCode:
|
||||
temp_frame_set = resolve_temp_frame_set(state_manager.get_item('target_path'))
|
||||
|
||||
if temp_frame_set:
|
||||
with tqdm(total = len(temp_frame_set), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
|
||||
progress.set_postfix(execution_providers = state_manager.get_item('execution_providers'))
|
||||
|
||||
read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
|
||||
|
||||
with ThreadPoolExecutor(max_workers = state_manager.get_item('execution_thread_count')) as executor:
|
||||
futures = []
|
||||
|
||||
for frame_number, temp_frame_path in temp_frame_set.items():
|
||||
future = executor.submit(process_disk_frame, temp_frame_path, frame_number, temp_frame_set)
|
||||
futures.append(future)
|
||||
|
||||
for future in as_completed(futures):
|
||||
if is_process_stopping():
|
||||
for __future__ in futures:
|
||||
__future__.cancel()
|
||||
|
||||
if not future.cancelled():
|
||||
future.result()
|
||||
progress.update()
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
processor_module.post_process()
|
||||
|
||||
if is_process_stopping():
|
||||
return 4
|
||||
else:
|
||||
logger.error(translator.get('temp_frames_not_found'), __name__)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
#todo: copy - dropped conditional resolution and fps helpers
|
||||
def merge_frames() -> ErrorCode:
|
||||
trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
output_video_resolution = scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
|
||||
@@ -127,13 +92,14 @@ def merge_frames() -> ErrorCode:
|
||||
if ffmpeg.merge_video(state_manager.get_item('target_path'), temp_video_fps, output_video_resolution, state_manager.get_item('output_video_fps'), trim_frame_start, trim_frame_end):
|
||||
logger.debug(translator.get('merging_video_succeeded'), __name__)
|
||||
else:
|
||||
if is_process_stopping():
|
||||
if core.is_process_stopping():
|
||||
return 4
|
||||
logger.error(translator.get('merging_video_failed'), __name__)
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
#todo: copy - clear_video_pool inline, move_temp_file takes target_path
|
||||
def restore_audio() -> ErrorCode:
|
||||
trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
|
||||
@@ -148,7 +114,7 @@ def restore_audio() -> ErrorCode:
|
||||
logger.debug(translator.get('replacing_audio_succeeded'), __name__)
|
||||
else:
|
||||
video_manager.clear_video_pool()
|
||||
if is_process_stopping():
|
||||
if core.is_process_stopping():
|
||||
return 4
|
||||
logger.warn(translator.get('replacing_audio_skipped'), __name__)
|
||||
move_temp_file(state_manager.get_item('target_path'), state_manager.get_item('output_path'))
|
||||
@@ -158,45 +124,13 @@ def restore_audio() -> ErrorCode:
|
||||
logger.debug(translator.get('restoring_audio_succeeded'), __name__)
|
||||
else:
|
||||
video_manager.clear_video_pool()
|
||||
if is_process_stopping():
|
||||
if core.is_process_stopping():
|
||||
return 4
|
||||
logger.warn(translator.get('restoring_audio_skipped'), __name__)
|
||||
move_temp_file(state_manager.get_item('target_path'), state_manager.get_item('output_path'))
|
||||
return 0
|
||||
|
||||
|
||||
#todo: needs review - [workflow] [critical: medium] shared processing path for disk and stream, receives vision frames instead of paths
|
||||
def process_target_frame(frame_number : int, target_vision_frames : List[VisionFrame], temp_vision_frame : VisionFrame) -> VisionFrame:
|
||||
trim_frame_start, _ = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
reference_vision_frame = read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
|
||||
source_vision_frames = read_static_images(state_manager.get_item('source_paths'))
|
||||
source_audio_path = get_first(filter_audio_paths(state_manager.get_item('source_paths')))
|
||||
temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
|
||||
temp_vision_mask = extract_vision_mask(temp_vision_frame)
|
||||
|
||||
source_audio_frame = get_audio_frame(source_audio_path, temp_video_fps, frame_number - trim_frame_start)
|
||||
source_voice_frame = get_voice_frame(source_audio_path, temp_video_fps, frame_number - trim_frame_start)
|
||||
|
||||
if not numpy.any(source_audio_frame):
|
||||
source_audio_frame = create_empty_audio_frame()
|
||||
if not numpy.any(source_voice_frame):
|
||||
source_voice_frame = create_empty_audio_frame()
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
temp_vision_frame, temp_vision_mask = processor_module.process_frame(
|
||||
{
|
||||
'reference_vision_frame': reference_vision_frame,
|
||||
'source_vision_frames': source_vision_frames,
|
||||
'source_audio_frame': source_audio_frame,
|
||||
'source_voice_frame': source_voice_frame,
|
||||
'target_vision_frames': target_vision_frames,
|
||||
'temp_vision_frame': temp_vision_frame[:, :, :3],
|
||||
'temp_vision_mask': temp_vision_mask
|
||||
})
|
||||
|
||||
return conditional_merge_vision_mask(temp_vision_frame, temp_vision_mask)
|
||||
|
||||
|
||||
#todo: needs review - [correctness] [critical: high] missing neighbor frames resolve to none paths and rely on read_static_image returning none
|
||||
def resolve_temp_vision_frames(frame_number : int, frame_amount : int, temp_frame_set : FrameSet) -> List[VisionFrame]:
|
||||
temp_vision_frames = []
|
||||
@@ -212,18 +146,10 @@ def resolve_temp_vision_frames(frame_number : int, frame_amount : int, temp_fram
|
||||
def process_disk_frame(temp_frame_path : str, frame_number : int, temp_frame_set : FrameSet) -> bool:
|
||||
target_vision_frames = resolve_temp_vision_frames(frame_number, state_manager.get_item('target_frame_amount'), temp_frame_set)
|
||||
temp_vision_frame = read_static_image(temp_frame_path, 'rgba')
|
||||
temp_vision_frame = process_target_frame(frame_number, target_vision_frames, temp_vision_frame)
|
||||
temp_vision_frame = core.process_target_frame(frame_number, target_vision_frames, temp_vision_frame)
|
||||
return write_image(temp_frame_path, temp_vision_frame)
|
||||
|
||||
|
||||
#todo: needs review - [memory] [critical: high] frame look ahead bound by a hardcoded 3gb budget
|
||||
def calculate_frame_look_ahead(temp_video_resolution : Resolution) -> int:
|
||||
width, height = temp_video_resolution
|
||||
frame_memory_budget = 3 * 1024 ** 3
|
||||
frame_memory_usage = width * height * 4 * 6
|
||||
return min(state_manager.get_item('execution_thread_count') * 2, max(2, frame_memory_budget // frame_memory_usage))
|
||||
|
||||
|
||||
#todo: needs review - [streaming] [critical: high] middle frame resized to temp resolution before processing, channels follow temp_pixel_format to match the writer pipe
|
||||
def process_stream_frame(frame_number : int, temp_video_resolution : Resolution) -> Tuple[int, VisionFrame]:
|
||||
target_vision_frames = select_video_frames(state_manager.get_item('target_path'), frame_number, state_manager.get_item('target_frame_amount'))
|
||||
@@ -232,7 +158,7 @@ def process_stream_frame(frame_number : int, temp_video_resolution : Resolution)
|
||||
|
||||
if not (target_vision_frame.shape[1], target_vision_frame.shape[0]) == temp_video_resolution:
|
||||
temp_vision_frame = cv2.resize(target_vision_frame, temp_video_resolution)
|
||||
temp_vision_frame = process_target_frame(frame_number, target_vision_frames, temp_vision_frame)
|
||||
temp_vision_frame = core.process_target_frame(frame_number, target_vision_frames, temp_vision_frame)
|
||||
|
||||
if state_manager.get_item('temp_pixel_format') == 'bgra':
|
||||
temp_vision_frame = cv2.cvtColor(temp_vision_frame, cv2.COLOR_BGR2BGRA)
|
||||
@@ -243,56 +169,28 @@ def process_stream_frame(frame_number : int, temp_video_resolution : Resolution)
|
||||
return frame_number, numpy.ascontiguousarray(temp_vision_frame)
|
||||
|
||||
|
||||
#todo: needs review - [streaming] [critical: high] ordered submit and drain keeps writes in order, failed close stops the process manager
|
||||
def process_stream_frames() -> ErrorCode:
|
||||
trim_frame_start, trim_frame_end = restrict_trim_frame(state_manager.get_item('target_path'), state_manager.get_item('trim_frame_start'), state_manager.get_item('trim_frame_end'))
|
||||
output_video_resolution = scale_resolution(detect_video_resolution(state_manager.get_item('target_path')), state_manager.get_item('output_video_scale'))
|
||||
temp_video_resolution = restrict_video_resolution(state_manager.get_item('target_path'), output_video_resolution)
|
||||
temp_video_fps = restrict_video_fps(state_manager.get_item('target_path'), state_manager.get_item('output_video_fps'))
|
||||
frame_range = range(trim_frame_start, trim_frame_end)
|
||||
|
||||
if frame_range:
|
||||
video_metadata = ffprobe.extract_static_video_metadata(state_manager.get_item('target_path'))
|
||||
video_writer = video_manager.get_writer(state_manager.get_item('target_path'), video_metadata, temp_video_fps, temp_video_resolution, output_video_resolution, state_manager.get_item('output_video_fps'))
|
||||
frame_look_ahead = calculate_frame_look_ahead(temp_video_resolution)
|
||||
|
||||
with tqdm(total = len(frame_range), desc = translator.get('processing'), unit = 'frame', ascii = ' =', disable = state_manager.get_item('log_level') in [ 'warn', 'error' ]) as progress:
|
||||
progress.set_postfix(execution_providers = state_manager.get_item('execution_providers'))
|
||||
|
||||
read_static_video_frame(state_manager.get_item('target_path'), state_manager.get_item('reference_frame_number'))
|
||||
|
||||
with ThreadPoolExecutor(max_workers = state_manager.get_item('execution_thread_count')) as executor:
|
||||
futures = []
|
||||
|
||||
for frame_number in frame_range:
|
||||
futures.append(executor.submit(process_stream_frame, frame_number, temp_video_resolution))
|
||||
write_stream_frames(video_writer, futures, frame_look_ahead, progress)
|
||||
|
||||
write_stream_frames(video_writer, futures, 1, progress)
|
||||
|
||||
if not video_manager.close_video_writer(video_writer):
|
||||
process_manager.stop()
|
||||
|
||||
for processor_module in get_processors_modules(state_manager.get_item('processors')):
|
||||
processor_module.post_process()
|
||||
|
||||
if is_process_stopping():
|
||||
return 4
|
||||
else:
|
||||
logger.error(translator.get('temp_frames_not_found'), __name__)
|
||||
return 1
|
||||
return 0
|
||||
#todo: needs review - [memory] [critical: high] frame look ahead bound by a hardcoded 3gb budget
|
||||
def calculate_frame_look_ahead(temp_video_resolution : Resolution) -> int:
|
||||
width, height = temp_video_resolution
|
||||
frame_memory_budget = 3 * 1024 ** 3
|
||||
frame_memory_usage = width * height * 4 * 6
|
||||
return min(state_manager.get_item('execution_thread_count') * 2, max(2, frame_memory_budget // frame_memory_usage))
|
||||
|
||||
|
||||
#todo: needs review - [streaming] [critical: medium] drains futures down to the look ahead, stopping skips writes without cancelling
|
||||
def write_stream_frames(video_writer : VideoWriter, futures : List[Future[Tuple[int, VisionFrame]]], frame_look_ahead : int, progress : tqdm) -> None:
|
||||
for _ in range(len(futures) - frame_look_ahead + 1):
|
||||
if not is_process_stopping():
|
||||
_, temp_vision_frame = futures.pop(0).result()
|
||||
video_manager.write_video_writer_frame(video_writer, temp_vision_frame)
|
||||
progress.update()
|
||||
def write_stream_frame(video_writer : VideoWriter, futures : List[Future[Tuple[int, VisionFrame]]], progress : tqdm) -> None:
|
||||
if core.is_process_stopping():
|
||||
for pending_future in futures:
|
||||
pending_future.cancel()
|
||||
|
||||
future = futures.pop(0)
|
||||
|
||||
if not future.cancelled():
|
||||
_, temp_vision_frame = future.result()
|
||||
video_manager.write_video_writer_frame(video_writer, temp_vision_frame)
|
||||
progress.update()
|
||||
|
||||
|
||||
#todo: copy
|
||||
def finalize_video(start_time : float) -> ErrorCode:
|
||||
if is_video(state_manager.get_item('output_path')):
|
||||
logger.info(translator.get('processing_video_succeeded').format(seconds = calculate_end_time(start_time)), __name__)
|
||||
|
||||
Reference in New Issue
Block a user