app.5.4.py

# Euia-AducSdr: Uma implementação aberta e funcional da arquitetura ADUC-SDR para geração de vídeo coerente.
# Copyright (C) 4 de Agosto de 2025  Carlos Rodrigues dos Santos
#
# Contato:
# Carlos Rodrigues dos Santos
# carlex22@gmail.com
# Rua Eduardo Carlos Pereira, 4125, B1 Ap32, Curitiba, PR, Brazil, CEP 8102025
#
# Repositórios e Projetos Relacionados:
# GitHub: https://github.com/carlex22/Aduc-sdr
# Hugging Face: https://huggingface.co/spaces/Carlexx/Ltx-SuperTime-60Secondos/
# Hugging Face: https://huggingface.co/spaces/Carlexxx/Novinho/
#
# Este programa é software livre: você pode redistribuí-lo e/ou modificá-lo
# sob os termos da Licença Pública Geral Affero da GNU como publicada pela
# Free Software Foundation, seja a versão 3 da Licença, ou
# (a seu critério) qualquer versão posterior.

# --- app.py (NOVIM-5.5: O Fator Humano) ---

# --- Ato 1: A Convocação da Orquestra (Importações) ---
import gradio as gr
import torch
import os
import yaml
from PIL import Image, ImageOps, ExifTags
import shutil
import gc
import subprocess
import google.generativeai as genai
import numpy as np
import imageio
from pathlib import Path
import huggingface_hub
import json
import time

from inference import create_ltx_video_pipeline, load_image_to_tensor_with_resize_and_crop, ConditioningItem, calculate_padding
from dreamo_helpers import dreamo_generator_singleton

# --- Ato 2: A Preparação do Palco (Configurações) ---
config_file_path = "configs/ltxv-13b-0.9.8-distilled.yaml"
with open(config_file_path, "r") as file: PIPELINE_CONFIG_YAML = yaml.safe_load(file)

LTX_REPO = "Lightricks/LTX-Video"
models_dir = "downloaded_models_gradio"
Path(models_dir).mkdir(parents=True, exist_ok=True)
WORKSPACE_DIR = "aduc_workspace"
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")

VIDEO_FPS = 24
TARGET_RESOLUTION = 420
MAX_KEYFRAMES_UI = 10 # Limite de abas de keyframe na UI

print("Criando pipelines LTX na CPU (estado de repouso)...")
distilled_model_actual_path = huggingface_hub.hf_hub_download(repo_id=LTX_REPO, filename=PIPELINE_CONFIG_YAML["checkpoint_path"], local_dir=models_dir, local_dir_use_symlinks=False)
pipeline_instance = create_ltx_video_pipeline(
    ckpt_path=distilled_model_actual_path,
    precision=PIPELINE_CONFIG_YAML["precision"],
    text_encoder_model_name_or_path=PIPELINE_CONFIG_YAML["text_encoder_model_name_or_path"],
    sampler=PIPELINE_CONFIG_YAML["sampler"],
    device='cpu'
)
print("Modelos LTX prontos (na CPU).")


# --- Ato 3: As Partituras dos Músicos (Funções de Geração e Análise) ---

def robust_json_parser(raw_text: str) -> dict:
    try:
        start_index = raw_text.find('{'); end_index = raw_text.rfind('}')
        if start_index != -1 and end_index != -1 and end_index > start_index:
            json_str = raw_text[start_index : end_index + 1]; return json.loads(json_str)
        else: raise ValueError("Nenhum objeto JSON válido encontrado na resposta da IA.")
    except json.JSONDecodeError as e: raise ValueError(f"Falha ao decodificar JSON: {e}")

