app(withdistilledcheckpoint).py

# =============================================================================
# Installation and Setup
# =============================================================================
import os
import subprocess
import sys

os.environ["TORCH_COMPILE_DISABLE"] = "1"
os.environ["TORCHDYNAMO_DISABLE"] = "1"

subprocess.run([sys.executable, "-m", "pip", "install", "xformers==0.0.32.post2", "--no-build-isolation"], check=False)

LTX_REPO_URL = "https://github.com/Lightricks/LTX-2.git"
LTX_REPO_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "LTX-2")
LTX_COMMIT = "ae855f8538843825f9015a419cf4ba5edaf5eec2"

if not os.path.exists(LTX_REPO_DIR):
    print(f"Cloning {LTX_REPO_URL}...")
    subprocess.run(["git", "clone", LTX_REPO_URL, LTX_REPO_DIR], check=True)
    subprocess.run(["git", "checkout", LTX_COMMIT], cwd=LTX_REPO_DIR, check=True)

print("Installing ltx-core and ltx-pipelines from cloned repo...")
subprocess.run(
    [sys.executable, "-m", "pip", "install", "--force-reinstall", "--no-deps", "-e",
     os.path.join(LTX_REPO_DIR, "packages", "ltx-core"),
     "-e", os.path.join(LTX_REPO_DIR, "packages", "ltx-pipelines")],
    check=True,
)

sys.path.insert(0, os.path.join(LTX_REPO_DIR, "packages", "ltx-pipelines", "src"))
sys.path.insert(0, os.path.join(LTX_REPO_DIR, "packages", "ltx-core", "src"))

# =============================================================================
# Imports
# =============================================================================
import logging
import random
import tempfile
from pathlib import Path
import gc
import hashlib

import torch
torch._dynamo.config.suppress_errors = True
torch._dynamo.config.disable = True

import spaces
import gradio as gr
import numpy as np
from huggingface_hub import hf_hub_download, snapshot_download
from safetensors.torch import load_file, save_file

from ltx_core.model.video_vae import TilingConfig, get_video_chunks_number
from ltx_core.model.audio_vae import decode_audio as vae_decode_audio
from ltx_core.model.video_vae import decode_video as vae_decode_video
from ltx_core.model.upsampler import upsample_video
from ltx_core.quantization import QuantizationPolicy
from ltx_core.loader import LoraPathStrengthAndSDOps, LTXV_LORA_COMFY_RENAMING_MAP
from ltx_core.components.guiders import MultiModalGuider, MultiModalGuiderParams
from ltx_core.components.noisers import GaussianNoiser
from ltx_core.components.diffusion_steps import Res2sDiffusionStep
from ltx_core.components.schedulers import LTX2Scheduler
from ltx_core.types import Audio, LatentState, VideoPixelShape, AudioLatentShape
from ltx_core.tools import VideoLatentShape

from ltx_pipelines.ti2vid_two_stages_hq import TI2VidTwoStagesHQPipeline
from ltx_pipelines.utils.args import ImageConditioningInput
from ltx_pipelines.utils.constants import LTX_2_3_HQ_PARAMS, STAGE_2_DISTILLED_SIGMA_VALUES
from ltx_pipelines.utils.media_io import encode_video
from ltx_pipelines.utils.helpers import (
    assert_resolution,
    cleanup_memory,
    combined_image_conditionings,
    encode_prompts,
    multi_modal_guider_denoising_func,
    simple_denoising_func,
    denoise_audio_video,
)

from ltx_pipelines.utils import res2s_audio_video_denoising_loop

# Patch xformers
try:
    from ltx_core.model.transformer import attention as _attn_mod
    from xformers.ops import memory_efficient_attention as _mea
    _attn_mod.memory_efficient_attention = _mea
    print("[ATTN] xformers patch applied")
except Exception as e:
    print(f"[ATTN] xformers patch failed: {e}")

logging.getLogger().setLevel(logging.INFO)

MAX_SEED = np.iinfo(np.int32).max
DEFAULT_PROMPT = (
    "A majestic eagle soaring over mountain peaks at sunset, "
    "wings spread wide against the orange sky, feathers catching the light, "
    "wind currents visible in the motion blur, cinematic slow motion, 4K quality"
)
DEFAULT_NEGATIVE_PROMPT = (
    "worst quality, inconsistent motion, blurry, jittery, distorted, "
    "deformed, artifacts, text, watermark, logo, frame, border, "
    "low resolution, pixelated, unnatural, fake, CGI, cartoon"
)
DEFAULT_FRAME_RATE = 24.0
MIN_DIM, MAX_DIM, STEP = 256, 1280, 64
MIN_FRAMES, MAX_FRAMES = 9, 721

# Resolution presets with high/low tiers
RESOLUTIONS = {
    "high": {"16:9": (1536, 1024), "9:16": (1024, 1536), "1:1": (1024, 1024)},
    "low": {"16:9": (768, 512), "9:16": (512, 768), "1:1": (768, 768)},
}

