Update api/ltx_server_refactored.py

api/ltx_server_refactored.py

@@ -1,72 +1,86 @@
-# ltx_server_refactored.py — VideoService (Modular Version with Simple Overlap Chunking)
-# Em api/ltx_server_refactored.py
+# FILE: ltx_server_refactored_complete.py
+# DESCRIPTION: Backend service for video generation using LTX-Video pipeline.
+#              Features modular generation, narrative chunking, and resource management.
 
-import warnings
-from huggingface_hub import logging
-import os, subprocess, shlex, tempfile
-import torch
+import gc
+import io
 import json
-import numpy as np
-import random
+import logging
 import os
-import io
-
-import shlex
-import yaml
-from typing import List, Dict
-from pathlib import Path
-import imageio
-from PIL import Image
-import tempfile
-from huggingface_hub import hf_hub_download
-import sys
-import subprocess
-import gc
+import random
 import shutil
-import contextlib
+import subprocess
+import sys
+import tempfile
 import time
 import traceback
-from api.gpu_manager import gpu_manager
+import warnings
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import torch
+import yaml
 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
+from huggingface_hub import hf_hub_download
+
+# ==============================================================================
+# --- INITIAL SETUP & CONFIGURATION ---
+# ==============================================================================
 
+# Suppress excessive logs from external libraries
+warnings.filterwarnings("ignore")
+logging.getLogger("huggingface_hub").setLevel(logging.ERROR)
+logging.basicConfig(level=logging.INFO, format='[%(levelname)s] %(message)s')
+
+# --- CONSTANTS ---
 DEPS_DIR = Path("/data")
 LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video"
-logging.set_verbosity_error()
-logging.set_verbosity_warning()
-logging.set_verbosity_info()
-logging.set_verbosity_debug()
-LTXV_DEBUG=1
-LTXV_FRAME_LOG_EVERY=8
-warnings.filterwarnings("ignore", category=UserWarning)
-warnings.filterwarnings("ignore", category=FutureWarning)
-warnings.filterwarnings("ignore", message=".*")
-
-# (Todas as funções de setup, helpers e inicialização da classe permanecem inalteradas)
-# ... (run_setup, add_deps_to_path, _query_gpu_processes_via_nvml, etc.)
-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()
+BASE_CONFIG_PATH = LTX_VIDEO_REPO_DIR / "configs"
+DEFAULT_CONFIG_FILE = BASE_CONFIG_PATH / "ltxv-13b-0.9.8-distilled-fp8.yaml"
+LTX_REPO_ID = "Lightricks/LTX-Video"
+RESULTS_DIR = Path("/app/output")
+DEFAULT_FPS = 24.0
+FRAMES_ALIGNMENT = 8
+
+# --- DEPENDENCY PATH SETUP ---
+# Ensures the LTX-Video library can be imported
 def add_deps_to_path():
+    """Adds the LTX repository directory to the Python system path."""
     repo_path = str(LTX_VIDEO_REPO_DIR.resolve())
-    if str(LTX_VIDEO_REPO_DIR.resolve()) not in sys.path:
+    if repo_path not in sys.path:
         sys.path.insert(0, repo_path)
-        print(f"[DEBUG] Repo adicionado ao sys.path: {repo_path}")
-def calculate_padding(orig_h, orig_w, target_h, target_w):
+        logging.info(f"Repo added to sys.path: {repo_path}")
+
+add_deps_to_path()
+
+# --- PROJECT IMPORTS ---
+# These must come after the path setup
+from api.gpu_manager import gpu_manager
+from ltx_video.models.autoencoders.vae_encode import (normalize_latents, un_normalize_latents)
+from ltx_video.pipelines.pipeline_ltx_video import (ConditioningItem, LTXMultiScalePipeline, adain_filter_latent)
+from ltx_video.pipelines.pipeline_ltx_video import create_ltx_video_pipeline, create_latent_upsampler
+from ltx_video.utils.inference_utils import load_image_to_tensor_with_resize_and_crop
+from managers.vae_manager import vae_manager_singleton
+from tools.video_encode_tool import video_encode_tool_singleton
+
+
+# ==============================================================================
+# --- UTILITY & HELPER FUNCTIONS ---
+# ==============================================================================
+
+def seed_everything(seed: int):
+    """Sets the seed for reproducibility across all relevant libraries."""
+    random.seed(seed)
+    os.environ['PYTHONHASHSEED'] = str(seed)
+    np.random.seed(seed)
+    torch.manual_seed(seed)
+    torch.cuda.manual_seed_all(seed)
+    # Potentially faster, but less reproducible
+    # torch.backends.cudnn.deterministic = False
+    # torch.backends.cudnn.benchmark = True
+
+def calculate_padding(orig_h: int, orig_w: int, target_h: int, target_w: int) -> Tuple[int, int, int, int]:
+    """Calculates symmetric padding values to reach a target dimension."""
     pad_h = target_h - orig_h
     pad_w = target_w - orig_w
     pad_top = pad_h // 2
@@ -74,651 +88,375 @@ def calculate_padding(orig_h, orig_w, target_h, target_w):
     pad_left = pad_w // 2
     pad_right = pad_w - pad_left
     return (pad_left, pad_right, pad_top, pad_bottom)
-def log_tensor_info(tensor, name="Tensor"):
+
+def log_tensor_info(tensor: torch.Tensor, name: str = "Tensor"):
+    """Logs detailed information about a PyTorch tensor for debugging."""
     if not isinstance(tensor, torch.Tensor):
-        print(f"\n[INFO] '{name}' não é tensor.")
+        logging.debug(f"'{name}' is not a tensor.")
         return
-    print(f"\n--- Tensor: {name} ---")
-    print(f"  - Shape: {tuple(tensor.shape)}")
-    print(f"  - Dtype: {tensor.dtype}")
-    print(f"  - Device: {tensor.device}")
+    
+    info_str = (
+        f"--- Tensor: {name} ---\n"
+        f"  - Shape: {tuple(tensor.shape)}\n"
+        f"  - Dtype: {tensor.dtype}\n"
+        f"  - Device: {tensor.device}\n"
+    )
     if tensor.numel() > 0:
         try:
-            print(f"  - Min: {tensor.min().item():.4f}  Max: {tensor.max().item():.4f}  Mean: {tensor.mean().item():.4f}")
+            info_str += (
+                f"  - Min: {tensor.min().item():.4f} | "
+                f"Max: {tensor.max().item():.4f} | "
+                f"Mean: {tensor.mean().item():.4f}\n"
+            )
         except Exception:
-            pass
-    print("------------------------------------------\n")
+            pass # Fails on some dtypes
+    logging.debug(info_str + "----------------------")
 
-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,
-)
+
+# ==============================================================================
+# --- VIDEO SERVICE CLASS ---
+# ==============================================================================
 
 class VideoService:
-    def _load_config(self):
-        base = LTX_VIDEO_REPO_DIR / "configs"
-        config_path = base / "ltxv-13b-0.9.8-distilled-fp8.yaml"
+    """
+    Backend service for orchestrating video generation using the LTX-Video pipeline.
+    Encapsulates model loading, state management, and the logic for multi-stage
+    video generation (low-resolution, upscale).
+    """
+
+    def __init__(self):
+        """Initializes the service, loads models, and configures the environment."""
+        t0 = time.perf_counter()
+        logging.info("Initializing VideoService...")
+        RESULTS_DIR.mkdir(parents=True, exist_ok=True)
+
+        self.config = self._load_config(DEFAULT_CONFIG_FILE)
+        self._tmp_dirs = set()
+        
+        self.pipeline, self.latent_upsampler = self._load_models_on_cpu()
+
+        target_device = gpu_manager.get_ltx_device()
+        self.device = torch.device("cpu") # Default device
+        self.move_to_device(target_device)
+
+        self._apply_precision_policy()
+        vae_manager_singleton.attach_pipeline(
+            self.pipeline,
+            device=self.device,
+            autocast_dtype=self.runtime_autocast_dtype
+        )
+        
+        logging.info(f"VideoService ready. Startup time: {time.perf_counter()-t0:.2f}s")
+
+    # ==========================================================================
+    # --- LIFECYCLE & MODEL MANAGEMENT ---
+    # ==========================================================================
+
+    def _load_config(self, config_path: Path) -> Dict:
+        """Loads the YAML configuration file."""
+        logging.info(f"Loading config from: {config_path}")
         with open(config_path, "r") as file:
             return yaml.safe_load(file)
 
-    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 [])
-        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}")
-            
-    def _load_models(self):
+    def _load_models_on_cpu(self) -> Tuple[LTXMultiScalePipeline, Optional[torch.nn.Module]]:
+        """Downloads and loads the pipeline and upsampler checkpoints onto the CPU."""
         t0 = time.perf_counter()
-        LTX_REPO = "Lightricks/LTX-Video"
-        print("[DEBUG] Baixando checkpoint principal...")
+        
+        logging.info("Downloading main checkpoint...")
         distilled_model_path = hf_hub_download(
-            repo_id=LTX_REPO,
+            repo_id=LTX_REPO_ID,
             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"],
+            device="cpu", # Load on CPU first
+            # Pass other config values directly
+            **{k: v for k, v in self.config.items() if k in create_ltx_video_pipeline.__code__.co_varnames}
         )
-        print("[DEBUG] Pipeline pronto.")
-
+        
         latent_upsampler = None
         if self.config.get("spatial_upscaler_model_path"):
-            print("[DEBUG] Construindo latent_upsampler...")
+            logging.info("Downloading spatial upscaler checkpoint...")
+            spatial_upscaler_path = hf_hub_download(
+                repo_id=LTX_REPO_ID,
+                filename=self.config["spatial_upscaler_model_path"],
+                token=os.getenv("HF_TOKEN")
+            )
+            self.config["spatial_upscaler_model_path"] = spatial_upscaler_path
             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 _apply_precision_policy(self):
-        prec = str(self.config.get("precision", "")).lower()
-        self.runtime_autocast_dtype = torch.float32
-        if prec in ["float8_e4m3fn", "bfloat16"]:
-            self.runtime_autocast_dtype = torch.bfloat16
-        elif prec == "mixed_precision":
-            self.runtime_autocast_dtype = torch.float16
 
-    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}")
+        logging.info(f"Models loaded on CPU in {time.perf_counter()-t0:.2f}s")
+        return pipeline, latent_upsampler
 
-    @torch.no_grad()
-    def _upsample_latents_internal(self, latents: torch.Tensor) -> torch.Tensor:
-        try:
-            if not self.latent_upsampler:
-                raise ValueError("Latent Upsampler não está carregado.")
-            latents_unnormalized = un_normalize_latents(latents, self.pipeline.vae, vae_per_channel_normalize=True)
-            upsampled_latents = self.latent_upsampler(latents_unnormalized)
-            return normalize_latents(upsampled_latents, self.pipeline.vae, vae_per_channel_normalize=True)
-        except Exception as e:
-            pass
-        finally:
+    def move_to_device(self, device_str: str):
+        """Moves all relevant models to the specified device (e.g., 'cuda:0' or 'cpu')."""
+        target_device = torch.device(device_str)
+        if self.device == target_device:
+            logging.info(f"Models are already on the target device: {device_str}")
+            return
+        
+        logging.info(f"Moving models to {device_str}...")
+        self.device = target_device
+        self.pipeline.to(self.device)
+        if self.latent_upsampler:
+            self.latent_upsampler.to(self.device)
+        
+        if device_str == "cpu" and torch.cuda.is_available():
             torch.cuda.empty_cache()
-            torch.cuda.ipc_collect()
-            self.finalize(keep_paths=[])
-
-    def _prepare_conditioning_tensor(self, filepath, height, width, padding_values):
-        tensor = load_image_to_tensor_with_resize_and_crop(filepath, height, width)
-        tensor = torch.nn.functional.pad(tensor, padding_values)
-        log_tensor_info(tensor, f"_prepare_conditioning_tensor")
-        return tensor.to(self.device, dtype=self.runtime_autocast_dtype)
-
             
