Update api/ltx_server_refactored_complete.py

api/ltx_server_refactored_complete.py

@@ -1,262 +1,378 @@
-# FILE: app.py
-# DESCRIPTION: Final Gradio web interface for the ADUC-SDR Video Suite.
-# Features dimension sliders locked to multiples of 8, a unified LTX workflow,
-# advanced controls, integrated SeedVR upscaling, and detailed debug logging.
+# FILE: api/ltx_server_refactored_complete.py
+# DESCRIPTION: Final orchestrator for LTX-Video generation.
+# This version includes the fix for the narrative generation overlap bug and
+# consolidates all previous refactoring and debugging improvements.
 
-import gradio as gr
-import traceback
-import sys
-import os
+import gc
+import json
 import logging
+import os
+import shutil
+import sys
+import tempfile
+import time
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+import random
+import torch
+import yaml
+import numpy as np
+from huggingface_hub import hf_hub_download
 
 # ==============================================================================
-# --- IMPORTAÇÃO DOS SERVIÇOS DE BACKEND E UTILS ---
+# --- SETUP E IMPORTAÇÕES DO PROJETO ---
 # ==============================================================================
 
-try:
-    # Serviço principal para geração LTX
-    from api.ltx_server_refactored_complete import video_generation_service
-    
-    # Nosso decorador de logging para depuração
-    from api.utils.debug_utils import log_function_io
+# 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')
 
-    # Serviço especialista para upscaling de resolução (SeedVR)
-    from api.seedvr_server import seedvr_server_singleton as seedvr_inference_server
-    
-    logging.info("All backend services (LTX, SeedVR) and debug utils imported successfully.")
+# --- 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
+def add_deps_to_path():
+    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}")
 
+add_deps_to_path()
+
+# --- Módulos da nossa Arquitetura ---
+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.ltx.ltx_utils import (
+        build_ltx_pipeline_on_cpu,
+        seed_everything,
+        load_image_to_tensor_with_resize_and_crop,
+        ConditioningItem,
+    )
+    from api.utils.debug_utils import log_function_io
 except ImportError as e:
-    def log_function_io(func): return func
-    logging.warning(f"Could not import a module, debug logger might be disabled. SeedVR might be unavailable. Details: {e}")
-    if 'video_generation_service' not in locals():
-        logging.critical(f"FATAL: Main LTX service failed to import.", exc_info=True)
-        sys.exit(1)
-    if 'seedvr_inference_server' not in locals():
-        seedvr_inference_server = None
-        logging.warning("SeedVR server could not be initialized. The SeedVR upscaling tab will be disabled.")
-except Exception as e:
-    logging.critical(f"FATAL ERROR: An unexpected error occurred during backend initialization. Details: {e}", exc_info=True)
+    logging.critical(f"A crucial import from the local API/architecture failed. Error: {e}", exc_info=True)
     sys.exit(1)
 
 # ==============================================================================
-# --- FUNÇÕES WRAPPER (PONTE ENTRE UI E BACKEND) ---
+# --- FUNÇÕES AUXILIARES DO ORQUESTRADOR ---
 # ==============================================================================
 
 @log_function_io
-def run_generate_base_video(
-    generation_mode: str, prompt: str, neg_prompt: str, start_img: str, 
-    height: int, width: int, duration: float,
-    fp_guidance_preset: str, fp_guidance_scale_list: str, fp_stg_scale_list: str,
-    fp_num_inference_steps: int, fp_skip_initial_steps: int, fp_skip_final_steps: int,
-    progress=gr.Progress(track_tqdm=True)
-) -> tuple:
-    """Wrapper para a geração do vídeo base LTX."""
-    try:
-        logging.info(f"[UI] Request received. Selected mode: {generation_mode}")
-        
-        initial_conditions = []
-        if start_img:
-            num_frames_estimate = int(duration * 24)
-            items_list = [[start_img, 0, 1.0]]
-            initial_conditions = video_generation_service.prepare_condition_items(
-                items_list, height, width, num_frames_estimate
-            )
-
-        ltx_configs = {
-            "guidance_preset": fp_guidance_preset,
-            "guidance_scale_list": fp_guidance_scale_list,
-            "stg_scale_list": fp_stg_scale_list,
-            "num_inference_steps": fp_num_inference_steps,
-            "skip_initial_inference_steps": fp_skip_initial_steps,
-            "skip_final_inference_steps": fp_skip_final_steps,
-        }
+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)
 
