# app.py # # Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos # # Versão 13.1.0 (UI Completa e Finalizada) # # Interface de usuário definitiva com fluxo de trabalho contínuo, chat unificado # e todas as etapas de produção e pós-produção conectadas. import gradio as gr import yaml import logging import os import sys import shutil import time import json from PIL import Image import gradio.themes as gt # --- 1. IMPORTAÇÃO DO FRAMEWORK E CONFIGURAÇÃO --- import aduc_framework from aduc_framework.types import PreProductionParams, ProductionParams # Configuração de Tema Cinemático cinematic_theme = gt.Soft( primary_hue="violet", secondary_hue="purple", neutral_hue="slate" ) # Configuração de Logging LOG_FILE_PATH = "aduc_log.txt" if os.path.exists(LOG_FILE_PATH): os.remove(LOG_FILE_PATH) log_format = '%(asctime)s - %(levelname)s - [%(name)s:%(funcName)s] - %(message)s' root_logger = logging.getLogger(); root_logger.setLevel(logging.INFO); root_logger.handlers.clear() stream_handler = logging.StreamHandler(sys.stdout); stream_handler.setFormatter(logging.Formatter(log_format)); root_logger.addHandler(stream_handler) file_handler = logging.FileHandler(LOG_FILE_PATH, mode='w', encoding='utf-8'); file_handler.setFormatter(logging.Formatter(log_format)); root_logger.addHandler(file_handler) logger = logging.getLogger(__name__) # Inicialização do Aduc Framework try: with open("config.yaml", 'r') as f: config = yaml.safe_load(f) WORKSPACE_DIR = config['application']['workspace_dir'] aduc = aduc_framework.create_aduc_instance(workspace_dir=WORKSPACE_DIR) logger.info("Interface Gradio inicializada e conectada ao Aduc Framework.") except Exception as e: logger.critical(f"ERRO CRÍTICO durante a inicialização: {e}", exc_info=True) with gr.Blocks(theme=cinematic_theme) as demo: gr.Markdown("# ERRO CRÍTICO NA INICIALIZAÇÃO") gr.Markdown("Não foi possível iniciar o Aduc Framework. Verifique os logs.") gr.Textbox(value=str(e), label="Detalhes do Erro", lines=10) demo.launch(); exit() # --- 2. FUNÇÕES WRAPPER E LÓGICA DA UI --- def chat_beautifier(role): if "Planner2D" in role: return "Arquiteto 2D" if "Composer" in role: return "Maestro Roteirista" if "Llama" in role: return "IA Criativa" if "Planner3D" in role: return "Arquiteto Visual 3D" if "Pintor3D" in role: return "Diretor de Arte" if "Deformes4D" in role: return "Motor Causal 4D" if "Maestro" in role: return "Maestro ADUC" return "Sistema" def process_chat_stream(generator, initial_chat_history=[]): chatbot_display_history = initial_chat_history.copy() fully_displayed_message_count = len(initial_chat_history) for update_package in generator: backend_chat_history = update_package.get("chat", []) while len(backend_chat_history) > fully_displayed_message_count: new_message_obj = backend_chat_history[fully_displayed_message_count] role = chat_beautifier(new_message_obj.get('role', 'Sistema')) content_to_type = new_message_obj.get('content', '') is_ai_message = any(tag in role for tag in ["IA Criativa", "Diretor de Arte", "Motor Causal"]) chatbot_display_history.append({"role": "assistant" if is_ai_message else "user", "content": ""}) full_typed_message = "" for char in f"**{role}:** {content_to_type}": full_typed_message += char chatbot_display_history[-1]["content"] = full_typed_message + "▌" yield { chat_history_chatbot: chatbot_display_history, dna_display: update_package.get("dna", gr.skip()), keyframe_gallery: gr.update(value=update_package.get("gallery", [])) } time.sleep(0.005) chatbot_display_history[-1]["content"] = full_typed_message fully_displayed_message_count += 1 update_dict = { chat_history_chatbot: chatbot_display_history, dna_display: update_package.get("dna", gr.skip()), keyframe_gallery: gr.update(value=update_package.get("gallery", [])), final_video_output: gr.update(value=update_package.get("final_video_path")), generation_state_holder: update_package.get("dna", gr.skip()), } if update_package.get("status") == "keyframes_complete": update_dict[step3_accordion] = gr.update(visible=True, open=True) elif update_package.get("status") == "production_complete": update_dict[step4_accordion] = gr.update(visible=True, open=True) update_dict[original_latents_paths_state] = update_package.get("latent_paths", []) update_dict[current_source_video_state] = update_package.get("final_video_path", "") yield update_dict def run_pre_production_and_keyframes(prompt, num_scenes, ref_files, duration_per_fragment): if not ref_files: raise gr.Error("Por favor, forneça pelo menos uma imagem de referência.") ref_paths = [aduc.process_image_for_story(f.name, f"ref_processed_{i}.jpg") for i, f in enumerate(ref_files)] params1 = PreProductionParams( prompt=prompt, num_scenes=int(num_scenes), ref_paths=ref_paths, resolution=512, # A resolução base ainda pode ser útil para o LTX duration_per_fragment=duration_per_fragment, fast_mode=False ) pre_prod_generator = aduc.task_pre_production(params1) yield from process_chat_stream(pre_prod_generator) current_state = aduc.get_current_state().model_dump(mode='json') dna_filepath = aduc.save_state_to_file() yield {dna_filepath_state: dna_filepath, generation_state_holder: current_state} keyframe_generator = aduc.task_generate_keyframes(current_state) yield from process_chat_stream(keyframe_generator, initial_chat_history=current_state.get("chat_history", [])) current_state = aduc.get_current_state().model_dump(mode='json') yield {generation_state_holder: current_state} def run_production_wrapper(state_dict, trim, handler, dest, guidance, stg, steps): params = ProductionParams( trim_percent=int(trim), handler_strength=handler, destination_convergence_strength=dest, guidance_scale=guidance, stg_scale=stg, inference_steps=int(steps) ) prod_generator = aduc.task_produce_original_movie(state_dict, params) yield from process_chat_stream(prod_generator, initial_chat_history=state_dict.get("chat_history", [])) # --- 3. DEFINIÇÃO DA UI --- with gr.Blocks(theme=cinematic_theme, css="style.css") as demo: generation_state_holder = gr.State({}) dna_filepath_state = gr.State() original_latents_paths_state = gr.State([]) current_source_video_state = gr.State() gr.Markdown("

