Update api/ltx_server_refactored_complete.py

api/ltx_server_refactored_complete.py

@@ -1,214 +1,443 @@
-# FILE: api/vince_pool_manager.py
-# DESCRIPTION: Singleton manager for a pool of VINCIE workers, integrated with a central GPU manager.
+# FILE: api/ltx_server_refactored_complete.py
+# DESCRIPTION: Final backend service for LTX-Video generation.
+#              Features dedicated VAE device logic, robust initialization, and narrative chunking.
 
-import os
-import sys
 import gc
+import io
+import json
+import logging
+import os
+import random
+import shutil
 import subprocess
-import threading
+import sys
+import tempfile
+import time
+import traceback
+import warnings
 from pathlib import Path
-from typing import List
+from typing import Dict, List, Optional, Tuple
 
 import torch
-from omegaconf import open_dict
+import yaml
+import numpy as np
+from einops import rearrange
+from huggingface_hub import hf_hub_download
+
+# ==============================================================================
+# --- INITIAL SETUP & CONFIGURATION ---
+# ==============================================================================
 
-# --- Import do Gerenciador Central de GPUs ---
-# Esta é a peça chave da integração. O Pool Manager perguntará a ele quais GPUs usar.
+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"
+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
+
+# --- CRITICAL: DEPENDENCY PATH INJECTION ---
+def add_deps_to_path():
+    """Adds the LTX repository directory to the Python system path for imports."""
+    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-Video repository added to sys.path: {repo_path}")
+
+add_deps_to_path()
+
+# --- PROJECT IMPORTS ---
 try:
     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, create_latent_upsampler, create_ltx_video_pipeline)
+    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
 except ImportError as e:
-    print(f"ERRO CRÍTICO: Não foi possível importar o gpu_manager. {e}", file=sys.stderr)
+    logging.critical(f"A crucial LTX import failed. Check LTX-Video repo integrity. Error: {e}")
     sys.exit(1)
 
-# --- Configurações Globais (Lidas do Ambiente) ---
-VINCIE_DIR = Path(os.getenv("VINCIE_DIR", "/data/VINCIE"))
-VINCIE_CKPT_DIR = Path(os.getenv("VINCIE_CKPT_DIR", "/data/ckpt/VINCIE-3B"))
-
-# --- Classe Worker (Gerencia uma única GPU de forma isolada) ---
-class VinceWorker:
-    """
-    Gerencia uma única instância da pipeline VINCIE em um dispositivo GPU específico.
-    Opera em um ambiente "isolado" para garantir que só veja sua própria GPU.
-    """
-    def __init__(self, device_id: str, config_path: str):
-        self.device_id_str = device_id
-        self.gpu_index_str = self.device_id_str.split(':')[-1]
-        self.config_path = config_path
-        self.gen = None
-        self.config = None
-        print(f"[VinceWorker-{self.device_id_str}] Inicializado. Mapeado para o índice de GPU físico {self.gpu_index_str}.")
-
-    def _execute_in_isolated_env(self, function_to_run, *args, **kwargs):
-        """
-        Wrapper crucial que define CUDA_VISIBLE_DEVICES para isolar a visibilidade da GPU.
-        Isso garante que o PyTorch e o VINCIE só possam usar a GPU designada para este worker.
-        """
-        original_cuda_visible = os.environ.get('CUDA_VISIBLE_DEVICES')
-        try:
-            os.environ['CUDA_VISIBLE_DEVICES'] = self.gpu_index_str
-            if torch.cuda.is_available():
-                # Dentro deste contexto, 'cuda:0' refere-se à nossa GPU alvo, pois é a única visível.
-                torch.cuda.set_device(0)
-            return function_to_run(*args, **kwargs)
-        finally:
-            # Restaura o ambiente original para não afetar outros threads/processos.
-            if original_cuda_visible is not None:
-                os.environ['CUDA_VISIBLE_DEVICES'] = original_cuda_visible
-            elif 'CUDA_VISIBLE_DEVICES' in os.environ:
-                del os.environ['CUDA_VISIBLE_DEVICES']
-
-    def _load_model_task(self):
-        """Tarefa de carregamento do modelo, executada no ambiente isolado."""
-        print(f"[VinceWorker-{self.device_id_str}] Carregando modelo para VRAM (GPU física visível: {self.gpu_index_str})...")
-        # O dispositivo para o VINCIE será 'cuda:0' porque é a única GPU que este processo pode ver.
-        device_for_vincie = 'cuda:0' if torch.cuda.is_available() else 'cpu'
-        
-        original_cwd = Path.cwd()
+# ==============================================================================
+# --- UTILITY & HELPER FUNCTIONS ---
+# ==============================================================================
+
+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 calculate_padding(orig_h: int, orig_w: int, target_h: int, target_w: int) -> Tuple[int, int, int, int]:
+    """Calculates symmetric padding values."""
+    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)
+
+def log_tensor_info(tensor: torch.Tensor, name: str = "Tensor"):
+    """Logs detailed debug information about a PyTorch tensor."""
+    if not isinstance(tensor, torch.Tensor):
+        logging.debug(f"'{name}' is not a tensor.")
+        return
+    
+    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:
-            # O código do VINCIE pode precisar ser executado de seu próprio diretório.
-            os.chdir(str(VINCIE_DIR))
-            # Adiciona o diretório ao path do sistema para encontrar os módulos do VINCIE.
-            if str(VINCIE_DIR) not in sys.path: sys.path.insert(0, str(VINCIE_DIR))
+            info_str += (
+                f"  - Min: {tensor.min().item():.4f} | "
+                f"Max: {tensor.max().item():.4f} | "
+                f"Mean: {tensor.mean().item():.4f}\n"
+            )
+        except Exception:
+            pass # Fails on some dtypes
+    logging.debug(info_str + "----------------------")
 
