feat: Implement self-contained specialist managers

aduc_framework/engineers/__init__.py

@@ -6,12 +6,10 @@ from .deformes2D_thinker import deformes2d_thinker_singleton
 from .deformes3D_thinker import deformes3d_thinker_singleton
 from .deformes3D import deformes3d_engine_singleton
 from .deformes4D import Deformes4DEngine
-from .deformes7D import deformes7d_engine_singleton
 
 __all__ = [
     "deformes2d_thinker_singleton",
     "deformes3d_thinker_singleton",
     "deformes3d_engine_singleton",
     "Deformes4DEngine",
-    "deformes7d_engine_singleton",
 ]

aduc_framework/engineers/deformes7D.py

@@ -1,233 +0,0 @@
-# aduc_framework/engineers/deformes7D.py
-#
-# Copyright (C) August 4, 2025  Carlos Rodrigues dos Santos
-#
-# Versão 3.2.3 (Framework-Compliant com Inicialização Explícita)
-#
-# Este é o motor de geração unificado. Ele intercala a criação de keyframes (3D)
-# e fragmentos de vídeo (4D) em um único processo contínuo, potencialmente
-# economizando recursos e melhorando a coerência.
-
-import os
-import time
-import imageio
-import numpy as np
-import torch
-import logging
-from PIL import Image, ImageOps
-import subprocess
-import gc
-import yaml
-import shutil
-from pathlib import Path
-from typing import List, Tuple, Dict, Generator, Callable, Optional
-
-# --- Imports Relativos Corrigidos ---
-from ..types import LatentConditioningItem
-from ..managers.ltx_manager import ltx_manager_singleton
-from ..managers.latent_enhancer_manager import latent_enhancer_specialist_singleton
-from ..managers.vae_manager import vae_manager_singleton
-from .deformes2D_thinker import deformes2d_thinker_singleton
-from .deformes3D_thinker import deformes3d_thinker_singleton
-from ..managers.seedvr_manager import seedvr_manager_singleton
-from ..managers.mmaudio_manager import mmaudio_manager_singleton
-from ..tools.video_encode_tool import video_encode_tool_singleton
-
-logger = logging.getLogger(__name__)
-
-ProgressCallback = Optional[Callable[[float, str], None]]
-
-class Deformes7DEngine:
-    """
-    Motor unificado 3D/4D para geração contínua e intercalada de keyframes e
-    fragmentos de vídeo.
-    """
-    def __init__(self):
-        """O construtor é leve e não recebe argumentos."""
-        self.workspace_dir: Optional[str] = None
-        self.device = 'cuda' if torch.cuda.is_available() else 'cpu'
-        logger.info("Deformes7DEngine instanciado (não inicializado).")
-
-    def initialize(self, workspace_dir: str):
-        """Inicializa o motor unificado com as configurações necessárias."""
-        if self.workspace_dir is not None:
-            return  # Evita reinicialização
-        self.workspace_dir = workspace_dir
-        os.makedirs(self.workspace_dir, exist_ok=True)
-        logger.info(f"Deformes7D Unified Engine inicializado com workspace: {self.workspace_dir}.")
-
-    def generate_full_movie_interleaved(
-        self,
-        generation_state: Dict[str, Any],
-        progress_callback: ProgressCallback = None
-    ) -> Dict[str, Any]:
-        """
-        Gera um filme completo de forma intercalada, lendo todos os parâmetros
-        do estado de geração.
-        """
-        if not self.workspace_dir:
-            raise RuntimeError("Deformes7DEngine não foi inicializado. Chame o método initialize() antes de usar.")
-
-        logger.info("--- DEFORMES 7D: INICIANDO PIPELINE DE RENDERIZAÇÃO INTERCALADA ---")
-        
-        # 1. Extrai todos os parâmetros do estado
-        pre_prod_params = generation_state.get("parametros_geracao", {}).get("pre_producao", {})
-        prod_params = generation_state.get("parametros_geracao", {}).get("producao", {})
-        storyboard = [ato["resumo_ato"] for ato in generation_state.get("Atos", [])]
-        global_prompt = generation_state.get("Promt_geral", "")
-        initial_ref_paths = [media["caminho"] for media in generation_state.get("midias_referencia", [])]
-        
-        video_resolution = pre_prod_params.get('resolution', 480)
-        seconds_per_fragment = pre_prod_params.get('duration_per_fragment', 4.0)
-        
-        trim_percent = prod_params.get('trim_percent', 50)
-        handler_strength = prod_params.get('handler_strength', 0.5)
-        dest_strength = prod_params.get('destination_convergence_strength', 0.75)
-        ltx_params = {
-            "guidance_scale": prod_params.get('guidance_scale', 2.0),
-            "stg_scale": prod_params.get('stg_scale', 0.025),
-            "num_inference_steps": prod_params.get('inference_steps', 20)
-        }
-
-        # 2. Inicia o processo de geração
-        run_timestamp = int(time.time())
-        temp_video_clips_dir = os.path.join(self.workspace_dir, f"temp_clips_7D_{run_timestamp}")
-        os.makedirs(temp_video_clips_dir, exist_ok=True)
-        FPS = 24
-        FRAMES_PER_LATENT_CHUNK = 8
-        resolution_tuple = (video_resolution, video_resolution)
-        generated_keyframe_paths, generated_keyframe_latents, generated_video_fragment_paths = [], [], []
-        
-        if progress_callback: progress_callback(0, "Bootstrap: Processando K0...")
-        k0_path = initial_ref_paths[0]
-        k0_pil = Image.open(k0_path).convert("RGB")
-        k0_processed_pil = self._preprocess_image(k0_pil, resolution_tuple)
-        k0_pixel_tensor = self._pil_to_pixel_tensor(k0_processed_pil)
-        k0_latent = vae_manager_singleton.encode(k0_pixel_tensor)
-        generated_keyframe_paths.append(k0_path)
-        generated_keyframe_latents.append(k0_latent)
-        
-        if progress_callback: progress_callback(0.01, "Bootstrap: Gerando K1...")
-        prompt_k1 = deformes2d_thinker_singleton.get_anticipatory_keyframe_prompt(
-            global_prompt, "Cena inicial.", storyboard[0], storyboard[1], k0_path, initial_ref_paths
-        )
-        k1_path, k1_latent = self._generate_next_causal_keyframe(k0_path, initial_ref_paths, prompt_k1, resolution_tuple)
-        generated_keyframe_paths.append(k1_path)
-        generated_keyframe_latents.append(k1_latent)
-        
-        story_history = ""
-        eco_latent_for_next_loop, dejavu_latent_for_next_loop = None, None
-        num_transitions = len(storyboard) - 1
-        base_4d_ltx_params = {"rescaling_scale": 0.15, "image_cond_noise_scale": 0.00, **ltx_params}
-
-        for i in range(1, num_transitions):
-            act_progress = i / num_transitions
-            if progress_callback: progress_callback(act_progress, f"Ato {i+1}/{num_transitions} (Gerando Keyframe)...")
-            
-            logger.info(f"--> Etapa 3D: Gerando Keyframe K{i+1}")
-            kx_path = generated_keyframe_paths[i]
-            prompt_ky = deformes2d_thinker_singleton.get_anticipatory_keyframe_prompt(
-                global_prompt, story_history, storyboard[i], storyboard[i+1], kx_path, initial_ref_paths
-            )
-            ky_path, ky_latent = self._generate_next_causal_keyframe(kx_path, initial_ref_paths, prompt_ky, resolution_tuple)
-            generated_keyframe_paths.append(ky_path)
-            generated_keyframe_latents.append(ky_latent)
-            
-            if progress_callback: progress_callback(act_progress + (0.5 / num_transitions), f"Ato {i+1}/{num_transitions} (Gerando Vídeo)...")
-            
-            logger.info(f"--> Etapa 4D: Gerando Fragmento de Vídeo V{i-1}")
-            kb_path, kx_path, ky_path = generated_keyframe_paths[i-1], generated_keyframe_paths[i], generated_keyframe_paths[i+1]
-            motion_prompt = deformes3d_thinker_singleton.get_enhanced_motion_prompt(
-                global_prompt, story_history, kb_path, kx_path, ky_path,
-                storyboard[i-1], storyboard[i], storyboard[i+1]
-            )
-            story_history += f"\n- Ato {i}: {motion_prompt}"
-            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
-            conditioning_items = []
-            if eco_latent_for_next_loop is None:
-                conditioning_items.append(LatentConditioningItem(generated_keyframe_latents[i], 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))
-            conditioning_items.append(LatentConditioningItem(ky_latent, DESTINATION_FRAME_TARGET, dest_strength))
-            
-            fragment_latents_brutos, _ = ltx_manager_singleton.generate_latent_fragment(
-                height=video_resolution, width=video_resolution,
-                conditioning_items_data=conditioning_items, motion_prompt=motion_prompt,
-                video_total_frames=total_frames_brutos, video_fps=FPS, **base_4d_ltx_params
-            )
-            
-            last_trim = fragment_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()
-            final_fragment_latents = fragment_latents_brutos[:, :, :-(latents_a_podar-1), :, :].clone()
-            final_fragment_latents = final_fragment_latents[:, :, 1:, :, :]
-            pixel_tensor = vae_manager_singleton.decode(final_fragment_latents)
-            fragment_path = os.path.join(temp_video_clips_dir, f"fragment_{i-1:04d}.mp4")
-            self.save_video_from_tensor(pixel_tensor, fragment_path, fps=FPS)
-            generated_video_fragment_paths.append(fragment_path)
-            logger.info(f"Fragmento de Vídeo V{i-1} salvo em {fragment_path}")
-
-        logger.info("--- Montagem Final dos Fragmentos de Vídeo ---")
-        if progress_callback: progress_callback(0.98, "Montando o filme final...")
-        final_video_path = os.path.join(self.workspace_dir, f"movie_7D_{run_timestamp}.mp4")
-        video_encode_tool_singleton.concatenate_videos(generated_video_fragment_paths, final_video_path, self.workspace_dir)
-        shutil.rmtree(temp_video_clips_dir)
-        logger.info(f"Filme completo gerado em: {final_video_path}")
-        return {"final_path": final_video_path, "all_keyframes": generated_keyframe_paths}
-
-    def _generate_next_causal_keyframe(self, base_keyframe_path: str, all_ref_paths: list, 
-                                       prompt: str, resolution_tuple: tuple) -> Tuple[str, torch.Tensor]:
-        ltx_context_paths = [base_keyframe_path] + [p for p in all_ref_paths if p != base_keyframe_path][:3]
-        ltx_conditioning_items = []
-        weight = 1.0
-        for path in ltx_context_paths:
-            img_pil = Image.open(path).convert("RGB")
-            img_processed = self._preprocess_image(img_pil, resolution_tuple)
-            pixel_tensor = self._pil_to_pixel_tensor(img_processed)
-            latent_tensor = vae_manager_singleton.encode(pixel_tensor)
-            ltx_conditioning_items.append(LatentConditioningItem(latent_tensor, 0, weight))
-            if weight == 1.0: weight = -0.2
-            else: weight -= 0.2
-        ltx_base_params = {"guidance_scale": 3.0, "stg_scale": 0.1, "num_inference_steps": 25}
-        generated_latents, _ = ltx_manager_singleton.generate_latent_fragment(
-            height=resolution_tuple[0], width=resolution_tuple[1],
-            conditioning_items_data=ltx_conditioning_items, motion_prompt=prompt,
-            video_total_frames=48, video_fps=24, **ltx_base_params
-        )
-        final_latent = generated_latents[:, :, -1:, :, :]
-        upscaled_latent = latent_enhancer_specialist_singleton.upscale(final_latent)
-        pixel_tensor_out = vae_manager_singleton.decode(upscaled_latent)
-        timestamp = int(time.time() * 1000)
-        output_path = os.path.join(self.workspace_dir, f"keyframe_7D_{timestamp}.png")
-        self._save_image_from_tensor(pixel_tensor_out, output_path)
-        return output_path, final_latent
-
-    def _preprocess_image(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_pixel_tensor(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)
-        return (tensor * 2.0) - 1.0
-        
-    def _save_image_from_tensor(self, pixel_tensor: torch.Tensor, path: str):
-        tensor_chw = pixel_tensor.squeeze(0).squeeze(1)
-        tensor_hwc = tensor_chw.permute(1, 2, 0)
-        tensor_hwc = (tensor_hwc.clamp(-1, 1) + 1) / 2.0
-        image_np = (tensor_hwc.cpu().float().numpy() * 255).astype(np.uint8)
-        Image.fromarray(image_np).save(path)
-
-    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
-
-# --- Instanciação Singleton ---
-deformes7d_engine_singleton = Deformes7DEngine()