ADUC-SDR 🎬 - O Diretor de Cinema IA

") gr.Markdown("

Uma interface para a tese 'A Genealogia da Falha'. Observe a equipe de IAs criar um filme a partir da sua ideia.

") with gr.Row(): with gr.Column(scale=2): with gr.Accordion("Etapa 1 & 2: Briefing e Geração de Cenas", open=True) as step1_and_2_accordion: prompt_input = gr.Textbox(label="Ideia Geral do Filme", value="Um leão majestoso caminha pela savana e ruge para o sol poente.") ref_image_input = gr.File(label="Imagens de Referência", file_count="multiple", file_types=["image"]) with gr.Row(): num_scenes_slider = gr.Slider(minimum=2, maximum=5, value=3, step=1, label="Número de Cenas") duration_per_fragment_slider = gr.Slider(label="Duração de cada Ato (s)", minimum=2.0, maximum=10.0, value=5.0, step=0.1) start_pre_prod_button = gr.Button("📝 Iniciar Roteiro e Keyframes (Etapas 1 & 2)", variant="primary") with gr.Accordion("Etapa 3: Parâmetros de Produção", open=False, visible=False) as step3_accordion: trim_percent_slider = gr.Slider(minimum=10, maximum=90, value=50, step=5, label="Poda Causal (%)") handler_strength_slider = gr.Slider(label="Força do Déjà-Vu", minimum=0.0, maximum=1.0, value=0.5, step=0.05) dest_strength_slider = gr.Slider(label="Força da Âncora Final", minimum=0.0, maximum=1.0, value=0.75, step=0.05) guidance_scale_slider = gr.Slider(minimum=1.0, maximum=10.0, value=2.0, step=0.1, label="Escala de Orientação") stg_scale_slider = gr.Slider(minimum=0.0, maximum=1.0, value=0.025, step=0.005, label="Escala STG") inference_steps_slider = gr.Slider(minimum=10, maximum=50, value=20, step=1, label="Passos de Inferência") produce_original_button = gr.Button("🎥 Produzir Vídeo (Etapa 3)", variant="primary") with gr.Accordion("Etapa 4: Pós-Produção (Opcional)", open=False, visible=False) as step4_accordion: gr.Markdown("Aplique melhorias ao filme. Cada etapa usa o resultado da anterior como fonte.") with gr.Accordion("A. Upscaler Latente 2x", open=True): upscaler_chunk_size_slider = gr.Slider(minimum=1, maximum=10, value=2, step=1, label="Fragmentos por Lote") run_upscaler_button = gr.Button("✨ Executar Upscaler Latente", variant="secondary") with gr.Column(scale=3): final_video_output = gr.Video(label="Filme Sendo Criado...", interactive=False) with gr.Accordion("🎥 Diário da Produção (Chat ao Vivo)", open=True): chat_history_chatbot = gr.Chatbot(label="Conversa da Equipe de IA", height=500, type='messages') with gr.Accordion("🖼️ Galeria de Keyframes", open=True): keyframe_gallery = gr.Gallery(label="Keyframes Gerados", object_fit="contain", height="auto") with gr.Accordion("🧬 DNA Digital (A Prova Viva dos Truques)", open=False): dna_display = gr.JSON(label="Estado Atual da Produção") # --- 4. CONEXÕES DE EVENTOS --- start_pre_prod_button.click( fn=run_pre_production_and_keyframes, inputs=[prompt_input, num_scenes_slider, ref_image_input, duration_per_fragment_slider], outputs=[ chat_history_chatbot, dna_display, keyframe_gallery, final_video_output, generation_state_holder, dna_filepath_state, step3_accordion, step4_accordion ] ) produce_original_button.click( fn=run_production_wrapper, inputs=[generation_state_holder, trim_percent_slider, handler_strength_slider, dest_strength_slider, guidance_scale_slider, stg_scale_slider, inference_steps_slider], outputs=[ chat_history_chatbot, dna_display, keyframe_gallery, final_video_output, generation_state_holder, original_latents_paths_state, current_source_video_state, step4_accordion ] ) generation_state_holder.change(fn=lambda s: s, inputs=generation_state_holder, outputs=dna_display) # --- 5. INICIALIZAÇÃO DA APLICAÇÃO --- if __name__ == "__main__": if os.path.exists(WORKSPACE_DIR): shutil.rmtree(WORKSPACE_DIR) os.makedirs(WORKSPACE_DIR) if not os.path.exists("./bot.png"): Image.new('RGB', (100, 100), color = '#6D28D9').save('bot.png') if not os.path.exists("./user.png"): Image.new('RGB', (100, 100), color = '#374151').save('user.png') logger.info("Aplicação Gradio iniciada. Lançando interface...") demo.queue().launch()