Aduc_sdr / app.py
Carlexxx
feat: Implement self-contained specialist managers
ac82132
raw
history blame
11.6 kB
# app.py
#
# Copyright (C) August 4, 2025 Carlos Rodrigues dos Santos
#
# Versão 3.0.0 (UI Head for Aduc Framework)
#
# Este arquivo implementa a interface de usuário com Gradio para o Aduc-Sdr.
# Ele atua como um cliente para o 'aduc_framework', que contém toda a
# lógica de negócio e orquestração.
import gradio as gr
import yaml
import logging
import os
import sys
import shutil
import time
import json
# --- 1. IMPORTAÇÃO DO FRAMEWORK E SEUS TIPOS ---
# A UI agora trata o Aduc-Sdr como um pacote que ela consome.
import aduc_framework
from aduc_framework.types import PreProductionParams, ProductionParams
# --- CUSTOM UI THEME E CONFIGURAÇÃO INICIAL ---
# (Esta seção permanece a mesma, pois é específica da aplicação)
cinematic_theme = gr.themes.Base(
primary_hue=gr.themes.colors.indigo,
secondary_hue=gr.themes.colors.purple,
neutral_hue=gr.themes.colors.slate,
font=(gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"),
).set(
body_background_fill="#111827", body_text_color="#E5E7EB",
button_primary_background_fill="linear-gradient(90deg, #4F46E5, #8B5CF6)",
button_primary_text_color="#FFFFFF", button_secondary_background_fill="#374151",
button_secondary_border_color="#4B5563", button_secondary_text_color="#E5E7EB",
block_background_fill="#1F2937", block_border_width="1px", block_border_color="#374151",
block_label_background_fill="#374151", block_label_text_color="#E5E7EB",
block_title_text_color="#FFFFFF", input_background_fill="#374151",
input_border_color="#4B5563", input_placeholder_color="#9CA3AF",
)
LOG_FILE_PATH = "aduc_log.txt"
if os.path.exists(LOG_FILE_PATH):
os.remove(LOG_FILE_PATH)
# Configuração de logging
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.setLevel(logging.INFO)
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.setLevel(logging.INFO)
file_handler.setFormatter(logging.Formatter(log_format))
root_logger.addHandler(file_handler)
logger = logging.getLogger(__name__)
# Carrega a configuração e inicializa o framework
try:
with open("config.yaml", 'r') as f: config = yaml.safe_load(f)
WORKSPACE_DIR = config['application']['workspace_dir']
# --- PONTO DE ENTRADA DO FRAMEWORK ---
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)
exit()
# --- 2. FUNÇÕES WRAPPER (CAMADA DE TRADUÇÃO UI <-> FRAMEWORK) ---
def run_pre_production_wrapper(prompt, num_keyframes, ref_files, resolution_str, duration_per_fragment, progress=gr.Progress()):
"""
Coleta dados da UI, os empacota em um objeto Pydantic e chama a tarefa de pré-produção do framework.
"""
if not ref_files:
raise gr.Error("Por favor, forneça pelo menos uma imagem de referência.")
# Etapa de UI: Processar e salvar os arquivos de referência
ref_paths = [aduc.process_image_for_story(f.name, 480, f"ref_processed_{i}.png") for i, f in enumerate(ref_files)]
# 1. Empacota os parâmetros da UI no modelo Pydantic que o framework espera
params = PreProductionParams(
prompt=prompt,
num_keyframes=int(num_keyframes),
ref_paths=ref_paths,
resolution=int(resolution_str.split('x')[0]),
duration_per_fragment=duration_per_fragment
)
# 2. Define a função de callback para o progresso
progress_callback = progress
# 3. Chama o framework
storyboard, final_keyframes, updated_state = aduc.task_pre_production(params, progress_callback)
# 4. Retorna os resultados desempacotados para os componentes corretos da UI
return updated_state.model_dump(), storyboard, final_keyframes, gr.update(visible=True, open=True)
def run_original_production_wrapper(current_state_dict, trim_percent, handler_strength, dest_strength, guidance_scale, stg_scale, steps, progress=gr.Progress()):
"""
Coleta os parâmetros da etapa de produção e o estado atual, e chama a tarefa de produção do framework.
"""
yield {
original_video_output: gr.update(value=None, visible=True, label="🎬 Produzindo seu filme..."),
final_video_output: gr.update(value=None, visible=True, label="🎬 Produção em progresso..."),
step4_accordion: gr.update(visible=False)
}
# 1. Empacota os parâmetros dos sliders no modelo Pydantic
production_params = ProductionParams(
trim_percent=int(trim_percent),
handler_strength=handler_strength,
destination_convergence_strength=dest_strength,
guidance_scale=guidance_scale,
stg_scale=stg_scale,
inference_steps=int(steps)
)
# 2. Chama a tarefa de produção no framework.
final_video_path, latent_paths, updated_state = aduc.task_produce_original_movie(
params=production_params,
progress_callback=progress
)
updated_state_dict = updated_state.model_dump()
# 3. Desempacota e retorna o resultado final para a UI
yield {
original_video_output: gr.update(value=final_video_path, label="✅ Filme Original Master"),
final_video_output: gr.update(value=final_video_path),
step4_accordion: gr.update(visible=True, open=True),
original_latents_paths_state: latent_paths,
original_video_path_state: final_video_path,
current_source_video_state: final_video_path,
generation_state_holder: updated_state_dict,
generation_data_output: updated_state_dict
}
def get_log_content():
try:
with open(LOG_FILE_PATH, "r", encoding="utf-8") as f:
return f.read()
except FileNotFoundError:
return "Arquivo de log ainda não criado. Inicie uma geração."
# --- 3. DEFINIÇÃO DA UI GRADIO ---
with gr.Blocks(theme=cinematic_theme, css="style.css") as demo:
# O gr.State é a "memória" da nossa UI. Ele armazena o JSON de estado entre os cliques.
generation_state_holder = gr.State(value={})
# Outros states para gerenciar caminhos de arquivos
original_latents_paths_state = gr.State(value=None)
original_video_path_state = gr.State(value=None)
current_source_video_state = gr.State(value=None)
upscaled_video_path_state = gr.State(value=None)
hd_video_path_state = gr.State(value=None)
gr.Markdown("<h1>ADUC-SDR 🎬 - O Diretor de Cinema IA</h1>")
gr.Markdown("<p>Crie um filme completo com vídeo e áudio, orquestrado por uma equipe de IAs especialistas.</p>")
with gr.Row():
lang_selector = gr.Radio(["🇧🇷", "🇺🇸", "🇨🇳"], value="🇧🇷", label="Idioma / Language")
resolution_selector = gr.Radio(["480x480", "720x720", "960x960"], value="480x480", label="Resolução Base")
with gr.Accordion("Etapa 1: Roteiro e Cenas-Chave (Pré-Produção)", open=True) as step1_accordion:
prompt_input = gr.Textbox(label="Ideia Geral do Filme", value="Um leão majestoso caminha pela savana, senta-se 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_keyframes_slider = gr.Slider(minimum=3, maximum=42, value=5, step=1, label="Número de Cenas-Chave")
duration_per_fragment_slider = gr.Slider(label="Duração de cada Clipe (s)", info="Duração alvo para cada fragmento de vídeo.", minimum=2.0, maximum=10.0, value=4.0, step=0.1)
storyboard_and_keyframes_button = gr.Button("Gerar Roteiro e Keyframes", variant="primary")
storyboard_output = gr.JSON(label="Roteiro Gerado (Storyboard)")
keyframe_gallery = gr.Gallery(label="Galeria de Cenas-Chave (Keyframes)", visible=True, object_fit="contain", height="auto", type="filepath")
with gr.Accordion("Etapa 3: Produção do Vídeo Original", 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 = gr.Slider(label="Força do Déjà-Vu", minimum=0.0, maximum=1.0, value=0.5, step=0.05)
dest_strength = 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 Original", variant="primary")
original_video_output = gr.Video(label="Filme Original Master", visible=False, interactive=False)
with gr.Accordion("Etapa 4: Pós-Produção (Opcional)", open=False, visible=False) as step4_accordion:
gr.Markdown("Aplique efeitos de melhoria ao vídeo mais recente.")
# ... (Componentes de pós-produção aqui) ...
with gr.Accordion("🧬 DNA Digital da Geração (JSON)", open=False) as data_accordion:
generation_data_output = gr.JSON(label="Estado de Geração Completo")
final_video_output = gr.Video(label="Filme Final (Resultado da Última Etapa)", visible=False, interactive=False)
with gr.Accordion("📝 Log de Geração (Detalhado)", open=False) as log_accordion:
log_display = gr.Textbox(label="Log da Sessão", lines=20, interactive=False, autoscroll=True)
update_log_button = gr.Button("Atualizar Log")
# --- 4. CONEXÕES DE EVENTOS DA UI ---
storyboard_and_keyframes_button.click(
fn=run_pre_production_wrapper,
inputs=[prompt_input, num_keyframes_slider, ref_image_input, resolution_selector, duration_per_fragment_slider],
outputs=[generation_state_holder, storyboard_output, keyframe_gallery, step3_accordion]
)
produce_original_button.click(
fn=run_original_production_wrapper,
inputs=[
generation_state_holder,
trim_percent_slider, handler_strength, dest_strength,
guidance_scale_slider, stg_scale_slider, inference_steps_slider
],
outputs=[
original_video_output, final_video_output, step4_accordion,
original_latents_paths_state, original_video_path_state, current_source_video_state,
generation_state_holder, generation_data_output
]
)
generation_state_holder.change(
fn=lambda state: state,
inputs=generation_state_holder,
outputs=generation_data_output
)
update_log_button.click(fn=get_log_content, inputs=[], outputs=[log_display])
# --- 5. INICIALIZAÇÃO DA APLICAÇÃO ---
if __name__ == "__main__":
if os.path.exists(WORKSPACE_DIR):
logger.info(f"Limpando workspace anterior em: {WORKSPACE_DIR}")
shutil.rmtree(WORKSPACE_DIR)
os.makedirs(WORKSPACE_DIR)
logger.info("Aplicação Gradio iniciada. Lançando interface...")
demo.queue().launch()