Delete app-old.py

app-old.py

@@ -1,215 +0,0 @@
-# --- app.py (O Painel de Controle do Maestro - Produção em Lote com Diário de Bordo) ---
-# By Carlex & Gemini
-
-# --- Ato 1: A Convocação da Orquestra (Importações) ---
-import gradio as gr
-import torch
-import spaces
-import os
-import yaml
-from PIL import Image
-import shutil
-import gc
-import traceback
-import subprocess
-import math
-import google.generativeai as genai
-import numpy as np
-import imageio
-import tempfile
-from pathlib import Path
-from huggingface_hub import hf_hub_download
-import json
-from facexlib.utils.face_restoration_helper import FaceRestoreHelper
-import huggingface_hub
-import spaces
-import argparse
-
-import spaces
-import argparse
-
-
-import cv2
-
-from facexlib.utils.face_restoration_helper import FaceRestoreHelper
-import huggingface_hub
-
-
-
-from dreamo.dreamo_pipeline import DreamOPipeline
-from dreamo.utils import img2tensor, resize_numpy_image_area, tensor2img, resize_numpy_image_long
-from tools import BEN2
-
-
-# --- Músicos Originais (Sua implementação) ---
-from inference import create_ltx_video_pipeline, load_image_to_tensor_with_resize_and_crop, seed_everething, calculate_padding
-from ltx_video.pipelines.pipeline_ltx_video import ConditioningItem
-
-# --- 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)
-
-# --- Constantes Globais ---
-LTX_REPO = "Lightricks/LTX-Video"
-models_dir = "downloaded_models_gradio_cpu_init"
-Path(models_dir).mkdir(parents=True, exist_ok=True)
-WORKSPACE_DIR = "aduc_workspace"
-GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")
-
-# --- Carregamento de Modelos LTX na CPU ---
-print("Baixando e criando pipelines LTX na CPU...")
-distilled_model_actual_path = 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.")
-
-
-# --- Ato 3: As Partituras dos Músicos (Funções) ---
-
-def get_storyboard_from_director_v2(num_fragments: int, prompt: str, initial_image_path: str, progress=gr.Progress()):
-    progress(0.5, desc="[Diretor Gemini] Criando o storyboard completo...")
-    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 (GEMINI_API_KEY) não configurada!")
-    genai.configure(api_key=GEMINI_API_KEY)
-    try:
-        with open("prompts/director_storyboard_v2.txt", "r", encoding="utf-8") as f: template = f.read()
-    except FileNotFoundError: raise gr.Error("'prompts/director_storyboard_v2.txt' não encontrado!")
-    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)
-        storyboard_list = storyboard_data.get("storyboard", [])
-        if not storyboard_list: raise gr.Error("A IA não retornou um storyboard válido.")
-        return storyboard_list
-    except (json.JSONDecodeError, KeyError, TypeError) as e:
-        raise gr.Error(f"O Diretor retornou uma resposta inesperada. Erro: {e}\nResposta Bruta: {response.text}")
-
-def run_ltx_animation(current_fragment_index, motion_prompt, input_frame_path, height, width, fps, seed, cfg, progress=gr.Progress()):
-    progress(0, desc=f"[Animador LTX] Aquecendo para a Cena {current_fragment_index}...")
-    target_device = "cuda"; output_path = os.path.join(WORKSPACE_DIR, f"fragment_{current_fragment_index}.mp4")
-    try:
-        pipeline_instance.to(target_device)
-        duration_fragment, target_frames_ideal = 3.0, 3.0 * fps
-        n_val = round((float(round(target_frames_ideal)) - 1.0) / 8.0); actual_num_frames = max(9, min(int(n_val * 8 + 1), 257))
-        num_frames_padded = ((actual_num_frames - 2) // 8 + 1) * 8 + 1
-        padded_h, padded_w = ((int(height) - 1) // 32 + 1) * 32, ((int(width) - 1) // 32 + 1) * 32
-        padding_vals = calculate_padding(int(height), int(width), padded_h, padded_w)
-        timesteps = PIPELINE_CONFIG_YAML.get("first_pass", {}).get("timesteps")
-        kwargs = {"prompt": motion_prompt, "negative_prompt": "blurry, distorted", "height": padded_h, "width": padded_w, "num_frames": num_frames_padded, "frame_rate": int(fps), "generator": torch.Generator(device=target_device).manual_seed(int(seed) + current_fragment_index), "output_type": "pt", "guidance_scale": float(cfg), "timesteps": timesteps, "vae_per_channel_normalize": True, "decode_timestep": PIPELINE_CONFIG_YAML["decode_timestep"], "decode_noise_scale": PIPELINE_CONFIG_YAML["decode_noise_scale"], "stochastic_sampling": PIPELINE_CONFIG_YAML["stochastic_sampling"], "image_cond_noise_scale": 0.15, "is_video": True, "mixed_precision": (PIPELINE_CONFIG_YAML["precision"] == "mixed_precision"), "offload_to_cpu": False, "enhance_prompt": False}
-        media_tensor = load_image_to_tensor_with_resize_and_crop(input_frame_path, int(height), int(width)); media_tensor = torch.nn.functional.pad(media_tensor, padding_vals); kwargs["conditioning_items"] = [ConditioningItem(media_tensor.to(target_device), 0, 1.0)]
-        result_tensor = pipeline_instance(**kwargs).images
-        pad_l, pad_r, pad_t, pad_b = padding_vals; slice_h, slice_w = (-pad_b if pad_b > 0 else None), (-pad_r if pad_r > 0 else None)
-        cropped_tensor = result_tensor[:, :, :actual_num_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=int(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
-    finally:
-        pipeline_instance.to("cpu"); gc.collect(); torch.cuda.empty_cache()
-
-def concatenate_masterpiece(fragment_paths: list, progress=gr.Progress()):
-    progress(0.5, desc="Montando a obra-prima final..."); 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 path in fragment_paths: f.write(f"file '{os.path.abspath(path)}'\n")
-    command = f"ffmpeg -y -f concat -safe 0 -i {list_file_path} -c copy {final_output_path}"
-    try:
-        subprocess.run(command, shell=True, check=True, capture_output=True, text=True); return final_output_path
-    except subprocess.CalledProcessError as e:
-        raise gr.Error(f"FFmpeg falhou ao unir os vídeos: {e.stderr}")
-
-def run_full_production(storyboard, ref_img_path, height, width, fps, seed, cfg):
-    if not storyboard: raise gr.Error("Nenhum roteiro para produzir.")
-    if not ref_img_path: raise gr.Error("Nenhuma imagem de referência definida.")
-    video_fragments, log_history = [], ""
-    for i, motion_prompt in enumerate(storyboard):
-        log_message = f"Iniciando produção da Cena {i+1}/{len(storyboard)}..."
-        log_history += log_message + "\n"
-        yield {production_log_output: gr.update(value=log_history)}
-        fragment_path = run_ltx_animation(i + 1, motion_prompt, ref_img_path, height, width, fps, seed, cfg, gr.Progress())
-        video_fragments.append(fragment_path)
-        log_message = f"Cena {i+1} concluída e salva em {os.path.basename(fragment_path)}."
-        log_history += log_message + "\n"
-        yield {production_log_output: gr.update(value=log_history), fragment_gallery_output: gr.update(value=video_fragments), fragment_list_state: video_fragments, final_fragments_display: gr.update(value=video_fragments)}
-    log_history += "\nProdução de todas as cenas concluída!"
-    yield {production_log_output: gr.update(value=log_history)}
-
-# --- Ato 4: A Apresentação (UI do Gradio) ---
-with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("# LTX Video - Storyboard em Vídeo (ADUC-SDR)\n*By Carlex & Gemini*")
-    
-    storyboard_state = gr.State([])
-    reference_image_state = gr.State("")
-    fragment_list_state = gr.State([])
-
-    if os.path.exists(WORKSPACE_DIR): shutil.rmtree(WORKSPACE_DIR)
-    os.makedirs(WORKSPACE_DIR)
-
-    with gr.Tabs():
-        with gr.TabItem("ETAPA 1: O DIRETOR (Roteiro Visual)"):
-            with gr.Row():
-                with gr.Column():
-                    num_fragments_input = gr.Slider(2, 10, 4, step=1, label="Número de Cenas (Fragmentos)")
-                    prompt_input = gr.Textbox(label="Ideia Geral (Prompt)")
-                    image_input = gr.Image(type="filepath", label="Imagem de Referência")
-                    director_button = gr.Button("▶️ Gerar Roteiro Visual (Gemini)", variant="primary")
-                with gr.Column():
-                    storyboard_output = gr.JSON(label="Roteiro Visual Gerado (Storyboard)")
-
-        with gr.TabItem("ETAPA 2: A PRODUÇÃO (Gerar Cenas em Vídeo)"):
-            with gr.Row():
-                with gr.Column():
-                    storyboard_to_render = gr.JSON(label="Roteiro a ser Produzido")
-                    animator_button = gr.Button("▶️ Produzir TODAS as Cenas (LTX)", variant="primary")
-                    production_log_output = gr.Textbox(label="Diário de Bordo da Produção", lines=5, interactive=False, placeholder="Aguardando início da produção...")
-                with gr.Column():
-                    fragment_gallery_output = gr.Gallery(label="Cenas Produzidas (Fragmentos de Vídeo)", object_fit="contain", height="auto")
-            with gr.Row():
-                height_slider = gr.Slider(256, 1024, 512, step=32, label="Altura")
-                width_slider = gr.Slider(256, 1024, 512, step=32, label="Largura")
-            with gr.Row():
-                fps_slider = gr.Slider(8, 24, 15, step=1, label="FPS")
-                seed_number = gr.Number(42, label="Seed")
-                cfg_slider = gr.Slider(1.0, 10.0, 2.5, step=0.1, label="CFG")
-
-        with gr.TabItem("ETAPA 3: PÓS-PRODUÇÃO"):
-            with gr.Row():
-                with gr.Column():
-                    final_fragments_display = gr.JSON(label="Vídeos a Concatenar")
-                    editor_button = gr.Button("▶️ Concatenar Tudo (FFmpeg)", variant="primary")
-                with gr.Column():
-                    final_video_output = gr.Video(label="A Obra-Prima Final")
-
-    # --- Ato 5: A Regência (Lógica de Conexão dos Botões) ---
-
-    def director_success(img_path, storyboard_json):
-        if not img_path: raise gr.Error("A imagem de referência é necessária.")
-        storyboard_list = storyboard_json if isinstance(storyboard_json, list) else storyboard_json.get("storyboard", [])
-        if not storyboard_list: raise gr.Error("O storyboard está vazio.")
-        return storyboard_list, img_path, gr.update(value=storyboard_json)
-
-    director_button.click(
-        fn=get_storyboard_from_director_v2,
-        inputs=[num_fragments_input, prompt_input, image_input],
-        outputs=[storyboard_output]
-    ).success(
-        fn=director_success,
-        inputs=[image_input, storyboard_output],
-        outputs=[storyboard_state, reference_image_state, storyboard_to_render]
-    )
-    
-    animator_button.click(
-        fn=run_full_production,
-        inputs=[storyboard_state, reference_image_state, height_slider, width_slider, fps_slider, seed_number, cfg_slider],
-        outputs=[production_log_output, fragment_gallery_output, fragment_list_state, final_fragments_display]
-    )
-
-    editor_button.click(
-        fn=concatenate_masterpiece,
-        inputs=[fragment_list_state],
-        outputs=[final_video_output]
-    )
-
-if __name__ == "__main__":
-    demo.queue().launch(server_name="0.0.0.0", share=True)