# 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 from audio_specialist import audio_specialist_singleton from ltx_manager_helpers import ltx_manager_singleton from gemini_helpers import gemini_singleton from upscaler_specialist import upscaler_specialist_singleton from ltx_video.models.autoencoders.causal_video_autoencoder import CausalVideoAutoencoder from ltx_video.models.autoencoders.vae_encode import vae_encode, vae_decode 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: Câmera Ψ e Destilador Δ) 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 def save_latent_tensor(self, tensor: torch.Tensor, path: str): torch.save(tensor.cpu(), path) def load_latent_tensor(self, path: str) -> torch.Tensor: return torch.load(path, map_location=self.device) @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) 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 _get_video_frame_count(self, video_path: str) -> int | None: if not os.path.exists(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] try: result = subprocess.run(cmd, check=True, capture_output=True, text=True, encoding='utf-8') return int(result.stdout.strip()) except Exception: return None def _trim_last_frame_ffmpeg(self, input_path: str, output_path: str) -> bool: frame_count = self._get_video_frame_count(input_path) if frame_count is None or frame_count < 2: if os.path.exists(input_path): os.rename(input_path, output_path) return True vf_filter = f"select='lt(n,{frame_count - 1})',setpts=PTS-STARTPTS" cmd_list = ['ffmpeg', '-y', '-i', input_path, '-vf', vf_filter, '-an', output_path] try: subprocess.run(cmd_list, check=True, capture_output=True, text=True, encoding='utf-8') return True except subprocess.CalledProcessError: return False 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] try: subprocess.run(cmd_list, check=True, capture_output=True, text=True) except subprocess.CalledProcessError as e: raise gr.Error(f"Falha na montagem final do vídeo. Detalhes: {e.stderr}") return output_path # --- PIPELINE DE PÓS-PRODUÇÃO LATENTE --- def _render_and_post_process_latents(self, low_res_latents: torch.Tensor, base_name: str, motion_prompt_for_refine: str, final_resolution: tuple ) -> str: logger.info(f"--- INICIANDO PIPELINE DE PÓS-PRODUÇÃO PARA '{base_name}' ---") high_res_latents = upscaler_specialist_singleton.upscale(low_res_latents) _, _, _, refined_h_latent, refined_w_latent = high_res_latents.shape video_h = refined_h_latent * self.ltx_manager.workers[0].pipeline.vae_scale_factor video_w = refined_w_latent * self.ltx_manager.workers[0].pipeline.vae_scale_factor num_latent_frames = high_res_latents.shape[2] num_video_frames = num_latent_frames * self.ltx_manager.workers[0].pipeline.video_scale_factor if isinstance(self.vae, CausalVideoAutoencoder): num_video_frames -= 1 refine_kwargs = { 'height': video_h, 'width': video_w, 'video_total_frames': num_video_frames, 'video_fps': 24, 'current_fragment_index': int(time.time()), 'motion_prompt': motion_prompt_for_refine, 'conditioning_items_data': [], 'denoise_strength': 0.4, 'refine_steps': 10 } final_latents, _ = self.ltx_manager.refine_latents(high_res_latents, **refine_kwargs) untrimmed_video_path = os.path.join(self.workspace_dir, f"{base_name}_untrimmed.mp4") trimmed_video_path = os.path.join(self.workspace_dir, f"{base_name}.mp4") pixel_tensor = self.latents_to_pixels(final_latents) logger.info(f"Redimensionando vídeo final de {pixel_tensor.shape[-2:]} para {final_resolution}") pixel_tensor_resized = torch.nn.functional.interpolate( pixel_tensor.squeeze(0), size=final_resolution, mode='bilinear', align_corners=False ).unsqueeze(0) self.save_video_from_tensor(pixel_tensor_resized, untrimmed_video_path, fps=24) del pixel_tensor, final_latents, high_res_latents, pixel_tensor_resized gc.collect() torch.cuda.empty_cache() success = self._trim_last_frame_ffmpeg(untrimmed_video_path, trimmed_video_path) if os.path.exists(untrimmed_video_path): os.remove(untrimmed_video_path) return trimmed_video_path if success else untrimmed_video_path def _generate_video_and_audio(self, silent_video_path: str, audio_prompt: str, base_name: str) -> str: 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) duration = float(result.stdout.strip()) except Exception: frame_count = self._get_video_frame_count(silent_video_path) duration = (frame_count / 24.0) if frame_count else 0 video_with_audio_path = audio_specialist_singleton.generate_audio_for_video( video_path=silent_video_path, prompt=audio_prompt, duration_seconds=duration) return video_with_audio_path # 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, video_resolution: int, use_continuity_director: bool, progress: gr.Progress = gr.Progress()): FPS = 24 FRAMES_PER_LATENT_CHUNK = 8 ECO_LATENT_CHUNKS = 2 total_frames_brutos = self._quantize_to_multiple(int(seconds_per_fragment * FPS), FRAMES_PER_LATENT_CHUNK) total_latents_brutos = total_frames_brutos // 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 if total_latents_brutos <= latents_a_podar + 1: raise gr.Error(f"A combinação de duração e poda é muito agressiva.") 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} keyframe_paths = [item[0] if isinstance(item, tuple) else item for item in keyframes] story_history = "" expected_height, expected_width = 768, 1152 downscale_factor = 2 / 3 downscaled_height = self._quantize_to_multiple(int(expected_height * downscale_factor), 8) downscaled_width = self._quantize_to_multiple(int(expected_width * downscale_factor), 8) target_resolution_tuple = (downscaled_height, downscaled_width) final_resolution_tuple = (expected_height, expected_width) eco_latent_for_next_loop = None dejavu_latent_for_next_loop = None num_transitions_to_generate = len(keyframe_paths) - 1 low_res_latent_fragments = [] for i in range(num_transitions_to_generate): fragment_index = i + 1 progress(i / (num_transitions_to_generate + 2), desc=f"Gerando Latentes do Fragmento {fragment_index}") 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[:, :, :ECO_LATENT_CHUNKS, :, :].clone() dejavu_latent_for_next_loop = last_trim[:, :, -1:, :, :].clone() latents_video = latents_brutos[:, :, :-(latents_a_podar-1), :, :].clone() latents_video = latents_video[:, :, 1:, :, :] if transition_type == "cut": eco_latent_for_next_loop, dejavu_latent_for_next_loop = None, None low_res_latent_fragments.append(latents_video) progress((num_transitions_to_generate) / (num_transitions_to_generate + 2), desc="Concatenando latentes...") tensors_para_concatenar = [] target_device = self.device for idx, tensor_frag in enumerate(low_res_latent_fragments): tensor_on_target_device = tensor_frag.to(target_device) if idx < len(low_res_latent_fragments) - 1: tensors_para_concatenar.append(tensor_on_target_device[:, :, :-1, :, :]) else: tensors_para_concatenar.append(tensor_on_target_device) final_concatenated_latents = torch.cat(tensors_para_concatenar, dim=2) progress((num_transitions_to_generate + 1) / (num_transitions_to_generate + 2), desc="Pós-produção (Upscale e Refinamento)...") base_name = f"final_movie_hq_{int(time.time())}" silent_video_path = self._render_and_post_process_latents( low_res_latents=final_concatenated_latents, base_name=base_name, motion_prompt_for_refine=global_prompt, final_resolution=final_resolution_tuple ) #progress((num_transitions_to_generate + 1.5) / (num_transitions_to_generate + 2), desc="Gerando paisagem sonora...") #video_with_audio_path = self._generate_video_and_audio( # silent_video_path=silent_video_path, # audio_prompt=global_prompt, # base_name=base_name #) yield {"final_path": silent_video_path} def _generate_latent_tensor_internal(self, conditioning_items, ltx_params, target_resolution, total_frames_to_generate): kwargs = { **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 } return self.ltx_manager.generate_latent_fragment(**kwargs) 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