Delete api/ltx_server (7).py

api/ltx_server (7).py

@@ -1,832 +0,0 @@
-# ltx_server.py — VideoService (beta 1.1)
-# Sempre output_type="latent"; no final: VAE (bloco inteiro) → pixels → MP4.
-# Ignora UserWarning/FutureWarning e injeta VAE no manager com dtype/device corretos.
-
-# --- 0. WARNINGS E AMBIENTE ---
-import warnings
-warnings.filterwarnings("ignore", category=UserWarning)
-warnings.filterwarnings("ignore", category=FutureWarning)
-warnings.filterwarnings("ignore", message=".*")
-
-from huggingface_hub import logging
-
-logging.set_verbosity_error()
-logging.set_verbosity_warning()
-logging.set_verbosity_info()
-logging.set_verbosity_debug()
-
-
-LTXV_DEBUG=1
-LTXV_FRAME_LOG_EVERY=8
-
-
-
-# --- 1. IMPORTAÇÕES ---
-import os, subprocess, shlex, tempfile
-import torch
-import json
-import numpy as np
-import random
-import os
-import shlex
-import yaml
-from typing import List, Dict
-from pathlib import Path
-import imageio
-import tempfile
-from huggingface_hub import hf_hub_download
-import sys
-import subprocess
-import gc
-import shutil
-import contextlib
-import time
-import traceback
-
-# Singletons (versões simples)
-from managers.vae_manager import vae_manager_singleton
-from tools.video_encode_tool import video_encode_tool_singleton
-
-# --- 2. GERENCIAMENTO DE DEPENDÊNCIAS E SETUP ---
-def _query_gpu_processes_via_nvml(device_index: int) -> List[Dict]:
-    try:
-        import psutil
-        import pynvml as nvml
-        nvml.nvmlInit()
-        handle = nvml.nvmlDeviceGetHandleByIndex(device_index)
-        try:
-            procs = nvml.nvmlDeviceGetComputeRunningProcesses_v3(handle)
-        except Exception:
-            procs = nvml.nvmlDeviceGetComputeRunningProcesses(handle)
-        results = []
-        for p in procs:
-            pid = int(p.pid)
-            used_mb = None
-            try:
-                if getattr(p, "usedGpuMemory", None) is not None and p.usedGpuMemory not in (0,):
-                    used_mb = max(0, int(p.usedGpuMemory) // (1024 * 1024))
-            except Exception:
-                used_mb = None
-            name = "unknown"
-            user = "unknown"
-            try:
-                import psutil
-                pr = psutil.Process(pid)
-                name = pr.name()
-                user = pr.username()
-            except Exception:
-                pass
-            results.append({"pid": pid, "name": name, "user": user, "used_mb": used_mb})
-        nvml.nvmlShutdown()
-        return results
-    except Exception:
-        return []
-
-def _query_gpu_processes_via_nvidiasmi(device_index: int) -> List[Dict]:
-    cmd = f"nvidia-smi -i {device_index} --query-compute-apps=pid,process_name,used_memory --format=csv,noheader,nounits"
-    try:
-        out = subprocess.check_output(shlex.split(cmd), stderr=subprocess.STDOUT, text=True, timeout=2.0)
-    except Exception:
-        return []
-    results = []
-    for line in out.strip().splitlines():
-        parts = [p.strip() for p in line.split(",")]
-        if len(parts) >= 3:
-            try:
-                pid = int(parts[0]); name = parts[1]; used_mb = int(parts[2])
-                user = "unknown"
-                try:
-                    import psutil
-                    pr = psutil.Process(pid)
-                    user = pr.username()
-                except Exception:
-                    pass
-                results.append({"pid": pid, "name": name, "user": user, "used_mb": used_mb})
-            except Exception:
-                continue
-    return results
-
-def _gpu_process_table(processes: List[Dict], current_pid: int) -> str:
-    if not processes:
-        return "  - Processos ativos: (nenhum)\n"
-    processes = sorted(processes, key=lambda x: (x.get("used_mb") or 0), reverse=True)
-    lines = ["  - Processos ativos (PID | USER | NAME | VRAM MB):"]
-    for p in processes:
-        star = "*" if p["pid"] == current_pid else " "
-        used_str = str(p["used_mb"]) if p.get("used_mb") is not None else "N/A"
-        lines.append(f"    {star} {p['pid']} | {p['user']} | {p['name']} | {used_str}")
-    return "\n".join(lines) + "\n"
-
-def run_setup():
-    setup_script_path = "setup.py"
-    if not os.path.exists(setup_script_path):
-        print("[DEBUG] 'setup.py' não encontrado. Pulando clonagem de dependências.")
-        return
-    try:
-        print("[DEBUG] Executando setup.py para dependências...")
-        subprocess.run([sys.executable, setup_script_path], check=True)
-        print("[DEBUG] Setup concluído com sucesso.")
-    except subprocess.CalledProcessError as e:
-        print(f"[DEBUG] ERRO no setup.py (code {e.returncode}). Abortando.")
-        sys.exit(1)
-
-from api.ltx.inference import (
-    create_ltx_video_pipeline,
-    create_latent_upsampler,
-    load_image_to_tensor_with_resize_and_crop,
-    seed_everething,
-    calculate_padding,
-    load_media_file,
-)
-
-DEPS_DIR = Path("/data")
-LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video"
-if not LTX_VIDEO_REPO_DIR.exists():
-    print(f"[DEBUG] Repositório não encontrado em {LTX_VIDEO_REPO_DIR}. Rodando setup...")
-    run_setup()
-
-def add_deps_to_path():
-    repo_path = str(LTX_VIDEO_REPO_DIR.resolve())
-    if str(LTX_VIDEO_REPO_DIR.resolve()) not in sys.path:
-        sys.path.insert(0, repo_path)
-        print(f"[DEBUG] Repo adicionado ao sys.path: {repo_path}")
-
-add_deps_to_path()
-
-# --- 3. IMPORTAÇÕES ESPECÍFICAS DO MODELO ---
-
-from ltx_video.pipelines.pipeline_ltx_video import ConditioningItem, LTXMultiScalePipeline
-from ltx_video.utils.skip_layer_strategy import SkipLayerStrategy
-
-# --- 4. FUNÇÕES HELPER DE LOG ---
-def log_tensor_info(tensor, name="Tensor"):
-    if not isinstance(tensor, torch.Tensor):
-        print(f"\n[INFO] '{name}' não é tensor.")
-        return
-    print(f"\n--- Tensor: {name} ---")
-    print(f"  - Shape: {tuple(tensor.shape)}")
-    print(f"  - Dtype: {tensor.dtype}")
-    print(f"  - Device: {tensor.device}")
-    if tensor.numel() > 0:
-        try:
-            print(f"  - Min: {tensor.min().item():.4f}  Max: {tensor.max().item():.4f}  Mean: {tensor.mean().item():.4f}")
-        except Exception:
-            pass
-    print("------------------------------------------\n")
-
-# --- 5. CLASSE PRINCIPAL DO SERVIÇO ---
-class VideoService:
-    def __init__(self):
-        t0 = time.perf_counter()
-        print("[DEBUG] Inicializando VideoService...")
-        self.debug = os.getenv("LTXV_DEBUG", "1") == "1"
-        self.frame_log_every = int(os.getenv("LTXV_FRAME_LOG_EVERY", "8"))
-        self.config = self._load_config()
-        print(f"[DEBUG] Config carregada (precision={self.config.get('precision')}, sampler={self.config.get('sampler')})")
-        self.device = "cuda" if torch.cuda.is_available() else "cpu"
-        print(f"[DEBUG] Device selecionado: {self.device}")
-        self.last_memory_reserved_mb = 0.0
-        self._tmp_dirs = set(); self._tmp_files = set(); self._last_outputs = []
-
-        self.pipeline, self.latent_upsampler = self._load_models()
-        print(f"[DEBUG] Pipeline e Upsampler carregados. Upsampler ativo? {bool(self.latent_upsampler)}")
-
-        print(f"[DEBUG] Movendo modelos para {self.device}...")
-        self.pipeline.to(self.device)
-        if self.latent_upsampler:
-            self.latent_upsampler.to(self.device)
-
-        self._apply_precision_policy()
-        print(f"[DEBUG] runtime_autocast_dtype = {getattr(self, 'runtime_autocast_dtype', None)}")
-
-        # Injeta pipeline/vae no manager (impede vae=None)
-        vae_manager_singleton.attach_pipeline(
-            self.pipeline,
-            device=self.device,
-            autocast_dtype=self.runtime_autocast_dtype
-        )
-        print(f"[DEBUG] VAE manager conectado: has_vae={hasattr(self.pipeline, 'vae')} device={self.device}")
-
-        if self.device == "cuda":
-            torch.cuda.empty_cache()
-            self._log_gpu_memory("Após carregar modelos")
-
-        print(f"[DEBUG] VideoService pronto. boot_time={time.perf_counter()-t0:.3f}s")
-
-    def _log_gpu_memory(self, stage_name: str):
-        if self.device != "cuda":
-            return
-        device_index = torch.cuda.current_device() if torch.cuda.is_available() else 0
-        current_reserved_b = torch.cuda.memory_reserved(device_index)
-        current_reserved_mb = current_reserved_b / (1024 ** 2)
-        total_memory_b = torch.cuda.get_device_properties(device_index).total_memory
-        total_memory_mb = total_memory_b / (1024 ** 2)
-        peak_reserved_mb = torch.cuda.max_memory_reserved(device_index) / (1024 ** 2)
-        delta_mb = current_reserved_mb - getattr(self, "last_memory_reserved_mb", 0.0)
-        processes = _query_gpu_processes_via_nvml(device_index) or _query_gpu_processes_via_nvidiasmi(device_index)
-        print(f"\n--- [LOG GPU] {stage_name} (cuda:{device_index}) ---")
-        print(f"  - Reservado: {current_reserved_mb:.2f} MB / {total_memory_mb:.2f} MB  (Δ={delta_mb:+.2f} MB)")
-        if peak_reserved_mb > getattr(self, "last_memory_reserved_mb", 0.0):
-            print(f"  - Pico reservado (nesta fase): {peak_reserved_mb:.2f} MB")
-        print(_gpu_process_table(processes, os.getpid()), end="")
-        print("--------------------------------------------------\n")
-        self.last_memory_reserved_mb = current_reserved_mb
-
-    def _register_tmp_dir(self, d: str):
-        if d and os.path.isdir(d):
-            self._tmp_dirs.add(d); print(f"[DEBUG] Registrado tmp dir: {d}")
-
-    def _register_tmp_file(self, f: str):
-        if f and os.path.exists(f):
-            self._tmp_files.add(f); print(f"[DEBUG] Registrado tmp file: {f}")
-
-    def finalize(self, keep_paths=None, extra_paths=None, clear_gpu=True):
-        print("[DEBUG] Finalize: iniciando limpeza...")
-        keep = set(keep_paths or []); extras = set(extra_paths or [])
-        removed_files = 0
-        for f in list(self._tmp_files | extras):
-            try:
-                if f not in keep and os.path.isfile(f):
-                    os.remove(f); removed_files += 1; print(f"[DEBUG] Removido arquivo tmp: {f}")
-            except Exception as e:
-                print(f"[DEBUG] Falha removendo arquivo {f}: {e}")
-            finally:
-                self._tmp_files.discard(f)
-        removed_dirs = 0
-        for d in list(self._tmp_dirs):
-            try:
-                if d not in keep and os.path.isdir(d):
-                    shutil.rmtree(d, ignore_errors=True); removed_dirs += 1; print(f"[DEBUG] Removido diretório tmp: {d}")
-            except Exception as e:
-                print(f"[DEBUG] Falha removendo diretório {d}: {e}")
-            finally:
-                self._tmp_dirs.discard(d)
-        print(f"[DEBUG] Finalize: arquivos removidos={removed_files}, dirs removidos={removed_dirs}")
-        gc.collect()
-        try:
-            if clear_gpu and torch.cuda.is_available():
-                torch.cuda.empty_cache()
-                try:
-                    torch.cuda.ipc_collect()
-                except Exception:
-                    pass
-        except Exception as e:
-            print(f"[DEBUG] Finalize: limpeza GPU falhou: {e}")
-        try:
-            self._log_gpu_memory("Após finalize")
-        except Exception as e:
-            print(f"[DEBUG] Log GPU pós-finalize falhou: {e}")
-
-    def _load_config(self):
-        base = LTX_VIDEO_REPO_DIR / "configs"
-        candidates = [
-            base / "ltxv-13b-0.9.8-dev-fp8.yaml",
-            base / "ltxv-13b-0.9.8-distilled-fp8.yaml",
-            base / "ltxv-13b-0.9.8-distilled.yaml",
-        ]
-        for cfg in candidates:
-            if cfg.exists():
-                print(f"[DEBUG] Config selecionada: {cfg}")
-                with open(cfg, "r") as file:
-                    return yaml.safe_load(file)
-        cfg = base / "ltxv-13b-0.9.8-distilled-fp8.yaml"
-        print(f"[DEBUG] Config fallback: {cfg}")
-        with open(cfg, "r") as file:
-            return yaml.safe_load(file)
-
-    def _load_models(self):
-        t0 = time.perf_counter()
-        LTX_REPO = "Lightricks/LTX-Video"
-        print("[DEBUG] Baixando checkpoint principal...")
-        distilled_model_path = hf_hub_download(
-            repo_id=LTX_REPO,
-            filename=self.config["checkpoint_path"],
-            local_dir=os.getenv("HF_HOME"),
-            cache_dir=os.getenv("HF_HOME_CACHE"),
-            token=os.getenv("HF_TOKEN"),
-        )
-        self.config["checkpoint_path"] = distilled_model_path
-        print(f"[DEBUG] Checkpoint em: {distilled_model_path}")
-
-        print("[DEBUG] Baixando upscaler espacial...")
-        spatial_upscaler_path = hf_hub_download(
-            repo_id=LTX_REPO,
-            filename=self.config["spatial_upscaler_model_path"],
-            local_dir=os.getenv("HF_HOME"),
-            cache_dir=os.getenv("HF_HOME_CACHE"),
-            token=os.getenv("HF_TOKEN")
-        )
-        self.config["spatial_upscaler_model_path"] = spatial_upscaler_path
-        print(f"[DEBUG] Upscaler em: {spatial_upscaler_path}")
-
-        print("[DEBUG] Construindo pipeline...")
-        pipeline = create_ltx_video_pipeline(
-            ckpt_path=self.config["checkpoint_path"],
-            precision=self.config["precision"],
-            text_encoder_model_name_or_path=self.config["text_encoder_model_name_or_path"],
-            sampler=self.config["sampler"],
-            device="cpu",
-            enhance_prompt=False,
-            prompt_enhancer_image_caption_model_name_or_path=self.config["prompt_enhancer_image_caption_model_name_or_path"],
-            prompt_enhancer_llm_model_name_or_path=self.config["prompt_enhancer_llm_model_name_or_path"],
-        )
-        print("[DEBUG] Pipeline pronto.")
-
-        latent_upsampler = None
-        if self.config.get("spatial_upscaler_model_path"):
-            print("[DEBUG] Construindo latent_upsampler...")
-            latent_upsampler = create_latent_upsampler(self.config["spatial_upscaler_model_path"], device="cpu")
-            print("[DEBUG] Upsampler pronto.")
-        print(f"[DEBUG] _load_models() tempo total={time.perf_counter()-t0:.3f}s")
-        return pipeline, latent_upsampler
-
-    def _promote_fp8_weights_to_bf16(self, module):
-        if not isinstance(module, torch.nn.Module):
-            print("[DEBUG] Promoção FP8→BF16 ignorada: alvo não é nn.Module.")
-            return
-        f8 = getattr(torch, "float8_e4m3fn", None)
-        if f8 is None:
-            print("[DEBUG] torch.float8_e4m3fn indisponível.")
-            return
-        p_cnt = b_cnt = 0
-        for _, p in module.named_parameters(recurse=True):
-            try:
-                if p.dtype == f8:
-                    with torch.no_grad():
-                        p.data = p.data.to(torch.bfloat16); p_cnt += 1
-            except Exception:
-                pass
-        for _, b in module.named_buffers(recurse=True):
-            try:
-                if hasattr(b, "dtype") and b.dtype == f8:
-                    b.data = b.data.to(torch.bfloat16); b_cnt += 1
-            except Exception:
-                pass
-        print(f"[DEBUG] FP8→BF16: params_promoted={p_cnt}, buffers_promoted={b_cnt}")
-
-    def _apply_precision_policy(self):
-        prec = str(self.config.get("precision", "")).lower()
-        self.runtime_autocast_dtype = torch.float32
-        print(f"[DEBUG] Aplicando política de precisão: {prec}")
-        if prec == "float8_e4m3fn":
-            self.runtime_autocast_dtype = torch.bfloat16
-            force_promote = os.getenv("LTXV_FORCE_BF16_ON_FP8", "0") == "1"
-            print(f"[DEBUG] FP8 detectado. force_promote={force_promote}")
-            if force_promote and hasattr(torch, "float8_e4m3fn"):
-                try:
-                    self._promote_fp8_weights_to_bf16(self.pipeline)
-                except Exception as e:
-                    print(f"[DEBUG] Promoção FP8→BF16 na pipeline falhou: {e}")
-                try:
-                    if self.latent_upsampler:
-                        self._promote_fp8_weights_to_bf16(self.latent_upsampler)
-                except Exception as e:
-                    print(f"[DEBUG] Promoção FP8→BF16 no upsampler falhou: {e}")
-        elif prec == "bfloat16":
-            self.runtime_autocast_dtype = torch.bfloat16
-        elif prec == "mixed_precision":
-            self.runtime_autocast_dtype = torch.float16
-        else:
-            self.runtime_autocast_dtype = torch.float32
-
-    def _prepare_conditioning_tensor(self, filepath, height, width, padding_values):
-        print(f"[DEBUG] Carregando condicionamento: {filepath}")
-        tensor = load_image_to_tensor_with_resize_and_crop(filepath, height, width)
-        tensor = torch.nn.functional.pad(tensor, padding_values)
-        out = tensor.to(self.device, dtype=self.runtime_autocast_dtype) if self.device == "cuda" else tensor.to(self.device)
-        print(f"[DEBUG] Cond shape={tuple(out.shape)} dtype={out.dtype} device={out.device}")
-        return out
-
-
-    def _dividir_latentes(self, latents, num_por_chunk: int, overlap: int = 1):
-        # latents: [B, C, T, H, W]
-        T = latents.shape[2]
-        L = num_por_chunk + overlap          # janela com stop exclusivo
-        stride = num_por_chunk - overlap     # avanço
-        print("================= DEBUG =================")  # debug
-        print(f"[DEBUG] T={T}  L={L}  stride={stride}")  # debug
-        print(f"[DEBUG] regra: slice [start:stop) com stop exclusivo")  # debug
-    
-        if T <= L:
-            start = max(0, T - L)
-            stop = start + L
-            chunk = latents[:, :, start:stop, :, :].clone().detach()
-            print(f"[DEBUG] único chunk: [:, :, {start}:{stop}, :, :]  len={stop-start}")  # debug
-            print("=========================================")  # debug
-            return [chunk]
-    
-        starts = []
-        s = 0
-        while s + L <= T:
-            starts.append(s)
-            s += stride
-        if starts[-1] + L < T:
-            starts.append(T - L)
-    
-        # remove duplicata contígua
-        uniq = []
-        for st in starts:
-            if not uniq or uniq[-1] != st:
-                uniq.append(st)
-    
-        chunks = []
-        for i, st in enumerate(uniq):
-            sp = st + L
-            chunk = latents[:, :, st:sp, :, :].clone().detach()
-            print(f"[DEBUG] chunk{i}: [:, :, {st}:{sp}, :, :]  len={sp-st}")  # debug
-            chunks.append(chunk)
-    
-        print("=========================================")  # debug
-        return chunks
-
-    def _get_total_frames(self, video_path: str) -> int:
-        cmd = [
-            "ffprobe",
-            "-v", "error",
-            "-select_streams", "v:0",
-            "-count_frames",
-            "-show_entries", "stream=nb_read_frames",
-            "-of", "default=nokey=1:noprint_wrappers=1",
-            video_path
-        ]
-        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
-        return int(result.stdout.strip())    
-
-    
-    
-    def _gerar_lista_com_transicoes(self, pasta: str, video_paths: list[str], crossfade_frames: int = 8) -> list[str]:
-        """
-        Gera uma nova lista de vídeos aplicando transições suaves (blend frame a frame)
-        seguindo exatamente a lógica linear de Carlos.
-        """
-        import os, subprocess, shutil
-    
-        poda = crossfade_frames
-        total_partes = len(video_paths)
-        video_fade_fim = None
-        video_fade_ini = None
-        nova_lista = []
-
-        print("===========CONCATECAO CAUSAL=============")
-     
-        print(f"[DEBUG] Iniciando pipeline com {total_partes} vídeos e {poda} frames de crossfade")
-    
-        for i in range(total_partes):
-            base = video_paths[i]
-        
-            # --- PODA ---
-            video_podado = os.path.join(pasta, f"{base}_podado_{i}.mp4")
-            
-
-            if i<total_partes-1:
-                end_frame = self._get_total_frames(base) - poda
-            else:
-                end_frame = self._get_total_frames(base)
-            
-            if i>0:
-                start_frame = poda
-            else:
-                start_frame = 0
-                
-            cmd_fim = (
-               f'ffmpeg -y -hide_banner -loglevel error -i "{base}" '
-               f'-vf "trim=start_frame={start_frame}:end_frame={end_frame},setpts=PTS-STARTPTS" '
-               f'-an "{video_podado}"'
-            )
-            subprocess.run(cmd_fim, shell=True, check=True)
-            
-        
-            # --- FADE_INI --- 
-            if i > 0:
-                video_fade_ini = os.path.join(pasta, f"{base}_fade_ini_{i}.mp4")
-                cmd_ini = (
-                    f'ffmpeg -y -hide_banner -loglevel error -i "{base}" '
-                    f'-vf "trim=end_frame={poda},setpts=PTS-STARTPTS" -an "{video_fade_ini}"'
-                )
-                subprocess.run(cmd_ini, shell=True, check=True)
-                
-            # --- TRANSIÇÃO ---
-            if video_fade_fim and video_fade_ini:
-                video_fade  = os.path.join(pasta, f"transicao_{i}_{i+1}.mp4")
-                cmd_blend = (
-                    f'ffmpeg -y -hide_banner -loglevel error '
-                    f'-i "{video_fade_fim}" -i "{video_fade_ini}" '
-                    f'-filter_complex "[0:v][1:v]blend=all_expr=\'A*(1-T/{poda})+B*(T/{poda})\',format=yuv420p" '
-                    f'-frames:v {poda} "{video_fade}"'
-                )
-                subprocess.run(cmd_blend, shell=True, check=True)
-                print(f"[DEBUG] transicao adicionada {i}/{i+1} {self._get_total_frames(video_fade)} frames ✅")
-                nova_lista.append(video_fade)
-                
-            # --- FADE_FIM ---
-            if i<=total_partes-1:
-                video_fade_fim = os.path.join(pasta, f"{base}_fade_fim_{i}.mp4")
-                cmd_fim = (
-                    f'ffmpeg -y -hide_banner -loglevel error -i "{base}" '
-                    f'-vf "trim=start_frame={end_frame-poda},setpts=PTS-STARTPTS" -an "{video_fade_fim}"'
-                )
-                subprocess.run(cmd_fim, shell=True, check=True)
-           
-            nova_lista.append(video_podado)
-            print(f"[DEBUG] Video podado {i+1} adicionado {self._get_total_frames(video_podado)} frames ✅")
-               
-
-
-        print("===========CONCATECAO CAUSAL=============")
-        print(f"[DEBUG] {nova_lista}")
-        return nova_lista
-    
-    def _concat_mp4s_no_reencode(self, mp4_list: List[str], out_path: str):
-        """
-        Concatena múltiplos MP4s sem reencode usando o demuxer do ffmpeg.
-        ATENÇÃO: todos os arquivos precisam ter mesmo codec, fps, resolução etc.
-        """
-        if not mp4_list or len(mp4_list) < 2:
-            raise ValueError("Forneça pelo menos dois arquivos MP4 para concatenar.")
-
-        
-        # Cria lista temporária para o ffmpeg
-        with tempfile.NamedTemporaryFile("w", delete=False, suffix=".txt") as f:
-            for mp4 in mp4_list:
-                f.write(f"file '{os.path.abspath(mp4)}'\n")
-            list_path = f.name
-    
-        cmd = f"ffmpeg -y -f concat -safe 0 -i {list_path} -c copy {out_path}"
-        print(f"[DEBUG] Concat: {cmd}")
-    
-        try:
-            subprocess.check_call(shlex.split(cmd))
-        finally:
-            try:
-                os.remove(list_path)
-            except Exception:
-                pass   
-
-    
-    def generate(
-        self,
-        prompt,
-        negative_prompt,
-        mode="text-to-video",
-        start_image_filepath=None,
-        middle_image_filepath=None,
-        middle_frame_number=None,
-        middle_image_weight=1.0,
-        end_image_filepath=None,
-        end_image_weight=1.0,
-        input_video_filepath=None,
-        height=512,
-        width=704,
-        duration=2.0,
-        frames_to_use=9,
-        seed=42,
-        randomize_seed=True,
-        guidance_scale=3.0,
-        improve_texture=True,
-        progress_callback=None,
-        # Sempre latent → VAE → MP4 (simples)
-        external_decode=True,
-    ):
-        t_all = time.perf_counter()
-        print(f"[DEBUG] generate() begin mode={mode} external_decode={external_decode} improve_texture={improve_texture}")
-        if self.device == "cuda":
-            torch.cuda.empty_cache(); torch.cuda.reset_peak_memory_stats()
-        self._log_gpu_memory("Início da Geração")
-
-        if mode == "image-to-video" and not start_image_filepath:
-            raise ValueError("A imagem de início é obrigatória para o modo image-to-video")
-        if mode == "video-to-video" and not input_video_filepath:
-            raise ValueError("O vídeo de entrada é obrigatório para o modo video-to-video")
-
-        used_seed = random.randint(0, 2**32 - 1) if randomize_seed else int(seed)
-        seed_everething(used_seed); print(f"[DEBUG] Seed usado: {used_seed}")
-
-        FPS = 24.0; MAX_NUM_FRAMES = 2570
-        target_frames_rounded = round(duration * FPS)
-        n_val = round((float(target_frames_rounded) - 1.0) / 8.0)
-        actual_num_frames = max(9, min(MAX_NUM_FRAMES, int(n_val * 8 + 1)))
-        print(f"[DEBUG] Frames alvo: {actual_num_frames} (dur={duration}s @ {FPS}fps)")
-
-        height_padded = ((height - 1) // 32 + 1) * 32
-        width_padded = ((width - 1) // 32 + 1) * 32
-        padding_values = calculate_padding(height, width, height_padded, width_padded)
-        print(f"[DEBUG] Dimensões: ({height},{width}) -> pad ({height_padded},{width_padded}); padding={padding_values}")
-
-        generator = torch.Generator(device=self.device).manual_seed(used_seed)
-        conditioning_items = []
-
-        if mode == "image-to-video":
-            start_tensor = self._prepare_conditioning_tensor(start_image_filepath, height, width, padding_values)
-            conditioning_items.append(ConditioningItem(start_tensor, 0, 1.0))
-            if middle_image_filepath and middle_frame_number is not None:
-                middle_tensor = self._prepare_conditioning_tensor(middle_image_filepath, height, width, padding_values)
-                safe_middle_frame = max(0, min(int(middle_frame_number), actual_num_frames - 1))
-                conditioning_items.append(ConditioningItem(middle_tensor, safe_middle_frame, float(middle_image_weight)))
-            if end_image_filepath:
-                end_tensor = self._prepare_conditioning_tensor(end_image_filepath, height, width, padding_values)
-                last_frame_index = actual_num_frames - 1
-                conditioning_items.append(ConditioningItem(end_tensor, last_frame_index, float(end_image_weight)))
-            print(f"[DEBUG] Conditioning items: {len(conditioning_items)}")
-
-        # Sempre pedimos latentes (simples)
-        call_kwargs = {
-            "prompt": prompt,
-            "negative_prompt": negative_prompt,
-            "height": height_padded,
-            "width": width_padded,
-            "num_frames": actual_num_frames,
-            "frame_rate": int(FPS),
-            "generator": generator,
-            "output_type": "latent",
-            "conditioning_items": conditioning_items if conditioning_items else None,
-            "media_items": None,
-            "decode_timestep": self.config["decode_timestep"],
-            "decode_noise_scale": self.config["decode_noise_scale"],
-            "stochastic_sampling": self.config["stochastic_sampling"],
-            "image_cond_noise_scale": 0.01,
-            "is_video": True,
-            "vae_per_channel_normalize": True,
-            "mixed_precision": (self.config["precision"] == "mixed_precision"),
-            "offload_to_cpu": False,
-            "enhance_prompt": False,
-            "skip_layer_strategy": SkipLayerStrategy.AttentionValues,
-        }
-        print(f"[DEBUG] output_type={call_kwargs['output_type']} skip_layer_strategy={call_kwargs['skip_layer_strategy']}")
-
-        if mode == "video-to-video":
-            media = load_media_file(
-                media_path=input_video_filepath,
-                height=height,
-                width=width,
-                max_frames=int(frames_to_use),
-                padding=padding_values,
-            ).to(self.device)
-            call_kwargs["media_items"] = media
-            print(f"[DEBUG] media_items shape={tuple(media.shape)}")
-
-        latents = None
-        multi_scale_pipeline = None
-
-        try:
-            if improve_texture:
-                if not self.latent_upsampler:
-                    raise ValueError("Upscaler espacial não carregado.")
-                print("[DEBUG] Multi-escala: construindo pipeline...")
-                multi_scale_pipeline = LTXMultiScalePipeline(self.pipeline, self.latent_upsampler)
-                first_pass_args = self.config.get("first_pass", {}).copy()
-                first_pass_args["guidance_scale"] = float(guidance_scale)
-                second_pass_args = self.config.get("second_pass", {}).copy()
-                second_pass_args["guidance_scale"] = float(guidance_scale)
-
-                multi_scale_call_kwargs = call_kwargs.copy()
-                multi_scale_call_kwargs.update(
-                    {
-                        "downscale_factor": self.config["downscale_factor"],
-                        "first_pass": first_pass_args,
-                        "second_pass": second_pass_args,
-                    }
-                )
-                print("[DEBUG] Chamando multi_scale_pipeline...")
-                t_ms = time.perf_counter()
-                ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext()
-                with ctx:
-                    result = multi_scale_pipeline(**multi_scale_call_kwargs)
-                print(f"[DEBUG] multi_scale_pipeline tempo={time.perf_counter()-t_ms:.3f}s")
-
-                if hasattr(result, "latents"):
-                    latents = result.latents
-                elif hasattr(result, "images") and isinstance(result.images, torch.Tensor):
-                    latents = result.images
-                else:
-                    latents = result
-                print(f"[DEBUG] Latentes (multi-escala): shape={tuple(latents.shape)}")
-            else:
-                single_pass_kwargs = call_kwargs.copy()
-                first_pass_config = self.config.get("first_pass", {})
-                single_pass_kwargs.update(
-                    {
-                        "guidance_scale": float(guidance_scale),
-                        "stg_scale": first_pass_config.get("stg_scale"),
-                        "rescaling_scale": first_pass_config.get("rescaling_scale"),
-                        "skip_block_list": first_pass_config.get("skip_block_list"),
-                    }
-                )
-                schedule = first_pass_config.get("timesteps") or first_pass_config.get("guidance_timesteps")
-                if mode == "video-to-video":
-                    schedule = [0.7]; print("[INFO] Modo video-to-video (etapa única): timesteps=[0.7]")
-                if isinstance(schedule, (list, tuple)) and len(schedule) > 0:
-                    single_pass_kwargs["timesteps"] = schedule
-                    single_pass_kwargs["guidance_timesteps"] = schedule
-                print(f"[DEBUG] Single-pass: timesteps_len={len(schedule) if schedule else 0}")
-
-                print("\n[INFO] Executando pipeline de etapa única...")
-                t_sp = time.perf_counter()
-                ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext()
-                with ctx:
-                    result = self.pipeline(**single_pass_kwargs)
-                print(f"[DEBUG] single-pass tempo={time.perf_counter()-t_sp:.3f}s")
-
-                if hasattr(result, "latents"):
-                    latents = result.latents
-                elif hasattr(result, "images") and isinstance(result.images, torch.Tensor):
-                    latents = result.images
-                else:
-                    latents = result
-                print(f"[DEBUG] Latentes (single-pass): shape={tuple(latents.shape)}")
-
-            # Staging e escrita MP4 (simples: VAE → pixels → MP4)
-            
-            latents_cpu = latents.detach().to("cpu", non_blocking=True)
-            torch.cuda.empty_cache()
-            try:
-                torch.cuda.ipc_collect()
-            except Exception:
-                pass
-                
-            latents_parts = self._dividir_latentes(latents_cpu,4,1)
-            
-            temp_dir = tempfile.mkdtemp(prefix="ltxv_"); self._register_tmp_dir(temp_dir)
-            results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
-
-            partes_mp4 = []
-            par = 0
-            
-            for latents in latents_parts:
-                print(f"[DEBUG] Partição {par}: {tuple(latents.shape)}")
-       
-                par = par + 1
-                output_video_path = os.path.join(temp_dir, f"output_{used_seed}_{par}.mp4")
-                final_output_path = None
-    
-                print("[DEBUG] Decodificando bloco de latentes com VAE → tensor de pixels...")
-                # Usar manager com timestep por item; previne target_shape e rota NoneType.decode
-                pixel_tensor = vae_manager_singleton.decode(
-                    latents.to(self.device, non_blocking=True),
-                    decode_timestep=float(self.config.get("decode_timestep", 0.05))
-                )
-                log_tensor_info(pixel_tensor, "Pixel tensor (VAE saída)")
-    
-                print("[DEBUG] Codificando MP4 a partir do tensor de pixels (bloco inteiro)...")
-                video_encode_tool_singleton.save_video_from_tensor(
-                    pixel_tensor,
-                    output_video_path,
-                    fps=call_kwargs["frame_rate"],
-                    progress_callback=progress_callback
-                )
-    
-                candidate = os.path.join(results_dir, f"output_par_{par}.mp4")
-                try:
-                    shutil.move(output_video_path, candidate)
-                    final_output_path = candidate
-                    print(f"[DEBUG] MP4 parte {par} movido para {final_output_path}")
-                    partes_mp4.append(final_output_path)
-                    
-                except Exception as e:
-                    final_output_path = output_video_path
-                    print(f"[DEBUG] Falha no move; usando tmp como final: {e}")
-
-            total_partes = len(partes_mp4)
-            if (total_partes>1):
-                final_vid = os.path.join(results_dir, f"concat_fim_{used_seed}.mp4")
-                partes_mp4_fade = self._gerar_lista_com_transicoes(pasta=results_dir, video_paths=partes_mp4, crossfade_frames=8)
-                self._concat_mp4s_no_reencode(partes_mp4_fade, final_vid)
-            else:
-                final_vid = partes_mp4[0]
-            
-            
-            self._log_gpu_memory("Fim da Geração")
-            return final_vid, used_seed
-            
-
-        except Exception as e:
-            print("[DEBUG] EXCEÇÃO NA GERAÇÃO:")
-            print("".join(traceback.format_exception(type(e), e, e.__traceback__)))
-            raise
-        finally:
-            try:
-                del latents
-            except Exception:
-                pass
-            try:
-                del multi_scale_pipeline
-            except Exception:
-                pass
-
-            gc.collect()
-            try:
-                if self.device == "cuda":
-                    torch.cuda.empty_cache()
-                    try:
-                        torch.cuda.ipc_collect()
-                    except Exception:
-                        pass
-            except Exception as e:
-                print(f"[DEBUG] Limpeza GPU no finally falhou: {e}")
-
-            try:
-                self.finalize(keep_paths=[])
-            except Exception as e:
-                print(f"[DEBUG] finalize() no finally falhou: {e}")
-
-print("Criando instância do VideoService. O carregamento do modelo começará agora...")
-video_generation_service = VideoService()