LTX_MODEL_REPO = "Lightricks/LTX-2.3"
GEMMA_REPO = "Lightricks/gemma-3-12b-it-qat-q4_0-unquantized"

# =============================================================================
# Custom HQ Pipeline with LoRA Cache Support
# =============================================================================

class HQPipelineWithCachedLoRA:
    """
    Custom HQ pipeline that:
    1. Creates ONE ModelLedger WITHOUT LoRAs
    2. Handles ALL LoRAs via cached state (distilled + 12 custom)
    3. Supports CFG/negative prompts and guidance parameters
    4. Reuses single transformer for both stages
    5. Uses 8 steps at half resolution + 3 steps at full resolution
    """
    
    def __init__(
        self,
        checkpoint_path: str,
        spatial_upsampler_path: str,
        gemma_root: str,
        quantization: QuantizationPolicy | None = None,
    ):
        from ltx_pipelines.utils import ModelLedger
        from ltx_pipelines.utils.types import PipelineComponents
        
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.dtype = torch.bfloat16
        
        print("    Creating ModelLedger (no LoRAs)...")
        self.model_ledger = ModelLedger(
            dtype=self.dtype,
            device=self.device,
            checkpoint_path=checkpoint_path,
            gemma_root_path=gemma_root,
            spatial_upsampler_path=spatial_upsampler_path,
            loras=(),
            quantization=quantization,
        )
        
        self.pipeline_components = PipelineComponents(
            dtype=self.dtype,
            device=self.device,
        )
        
        self._cached_state = None
    
    def apply_cached_lora_state(self, state_dict):
        """Apply pre-cached LoRA state to transformer."""
        self._cached_state = state_dict
    
    @torch.inference_mode()
    def __call__(  # noqa: PLR0913
        self,
        prompt: str,
        negative_prompt: str,
        seed: int,
        height: int,
        width: int,
        num_frames: int,
        frame_rate: float,
        video_guider_params: MultiModalGuiderParams,
        audio_guider_params: MultiModalGuiderParams,
        images: list,
        tiling_config: TilingConfig | None = None,
    ):
        from ltx_pipelines.utils import assert_resolution, cleanup_memory, combined_image_conditionings, encode_prompts, res2s_audio_video_denoising_loop, multi_modal_guider_denoising_func, simple_denoising_func, denoise_audio_video
        from ltx_core.tools import VideoLatentShape
        from ltx_core.components.noisers import GaussianNoiser
        from ltx_core.components.diffusion_steps import Res2sDiffusionStep
        from ltx_core.types import VideoPixelShape
        from ltx_core.model.upsampler import upsample_video
        from ltx_core.model.video_vae import decode_video as vae_decode_video
        from ltx_core.model.audio_vae import decode_audio as vae_decode_audio
        
        assert_resolution(height=height, width=width, is_two_stage=True)
        
        device = self.device
        dtype = self.dtype
        generator = torch.Generator(device=device).manual_seed(seed)
        noiser = GaussianNoiser(generator=generator)
        
        # NO LoRA application here - done in apply_prepared_lora_state_to_pipeline()
        
        ctx_p, ctx_n = encode_prompts(
            [prompt, negative_prompt],
            self.model_ledger,
        )
        
        v_context_p, a_context_p = ctx_p.video_encoding, ctx_p.audio_encoding
        v_context_n, a_context_n = ctx_n.video_encoding, ctx_n.audio_encoding
        
        # ===================== STAGE 1: 8 steps at half resolution =====================
        stage_1_output_shape = VideoPixelShape(
            batch=1, frames=num_frames,
            width=width // 2, height=height // 2, fps=frame_rate
        )
        
        video_encoder = self.model_ledger.video_encoder()
        stage_1_conditionings = combined_image_conditionings(
            images=images,
            height=stage_1_output_shape.height,
            width=stage_1_output_shape.width,
            video_encoder=video_encoder,
            dtype=dtype,
            device=device,
        )
        torch.cuda.synchronize()
        del video_encoder
        cleanup_memory()
        
        transformer = self.model_ledger.transformer()
        
        # Use DISTILLED_SIGMA_VALUES for 8 steps at half resolution
        from ltx_pipelines.utils.constants import DISTILLED_SIGMA_VALUES
        stage_1_sigmas = torch.tensor(DISTILLED_SIGMA_VALUES, device=device)
        stepper = Res2sDiffusionStep()
        
        def first_stage_denoising_loop(sigmas, video_state, audio_state, stepper):
            return res2s_audio_video_denoising_loop(
                sigmas=sigmas,
                video_state=video_state,
                audio_state=audio_state,
                stepper=stepper,
                denoise_fn=multi_modal_guider_denoising_func(
                    video_guider=MultiModalGuider(params=video_guider_params, negative_context=v_context_n),
                    audio_guider=MultiModalGuider(params=audio_guider_params, negative_context=a_context_n),
                    v_context=v_context_p,
                    a_context=a_context_p,
                    transformer=transformer,
                ),
            )
        
        video_state, audio_state = denoise_audio_video(
            output_shape=stage_1_output_shape,
            conditionings=stage_1_conditionings,
            noiser=noiser,
            sigmas=stage_1_sigmas,
            stepper=stepper,
            denoising_loop_fn=first_stage_denoising_loop,
            components=self.pipeline_components,
            dtype=dtype,
            device=device,
        )
        
        torch.cuda.synchronize()
        del transformer
        cleanup_memory()
        
        # ===================== UPSCALING =====================
        video_encoder = self.model_ledger.video_encoder()
        upscaled_video_latent = upsample_video(
            latent=video_state.latent[:1],
            video_encoder=video_encoder,
            upsampler=self.model_ledger.spatial_upsampler(),
        )
        
        stage_2_output_shape = VideoPixelShape(batch=1, frames=num_frames, width=width, height=height, fps=frame_rate)
        stage_2_conditionings = combined_image_conditionings(
            images=images,
            height=stage_2_output_shape.height,
            width=stage_2_output_shape.width,
            video_encoder=video_encoder,
            dtype=dtype,
            device=device,
        )
        torch.cuda.synchronize()
        del video_encoder
        cleanup_memory()
        
        # ===================== STAGE 2: 3 steps at full resolution =====================
        transformer = self.model_ledger.transformer()
        
        from ltx_pipelines.utils.constants import STAGE_2_DISTILLED_SIGMA_VALUES
        stage_2_sigmas = torch.tensor(STAGE_2_DISTILLED_SIGMA_VALUES, device=device)
        
        def second_stage_denoising_loop(sigmas, video_state, audio_state, stepper):
            return res2s_audio_video_denoising_loop(
                sigmas=sigmas,
                video_state=video_state,
                audio_state=audio_state,
                stepper=stepper,
                denoise_fn=simple_denoising_func(
                    video_context=v_context_p,
                    audio_context=a_context_p,
                    transformer=transformer,
                ),
            )
        
        video_state, audio_state = denoise_audio_video(
            output_shape=stage_2_output_shape,
            conditionings=stage_2_conditionings,
            noiser=noiser,
            sigmas=stage_2_sigmas,
            stepper=stepper,
            denoising_loop_fn=second_stage_denoising_loop,
            components=self.pipeline_components,
            dtype=dtype,
            device=device,
            noise_scale=stage_2_sigmas[0],
            initial_video_latent=upscaled_video_latent,
            initial_audio_latent=audio_state.latent,
        )
        
        torch.cuda.synchronize()
        del transformer
        cleanup_memory()
        
        # ===================== DECODE =====================
        decoded_video = vae_decode_video(
            video_state.latent, self.model_ledger.video_decoder(), tiling_config, generator
        )
        decoded_audio = vae_decode_audio(
            audio_state.latent, self.model_ledger.audio_decoder(), self.model_ledger.vocoder()
        )
        
        return decoded_video, decoded_audio