-        video_path, tensor_path, final_seed = video_generation_service.generate_low_resolution(
-            prompt=prompt, negative_prompt=neg_prompt,
-            height=height, width=width, duration=duration,
-            initial_conditions=initial_conditions, ltx_configs_override=ltx_configs
-        )
+# ==============================================================================
+# --- CLASSE DE SERVIÇO (O ORQUESTRADOR) ---
+# ==============================================================================
+
+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.")
         
-        if not video_path: raise RuntimeError("Backend failed to return a valid video path.")
+        is_narrative = len(prompt_list) > 1
+        logging.info(f"Generation mode detected: {'Narrative' if is_narrative else 'Simple'} ({len(prompt_list)} scene(s)).")
 
-        new_state = {"low_res_video": video_path, "low_res_latents": tensor_path, "used_seed": final_seed}
-        logging.info(f"[UI] Base video generation successful. Seed used: {final_seed}, Path: {video_path}")
-        return video_path, new_state, gr.update(visible=True)
+        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 must be N*8+1 frames. 9 is the smallest practical value.
+        overlap_frames = 9 if is_narrative else 0
+        if is_narrative:
+            logging.info(f"Narrative mode: Using overlap of {overlap_frames} frames between chunks.")
+        
+        temp_latent_paths = []
+        overlap_condition_item = None
         
-    except Exception as e:
-        error_message = f"❌ An error occurred during base generation:\n{e}"
-        logging.error(f"{error_message}\nDetails: {traceback.format_exc()}", exc_info=True)
-        raise gr.Error(error_message)
+        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:
+                    current_frames_base = frames_per_chunk
+                else: # Last chunk takes all remaining frames
+                    processed_frames_base = (num_chunks - 1) * frames_per_chunk
+                    current_frames_base = total_frames - processed_frames_base
+                
+                current_frames = current_frames_base + (overlap_frames if i > 0 else 0)
+                # Ensure final frame count for generation is N*8+1
+                current_frames = self._align(current_frames, alignment_rule='n*8+1')
 
-@log_function_io
-def run_ltx_refinement(state: dict, prompt: str, neg_prompt: str, progress=gr.Progress(track_tqdm=True)) -> tuple:
-    """Wrapper para o refinamento de textura LTX."""
-    if not state or not state.get("low_res_latents"):
-        raise gr.Error("Error: Please generate a base video in Step 1 before refining.")
-    
-    try:
-        logging.info(f"[UI] Requesting LTX refinement for latents: {state.get('low_res_latents')}")
-        video_path, tensor_path = video_generation_service.generate_upscale_denoise(
-            latents_path=state["low_res_latents"],
-            prompt=prompt,
-            negative_prompt=neg_prompt,
-            seed=state["used_seed"]
-        )
-        state["refined_video_ltx"] = video_path
-        state["refined_latents_ltx"] = tensor_path
-        logging.info(f"[UI] LTX refinement successful. Path: {video_path}")
-        return video_path, state
-    except Exception as e:
-        error_message = f"❌ An error occurred during LTX Refinement:\n{e}"
-        logging.error(f"{error_message}\nDetails: {traceback.format_exc()}", exc_info=True)
-        raise gr.Error(error_message)
+                current_conditions = kwargs.get("initial_conditions", []) if i == 0 else []
+                if overlap_condition_item: current_conditions.append(overlap_condition_item)
 
