Update tools/video_encode_tool.py

tools/video_encode_tool.py

@@ -38,4 +38,167 @@ class _SimpleVideoEncoder:
                     print(f"[DEBUG] [Encoder] frame {i}/{T} gravado ({H}x{W}@{fps}fps)")
 
 # Singleton global de uso simples
-video_encode_tool_singleton = _SimpleVideoEncoder()
+video_encode_tool_singleton = _SimpleVideoEncoder()
+
+
+
+# aduc_framework/tools/video_encode_tool.py
+#
+# Versão 1.4.0 (Conjunto de Ferramentas de Vídeo Finalizado)
+# Copyright (C) August 4, 2025  Carlos Rodrigues dos Santos
+#
+# Este módulo atua como o especialista central para todas as operações de
+# manipulação e codificação de vídeo. Ele abstrai as interações com
+# FFmpeg e imageio, fornecendo uma API limpa e robusta para o resto do framework.
+# - save_video_from_tensor: Converte um tensor de pixel em um arquivo .mp4.
+# - extract_..._frame: Extrai frames específicos de clipes de vídeo.
+# - concatenate_videos: Monta o filme final a partir dos clipes de cena.
+
+import os
+import subprocess
+import logging
+import random
+import time
+import shutil
+from typing import List, Optional, Tuple
+
+import imageio
+import numpy as np
+import torch
+
+logger = logging.getLogger(__name__)
+
+class VideoToolError(Exception):
+    """Exceção personalizada para erros originados do VideoEncodeTool."""
+    pass
+
+class VideoEncodeTool:
+    """
+    Um especialista para lidar com tarefas de codificação e manipulação de vídeo.
+    """
+
+    def save_video_from_tensor(self, video_tensor: torch.Tensor, path: str, fps: int = 24, progress_callback=None):
+        """
+        Salva um tensor de pixel como um arquivo de vídeo .mp4 usando parâmetros otimizados.
+        Espera um tensor no formato (B, C, F, H, W) onde B=1.
+        """
+        # Verificações de robustez para garantir que o tensor é válido
+        if video_tensor is None or video_tensor.ndim != 5 or video_tensor.shape[0] != 1 or video_tensor.shape[2] == 0:
+            logger.warning(f"Tensor de vídeo inválido ou vazio recebido. Shape: {video_tensor.shape if video_tensor is not None else 'None'}. Pulando salvamento de vídeo para '{path}'.")
+            return
+
+        logger.info(f"Salvando tensor de vídeo com shape {video_tensor.shape} para '{os.path.basename(path)}'...")
+        
+        try:
+            # Squeeze: (1, C, F, H, W) -> (C, F, H, W)
+            # Permute: (C, F, H, W) -> (F, H, W, C) - formato esperado por imageio
+            video_tensor_permuted = video_tensor.squeeze(0).permute(1, 2, 3, 0)
+            
+            # Desnormaliza de [-1, 1] para [0, 1]
+            video_tensor_normalized = (video_tensor_permuted.clamp(-1, 1) + 1) / 2.0
+            
+            # Converte para [0, 255], move para CPU e converte para numpy uint8
+            video_np = (video_tensor_normalized.detach().cpu().float().numpy() * 255).astype(np.uint8)
+            
+            # Salva o vídeo com parâmetros de alta compatibilidade
+            with imageio.get_writer(
+                path, 
+                fps=fps, 
+                codec='libx264', 
+                quality=8, # Qualidade boa (0-10, onde 10 é a melhor)
+                output_params=['-pix_fmt', 'yuv420p'] # Formato de pixel para compatibilidade máxima
+            ) as writer:
+                for frame in video_np:
+                    writer.append_data(frame)
+                    if progress_callback:
+                       progress_callback(i + 1, T)
+                          if i % self.frame_log_every == 0:
+                       print(f"[DEBUG] [Encoder] frame {i}/{T} gravado ({H}x{W}@{fps}fps)")
+                    
+            
+            logger.info(f"Vídeo salvo com sucesso em: {path}")
+        except Exception as e:
+            logger.error(f"Falha ao salvar vídeo com imageio para '{path}': {e}", exc_info=True)
+            raise VideoToolError(f"Não foi possível escrever o arquivo de vídeo: {e}")
+
+    def extract_first_frame(self, video_path: str, output_image_path: str) -> str:
+        """
+        Extrai o primeiro frame de um arquivo de vídeo e o salva como uma imagem.
+        """
+        logger.info(f"Extraindo primeiro frame de '{os.path.basename(video_path)}'...")
+        cmd = ['ffmpeg', '-y', '-v', 'error', '-i', video_path, '-vframes', '1', '-q:v', '2', output_image_path]
+        try:
+            subprocess.run(cmd, check=True, capture_output=True, text=True)
+            return output_image_path
+        except subprocess.CalledProcessError as e:
+            logger.error(f"FFmpeg (extract_first_frame) falhou: {e.stderr}")
+            raise VideoToolError(f"Falha ao extrair o primeiro frame de {video_path}")
+
+    def extract_last_frame(self, video_path: str, output_image_path: str) -> str:
+        """
+        Extrai o último frame de um arquivo de vídeo e o salva como uma imagem.
+        """
+        logger.info(f"Extraindo último frame de '{os.path.basename(video_path)}'...")
+        cmd = ['ffmpeg', '-y', '-v', 'error', '-sseof', '-0.1', '-i', video_path, '-vframes', '1', '-q:v', '2', output_image_path]
+        try:
+            subprocess.run(cmd, check=True, capture_output=True, text=True)
+            return output_image_path
+        except subprocess.CalledProcessError as e:
+            logger.error(f"FFmpeg (extract_last_frame) falhou: {e.stderr}")
+            raise VideoToolError(f"Falha ao extrair o último frame de {video_path}")
+
+    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:
+        """
+        Cria um clipe de vídeo curto que transiciona entre duas imagens estáticas.
+        """
+        output_path = os.path.join(workspace_dir, f"bridge_{int(time.time())}_{random.randint(100, 999)}.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"Criando ponte de transição com efeito '{selected_effect}'...")
+        try:
+            subprocess.run(cmd, shell=True, check=True, text=True)
+        except subprocess.CalledProcessError as e:
+            raise VideoToolError(f"Falha ao criar vídeo de transição: {e.stderr}")
+        return output_path
+
+    def concatenate_videos(self, video_paths: List[str], output_path: str, workspace_dir: str) -> str:
+        """
+        Concatena múltiplos clipes de vídeo em um único arquivo sem re-codificar.
+        """
+        if not video_paths:
+            raise VideoToolError("Nenhum fragmento de vídeo fornecido para concatenação.")
+        # Se houver apenas um clipe, apenas o copie para o destino final.
+        if len(video_paths) == 1:
+            shutil.copy(video_paths[0], output_path)
+            logger.info(f"Apenas um clipe fornecido. Copiado para '{output_path}'.")
+            return output_path
+            
+        list_file_path = os.path.join(workspace_dir, f"concat_list_{int(time.time())}.txt")
+        try:
+            with open(list_file_path, 'w', encoding='utf-8') as f:
+                for path in video_paths:
+                    # Garante que o caminho seja absoluto para o ffmpeg encontrar
+                    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"Concatenando {len(video_paths)} clipes para '{os.path.basename(output_path)}'...")
+            subprocess.run(cmd_list, check=True, capture_output=True, text=True)
+            logger.info("Concatenação FFmpeg bem-sucedida.")
+            return output_path
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Falha ao montar o vídeo final com FFmpeg: {e.stderr}")
+            raise VideoToolError(f"Falha ao montar o vídeo final com FFmpeg.")
+        finally:
+            if os.path.exists(list_file_path):
+                os.remove(list_file_path)
+
+# --- Instância Singleton ---
+video_encode_tool_singleton = VideoEncodeTool()