api/ltx/ltx_aduc_pipeline.py

# FILE: api/ltx/ltx_aduc_pipeline.py
# DESCRIPTION: Final orchestrator for LTX-Video generation.
# This version internalizes conditioning item preparation, accepting a raw
# list of media items directly in its main generation function for maximum simplicity and encapsulation.

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, Union

import torch
import yaml
import numpy as np
from PIL import Image
from huggingface_hub import hf_hub_download

# ==============================================================================
# --- SETUP E IMPORTAÇÕES DO PROJETO ---
# ==============================================================================

# Configuração de logging e supressão de warnings
import warnings
warnings.filterwarnings("ignore")
logging.getLogger("huggingface_hub").setLevel(logging.ERROR)
log_level = os.environ.get("ADUC_LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=log_level, format='[%(levelname)s] [%(name)s] %(message)s')

# --- Constantes de Configuração ---
DEPS_DIR = Path("/data")
LTX_VIDEO_REPO_DIR = DEPS_DIR / "LTX-Video"
RESULTS_DIR = Path("/app/output")
DEFAULT_FPS = 24.0
FRAMES_ALIGNMENT = 8
LTX_REPO_ID = "Lightricks/LTX-Video"


# --- Módulos da nossa Arquitetura ---
try:
    from managers.gpu_manager import gpu_manager
    from api.ltx.vae_aduc_pipeline import vae_ltx_aduc_pipeline
    from tools.video_encode_tool import video_encode_tool_singleton
    from api.ltx.ltx_utils import build_ltx_pipeline_on_cpu, seed_everything
    from api.ltx.ltx_aduc_manager import LatentConditioningItem, ltx_aduc_manager
    from utils.debug_utils import log_function_io
except ImportError as e:
    logging.critical(f"A crucial import from the local API/architecture failed. Error: {e}", exc_info=True)
    sys.exit(1)

# ==============================================================================
# --- CLASSE DE SERVIÇO (O ORQUESTRADOR) ---
# ==============================================================================


class LtxAducPipeline:
    """
    Orchestrates the high-level logic of video generation, with internalized
    conditioning item preparation.
    """

    @log_function_io
    def __init__(self):
        t0 = time.time()
        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.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()
        logging.info(f"VideoService ready. Startup time: {time.time() - 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"
        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
            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
        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)
        self.main_device = target_main_device
        self.vae_device = target_vae_device
        self.pipeline.to(self.main_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_list: List[str],
        initial_media_items: Optional[List[Tuple[Union[str, Image.Image, torch.Tensor], int, float]]] = None,
        **kwargs
    ) -> Tuple[Optional[str], Optional[str], Optional[int]]:
        """
        [UNIFIED ORCHESTRATOR] Generates a low-resolution video from a prompt and a raw list of media items.
        """
        logging.info("Starting unified low-resolution generation...")
        used_seed = self._get_random_seed()
        seed_everything(used_seed)
        logging.info(f"Using randomly generated seed: {used_seed}")

        if not prompt_list: raise ValueError("Prompt is empty or contains no valid lines.")
        
        is_narrative = len(prompt_list) > 1
        num_chunks = len(prompt_list)
        total_frames = self._calculate_aligned_frames(kwargs.get("duration", 4.0))
        frames_per_chunk = max(FRAMES_ALIGNMENT, (total_frames // num_chunks // FRAMES_ALIGNMENT) * FRAMES_ALIGNMENT)
        overlap_frames = 9 if is_narrative else 0
        
        initial_conditions = []
        if initial_media_items:
            logging.info("Preparing initial conditioning items from raw media list...")
            initial_conditions = vae_ltx_aduc_pipeline.generate_conditioning_items(
                media_items=[item[0] for item in initial_media_items],
                target_frames=[item[1] for item in initial_media_items],
                strengths=[item[2] for item in initial_media_items],
                target_resolution=(kwargs['height'], kwargs['width'])
            )
        
        temp_latent_paths = []
        overlap_condition_item: Optional[LatentConditioningItem] = None
        
        try:
            for i, chunk_prompt in enumerate(prompt_list):
                logging.info(f"Processing scene {i+1}/{num_chunks}: '{chunk_prompt[:50]}...'")
                
                if i < num_chunks - 1:
                    current_frames_base = frames_per_chunk
                else:
                    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)
                current_frames = self._align(current_frames, alignment_rule='n*8+1')

                current_conditions = initial_conditions if i == 0 else []
                if overlap_condition_item: current_conditions.append(overlap_condition_item)

                chunk_latents = self._generate_single_chunk_low(
                    prompt=chunk_prompt, num_frames=current_frames, seed=used_seed + i,
                    conditioning_items=current_conditions, **kwargs
                )
                if chunk_latents is None: raise RuntimeError(f"Failed to generate latents for scene {i+1}.")

                if is_narrative and i < num_chunks - 1:
                    overlap_latents = chunk_latents[:, :, -overlap_frames:, :, :].clone()
                    overlap_condition_item = LatentConditioningItem(
                        latent_tensor=overlap_latents.cpu(),
                        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"
            all_tensors_cpu = [torch.load(p) for p in temp_latent_paths]
            final_latents = torch.cat(all_tensors_cpu, dim=2)

            video_path, latents_path = self._finalize_generation(final_latents, base_filename, used_seed)
            return video_path, latents_path, 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 ---
    # ==========================================================================
    
    def _log_conditioning_items(self, items: List[LatentConditioningItem]):
        """
        Logs detailed information about a list of ConditioningItem objects.
        This is a dedicated debug helper function.
        """
        # Só imprime o log se o nível de logging for DEBUG
        if logging.getLogger().isEnabledFor(logging.INFO):
            log_str = ["\n" + "="*25 + " INFO: Conditioning Items " + "="*25]
            if not items:
                log_str.append("  -> Lista de conditioning_items está vazia.")
            else:
                for i, item in enumerate(items):
                    if hasattr(item, 'media_item') and isinstance(item.media_item, torch.Tensor):
                        t = item.media_item
                        log_str.append(
                            f"  -> Item [{i}]: "
                            f"Tensor(shape={list(t.shape)}, "
                            f"device='{t.device}', "
                            f"dtype={t.dtype}), "
                            f"Target Frame = {item.media_frame_number}, "
                            f"Strength = {item.conditioning_strength:.2f}"
                        )
                    else:
                        log_str.append(f"  -> Item [{i}]: Não contém um tensor válido.")
            log_str.append("="*75 + "\n")
            
            # Usa o logger de debug para imprimir a mensagem completa
            logging.info("\n".join(log_str))

    @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)

        # 1. Começa com a configuração padrão
        first_pass_config = self.config.get("first_pass", {}).copy()
        
        # 2. Aplica os overrides da UI, se existirem
        if kwargs.get("ltx_configs_override"):
            self._apply_ui_overrides(first_pass_config, kwargs.get("ltx_configs_override"))

        # 3. Monta o dicionário de argumentos SEM conditioning_items primeiro
        pipeline_kwargs = {
            "prompt": kwargs['prompt'],
            "negative_prompt": kwargs['negative_prompt'],
            "height": downscaled_height,
            "width": downscaled_width,
            "num_frames": kwargs['num_frames'],
            "frame_rate": int(DEFAULT_FPS),
            "generator": torch.Generator(device=self.main_device).manual_seed(kwargs['seed']),
            "output_type": "latent",
            #"conditioning_items": conditioning_items if conditioning_items else None,
            "media_items": None,
            "decode_timestep": self.config["decode_timestep"],
            "decode_noise_scale": self.config["decode_noise_scale"],
            "stochastic_sampling": self.config["stochastic_sampling"],
            "image_cond_noise_scale": 0.01,
            "is_video": True,
            "vae_per_channel_normalize": True,
            "mixed_precision": (self.config["precision"] == "mixed_precision"),
            "offload_to_cpu": False,
            "enhance_prompt": False,
            #"skip_layer_strategy": SkipLayerStrategy.AttentionValues,
            **first_pass_config
        }
        
        # --- Bloco de Logging para Depuração ---
        # 4. Loga os argumentos do pipeline (sem os tensores de condição)
        logging.info(f"\n[Info] Pipeline Arguments (BASE):\n {json.dumps(pipeline_kwargs, indent=2, default=str)}\n")
        
        # Loga os conditioning_items separadamente com a nossa função helper
        conditioning_items_list = kwargs.get('conditioning_items')
        self._log_conditioning_items(conditioning_items_list)
        # --- Fim do Bloco de Logging ---

        # 5. Adiciona os conditioning_items ao dicionário
        pipeline_kwargs['conditioning_items'] = conditioning_items_list
        
        # 6. Executa o pipeline com o dicionário completo
        with torch.autocast(device_type=self.main_device.type, dtype=self.runtime_autocast_dtype, enabled="cuda" in self.main_device.type):
            latents_raw = self.pipeline(**pipeline_kwargs).images
        
        return latents_raw.to(self.main_device)

    @log_function_io
    def _finalize_generation(self, final_latents: torch.Tensor, base_filename: str, seed: int) -> Tuple[str, str]:
        """Consolidates latents, decodes them to video, and saves final artifacts."""
        logging.info("Finalizing generation: decoding latents to video.")
        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_ltx_aduc_pipeline.decode_to_pixels(
            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)    

    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.")
        
    def _save_and_log_video(self, pixel_tensor: torch.Tensor, base_filename: str) -> Path:
        with tempfile.TemporaryDirectory() as temp_dir:
            temp_path = os.path.join(temp_dir, f"{base_filename}.mp4")
            video_encode_tool_singleton.save_video_from_tensor(pixel_tensor, temp_path, fps=DEFAULT_FPS)
            final_path = RESULTS_DIR / f"{base_filename}.mp4"
            shutil.move(temp_path, final_path)
            logging.info(f"Video saved successfully to: {final_path}")
            return final_path
    
    def _apply_precision_policy(self):
        precision = str(self.config.get("precision", "bfloat16")).lower()
        if precision in ["float8_e4m3fn", "bfloat16"]: self.runtime_autocast_dtype = torch.bfloat16
        elif precision == "mixed_precision": self.runtime_autocast_dtype = torch.float16
        else: self.runtime_autocast_dtype = torch.float32
        logging.info(f"Runtime precision policy set for autocast: {self.runtime_autocast_dtype}")

    def _align(self, dim: int, alignment: int = FRAMES_ALIGNMENT, 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
    
    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)

# ==============================================================================
# --- INSTANCIAÇÃO SINGLETON ---
# ==============================================================================
try:
    ltx_aduc_pipeline = LtxAducPipeline()
    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)