-@log_function_io
-def run_seedvr_upscaling(state: dict, seed: int, resolution: int, batch_size: int, fps: int, progress=gr.Progress(track_tqdm=True)) -> tuple:
-    """Wrapper para o upscale de resolução SeedVR."""
-    if not state or not state.get("low_res_video"):
-        raise gr.Error("Error: Please generate a base video in Step 1 before upscaling.")
-    if not seedvr_inference_server:
-        raise gr.Error("Error: The SeedVR upscaling server is not available.")
-    
-    try:
-        logging.info(f"[UI] Requesting SeedVR upscaling for video: {state.get('low_res_video')}")
-        def progress_wrapper(p, desc=""): progress(p, desc=desc)
+                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
+        }
         
-        output_filepath = seedvr_inference_server.run_inference(
-            file_path=state["low_res_video"], seed=int(seed), resolution=int(resolution),
-            batch_size=int(batch_size), fps=float(fps), progress=progress_wrapper
-        )
+        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
         
-        status_message = f"✅ Upscaling complete!\nSaved to: {output_filepath}"
-        logging.info(f"[UI] SeedVR upscaling successful. Path: {output_filepath}")
-        return gr.update(value=output_filepath), gr.update(value=status_message)
-    except Exception as e:
-        error_message = f"❌ An error occurred during SeedVR Upscaling:\n{e}"
-        logging.error(f"{error_message}\nDetails: {traceback.format_exc()}", exc_info=True)
-        return None, gr.update(value=error_message)
+        return latents_raw.to(self.main_device)
 
-# ==============================================================================
-# --- CONSTRUÇÃO DA INTERFACE GRADIO ---
-# ==============================================================================
+    @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
 
