# deformes4D_engine.py # Copyright (C) 4 de Agosto de 2025 Carlos Rodrigues dos Santos # # MODIFICATIONS FOR ADUC-SDR: # Copyright (C) 2025 Carlos Rodrigues dos Santos. All rights reserved. # # This file is part of the ADUC-SDR project. It contains the core logic for # video fragment generation, latent manipulation, and dynamic editing, # governed by the ADUC orchestrator. # This component is licensed under the GNU Affero General Public License v3.0. import os import time import imageio import numpy as np import torch import logging from PIL import Image, ImageOps from dataclasses import dataclass import gradio as gr import subprocess import random import gc from ltx_manager_helpers import ltx_manager_singleton from gemini_helpers import gemini_singleton from ltx_video.models.autoencoders.vae_encode import vae_encode, vae_decode logger = logging.getLogger(__name__) @dataclass class LatentConditioningItem: latent_tensor: torch.Tensor media_frame_number: int conditioning_strength: float class Deformes4DEngine: def __init__(self, ltx_manager, workspace_dir="deformes_workspace"): self.ltx_manager = ltx_manager self.workspace_dir = workspace_dir self._vae = None self.device = 'cuda' if torch.cuda.is_available() else 'cpu' logger.info("Especialista Deformes4D (SDR Executor) inicializado.") @property def vae(self): if self._vae is None: self._vae = self.ltx_manager.workers[0].pipeline.vae self._vae.to(self.device); self._vae.eval() return self._vae # ... (métodos auxiliares como save/load/pixels_to_latents permanecem os mesmos) ... def save_latent_tensor(self, tensor: torch.Tensor, path: str): torch.save(tensor.cpu(), path) logger.info(f"Tensor latente salvo em: {path}") def load_latent_tensor(self, path: str) -> torch.Tensor: tensor = torch.load(path, map_location=self.device) logger.info(f"Tensor latente carregado de: {path} para o dispositivo {self.device}") return tensor @torch.no_grad() def pixels_to_latents(self, tensor: torch.Tensor) -> torch.Tensor: tensor = tensor.to(self.device, dtype=self.vae.dtype) return vae_encode(tensor, self.vae, vae_per_channel_normalize=True) @torch.no_grad() def latents_to_pixels(self, latent_tensor: torch.Tensor, decode_timestep: float = 0.05) -> torch.Tensor: latent_tensor = latent_tensor.to(self.device, dtype=self.vae.dtype) timestep_tensor = torch.tensor([decode_timestep] * latent_tensor.shape[0], device=self.device, dtype=latent_tensor.dtype) return vae_decode(latent_tensor, self.vae, is_video=True, timestep=timestep_tensor, vae_per_channel_normalize=True) def save_video_from_tensor(self, video_tensor: torch.Tensor, path: str, fps: int = 24): if video_tensor is None or video_tensor.ndim != 5 or video_tensor.shape[2] == 0: logger.warning("Tentativa de salvar um tensor de vídeo inválido. Abortando.") return video_tensor = video_tensor.squeeze(0).permute(1, 2, 3, 0) video_tensor = (video_tensor.clamp(-1, 1) + 1) / 2.0 video_np = (video_tensor.detach().cpu().float().numpy() * 255).astype(np.uint8) with imageio.get_writer(path, fps=fps, codec='libx264', quality=8) as writer: for frame in video_np: writer.append_data(frame) logger.info(f"Vídeo salvo em: {path}") def _preprocess_image_for_latent_conversion(self, image: Image.Image, target_resolution: tuple) -> Image.Image: if image.size != target_resolution: logger.info(f" - AÇÃO: Redimensionando imagem de {image.size} para {target_resolution} antes da conversão para latente.") return ImageOps.fit(image, target_resolution, Image.Resampling.LANCZOS) return image def pil_to_latent(self, pil_image: Image.Image) -> torch.Tensor: image_np = np.array(pil_image).astype(np.float32) / 255.0 tensor = torch.from_numpy(image_np).permute(2, 0, 1).unsqueeze(0).unsqueeze(2) tensor = (tensor * 2.0) - 1.0 return self.pixels_to_latents(tensor) def _generate_video_from_latents(self, latent_tensor, base_name): silent_video_path = os.path.join(self.workspace_dir, f"{base_name}_silent.mp4") pixel_tensor = self.latents_to_pixels(latent_tensor) self.save_video_from_tensor(pixel_tensor, silent_video_path, fps=24) del pixel_tensor; gc.collect() return silent_video_path def _get_video_frame_count(self, video_path: str) -> int | None: """ Obtém o número total de frames de um arquivo de vídeo usando ffprobe. Args: video_path (str): O caminho para o arquivo de vídeo. Returns: int | None: O número de frames, ou None se ocorrer um erro. """ if not os.path.exists(video_path): logger.error(f"Arquivo de vídeo não encontrado para contagem de frames: {video_path}") return None cmd = [ 'ffprobe', '-v', 'error', '-select_streams', 'v:0', '-count_frames', '-show_entries', 'stream=nb_read_frames', '-of', 'default=nokey=1:noprint_wrappers=1', video_path ] logger.debug(f"Executando ffprobe para contar frames: {' '.join(cmd)}") try: result = subprocess.run(cmd, check=True, capture_output=True, text=True, encoding='utf-8') frame_count = int(result.stdout.strip()) return frame_count except FileNotFoundError: logger.error("Comando 'ffprobe' não encontrado. Verifique se o FFmpeg está instalado e no PATH do sistema.") return None except subprocess.CalledProcessError as e: logger.error(f"Erro ao executar ffprobe para contar frames: {e.stderr}") return None except ValueError: logger.error(f"Não foi possível converter a saída do ffprobe em um número. Saída: '{result.stdout}'") return None def _generate_video_from_latents(self, latent_tensor, base_name, nPoda_inicio: int = 0, nPoda_fim: int = 0): """ Gera um vídeo a partir de um tensor latente, com poda opcional de frames no início e no fim. A contagem de frames para a poda é obtida diretamente do arquivo de vídeo gerado. """ # 1. Gera o vídeo inicial (completo) a partir dos latentes. O tensor é usado aqui e depois descartado. untrimmed_video_path = os.path.join(self.workspace_dir, f"{base_name}_untrimmed.mp4") pixel_tensor = self.latents_to_pixels(latent_tensor) self.save_video_from_tensor(pixel_tensor, untrimmed_video_path, fps=24) del pixel_tensor; gc.collect() # Se não houver poda, apenas renomeia e retorna if nPoda_inicio == 0 and nPoda_fim == 0: logger.info("Nenhuma poda necessária. Retornando vídeo gerado diretamente.") final_video_path = os.path.join(self.workspace_dir, f"{base_name}_final.mp4") os.rename(untrimmed_video_path, final_video_path) return final_video_path # 2. OBTÉM A CONTAGEM DE FRAMES DO ARQUIVO DE VÍDEO USANDO FFPROBE logger.info(f"Obtendo contagem de frames do arquivo: {untrimmed_video_path}") total_frames = self._get_video_frame_count(untrimmed_video_path) # Validação: Verifica se a contagem de frames funcionou if total_frames is None: logger.error("Não foi possível obter a contagem de frames do vídeo. " "Retornando o vídeo original sem poda.") return untrimmed_video_path # 3. Prepara o comando FFmpeg para a poda com base nos dados do arquivo if nPoda_inicio + nPoda_fim >= total_frames: logger.error(f"Erro de poda: A soma dos frames a podar ({nPoda_inicio + nPoda_fim}) " f"é maior ou igual ao total de frames ({total_frames}). " "Retornando o vídeo original sem poda.") return untrimmed_video_path final_video_path = os.path.join(self.workspace_dir, f"{base_name}_trimmed.mp4") start_frame_to_keep = nPoda_inicio end_frame_to_keep = total_frames - nPoda_fim vf_filter = f"select='gte(n,{start_frame_to_keep})*lt(n,{end_frame_to_keep})',setpts=PTS-STARTPTS" cmd_list = [ 'ffmpeg', '-y', '-i', untrimmed_video_path, '-vf', vf_filter, '-an', final_video_path ] logger.info(f"Executando poda com FFmpeg. Total de frames (detectado): {total_frames}. " f"Removendo {nPoda_inicio} do início e {nPoda_fim} do fim.") logger.debug(f"Comando FFmpeg: {' '.join(cmd_list)}") # 4. Executa o comando e lida com possíveis erros try: subprocess.run(cmd_list, check=True, capture_output=True, text=True, encoding='utf-8') logger.info(f"Poda concluída. Vídeo final salvo em: {final_video_path}") os.remove(untrimmed_video_path) return final_video_path except subprocess.CalledProcessError as e: logger.error(f"Erro no FFmpeg durante a poda: {e.stderr}") return untrimmed_video_path def _generate_latent_tensor_internal(self, conditioning_items, ltx_params, target_resolution, total_frames_to_generate): final_ltx_params = {**ltx_params, 'width': target_resolution[0], 'height': target_resolution[1], 'video_total_frames': total_frames_to_generate, 'video_fps': 24, 'current_fragment_index': int(time.time()), 'conditioning_items_data': conditioning_items} new_full_latents, _ = self.ltx_manager.generate_latent_fragment(**final_ltx_params) return new_full_latents def concatenate_videos_ffmpeg(self, video_paths: list[str], output_path: str) -> str: if not video_paths: raise gr.Error("Nenhum fragmento de vídeo para montar.") list_file_path = os.path.join(self.workspace_dir, "concat_list.txt") 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("Executando concatenação FFmpeg...") try: subprocess.run(cmd_list, check=True, capture_output=True, text=True) except subprocess.CalledProcessError as e: logger.error(f"Erro no FFmpeg: {e.stderr}") raise gr.Error(f"Falha na montagem final do vídeo. Detalhes: {e.stderr}") return output_path def generate_full_movie(self, keyframes: list, global_prompt: str, storyboard: list, seconds_per_fragment: float, trim_percent: int, handler_strength: float, destination_convergence_strength: float, video_resolution: int, use_continuity_director: bool, progress: gr.Progress = gr.Progress()): # 1. Calcula o número de chunks normalmente num_chunks = 12 # 2. Calcular o número de chunks a podar com base na porcentagem, com mínimo de 4 trim_chunks = 8 HANDLER_CHUNK_INDICES = -1 ECO_CHUNK_INDICES = 1 HANDLER_FRAME_TARGET = 4 * 8 FRAMES_TO_GENERATE = (num_chunks * 8) DESTINATION_FRAME_TARGET = FRAMES_TO_GENERATE-1 base_ltx_params = {"guidance_scale": 2.0, "stg_scale": 0.025, "rescaling_scale": 0.15, "num_inference_steps": 20, "image_cond_noise_scale": 0.00} keyframe_paths = [item[0] if isinstance(item, tuple) else item for item in keyframes] video_clips_paths, story_history = [], "" target_resolution_tuple = (video_resolution, video_resolution) eco_latent_for_next_loop = None handler_latent_for_next_loop = None if len(keyframe_paths) < 3: raise gr.Error(f"O modelo de geração requer no mínimo 3 keyframes (Passado, Presente, Futuro). Você forneceu {len(keyframe_paths)}.") num_transitions_to_generate = len(keyframe_paths) - 1 for i in range(num_transitions_to_generate): start_keyframe_index = i logger.info(f"--- INICIANDO FRAGMENTO {i+1}/{num_transitions_to_generate} ---") progress((i + 1) / num_transitions_to_generate, desc=f"Produzindo Transição {i+1}/{num_transitions_to_generate}") if i > 0: past_keyframe_path = keyframe_paths[start_keyframe_index - 1] else: past_keyframe_path = keyframe_paths[start_keyframe_index] start_keyframe_path = keyframe_paths[start_keyframe_index] destination_keyframe_path = keyframe_paths[start_keyframe_index + 1] future_story_prompt = storyboard[start_keyframe_index + 1] if (start_keyframe_index + 1) < len(storyboard) else "A cena final." decision = gemini_singleton.get_cinematic_decision( global_prompt, story_history, past_keyframe_path, start_keyframe_path, destination_keyframe_path, storyboard[start_keyframe_index - 1], storyboard[start_keyframe_index], future_story_prompt ) transition_type, motion_prompt = decision["transition_type"], decision["motion_prompt"] story_history += f"\n- Ato {i+1}: {motion_prompt}" conditioning_items = [] logger.info(" [0. PREPARAÇÃO] Montando itens de condicionamento...") is_first_fragment = eco_latent_for_next_loop is None if is_first_fragment: img_start = self._preprocess_image_for_latent_conversion(Image.open(start_keyframe_path).convert("RGB"), target_resolution_tuple) conditioning_items.append(LatentConditioningItem(self.pil_to_latent(img_start), 0, 1.0)) else: conditioning_items.append(LatentConditioningItem(eco_latent_for_next_loop, 0, 1.0)) conditioning_items.append(LatentConditioningItem(handler_latent_for_next_loop, HANDLER_FRAME_TARGET, handler_strength)) img_dest = self._preprocess_image_for_latent_conversion(Image.open(destination_keyframe_path).convert("RGB"), target_resolution_tuple) conditioning_items.append(LatentConditioningItem(self.pil_to_latent(img_dest), DESTINATION_FRAME_TARGET, destination_convergence_strength)) current_ltx_params = {**base_ltx_params, "motion_prompt": motion_prompt} new_full_latents = self._generate_latent_tensor_internal(conditioning_items, current_ltx_params, target_resolution_tuple, FRAMES_TO_GENERATE) logger.info(f" [1. GERAÇÃO] Tensor latente bruto gerado com shape: {new_full_latents.shape}.") trim_chunks = 3 # Pega os últimos 4 elementos eco_latent_for_next_loop = new_full_latents[:, :, -trim_chunks:, :, :].clone() # Desses 4, pega os 2 primeiros (índices 0 e 1) eco_latent_for_next_loop = eco_latent_for_next_loop[:, :, :2, :, :] handler_latent_for_next_loop = new_full_latents[:, :, -1:, :, :].clone() latents_for_video = new_full_latents.clone() #if transition_type == "cut": # eco_latent_for_next_loop = None # handler_latent_for_next_loop = None base_name = f"fragment_{i}_{int(time.time())}" frames_poda_inicio = 8 # Ex: Você também quer podar os 2 últimos chunks da cauda gerada. frames_poda_fim = 8*4 # A chamada à função agora inclui os novos parâmetros video_path = self._generate_video_from_latents( latents_for_video, base_name, nPoda_inicio=frames_poda_inicio, nPoda_fim=frames_poda_fim ) video_path = self._generate_video_from_latents(latents_for_video, base_name) video_clips_paths.append(video_path) yield {"fragment_path": video_path} final_movie_path = os.path.join(self.workspace_dir, f"final_movie_silent_{int(time.time())}.mp4") self.concatenate_videos_ffmpeg(video_clips_paths, final_movie_path) logger.info(f"Filme completo salvo em: {final_movie_path}") yield {"final_path": final_movie_path} def _quantize_to_multiple(self, n, m): if m == 0: return n quantized = int(round(n / m) * m) return m if n > 0 and quantized == 0 else quantized