# 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 gc import shutil from ltx_manager_helpers import ltx_manager_singleton from gemini_helpers import gemini_singleton # [REATORADO] Importa o novo especialista from latent_enhancer_specialist import latent_enhancer_specialist_singleton from hd_specialist import hd_specialist_singleton from ltx_video.models.autoencoders.vae_encode import vae_encode, vae_decode from audio_specialist import audio_specialist_singleton logger = logging.getLogger(__name__) @dataclass class LatentConditioningItem: """Representa uma âncora de condicionamento no espaço latente para a Câmera (Ψ).""" latent_tensor: torch.Tensor media_frame_number: int conditioning_strength: float class Deformes4DEngine: """ Implementa a Câmera (Ψ) e o Destilador (Δ) da arquitetura ADUC-SDR. Orquestra a geração, pós-produção latente e renderização final dos fragmentos de vídeo. """ 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 (Executor ADUC-SDR) inicializado.") # Cria o diretório de workspace se não existir os.makedirs(self.workspace_dir, exist_ok=True) @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 --- @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: 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, output_params=['-pix_fmt', 'yuv420p']) as writer: for frame in video_np: writer.append_data(frame) def _preprocess_image_for_latent_conversion(self, image: Image.Image, target_resolution: tuple) -> Image.Image: if image.size != target_resolution: 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 concatenate_videos_ffmpeg(self, video_paths: list[str], output_path: 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") # Tenta usar aceleração de hardware (GPU) para a concatenação, se disponível cmd_list = ['ffmpeg', '-y', '-hwaccel', 'auto', '-f', 'concat', '-safe', '0', '-i', list_file_path, '-c', 'copy', output_path] logger.info(f"Concatenando {len(video_paths)} clipes de vídeo em {output_path}...") 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}") # Tenta novamente sem aceleração de hardware como fallback logger.info("Tentando concatenar novamente sem aceleração de hardware...") cmd_list = ['ffmpeg', '-y', '-f', 'concat', '-safe', '0', '-i', list_file_path, '-c', 'copy', output_path] try: subprocess.run(cmd_list, check=True, capture_output=True, text=True) except subprocess.CalledProcessError as e_fallback: logger.error(f"Erro no FFmpeg (fallback): {e_fallback.stderr}") raise gr.Error(f"Falha na montagem final do vídeo. Detalhes: {e_fallback.stderr}") # --- NÚCLEO DA LÓGICA ADUC-SDR --- 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, use_upscaler: bool, use_refiner: bool, use_hd: bool, use_audio: bool, video_resolution: int, use_continuity_director: bool, progress: gr.Progress = gr.Progress()): # --- ETAPA 0: SETUP --- FPS = 24 FRAMES_PER_LATENT_CHUNK = 8 ECO_LATENT_CHUNKS = 2 LATENT_PROCESSING_CHUNK_SIZE = 10 # Processa 10 fragmentos latentes por vez para economizar memória run_timestamp = int(time.time()) temp_latent_dir = os.path.join(self.workspace_dir, f"temp_latents_{run_timestamp}") temp_video_clips_dir = os.path.join(self.workspace_dir, f"temp_clips_{run_timestamp}") os.makedirs(temp_latent_dir, exist_ok=True) os.makedirs(temp_video_clips_dir, exist_ok=True) total_frames_brutos = self._quantize_to_multiple(int(seconds_per_fragment * FPS), FRAMES_PER_LATENT_CHUNK) frames_a_podar = self._quantize_to_multiple(int(total_frames_brutos * (trim_percent / 100)), FRAMES_PER_LATENT_CHUNK) latents_a_podar = frames_a_podar // FRAMES_PER_LATENT_CHUNK DEJAVU_FRAME_TARGET = frames_a_podar - 1 if frames_a_podar > 0 else 0 DESTINATION_FRAME_TARGET = total_frames_brutos - 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} refine_ltx_params = {"motion_prompt": "", "guidance_scale": 1.0, "denoise_strength": 0.35, "refine_steps": 12} keyframe_paths = [item[0] if isinstance(item, tuple) else item for item in keyframes] story_history = "" target_resolution_tuple = (video_resolution, video_resolution) eco_latent_for_next_loop, dejavu_latent_for_next_loop = None, None latent_fragment_paths = [] # Lista para armazenar caminhos dos latentes salvos no disco if len(keyframe_paths) < 2: raise gr.Error(f"A geração requer no mínimo 2 keyframes. Você forneceu {len(keyframe_paths)}.") num_transitions_to_generate = len(keyframe_paths) - 1 # --- ETAPA 1: GERAR FRAGMENTOS LATENTES E SALVAR EM DISCO --- logger.info("--- INICIANDO ETAPA 1: Geração de Fragmentos Latentes ---") for i in range(num_transitions_to_generate): fragment_index = i + 1 progress(i / num_transitions_to_generate, desc=f"Gerando Latente {fragment_index}/{num_transitions_to_generate}") # (Lógica de decisão do Gemini e preparação de âncoras - inalterada) past_keyframe_path = keyframe_paths[i - 1] if i > 0 else keyframe_paths[i] start_keyframe_path = keyframe_paths[i] destination_keyframe_path = keyframe_paths[i + 1] future_story_prompt = storyboard[i + 1] if (i + 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[i - 1] if i > 0 else "O início.", storyboard[i], future_story_prompt) transition_type, motion_prompt = decision["transition_type"], decision["motion_prompt"] story_history += f"\n- Ato {fragment_index}: {motion_prompt}" conditioning_items = [] if eco_latent_for_next_loop is None: 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(dejavu_latent_for_next_loop, DEJAVU_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} latents_brutos = self._generate_latent_tensor_internal(conditioning_items, current_ltx_params, target_resolution_tuple, total_frames_brutos) last_trim = latents_brutos[:, :, -(latents_a_podar+1):, :, :].clone() eco_latent_for_next_loop = last_trim[:, :, :2, :, :].clone() dejavu_latent_for_next_loop = last_trim[:, :, -1:, :, :].clone() latents_video = latents_brutos[:, :, :-(latents_a_podar-1), :, :].clone() latents_video = latents_video[:, :, 1:, :, :] del last_trim, latents_brutos gc.collect(); torch.cuda.empty_cache() if transition_type == "cut": eco_latent_for_next_loop, dejavu_latent_for_next_loop = None, None # [REATORADO] Mover latente para CPU e salvar no disco para liberar VRAM cpu_latent = latents_video.cpu() latent_path = os.path.join(temp_latent_dir, f"latent_fragment_{i:04d}.pt") torch.save(cpu_latent, latent_path) latent_fragment_paths.append(latent_path) del latents_video, cpu_latent gc.collect() del eco_latent_for_next_loop, dejavu_latent_for_next_loop gc.collect(); torch.cuda.empty_cache() # --- ETAPA 2: PROCESSAR LATENTES EM LOTES (CHUNKS) --- logger.info(f"--- INICIANDO ETAPA 2: Processamento de {len(latent_fragment_paths)} latentes em lotes de {LATENT_PROCESSING_CHUNK_SIZE} ---") final_video_clip_paths = [] num_chunks = -(-len(latent_fragment_paths) // LATENT_PROCESSING_CHUNK_SIZE) # Ceiling division for i in range(num_chunks): chunk_start_index = i * LATENT_PROCESSING_CHUNK_SIZE chunk_end_index = chunk_start_index + LATENT_PROCESSING_CHUNK_SIZE chunk_paths = latent_fragment_paths[chunk_start_index:chunk_end_index] progress(i / num_chunks, desc=f"Processando Lote {i+1}/{num_chunks}") # Carrega os tensores do lote atual do disco para a GPU tensors_in_chunk = [torch.load(p, map_location=self.device) for p in chunk_paths] # Concatena os tensores do lote, removendo o latente de sobreposição tensors_para_concatenar = [ frag[:, :, :-1, :, :] if j < len(tensors_in_chunk) - 1 else frag for j, frag in enumerate(tensors_in_chunk) ] sub_group_latent = torch.cat(tensors_para_concatenar, dim=2) del tensors_in_chunk, tensors_para_concatenar gc.collect(); torch.cuda.empty_cache() logger.info(f"Lote {i+1} concatenado. Shape do sub-latente: {sub_group_latent.shape}") # 1. (Opcional) Upscaler Latente if use_upscaler: logger.info(f"Aplicando Upscaler no lote {i+1}...") sub_group_latent = latent_enhancer_specialist_singleton.upscale(sub_group_latent) gc.collect(); torch.cuda.empty_cache() # 2. Decodificar Latente para Vídeo (com ou sem áudio) base_name = f"clip_{i:04d}_{run_timestamp}" current_clip_path = os.path.join(temp_video_clips_dir, f"{base_name}_temp.mp4") if use_audio: # O áudio é gerado para o prompt global por enquanto. Pode ser adaptado. current_clip_path = self._generate_video_and_audio_from_latents(sub_group_latent, global_prompt, base_name) else: pixel_tensor = self.latents_to_pixels(sub_group_latent) self.save_video_from_tensor(pixel_tensor, current_clip_path, fps=FPS) del pixel_tensor del sub_group_latent gc.collect(); torch.cuda.empty_cache() # 3. (Opcional) Masterização HD if use_hd: logger.info(f"Aplicando masterização HD no clipe {i+1}...") hd_clip_path = os.path.join(temp_video_clips_dir, f"{base_name}_hd.mp4") try: hd_specialist_singleton.process_video(input_video_path=current_clip_path, output_video_path=hd_clip_path, prompt=global_prompt) # Apaga o clipe não-HD para economizar espaço if os.path.exists(current_clip_path) and current_clip_path != hd_clip_path: os.remove(current_clip_path) current_clip_path = hd_clip_path except Exception as e: logger.error(f"Falha na masterização HD do clipe {i+1}: {e}. Usando versão padrão.") # 4. Adicionar caminho do clipe final à lista final_video_clip_paths.append(current_clip_path) #if use_refiner: # progress(0.8, desc="Refinando continuidade visual...") # # [REATORADO] Chamada para o novo especialista # # OBS: Refinamento foi desativado conforme solicitado por degradar a lógica das keyframes. # --- ETAPA 3: MONTAGEM FINAL --- progress(0.98, desc="Montagem final dos clipes...") final_video_path = os.path.join(self.workspace_dir, f"filme_final_{run_timestamp}.mp4") self.concatenate_videos_ffmpeg(final_video_clip_paths, final_video_path) # --- ETAPA 4: LIMPEZA --- logger.info("Limpando arquivos temporários...") try: shutil.rmtree(temp_latent_dir) shutil.rmtree(temp_video_clips_dir) concat_list_path = os.path.join(self.workspace_dir, "concat_list.txt") if os.path.exists(concat_list_path): os.remove(concat_list_path) except OSError as e: logger.warning(f"Não foi possível remover os diretórios temporários: {e}") logger.info(f"Processo concluído! Vídeo final salvo em: {final_video_path}") yield {"final_path": final_video_path} def _generate_video_and_audio_from_latents(self, latent_tensor, audio_prompt, base_name): # Este método agora opera em um diretório temporário para os clipes temp_video_clips_dir = os.path.dirname(os.path.join(self.workspace_dir, base_name)) # Hack para obter o diretório correto silent_video_path = os.path.join(temp_video_clips_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(); torch.cuda.empty_cache() try: result = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", silent_video_path], capture_output=True, text=True, check=True) frag_duration = float(result.stdout.strip()) except (subprocess.CalledProcessError, ValueError, FileNotFoundError): logger.warning(f"ffprobe falhou. Calculando duração manualmente a partir dos latentes.") # O VAE interpola, então o número de frames é (num_latentes - 1) * 8 + 1 (aproximadamente) num_pixel_frames = (latent_tensor.shape[2] - 1) * 8 + 1 frag_duration = num_pixel_frames / 24.0 # Salva o vídeo com áudio no mesmo diretório temporário video_with_audio_path = audio_specialist_singleton.generate_audio_for_video( video_path=silent_video_path, prompt=audio_prompt, duration_seconds=frag_duration, output_path_override=os.path.join(temp_video_clips_dir, f"{base_name}_with_audio.mp4") ) if os.path.exists(silent_video_path): os.remove(silent_video_path) return video_with_audio_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) gc.collect() torch.cuda.empty_cache() return new_full_latents 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