# =============================================================================
# Model Download
# =============================================================================

print("=" * 80)
print("Downloading LTX-2.3 HQ models...")
print("=" * 80)

weights_dir = Path("weights")
weights_dir.mkdir(exist_ok=True)

checkpoint_path = hf_hub_download(
    repo_id=LTX_MODEL_REPO,
    filename="ltx-2.3-22b-distilled-1.1.safetensors",
    local_dir=str(weights_dir),
    local_dir_use_symlinks=False,  # Ensure actual file copy, not symlink
)

# Force download if not present
if not os.path.exists(checkpoint_path):
    print(f"Re-downloading checkpoint to {weights_dir}...")
    checkpoint_path = hf_hub_download(
        repo_id=LTX_MODEL_REPO,
        filename="ltx-2.3-22b-distilled-1.1.safetensors",
        local_dir=str(weights_dir),
        local_dir_use_symlinks=False,
        force_download=True,
    )

print(f"Checkpoint at: {checkpoint_path}")
print(f"File exists: {os.path.exists(checkpoint_path)}")
print(f"File size: {os.path.getsize(checkpoint_path) / 1024**3:.2f} GB")

spatial_upsampler_path = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-spatial-upscaler-x2-1.1.safetensors")
distilled_lora_path = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-22b-distilled-lora-384.safetensors")
gemma_root = snapshot_download(repo_id=GEMMA_REPO)

print(f"Dev checkpoint: {checkpoint_path}")
print(f"Spatial upsampler: {spatial_upsampler_path}")
print(f"Distilled LoRA: {distilled_lora_path}")
print(f"Gemma root: {gemma_root}")

# =============================================================================
# Download Custom LoRAs
# =============================================================================

LORA_REPO = "dagloop5/LoRA"

print("=" * 80)
print("Downloading custom LoRA adapters...")
print("=" * 80)

