Spaces:
Paused
Paused
| # 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() |