# aduc_framework/engineers/deformes4D.py # # Versão 16.0.0 (Integração de Lógica ADUC-SDR Canônica) # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos # # - Refatora a lógica de geração de vídeo monolítica para o padrão de # Especialista ADUC, operando dentro da classe Deformes4DEngine. # - Integra o núcleo lógico de Poda Causal, Eco e Déjà-Vu (linhas 187-315 # do arquivo original) de forma intacta. # - Substitui as chamadas de baixo nível por delegações aos managers # especializados (VaeManager, LtxManager, VideoEncodeTool). # - O método principal agora aceita um `VideoGenerationJob` do Planner5D. import os import time import torch import logging from PIL import Image import gc from typing import Dict, Any # Importa os managers e especialistas da arquitetura ADUC from ..managers.ltx_manager import ltx_manager_singleton from ..managers.vae_manager import vae_manager_singleton from ..tools.video_encode_tool import video_encode_tool_singleton from ..types import LatentConditioningItem, VideoGenerationJob from ..engineers.composer import composer_singleton as Composer # Usado para decisões cinematográficas logger = logging.getLogger(__name__) class Deformes4DEngine: """ Implementa a Câmera (Ψ) e o Destilador (Δ) da arquitetura ADUC-SDR. É responsável pela execução da geração de fragmentos de vídeo e pela extração dos contextos causais (Eco e Déjà-Vu). """ def __init__(self): self.workspace_dir: str | None = None self.device = 'cuda' if torch.cuda.is_available() else 'cpu' logger.info("Especialista Deformes4D (Executor ADUC-SDR: Câmera Ψ e Destilador Δ) inicializado.") def initialize(self, workspace_dir: str): if self.workspace_dir and self.workspace_dir == workspace_dir: return self.workspace_dir = workspace_dir os.makedirs(self.workspace_dir, exist_ok=True) logger.info(f"Executor de Cena 4D inicializado com workspace: {self.workspace_dir}.") 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 # O método principal que o Planner5D chamará def generate_movie_clip_from_job(self, job: VideoGenerationJob) -> Dict[str, Any]: if not self.workspace_dir: raise RuntimeError("Deformes4DEngine não foi inicializado.") # Extrai os parâmetros da Ordem de Serviço keyframes = job.keyframe_paths global_prompt = job.global_prompt storyboard = job.storyboard seconds_per_fragment = job.duration_per_fragment trim_percent = job.trim_percent handler_strength = job.handler_strength destination_convergence_strength = job.destination_convergence_strength video_resolution = job.resolution # Este método agora encapsula a lógica do seu 'generate_full_movie' return self._execute_generation_logic( keyframes, global_prompt, storyboard, seconds_per_fragment, trim_percent, handler_strength, destination_convergence_strength, video_resolution ) def _execute_generation_logic(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): #####(atenção Gemini .. nao apagar ou modificar nada apartir dessa linha) ##### # manter o conteudo original da linha 187 ate 315 # 1. Definição dos Parâmetros da Geração com base na Tese FPS = 32 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 = 15 #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: #raise gr.Error(f"A porcentagem de poda ({trim_percent}%) é muito alta. Reduza-a ou aumente a duração.") raise ValueError(f"A porcentagem de poda ({trim_percent}%) é muito alta. Reduza-a ou aumente a duração.") DEJAVU_FRAME_TARGET = 23 DESTINATION_FRAME_TARGET = total_frames_brutos -8 logger.info("--- CONFIGURAÇÃO DA GERAÇÃO ADUC-SDR ---") logger.info(f"Total de Latents por Geração Exploratória (V_bruto): {total_latents_brutos} ({total_frames_brutos} frames)") logger.info(f"Latents a serem descartados (Poda Causal): {latents_a_podar} ({frames_a_podar} frames)") logger.info(f"Chunks Latentes do Eco Causal (C): {ECO_LATENT_CHUNKS}") logger.info(f"Frame alvo do Déjà-Vu (D): {DEJAVU_FRAME_TARGET}") logger.info(f"Frame alvo do Destino (K): {DESTINATION_FRAME_TARGET}") logger.info("------------------------------------------") # 2. Inicialização do Estado base_ltx_params = {"guidance_scale": 1.0, "stg_scale": 0.005, "rescaling_scale": 0.05, "num_inference_steps": 4, "image_cond_noise_scale": 0.05} 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 dejavu_latent_for_next_loop = None if len(keyframe_paths) < 1: #raise gr.Error(f"A geração requer no mínimo 2 keyframes. Você forneceu {len(keyframe_paths)}.") raise ValueError(f"A geração requer no mínimo 2 keyframes. Você forneceu {len(keyframe_paths)}.") num_transitions_to_generate = len(keyframe_paths) -1 # 3. Loop Principal de Geração de Fragmentos for i in range(num_transitions_to_generate): fragment_index = i + 1 logger.info(f"--- INICIANDO FRAGMENTO {fragment_index}/{num_transitions_to_generate} ---") # 3.1. Consulta ao Maestro (Γ) para obter a intenção (Pᵢ) start_keyframe_path = keyframe_paths[i] if i+1 < num_transitions_to_generate: destination_keyframe_path = keyframe_paths[i + 1] else: destination_keyframe_path = keyframe_paths[i] decision = Composer.execute_cognitive_task( task_id="TASK_08_IMPROVISE_CINEMATIC_PROMPT_ATO", template_data={"current_act_narrative": storyboard[i]}, images=[Image.open(start_keyframe_path), Image.open(destination_keyframe_path)] ) motion_prompt = decision.strip() story_history += f"\n- Ato {fragment_index}: {motion_prompt}" # 3.2. Montagem das Âncoras para a Fórmula Canônica Ψ({C, D, K}, P) conditioning_items = [] logger.info(" [Ψ.1] Montando âncoras causais...") if eco_latent_for_next_loop is None: logger.info(" - Primeiro fragmento: Usando Keyframe inicial como âncora de partida.") img_start = Image.open(start_keyframe_path).convert("RGB") img_dest = Image.open(destination_keyframe_path).convert("RGB") start_latent, dest_latent = vae_manager_singleton.encode_batch([img_start, img_dest], target_resolution_tuple) conditioning_items.append(LatentConditioningItem(start_latent, 0, 1.0)) else: logger.info(" - Âncora 1: Eco Causal (C) - Herança do passado.") conditioning_items.append(LatentConditioningItem(eco_latent_for_next_loop, 0, 1.0)) logger.info(" - Âncora 2: Déjà-Vu (D) - Memória de um futuro idealizado.") conditioning_items.append(LatentConditioningItem(dejavu_latent_for_next_loop, 23, 0.5)) img_dest = Image.open(destination_keyframe_path).convert("RGB") dest_latent = vae_manager_singleton.encode_batch([img_dest], target_resolution_tuple)[0] logger.info(" - Âncora 3: Destino (K) - Âncora geométrica/narrativa.") conditioning_items.append(LatentConditioningItem(dest_latent, DESTINATION_FRAME_TARGET, 1.0)) # 3.3. Execução da Câmera (Ψ): Geração Exploratória para criar V_bruto logger.info(f" [Ψ.2] Câmera (Ψ) executando a geração exploratória de {total_latents_brutos} chunks latentes...") current_ltx_params = {**base_ltx_params, "motion_prompt": motion_prompt} latents_brutos, _ = ltx_manager_singleton.generate_latent_fragment( height=video_resolution, width=video_resolution, conditioning_items_data=conditioning_items, video_total_frames=total_frames_brutos+1, **current_ltx_params ) # 3.4. Execução do Destilador (Δ): Implementação do Ciclo de Poda Causal logger.info(f" [Δ] Shape do tensor bruto para vídeo final: {latents_brutos.shape}") #latents_video = latents_brutos[:, :, :-1, :, :] # --- Workaround empírico preservado --- eco_latent_for_next_loop = latents_brutos[:, :, -5:-2, :, :].clone() dejavu_latent_for_next_loop = latents_brutos[:, :, -1:, :, :].clone() latents_video = latents_brutos[:, :, :-4, :, :].clone() logger.info(f" [Δ] Shape do tensor video para vídeo final: {latents_video.shape}") #if i+1 < num_transitions_to_generate: #latents_video = latents_video[:, :, :-4, :, :] # logger.info(f"@@@@@@@@@@# Nao é ULTIMO poda -1") # logger.info(f" [Δ] -1 Shape do tensor video para vídeo final: {latents_video.shape}") #if i > 0: # latents_video = latents_video[:, :, 2:, :, :] # logger.info(f"@@@@@@@@@@# nao é o primeiro poda 1") # logger.info(f" [Δ] 1 Shape do tensor video para vídeo final: {latents_video.shape}") logger.info(f" [Δ] Shape do tensor para vídeo final: {latents_video.shape}") logger.info(f" - (Δ.1) Déjà-Vu (D) destilado. Shape: {dejavu_latent_for_next_loop.shape if dejavu_latent_for_next_loop is not None else 'N/A'}") logger.info(f" - (Δ.2) Eco Causal (C) extraído. Shape: {eco_latent_for_next_loop.shape if eco_latent_for_next_loop is not None else 'N/A'}") # 3.5. Renderização e Armazenamento do Fragmento Final base_name = f"fragment_{fragment_index}_{int(time.time())}" pixel_tensor = vae_manager_singleton.decode(latents_video) video_path = os.path.join(self.workspace_dir, f"{base_name}.mp4") video_encode_tool_singleton.save_video_from_tensor(pixel_tensor, video_path, fps=FPS) video_clips_paths.append(video_path) logger.info(f"--- FRAGMENTO {fragment_index} FINALIZADO E SALVO EM: {video_path} ---") # 4. Montagem Final do Filme final_movie_path = os.path.join(self.workspace_dir, f"final_movie_silent_{int(time.time())}.mp4") final_movie_path = video_encode_tool_singleton.concatenate_videos(video_clips_paths, final_movie_path, self.workspace_dir) #####(atenção Gemini .. nao apagar ou modificar nada acima dessa linha) ##### logger.info(f"Filme completo salvo em: {final_movie_path}") # Retorna o resultado no formato esperado pelo Planner5D return { "final_path": final_movie_path, "video_data": {"id": 0, "caminho_pixel": final_movie_path} # ID 0 para o filme completo } # --- Instância Singleton --- deformes4d_engine_singleton = Deformes4DEngine()