pose_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="LTX2_3_NSFW_furry_concat_v2.safetensors")
general_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="LTX2.3_reasoning_I2V_V3.safetensors")
motion_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="motion_helper.safetensors")
dreamlay_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="DR34ML4Y_LTXXX_PREVIEW_RC1.safetensors")
mself_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="Furry Hyper Masturbation - LTX-2 I2V v1.safetensors")
dramatic_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="LTX-2.3 - Orgasm.safetensors")
fluid_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="cr3ampi3_animation_i2v_ltx2_v1.0.safetensors")
liquid_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="liquid_wet_dr1pp_ltx2_v1.0_scaled.safetensors")
demopose_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="clapping-cheeks-audio-v001-alpha.safetensors")
voice_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="hentai_voice_ltx23.safetensors")
realism_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="FurryenhancerLTX2.3V1.215.safetensors")
transition_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="LTX-2_takerpov_lora_v1.2.safetensors")

print(f"All 12 custom LoRAs downloaded + distilled LoRA")
print("=" * 80)

# =============================================================================
# Pipeline Initialization
# =============================================================================

print("Initializing HQ Pipeline...")

pipeline = HQPipelineWithCachedLoRA(
    checkpoint_path=checkpoint_path,
    spatial_upsampler_path=spatial_upsampler_path,
    gemma_root=gemma_root,
    quantization=QuantizationPolicy.fp8_cast(),
)

print("Pipeline initialized!")
print("=" * 80)

# =============================================================================
# ZeroGPU Tensor Preloading - Single Transformer
# =============================================================================

print("Preloading models for ZeroGPU tensor packing...")

# Load shared components
_video_encoder = pipeline.model_ledger.video_encoder()
_video_decoder = pipeline.model_ledger.video_decoder()
_audio_encoder = pipeline.model_ledger.audio_encoder()
_audio_decoder = pipeline.model_ledger.audio_decoder()
_vocoder = pipeline.model_ledger.vocoder()
_spatial_upsampler = pipeline.model_ledger.spatial_upsampler()
_text_encoder = pipeline.model_ledger.text_encoder()
_embeddings_processor = pipeline.model_ledger.gemma_embeddings_processor()

# Load the SINGLE transformer
_transformer = pipeline.model_ledger.transformer()

# Replace ledger methods with lambdas returning cached instances
pipeline.model_ledger.video_encoder = lambda: _video_encoder
pipeline.model_ledger.video_decoder = lambda: _video_decoder
pipeline.model_ledger.audio_encoder = lambda: _audio_encoder
pipeline.model_ledger.audio_decoder = lambda: _audio_decoder
pipeline.model_ledger.vocoder = lambda: _vocoder
pipeline.model_ledger.spatial_upsampler = lambda: _spatial_upsampler
pipeline.model_ledger.text_encoder = lambda: _text_encoder
pipeline.model_ledger.gemma_embeddings_processor = lambda: _embeddings_processor
pipeline.model_ledger.transformer = lambda: _transformer

print("All models preloaded for ZeroGPU tensor packing!")
print("=" * 80)
print("Pipeline ready!")
print("=" * 80)

# =============================================================================
# LoRA Cache Functions
# =============================================================================

LORA_CACHE_DIR = Path("lora_cache")
LORA_CACHE_DIR.mkdir(exist_ok=True)

