Update app_ltx.py

app_ltx.py

@@ -1,110 +1,106 @@
 import os
-import subprocess
-import sys
+import gradio as gr
 from pathlib import Path
-import time  # <<< CORREÇÃO AQUI: Importa o módulo 'time'
-from typing import Optional, Tuple
 
-from huggingface_hub import snapshot_download
+# Importa o singleton do nosso novo servidor LTX
+try:
+    from services.ltx_server import ltx_server_singleton as server
+except Exception as e:
+    print(f"ERRO FATAL: Não foi possível importar o LTXServer. A aplicação não pode iniciar.")
+    print(f"Detalhe do erro: {e}")
+    raise
 
-class LTXServer:
-    """
-    Gerencia o setup e a execução da inferência para o LTX-Video Q8.
-    """
-    _instance = None
+# --- Função de Callback da UI ---
+def generate_video_from_image(
+    prompt: str,
+    image_input: str,
+    height: int,
+    width: int,
+    num_frames: int,
+    seed: int,
+    progress=gr.Progress(track_tqdm=True) # <<< ADICIONADO PROGRESSO AQUI
+):
+    """Callback para a UI que chama o backend LTXServer."""
+    progress(0.1, desc="Validando entradas...")
+    if not image_input or not Path(image_input).exists():
+        gr.Warning("Por favor, faça o upload de uma imagem de entrada.")
+        return None
+    if not prompt or not prompt.strip():
+        gr.Warning("Por favor, insira um prompt.")
+        return None
 
-    def __new__(cls, *args, **kwargs):
-        if not cls._instance:
-            cls._instance = super(LTXServer, cls).__new__(cls)
-        return cls._instance
-
-    def __init__(self):
-        if hasattr(self, '_initialized') and self._initialized:
-            return
-        
-        print("🚀 LTXServer (Q8) inicializando e preparando o ambiente...")
+    try:
+        progress(0.5, desc="Enviando tarefa para o backend LTX (Q8). A inferência pode demorar um pouco...")
         
-        self.LTX_REPO_DIR = Path(os.getenv("LTX_REPO_DIR", "/data/LTX-Video"))
-        self.LTX_CKPT_DIR = Path(os.getenv("LTX_CKPT_DIR", "/data/ckpt/ltxvideo_q8"))
-        self.OUTPUT_ROOT = Path(os.getenv("OUTPUT_ROOT", "/app/outputs/ltx"))
-        self.HF_HOME_CACHE = Path(os.getenv("HF_HOME", "/data/.cache/huggingface"))
+        video_path = server.run_inference(
+            prompt=prompt,
+            image_path=image_input,
+            height=int(height),
+            width=int(width),
+            num_frames=int(num_frames),
+            seed=int(seed)
+        )
         
-        self.REPO_URL = "https://github.com/KONAKONA666/LTX-Video.git"
-        self.MODEL_REPO_ID = "konakona/ltxvideo_q8"
+        progress(1.0, desc="Inferência concluída!")
+        return video_path
         
-        for p in [self.LTX_REPO_DIR.parent, self.LTX_CKPT_DIR, self.OUTPUT_ROOT, self.HF_HOME_CACHE]:
-            p.mkdir(parents=True, exist_ok=True)
-            
-        self.setup_dependencies()
-        self._initialized = True
-        print("✅ LTXServer (Q8) pronto.")
-
-    def setup_dependencies(self):
-        self._ensure_repo()
-        self._ensure_model()
-
-    def _ensure_repo(self) -> None:
-        if not (self.LTX_REPO_DIR / ".git").exists():
-            print(f"[LTXServer] Clonando repositório para {self.LTX_REPO_DIR}...")
-            subprocess.run(["git", "clone", "--depth", "1", self.REPO_URL, str(self.LTX_REPO_DIR)], check=True)
-        else:
-            print("[LTXServer] Repositório LTX-Video já existe.")
+    except Exception as e:
+        print(f"[UI LTX ERROR] A inferência falhou: {e}")
+        gr.Error(f"Erro na Geração: {e}")
+        return None
 
