Upload ltx_utils.py

api/ltx/ltx_utils.py

@@ -1,6 +1,6 @@
 # FILE: api/ltx/ltx_utils.py
 # DESCRIPTION: Comprehensive, self-contained utility module for the LTX pipeline.
-# Handles dependency path injection, model loading, data structures, and helper functions.
+# Handles dependency path injection, model loading, pipeline creation, and tensor preparation.
 
 import os
 import random
@@ -10,8 +10,6 @@ import time
 import sys
 from pathlib import Path
 from typing import Dict, Optional, Tuple, Union
-from dataclasses import dataclass
-from enum import Enum, auto
 
 import numpy as np
 import torch
@@ -51,34 +49,11 @@ try:
     from ltx_video.models.transformers.transformer3d import Transformer3DModel
     from ltx_video.models.transformers.symmetric_patchifier import SymmetricPatchifier
     from ltx_video.schedulers.rf import RectifiedFlowScheduler
-    from ltx_video.models.autoencoders.vae_encode import un_normalize_latents, normalize_latents
     import ltx_video.pipelines.crf_compressor as crf_compressor
 except ImportError as e:
     raise ImportError(f"Could not import from LTX-Video library even after setting sys.path. Check repo integrity at '{LTX_VIDEO_REPO_DIR}'. Error: {e}")
 
 
-# ==============================================================================
-# --- ESTRUTURAS DE DADOS E ENUMS (Centralizadas aqui) ---
-# ==============================================================================
-
-@dataclass
-class ConditioningItem:
-    """Define a single frame-conditioning item, used to guide the generation pipeline."""
-    media_item: torch.Tensor
-    media_frame_number: int
-    conditioning_strength: float
-    media_x: Optional[int] = None
-    media_y: Optional[int] = None
-
-
-class SkipLayerStrategy(Enum):
-    """Defines the strategy for how spatio-temporal guidance is applied across transformer blocks."""
-    AttentionSkip = auto()
-    AttentionValues = auto()
-    Residual = auto()
-    TransformerBlock = auto()
-
-
 # ==============================================================================
 # --- FUNÇÕES DE CONSTRUÇÃO DE MODELO E PIPELINE ---
 # ==============================================================================
@@ -141,22 +116,9 @@ def build_ltx_pipeline_on_cpu(config: Dict) -> Tuple[LTXVideoPipeline, Optional[
 
 
 # ==============================================================================
-# --- FUNÇÕES AUXILIARES (Latent Processing, Seed, Image Prep) ---
+# --- FUNÇÕES AUXILIARES (Seed, Preparação de Imagem) ---
 # ==============================================================================
 
-def adain_filter_latent(
-    latents: torch.Tensor, reference_latents: torch.Tensor, factor=1.0
-) -> torch.Tensor:
-    """Applies AdaIN to transfer the style from a reference latent to another."""
-    result = latents.clone()
-    for i in range(latents.size(0)):
-        for c in range(latents.size(1)):
-            r_sd, r_mean = torch.std_mean(reference_latents[i, c], dim=None)
-            i_sd, i_mean = torch.std_mean(result[i, c], dim=None)
-            if i_sd > 1e-6:
-                result[i, c] = ((result[i, c] - i_mean) / i_sd) * r_sd + r_mean
-    return torch.lerp(latents, result, factor)
-
 def seed_everything(seed: int):
     """Sets the seed for reproducibility."""
     random.seed(seed)
@@ -172,7 +134,7 @@ def load_image_to_tensor_with_resize_and_crop(
     target_height: int,
     target_width: int,
 ) -> torch.Tensor:
-    """Loads and processes an image into a 5D tensor compatible with the LTX pipeline."""
+    """Loads and processes an image into a 5D pixel tensor compatible with the LTX pipeline."""
     if isinstance(image_input, str):
         image = Image.open(image_input).convert("RGB")
     elif isinstance(image_input, Image.Image):
@@ -194,13 +156,13 @@ def load_image_to_tensor_with_resize_and_crop(
     image = image.crop((x_start, y_start, x_start + new_width, y_start + new_height))
     image = image.resize((target_width, target_height), Image.Resampling.LANCZOS)
 
-    frame_tensor = TVF.to_tensor(image)
+    frame_tensor = TVF.to_tensor(image)  # PIL -> tensor (C, H, W) in [0, 1] range
     frame_tensor = TVF.gaussian_blur(frame_tensor, kernel_size=(3, 3))
     
     frame_tensor_hwc = frame_tensor.permute(1, 2, 0)
     frame_tensor_hwc = crf_compressor.compress(frame_tensor_hwc)
     frame_tensor = frame_tensor_hwc.permute(2, 0, 1)
-    # Normalize to [-1, 1] range
+    # Normalize to [-1, 1] range, which the VAE expects for encoding
     frame_tensor = (frame_tensor * 2.0) - 1.0
     
     # Create 5D tensor: (batch_size=1, channels=3, num_frames=1, height, width)