# aduc_framework/engineers/deformes4D.py # # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos # # Versão 6.0.0 (Causal Physics Engine) # # Este engenheiro atua como a Câmera (Ψ) e a Sala de Edição (Δ). Sua única responsabilidade # é executar a Fórmula Canônica da ADUC-SDR: ele lê a sequência de Keyframes, # gera os fragmentos de vídeo brutos (realidade exploratória), e aplica o Ciclo de # Poda Causal para destilar os fragmentos canônicos (realidade observável). import os import time import imageio import numpy as np import torch import logging from PIL import Image, ImageOps import gc import shutil from typing import List, Tuple, Dict, Any, Callable, Optional from ..types import LatentConditioningItem from ..managers.ltx_manager import ltx_manager_singleton from ..managers.vae_manager import vae_manager_singleton from .deformes2D_thinker import deformes2d_thinker_singleton from ..tools.video_encode_tool import video_encode_tool_singleton logger = logging.getLogger(__name__) ProgressCallback = Optional[Callable[[float, str], None]] class Deformes4DEngine: """ Motor Causal ADUC-SDR. Orquestra a geração e a costura de fragmentos de vídeo no espaço latente, respeitando as decisões de corte e continuidade do Diretor Autônomo (Deformes2DThinker). """ _EDITOR_PARAMS = { "fps": 24, "ltx_frames_per_latent": 8, } def __init__(self): self.workspace_dir: Optional[str] = None self.device = 'cuda' if torch.cuda.is_available() else 'cpu' logger.info("Deformes4DEngine (Causal Physics Engine) instanciado.") def initialize(self, workspace_dir: str): self.workspace_dir = workspace_dir os.makedirs(self.workspace_dir, exist_ok=True) logger.info(f"Deformes4D Engine inicializado com workspace: {self.workspace_dir}.") def generate_original_movie( self, full_generation_state: Dict[str, Any], progress_callback: ProgressCallback = None ) -> Dict[str, Any]: """ Ponto de entrada principal. Gera o filme completo a partir do estado de geração. Etapa 1: Gera todos os fragmentos latentes canônicos usando o Ciclo de Poda Causal. Etapa 2: Decodifica os fragmentos latentes em clipes de vídeo, aplicando a "costura causal". Etapa 3: Monta os clipes de vídeo no filme final. """ if not self.workspace_dir: raise RuntimeError("Deformes4DEngine não foi inicializado. Chame o método initialize().") run_timestamp = int(time.time()) temp_latent_dir = os.path.join(self.workspace_dir, f"temp_latents_{run_timestamp}") os.makedirs(temp_latent_dir, exist_ok=True) # ETAPA 1: Geração dos fragmentos latentes estáveis (V_final) all_latent_paths, video_fragments_data = self._generate_all_latent_fragments( full_generation_state, temp_latent_dir, progress_callback ) # ETAPA 2: Decodificação dos fragmentos em clipes de vídeo video_clip_paths = self._decode_latents_to_clips( all_latent_paths, run_timestamp, progress_callback ) # ETAPA 3: Montagem final dos clipes final_video_path = os.path.join(self.workspace_dir, f"original_movie_{run_timestamp}.mp4") video_encode_tool_singleton.concatenate_videos(video_clip_paths, final_video_path, self.workspace_dir) logger.info(f"Processo de edição completo! Vídeo original salvo em: {final_video_path}") return { "final_path": final_video_path, "latent_paths": all_latent_paths, "video_data": { "id": 0, "caminho_pixel": final_video_path, "caminhos_latentes_fragmentos": all_latent_paths, "fragmentos_componentes": video_fragments_data } } def _generate_all_latent_fragments(self, state: Dict[str, Any], temp_dir: str, progress: ProgressCallback) -> Tuple[List[str], List[Dict]]: """ Implementa o "Ciclo de Poda Causal" para gerar a sequência de fragmentos latentes canônicos. Este é o motor da Câmera (Ψ) e do Destilador (Δ) operando em conjunto. """ # --- Configuração do Universo Físico --- keyframes_data = state.get("Keyframe_atos", []) if len(keyframes_data) < 2: raise ValueError("Geração de vídeo requer pelo menos 2 keyframes.") pre_prod_params = state.get("parametros_geracao", {}).get("pre_producao", {}) prod_params = state.get("parametros_geracao", {}).get("producao", {}) seconds_per_fragment = pre_prod_params.get('duration_per_fragment', 4.0) resolution = pre_prod_params.get('resolution', 512) trim_percent = prod_params.get('trim_percent', 50) # Realidade Exploratória: O comprimento da geração bruta para chegar ao destino. total_frames_brutos = self._quantize_to_multiple(int(seconds_per_fragment * self._EDITOR_PARAMS["fps"]), self._EDITOR_PARAMS["ltx_frames_per_latent"]) # Poda Causal: A quantidade de futuro entrópico a ser descartada. frames_a_podar = self._quantize_to_multiple(int(total_frames_brutos * (trim_percent / 100)), self._EDITOR_PARAMS["ltx_frames_per_latent"]) latents_a_podar = frames_a_podar // self._EDITOR_PARAMS["ltx_frames_per_latent"] # --- Inicialização do Loop de Geração --- eco_latent, dejavu_latent = None, None motion_history, all_latent_paths, video_fragments_data = "", [], [] num_transitions = len(keyframes_data) - 1 for i in range(num_transitions): if progress: progress(i / num_transitions * 0.7, f"Filmando Clipe {i+1}/{num_transitions}") start_kf, end_kf = keyframes_data[i], keyframes_data[i+1] # Se o Diretor marcou um "corte", a memória causal é resetada. if end_kf.get("is_cut_point", False): logger.info(f"Transição {i+1}: Diretor marcou 'CUT'. Resetando memória Déjà-Vu e Eco.") eco_latent, dejavu_latent = None, None # 1. Planejamento (Maestro Γ): Gerar a intenção (prompt) para o fragmento. motion_prompt = deformes2d_thinker_singleton.get_motion_decision(start_kf, end_kf, motion_history) motion_history += f"\n- {motion_prompt}" logger.info(f"Cineasta (Γ) decidiu o movimento para a transição {i+1}: '{motion_prompt[:50]}...'") # 2. Montagem das Âncoras (Fórmula Canônica): conditioning_items, dna_data = self._prepare_ltx_conditioning( start_kf, end_kf, (eco_latent, dejavu_latent), resolution, total_frames_brutos, prod_params ) # 3. Execução (Câmera Ψ): Gera a realidade exploratória (V_bruto). latents_brutos, _ = ltx_manager_singleton.generate_latent_fragment( height=resolution, width=resolution, conditioning_items_data=conditioning_items, motion_prompt=motion_prompt, video_total_frames=total_frames_brutos, video_fps=self._EDITOR_PARAMS["fps"], **prod_params ) # 4. Destilação e Poda (Destilador Δ): last_trim = latents_brutos[:, :, -(latents_a_podar + 1):, :, :].clone() eco_latent = last_trim[:, :, :2, :, :].clone() # Eco Causal (Cᵢ) dejavu_latent = last_trim[:, :, -1:, :, :].clone() # Déjà-Vu (Dᵢ) latents_video = latents_brutos[:, :, :-(latents_a_podar-1), :, :].clone() # Poda Causal latents_video = latents_video[:, :, 1:, :, :] # Normalização do frame inicial # 5. Armazenamento: Salva a realidade canônica (V_final) latent_path = os.path.join(temp_dir, f"latent_fragment_{i:04d}.pt") torch.save(latents_video.cpu(), latent_path) all_latent_paths.append(latent_path) video_fragments_data.append({"id": i, "prompt_video": motion_prompt, "entradas_latentes": dna_data}) return all_latent_paths, video_fragments_data def _decode_latents_to_clips(self, latent_paths: List[str], run_timestamp: int, progress: ProgressCallback) -> List[str]: """ Decodifica uma lista de fragmentos latentes em arquivos de vídeo .mp4 individuais. A lógica de concatenação e costura causal foi movida para o orquestrador. """ temp_dir = os.path.join(self.workspace_dir, f"temp_clips_{run_timestamp}") os.makedirs(temp_dir, exist_ok=True) video_clip_paths = [] num_latents = len(latent_paths) for i, latent_path in enumerate(latent_paths): if progress: progress(0.7 + (i / num_latents * 0.3), f"Decodificando Clipe {i+1}/{num_latents}") latent_tensor = torch.load(latent_path, map_location=self.device) pixel_tensor = vae_manager_singleton.decode(latent_tensor) clip_path = os.path.join(temp_dir, f"clip_{i:04d}.mp4") self._save_video_from_tensor(pixel_tensor, clip_path, fps=self._EDITOR_PARAMS["fps"]) video_clip_paths.append(clip_path) # Devolve o diretório temporário para o orquestrador gerenciar a limpeza. return video_clip_paths def _prepare_ltx_conditioning(self, start_kf, end_kf, dejavu_memory, res, total_frames, prod_params) -> Tuple[List[LatentConditioningItem], List[Dict]]: """Prepara os condicionais (Eco, Déjà-Vu, Keyframe) e os dados de DNA para o motor LTX.""" items, dna_data = [], [] res_tuple = (res, res) eco_latent, dejavu_latent = dejavu_memory def to_latent_tensor(path): pil = self._preprocess_image_for_latent_conversion(Image.open(path).convert("RGB"), res_tuple) tensor = self._pil_to_pixel_tensor(pil) return vae_manager_singleton.encode(tensor.to(self.device)) # Âncora 1: Condição Inicial (Herança do Passado - Eco Causal) if eco_latent is None: start_path = start_kf['caminho_pixel'] items.append(LatentConditioningItem(to_latent_tensor(start_path), 0, 1.0)) dna_data.append({"caminho_origem": start_path, "frame_alvo": 0, "forca_condicionamento": 1.0}) else: items.append(LatentConditioningItem(eco_latent, 0, 1.0)) dna_data.append({"caminho_origem": f"Memória Eco (C) do kf_id:{start_kf['id']-1}", "frame_alvo": 0, "forca_condicionamento": 1.0}) # Âncora 2: Atrator de Trajetória (Memória do Futuro - Déjà-Vu) dejavu_frame_target = self._quantize_to_multiple(int(total_frames * (prod_params.get('trim_percent', 50) / 100)), self._EDITOR_PARAMS["ltx_frames_per_latent"]) - 1 dejavu_frame_target = max(0, dejavu_frame_target) handler_strength = prod_params.get('handler_strength', 0.5) items.append(LatentConditioningItem(dejavu_latent, dejavu_frame_target, handler_strength)) dna_data.append({"caminho_origem": f"Memória Déjà-Vu (D) do kf_id:{start_kf['id']-1}", "frame_alvo": dejavu_frame_target, "forca_condicionamento": handler_strength}) # Âncora 3: Condição Final (Destino - Âncora Geométrica) end_path = end_kf['caminho_pixel'] destination_strength = prod_params.get('destination_convergence_strength', 0.75) destination_frame = total_frames - 1 items.append(LatentConditioningItem(to_latent_tensor(end_path), destination_frame, destination_strength)) dna_data.append({"caminho_origem": end_path, "frame_alvo": destination_frame, "forca_condicionamento": destination_strength}) return items, dna_data # --- Funções Utilitárias de Baixo Nível --- def _save_video_from_tensor(self, video_tensor: torch.Tensor, path: str, fps: int): if video_tensor is None or video_tensor.ndim != 5: return video = (video_tensor.squeeze(0).permute(1, 2, 3, 0).clamp(-1, 1) + 1) / 2.0 video_np = (video.cpu().float().numpy() * 255).astype(np.uint8) imageio.mimwrite(path, video_np, fps=fps, codec='libx264', quality=8, output_params=['-pix_fmt', 'yuv420p']) def _preprocess_image_for_latent_conversion(self, image: Image.Image, res: tuple): return ImageOps.fit(image, res, Image.Resampling.LANCZOS) if image.size != res else image def _pil_to_pixel_tensor(self, pil_image: Image.Image) -> torch.Tensor: arr = np.array(pil_image, dtype=np.float32) / 255.0 tensor = torch.from_numpy(arr).permute(2, 0, 1).unsqueeze(0).unsqueeze(2) return (tensor * 2.0) - 1.0 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