def run_storyboard_generation(num_fragments: int, prompt: str, initial_image_path: str):
    if not initial_image_path: raise gr.Error("Por favor, forneça uma imagem de referência inicial.")
    if not GEMINI_API_KEY: raise gr.Error("Chave da API Gemini não configurada!")
    prompt_file = "prompts/unified_storyboard_prompt.txt"
    with open(os.path.join(os.path.dirname(__file__), prompt_file), "r", encoding="utf-8") as f: template = f.read()
    director_prompt = template.format(user_prompt=prompt, num_fragments=int(num_fragments), image_metadata="")
    genai.configure(api_key=GEMINI_API_KEY)
    model = genai.GenerativeModel('gemini-1.5-flash'); img = Image.open(initial_image_path)
    response = model.generate_content([director_prompt, img])
    try:
        storyboard_data = robust_json_parser(response.text)
        storyboard = storyboard_data.get("scene_storyboard", [])
        if not storyboard or len(storyboard) != int(num_fragments): raise ValueError(f"A IA não gerou o número correto de cenas. Esperado: {num_fragments}, Recebido: {len(storyboard)}")
        return storyboard
    except Exception as e: raise gr.Error(f"O Roteirista (Gemini) falhou: {e}. Resposta: {response.text}")

def get_dreamo_prompt_for_transition(previous_image_path: str, target_scene_description: str) -> str:
    genai.configure(api_key=GEMINI_API_KEY)
    prompt_file = "prompts/img2img_evolution_prompt.txt"
    with open(os.path.join(os.path.dirname(__file__), prompt_file), "r", encoding="utf-8") as f: template = f.read()
    director_prompt = template.format(target_scene_description=target_scene_description)
    model = genai.GenerativeModel('gemini-1.5-flash'); img = Image.open(previous_image_path)
    response = model.generate_content([director_prompt, "Previous Image:", img])
    return response.text.strip().replace("\"", "")