def prepare_lora_cache(
    distilled_strength: float,
    pose_strength: float, general_strength: float, motion_strength: float,
    dreamlay_strength: float, mself_strength: float, dramatic_strength: float,
    fluid_strength: float, liquid_strength: float, demopose_strength: float,
    voice_strength: float, realism_strength: float, transition_strength: float,
    progress=gr.Progress(track_tqdm=True),
):
    """Build cached LoRA state for single transformer."""
    global pipeline
    
    print("[LoRA] === Starting LoRA Cache Preparation ===")
    progress(0.05, desc="Preparing LoRA cache...")
    
    # Validate all LoRA files exist
    print("[LoRA] Validating LoRA file paths...")
    lora_files = [
        ("Distilled", distilled_lora_path, distilled_strength),
        ("Pose", pose_lora_path, pose_strength),
        ("General", general_lora_path, general_strength),
        ("Motion", motion_lora_path, motion_strength),
        ("Dreamlay", dreamlay_lora_path, dreamlay_strength),
        ("Mself", mself_lora_path, mself_strength),
        ("Dramatic", dramatic_lora_path, dramatic_strength),
        ("Fluid", fluid_lora_path, fluid_strength),
        ("Liquid", liquid_lora_path, liquid_strength),
        ("Demopose", demopose_lora_path, demopose_strength),
        ("Voice", voice_lora_path, voice_strength),
        ("Realism", realism_lora_path, realism_strength),
        ("Transition", transition_lora_path, transition_strength),
    ]
    
    active_loras = []
    for name, path, strength in lora_files:
        if path is not None and float(strength) != 0.0:
            active_loras.append((name, path, strength))
            print(f"[LoRA]   - {name}: strength={strength}")
    
    print(f"[LoRA] Active LoRAs: {len(active_loras)}")
    
    key_str = f"{checkpoint_path}:{distilled_strength}:{pose_strength}:{general_strength}:{motion_strength}:{dreamlay_strength}:{mself_strength}:{dramatic_strength}:{fluid_strength}:{liquid_strength}:{demopose_strength}:{voice_strength}:{realism_strength}:{transition_strength}"
    key = hashlib.sha256(key_str.encode()).hexdigest()
    
    cache_path = LORA_CACHE_DIR / f"{key}.safetensors"
    print(f"[LoRA] Cache key: {key[:16]}...")
    print(f"[LoRA] Cache path: {cache_path}")
    
    if cache_path.exists():
        print("[LoRA] Loading from existing cache...")
        progress(0.20, desc="Loading cached LoRA state...")
        state = load_file(str(cache_path))
        print(f"[LoRA] Loaded state dict with {len(state)} keys, size: {sum(v.numel() * v.element_size() for v in state.values()) / 1024**3:.2f} GB")
        pipeline.apply_cached_lora_state(state)
        print("[LoRA] State applied to pipeline._cached_state")
        print("[LoRA] === LoRA Cache Preparation Complete ===")
        return f"Loaded cached LoRA state: {cache_path.name} ({len(state)} keys)"
    
    if not active_loras:
        print("[LoRA] No non-zero LoRA strengths selected; nothing to prepare.")
        print("[LoRA] === LoRA Cache Preparation Complete (no LoRAs) ===")
        return "No non-zero LoRA strengths selected; nothing to prepare."
    
    entries = [
        (distilled_lora_path, distilled_strength),
        (pose_lora_path, pose_strength),
        (general_lora_path, general_strength),
        (motion_lora_path, motion_strength),
        (dreamlay_lora_path, dreamlay_strength),
        (mself_lora_path, mself_strength),
        (dramatic_lora_path, dramatic_strength),
        (fluid_lora_path, fluid_strength),
        (liquid_lora_path, liquid_strength),
        (demopose_lora_path, demopose_strength),
        (voice_lora_path, voice_strength),
        (realism_lora_path, realism_strength),
        (transition_lora_path, transition_strength),
    ]
    
    loras_for_builder = [
        LoraPathStrengthAndSDOps(path, strength, LTXV_LORA_COMFY_RENAMING_MAP)
        for path, strength in entries
        if path is not None and float(strength) != 0.0
    ]
    
    print(f"[LoRA] Building fused state on CPU with {len(loras_for_builder)} LoRAs...")
    print("[LoRA] This may take several minutes (do not close the Space)...")
    progress(0.35, desc="Building fused state (CPU)...")
    
    import time
    start_time = time.time()
    
    tmp_ledger = pipeline.model_ledger.__class__(
        dtype=torch.bfloat16,
        device=torch.device("cpu"),
        checkpoint_path=str(checkpoint_path),
        spatial_upsampler_path=str(spatial_upsampler_path),
        gemma_root_path=str(gemma_root),
        loras=tuple(loras_for_builder),
        quantization=None,
    )
    print(f"[LoRA] Temporary ledger created in {time.time() - start_time:.1f}s")
    
    print("[LoRA] Loading transformer with LoRAs applied...")
    transformer = tmp_ledger.transformer()
    print(f"[LoRA] Transformer loaded in {time.time() - start_time:.1f}s")
    
    print("[LoRA] Extracting state dict...")
    progress(0.70, desc="Extracting fused stateDict")
    state = {k: v.detach().cpu().contiguous() for k, v in transformer.state_dict().items()}
    print(f"[LoRA] State dict extracted: {len(state)} keys")
    
    print(f"[LoRA] Saving to cache: {cache_path}")
    save_file(state, str(cache_path))
    print(f"[LoRA] Cache saved, size: {sum(v.numel() * v.element_size() for v in state.values()) / 1024**3:.2f} GB")
    
    print("[LoRA] Cleaning up temporary ledger...")
    del transformer, tmp_ledger
    gc.collect()
    print(f"[LoRA] Cleanup complete in {time.time() - start_time:.1f}s total")
    
    print("[LoRA] Applying state to pipeline._cached_state...")
    progress(0.90, desc="Applying LoRA state to pipeline...")
    pipeline.apply_cached_lora_state(state)
    
    progress(1.0, desc="Done!")
    print("[LoRA] === LoRA Cache Preparation Complete ===")
    return f"Built and cached LoRA state: {cache_path.name} ({len(state)} keys, {time.time() - start_time:.1f}s)"

# =============================================================================
# LoRA State Application (called BEFORE pipeline generation)
# =============================================================================

def apply_prepared_lora_state_to_pipeline():
    """
    Apply the prepared LoRA state from pipeline._cached_state to the preloaded
    transformer. This should be called BEFORE pipeline generation, not during.
    """
    print("[LoRA] === Applying LoRA State to Transformer ===")
    
    if pipeline._cached_state is None:
        print("[LoRA] No prepared LoRA state available; skipping.")
        print("[LoRA] === LoRA Application Complete (no state) ===")
        return False
    
    try:
        existing_transformer = _transformer  # The preloaded transformer from globals
        state = pipeline._cached_state
        print(f"[LoRA] Applying state dict with {len(state)} keys...")
        print(f"[LoRA] State dict size: {sum(v.numel() * v.element_size() for v in state.values()) / 1024**3:.2f} GB")
        
        import time
        start_time = time.time()
        
        with torch.no_grad():
            missing, unexpected = existing_transformer.load_state_dict(state, strict=False)
        
        print(f"[LoRA] load_state_dict completed in {time.time() - start_time:.1f}s")
        
        if missing:
            print(f"[LoRA] WARNING: {len(missing)} keys missing from state dict")
        if unexpected:
            print(f"[LoRA] WARNING: {len(unexpected)} unexpected keys in state dict")
        
        if not missing and not unexpected:
            print("[LoRA] State dict loaded successfully with no mismatches!")
        
        print("[LoRA] === LoRA Application Complete (success) ===")
        return True
    except Exception as e:
        print(f"[LoRA] FAILED to apply LoRA state: {type(e).__name__}: {e}")
        print("[LoRA] === LoRA Application Complete (FAILED) ===")
        return False