-    def _save_and_log_video(self, pixel_tensor, base_filename, fps, temp_dir, results_dir, used_seed, progress_callback=None):
-        output_path = os.path.join(temp_dir, f"{base_filename}_.mp4")
-        video_encode_tool_singleton.save_video_from_tensor(
-            pixel_tensor, output_path, fps=fps, progress_callback=progress_callback
-        )
-        final_path = os.path.join(results_dir, f"{base_filename}_.mp4")
-        shutil.move(output_path, final_path)
-        print(f"[DEBUG] Vídeo salvo em: {final_path}")
-        return final_path
-
-    def _load_tensor(self, caminho):
-        # Se já é um tensor, retorna diretamente
-        if isinstance(caminho, torch.Tensor):
-            return caminho
-        # Se é bytes, carrega do buffer
-        if isinstance(caminho, (bytes, bytearray)):
-            return torch.load(io.BytesIO(caminho))
-        # Caso contrário, assume que é um caminho de arquivo
-        return torch.load(caminho)
-
-
-    # ==============================================================================
-    # --- FUNÇÕES MODULARES COM A LÓGICA DE CHUNKING SIMPLIFICADA ---
-    # ==============================================================================
-
-    def prepare_condition_items(self, items_list: List, height: int, width: int, num_frames: int):
-        if not items_list: return []
-        height_padded = ((height - 1) // 8 + 1) * 8
-        width_padded = ((width - 1) // 8 + 1) * 8
-        padding_values = calculate_padding(height, width, height_padded, width_padded)
-        conditioning_items = []
-        for media, frame, weight in items_list:
-            tensor = self._prepare_conditioning_tensor(media, height, width, padding_values) if isinstance(media, str) else media.to(self.device, dtype=self.runtime_autocast_dtype)
-            safe_frame = max(0, min(int(frame), num_frames - 1))
-            conditioning_items.append(ConditioningItem(tensor, safe_frame, float(weight)))
-        return conditioning_items
+        logging.info(f"Models successfully moved to {self.device}.")
 
-    def _prepare_condition_items_latent(self, items_list: List):
-        if not items_list: 
-            return []
-        conditioning_items = []
-        for tensor_patch, frame, weight in items_list:
-            # Verifica se já é um tensor
-            if isinstance(tensor_patch, torch.Tensor):
-                tensor = tensor_patch.to(self.device)
-            # Se é bytes, carrega do buffer
-            elif isinstance(tensor_patch, (bytes, bytearray)):
-                tensor = torch.load(io.BytesIO(tensor_patch)).to(self.device)
-            # Caso contrário, assume que é um caminho de arquivo
-            else:
-                tensor = torch.load(tensor_patch).to(self.device)
-            safe_frame = max(0, int(frame))
-            conditioning_items.append(ConditioningItem(tensor, safe_frame, float(weight)))
-        return conditioning_items
-
-    def generate_low(self, prompt, negative_prompt, height, width, duration, guidance_scale, seed, conditioning_items=None):
-        used_seed = random.randint(0, 2**32 - 1) if seed is None else int(seed)
-        seed_everething(used_seed)
-        FPS = 24.0
+    def finalize(self, keep_paths: Optional[List[str]] = None):
+        """Cleans up GPU memory and temporary directories."""
+        logging.debug("Finalizing resources...")
+        gc.collect()
+        if torch.cuda.is_available():
+            torch.cuda.empty_cache()
+            try:
+                torch.cuda.ipc_collect()
+            except Exception:
+                pass
         
-        actual_num_frames = max(9, int(round((round(duration * FPS) - 1) / 8.0) * 8 + 1))
-        height_padded = ((height - 1) // 8 + 1) * 8
-        width_padded = ((width - 1) // 8 + 1) * 8
-        temp_dir = tempfile.mkdtemp(prefix="ltxv_low_"); self._register_tmp_dir(temp_dir)
-        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
-        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)
-        first_pass_kwargs = {
-            "prompt": prompt, "negative_prompt": negative_prompt, "height": downscaled_height, "width": downscaled_width,
-            "num_frames": actual_num_frames, "frame_rate": int(FPS), "generator": torch.Generator(device=self.device).manual_seed(used_seed),
-            "output_type": "latent", 
-            #"conditioning_items": conditioning_items, 
-            #"guidance_scale": float(guidance_scale),
-            **(self.config.get("first_pass", {}))
-        }
+        # Optional: Clean up temporary directories if needed (logic can be added here)
 
-        print(f"[DEBUG]  generate_low.first_pass_kwargs: {first_pass_kwargs}")
-        try: 
-            with torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype, enabled=self.device == 'cuda'):
-                latents = self.pipeline(**first_pass_kwargs).images
-                pixel_tensor = vae_manager_singleton.decode(latents.clone(), decode_timestep=float(self.config.get("decode_timestep", 0.05)))
-                video_path = self._save_and_log_video(pixel_tensor, "low_res_video", FPS, temp_dir, results_dir, used_seed)
-                latents_cpu = latents.detach().to("cpu")
-                tensor_path = os.path.join(results_dir, f"latents_low_res_{used_seed}.pt")
-                torch.save(latents_cpu, tensor_path)
-            return video_path, tensor_path, used_seed
-        
-        except Exception as e:
-            print(f"[DEBUG] falhou: {e}")
-        finally:
-            torch.cuda.empty_cache()
-            torch.cuda.ipc_collect()
-            self.finalize(keep_paths=[])
 
-    # ==============================================================================
-    # --- FUNÇÃO #1: GERADOR DE CHUNK ÚNICO (AUXILIAR INTERNA) ---
-    # ==============================================================================
-    def _generate_single_chunk_low(
-        self, prompt, negative_prompt, 
-        height, width, num_frames, guidance_scale, 
-        seed, itens_conditions_itens, 
-        ltx_configs_override=None):
+    # ==========================================================================
+    # --- PUBLIC ORCHESTRATORS ---
+    # These are the main entry points called by the frontend.
+    # ==========================================================================
+
+    def generate_narrative_low(self, prompt: str, **kwargs) -> Tuple[Optional[str], Optional[str], Optional[int]]:
         """
-        [NÓ DE GERAÇÃO]
-        Gera um ÚNICO chunk de latentes brutos. Esta é a unidade de trabalho fundamental.
+        [ORCHESTRATOR] Generates a video from a multi-line prompt, creating a sequence of scenes.
+        
+        Returns:
+            A tuple of (video_path, latents_path, used_seed).
         """
-        print("\n" + "-"*20 + " INÍCIO: _generate_single_chunk_low " + "-"*20)
+        logging.info("Starting narrative low-res generation...")
+        used_seed = self._resolve_seed(kwargs.get("seed"))
+        seed_everything(used_seed)
 
-        num_frames = ((num_frames - 1)//8)*8 + 1
-        
-        used_seed = random.randint(0, 2**32 - 1) if seed is None else int(seed)
-        seed_everething(used_seed)
-        
-        # --- NÓ 1.1: SETUP DE PARÂMETROS ---
-        height_padded = ((height - 1) // 8 + 1) * 8
-        width_padded = ((width - 1) // 8 + 1) * 8
-        generator = torch.Generator(device=self.device).manual_seed(seed)
-        
-        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)
+        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.")
+
+        num_chunks = len(prompt_list)
+        total_frames = self._calculate_aligned_frames(kwargs.get("duration", 4.0))
+        frames_per_chunk = (total_frames // num_chunks // FRAMES_ALIGNMENT) * FRAMES_ALIGNMENT
+        overlap_frames = self.config.get("overlap_frames", 8)
         
-        # --- NÓ 1.2: MONTAGEM DE CONDIÇÕES E OVERRIDES ---
+        all_latents_paths = []
+        overlap_condition_item = None
         
-        first_pass_config = self.config.get("first_pass", {}).copy()
+        try:
+            for i, chunk_prompt in enumerate(prompt_list):
+                logging.info(f"Generating narrative chunk {i+1}/{num_chunks}: '{chunk_prompt[:50]}...'")
 
-        if ltx_configs_override:
-            print("[DEBUG] Sobrepondo configurações do LTX com valores da UI...")
-            preset = ltx_configs_override.get("guidance_preset")
-            if preset == "Customizado":
-                try:
-                    first_pass_config["guidance_scale"] = json.loads(ltx_configs_override["guidance_scale_list"])
-                    first_pass_config["stg_scale"] = json.loads(ltx_configs_override["stg_scale_list"])
-                    #first_pass_config["guidance_timesteps"] = json.loads(ltx_configs_override["timesteps_list"])
-                except Exception as e:
-                    print(f"  > ERRO ao parsear valores customizados: {e}. Usando Padrão como fallback.")
-            elif preset == "Agressivo":
-                first_pass_config["guidance_scale"] = [1, 2, 8, 12, 8, 2, 1]
-                first_pass_config["stg_scale"] = [0, 0, 5, 6, 5, 3, 2]
-            elif preset == "Suave":
-                first_pass_config["guidance_scale"] = [1, 1, 4, 5, 4, 1, 1]
-                first_pass_config["stg_scale"] = [0, 0, 2, 2, 2, 1, 0]
-                                        
-        first_pass_kwargs = {
-            "prompt": prompt, "negative_prompt": negative_prompt, "height": downscaled_height, "width": downscaled_width,
-            "num_frames": num_frames, "frame_rate": 24, "generator": generator, "output_type": "latent",
-            "conditioning_items": itens_conditions_itens,
-            **first_pass_config,
-        }
+                current_frames = frames_per_chunk
+                if i > 0:
+                    current_frames += overlap_frames
+                
+                # Use initial image conditions only for the first chunk
+                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 chunk {i+1}.")
 
-        
-        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
-    
-        
-        # --- NÓ 1.3: CHAMADA AO PIPELINE ---
-        try: 
-            with torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype, enabled=self.device.type == 'cuda'):
-                latents_bruto = self.pipeline(**first_pass_kwargs).images
-                #latents_cpu_bruto = latents_bruto.detach().to("cpu")
-                #tensor_path_cpu = os.path.join(results_dir, f"latents_low_res.pt")
-                #torch.save(latents_cpu_bruto, tensor_path_cpu)
-                log_tensor_info(latents_bruto, f"Latente Bruto Gerado para: '{prompt[:40]}...'")
-        
-            print("-" * 20 + " FIM: _generate_single_chunk_low " + "-"*20)
-            return latents_bruto
+                # Create overlap for the next chunk
+                if i < num_chunks - 1:
+                    overlap_latents = chunk_latents[:, :, -overlap_frames:, :, :].clone()
+                    log_tensor_info(overlap_latents, f"Overlap Latents from chunk {i+1}")
+                    overlap_condition_item = ConditioningItem(
+                        media_item=overlap_latents, media_frame_number=0, conditioning_strength=1.0
+                    )
+                
+                # Trim the overlap from the current chunk before saving
+                if i > 0:
+                    chunk_latents = chunk_latents[:, :, overlap_frames:, :, :]
+                
+                # Save chunk latents to disk to manage memory
+                chunk_path = RESULTS_DIR / f"chunk_{i}_{used_seed}.pt"
+                torch.save(chunk_latents.cpu(), chunk_path)
+                all_latents_paths.append(chunk_path)
+            
+            # Concatenate, decode, and save the final video
+            return self._finalize_generation(all_latents_paths, "narrative_video", used_seed)
 
         except Exception as e:
-            print("-" * 20 + " ERRO: _generate_single_chunk_low --------------------")
+            logging.error(f"Error during narrative generation: {e}")
             traceback.print_exc()
-            print("-" * 20 + " ----------------------------------------------")
-            return None
+            return None, None, None
         finally:
-            torch.cuda.empty_cache()
-            torch.cuda.ipc_collect()
-            self.finalize(keep_paths=[])
-
-    # ==============================================================================
-    # --- FUNÇÃO #2: ORQUESTRADOR NARRATIVO (MÚLTIPLOS PROMPTS) ---
-    # ==============================================================================
-    def generate_narrative_low(
-        self, prompt: str, negative_prompt, 
-        height, width, duration, guidance_scale, 
-        seed, initial_conditions, overlap_frames: int = 4,
-        ltx_configs_override: dict = None):
-        """
-        [ORQUESTRADOR NARRATIVO]
-        Gera um vídeo em múltiplos chunks sequenciais a partir de um prompt com várias linhas.
-        """
-        print("\n" + "="*80)
-        print("======           INICIANDO GERAÇÃO NARRATIVA EM CHUNKS (LOW-RES)           ======")
-        print("="*80)
-        
-        
-        used_seed = random.randint(0, 2**32 - 1) if seed is None else int(seed)
-        seed_everething(used_seed)
-        FPS = 24.0
+            # Clean up intermediate chunk files
+            for path in all_latents_paths:
+                if os.path.exists(path):
+                    os.remove(path)
+            self.finalize()
 
-        prompt_list = [p.strip() for p in prompt.splitlines() if p.strip()]
-        num_chunks = len(prompt_list)
-        if num_chunks == 0: raise ValueError("O prompt está vazio ou não contém linhas válidas.")
-        
-        total_actual_frames = max(8, int(round((round(duration * FPS) ) / 8.0) * 8 ))
 
-                
-        frames_per_chunk = total_actual_frames//num_chunks
-        #frames_per_chunk_last = max(9, frames_per_chunk_last)
-        
-        poda_latents_num = overlap_frames 
-        
+    def generate_single_low(self, **kwargs) -> Tuple[Optional[str], Optional[str], Optional[int]]:
+        """
+        [ORCHESTRATOR] Generates a video from a single prompt in one go.
         
-        latentes_chunk_video = []
-        overlap_condition = []
-        overlap_latents = None
-        lista_patch_latentes_chunk = []
-        condition_item_latent_overlap = None
-        temp_dir = tempfile.mkdtemp(prefix="ltxv_narrative_"); self._register_tmp_dir(temp_dir)
-        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
+        Returns:
+            A tuple of (video_path, latents_path, used_seed).
+        """
+        logging.info("Starting single-prompt low-res generation...")
+        used_seed = self._resolve_seed(kwargs.get("seed"))
+        seed_everything(used_seed)
         
-        for i, chunk_prompt in enumerate(prompt_list):
-            print(f"\n--- Gerando Chunk Narrativo {i+1}/{num_chunks}: '{chunk_prompt}' ---")
-
-            #current_image_conditions = []
-            #if initial_image_conditions:
-            #    cond_item_original = initial_image_conditions[0]
-            #    if i == 0:
-            #        current_image_conditions.append(cond_item_original)
-            #    else:
-            #        cond_item_fraco = ConditioningItem(
-            #            media_item=cond_item_original.media_item, media_frame_number=0, conditioning_strength=0.1
-            #        )
-            #        current_image_conditions.append(cond_item_fraco)
-
-
-            poda_latents_num = 8
-            
-            if i > 0 and poda_latents_num > 0:
-                frames_per_chunk += poda_latents_num
-            else:
-                frames_per_chunk = frames_per_chunk
-           
-            if i == num_chunks - 1:
-                frames_per_chunk = frames_per_chunk+poda_latents_num
-
-            #frames_per_chunk = ((frames_per_chunk - 1)//8)*8 + 1
-
-
-            if i> 0:
-                 initial_conditions = []
-
-            if i > 0:
-                initial_conditions = []
+        try:
+            total_frames = self._calculate_aligned_frames(kwargs.get("duration", 4.0), min_frames=9)
             
-            if overlap_latents is not None:
-                # Já é um tensor, usa diretamente
-                overlap_latents_r = overlap_latents
-                items_list = [[overlap_latents_r, 0, 1.0]]
-                overlap_condition = self._prepare_condition_items_latent(items_list)
-                        
-            itens_conditions_itens = initial_conditions + overlap_condition
-
-            latentes_bruto_r = self._generate_single_chunk_low(
-                prompt=chunk_prompt, negative_prompt=negative_prompt, height=height, width=width,
-                num_frames=frames_per_chunk, guidance_scale=guidance_scale, seed=used_seed + i,
-                itens_conditions_itens=itens_conditions_itens, 
-                ltx_configs_override=ltx_configs_override
+            final_latents = self._generate_single_chunk_low(
+                num_frames=total_frames,
+                seed=used_seed,
+                conditioning_items=kwargs.get("initial_conditions", []),
+                **kwargs
             )
-
-
-            print(f"[DEBUG]  generate_narrative_low.frames_per_chunk: {frames_per_chunk}")
-            log_tensor_info(latentes_bruto_r, f"latentes_bruto_r recebidk: {i}...'")
-        
-            #latent_path_bufer = load_tensor(latent_path)
-            #final_latents = torch.cat(lista_tensores, dim=2).to(self.device)
-            
             
-            #poda inicio overlap
-            if i > 0 and poda_latents_num > 0:
-                 latentes_bruto = latentes_bruto_r[:, :, poda_latents_num:, :, :].clone()
-            else: 
-                latentes_bruto = latentes_bruto_r[:, :, :, :, :].clone()
-    
-            log_tensor_info(latentes_bruto, f"latentes_bruto recebidk: {i}...'")
-        
-            # cria estado overlap para proximo
-            if i < num_chunks - 1 and poda_latents_num > 0:
-                overlap_latents = latentes_bruto_r[:, :, -poda_latents_num:, :, :].clone()
-                log_tensor_info(overlap_latents, f"overlap_latents recebidk: {i}...'")
-                overlap_latents = overlap_latents.detach().to(self.device)
-                condition_item_latent_overlap = ConditioningItem(
-                    media_item=overlap_latents, media_frame_number=0, conditioning_strength=1.0
-                )
-
+            if final_latents is None:
+                raise RuntimeError("Failed to generate latents.")
 
-            #adiciona a lista
-            tensor_path_podado = os.path.join(results_dir, f"latents_poda{i}_res.pt")
-            torch.save(latentes_bruto, tensor_path_podado)
-            lista_patch_latentes_chunk.append(tensor_path_podado)
+            # Save latents to a single file, then decode and save video
+            latents_path = RESULTS_DIR / f"single_{used_seed}.pt"
+            torch.save(final_latents.cpu(), latents_path)
+            return self._finalize_generation([latents_path], "single_video", used_seed)
 
-        print("\n--- Finalizando Narrativa: Concatenando chunks ---")
-        
-        # Carrega cada tensor do disco
-        lista_tensores = [self._load_tensor(c) for c in lista_patch_latentes_chunk]
-        final_latents = torch.cat(lista_tensores, dim=2).to(self.device)
-        log_tensor_info(final_latents, "Tensor de Latentes Final Concatenado")
-        
-        try: 
-            with torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype, enabled=self.device.type == 'cuda'):
-                pixel_tensor = vae_manager_singleton.decode(final_latents, decode_timestep=float(self.config.get("decode_timestep", 0.05)))
-                pixel_tensor_cpu = pixel_tensor.detach().to("cpu")
-                video_path = self._save_and_log_video(pixel_tensor_cpu, "narrative_video", FPS, temp_dir, results_dir, used_seed)
-                final_latents_cpu = final_latents.detach().to("cpu")
-                final_latents_patch = os.path.join(results_dir, f"latents_low_fim.pt")
-                torch.save(final_latents_cpu, final_latents_patch)
-            return video_path, final_latents_patch, used_seed
-        
         except Exception as e:
-            print("-" * 20 + " ERRO: generate_narrative_low --------------------")
+            logging.error(f"Error during single generation: {e}")
             traceback.print_exc()
-            print("-" * 20 + " ----------------------------------------------")
             return None, None, None
         finally:
-            torch.cuda.empty_cache()
-            torch.cuda.ipc_collect()
-            self.finalize(keep_paths=[])
-            
-    # ==============================================================================
-    # --- FUNÇÃO #3: ORQUESTRADOR SIMPLES (PROMPT ÚNICO) ---
-    # ==============================================================================
-    def generate_single_low(
-        self, prompt: str, negative_prompt, 
-        height, width, duration, guidance_scale, 
-        seed, initial_conditions=None,
-        ltx_configs_override: dict = None):
+            self.finalize()
+    
+
+    # ==========================================================================
+    # --- INTERNAL WORKER UNITS ---
+    # ==========================================================================
+
+    def _generate_single_chunk_low(
+        self, prompt: str, negative_prompt: str, height: int, width: int, num_frames: int, seed: int,
+        conditioning_items: List[ConditioningItem], ltx_configs_override: Optional[Dict], **kwargs
+    ) -> Optional[torch.Tensor]:
         """
-        [ORQUESTRADOR SIMPLES]
-        Gera um vídeo completo em um único chunk. Ideal para prompts simples e curtos.
+        [WORKER] Generates a single chunk of latents. This is the core generation unit.
+        Returns the raw latents tensor on the target device, or None on failure.
         """
-        print("\n" + "="*80)
-        print("======             INICIANDO GERAÇÃO SIMPLES EM CHUNK ÚNICO (LOW-RES)             ======")
-        print("="*80)
+        height_padded, width_padded = (self._align(d) for d in (height, width))
+        downscale_factor = self.config.get("downscale_factor", 0.6666666)
+        vae_scale_factor = self.pipeline.vae_scale_factor
 
-        used_seed = random.randint(0, 2**32 - 1) if seed is None else int(seed)
-        seed_everething(used_seed)
-        FPS = 24.0
+        downscaled_height = self._align(int(height_padded * downscale_factor), vae_scale_factor)
+        downscaled_width = self._align(int(width_padded * downscale_factor), vae_scale_factor)
 
-        total_actual_frames = max(9, int(round((round(duration * FPS) - 1) / 8.0) * 8 + 1))
+        first_pass_config = self.config.get("first_pass", {}).copy()
+        if ltx_configs_override:
+             first_pass_config.update(self._prepare_guidance_overrides(ltx_configs_override))
+
+        pipeline_kwargs = {
+            "prompt": prompt,
+            "negative_prompt": negative_prompt,
+            "height": downscaled_height,
+            "width": downscaled_width,
+            "num_frames": num_frames,
+            "frame_rate": DEFAULT_FPS,
+            "generator": torch.Generator(device=self.device).manual_seed(seed),
+            "output_type": "latent",
+            "conditioning_items": conditioning_items,
+            **first_pass_config
+        }
         
-        temp_dir = tempfile.mkdtemp(prefix="ltxv_single_"); self._register_tmp_dir(temp_dir)
-        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
-
-        # Chama a função de geração de chunk único para fazer todo o trabalho
-        final_latents = self._generate_single_chunk_low(
-            prompt=prompt, negative_prompt=negative_prompt, height=height, width=width,
-            num_frames=total_actual_frames, guidance_scale=guidance_scale, seed=used_seed,
-            itens_conditions_itens=initial_conditions, 
-            ltx_configs_override=ltx_configs_override
-        )
-
-        print("\n--- Finalizando Geração Simples: Salvando e decodificando ---")
-        log_tensor_info(final_latents, "Tensor de Latentes Final")
-
-        try: 
-            with torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype, enabled=self.device.type == 'cuda'):
-                pixel_tensor = vae_manager_singleton.decode(final_latents, decode_timestep=float(self.config.get("decode_timestep", 0.05)))
-                pixel_tensor_cpu = pixel_tensor.detach().to("cpu")
-                video_path = self._save_and_log_video(pixel_tensor_cpu, "narrative_video", FPS, temp_dir, results_dir, used_seed)
-                final_latents_cpu = final_latents.detach().to("cpu")
-                final_latents_patch = os.path.join(results_dir, f"latents_low_fim.pt")
-                torch.save(final_latents_cpu, final_latents_patch)
-            return video_path, final_latents_patch, used_seed
-        except Exception as e:
-            print("-" * 20 + " ERRO: generate_single_low --------------------")
-            traceback.print_exc()
-            print("-" * 20 + " ----------------------------------------------")
-            return None, None, None
-        finally:
-            torch.cuda.empty_cache()
-            torch.cuda.ipc_collect()
-            self.finalize(keep_paths=[])
+        logging.debug(f"Pipeline call args: { {k: v for k, v in pipeline_kwargs.items() if k != 'conditioning_items'} }")
         
-
+        with torch.autocast(device_type=self.device.type, dtype=self.runtime_autocast_dtype, enabled=self.device.type == 'cuda'):
+            latents_raw = self.pipeline(**pipeline_kwargs).images
         
-    # ==============================================================================
-    # --- FUNÇÃO #4: ORQUESTRADOR  (Upscaler + texturas hd) ---
-    # ==============================================================================
-    def generate_upscale_denoise(
-        self, latents_path, prompt, negative_prompt, 
-        guidance_scale, seed,
-    ):
-            used_seed = random.randint(0, 2**32 - 1) if seed is None else int(seed)
-            seed_everething(used_seed)
-            temp_dir = tempfile.mkdtemp(prefix="ltxv_up_"); self._register_tmp_dir(temp_dir)
-            results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
-            latents_low = torch.load(latents_path).to(self.device)
-            with torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype, enabled=self.device == 'cuda'):
-                upsampled_latents = self._upsample_latents_internal(latents_low)
-                upsampled_latents = adain_filter_latent(latents=upsampled_latents, reference_latents=latents_low)
-                del latents_low; torch.cuda.empty_cache()
-                
-                # --- LÓGICA DE DIVISÃO SIMPLES COM OVERLAP ---
-                total_frames = upsampled_latents.shape[2]
-                # Garante que mid_point seja pelo menos 1 para evitar um segundo chunk vazio se houver poucos frames
-                mid_point = max(1, total_frames // 2) 
-                chunk1 = upsampled_latents[:, :, :mid_point, :, :]
-                # O segundo chunk começa um frame antes para criar o overlap
-                chunk2 = upsampled_latents[:, :, mid_point - 1:, :, :]
-                
-                final_latents_list = []
-                for i, chunk in enumerate([chunk1, chunk2]):
-                    if chunk.shape[2] <= 1: continue # Pula chunks inválidos ou vazios
-                    second_pass_height = chunk.shape[3] * self.pipeline.vae_scale_factor
-                    second_pass_width = chunk.shape[4] * self.pipeline.vae_scale_factor
-                    second_pass_kwargs = {
-                        "prompt": prompt, "negative_prompt": negative_prompt, "height": second_pass_height, "width": second_pass_width,
-                        "num_frames": chunk.shape[2], "latents": chunk, 
-                        #"guidance_scale": float(guidance_scale),
-                        "output_type": "latent", "generator": torch.Generator(device=self.device).manual_seed(used_seed),
-                        **(self.config.get("second_pass", {}))
-                    }
-                    refined_chunk = self.pipeline(**second_pass_kwargs).images
-                    # Remove o overlap do primeiro chunk refinado antes de juntar
-                    if i == 0:
-                        final_latents_list.append(refined_chunk[:, :, :-1, :, :])
-                    else:
-                        final_latents_list.append(refined_chunk)
-    
-            final_latents = torch.cat(final_latents_list, dim=2)
-            log_tensor_info(final_latents, "Latentes Upscaled/Refinados Finais")
+        log_tensor_info(latents_raw, f"Raw Latents for '{prompt[:40]}...'")
+        return latents_raw
+
+
+    # ==========================================================================
+    # --- HELPERS & UTILITY METHODS ---
+    # ==========================================================================
     
-            latents_cpu = final_latents.detach().to("cpu")
-            tensor_path = os.path.join(results_dir, f"latents_refined_{used_seed}.pt")
-            torch.save(latents_cpu, tensor_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, "refined_video", 24.0, temp_dir, results_dir, used_seed)
-            return video_path, tensor_path    
-
-    def encode_mp4(self, latents_path: str, fps: int = 24):
-        latents = torch.load(latents_path)
-        seed = random.randint(0, 99999)
-        temp_dir = tempfile.mkdtemp(prefix="ltxv_enc_"); self._register_tmp_dir(temp_dir)
-        results_dir = "/app/output"; os.makedirs(results_dir, exist_ok=True)
+    def _finalize_generation(self, latents_paths: List[Path], base_filename: str, seed: int) -> Tuple[str, str, int]:
+        """
+        Loads latents from paths, concatenates them, decodes to video, and saves both.
+        """
+        logging.info("Finalizing generation: decoding latents to video.")
+        # Load all tensors and concatenate them on the CPU first
+        all_tensors_cpu = [torch.load(p) for p in latents_paths]
+        final_latents_cpu = torch.cat(all_tensors_cpu, dim=2)
         
-        # --- LÓGICA DE DIVISÃO SIMPLES COM OVERLAP ---
-        total_frames = latents.shape[2]
-        mid_point = max(1, total_frames // 2)
-        chunk1_latents = latents[:, :, :mid_point, :, :]
-        chunk2_latents = latents[:, :, mid_point - 1:, :, :]
+        # Save final combined latents
+        final_latents_path = RESULTS_DIR / f"latents_{base_filename}_{seed}.pt"
+        torch.save(final_latents_cpu, final_latents_path)
+        logging.info(f"Final latents saved to: {final_latents_path}")
         
-        video_parts = []
-        pixel_chunks_to_concat = []
-        with torch.autocast(device_type="cuda", dtype=self.runtime_autocast_dtype, enabled=self.device == 'cuda'):
-            for i, chunk in enumerate([chunk1_latents, chunk2_latents]):
-                if chunk.shape[2] == 0: continue
-                pixel_chunk = vae_manager_singleton.decode(chunk.to(self.device), decode_timestep=float(self.config.get("decode_timestep", 0.05)))
-                # Remove o overlap do primeiro chunk de pixels
-                if i == 0:
-                    pixel_chunks_to_concat.append(pixel_chunk[:, :, :-1, :, :])
-                else:
-                    pixel_chunks_to_concat.append(pixel_chunk)
+        # Move to GPU for decoding
+        final_latents_gpu = final_latents_cpu.to(self.device)
+        log_tensor_info(final_latents_gpu, "Final Concatenated Latents")
+
+        with torch.autocast(device_type=self.device.type, dtype=self.runtime_autocast_dtype, enabled=self.device.type == 'cuda'):
+            pixel_tensor = vae_manager_singleton.decode(
+                final_latents_gpu,
+                decode_timestep=float(self.config.get("decode_timestep", 0.05))
+            )
         
-        final_pixel_tensor = torch.cat(pixel_chunks_to_concat, dim=2)
-        final_video_path = self._save_and_log_video(final_pixel_tensor, f"final_concatenated_{seed}", fps, temp_dir, results_dir, seed)
-        return final_video_path
+        video_path = self._save_and_log_video(pixel_tensor, f"{base_filename}_{seed}")
+        return str(video_path), str(final_latents_path), seed
 
-    def __init__(self):
-        t0 = time.perf_counter()
-        print("[DEBUG] Inicializando VideoService...")
+    def prepare_condition_items(self, items_list: List, height: int, width: int, num_frames: int) -> List[ConditioningItem]:
+        """Prepares a list of ConditioningItem objects from file paths or tensors."""
+        if not items_list:
+            return []
         
-        # 1. Obter o dispositivo alvo a partir do gerenciador
-        #    Não definimos `self.device` ainda, apenas guardamos o alvo.
-        target_device = gpu_manager.get_ltx_device()
-        print(f"[DEBUG] LTX foi alocado para o dispositivo: {target_device}")
-
-        # 2. Carregar a configuração e os modelos (na CPU, como a função _load_models faz)
-        self.config = self._load_config()
-        self.pipeline, self.latent_upsampler = self._load_models()
+        height_padded, width_padded = self._align(height), self._align(width)
+        padding_values = calculate_padding(height, width, height_padded, width_padded)
+        
+        conditioning_items = []
+        for media, frame, weight in items_list:
+            tensor = self._prepare_conditioning_tensor(media, height, width, padding_values)
+            safe_frame = max(0, min(int(frame), num_frames - 1))
+            conditioning_items.append(ConditioningItem(tensor, safe_frame, float(weight)))
+        return conditioning_items
 
-        # 3. Mover os modelos para o dispositivo alvo e definir `self.device`
-        self.move_to_device(target_device) # Usando a função que já criamos!
+    def _prepare_conditioning_tensor(self, media_path: str, height: int, width: int, padding: Tuple) -> torch.Tensor:
+        """Loads and processes an image to be a conditioning tensor."""
+        tensor = load_image_to_tensor_with_resize_and_crop(media_path, height, width)
+        tensor = torch.nn.functional.pad(tensor, padding)
+        log_tensor_info(tensor, f"Prepared Conditioning Tensor from {media_path}")
+        return tensor.to(self.device, dtype=self.runtime_autocast_dtype)
 
-        # 4. Configurar o resto dos componentes com o dispositivo correto
-        self._apply_precision_policy()
-        vae_manager_singleton.attach_pipeline(
-            self.pipeline,
-            device=self.device, # Agora `self.device` está correto
-            autocast_dtype=self.runtime_autocast_dtype
-        )
-        self._tmp_dirs = set()
-        print(f"[DEBUG] VideoService pronto. boot_time={time.perf_counter()-t0:.3f}s")
-    
-    # A função move_to_device que criamos antes é essencial aqui
-    def move_to_device(self, device):
-        """Move os modelos do pipeline para o dispositivo especificado."""
-        print(f"[LTX] Movendo modelos para {device}...")
-        self.device = torch.device(device) # Garante que é um objeto torch.device
-        self.pipeline.to(self.device)
-        if self.latent_upsampler:
-            self.latent_upsampler.to(self.device)
-        print(f"[LTX] Modelos agora estão em {self.device}.")
+    def _prepare_guidance_overrides(self, ltx_configs: Dict) -> Dict:
+        """Parses UI presets for guidance into pipeline-compatible arguments."""
+        overrides = {}
+        preset = ltx_configs.get("guidance_preset", "Padrão (Recomendado)")
         
-    def move_to_cpu(self):
-        """Move os modelos para a CPU para liberar VRAM."""
-        self.move_to_device(torch.device("cpu"))
-        if torch.cuda.is_available():
-            torch.cuda.empty_cache()
-
-    
-# Instanciação limpa, sem usar `self` fora da classe.
-print("Criando instância do VideoService...")
-video_generation_service = VideoService()
-print("Instância do VideoService pronta.")
+        # Default LTX values are used if preset is 'Padrão'
+        if preset == "Agressivo":
+            overrides["guidance_scale"] = [1, 2, 8, 12, 8, 2, 1]
+            overrides["stg_scale"] = [0, 0, 5, 6, 5, 3, 2]
+        elif preset == "Suave":
+            overrides["guidance_scale"] = [1, 1, 4, 5, 4, 1, 1]
+            overrides["stg_scale"] = [0, 0, 2, 2, 2, 1, 0]
+        elif preset == "Customizado":
+            try:
+                overrides["guidance_scale"] = json.loads(ltx_configs["guidance_scale_list"])
+                overrides["stg_scale"] = json.loads(ltx_configs["stg_scale_list"])
+            except (json.JSONDecodeError, KeyError