Update api/ltx/ltx_utils.py

api/ltx/ltx_utils.py

@@ -1,361 +1,207 @@
-# FILE: api/ltx_server_refactored_complete.py
-# DESCRIPTION: Final high-level orchestrator for LTX-Video generation.
-# This version features a unified generation workflow, random seed generation,
-# delegation to specialized modules, and advanced debugging capabilities.
+# 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.
 
-import gc
+import os
+import random
 import json
 import logging
-import os
-import shutil
-import sys
-import tempfile
 import time
+import sys
 from pathlib import Path
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, Optional, Tuple, Union
+from dataclasses import dataclass
+from enum import Enum, auto
 
-import torch
-import yaml
 import numpy as np
-from huggingface_hub import hf_hub_download
+import torch
+import torchvision.transforms.functional as TVF
+from PIL import Image
+from safetensors import safe_open
+from transformers import T5EncoderModel, T5Tokenizer
 
 # ==============================================================================
-# --- SETUP E IMPORTAÇÕES DO PROJETO ---
+# --- CRITICAL: DEPENDENCY PATH INJECTION ---
 # ==============================================================================
 
-# Configuração de logging e supressão de warnings
-import warnings
-warnings.filterwarnings("ignore")
-logging.getLogger("huggingface_hub").setLevel(logging.ERROR)
-log_level = os.environ.get("ADUC_LOG_LEVEL", "INFO").upper()
-logging.basicConfig(level=log_level, format='[%(levelname)s] [%(name)s] %(message)s')
-
-# --- Constantes de Configuração ---
-DEPS_DIR = Path("/data")
-LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video"
-RESULTS_DIR = Path("/app/output")
-DEFAULT_FPS = 24.0
-FRAMES_ALIGNMENT = 8
-LTX_REPO_ID = "Lightricks/LTX-Video"
-
-# Garante que a biblioteca LTX-Video seja importável
+# Define o caminho para o repositório clonado
+LTX_VIDEO_REPO_DIR = Path("/data/LTX-Video")
+
 def add_deps_to_path():
+    """
+    Adiciona o diretório do repositório LTX ao sys.path para garantir que suas
+    bibliotecas possam ser importadas.
+    """
     repo_path = str(LTX_VIDEO_REPO_DIR.resolve())
     if repo_path not in sys.path:
         sys.path.insert(0, repo_path)
-        logging.info(f"[ltx_server] LTX-Video repository added to sys.path: {repo_path}")
+        logging.info(f"[ltx_utils] LTX-Video repository added to sys.path: {repo_path}")
 
+# Executa a função imediatamente para configurar o ambiente antes de qualquer importação.
 add_deps_to_path()
 
-# --- Módulos da nossa Arquitetura ---
+
+# ==============================================================================
+# --- IMPORTAÇÕES DA BIBLIOTECA LTX-VIDEO (Após configuração do path) ---
+# ==============================================================================
 try:
-    from api.gpu_manager import gpu_manager
-    from managers.vae_manager import vae_manager_singleton
-    from tools.video_encode_tool import video_encode_tool_singleton
-    from api.utils.debug_utils import log_function_io
+    from ltx_video.pipelines.pipeline_ltx_video import LTXVideoPipeline
+    from ltx_video.models.autoencoders.latent_upsampler import LatentUpsampler
+    from ltx_video.models.autoencoders.causal_video_autoencoder import CausalVideoAutoencoder
+    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:
-    logging.critical(f"A crucial import from the local API/architecture failed. Error: {e}", exc_info=True)
-    sys.exit(1)
+    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}")
+
 
 # ==============================================================================
-# --- FUNÇÕES AUXILIARES DO ORQUESTRADOR ---
+# --- ESTRUTURAS DE DADOS E ENUMS (Centralizadas aqui) ---
 # ==============================================================================
 
-@log_function_io
-def calculate_padding(orig_h: int, orig_w: int, target_h: int, target_w: int) -> Tuple[int, int, int, int]:
-    """Calculates symmetric padding required to meet target dimensions."""
-    pad_h = target_h - orig_h
-    pad_w = target_w - orig_w
-    pad_top = pad_h // 2
-    pad_bottom = pad_h - pad_top
-    pad_left = pad_w // 2
-    pad_right = pad_w - pad_left
-    return (pad_left, pad_right, pad_top, pad_bottom)
+@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()
+
 
 # ==============================================================================