# =============================================================================
# Helper Functions
# =============================================================================

def log_memory(tag: str):
    if torch.cuda.is_available():
        allocated = torch.cuda.memory_allocated() / 1024**3
        peak = torch.cuda.max_memory_allocated() / 1024**3
        free, total = torch.cuda.mem_get_info()
        print(f"[VRAM {tag}] allocated={allocated:.2f}GB peak={peak:.2f}GB free={free / 1024**3:.2f}GB total={total / 1024**3:.2f}GB")


def calculate_frames(duration: float, frame_rate: float = DEFAULT_FRAME_RATE) -> int:
    ideal_frames = int(duration * frame_rate)
    ideal_frames = max(ideal_frames, MIN_FRAMES)
    k = round((ideal_frames - 1) / 8)
    frames = k * 8 + 1
    return min(frames, MAX_FRAMES)

def detect_aspect_ratio(image) -> str:
    if image is None:
        return "16:9"
    if hasattr(image, "size"):
        w, h = image.size
    elif hasattr(image, "shape"):
        h, w = image.shape[:2]
    else:
        return "16:9"
    ratio = w / h
    candidates = {"16:9": 16 / 9, "9:16": 9 / 16, "1:1": 1.0}
    return min(candidates, key=lambda k: abs(ratio - candidates[k]))

def on_image_upload(first_image, last_image, high_res):
    ref_image = first_image if first_image is not None else last_image
    aspect = detect_aspect_ratio(ref_image)
    tier = "high" if high_res else "low"
    w, h = RESOLUTIONS[tier][aspect]
    return gr.update(value=w), gr.update(value=h)


def on_highres_toggle(first_image, last_image, high_res):
    ref_image = first_image if first_image is not None else last_image
    aspect = detect_aspect_ratio(ref_image)
    tier = "high" if high_res else "low"
    w, h = RESOLUTIONS[tier][aspect]
    return gr.update(value=w), gr.update(value=h)


def get_gpu_duration(
    first_image,
    last_image,
    prompt: str,
    negative_prompt: str,
    duration: float,
    gpu_duration: float,
    seed: int = 42,
    randomize_seed: bool = True,
    height: int = 1024,
    width: int = 1536,
    video_cfg_scale: float = 1.0,
    video_stg_scale: float = 0.0,
    video_rescale_scale: float = 0.45,
    video_a2v_scale: float = 3.0,
    audio_cfg_scale: float = 1.0,
    audio_stg_scale: float = 0.0,
    audio_rescale_scale: float = 1.0,
    audio_v2a_scale: float = 3.0,
    distilled_strength: float = 0.0,
    pose_strength: float = 0.0,
    general_strength: float = 0.0,
    motion_strength: float = 0.0,
    dreamlay_strength: float = 0.0,
    mself_strength: float = 0.0,
    dramatic_strength: float = 0.0,
    fluid_strength: float = 0.0,
    liquid_strength: float = 0.0,
    demopose_strength: float = 0.0,
    voice_strength: float = 0.0,
    realism_strength: float = 0.0,
    transition_strength: float = 0.0,
    progress=None,
) -> int:
    return int(gpu_duration)


