Create video_encode_tool.py

tools/video_encode_tool.py

@@ -0,0 +1,143 @@
+# tools/video_encode_tool.py
+#
+# Copyright (C) August 4, 2025  Carlos Rodrigues dos Santos
+#
+# Version: 1.1.0
+#
+# 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.
+
+import os
+import subprocess
+import logging
+import gradio as gr
+from typing import List, Optional, Tuple
+import random
+import time
+import shutil
+
+logger = logging.getLogger(__name__)
+
+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.
+        """
+        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)
+        
+        # The duration of each image loop and the xfade itself should match the total desired duration.
+        transition_duration = max(0.1, duration) # Ensure duration is not zero
+
+        # Construct the FFmpeg command
+        # -v error: Suppress all console output except for errors.
+        # -loop 1: Loop the input image.
+        # -t {duration}: Set the duration for the looped image.
+        # -filter_complex: Defines a complex filtergraph.
+        #   - scale,setsar: Pre-process each image to the target resolution and aspect ratio.
+        #   - xfade: Apply the crossfade transition.
+        #     - offset=0: Start the transition immediately.
+        # -map "[out]": Map the output of the filtergraph to the final video.
+        # -c:v libx264 -r {fps} -pix_fmt yuv420p: Standard high-compatibility video encoding settings.
+        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 gr.Error(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.
+        """
+        if not video_paths:
+            raise gr.Error("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.")
+            # If there's only one clip, a simple copy is much faster.
+            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 gr.Error(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 gr.Error("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()```
+
+O especialista `VideoEncodeTool` está agora mais poderoso. O `Deformes4D_engine` poderá chamá-lo tanto para montar os clipes gerados pelo LTX quanto para criar transições rápidas e eficientes para os "cortes", deixando a lógica de geração de latentes focada nos segmentos que precisam de movimento complexo.
+
+Essa abstração está deixando nosso `Deformes4D_engine` cada vez mais limpo e focado em sua tarefa de orquestração de alto nível.