File size: 6,362 Bytes
640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 640c04f 23d1ce3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
# tools/video_encode_tool.py
#
# Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos
#
# Version: 1.1.1
#
# This file defines the VideoEncodeTool specialist. Its purpose is to abstract away
# the underlying command-line tools (like FFmpeg) used for video manipulation tasks
# such as concatenation and creating transitions. By encapsulating this logic, the core
# Deformes4D engine can remain agnostic to the specific tool being used, allowing for easier
# maintenance and future replacement with other libraries or tools.
import os
import subprocess
import logging
import random
import time
import shutil
from typing import List, Optional, Tuple
logger = logging.getLogger(__name__)
class VideoToolError(Exception):
"""Custom exception for errors originating from the VideoEncodeTool."""
pass
class VideoEncodeTool:
"""
A specialist for handling video encoding and manipulation tasks.
Currently uses FFmpeg as the backend.
"""
def create_transition_bridge(self, start_image_path: str, end_image_path: str,
duration: float, fps: int, target_resolution: Tuple[int, int],
workspace_dir: str, effect: Optional[str] = None) -> str:
"""
Creates a short video clip that transitions between two static images using FFmpeg's xfade filter.
This is useful for creating a "bridge" during a hard "cut" decided by the cinematic director.
Args:
start_image_path (str): The file path to the starting image.
end_image_path (str): The file path to the ending image.
duration (float): The desired duration of the transition in seconds.
fps (int): The frames per second for the output video.
target_resolution (Tuple[int, int]): The (width, height) of the output video.
workspace_dir (str): The directory to save the output video.
effect (Optional[str], optional): The specific xfade effect to use. If None, a random
effect is chosen. Defaults to None.
Returns:
str: The file path to the generated transition video clip.
Raises:
VideoToolError: If the FFmpeg command fails.
"""
output_path = os.path.join(workspace_dir, f"bridge_{int(time.time())}.mp4")
width, height = target_resolution
fade_effects = [
"fade", "wipeleft", "wiperight", "wipeup", "wipedown", "dissolve",
"fadeblack", "fadewhite", "radial", "rectcrop", "circleopen",
"circleclose", "horzopen", "horzclose"
]
selected_effect = effect if effect and effect.strip() else random.choice(fade_effects)
transition_duration = max(0.1, duration)
cmd = (
f"ffmpeg -y -v error -loop 1 -t {transition_duration} -i \"{start_image_path}\" -loop 1 -t {transition_duration} -i \"{end_image_path}\" "
f"-filter_complex \"[0:v]scale={width}:{height},setsar=1[v0];[1:v]scale={width}:{height},setsar=1[v1];"
f"[v0][v1]xfade=transition={selected_effect}:duration={transition_duration}:offset=0[out]\" "
f"-map \"[out]\" -c:v libx264 -r {fps} -pix_fmt yuv420p \"{output_path}\""
)
logger.info(f"Creating FFmpeg transition bridge with effect: '{selected_effect}' | Duration: {transition_duration}s")
try:
subprocess.run(cmd, shell=True, check=True, text=True)
except subprocess.CalledProcessError as e:
logger.error(f"FFmpeg bridge creation failed. Return code: {e.returncode}")
logger.error(f"FFmpeg command: {cmd}")
logger.error(f"FFmpeg stderr: {e.stderr}")
raise VideoToolError(f"Failed to create transition video. Details: {e.stderr}")
return output_path
def concatenate_videos(self, video_paths: List[str], output_path: str, workspace_dir: str):
"""
Concatenates multiple video clips into a single file without re-encoding.
Args:
video_paths (List[str]): A list of absolute paths to the video clips to be concatenated.
output_path (str): The absolute path for the final output video.
workspace_dir (str): The directory to use for temporary files, like the concat list.
Raises:
VideoToolError: If no video paths are provided or if the FFmpeg command fails.
"""
if not video_paths:
raise VideoToolError("VideoEncodeTool: No video fragments provided for concatenation.")
if len(video_paths) == 1:
logger.info("Only one video clip found. Skipping concatenation and just copying the file.")
shutil.copy(video_paths[0], output_path)
return
list_file_path = os.path.join(workspace_dir, "concat_list.txt")
try:
with open(list_file_path, 'w', encoding='utf-8') as f:
for path in video_paths:
f.write(f"file '{os.path.abspath(path)}'\n")
cmd_list = ['ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', list_file_path, '-c', 'copy', output_path]
logger.info(f"Concatenating {len(video_paths)} video clips into {output_path} using FFmpeg...")
subprocess.run(cmd_list, check=True, capture_output=True, text=True)
logger.info(f"FFmpeg concatenation successful. Final video is at: {output_path}")
except subprocess.CalledProcessError as e:
logger.error(f"FFmpeg concatenation failed. Return code: {e.returncode}")
logger.error(f"FFmpeg stderr: {e.stderr}")
raise VideoToolError(f"Failed to assemble the final video using FFmpeg. Details: {e.stderr}")
except Exception as e:
logger.error(f"An unexpected error occurred during video concatenation: {e}", exc_info=True)
raise VideoToolError("An unexpected error occurred during the final video assembly.")
finally:
if os.path.exists(list_file_path):
os.remove(list_file_path)
# --- Singleton Instance ---
# We create a single instance of the tool to be imported by other modules.
video_encode_tool_singleton = VideoEncodeTool() |