-def build_ui():
-    """Constrói a interface completa do Gradio."""
-    with gr.Blocks(theme=gr.themes.Soft(primary_hue="indigo")) as demo:
-        app_state = gr.State(value={"low_res_video": None, "low_res_latents": None, "used_seed": None})
-        ui_components = {}
-        gr.Markdown("# ADUC-SDR Video Suite - LTX & SeedVR Workflow", elem_id="main-title")
-        with gr.Row():
-            with gr.Column(scale=1): _build_generation_controls(ui_components)
-            with gr.Column(scale=1):
-                gr.Markdown("### Etapa 1: Vídeo Base Gerado")
-                ui_components['low_res_video_output'] = gr.Video(label="O resultado aparecerá aqui", interactive=False)
-                ui_components['used_seed_display'] = gr.Textbox(label="Seed Utilizada", interactive=False)
-        _build_postprod_controls(ui_components)
-        _register_event_handlers(app_state, ui_components)
-    return demo
-
-def _build_generation_controls(ui: dict):
-    """Constrói os componentes da UI para a Etapa 1: Geração."""
-    gr.Markdown("### Configurações de Geração")
-    ui['generation_mode'] = gr.Radio(label="Modo de Geração", choices=["Simples (Prompt Único)", "Narrativa (Múltiplos Prompts)"], value="Narrativa (Múltiplos Prompts)", info="Simples para uma ação contínua, Narrativa para uma sequência (uma cena por linha).")
-    ui['prompt'] = gr.Textbox(label="Prompt(s)", value="Um leão majestoso caminha pela savana\nEle sobe em uma grande pedra e olha o horizonte", lines=4)
-    ui['neg_prompt'] = gr.Textbox(label="Negative Prompt", value="blurry, low quality, bad anatomy, deformed", lines=2)
-    ui['start_image'] = gr.Image(label="Imagem de Início (Opcional)", type="filepath", sources=["upload"])
-    
-    with gr.Accordion("Parâmetros Principais", open=True):
-        ui['duration'] = gr.Slider(label="Duração Total (s)", value=4, step=1, minimum=1, maximum=30)
-        with gr.Row():
-            ui['height'] = gr.Slider(label="Height", value=432, step=8, minimum=256, maximum=1024)
-            ui['width'] = gr.Slider(label="Width", value=768, step=8, minimum=256, maximum=1024)
-
-    with gr.Accordion("Opções Avançadas LTX", open=False):
-        gr.Markdown("#### Configurações de Passos de Inferência (First Pass)")
-        gr.Markdown("*Deixe o valor padrão (ex: 20) ou 0 para usar a configuração do `config.yaml`.*")
-        ui['fp_num_inference_steps'] = gr.Slider(label="Número de Passos", minimum=0, maximum=100, step=1, value=20, info="Padrão LTX: 20.")
-        ui['fp_skip_initial_steps'] = gr.Slider(label="Pular Passos Iniciais", minimum=0, maximum=100, step=1, value=0)
-        ui['fp_skip_final_steps'] = gr.Slider(label="Pular Passos Finais", minimum=0, maximum=100, step=1, value=0)
-        with gr.Tabs():
-            with gr.TabItem("Configurações de Guiagem (First Pass)"):
-                ui['fp_guidance_preset'] = gr.Dropdown(label="Preset de Guiagem", choices=["Padrão (Recomendado)", "Agressivo", "Suave", "Customizado"], value="Padrão (Recomendado)", info="Controla o comportamento da guiagem durante a difusão.")
-                with gr.Group(visible=False) as ui['custom_guidance_group']:
-                    gr.Markdown("⚠️ Edite as listas em formato JSON. Ex: `[1.0, 2.5, 3.0]`")
-                    ui['fp_guidance_scale_list'] = gr.Textbox(label="Lista de Guidance Scale", value="[1, 1, 6, 8, 6, 1, 1]")
-                    ui['fp_stg_scale_list'] = gr.Textbox(label="Lista de STG Scale (Movimento)", value="[0, 0, 4, 4, 4, 2, 1]")
-    
-    ui['generate_low_btn'] = gr.Button("1. Gerar Vídeo Base", variant="primary")
-
-def _build_postprod_controls(ui: dict):
-    """Constrói os componentes da UI para a Etapa 2: Pós-Produção."""
-    with gr.Group(visible=False) as ui['post_prod_group']:
-        gr.Markdown("--- \n## Etapa 2: Pós-Produção")
-        with gr.Tabs():
-            with gr.TabItem("🚀 Upscaler de Textura (LTX)"):
-                with gr.Row():
-                    with gr.Column(scale=1):
-                         gr.Markdown("Usa o prompt e a semente originais para refinar o vídeo, adicionando detalhes e texturas de alta qualidade.")
-                         ui['ltx_refine_btn'] = gr.Button("2. Aplicar Refinamento LTX", variant="primary")
-                    with gr.Column(scale=1):
-                        ui['ltx_refined_video_output'] = gr.Video(label="Vídeo com Textura Refinada", interactive=False)
+    @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
             
-            with gr.TabItem("✨ Upscaler de Resolução (SeedVR)"):
-                is_seedvr_available = seedvr_inference_server is not None
-                if not is_seedvr_available:
-                    gr.Markdown("🔴 **AVISO: O serviço SeedVR não está disponível.**")
-                with gr.Row():
-                    with gr.Column(scale=1):
-                        ui['seedvr_seed'] = gr.Slider(minimum=0, maximum=999999, value=42, step=1, label="Seed")
-                        ui['seedvr_resolution'] = gr.Slider(minimum=720, maximum=2160, value=1080, step=8, label="Resolução Vertical Alvo")
-                        ui['seedvr_batch_size'] = gr.Slider(minimum=1, maximum=16, value=4, step=1, label="Batch Size por GPU")
-                        ui['seedvr_fps'] = gr.Number(label="FPS de Saída (0 = original)", value=0)
-                        ui['run_seedvr_btn'] = gr.Button("2. Iniciar Upscaling SeedVR", variant="primary", interactive=is_seedvr_available)
-                    with gr.Column(scale=1):
-                        ui['seedvr_video_output'] = gr.Video(label="Vídeo com Upscale SeedVR", interactive=False)
-                        ui['seedvr_status_box'] = gr.Textbox(label="Status do SeedVR", value="Aguardando...", lines=3, interactive=False)
-
-def _register_event_handlers(app_state: gr.State, ui: dict):
-    """Registra todos os manipuladores de eventos do Gradio."""
-    def toggle_custom_guidance(preset_choice: str) -> gr.update:
-        return gr.update(visible=(preset_choice == "Customizado"))
-    
-    ui['fp_guidance_preset'].change(fn=toggle_custom_guidance, inputs=ui['fp_guidance_preset'], outputs=ui['custom_guidance_group'])
-
-    def update_seed_display(state):
-        return state.get("used_seed", "N/A")
-
-    gen_inputs = [
-        ui['generation_mode'], ui['prompt'], ui['neg_prompt'], ui['start_image'],
-        ui['height'], ui['width'], ui['duration'],
-        ui['fp_guidance_preset'], ui['fp_guidance_scale_list'], ui['fp_stg_scale_list'],
-        ui['fp_num_inference_steps'], ui['fp_skip_initial_steps'], ui['fp_skip_final_steps'],
-    ]
-    gen_outputs = [ui['low_res_video_output'], app_state, ui['post_prod_group']]
+            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
     
