# aduc_framework/orchestrator.py # # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos # # Versão 3.1.0 (Framework Core com Inicialização Centralizada) # # Esta versão representa a camada de orquestração do Aduc Framework. # Ela é agnóstica a qualquer interface (UI ou API) e opera com # tipos de dados bem definidos (Pydantic) e um estado de geração central. # Agora, é responsável por inicializar os singletons dos engenheiros. import logging from typing import List, Dict, Any, Tuple, Callable, Optional from PIL import Image, ImageOps import os # Importa componentes internos do framework from .director import AducDirector from .types import GenerationState, PreProductionParams, ProductionParams from .engineers import ( deformes2d_thinker_singleton, deformes3d_engine_singleton, Deformes4DEngine ) logger = logging.getLogger(__name__) # Define um tipo para o callback de progresso para clareza ProgressCallback = Optional[Callable[[float, str], None]] class AducOrchestrator: """ Implementa o Maestro (Γ), a camada de orquestração central do Aduc Framework. Ele recebe solicitações, atualiza o estado de geração, delega tarefas para os engenheiros especialistas e retorna o estado atualizado. """ def __init__(self, workspace_dir: str): """ Inicializa o Maestro e seus componentes principais. Args: workspace_dir (str): O diretório raiz para salvar todos os artefatos gerados. """ self.director = AducDirector(workspace_dir) self.editor = Deformes4DEngine() # Instancia o editor self.editor.initialize(workspace_dir) # E o inicializa # Inicializa os singletons globais com a configuração necessária self.painter = deformes3d_engine_singleton self.painter.initialize(workspace_dir) logger.info("ADUC Maestro (Framework Core) está no pódio. Engenheiros especialistas prontos.") def get_current_state(self) -> GenerationState: """ Retorna o objeto de estado Pydantic completo e atual. Returns: GenerationState: O modelo Pydantic representando o estado completo. """ return self.director.get_full_state() def process_image_for_story(self, image_path: str, size: int, filename: str) -> str: """ Pré-processa uma imagem de referência, padronizando-a para uso pelos Especialistas. Este é um método utilitário exposto pelo framework. """ 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"Imagem de referência processada e salva em: {processed_path}") return processed_path # --- TAREFAS PRINCIPAIS DO FRAMEWORK --- def task_pre_production(self, params: PreProductionParams, progress_callback: ProgressCallback = None) -> Tuple[List[str], List[str], GenerationState]: """ Executa o fluxo completo de pré-produção: storyboard e geração de keyframes. """ logger.info("Maestro: Iniciando tarefa de Pré-Produção.") self.director.update_parameters("pre_producao", params) if progress_callback: progress_callback(0.1, "Gerando storyboard...") storyboard_list = deformes2d_thinker_singleton.generate_storyboard( prompt=params.prompt, num_keyframes=params.num_keyframes, ref_image_paths=params.ref_paths ) self.director.update_pre_production_state(params.prompt, params.ref_paths, storyboard_list) if progress_callback: progress_callback(0.2, "Iniciando geração de keyframes...") keyframes_detailed_data = self.painter.generate_keyframes_from_storyboard( generation_state=self.director.get_full_state_as_dict(), progress_callback=progress_callback ) self.director.update_keyframes_state(keyframes_detailed_data) final_keyframe_paths = [kf["caminho_pixel"] for kf in keyframes_detailed_data] final_state = self.director.get_full_state() logger.info("Maestro: Tarefa de Pré-Produção concluída.") return storyboard_list, final_keyframe_paths, final_state def task_produce_original_movie(self, params: ProductionParams, progress_callback: ProgressCallback = None) -> Tuple[str, List[str], GenerationState]: """ Executa o fluxo de produção do vídeo principal. """ logger.info("Maestro: Iniciando tarefa de Produção do Filme Original.") self.director.update_parameters("producao", params) result_data = self.editor.generate_original_movie( full_generation_state=self.director.get_full_state_as_dict(), progress_callback=progress_callback ) self.director.update_video_state(result_data["video_data"]) final_video_path = result_data["final_path"] latent_paths = result_data["latent_paths"] final_state = self.director.get_full_state() logger.info("Maestro: Tarefa de Produção do Filme Original concluída.") return final_video_path, latent_paths, final_state # --- TAREFAS DE PÓS-PRODUÇÃO (Exemplos para futura refatoração) --- def task_run_hd_mastering(self, source_video_path: str, hd_params: Dict[str, Any], progress_callback: ProgressCallback = None) -> str: logger.info(f"Maestro: Delegando tarefa de masterização HD.") # Lógica futura: # self.director.update_parameters("pos_producao", {"hd_mastering": hd_params}) # final_path = self.editor.master_video_hd(...) # self.director.update_pos_production_results(...) # return final_path return "" # Placeholder def task_run_audio_generation(self, source_video_path: str, audio_params: Dict[str, Any], progress_callback: ProgressCallback = None) -> str: logger.info(f"Maestro: Delegando tarefa de geração de áudio.") # Lógica futura: # self.director.update_parameters("pos_producao", {"audio_generation": audio_params}) # final_path = self.editor.generate_audio_for_final_video(...) # self.director.update_pos_production_results(...) # return final_path return "" # Placeholder