-    def _ensure_model(self) -> None:
-        print(f"[LTXServer] Verificando checkpoints em {self.LTX_CKPT_DIR}...")
-        snapshot_download(
-            repo_id=self.MODEL_REPO_ID,
-            local_dir=str(self.LTX_CKPT_DIR),
-            cache_dir=str(self.HF_HOME_CACHE),
-            repo_type='model',
-            token=os.getenv("HF_TOKEN")
-        )
-        print("[LTXServer] Checkpoints Q8 prontos.")
+# --- Definição da Interface Gráfica com Gradio ---
+with gr.Blocks(title="LTX-Video (Q8 Img2Vid)") as demo:
+    gr.HTML(
+        """
+        <div style='text-align:center; margin-bottom: 20px;'>
+            <h1>LTX-Video Q8 - Imagem para Vídeo</h1>
+            <p>Interface de teste isolada para o modelo LTX-Video quantizado.</p>
+        </div>
+        """
+    )
+    
+    with gr.Row():
+        with gr.Column(scale=1):
+            image_in = gr.Image(type="filepath", label="Imagem de Entrada")
+            prompt_in = gr.Textbox(label="Prompt", lines=3, placeholder="Ex: a cinematic shot of a woman smiling")
+            
+            with gr.Accordion("Parâmetros de Geração", open=True):
+                with gr.Row():
+                    height_in = gr.Slider(label="Altura (Height)", minimum=256, maximum=1024, step=64, value=512)
+                    width_in = gr.Slider(label="Largura (Width)", minimum=256, maximum=1024, step=64, value=512)
+                with gr.Row():
+                    frames_in = gr.Slider(label="Número de Frames", minimum=16, maximum=128, step=8, value=32)
+                    seed_in = gr.Number(label="Seed", value=42, precision=0)
 
-    def run_inference(self, prompt: str, image_path: str, height: int, width: int, num_frames: int, seed: int) -> str:
-        if not Path(image_path).exists():
-            raise FileNotFoundError(f"Arquivo de imagem de entrada não encontrado: {image_path}")
+            run_button = gr.Button("Gerar Vídeo", variant="primary")
 
-        script_path = self.LTX_REPO_DIR / "inference.py"
-        if not script_path.exists():
-            raise FileNotFoundError(f"Script de inferência não encontrado em: {script_path}")
+        with gr.Column(scale=1):
+            video_out = gr.Video(label="Vídeo Gerado")
 
-        job_output_dir = self.OUTPUT_ROOT / f"run_{int(time.time())}_{os.urandom(4).hex()}"
-        job_output_dir.mkdir(parents=True)
+    run_button.click(
+        fn=generate_video_from_image,
+        inputs=[prompt_in, image_in, height_in, width_in, frames_in, seed_in],
+        outputs=[video_out],
+    )
 
-        cmd = [
-            "python", str(script_path),
-            "--ckpt_dir", str(self.LTX_CKPT_DIR),
-            "--output_dir", str(job_output_dir),
-            "--low_vram",
-            "--transformer_type=q8_kernels",
-            "--prompt", prompt,
-            "--input_image_path", image_path,
-            "--height", str(height),
-            "--width", str(width),
-            "--num_frames", str(num_frames),
-            "--seed", str(seed),
-        ]
+    gr.Markdown("---")
+    gr.Examples(
+        examples=[["A beautiful woman with a gentle smile, cinematic lighting", "frame_1.png", 512, 512, 32, 123]],
+        inputs=[prompt_in, image_in, height_in, width_in, frames_in, seed_in],
+    )
 
-        print("[LTXServer] Comando de execução:", " ".join(cmd))
-        
+if __name__ == "__main__":
+    if not os.path.exists("frame_1.png"):
         try:
-            subprocess.run(
-                cmd, 
-                cwd=str(self.LTX_REPO_DIR),
-                check=True, 
-                env=os.environ.copy(),
-                stdout=sys.stdout,
-                stderr=sys.stderr
-            )
-        except Exception as e:
-            print(f"[LTXServer] Erro na execução da inferência: {e}")
-            raise
-            
-        output_videos = sorted(job_output_dir.glob("*.mp4"))
-        if not output_videos:
-            raise FileNotFoundError(f"Nenhum vídeo foi gerado no diretório de saída: {job_output_dir}")
-            
-        return str(output_videos[0])
+            from PIL import Image
+            img = Image.new('RGB', (512, 512), color = 'grey')
+            img.save('frame_1.png')
+        except:
+            pass
 
-ltx_server_singleton = LTXServer()
+    demo.launch(
+        server_name=os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"),
+        server_port=int(os.getenv("GRADIO_SERVER_PORT", "7861")),
+        show_error=True,
+    )