# 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 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 from PIL import Image # Import adicionado para handle_media_upload_for_dims import tempfile from huggingface_hub import hf_hub_download import sys import subprocess import gc import shutil import contextlib import time import traceback from einops import rearrange import torch.nn.functional as F from managers.vae_manager import vae_manager_singleton from tools.video_encode_tool import video_encode_tool_singleton DEPS_DIR = Path("/data") LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video" # CORREÇÃO: Movido run_setup para o início para garantir que seja definido antes de ser chamado. 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) 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}") 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 calculate_new_dimensions(orig_w, orig_h, divisor=8): if orig_w == 0 or orig_h == 0: return 512, 512 if orig_w >= orig_h: aspect_ratio = orig_w / orig_h new_h = 512 new_w = new_h * aspect_ratio else: aspect_ratio = orig_h / orig_w new_w = 512 new_h = new_w * aspect_ratio final_w = int(round(new_w / divisor)) * divisor final_h = int(round(new_h / divisor)) * divisor final_w = max(divisor, final_w) final_h = max(divisor, final_h) print(f"[Dimension Calc] Original: {orig_w}x{orig_h} -> Calculado: {new_w:.0f}x{new_h:.0f} -> Final (divisível por {divisor}): {final_w}x{final_h}") return final_h, final_w def handle_media_upload_for_dims(filepath, current_h, current_w): # CORREÇÃO: Gradio (`gr`) não deve ser usado no backend. Retornando tupla diretamente. if not filepath or not os.path.exists(str(filepath)): return current_h, current_w try: if str(filepath).lower().endswith(('.png', '.jpg', '.jpeg', '.webp')): with Image.open(filepath) as img: orig_w, orig_h = img.size else: with imageio.get_reader(filepath) as reader: meta = reader.get_meta_data() orig_w, orig_h = meta.get('size', (current_w, current_h)) new_h, new_w = calculate_new_dimensions(orig_w, orig_h) return new_h, new_w except Exception as e: print(f"Erro ao processar mídia para dimensões: {e}") return current_h, current_w 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 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") add_deps_to_path() from ltx_video.pipelines.pipeline_ltx_video import ConditioningItem, LTXMultiScalePipeline from ltx_video.utils.skip_layer_strategy import SkipLayerStrategy from ltx_video.models.autoencoders.vae_encode import un_normalize_latents, normalize_latents from ltx_video.pipelines.pipeline_ltx_video import adain_filter_latent 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, ) 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)}") 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}") @torch.no_grad() def _upsample_latents_internal(self, latents: torch.Tensor) -> torch.Tensor: if not self.latent_upsampler: raise ValueError("Latent Upsampler não está carregado.") self.latent_upsampler.to(self.device) self.pipeline.vae.to(self.device) print(f"[DEBUG-UPSAMPLE] Shape de entrada: {tuple(latents.shape)}") latents = un_normalize_latents(latents, self.pipeline.vae, vae_per_channel_normalize=True) upsampled_latents = self.latent_upsampler(latents) upsampled_latents = normalize_latents(upsampled_latents, self.pipeline.vae, vae_per_channel_normalize=True) print(f"[DEBUG-UPSAMPLE] Shape de saída: {tuple(upsampled_latents.shape)}") return upsampled_latents 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_por_tamanho(self, latents_brutos, num_latente_por_chunk: int, overlap: int = 1): sum_latent = latents_brutos.shape[2] chunks = [] if num_latente_por_chunk >= sum_latent: return [latents_brutos.clone().detach()] # CORREÇÃO: Retornar uma lista e clonar # CORREÇÃO: Lógica de chunking simplificada e corrigida para evitar estouro de índice start = 0 while start < sum_latent: end = min(start + num_latente_por_chunk, sum_latent) # Para o overlap, pegamos um pouco do chunk anterior, exceto para o primeiro overlap_start = max(0, start - overlap) # O chunk a ser processado vai de `overlap_start` até `end` # mas o chunk "real" para junção posterior seria de `start` a `end` # A lógica atual já faz um overlap simples, vamos refinar effective_end = min(start + num_latente_por_chunk, sum_latent) chunk = latents_brutos[:, :, start:effective_end, :, :].clone().detach() # Adiciona overlap no final se não for o último chunk if effective_end < sum_latent: overlap_end = min(effective_end + overlap, sum_latent) chunk = latents_brutos[:, :, start:overlap_end, :, :].clone().detach() print(f"[DEBUG] Chunk: start={start}, end={chunk.shape[2]}, total_latents={sum_latent}") chunks.append(chunk) # Avança para o próximo chunk if start + num_latente_por_chunk >= sum_latent: break start += num_latente_por_chunk 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]: # Esta função parece complexa e propensa a erros com nomes de arquivo. # Por segurança, mantendo a lógica original, mas corrigindo possíveis bugs de `shell=True` # e garantindo que os arquivos existam. if len(video_paths) <= 1: return video_paths # Não há o que fazer nova_lista_intermediaria = [] # Primeiro, cria todos os vídeos podados videos_podados = [] for i, base in enumerate(video_paths): video_podado = os.path.join(pasta, f"podado_{i}.mp4") total_frames = self._get_total_frames(base) start_frame = crossfade_frames if i > 0 else 0 end_frame = total_frames - crossfade_frames if i < len(video_paths) - 1 else total_frames # Pular poda se não houver frames suficientes if start_frame >= end_frame: continue cmd = [ 'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', base, '-vf', f'trim=start_frame={start_frame}:end_frame={end_frame},setpts=PTS-STARTPTS', '-an', video_podado ] subprocess.run(cmd, check=True) videos_podados.append(video_podado) # Agora, cria as transições e monta a lista final lista_final = [videos_podados[0]] for i in range(len(video_paths) - 1): video_anterior = video_paths[i] video_seguinte = video_paths[i+1] # Extrai fade_fim do anterior fade_fim_path = os.path.join(pasta, f"fade_fim_{i}.mp4") total_frames_anterior = self._get_total_frames(video_anterior) cmd_fim = [ 'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_anterior, '-vf', f'trim=start_frame={total_frames_anterior - crossfade_frames},setpts=PTS-STARTPTS', '-an', fade_fim_path ] subprocess.run(cmd_fim, check=True) # Extrai fade_ini do seguinte fade_ini_path = os.path.join(pasta, f"fade_ini_{i+1}.mp4") cmd_ini = [ 'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', video_seguinte, '-vf', f'trim=end_frame={crossfade_frames},setpts=PTS-STARTPTS', '-an', fade_ini_path ] subprocess.run(cmd_ini, check=True) # Cria a transição transicao_path = os.path.join(pasta, f"transicao_{i}_{i+1}.mp4") cmd_blend = [ 'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', '-i', fade_fim_path, '-i', fade_ini_path, '-filter_complex', f'[0:v][1:v]blend=all_expr=\'A*(1-T/{crossfade_frames})+B*(T/{crossfade_frames})\',format=yuv420p', '-frames:v', str(crossfade_frames), transicao_path ] subprocess.run(cmd_blend, check=True) lista_final.append(transicao_path) lista_final.append(videos_podados[i+1]) return lista_final def _concat_mp4s_no_reencode(self, mp4_list: List[str], out_path: str): if not mp4_list: raise ValueError("A lista de MP4s para concatenar está vazia.") # Se houver apenas um vídeo, apenas o copie/mova if len(mp4_list) == 1: shutil.move(mp4_list[0], out_path) print(f"[DEBUG] Apenas um vídeo, movido para: {out_path}") return 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, # Parâmetro não utilizado, mas mantido por consistência seed=42, randomize_seed=True, guidance_scale=3.0, improve_texture=True, progress_callback=None, external_decode=True, # Parâmetro não utilizado, mas mantido ): t_all = time.perf_counter() print(f"[DEBUG] generate() begin mode={mode} 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") 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))) height_padded = ((height - 1) // 8 + 1) * 8 width_padded = ((width - 1) // 8 + 1) * 8 padding_values = calculate_padding(height, width, height_padded, width_padded) 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)}") 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, } # CORREÇÃO: Inicialização de listas latents_list = [] temp_dir = tempfile.mkdtemp(prefix="ltxv_"); self._register_tmp_dir(temp_dir) results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True) try: if improve_texture: ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext() with ctx: if not self.latent_upsampler: raise ValueError("Upscaler espacial não carregado, mas 'improve_texture' está ativo.") print("\n--- INICIANDO ETAPA 1: GERAÇÃO BASE (FIRST PASS) ---") t_pass1 = time.perf_counter() first_pass_config = self.config.get("first_pass", {}).copy() first_pass_config.pop("num_inference_steps", None) downscale_factor = self.config.get("downscale_factor", 0.6666666) vae_scale_factor = self.pipeline.vae_scale_factor x_width = int(width_padded * downscale_factor) downscaled_width = x_width - (x_width % vae_scale_factor) x_height = int(height_padded * downscale_factor) downscaled_height = x_height - (x_height % vae_scale_factor) print(f"[DEBUG] First Pass Dims: Original Pad ({width_padded}x{height_padded}) -> Downscaled ({downscaled_width}x{downscaled_height})") first_pass_kwargs = call_kwargs.copy() first_pass_kwargs.update({ "output_type": "latent", "width": downscaled_width, "height": downscaled_height, "guidance_scale": float(guidance_scale), **first_pass_config }) print(f"[DEBUG] First Pass: Gerando em {downscaled_width}x{downscaled_height}...") # CORREÇÃO: Usar self.pipeline, não a variável deletada 'pipeline' latents = self.pipeline(**first_pass_kwargs).images log_tensor_info(latents, "Latentes Base (First Pass)") print(f"[DEBUG] First Pass concluída em {time.perf_counter() - t_pass1:.2f}s") with ctx: print("\n--- INICIANDO ETAPA 2: UPSCALE DOS LATENTES ---") t_upscale = time.perf_counter() upsampled_latents = self._upsample_latents_internal(latents) upsampled_latents = adain_filter_latent(latents=upsampled_latents, reference_latents=latents) print(f"[DEBUG] Upscale de Latentes concluído em {time.perf_counter() - t_upscale:.2f}s") # CORREÇÃO: Manter latentes originais para AdaIN e passar latentes com upscale para o second pass reference_latents_cpu = latents.detach().to("cpu", non_blocking=True) latents_to_refine = upsampled_latents del upsampled_latents; del latents; gc.collect(); torch.cuda.empty_cache() # CORREÇÃO: Lógica de chunking para o second pass latents_parts = self._dividir_latentes_por_tamanho(latents_to_refine, 32, 8) # Exemplo: chunks de 32 frames com 8 de overlap del latents_to_refine with ctx: for i, latents_chunk in enumerate(latents_parts): print(f"\n--- INICIANDO ETAPA 3.{i+1}: REFINAMENTO DE TEXTURA (SECOND PASS) ---") # CORREÇÃO: AdaIN precisa de latents de referência com mesmo H/W, o que não é o caso aqui. # Vamos aplicar AdaIN com o próprio chunk para normalização, ou pular. Pulando por simplicidade. second_pass_config = self.config.get("second_pass", {}).copy() second_pass_config.pop("num_inference_steps", None) # O tamanho do second pass deve ser o tamanho do latente de entrada (após upscale) second_pass_height, second_pass_width = latents_chunk.shape[3] * 8, latents_chunk.shape[4] * 8 print(f"[DEBUG] Second Pass Dims: Target ({second_pass_width}x{second_pass_height})") t_pass2 = time.perf_counter() second_pass_kwargs = call_kwargs.copy() second_pass_kwargs.update({ "output_type": "latent", "width": second_pass_width, "height": second_pass_height, "latents": latents_chunk.to(self.device), # Mover chunk para GPU "guidance_scale": float(guidance_scale), "num_frames": latents_chunk.shape[2], # Usar o número de frames do chunk **second_pass_config }) print(f"[DEBUG] Second Pass: Refinando chunk {i+1}/{len(latents_parts)}...") final_latents = self.pipeline(**second_pass_kwargs).images log_tensor_info(final_latents, "Latentes Finais (Pós-Second Pass)") print(f"[DEBUG] Second part Pass concluída em {time.perf_counter() - t_pass2:.2f}s") latents_cpu = final_latents.detach().to("cpu", non_blocking=True) latents_list.append(latents_cpu) del final_latents; del latents_chunk; gc.collect(); torch.cuda.empty_cache() else: ctx = torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype) if self.device == "cuda" else contextlib.nullcontext() with ctx: print("\n--- INICIANDO GERAÇÃO DE ETAPA ÚNICA ---") t_single = time.perf_counter() single_pass_call_kwargs = call_kwargs.copy() # CORREÇÃO: `pipeline_instance` não existe, usar `self.pipeline`. latents_single_pass = self.pipeline(**single_pass_call_kwargs).images log_tensor_info(latents_single_pass, "Latentes Finais (Etapa Única)") print(f"[DEBUG] Etapa única concluída em {time.perf_counter() - t_single:.2f}s") latents_cpu = latents_single_pass.detach().to("cpu", non_blocking=True) latents_list.append(latents_cpu) # CORREÇÃO: aqui deve ser latents_cpu, não latents_single_pass del latents_single_pass; gc.collect(); torch.cuda.empty_cache() # --- ETAPA FINAL: DECODIFICAÇÃO E CODIFICAÇÃO MP4 --- print("\n--- INICIANDO ETAPA FINAL: DECODIFICAÇÃO E MONTAGEM ---") partes_mp4 = [] for i, latents in enumerate(latents_list): print(f"[DEBUG] Decodificando partição {i+1}/{len(latents_list)}: {tuple(latents.shape)}") output_video_path = os.path.join(temp_dir, f"output_{used_seed}_{i}.mp4") 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)") video_encode_tool_singleton.save_video_from_tensor( pixel_tensor, output_video_path, fps=call_kwargs["frame_rate"], progress_callback=progress_callback ) partes_mp4.append(output_video_path) del pixel_tensor; del latents; gc.collect(); torch.cuda.empty_cache() final_vid = os.path.join(results_dir, f"final_video_{used_seed}.mp4") if len(partes_mp4) > 1: # A função _gerar_lista_com_transicoes é complexa, usando uma concatenação direta como fallback robusto. # Para usar a transição, a lógica de overlap na divisão de latentes precisa ser perfeita. print("[DEBUG] Múltiplas partes geradas, concatenando...") partes_mp4_fade = self._gerar_lista_com_transicoes(pasta=temp_dir, video_paths=partes_mp4, crossfade_frames=8) self._concat_mp4s_no_reencode(partes_mp4_fade, final_vid) else: shutil.move(partes_mp4[0], final_vid) 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: gc.collect() if torch.cuda.is_available(): torch.cuda.empty_cache() torch.cuda.ipc_collect() self.finalize(keep_paths=[]) # O resultado final já foi movido print("Criando instância do VideoService. O carregamento do modelo começará agora...") video_generation_service = VideoService()