# aduc_framework/engineers/planner_5D.py # # Versão 1.1.0 (Diretor de Produção Alinhado) # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos # # - Versão finalizada e alinhada com os especialistas Deformes v12+ e VaeManager v2. # - Cria as Ordens de Serviço (Jobs) usando as definições mais recentes de `types.py`. # - Delega os cálculos técnicos (ex: poda de frames) para os especialistas, # focando-se na orquestração de alto nível e no gerenciamento do loop de feedback visual. import logging import os from typing import Generator, Dict, Any # Importa os especialistas que serão dirigidos from ..engineers.deformes3D import deformes3d_engine_singleton as Deformes3D from ..engineers.deformes4D import deformes4d_engine_singleton as Deformes4D from ..tools.video_encode_tool import video_encode_tool_singleton as VideoTool # Importa as estruturas de dados (o "DNA" e as "Ordens de Serviço") from ..types import ( GenerationState, KeyframeGenerationJob, VideoGenerationJob, KeyframeData, VideoData ) logger = logging.getLogger(__name__) class Planner5D: """ Atua como o Diretor de Produção. Orquestra a geração de keyframes e vídeos cena a cena, implementando um loop de feedback para garantir continuidade visual. """ def __init__(self): self.dna: GenerationState | None = None self.workspace: str | None = None self.production_params: Dict[str, Any] = {} self.pre_prod_params: Dict[str, Any] = {} def produce_movie_by_scene(self, dna: GenerationState) -> Generator[GenerationState, None, None]: """ Ponto de entrada para a produção do filme. Executa o loop de produção cena a cena. """ self.dna = dna self.workspace = self.dna.workspace_dir if not (self.dna.parametros_geracao and self.dna.parametros_geracao.pre_producao and self.dna.parametros_geracao.producao): raise ValueError("Parâmetros de pré-produção ou produção ausentes no DNA. O processo não pode continuar.") self.pre_prod_params = self.dna.parametros_geracao.pre_producao.model_dump() self.production_params = self.dna.parametros_geracao.producao.model_dump() Deformes3D.initialize(self.workspace) Deformes4D.initialize(self.workspace) all_final_scene_clips = [] last_scene_final_frame_path = None self.dna.chat_history.append({"role": "Planner5D", "content": "Ok, storyboard recebido. Iniciando a produção do filme, cena por cena..."}) yield self.dna for scene_idx, scene in enumerate(self.dna.storyboard_producao): scene_title = scene.titulo_cena or f"Cena {scene.id_cena}" logger.info(f"--- PLANNER 5D: Iniciando Cena {scene.id_cena}: '{scene_title}' ---") self.dna.chat_history.append({"role": "Planner5D", "content": f"Preparando para filmar a Cena {scene.id_cena}: '{scene_title}'..."}) yield self.dna # ETAPA 1: GERAÇÃO DE KEYFRAMES reference_paths = [ref.caminho for ref in self.dna.midias_referencia if ref.caminho] ref_id_to_path_map = {i: path for i, path in enumerate(reference_paths)} keyframe_job = KeyframeGenerationJob( storyboard=[ato.NARRATIVA_VISUAL for ato in scene.atos], global_prompt=self.dna.texto_global_historia or "", ref_image_paths=reference_paths, ref_id_to_path_map=ref_id_to_path_map, available_ref_ids=list(ref_id_to_path_map.keys()), keyframe_prefix=f"s{scene.id_cena:02d}", params=self.pre_prod_params ) generated_keyframes_data = Deformes3D.generate_keyframes_from_job(keyframe_job) self.dna.storyboard_producao[scene_idx].keyframes = [KeyframeData(**kf) for kf in generated_keyframes_data] self.dna.chat_history.append({"role": "Planner5D", "content": f"Cena {scene.id_cena}: {len(generated_keyframes_data)} keyframes criados."}) yield self.dna for scene_idx, scene in enumerate(self.dna.storyboard_producao): scene_title = scene.titulo_cena or f"Cena {scene.id_cena}" logger.info(f"--- PLANNER 5D: Iniciando Gravação de Cena {scene.id_cena}: '{scene_title}' ---") self.dna.chat_history.append({"role": "Planner5D", "content": f"Filmando a Cena {scene.id_cena}: '{scene_title}'..."}) yield self.dna # ETAPA 2: GERAÇÃO DO VÍDEO if len(generated_keyframes_data) < 1: logger.warning(f"Cena {scene.id_cena} tem menos de 1 keyframes. Pulando geração de vídeo.") self.dna.chat_history.append({"role": "Planner5D", "content": f"AVISO: Cena {scene.id_cena} não pôde ser filmada (keyframes insuficientes)."}) yield self.dna continue self.dna.chat_history.append({"role": "Planner5D", "content": f"Cena {scene.id_cena}: Luz, câmera, ação! Gerando o clipe de vídeo..."}) yield self.dna # Criando a Ordem de Serviço para Deformes4D com os parâmetros de alto nível video_job = VideoGenerationJob( scene_id=scene.id_cena, global_prompt=self.dna.texto_global_historia or "", storyboard=[ato.NARRATIVA_VISUAL for ato in scene.atos], keyframe_paths=[kf['caminho_pixel'] for kf in generated_keyframes_data], **self.pre_prod_params, **self.production_params ) video_result = Deformes4D.generate_movie_clip_from_job(video_job) scene_clip_path = video_result.get("final_path") if scene_clip_path: all_final_scene_clips.append(scene_clip_path) video_data = video_result.get("video_data", {}) self.dna.storyboard_producao[scene_idx].video_gerado = VideoData(**video_data) self.dna.caminho_filme_final_bruto = scene_clip_path self.dna.chat_history.append({"role": "Planner5D", "content": f"Cena {scene.id_cena} filmada! O resultado será usado como ponto de partida para a próxima cena."}) yield self.dna # ETAPA FINAL: MONTAGEM DO FILME if not all_final_scene_clips: self.dna.chat_history.append({"role": "Planner5D", "content": "Produção falhou. Nenhum clipe de vídeo foi gerado."}) yield self.dna return self.dna.chat_history.append({"role": "Planner5D", "content": "Todas as cenas foram filmadas. Iniciando a montagem final do filme..."}) yield self.dna final_movie_path = os.path.join(self.workspace, "filme_final.mp4") final_movie_path = VideoTool.concatenate_videos(all_final_scene_clips, final_movie_path, self.workspace) self.dna.caminho_filme_final_bruto = final_movie_path self.dna.chat_history.append({"role": "Maestro", "content": "Produção concluída! O filme final está pronto para ser assistido."}) yield self.dna # --- Instância Singleton --- planner_5d_singleton = Planner5D()