tools/video_encode_tool.py

# 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.