@spaces.GPU(duration=get_gpu_duration)
@torch.inference_mode()
def generate_video(
    first_image,
    last_image,
    prompt: str,
    negative_prompt: str,
    duration: float,
    gpu_duration: float,
    seed: int = 42,
    randomize_seed: bool = True,
    height: int = 1024,
    width: int = 1536,
    video_cfg_scale: float = 1.0,
    video_stg_scale: float = 0.0,
    video_rescale_scale: float = 0.45,
    video_a2v_scale: float = 3.0,
    audio_cfg_scale: float = 1.0,
    audio_stg_scale: float = 0.0,
    audio_rescale_scale: float = 1.0,
    audio_v2a_scale: float = 3.0,
    distilled_strength: float = 0.0,
    pose_strength: float = 0.0,
    general_strength: float = 0.0,
    motion_strength: float = 0.0,
    dreamlay_strength: float = 0.0,
    mself_strength: float = 0.0,
    dramatic_strength: float = 0.0,
    fluid_strength: float = 0.0,
    liquid_strength: float = 0.0,
    demopose_strength: float = 0.0,
    voice_strength: float = 0.0,
    realism_strength: float = 0.0,
    transition_strength: float = 0.0,
    progress=gr.Progress(track_tqdm=True),
):
    try:
        torch.cuda.reset_peak_memory_stats()
        log_memory("start")

        current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)
        print(f"Using seed: {current_seed}")

        print(f"Resolution: {width}x{height}")

        num_frames = calculate_frames(duration, DEFAULT_FRAME_RATE)
        print(f"Frames: {num_frames} ({duration}s @ {DEFAULT_FRAME_RATE}fps)")

        images = []
        output_dir = Path("outputs")
        output_dir.mkdir(exist_ok=True)
        
        if first_image is not None:
            temp_first_path = output_dir / f"temp_first_{current_seed}.jpg"
            if hasattr(first_image, "save"):
                first_image.save(temp_first_path)
            else:
                import shutil
                shutil.copy(first_image, temp_first_path)
            images.append(ImageConditioningInput(path=str(temp_first_path), frame_idx=0, strength=1.0))
        
        if last_image is not None:
            temp_last_path = output_dir / f"temp_last_{current_seed}.jpg"
            if hasattr(last_image, "save"):
                last_image.save(temp_last_path)
            else:
                import shutil
                shutil.copy(last_image, temp_last_path)
            images.append(ImageConditioningInput(path=str(temp_last_path), frame_idx=num_frames - 1, strength=1.0))

        tiling_config = TilingConfig.default()
        video_chunks_number = get_video_chunks_number(num_frames, tiling_config)

        video_guider_params = MultiModalGuiderParams(
            cfg_scale=video_cfg_scale,
            stg_scale=video_stg_scale,
            rescale_scale=video_rescale_scale,
            modality_scale=video_a2v_scale,
            skip_step=0,
            stg_blocks=[],
        )

        audio_guider_params = MultiModalGuiderParams(
            cfg_scale=audio_cfg_scale,
            stg_scale=audio_stg_scale,
            rescale_scale=audio_rescale_scale,
            modality_scale=audio_v2a_scale,
            skip_step=0,
            stg_blocks=[],
        )

        log_memory("before pipeline call")

        apply_prepared_lora_state_to_pipeline()

        video, audio = pipeline(
            prompt=prompt,
            negative_prompt=negative_prompt,
            seed=current_seed,
            height=height,
            width=width,
            num_frames=num_frames,
            frame_rate=DEFAULT_FRAME_RATE,
            video_guider_params=video_guider_params,
            audio_guider_params=audio_guider_params,
            images=images,
            tiling_config=tiling_config,
        )

        log_memory("after pipeline call")

        output_path = tempfile.mktemp(suffix=".mp4")
        encode_video(
            video=video,
            fps=DEFAULT_FRAME_RATE,
            audio=audio,
            output_path=output_path,
            video_chunks_number=video_chunks_number,
        )

        log_memory("after encode_video")
        return str(output_path), current_seed

    except Exception as e:
        import traceback
        log_memory("on error")
        print(f"Error: {str(e)}\n{traceback.format_exc()}")
        return None, current_seed


# =============================================================================
# Gradio UI
# =============================================================================

css = """
.fillable {max-width: 1200px !important}
.progress-text {color: black}
"""