def run_keyframe_generation(storyboard, ref_images_tasks, progress=gr.Progress()):
    if not storyboard: raise gr.Error("Nenhum roteiro para gerar keyframes.")
    initial_ref_image_path = ref_images_tasks[0]['image']
    if not initial_ref_image_path or not os.path.exists(initial_ref_image_path): raise gr.Error("A imagem de referência principal (à esquerda) é obrigatória.")
    log_history = ""; keyframe_paths = []
    try:
        pipeline_instance.to('cpu'); gc.collect(); torch.cuda.empty_cache()
        dreamo_generator_singleton.to_gpu()
        with Image.open(initial_ref_image_path) as img: width, height = (img.width // 32) * 32, (img.height // 32) * 32
        current_ref_image_path = initial_ref_image_path
        for i, scene_description in enumerate(storyboard):
            progress(i / len(storyboard), desc=f"Pintando Keyframe {i+1}/{len(storyboard)}")
            log_history += f"\n--- PINTANDO KEYFRAME {i+1}/{len(storyboard)} ---\n"
            dreamo_prompt = get_dreamo_prompt_for_transition(current_ref_image_path, scene_description)
            reference_items = []
            for item in ref_images_tasks:
                if item['image'] and os.path.exists(item['image']):
                    reference_items.append({'image_np': np.array(Image.open(item['image']).convert("RGB")), 'task': item['task']})
            log_history += f"  - Roteiro: '{scene_description}'\n  - Usando {len(reference_items)} referências visuais.\n  - Prompt do D.A.: \"{dreamo_prompt}\"\n"
            yield {keyframe_log_output: gr.update(value=log_history)}
            output_path = os.path.join(WORKSPACE_DIR, f"keyframe_{i+1}.png")
            image = dreamo_generator_singleton.generate_image_with_gpu_management(reference_items=reference_items, prompt=dreamo_prompt, width=width, height=height)
            image.save(output_path)
            keyframe_paths.append(output_path); current_ref_image_path = output_path
    except Exception as e: raise gr.Error(f"O Pintor (DreamO) ou Diretor de Arte (Gemini) falhou: {e}")
    finally: dreamo_generator_singleton.to_cpu(); gc.collect(); torch.cuda.empty_cache()
    log_history += "\nPintura de todos os keyframes concluída.\n"
    yield {keyframe_log_output: gr.update(value=log_history), keyframe_images_state: keyframe_paths}

def get_motion_prompt(user_prompt, start_path, end_path, scene_desc):
    return f"A smooth, cinematic transition from the start image towards the end image, focusing on: {scene_desc}"

def run_video_production(
    video_duration_seconds, video_fps, end_cond_strength,
    prompt_geral, keyframe_paths_from_ui, scene_storyboard, cfg, 
    progress=gr.Progress()
):
    valid_keyframes = [p for p in keyframe_paths_from_ui if p is not None and os.path.exists(p)]
    if not valid_keyframes or len(valid_keyframes) < 2: raise gr.Error("São necessários pelo menos 2 keyframes válidos para produzir um vídeo.")
    
    log_history = "\n--- FASE 3: Iniciando Produção...\n"
    yield {production_log_output: log_history, video_gallery_glitch: []}
    
    video_total_frames = int(video_duration_seconds * video_fps)
    seed = int(time.time())
    try:
        pipeline_instance.to('cuda')
        video_fragments = []; kinetic_memory_path = valid_keyframes[0]
        with Image.open(kinetic_memory_path) as img: width, height = img.size
        
        for i in range(len(valid_keyframes) - 1):
            fragment_num = i + 1
            progress(i / (len(valid_keyframes) - 1), desc=f"Filmando Fragmento {fragment_num}")
            
            start_path = kinetic_memory_path
            destination_path = valid_keyframes[i+1]
            
            motion_prompt = get_motion_prompt(prompt_geral, start_path, destination_path, scene_storyboard[i])
            
            conditioning_items_data = [(start_path, 0, 1.0), (destination_path, video_total_frames - 1, end_cond_strength)]
            
            fragment_path, _ = run_ltx_animation(
                current_fragment_index=fragment_num, motion_prompt=motion_prompt, 
                conditioning_items_data=conditioning_items_data, width=width, height=height, 
                seed=seed, cfg=cfg, progress=progress,
                video_total_frames=video_total_frames, video_fps=video_fps, use_attention_slicing=True, num_inference_steps=30
            )
            
            video_fragments.append(fragment_path)
            eco_output_path = os.path.join(WORKSPACE_DIR, f"eco_from_frag_{fragment_num}.png")
            kinetic_memory_path = extract_last_frame_as_image(fragment_path, eco_output_path)
            
            log_history += f"Fragmento {fragment_num} concluído.\n"
            yield {production_log_output: log_history, video_gallery_glitch: video_fragments}
            
        yield {production_log_output: log_history + "\nProdução concluída.", video_gallery_glitch: video_fragments, fragment_list_state: video_fragments}
    finally:
        pipeline_instance.to('cpu'); gc.collect(); torch.cuda.empty_cache()

def process_image_to_square(image_path: str, size: int = TARGET_RESOLUTION) -> str:
    if not image_path: return None
    try:
        img = Image.open(image_path).convert("RGB"); img_square = ImageOps.fit(img, (size, size), Image.Resampling.LANCZOS)
        output_path = os.path.join(WORKSPACE_DIR, f"initial_ref_{size}x{size}.png"); img_square.save(output_path)
        return output_path
    except Exception as e: raise gr.Error(f"Falha ao processar a imagem de referência: {e}")

def load_conditioning_tensor(media_path: str, height: int, width: int) -> torch.Tensor:
    return load_image_to_tensor_with_resize_and_crop(media_path, height, width)

def run_ltx_animation(
    current_fragment_index, motion_prompt, conditioning_items_data, 
    width, height, seed, cfg, progress,
    video_total_frames, video_fps, use_attention_slicing, num_inference_steps
):
    progress(0, desc=f"[Câmera LTX] Filmando Cena {current_fragment_index}...");
    output_path = os.path.join(WORKSPACE_DIR, f"fragment_{current_fragment_index}_full.mp4"); target_device = 'cuda' if torch.cuda.is_available() else 'cpu'
    try:
        if use_attention_slicing: pipeline_instance.enable_attention_slicing()
        conditioning_items = [ConditioningItem(load_conditioning_tensor(p, height, width).to(target_device), s, t) for p, s, t in conditioning_items_data]
        actual_num_frames = int(round((float(video_total_frames) - 1.0) / 8.0) * 8 + 1)
        padded_h, padded_w = ((height - 1) // 32 + 1) * 32, ((width - 1) // 32 + 1) * 32
        padding_vals = calculate_padding(height, width, padded_h, padded_w)
        for item in conditioning_items: item.media_item = torch.nn.functional.pad(item.media_item, padding_vals)
        
        first_pass_config = PIPELINE_CONFIG_YAML.get("first_pass", {}).copy()
        first_pass_config['num_inference_steps'] = int(num_inference_steps)

        kwargs = {"prompt": motion_prompt, "negative_prompt": "blurry, distorted, bad quality, artifacts", "height": padded_h, "width": padded_w, "num_frames": actual_num_frames, "frame_rate": video_fps, "generator": torch.Generator(device=target_device).manual_seed(int(seed) + current_fragment_index), "output_type": "pt", "guidance_scale": float(cfg), "timesteps": first_pass_config.get("timesteps"), "conditioning_items": conditioning_items, "decode_timestep": PIPELINE_CONFIG_YAML.get("decode_timestep"), "decode_noise_scale": PIPELINE_CONFIG_YAML.get("decode_noise_scale"), "stochastic_sampling": PIPELINE_CONFIG_YAML.get("stochastic_sampling"), "image_cond_noise_scale": 0.15, "is_video": True, "vae_per_channel_normalize": True, "mixed_precision": (PIPELINE_CONFIG_YAML.get("precision") == "mixed_precision"), "enhance_prompt": False, "decode_every": 4, "num_inference_steps": int(num_inference_steps)}
        
        result_tensor = pipeline_instance(**kwargs).images
        
        pad_l, pad_r, pad_t, pad_b = map(int, padding_vals); slice_h = -pad_b if pad_b > 0 else None; slice_w = -pad_r if pad_r > 0 else None
        cropped_tensor = result_tensor[:, :, :video_total_frames, pad_t:slice_h, pad_l:slice_w]; video_np = (cropped_tensor[0].permute(1, 2, 3, 0).cpu().float().numpy() * 255).astype(np.uint8)
        with imageio.get_writer(output_path, fps=video_fps, codec='libx264', quality=8) as writer:
            for i, frame in enumerate(video_np): writer.append_data(frame)
        return output_path, actual_num_frames
    finally:
        if use_attention_slicing: pipeline_instance.disable_attention_slicing()

def trim_video_to_frames(input_path: str, output_path: str, frames_to_keep: int) -> str:
    try:
        subprocess.run(f"ffmpeg -y -v error -i \"{input_path}\" -vf \"select='lt(n,{frames_to_keep})'\" -an \"{output_path}\"", shell=True, check=True, text=True)
        return output_path
    except subprocess.CalledProcessError as e: raise gr.Error(f"FFmpeg falhou ao cortar vídeo: {e.stderr}")

def extract_last_frame_as_image(video_path: str, output_image_path: str) -> str:
    try:
        subprocess.run(f"ffmpeg -y -v error -sseof -1 -i \"{video_path}\" -update 1 -q:v 1 \"{output_image_path}\"", shell=True, check=True, text=True)
        return output_image_path
    except subprocess.CalledProcessError as e: raise gr.Error(f"FFmpeg falhou ao extrair último frame: {e.stderr}")

def concatenate_and_trim_masterpiece(fragment_paths: list, progress=gr.Progress()):
    if not fragment_paths: raise gr.Error("Nenhum fragmento de vídeo para concatenar.")
    progress(0.5, desc="Montando a obra-prima final...");
    try:
        list_file_path = os.path.join(WORKSPACE_DIR, "concat_list.txt"); final_output_path = os.path.join(WORKSPACE_DIR, "masterpiece_final.mp4")
        with open(list_file_path, "w") as f:
            for p in fragment_paths: f.write(f"file '{os.path.abspath(p)}'\n")
        subprocess.run(f"ffmpeg -y -v error -f concat -safe 0 -i \"{list_file_path}\" -c copy \"{final_output_path}\"", shell=True, check=True, text=True)
        progress(1.0, desc="Montagem concluída!")
        return final_output_path
    except subprocess.CalledProcessError as e: raise gr.Error(f"FFmpeg falhou na concatenação final: {e.stderr}")

with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("# NOVIM-5.5 (O Fator Humano)\n*By Carlex & Gemini & DreamO*")
    if os.path.exists(WORKSPACE_DIR): shutil.rmtree(WORKSPACE_DIR)
    os.makedirs(WORKSPACE_DIR)
    
    scene_storyboard_state, keyframe_images_state, fragment_list_state = gr.State([]), gr.State([]), gr.State([])
    prompt_geral_state, processed_ref_path_state = gr.State(""), gr.State("")

    gr.Markdown("--- \n ## ETAPA 1: O ROTEIRO (IA Roteirista)")
    with gr.Row():
        with gr.Column(scale=1):
            prompt_input = gr.Textbox(label="Ideia Geral (Prompt)")
            num_fragments_input = gr.Slider(2, MAX_KEYFRAMES_UI, 4, step=1, label="Número de Atos (Keyframes)")
            image_input = gr.Image(type="filepath", label=f"Imagem de Referência Principal (será {TARGET_RESOLUTION}x{TARGET_RESOLUTION})")
            director_button = gr.Button("▶️ 1. Gerar Roteiro", variant="primary")
        with gr.Column(scale=2):
            storyboard_to_show = gr.JSON(label="Roteiro de Cenas Gerado")

    gr.Markdown("--- \n ## ETAPA 2: OS KEYFRAMES (IA Pintor & Diretor de Arte)")
    with gr.Row():
        with gr.Column(scale=2):
            with gr.Row():
                ref1_image = gr.Image(label="Referência Principal (Conteúdo/ID)", type="filepath")
                ref1_task = gr.Dropdown(choices=["ip", "id", "style"], value="ip", label="Tarefa da Ref. Principal")
            with gr.Row():
                ref2_image = gr.Image(label="Referência Secundária (Opcional)", type="filepath")
                ref2_task = gr.Dropdown(choices=["ip", "id", "style"], value="style", label="Tarefa da Ref. Secundária")
            photographer_button = gr.Button("▶️ 2. Pintar Imagens-Chave em Cadeia", variant="primary")
            keyframe_log_output = gr.Textbox(label="Diário de Bordo do Pintor", lines=10, interactive=False)
        with gr.Column(scale=1):
            gr.Markdown("### Painel de Edição de Keyframes")
            keyframe_ui_slots = []
            keyframe_ui_tabs_visibility = []
            with gr.Tabs() as keyframe_tabs:
                for i in range(MAX_KEYFRAMES_UI):
                    with gr.TabItem(f"Keyframe {i+1}", visible=(i<2)) as keyframe_tab:
                        keyframe_ui_slots.append(gr.Image(label=f"Conteúdo do Keyframe {i+1}", type="filepath", interactive=True))
                        keyframe_ui_tabs_visibility.append(keyframe_tab)

    gr.Markdown("--- \n ## ETAPA 3: A PRODUÇÃO (IA Cineasta & Câmera)")
    with gr.Row():
        with gr.Column(scale=1):
            cfg_slider = gr.Slider(1.0, 10.0, 7.5, step=0.1, label="CFG")
            end_cond_strength_slider = gr.Slider(label="Força de Convergência do Destino", minimum=0.1, maximum=1.0, value=1.0, step=0.05)
            with gr.Accordion("Controles Avançados de Timing", open=False):
                video_duration_slider = gr.Slider(label="Duração da Cena (segundos)", minimum=2.0, maximum=10.0, value=4.0, step=0.5)
                video_fps_slider = gr.Slider(label="FPS do Vídeo", minimum=12, maximum=36, value=VIDEO_FPS, step=1)
            animator_button = gr.Button("▶️ 3. Produzir Cenas (Handoff Cinético)", variant="primary")
            production_log_output = gr.Textbox(label="Diário de Bordo da Produção", lines=10, interactive=False)
        with gr.Column(scale=1):
            video_gallery_glitch = gr.Gallery(label="Fragmentos Gerados", object_fit="contain", height="auto", type="video")
    
    gr.Markdown(f"--- \n ## ETAPA 4: PÓS-PRODUÇÃO (Editor)")
    editor_button = gr.Button("▶️ 4. Montar Vídeo Final", variant="primary")
    final_video_output = gr.Video(label="A Obra-Prima Final", width=TARGET_RESOLUTION)
    
    def process_and_update_storyboard(num_fragments, prompt, image_path):
        processed_path = process_image_to_square(image_path)
        if not processed_path: raise gr.Error("A imagem de referência é inválida.")
        storyboard = run_storyboard_generation(num_fragments, prompt, processed_path)
        tab_updates = [gr.update(visible=(i < num_fragments)) for i in range(MAX_KEYFRAMES_UI)]
        return storyboard, prompt, processed_path, storyboard, processed_path, *tab_updates

    director_button.click(
        fn=process_and_update_storyboard, 
        inputs=[num_fragments_input, prompt_input, image_input], 
        outputs=[scene_storyboard_state, prompt_geral_state, processed_ref_path_state, storyboard_to_show, ref1_image] + keyframe_ui_tabs_visibility
    )
    
    def run_keyframe_generation_wrapper(storyboard, ref1_img, ref1_tsk, ref2_img, ref2_tsk, progress=gr.Progress()):
        ref_data = [{'image': ref1_img, 'task': ref1_tsk}, {'image': ref2_img, 'task': ref2_tsk}]
        final_update = {}
        for update in run_keyframe_generation(storyboard, ref_data, progress):
            final_update = update
        final_paths = final_update.get('keyframe_images_state', [])
        updates = [gr.update(value=final_paths[i] if i < len(final_paths) else None) for i in range(MAX_KEYFRAMES_UI)]
        return final_update.get('keyframe_log_output', ''), final_paths, *updates

    photographer_button.click(
        fn=run_keyframe_generation_wrapper,
        inputs=[scene_storyboard_state, ref1_image, ref1_task, ref2_image, ref2_task],
        outputs=[keyframe_log_output, keyframe_images_state] + keyframe_ui_slots
    )
    
    # A lista de inputs para a produção de vídeo agora coleta os keyframes das abas
    video_prod_inputs = [
        video_duration_slider, video_fps_slider, end_cond_strength_slider,
        prompt_geral_state, 
        scene_storyboard_state, cfg_slider
    ] + keyframe_ui_slots

    # A função wrapper é necessária para coletar os valores dos slots de keyframe
    def run_video_production_wrapper(duration, fps, strength, prompt, storyboard, cfg, *keyframes, progress=gr.Progress()):
        # Filtra os keyframes que não são None
        valid_keyframes = [k for k in keyframes if k]
        yield from run_video_production(duration, fps, strength, prompt, valid_keyframes, storyboard, cfg, progress)

    animator_button.click(
        fn=run_video_production_wrapper,
        inputs=video_prod_inputs,
        outputs=[production_log_output, video_gallery_glitch, fragment_list_state]
    )
    
    editor_button.click(
        fn=concatenate_and_trim_masterpiece, 
        inputs=[fragment_list_state], 
        outputs=[final_video_output]
    )

if __name__ == "__main__":
    demo.queue().launch(server_name="0.0.0.0", share=True)