-            from common.config import load_config, create_object
-            
-            cfg = load_config(self.config_path, [f"device='{device_for_vincie}'"])
-            self.gen = create_object(cfg)
-            self.config = cfg
 
-            # Executa os passos de configuração internos do VINCIE.
-            for name in ("configure_persistence", "configure_models", "configure_diffusion"):
-                getattr(self.gen, name)()
+# ==============================================================================
+# --- VIDEO SERVICE CLASS ---
+# ==============================================================================
+
+class VideoService:
+    """Backend service for orchestrating video generation using the LTX-Video pipeline."""
+
+    def __init__(self):
+        """Initializes the service with dedicated GPU logic for main pipeline and VAE."""
+        t0 = time.perf_counter()
+        logging.info("Initializing VideoService...")
+        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.pipeline, self.latent_upsampler = self._load_models()
+
+        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
+        )
+        self._tmp_dirs = set()
+        logging.info(f"VideoService ready. Startup time: {time.perf_counter()-t0:.2f}s")
+
+    # ==========================================================================
+    # --- LIFECYCLE & MODEL MANAGEMENT ---
+    # ==========================================================================
+
+    def _load_config(self) -> Dict:
+        """Loads the YAML configuration file."""
+        config_path = DEFAULT_CONFIG_FILE
+        logging.info(f"Loading config from: {config_path}")
+        with open(config_path, "r") as file:
+            return yaml.safe_load(file)
+
+    def _load_models(self) -> Tuple[LTXMultiScalePipeline, Optional[torch.nn.Module]]:
+        """Loads models from cache to CPU."""
+        t0 = time.perf_counter()
+        logging.info("Loading LTX models from cache to CPU...")
+        
+        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,
+        )
+        
+        latent_upsampler = None
+        if self.config.get("spatial_upscaler_model_path"):
+            spatial_path = self.config["spatial_upscaler_model_path"]
+            latent_upsampler = create_latent_upsampler(spatial_path, device="cpu")
+
+        logging.info(f"Models loaded on CPU in {time.perf_counter()-t0:.2f}s")
+        return pipeline, latent_upsampler
+
+    def move_to_device(self, main_device_str: str, vae_device_str: str):
+        """Moves pipeline components to their 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)
             
-            self.gen.to(torch.device(device_for_vincie))
-            print(f"[VinceWorker-{self.device_id_str}] ✅ Modelo VINCIE 'quente' e pronto na GPU física {self.gpu_index_str}.")
-        finally:
-            os.chdir(original_cwd) # Restaura o diretório de trabalho original.
-    
-    def load_model_to_gpu(self):
-        """Método público para carregar o modelo, garantindo o isolamento da GPU."""
-        if self.gen is None:
-            self._execute_in_isolated_env(self._load_model_task)
-
-    def _infer_task(self, **kwargs) -> Path:
-        """Tarefa de inferência, executada no ambiente isolado."""
-        original_cwd = Path.cwd()
+        logging.info("LTX models successfully moved to target devices.")
+
+    def move_to_cpu(self):
+        """Moves all LTX components to CPU to free VRAM."""
+        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
+
+    # ==========================================================================
+    # --- PUBLIC ORCHESTRATORS ---
+    # ==========================================================================
+
+    def generate_narrative_low(self, prompt: str, **kwargs) -> Tuple[Optional[str], Optional[str], Optional[int]]:
+        """[ORCHESTRATOR] Generates a video from a multi-line prompt (sequence of scenes)."""
+        logging.info("Starting narrative low-res generation...")
+        used_seed = self._resolve_seed(kwargs.get("seed"))
+        seed_everything(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.")
+
+        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)
+        
+        all_latents_paths = []
+        overlap_condition_item = None
+        
         try:
-            os.chdir(str(VINCIE_DIR))
-
-            # Atualiza a configuração do gerador com os parâmetros da chamada atual.
-            with open_dict(self.gen.config):
-                self.gen.config.generation.output.dir = str(kwargs["output_dir"])
-                image_paths = kwargs.get("image_path", [])
-                self.gen.config.generation.positive_prompt.image_path = [str(p) for p in image_paths] if isinstance(image_paths, list) else [str(image_paths)]
-                if "prompts" in kwargs:
-                    self.gen.config.generation.positive_prompt.prompts = list(kwargs["prompts"])
-                if "cfg_scale" in kwargs and kwargs["cfg_scale"] is not None:
-                    self.gen.config.diffusion.cfg.scale = float(kwargs["cfg_scale"])
+            for i, chunk_prompt in enumerate(prompt_list):
+                logging.info(f"Generating narrative chunk {i+1}/{num_chunks}: '{chunk_prompt[:50]}...'")
+
+                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 chunk {i+1}.")
+
+                if 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)
+                all_latents_paths.append(chunk_path)
             
-            # Inicia o loop de inferência do VINCIE.
-            self.gen.inference_loop()
-            return Path(kwargs["output_dir"])
+            return self._finalize_generation(all_latents_paths, "narrative_video", used_seed)
+
+        except Exception as e:
+            logging.error(f"Error during narrative generation: {e}", exc_info=True)
+            return None, None, None
         finally:
-            os.chdir(original_cwd)
-            # Limpeza de memória após a inferência.
-            gc.collect()
-            if torch.cuda.is_available():
-                torch.cuda.empty_cache()
-
-    def infer(self, **kwargs) -> Path:
-        """Método público para iniciar a inferência, garantindo o isolamento da GPU."""
-        if self.gen is None:
-            raise RuntimeError(f"Modelo no worker {self.device_id_str} não foi carregado.")
-        return self._execute_in_isolated_env(self._infer_task, **kwargs)
-
-
-# --- Classe Pool Manager (A Orquestradora Singleton) ---
-class VincePoolManager:
-    _instance = None
-    _lock = threading.Lock()
-
-    def __new__(cls, *args, **kwargs):
-        with cls._lock:
-            if cls._instance is None:
-                cls._instance = super().__new__(cls)
-                cls._instance._initialized = False
-            return cls._instance
-
-    def __init__(self, output_root: str = "/app/outputs"):
-        if self._initialized: return
-        with self._lock:
-            if self._initialized: return
-
-            print("⚙️ Inicializando o VincePoolManager Singleton...")
-            self.output_root = Path(output_root)
-            self.output_root.mkdir(parents=True, exist_ok=True)
-            self.worker_lock = threading.Lock()
-            self.next_worker_idx = 0
-
-            # Pergunta ao gerenciador central quais GPUs ele pode usar.
-            self.allocated_gpu_indices = gpu_manager.get_vincie_devices()
+            for path in all_latents_paths:
+                if path.exists(): path.unlink()
+            self.finalize()
+
+
+    def generate_single_low(self, **kwargs) -> Tuple[Optional[str], Optional[str], Optional[int]]:
+        """[ORCHESTRATOR] Generates a video from a single prompt in one go."""
+        logging.info("Starting single-prompt low-res generation...")
+        used_seed = self._resolve_seed(kwargs.get("seed"))
+        seed_everything(used_seed)
+        
+        try:
+            total_frames = self._calculate_aligned_frames(kwargs.get("duration", 4.0), min_frames=9)
             
-            if not self.allocated_gpu_indices:
-                # Se não houver GPUs alocadas, não podemos continuar.
-                # O setup.py já deve ter sido executado, então não precisamos verificar dependências aqui.
-                print("AVISO: Nenhuma GPU alocada para o VINCIE pelo GPUManager. O serviço VINCIE estará inativo.")
-                self.workers = []
-                self._initialized = True
-                return
-
-            devices = [f'cuda:{i}' for i in self.allocated_gpu_indices]
-            vincie_config_path = VINCIE_DIR / "configs/generate.yaml"
-            if not vincie_config_path.exists():
-                raise FileNotFoundError(f"Arquivo de configuração do VINCIE não encontrado em {vincie_config_path}")
-
-            self.workers = [VinceWorker(dev_id, str(vincie_config_path)) for dev_id in devices]
-
-            print(f"Iniciando carregamento dos modelos em paralelo para {len(self.workers)} GPUs VINCIE...")
-            threads = [threading.Thread(target=worker.load_model_to_gpu) for worker in self.workers]
-            for t in threads: t.start()
-            for t in threads: t.join()
+            final_latents = self._generate_single_chunk_low(
+                num_frames=total_frames,
+                seed=used_seed,
+                conditioning_items=kwargs.get("initial_conditions", []),
+                **kwargs
+            )
             
-            self._initialized = True
-            print(f"✅ VincePoolManager pronto com {len(self.workers)} workers 'quentes'.")
-
-    def _get_next_worker(self) -> VinceWorker:
-        """Seleciona o próximo worker disponível usando uma estratégia round-robin."""
-        if not self.workers:
-            raise RuntimeError("Não há workers VINCIE disponíveis para processar a tarefa.")
-        
-        with self.worker_lock:
-            worker = self.workers[self.next_worker_idx]
-            self.next_worker_idx = (self.next_worker_idx + 1) % len(self.workers)
-            print(f"Tarefa despachada para o worker: {worker.device_id_str}")
-            return worker
-
-    def generate_multi_turn(self, input_image: str, turns: List[str], **kwargs) -> Path:
-        """Gera um vídeo a partir de uma imagem e uma sequência de prompts (turnos)."""
-        worker = self._get_next_worker()
-        out_dir = self.output_root / f"vince_multi_turn_{Path(input_image).stem}_{os.urandom(4).hex()}"
-        out_dir.mkdir(parents=True)
-        
-        infer_kwargs = {"output_dir": out_dir, "image_path": input_image, "prompts": turns, **kwargs}
-        return worker.infer(**infer_kwargs)
+            if final_latents is None: raise RuntimeError("Failed to generate latents.")
+
+            latents_path = RESULTS_DIR / f"temp_single_{used_seed}.pt"
+            torch.save(final_latents.cpu(), latents_path)
+            return self._finalize_generation([latents_path], "single_video", used_seed)
+
+        except Exception as e:
+            logging.error(f"Error during single generation: {e}", exc_info=True)
+            return None, None, None
+        finally:
+            self.finalize()
+
+    # ==========================================================================
+    # --- INTERNAL WORKER & HELPER METHODS ---
+    # ==========================================================================
+
+    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]:
+        """[WORKER] Generates a single chunk of latents. This is the core generation unit."""
+        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
+
+        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 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.main_device).manual_seed(seed),
+            "output_type": "latent", "conditioning_items": 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
+        
+        log_tensor_info(latents_raw, f"Raw Latents for '{prompt[:40]}...'")
+        return latents_raw
+
+    def _finalize_generation(self, latents_paths: List[Path], base_filename: str, seed: int) -> Tuple[str, str, int]:
+        """Loads latents, concatenates, decodes to video, and saves both."""
+        logging.info("Finalizing generation: decoding latents to video.")
+        all_tensors_cpu = [torch.load(p) for p in latents_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}")
+        
+        # The decode method in vae_manager now handles moving the tensor to the correct VAE device.
+        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
+
+    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 []
+        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
+
+    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)
+        # Conditioning tensors are needed on the main device for the transformer pass
+        return tensor.to(self.main_device, dtype=self.runtime_autocast_dtype)
+
+    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)")
+        
+        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) as e:
+                logging.warning(f"Failed to parse custom guidance values: {e}. Falling back to defaults.")
+        
+        if overrides: logging.info(f"Applying '{preset}' guidance preset overrides.")
+        return overrides
+
+    def _save_and_log_video(self, pixel_tensor: torch.Tensor, base_filename: str) -> Path:
+        """Saves a pixel tensor (on CPU) to an MP4 file."""
+        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 generate_multi_concept(self, concept_images: List[str], concept_prompts: List[str], final_prompt: str, **kwargs) -> Path:
-        """Gera um vídeo a partir de múltiplas imagens-conceito e um prompt final."""
-        worker = self._get_next_worker()
-        out_dir = self.output_root / f"vince_multi_concept_{os.urandom(4).hex()}"
-        out_dir.mkdir(parents=True)
-        
-        all_prompts = concept_prompts + [final_prompt]
-        infer_kwargs = {"output_dir": out_dir, "image_path": concept_images, "prompts": all_prompts, **kwargs}
-        return worker.infer(**infer_kwargs)
-
-# --- Instância Singleton Global ---
-# A inicialização é envolvida em um try-except para evitar que a aplicação inteira quebre
-# se o VINCIE não puder ser inicializado por algum motivo.
+    def _apply_precision_policy(self):
+        """Sets the autocast dtype based on the configuration file."""
+        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:
+        """Aligns a dimension to the nearest multiple of `alignment`."""
+        return ((dim - 1) // alignment + 1) * alignment
+    
+    def _calculate_aligned_frames(self, duration_s: float, min_frames: int = 1) -> int:
+        """Calculates total frames based on duration, ensuring alignment."""
+        num_frames = int(round(duration_s * DEFAULT_FPS))
+        aligned_frames = self._align(num_frames)
+        return max(aligned_frames + 1, min_frames)
+
+    def _resolve_seed(self, seed: Optional[int]) -> int:
+        """Returns the given seed or generates a new random one."""
+        return random.randint(0, 2**32 - 1) if seed is None else int(seed)
+
+# ==============================================================================
+# --- SINGLETON INSTANTIATION ---
+# ==============================================================================
 try:
-    output_root_path = os.getenv("OUTPUT_ROOT", "/app/outputs")
-    vince_pool_manager_singleton = VincePoolManager(output_root=output_root_path)
+    video_generation_service = VideoService()
+    logging.info("Global VideoService instance created successfully.")
 except Exception as e:
-    print(f"ERRO CRÍTICO ao inicializar o VincePoolManager: {e}", file=sys.stderr)
-    traceback.print_exc()
-    vince_pool_manager_singleton = None
+    logging.critical(f"Failed to initialize VideoService: {e}", exc_info=True)
+    sys.exit(1)