-# --- CLASSE DE SERVIÇO (O ORQUESTRADOR) ---
+# --- FUNÇÕES DE CONSTRUÇÃO DE MODELO E PIPELINE ---
 # ==============================================================================
 
-class VideoService:
-    """
-    Orchestrates the high-level logic of video generation, delegating low-level
-    tasks to specialized managers and utility modules.
-    """
-
-    @log_function_io
-    def __init__(self):
-        t0 = time.perf_counter()
-        logging.info("Initializing VideoService Orchestrator...")
-        RESULTS_DIR.mkdir(parents=True, exist_ok=True)
-
-        target_main_device_str = str(gpu_manager.get_ltx_device())
-        target_vae_device_str = str(gpu_manager.get_ltx_vae_device())
-        logging.info(f"LTX allocated to devices: Main='{target_main_device_str}', VAE='{target_vae_device_str}'")
-
-        self.config = self._load_config()
-        self._resolve_model_paths_from_cache()
-
-        self.pipeline, self.latent_upsampler = build_ltx_pipeline_on_cpu(self.config)
-
-        self.main_device = torch.device("cpu")
-        self.vae_device = torch.device("cpu")
-        self.move_to_device(main_device_str=target_main_device_str, vae_device_str=target_vae_device_str)
-
-        self._apply_precision_policy()
-        vae_manager_singleton.attach_pipeline(self.pipeline, device=self.vae_device, autocast_dtype=self.runtime_autocast_dtype)
-        logging.info(f"VideoService ready. Startup time: {time.perf_counter()-t0:.2f}s")
-
-    def _load_config(self) -> Dict:
-        """Loads the YAML configuration file."""
-        config_path = LTX_VIDEO_REPO_DIR / "configs" / "ltxv-13b-0.9.8-distilled-fp8.yaml"
-        logging.info(f"Loading config from: {config_path}")
-        with open(config_path, "r") as file:
-            return yaml.safe_load(file)
-
-    def _resolve_model_paths_from_cache(self):
-        """Finds the absolute paths to model files in the cache and updates the in-memory config."""
-        logging.info("Resolving model paths from Hugging Face cache...")
-        cache_dir = os.environ.get("HF_HOME")
-        try:
-            main_ckpt_path = hf_hub_download(repo_id=LTX_REPO_ID, filename=self.config["checkpoint_path"], cache_dir=cache_dir)
-            self.config["checkpoint_path"] = main_ckpt_path
-            logging.info(f"  -> Main checkpoint resolved to: {main_ckpt_path}")
-
-            if self.config.get("spatial_upscaler_model_path"):
-                upscaler_path = hf_hub_download(repo_id=LTX_REPO_ID, filename=self.config["spatial_upscaler_model_path"], cache_dir=cache_dir)
-                self.config["spatial_upscaler_model_path"] = upscaler_path
-                logging.info(f"  -> Spatial upscaler resolved to: {upscaler_path}")
-        except Exception as e:
-            logging.critical(f"Failed to resolve model paths. Ensure setup.py ran correctly. Error: {e}", exc_info=True)
-            sys.exit(1)
-
-    @log_function_io
-    def move_to_device(self, main_device_str: str, vae_device_str: str):
-        """Moves pipeline components to their designated target devices."""
-        target_main_device = torch.device(main_device_str)
-        target_vae_device = torch.device(vae_device_str)
-        logging.info(f"Moving LTX models -> Main Pipeline: {target_main_device}, VAE: {target_vae_device}")
-
-        self.main_device = target_main_device
-        self.pipeline.to(self.main_device)
-        self.vae_device = target_vae_device
-        self.pipeline.vae.to(self.vae_device)
-        if self.latent_upsampler: self.latent_upsampler.to(self.main_device)
-        logging.info("LTX models successfully moved to target devices.")
-
-    def move_to_cpu(self):
-        """Moves all LTX components to CPU to free VRAM for other services."""
-        self.move_to_device(main_device_str="cpu", vae_device_str="cpu")
-        if torch.cuda.is_available(): torch.cuda.empty_cache()
-
-    def finalize(self):
-        """Cleans up GPU memory after a generation task."""
-        gc.collect()
-        if torch.cuda.is_available():
-            torch.cuda.empty_cache()
-            try: torch.cuda.ipc_collect();
-            except Exception: pass
-
-    # ==========================================================================
-    # --- LÓGICA DE NEGÓCIO: ORQUESTRADOR PÚBLICO UNIFICADO ---
-    # ==========================================================================
-
-    @log_function_io
-    def generate_low_resolution(self, prompt: str, **kwargs) -> Tuple[Optional[str], Optional[str], Optional[int]]:
-        """
-        [UNIFIED ORCHESTRATOR] Generates a low-resolution video from a prompt.
-        Handles both single-line and multi-line prompts transparently.
-        """
-        logging.info("Starting unified low-resolution generation (random seed)...")
-        used_seed = self._get_random_seed()
-        seed_everything(used_seed)
-        logging.info(f"Using randomly generated seed: {used_seed}")
-
-        prompt_list = [p.strip() for p in prompt.splitlines() if p.strip()]
-        if not prompt_list: raise ValueError("Prompt is empty or contains no valid lines.")
-        
-        is_narrative = len(prompt_list) > 1
-        logging.info(f"Generation mode detected: {'Narrative' if is_narrative else 'Simple'} ({len(prompt_list)} scene(s)).")
-
-        num_chunks = len(prompt_list)
-        total_frames = self._calculate_aligned_frames(kwargs.get("duration", 4.0))
-        frames_per_chunk = max(FRAMES_ALIGNMENT, (total_frames // num_chunks // FRAMES_ALIGNMENT) * FRAMES_ALIGNMENT)
-        overlap_frames = self.config.get("overlap_frames", 8) if is_narrative else 0
-        
-        temp_latent_paths = []
-        overlap_condition_item = None
-        
-        try:
-            for i, chunk_prompt in enumerate(prompt_list):
-                logging.info(f"Processing scene {i+1}/{num_chunks}: '{chunk_prompt[:50]}...'")
-                
-                if i == num_chunks - 1:
-                    processed_frames = (num_chunks - 1) * frames_per_chunk
-                    current_frames = total_frames - processed_frames
-                else:
-                    current_frames = frames_per_chunk
-                
-                if i > 0: current_frames += overlap_frames
-                
-                current_conditions = kwargs.get("initial_conditions", []) if i == 0 else []
-                if overlap_condition_item: current_conditions.append(overlap_condition_item)
-
-                chunk_latents = self._generate_single_chunk_low(
-                    prompt=chunk_prompt, num_frames=current_frames, seed=used_seed + i,
-                    conditioning_items=current_conditions, **kwargs
-                )
-                if chunk_latents is None: raise RuntimeError(f"Failed to generate latents for scene {i+1}.")
-
-                if is_narrative and i < num_chunks - 1:
-                    overlap_latents = chunk_latents[:, :, -overlap_frames:, :, :].clone()
-                    overlap_condition_item = ConditioningItem(media_item=overlap_latents, media_frame_number=0, conditioning_strength=1.0)
-                
-                if i > 0: chunk_latents = chunk_latents[:, :, overlap_frames:, :, :]
-                
-                chunk_path = RESULTS_DIR / f"temp_chunk_{i}_{used_seed}.pt"
-                torch.save(chunk_latents.cpu(), chunk_path)
-                temp_latent_paths.append(chunk_path)
-            
-            base_filename = "narrative_video" if is_narrative else "single_video"
-            return self._finalize_generation(temp_latent_paths, base_filename, used_seed)
-        except Exception as e:
-            logging.error(f"Error during unified generation: {e}", exc_info=True)
-            return None, None, None
-        finally:
-            for path in temp_latent_paths:
-                if path.exists(): path.unlink()
-            self.finalize()
-
-    # ==========================================================================
-    # --- UNIDADES DE TRABALHO E HELPERS INTERNOS ---
-    # ==========================================================================
-
-    @log_function_io
-    def _generate_single_chunk_low(self, **kwargs) -> Optional[torch.Tensor]:
-        """[WORKER] Calls the pipeline to generate a single chunk of latents."""
-        height_padded, width_padded = (self._align(d) for d in (kwargs['height'], kwargs['width']))
-        downscale_factor = self.config.get("downscale_factor", 0.6666666)
-        vae_scale_factor = self.pipeline.vae_scale_factor
-        downscaled_height = self._align(int(height_padded * downscale_factor), vae_scale_factor)
-        downscaled_width = self._align(int(width_padded * downscale_factor), vae_scale_factor)
-
-        first_pass_config = self.config.get("first_pass", {}).copy()
-        if kwargs.get("ltx_configs_override"):
-            self._apply_ui_overrides(first_pass_config, kwargs["ltx_configs_override"])
-
-        pipeline_kwargs = {
-            "prompt": kwargs['prompt'], "negative_prompt": kwargs['negative_prompt'],
-            "height": downscaled_height, "width": downscaled_width, "num_frames": kwargs['num_frames'],
-            "frame_rate": DEFAULT_FPS, "generator": torch.Generator(device=self.main_device).manual_seed(kwargs['seed']),
-            "output_type": "latent", "conditioning_items": kwargs['conditioning_items'], **first_pass_config
-        }
-        
-        with torch.autocast(device_type=self.main_device.type, dtype=self.runtime_autocast_dtype, enabled="cuda" in self.main_device.type):
-            latents_raw = self.pipeline(**pipeline_kwargs).images
-        
-        return latents_raw.to(self.main_device)
-
-    @log_function_io
-    def _finalize_generation(self, temp_latent_paths: List[Path], base_filename: str, seed: int) -> Tuple[str, str, int]:
-        """Consolidates latents, decodes them to video, and saves final artifacts."""
-        logging.info("Finalizing generation: decoding latents to video.")
-        all_tensors_cpu = [torch.load(p) for p in temp_latent_paths]
-        final_latents = torch.cat(all_tensors_cpu, dim=2)
-        
-        final_latents_path = RESULTS_DIR / f"latents_{base_filename}_{seed}.pt"
-        torch.save(final_latents, final_latents_path)
-        logging.info(f"Final latents saved to: {final_latents_path}")
-        
-        pixel_tensor = vae_manager_singleton.decode(
-            final_latents, decode_timestep=float(self.config.get("decode_timestep", 0.05))
-        )
-        video_path = self._save_and_log_video(pixel_tensor, f"{base_filename}_{seed}")
-        return str(video_path), str(final_latents_path), seed
-
-    @log_function_io
-    def prepare_condition_items(self, items_list: List, height: int, width: int, num_frames: int) -> List[ConditioningItem]:
-        """[UNIFIED] Prepares ConditioningItems from a mixed list of file paths and tensors."""
-        if not items_list: return []
-        height_padded, width_padded = self._align(height), self._align(width)
-        padding_values = calculate_padding(height, width, height_padded, width_padded)
-        
-        conditioning_items = []
-        for media_item, frame, weight in items_list:
-            if isinstance(media_item, str):
-                tensor = load_image_to_tensor_with_resize_and_crop(media_item, height, width)
-                tensor = torch.nn.functional.pad(tensor, padding_values)
-                tensor = tensor.to(self.main_device, dtype=self.runtime_autocast_dtype)
-            elif isinstance(media_item, torch.Tensor):
-                tensor = media_item.to(self.main_device, dtype=self.runtime_autocast_dtype)
-            else:
-                logging.warning(f"Unknown conditioning media type: {type(media_item)}. Skipping.")
-                continue
-            
-            safe_frame = max(0, min(int(frame), num_frames - 1))
-            conditioning_items.append(ConditioningItem(tensor, safe_frame, float(weight)))
-        return conditioning_items
-
-    def _apply_ui_overrides(self, config_dict: Dict, overrides: Dict):
-        """Applies advanced settings from the UI to a config dictionary."""
-        # Override step counts
-        for key in ["num_inference_steps", "skip_initial_inference_steps", "skip_final_inference_steps"]:
-            ui_value = overrides.get(key)
-            if ui_value and ui_value > 0:
-                config_dict[key] = ui_value
-                logging.info(f"Override: '{key}' set to {ui_value} by UI.")
-        
-        # Override guidance settings
-        preset = overrides.get("guidance_preset", "Padrão (Recomendado)")
-        guidance_overrides = {}
-        if preset == "Agressivo":
-            guidance_overrides = {"guidance_scale": [1, 2, 8, 12, 8, 2, 1], "stg_scale": [0, 0, 5, 6, 5, 3, 2]}
-        elif preset == "Suave":
-            guidance_overrides = {"guidance_scale": [1, 1, 4, 5, 4, 1, 1], "stg_scale": [0, 0, 2, 2, 2, 1, 0]}
-        elif preset == "Customizado":
-            try:
-                guidance_overrides["guidance_scale"] = json.loads(overrides["guidance_scale_list"])
-                guidance_overrides["stg_scale"] = json.loads(overrides["stg_scale_list"])
-            except Exception as e:
-                logging.warning(f"Failed to parse custom guidance values: {e}. Using defaults.")
-        
-        if guidance_overrides:
-            config_dict.update(guidance_overrides)
-            logging.info(f"Applying '{preset}' guidance preset overrides.")
-
-    def _save_and_log_video(self, pixel_tensor: torch.Tensor, base_filename: str) -> Path:
-        with tempfile.TemporaryDirectory() as temp_dir:
-            temp_path = os.path.join(temp_dir, f"{base_filename}.mp4")
-            video_encode_tool_singleton.save_video_from_tensor(pixel_tensor, temp_path, fps=DEFAULT_FPS)
-            final_path = RESULTS_DIR / f"{base_filename}.mp4"
-            shutil.move(temp_path, final_path)
-            logging.info(f"Video saved successfully to: {final_path}")
-            return final_path
+def create_latent_upsampler(latent_upsampler_model_path: str, device: str) -> LatentUpsampler:
+    """Loads the Latent Upsampler model from a checkpoint path."""
+    logging.info(f"Loading Latent Upsampler from: {latent_upsampler_model_path} to device: {device}")
+    latent_upsampler = LatentUpsampler.from_pretrained(latent_upsampler_model_path)
+    latent_upsampler.to(device)
+    latent_upsampler.eval()
+    return latent_upsampler
+
+def build_ltx_pipeline_on_cpu(config: Dict) -> Tuple[LTXVideoPipeline, Optional[torch.nn.Module]]:
+    """Builds the complete LTX pipeline and upsampler on the CPU."""
+    t0 = time.perf_counter()
+    logging.info("Building LTX pipeline on CPU...")
+
+    ckpt_path = Path(config["checkpoint_path"])
+    if not ckpt_path.is_file():
+        raise FileNotFoundError(f"Main checkpoint file not found: {ckpt_path}")
+
+    with safe_open(ckpt_path, framework="pt") as f:
+        metadata = f.metadata() or {}
+        config_str = metadata.get("config", "{}")
+        configs = json.loads(config_str)
+        allowed_inference_steps = configs.get("allowed_inference_steps")
+
+    vae = CausalVideoAutoencoder.from_pretrained(ckpt_path).to("cpu")
+    transformer = Transformer3DModel.from_pretrained(ckpt_path).to("cpu")
+    scheduler = RectifiedFlowScheduler.from_pretrained(ckpt_path)
     
-    def _apply_precision_policy(self):
-        precision = str(self.config.get("precision", "bfloat16")).lower()
-        if precision in ["float8_e4m3fn", "bfloat16"]: self.runtime_autocast_dtype = torch.bfloat16
-        elif precision == "mixed_precision": self.runtime_autocast_dtype = torch.float16
-        else: self.runtime_autocast_dtype = torch.float32
-        logging.info(f"Runtime precision policy set for autocast: {self.runtime_autocast_dtype}")
-
-    def _align(self, dim: int, alignment: int = FRAMES_ALIGNMENT) -> int:
-        return ((dim - 1) // alignment + 1) * alignment
+    text_encoder_path = config["text_encoder_model_name_or_path"]
+    text_encoder = T5EncoderModel.from_pretrained(text_encoder_path, subfolder="text_encoder").to("cpu")
+    tokenizer = T5Tokenizer.from_pretrained(text_encoder_path, subfolder="tokenizer")
+    patchifier = SymmetricPatchifier(patch_size=1)
+
+    precision = config.get("precision", "bfloat16")
+    if precision == "bfloat16":
+        vae.to(torch.bfloat16)
+        transformer.to(torch.bfloat16)
+        text_encoder.to(torch.bfloat16)
     
-    def _calculate_aligned_frames(self, duration_s: float, min_frames: int = 1) -> int:
-        num_frames = int(round(duration_s * DEFAULT_FPS))
-        aligned_frames = self._align(num_frames)
-        return max(aligned_frames, min_frames) # Use max(aligned_frames) para garantir alinhamento
+    pipeline = LTXVideoPipeline(
+        transformer=transformer, patchifier=patchifier, text_encoder=text_encoder,
+        tokenizer=tokenizer, scheduler=scheduler, vae=vae,
+        allowed_inference_steps=allowed_inference_steps,
+        prompt_enhancer_image_caption_model=None, prompt_enhancer_image_caption_processor=None,
+        prompt_enhancer_llm_model=None, prompt_enhancer_llm_tokenizer=None,
+    )
+
+    latent_upsampler = None
+    if config.get("spatial_upscaler_model_path"):
+        spatial_path = config["spatial_upscaler_model_path"]
+        latent_upsampler = create_latent_upsampler(spatial_path, device="cpu")
+        if precision == "bfloat16":
+            latent_upsampler.to(torch.bfloat16)
+
+    logging.info(f"LTX pipeline built on CPU in {time.perf_counter() - t0:.2f}s")
+    return pipeline, latent_upsampler
 
-    def _get_random_seed(self) -> int:
-        """Always generates and returns a new random seed."""
-        return random.randint(0, 2**32 - 1)
 
 # ==============================================================================
-# --- INSTANCIAÇÃO SINGLETON ---
+# --- FUNÇÕES AUXILIARES (Latent Processing, Seed, Image Prep) ---
 # ==============================================================================
-try:
-    video_generation_service = VideoService()
-    logging.info("Global VideoService orchestrator instance created successfully.")
-except Exception as e:
-    logging.critical(f"Failed to initialize VideoService: {e}", exc_info=True)
-    sys.exit(1)
+
+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)
+    os.environ['PYTHONHASHSEED'] = str(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    torch.cuda.manual_seed_all(seed)
+    torch.backends.cudnn.deterministic = True
+    torch.backends.cudnn.benchmark = False
+
+def load_image_to_tensor_with_resize_and_crop(
+    image_input: Union[str, Image.Image],
+    target_height: int,
+    target_width: int,
+) -> torch.Tensor:
+    """Loads and processes an image into a 5D tensor compatible with the LTX pipeline."""
+    if isinstance(image_input, str):
+        image = Image.open(image_input).convert("RGB")
+    elif isinstance(image_input, Image.Image):
+        image = image_input
+    else:
+        raise ValueError("image_input must be a file path or a PIL Image object")
+
+    input_width, input_height = image.size
+    aspect_ratio_target = target_width / target_height
+    aspect_ratio_frame = input_width / input_height
+
+    if aspect_ratio_frame > aspect_ratio_target:
+        new_width, new_height = int(input_height * aspect_ratio_target), input_height
+        x_start, y_start = (input_width - new_width) // 2, 0
+    else:
+        new_width, new_height = input_width, int(input_width / aspect_ratio_target)
+        x_start, y_start = 0, (input_height - new_height) // 2
+
+    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.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
+    frame_tensor = (frame_tensor * 2.0) - 1.0
+    
+    # Create 5D tensor: (batch_size=1, channels=3, num_frames=1, height, width)
+    return frame_tensor.unsqueeze(0).unsqueeze(2)