aduc_orchestrator.py

# aduc_orchestrator.py
#
# AducSdr: Uma implementação aberta e funcional da arquitetura ADUC-SDR
# Copyright (C) 4 de Agosto de 2025  Carlos Rodrigues dos Santos
#
# Contato:
# Carlos Rodrigues dos Santos
# carlex22@gmail.com
# Rua Eduardo Carlos Pereira, 4125, B1 Ap32, Curitiba, PR, Brazil, CEP 8102025
#
# Repositórios e Projetos Relacionados:
# GitHub: https://github.com/carlex22/Aduc-sdr
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License...
# PENDING PATENT NOTICE: Please see NOTICE.md.
#
# Version: 3.0.0
#
# This version adapts the Maestro to command the new unified "Turbo Intergalactic
# Multiverse Engine": the Deformes7DEngine. The orchestrator is now a lean
# command layer, delegating all complex production and post-production tasks
# to its single, powerful Chief Engineer.

import os
import logging
from typing import List, Dict, Any, Generator, Tuple

import gradio as gr
from PIL import Image, ImageOps

# O Orquestrador agora só precisa de dois engenheiros:
# - O Pensador (2D) para a lógica criativa
# - O Engenheiro-Chefe (7D) para toda a execução
from engineers.deformes2D_thinker import deformes2d_thinker_singleton
from engineers.deformes7D import deformes7d_engine_singleton

logger = logging.getLogger(__name__)

class AducDirector:
    """
    Representa o Diretor de Cena, gerenciando o estado da produção.
    """
    def __init__(self, workspace_dir: str):
        self.workspace_dir = workspace_dir
        os.makedirs(self.workspace_dir, exist_ok=True)
        self.state: Dict[str, Any] = {}
        logger.info(f"The stage is set. Workspace at '{self.workspace_dir}'.")

    def update_state(self, key: str, value: Any) -> None:
        logger.info(f"Notating on the score: State '{key}' updated.")
        self.state[key] = value

    def get_state(self, key: str, default: Any = None) -> Any:
        return self.state.get(key, default)

class AducOrchestrator:
    """
    Implementa o Maestro (Γ), a camada de orquestração que comanda
    o Engenheiro-Chefe Deformes7D.
    """
    def __init__(self, workspace_dir: str):
        """
        Inicializa o Maestro e seu Engenheiro-Chefe.
        """
        self.director = AducDirector(workspace_dir)
        self.thinker = deformes2d_thinker_singleton
        self.chief_engineer = deformes7d_engine_singleton
        logger.info("ADUC Maestro is on the podium with the Chief Engineer (Deformes7D) ready.")

    def process_image_for_story(self, image_path: str, size: int, filename: str) -> str:
        """
        Pré-processa uma imagem de referência, padronizando-a.
        """
        img = Image.open(image_path).convert("RGB")
        img_square = ImageOps.fit(img, (size, size), Image.Resampling.LANCZOS)
        processed_path = os.path.join(self.director.workspace_dir, filename)
        img_square.save(processed_path)
        logger.info(f"Reference image processed and saved to: {processed_path}")
        return processed_path

    # --- TAREFAS DE PRÉ-PRODUÇÃO (Delegadas ao Pensador 2D) ---

    def task_generate_storyboard(self, prompt: str, num_keyframes: int, ref_image_paths: List[str],
                                 progress: gr.Progress) -> Tuple[List[str], str, Any]:
        """
        Delega a criação do storyboard para o Deformes2DThinker.
        """
        logger.info(f"Act 1, Scene 1: Delegating storyboard creation to the Thinker.")
        progress(0.2, desc="Consulting the Thinker for the script...")

        storyboard = self.thinker.generate_storyboard(prompt, num_keyframes, ref_image_paths)

        logger.info(f"The Thinker returned the script: {storyboard}")
        self.director.update_state("storyboard", storyboard)
        self.director.update_state("processed_ref_paths", ref_image_paths)
        return storyboard, ref_image_paths[0], gr.update(visible=True, open=True)

    # A geração de keyframes agora é parte da produção principal, então esta tarefa é removida.
    # A seleção de keyframes para o modo fotógrafo permanece.
    def task_select_keyframes(self, storyboard: List[str], base_ref_paths: List[str],
                              pool_ref_paths: List[str]) -> List[str]:
        logger.info(f"Act 1, Scene 2 (Photographer Mode): Delegating keyframe selection to the Thinker.")
        selected_paths = self.thinker.select_keyframes_from_pool(storyboard, base_ref_paths, pool_ref_paths)
        self.director.update_state("keyframes", selected_paths)
        return selected_paths


    # --- TAREFA DE PRODUÇÃO UNIFICADA (Delegada ao Engenheiro-Chefe 7D) ---

    def task_produce_full_movie(self, initial_ref_paths: List[str], global_prompt: str,
                                video_resolution: int, seconds_per_fragment: float,
                                # ... outros parâmetros do Deformes4D
                                trim_percent: int, handler_strength: float, dest_strength: float,
                                guidance_scale: float, stg_scale: float, inference_steps: int,
                                progress: gr.Progress) -> Dict[str, Any]:
        """
        Delega a produção completa do filme para o Deformes7DEngine.
        """
        logger.info("Maestro: All systems go. Engaging the Turbo Intergalactic Engine (Deformes7D)...")
        storyboard = self.director.get_state("storyboard", [])
        if not storyboard:
            raise gr.Error("Storyboard not generated. Please complete Step 1 first.")
            
        ltx_params = {
            "guidance_scale": guidance_scale,
            "stg_scale": stg_scale,
            "num_inference_steps": inference_steps
        }

        # A chamada agora é para a função unificada do motor 7D
        result = self.chief_engineer.generate_full_movie_interleaved(
            initial_ref_paths=initial_ref_paths,
            storyboard=storyboard,
            global_prompt=global_prompt,
            video_resolution=video_resolution,
            seconds_per_fragment=seconds_per_fragment,
            trim_percent=trim_percent,
            handler_strength=handler_strength,
            dest_strength=dest_strength,
            ltx_params=ltx_params,
            progress=progress
        )
        
        self.director.update_state("final_video_path", result["final_path"])
        self.director.update_state("all_keyframes", result["all_keyframes"])
        logger.info("Maestro: Deformes7D has completed the main production run.")
        return result

    # --- TAREFAS DE PÓS-PRODUÇÃO (Delegadas ao Engenheiro-Chefe 7D) ---

    def task_run_hd_mastering(self, source_video_path: str, model_version: str, steps: int, prompt: str, progress: gr.Progress) -> Generator[Dict[str, Any], None, None]:
        logger.info(f"Maestro: Delegating HD mastering to the Chief Engineer.")
        for update in self.chief_engineer.master_video_hd(
            source_video_path=source_video_path, model_version=model_version,
            steps=steps, prompt=prompt, progress=progress
        ):
            yield update
    
    def task_run_audio_generation(self, source_video_path: str, audio_prompt: str, progress: gr.Progress) -> Generator[Dict[str, Any], None, None]:
        logger.info(f"Maestro: Delegating audio generation to the Chief Engineer.")
        for update in self.chief_engineer.generate_audio(
            source_video_path=source_video_path, audio_prompt=audio_prompt, progress=progress
        ):
            yield update