-    (ui['generate_low_btn'].click(fn=run_generate_base_video, inputs=gen_inputs, outputs=gen_outputs)
-     .then(fn=update_seed_display, inputs=[app_state], outputs=[ui['used_seed_display']]))
+    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}")
 
-    refine_inputs = [app_state, ui['prompt'], ui['neg_prompt']]
-    refine_outputs = [ui['ltx_refined_video_output'], app_state]
-    ui['ltx_refine_btn'].click(fn=run_ltx_refinement, inputs=refine_inputs, outputs=refine_outputs)
+    def _align(self, dim: int, alignment: int = FRAMES_ALIGNMENT, alignment_rule: str = 'default') -> int:
+        """Aligns a dimension to the nearest multiple of `alignment`."""
+        if alignment_rule == 'n*8+1':
+             return ((dim - 1) // alignment) * alignment + 1
+        return ((dim - 1) // alignment + 1) * alignment
     
-    if 'run_seedvr_btn' in ui and ui['run_seedvr_btn'].interactive:
-        seedvr_inputs = [app_state, ui['seedvr_seed'], ui['seedvr_resolution'], ui['seedvr_batch_size'], ui['seedvr_fps']]
-        seedvr_outputs = [ui['seedvr_video_output'], ui['seedvr_status_box']]
-        ui['run_seedvr_btn'].click(fn=run_seedvr_upscaling, inputs=seedvr_inputs, outputs=seedvr_outputs)
+    def _calculate_aligned_frames(self, duration_s: float, min_frames: int = 1) -> int:
+        num_frames = int(round(duration_s * DEFAULT_FPS))
+        # Para a duração total, sempre arredondamos para cima para o múltiplo de 8 mais próximo
+        aligned_frames = self._align(num_frames, alignment=FRAMES_ALIGNMENT)
+        return max(aligned_frames, min_frames)
+
+    def _get_random_seed(self) -> int:
+        """Always generates and returns a new random seed."""
+        return random.randint(0, 2**32 - 1)
 
 # ==============================================================================
-# --- PONTO DE ENTRADA DA APLICAÇÃO ---
+# --- INSTANCIAÇÃO SINGLETON ---
 # ==============================================================================
-
-if __name__ == "__main__":
-    log_level = os.environ.get("ADUC_LOG_LEVEL", "INFO").upper()
-    logging.basicConfig(level=log_level, format='[%(levelname)s] [%(name)s] %(message)s')
-    
-    print("Building Gradio UI...")
-    gradio_app = build_ui()
-    print("Launching Gradio app...")
-    gradio_app.queue().launch(
-        server_name=os.getenv("GRADIO_SERVER_NAME", "0.0.0.0"), 
-        server_port=int(os.getenv("GRADIO_SERVER_PORT", "7860")),
-        show_error=True
-    )
+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)