aduc_framework/engineers/deformes4D.py

# 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()