# aduc_framework/engineers/planner_4d.py # # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos # # Versão 6.5.0 (Montagem com Pontes de Transição) # # - Implementa uma nova etapa de pós-produção antes da concatenação final. # - Para cada par de clipes de cena, o planejador agora extrai o último frame # do primeiro clipe e o primeiro frame do segundo. # - Usa esses frames para gerar um pequeno vídeo de "ponte de transição". # - O filme final é montado intercalando os clipes originais com essas novas # transições, resultando em um produto final mais suave e profissional. import logging import os import time from typing import List, Dict, Any, Generator from ..types import VideoGenerationJob, VideoData from ..tools.video_encode_tool import video_encode_tool_singleton logger = logging.getLogger(__name__) class Planner4D: """ Atua como o Diretor de Fotografia, orquestrando a renderização de cenas e a montagem final com transições inteligentes. """ def _quantize_to_multiple(self, n: int, m: int) -> int: if m == 0: return n quantized = int(round(n / m) * m) return m if n > 0 and quantized == 0 else quantized def produce_movie_by_scene( self, generation_state: Dict[str, Any], initial_chat_history: List[Dict[str, Any]] ) -> Generator[Dict[str, Any], None, None]: from .deformes4D import deformes4d_engine_singleton as editor workspace_dir = generation_state.get("workspace_dir", "deformes_workspace") editor.initialize(workspace_dir) chat_history = initial_chat_history.copy() chat_history.append({"role": "Planner4D", "content": "Produção iniciada. Renderizando clipes de cena..."}) yield {"chat": chat_history} scenes = generation_state.get("scenes", []) pre_prod_params = generation_state.get("parametros_geracao", {}).get("pre_producao", {}) prod_params = generation_state.get("parametros_geracao", {}).get("producao", {}) global_prompt = generation_state.get("prompt_geral", "") FPS = 24 FRAMES_PER_LATENT_CHUNK = 8 seconds_per_fragment = pre_prod_params.get('duration_per_fragment', 4.0) trim_percent = prod_params.get('trim_percent', 50) total_frames_brutos_pixels = self._quantize_to_multiple(int(seconds_per_fragment * FPS), FRAMES_PER_LATENT_CHUNK) pixels_a_podar = self._quantize_to_multiple(int(total_frames_brutos_pixels * (trim_percent / 100)), FRAMES_PER_LATENT_CHUNK) latent_frames_a_podar = pixels_a_podar // FRAMES_PER_LATENT_CHUNK all_scene_clips_paths: List[str] = [] # --- ETAPA 1: GERAÇÃO DOS CLIPES DE CENA --- for i, scene in enumerate(scenes): # ... (Lógica de geração de job e chamada ao editor permanece a mesma) scene_id = scene.get('id') chat_history.append({"role": "Planner4D", "content": f"Preparando para filmar a Cena {scene_id}/{len(scenes)}..."}) yield {"chat": chat_history} storyboard_da_cena = [ato.get("resumo_ato", "") for ato in scene.get("atos", [])] keyframes_para_cena = scene.get("keyframes", []) if len(keyframes_para_cena) < 2: chat_history.append({"role": "Planner4D", "content": f"Cena {scene_id} pulada (keyframes insuficientes)."}) yield {"chat": chat_history} continue job = VideoGenerationJob( scene_id=scene_id, global_prompt=global_prompt, storyboard=storyboard_da_cena, keyframe_paths=[kf['caminho_pixel'] for kf in keyframes_para_cena], video_resolution=pre_prod_params.get('resolution', 480), handler_strength=prod_params.get('handler_strength', 0.5), destination_convergence_strength=prod_params.get('destination_convergence_strength', 0.75), 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), total_frames_brutos=total_frames_brutos_pixels, latents_a_podar=pixels_a_podar, latent_frames_a_podar=latent_frames_a_podar, DEJAVU_FRAME_TARGET=pixels_a_podar - 1 if pixels_a_podar > 0 else 0, DESTINATION_FRAME_TARGET=total_frames_brutos_pixels - 1, ) clip_result = editor.generate_movie_clip_from_job(job=job) if clip_result and clip_result.get("final_path"): final_clip_path = clip_result["final_path"] all_scene_clips_paths.append(final_clip_path) video_data_obj = VideoData(**clip_result["video_data"]).model_dump() if "videos" not in generation_state["scenes"][i]: generation_state["scenes"][i]["videos"] = [] generation_state["scenes"][i]["videos"].append(video_data_obj) chat_history.append({"role": "Planner4D", "content": f"Cena {scene_id} filmada com sucesso!"}) yield {"chat": chat_history, "final_video_path": final_clip_path, "dna": generation_state} if not all_scene_clips_paths: chat_history.append({"role": "Planner4D", "content": "Nenhum clipe de cena foi gerado."}) yield {"chat": chat_history, "status": "production_complete"} return # --- ETAPA 2: CRIAÇÃO DE PONTES DE TRANSIÇÃO (NOVA LÓGICA) --- chat_history.append({"role": "Planner4D", "content": "Todas as cenas filmadas. Criando transições suaves..."}) yield {"chat": chat_history} final_concat_list = [] temp_frames_to_delete = [] if len(all_scene_clips_paths) > 1: # Adiciona o primeiro clipe e entra no loop para criar pontes final_concat_list.append(all_scene_clips_paths[0]) for i in range(len(all_scene_clips_paths) - 1): clip_a_path = all_scene_clips_paths[i] clip_b_path = all_scene_clips_paths[i+1] # Define caminhos para os frames temporários last_frame_path = os.path.join(workspace_dir, f"temp_last_frame_{i}.png") first_frame_path = os.path.join(workspace_dir, f"temp_first_frame_{i+1}.png") temp_frames_to_delete.extend([last_frame_path, first_frame_path]) # Extrai os frames video_encode_tool_singleton.extract_last_frame(clip_a_path, last_frame_path) video_encode_tool_singleton.extract_first_frame(clip_b_path, first_frame_path) # Cria a ponte de transição bridge_video_path = video_encode_tool_singleton.create_transition_bridge( start_image_path=last_frame_path, end_image_path=first_frame_path, duration=0.3, # 1 segundo de transição fps=FPS, target_resolution=(pre_prod_params.get('resolution', 480), pre_prod_params.get('resolution', 480)), workspace_dir=workspace_dir ) # Adiciona a ponte e o próximo clipe à lista de montagem final_concat_list.append(bridge_video_path) final_concat_list.append(clip_b_path) else: # Se houver apenas um clipe, não há transições a criar final_concat_list = all_scene_clips_paths # --- ETAPA 3: MONTAGEM FINAL --- chat_history.append({"role": "Planner4D", "content": "Transições criadas. Montando o filme final..."}) yield {"chat": chat_history} final_video_path = video_encode_tool_singleton.concatenate_videos( video_paths=final_concat_list, output_path=f"{workspace_dir}/filme_final_com_transicoes.mp4", workspace_dir=workspace_dir ) # Limpa os frames temporários for frame_path in temp_frames_to_delete: if os.path.exists(frame_path): os.remove(frame_path) # Atualiza o DNA com o resultado final final_movie_data = VideoData(id=0, caminho_pixel=final_video_path, prompt_video=[]).model_dump() generation_state["filme_final"] = final_movie_data generation_state["chat_history"] = chat_history chat_history.append({"role": "Maestro", "content": "Produção concluída! O filme final está pronto."}) logger.info(f"Planner4D: Produção finalizada. Enviando filme '{final_video_path}' para a UI.") yield { "chat": chat_history, "final_video_path": final_video_path, "status": "production_complete", "dna": generation_state } planner_4d_singleton = Planner4D()