Delete app.5.4.py

app.5.4.py

@@ -1,356 +0,0 @@
-# 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
-# [email protected]
-# 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)