with gr.Blocks(title="LTX-2.3 Two-Stage HQ with LoRA Cache") as demo:
    gr.Markdown("# LTX-2.3 Two-Stage HQ Video Generation with LoRA Cache")
    gr.Markdown(
        "High-quality text/image-to-video with cached LoRA state + CFG guidance. "
        "[[Model]](https://huggingface.co/Lightricks/LTX-2.3)"
    )

    with gr.Row():
        # LEFT SIDE: Input Controls
        with gr.Column():
            with gr.Row():
                first_image = gr.Image(label="First Frame (Optional)", type="pil")
                last_image = gr.Image(label="Last Frame (Optional)", type="pil")
            
            prompt = gr.Textbox(
                label="Prompt",
                value=DEFAULT_PROMPT,
                lines=3,
            )
            
            negative_prompt = gr.Textbox(
                label="Negative Prompt",
                value=DEFAULT_NEGATIVE_PROMPT,
                lines=2,
            )
            
            duration = gr.Slider(
                label="Duration (seconds)",
                minimum=1.0, maximum=30.0, value=10.0, step=0.1,
            )
            
            with gr.Row():
                seed = gr.Number(label="Seed", value=42, precision=0, minimum=0, maximum=MAX_SEED)
                randomize_seed = gr.Checkbox(label="Randomize Seed", value=True)
            
            with gr.Row():
                high_res = gr.Checkbox(label="High Resolution", value=True)
            
            with gr.Row():
                width = gr.Number(label="Width", value=1536, precision=0)
                height = gr.Number(label="Height", value=1024, precision=0)
            
            generate_btn = gr.Button("Generate Video", variant="primary", size="lg")

            with gr.Accordion("Advanced Settings", open=False):
                gr.Markdown("### Video Guidance Parameters")
                
                with gr.Row():
                    video_cfg_scale = gr.Slider(
                        label="Video CFG Scale", minimum=1.0, maximum=10.0,
                        value=LTX_2_3_HQ_PARAMS.video_guider_params.cfg_scale, step=0.1
                    )
                    video_stg_scale = gr.Slider(
                        label="Video STG Scale", minimum=0.0, maximum=2.0, value=0.0, step=0.1
                    )
                
                with gr.Row():
                    video_rescale_scale = gr.Slider(
                        label="Video Rescale", minimum=0.0, maximum=2.0, value=0.45, step=0.1
                    )
                    video_a2v_scale = gr.Slider(
                        label="A2V Scale", minimum=0.0, maximum=5.0, value=3.0, step=0.1
                    )
                
                gr.Markdown("### Audio Guidance Parameters")
                
                with gr.Row():
                    audio_cfg_scale = gr.Slider(
                        label="Audio CFG Scale", minimum=1.0, maximum=15.0,
                        value=LTX_2_3_HQ_PARAMS.audio_guider_params.cfg_scale, step=0.1
                    )
                    audio_stg_scale = gr.Slider(
                        label="Audio STG Scale", minimum=0.0, maximum=2.0, value=0.0, step=0.1
                    )
                
                with gr.Row():
                    audio_rescale_scale = gr.Slider(
                        label="Audio Rescale", minimum=0.0, maximum=2.0, value=1.0, step=0.1
                    )
                    audio_v2a_scale = gr.Slider(
                        label="V2A Scale", minimum=0.0, maximum=5.0, value=3.0, step=0.1
                    )

        # RIGHT SIDE: Output and LoRA
        with gr.Column():
            output_video = gr.Video(label="Generated Video", autoplay=False)
            
            gpu_duration = gr.Slider(
                label="ZeroGPU duration (seconds)",
                minimum=30.0, maximum=240.0, value=90.0, step=1.0,
                info="Increase for longer videos, higher resolution, or LoRA usage"
            )
            
            gr.Markdown("### LoRA Adapter Strengths")
            gr.Markdown("Set to 0 to disable, then click 'Prepare LoRA Cache'")
            
            with gr.Row():
                distilled_strength = gr.Slider(label="Distilled LoRA", minimum=0.0, maximum=1.5, value=0.0, step=0.01)
                pose_strength = gr.Slider(label="Anthro Enhancer", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
            
            with gr.Row():
                general_strength = gr.Slider(label="Reasoning Enhancer", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
                motion_strength = gr.Slider(label="Anthro Posing", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
            
            with gr.Row():
                dreamlay_strength = gr.Slider(label="Dreamlay", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
                mself_strength = gr.Slider(label="Mself", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
            
            with gr.Row():
                dramatic_strength = gr.Slider(label="Dramatic", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
                fluid_strength = gr.Slider(label="Fluid Helper", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
            
            with gr.Row():
                liquid_strength = gr.Slider(label="Liquid Helper", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
                demopose_strength = gr.Slider(label="Audio Helper", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
            
            with gr.Row():
                voice_strength = gr.Slider(label="Voice Helper", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
                realism_strength = gr.Slider(label="Anthro Realism", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
            
            with gr.Row():
                transition_strength = gr.Slider(label="POV", minimum=0.0, maximum=2.0, value=0.0, step=0.01)
                gr.Markdown("")  # Spacer for alignment
            
            prepare_lora_btn = gr.Button("Prepare / Load LoRA Cache", variant="secondary")
            lora_status = gr.Textbox(
                label="LoRA Cache Status",
                value="No LoRA state prepared yet.",
                interactive=False,
            )

    # Event handlers
    first_image.change(fn=on_image_upload, inputs=[first_image, last_image, high_res], outputs=[width, height])
    last_image.change(fn=on_image_upload, inputs=[first_image, last_image, high_res], outputs=[width, height])
    high_res.change(fn=on_highres_toggle, inputs=[first_image, last_image, high_res], outputs=[width, height])

    prepare_lora_btn.click(
        fn=prepare_lora_cache,
        inputs=[distilled_strength, pose_strength, general_strength, motion_strength, dreamlay_strength,
                mself_strength, dramatic_strength, fluid_strength, liquid_strength,
                demopose_strength, voice_strength, realism_strength, transition_strength],
        outputs=[lora_status],
    )

    generate_btn.click(
        fn=generate_video,
        inputs=[
            first_image, last_image, prompt, negative_prompt, duration, gpu_duration,
            seed, randomize_seed, height, width,
            video_cfg_scale, video_stg_scale, video_rescale_scale, video_a2v_scale,
            audio_cfg_scale, audio_stg_scale, audio_rescale_scale, audio_v2a_scale,
            distilled_strength, pose_strength, general_strength, motion_strength,
            dreamlay_strength, mself_strength, dramatic_strength, fluid_strength,
            liquid_strength, demopose_strength, voice_strength, realism_strength,
            transition_strength,
        ],
        outputs=[output_video, seed],
    )


if __name__ == "__main__":
    demo.queue().launch(theme=gr.themes.Citrus(), css=css, mcp_server=False)