# 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. # # Este programa é distribuído na esperança de que seja útil, # mas SEM QUALQUER GARANTIA; sem mesmo a garantia implícita de # COMERCIALIZAÇÃO ou ADEQUAÇÃO A UM DETERMINADO FIM. Consulte a # Licença Pública Geral Affero da GNU para mais detalhes. # # Você deve ter recebido uma cópia da Licença Pública Geral Affero da GNU # junto com este programa. Se não, veja . # --- app.py (NOVINHO-4.2: Versão Final - Arquitetura "Memória, Caminho, Destino") --- # --- Ato 1: A Convocação da Orquestra (Importações) --- import gradio as gr import torch import os import yaml from PIL import Image, ImageOps 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 VIDEO_DURATION_SECONDS = 4 VIDEO_TOTAL_FRAMES = VIDEO_DURATION_SECONDS * VIDEO_FPS CONVERGENCE_FRAMES = 8 TARGET_RESOLUTION = 720 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 Corrigidas, Otimizadas e Documentadas) --- def load_conditioning_tensor(media_path: str, height: int, width: int) -> torch.Tensor: if not media_path: raise ValueError("Caminho da mídia de condicionamento não pode ser nulo.") # A lógica agora só precisa lidar com imagens, simplificando o processo 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=gr.Progress()): progress(0, desc=f"[TECPIX 5000] Filmando Cena {current_fragment_index}..."); output_path = os.path.join(WORKSPACE_DIR, f"fragment_{current_fragment_index}_full.mp4"); target_device = pipeline_instance.device try: conditioning_items = [] for (path, start_frame, strength) in conditioning_items_data: tensor = load_conditioning_tensor(path, height, width) conditioning_items.append(ConditioningItem(tensor.to(target_device), start_frame, strength)) n_val = round((float(VIDEO_TOTAL_FRAMES) - 1.0) / 8.0); actual_num_frames = int(n_val * 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 cond_item in conditioning_items: cond_item.media_item = torch.nn.functional.pad(cond_item.media_item, padding_vals) decode_every_val = 4 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": PIPELINE_CONFIG_YAML.get("first_pass", {}).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": decode_every_val } 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): progress(i / len(video_np), desc=f"Renderizando frame {i+1}/{len(video_np)}..."); writer.append_data(frame) return output_path, actual_num_frames except Exception as e: raise e def trim_video_to_frames(input_path: str, output_path: str, frames_to_keep: int) -> str: if not os.path.exists(input_path): raise gr.Error(f"Erro Interno: Vídeo de entrada para corte não encontrado: {input_path}") try: trim_cmd = (f"ffmpeg -y -v error -i \"{input_path}\" -vf \"select='lt(n,{frames_to_keep})'\" -an \"{output_path}\"") subprocess.run(trim_cmd, shell=True, check=True, capture_output=True, text=True) return output_path except subprocess.CalledProcessError as e: error_message = f"Editor Mágico (FFmpeg) falhou ao cortar o vídeo para {frames_to_keep} frames: {e}" if hasattr(e, 'stderr'): error_message += f"\nDetalhes: {e.stderr}" raise gr.Error(error_message) def extract_last_frame_as_image(video_path: str, output_image_path: str) -> str: if not os.path.exists(video_path): raise gr.Error(f"Erro Interno: Vídeo de entrada para extração de frame não encontrado: {video_path}") try: command = (f"ffmpeg -y -v error -sseof -1 -i \"{video_path}\" -update 1 -q:v 1 \"{output_image_path}\"") subprocess.run(command, shell=True, check=True, capture_output=True, text=True) return output_image_path except subprocess.CalledProcessError as e: error_message = f"Editor Mágico (FFmpeg) falhou ao extrair o último frame: {e}" if hasattr(e, 'stderr'): error_message += f"\nDetalhes: {e.stderr}" raise gr.Error(error_message) def process_image_to_square(image_path: str, size: int = TARGET_RESOLUTION) -> str: if not image_path or not os.path.exists(image_path): return None try: img = Image.open(image_path).convert("RGB") img_square = ImageOps.fit(img, (size, size), Image.Resampling.LANCZOS) output_filename = f"initial_ref_{size}x{size}.png" output_path = os.path.join(WORKSPACE_DIR, output_filename) 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 get_static_scenes_storyboard(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!") genai.configure(api_key=GEMINI_API_KEY) prompt_file = "prompts/photographer_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)) model = genai.GenerativeModel('gemini-2.0-flash'); img = Image.open(initial_image_path) response = model.generate_content([director_prompt, img]) try: cleaned_response = response.text.strip().replace("```json", "").replace("```", "") storyboard_data = json.loads(cleaned_response) return storyboard_data.get("scene_storyboard", []) except Exception as e: raise gr.Error(f"O Sonhador (Gemini) falhou ao criar o roteiro: {e}. Resposta: {response.text}") def run_keyframe_generation(storyboard, initial_ref_image_path, sequential_ref_task): if not storyboard: raise gr.Error("Nenhum roteiro para gerar imagens-chave.") if not initial_ref_image_path or not os.path.exists(initial_ref_image_path): raise gr.Error("A imagem de referência principal é obrigatória.") log_history = "" try: print("Pintor (DreamO): Movendo a Câmera (LTX) para a CPU para liberar VRAM...") pipeline_instance.to('cpu') gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() print("Pintor (DreamO): VRAM liberada. Movendo o Pintor para a GPU...") dreamo_generator_singleton.to_gpu() with Image.open(initial_ref_image_path) as img: width, height = img.size width, height = (width // 32) * 32, (height // 32) * 32 keyframe_paths, current_ref_image_path = [], initial_ref_image_path for i, prompt in enumerate(storyboard): log_history += f"\nPintando Cena {i+1}/{len(storyboard)}...\n" yield {keyframe_log_output: gr.update(value=log_history), keyframe_gallery_output: gr.update(value=keyframe_paths)} reference_items_for_dreamo = [{'image_np': np.array(Image.open(current_ref_image_path).convert("RGB")), 'task': sequential_ref_task}] log_history += f" - Usando referência: {os.path.basename(current_ref_image_path)} (Tarefa: {sequential_ref_task})\n" 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_for_dreamo, prompt=prompt, width=width, height=height) image.save(output_path) keyframe_paths.append(output_path) current_ref_image_path = output_path yield {keyframe_log_output: gr.update(value=log_history), keyframe_gallery_output: gr.update(value=keyframe_paths)} except Exception as e: raise gr.Error(f"O Pintor (DreamO) encontrou um erro: {e}") finally: print("Pintor (DreamO): Trabalho concluído. Movendo o Pintor de volta para a CPU.") dreamo_generator_singleton.to_cpu() gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() log_history += "\nPintura de todos os keyframes concluída.\n" yield {keyframe_log_output: gr.update(value=log_history), keyframe_gallery_output: gr.update(value=keyframe_paths), keyframe_images_state: keyframe_paths} def get_single_motion_prompt(user_prompt: str, story_history: str, start_image_path: str, middle_image_path: str, end_image_path: str): if not GEMINI_API_KEY: raise gr.Error("Chave da API Gemini não configurada!") try: genai.configure(api_key=GEMINI_API_KEY) model = genai.GenerativeModel('gemini-2.0-flash') start_img, middle_img, end_img = Image.open(start_image_path), Image.open(middle_image_path), Image.open(end_image_path) prompt_file_path = os.path.join(os.path.dirname(__file__), "prompts", "director_motion_prompt_three_act.txt") with open(prompt_file_path, "r", encoding="utf-8") as f: template = f.read() director_prompt = template.format(user_prompt=user_prompt, story_history=story_history) model_contents = [director_prompt, "INÍCIO:", start_img, "MEIO:", middle_img, "FIM:", end_img] response = model.generate_content(model_contents) cleaned_text = response.text.strip() if cleaned_text.startswith("```json"): cleaned_text = cleaned_text[len("```json"):].strip() if cleaned_text.endswith("```"): cleaned_text = cleaned_text[:-len("```")].strip() try: motion_data = json.loads(cleaned_text) final_prompt = motion_data.get("motion_prompt", "") if not final_prompt: raise ValueError("Prompt de movimento vazio no JSON.") return final_prompt except (json.JSONDecodeError, ValueError): return cleaned_text.replace("\"", "").replace("{", "").replace("}", "").replace("motion_prompt:", "").strip() except Exception as e: response_text = getattr(e, 'text', 'Nenhuma resposta de texto disponível.') raise gr.Error(f"O Cineasta (Gemini) falhou ao criar o prompt de movimento de 3 atos: {e}. Resposta: {response_text}") def run_video_production(prompt_geral, keyframe_images_state, scene_storyboard, seed, cfg, cut_frames_value, progress=gr.Progress()): if not keyframe_images_state or len(keyframe_images_state) < 2: raise gr.Error("Pinte pelo menos 2 keyframes na Etapa 2 para produzir as transições.") log_history = "\n--- FASE 3/4: A Câmera e o Cineasta estão filmando em sequência just-in-time...\n" yield {production_log_output: log_history, video_gallery_glitch: []} target_device = 'cuda' if torch.cuda.is_available() else 'cpu' try: print(f"Câmera (LTX): Movendo para a {target_device} para a produção em lote.") pipeline_instance.to(target_device) if target_device == 'cuda': if hasattr(pipeline_instance, 'disable_model_cpu_offload'): pipeline_instance.disable_model_cpu_offload() if hasattr(pipeline_instance, 'disable_attention_slicing'): pipeline_instance.disable_attention_slicing() if hasattr(pipeline_instance.vae, 'disable_slicing'): pipeline_instance.vae.disable_slicing() if hasattr(pipeline_instance.vae, 'disable_z_tiling'): pipeline_instance.vae.disable_z_tiling() video_fragments, story_history = [], "" previous_fragment_last_frame_path = keyframe_images_state[0] with Image.open(keyframe_images_state[0]) as img: width, height = img.size num_transitions = len(keyframe_images_state) - 1 for i in range(num_transitions): start_image_path = previous_fragment_last_frame_path middle_image_path = keyframe_images_state[i] end_image_path = keyframe_images_state[i+1] fragment_num = i + 1 is_last_fragment = (i == num_transitions - 1) progress(i / num_transitions, desc=f"Planejando e Filmando Fragmento {fragment_num}/{num_transitions}") log_history += f"\n--- FRAGMENTO {fragment_num} ---\n" story_history += f"\n- Transição de '{scene_storyboard[i]}' para '{scene_storyboard[i+1]}'." current_motion_prompt = get_single_motion_prompt(prompt_geral, story_history, start_image_path, middle_image_path, end_image_path) log_history += f"Instrução do Cineasta (3 Atos): '{current_motion_prompt}'\n" yield {production_log_output: log_history} foreshadow_frame, foreshadow_strength = 54, 0.3 end_frame_index = VIDEO_TOTAL_FRAMES - CONVERGENCE_FRAMES conditioning_items_data = [(start_image_path, 0, 1.0), (end_image_path, foreshadow_frame, foreshadow_strength), (end_image_path, end_frame_index, 1.0)] full_fragment_path, frames_gerados = run_ltx_animation(fragment_num, current_motion_prompt, conditioning_items_data, width, height, seed, cfg, progress) log_history += f" - Gerado: {frames_gerados} frames\n" if not is_last_fragment: cut_frames = int(cut_frames_value) final_fragment_path = os.path.join(WORKSPACE_DIR, f"fragment_{fragment_num}_final_{cut_frames}f.mp4") trim_video_to_frames(full_fragment_path, final_fragment_path, cut_frames) output_frame_path = os.path.join(WORKSPACE_DIR, f"last_frame_of_frag_{fragment_num}.png") previous_fragment_last_frame_path = extract_last_frame_as_image(final_fragment_path, output_frame_path) log_history += f" - Cortado para: {cut_frames} frames\n" log_history += f" - Memória para próxima cena: Último frame extraído\n" else: final_fragment_path = full_fragment_path log_history += f" - Último fragmento, mantendo duração total: {frames_gerados} frames\n" video_fragments.append(final_fragment_path) yield {production_log_output: log_history, video_gallery_glitch: video_fragments} log_history += "\nFilmagem de todos os fragmentos de transição concluída.\n" progress(1.0, desc="Produção Concluída.") yield {production_log_output: log_history, video_gallery_glitch: video_fragments, fragment_list_state: video_fragments} finally: print(f"Câmera (LTX): Produção em lote concluída. Movendo para a CPU para liberar VRAM.") pipeline_instance.to('cpu') gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() 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, final_output_path = os.path.join(WORKSPACE_DIR, "concat_list.txt"), os.path.join(WORKSPACE_DIR, "obra_prima_final.mp4") with open(list_file_path, "w") as f: for p in fragment_paths: f.write(f"file '{os.path.abspath(p)}'\n") concat_cmd = f"ffmpeg -y -v error -f concat -safe 0 -i \"{list_file_path}\" -c copy \"{final_output_path}\"" subprocess.run(concat_cmd, shell=True, check=True, capture_output=True, text=True) progress(1.0, desc="Montagem concluída!") return final_output_path except (subprocess.CalledProcessError, ValueError) as e: error_message = f"FFmpeg falhou durante a concatenação final: {e}" if hasattr(e, 'stderr'): error_message += f"\nDetalhes do erro do FFmpeg: {e.stderr}" raise gr.Error(error_message) # --- Ato 5: A Interface com o Mundo (A UI Restaurada e Aprimorada) --- with gr.Blocks(theme=gr.themes.Soft()) as demo: gr.Markdown("# NOVINHO-4.2 (Piloto de Testes - Arquitetura 'Memória, Caminho, Destino')\n*By Carlex & Gemini & DreamO*") if os.path.exists(WORKSPACE_DIR): shutil.rmtree(WORKSPACE_DIR) os.makedirs(WORKSPACE_DIR) Path("examples").mkdir(exist_ok=True) 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 (Sonhador)") with gr.Row(): with gr.Column(scale=1): prompt_input = gr.Textbox(label="Ideia Geral (Prompt)") num_fragments_input = gr.Slider(2, 10, 4, step=1, label="Número de Cenas") 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 de Cenas", variant="primary") with gr.Column(scale=2): storyboard_to_show = gr.JSON(label="Roteiro de Cenas Gerado") gr.Markdown("--- \n ## ETAPA 2: OS KEYFRAMES (Pintor)") with gr.Row(): with gr.Column(scale=2): gr.Markdown("### Controles do Pintor (DreamO)\n**Tarefas:** `style` (estilo), `ip` (conteúdo), `id` (identidade).") ref_image_inputs, ref_task_inputs = [], [] with gr.Group(): with gr.Row(): ref_image_inputs.append(gr.Image(label="Referência Inicial / Sequencial (Automática)", type="filepath", interactive=False)) ref_task_inputs.append(gr.Dropdown(choices=["ip", "id", "style"], value="ip", label="Tarefa da Referência")) photographer_button = gr.Button("▶️ 2. Pintar Imagens-Chave em Cadeia", variant="primary") with gr.Column(scale=1): keyframe_log_output = gr.Textbox(label="Diário de Bordo do Pintor", lines=15, interactive=False) keyframe_gallery_output = gr.Gallery(label="Imagens-Chave Pintadas", object_fit="contain", height="auto", type="filepath") gr.Markdown("--- \n ## ETAPA 3: A PRODUÇÃO (Cineasta e Câmera)") with gr.Row(): with gr.Column(scale=1): with gr.Row(): seed_number = gr.Number(42, label="Seed") cfg_slider = gr.Slider(1.0, 10.0, 2.5, step=0.1, label="CFG") cut_frames_slider = gr.Slider(label="Duração do Fragmento (Frames)", minimum=36, maximum=VIDEO_TOTAL_FRAMES, value=90, step=1) animator_button = gr.Button("▶️ 3. Produzir Cenas em Vídeo", variant="primary") production_log_output = gr.Textbox(label="Diário de Bordo da Produção", lines=15, 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) gr.Markdown( """ --- ### A Arquitetura "Memória, Caminho, Destino" Nossa geração de vídeo é governada por uma orquestração de IAs, onde cada fragmento (`V_i`) é criado com base em três pilares: * **Memória (`M_(i-1)`):** O último frame do fragmento anterior. Garante a **continuidade** visual. * **Caminho (`Γ_i`):** Uma instrução de movimento gerada pelo "Cineasta" (Gemini) ao analisar a Memória, o Keyframe atual e o Destino. Define a **narrativa** da transição. * **Destino (`K_(i+1)`):** O próximo keyframe a ser alcançado. Define o **objetivo** da animação. A Câmera (LTX) recebe esses três elementos para construir cada cena, resultando em um vídeo coeso e com propósito. """ ) # --- Ato 6: A Regência (Lógica de Conexão dos Botões) --- director_button.click( fn=get_static_scenes_storyboard, inputs=[num_fragments_input, prompt_input, image_input], outputs=[scene_storyboard_state] ).success( fn=lambda s, p: (s, p), inputs=[scene_storyboard_state, prompt_input], outputs=[storyboard_to_show, prompt_geral_state] ).success( fn=process_image_to_square, inputs=[image_input], outputs=[processed_ref_path_state] ).success( fn=lambda p: p, inputs=[processed_ref_path_state], outputs=[ref_image_inputs[0]] ) photographer_button.click( fn=run_keyframe_generation, inputs=[scene_storyboard_state, processed_ref_path_state, ref_task_inputs[0]], outputs=[keyframe_log_output, keyframe_gallery_output, keyframe_images_state] ) animator_button.click( fn=run_video_production, inputs=[prompt_geral_state, keyframe_images_state, scene_storyboard_state, seed_number, cfg_slider, cut_frames_slider], 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)