Create app.py

app.py

@@ -0,0 +1,921 @@
+# =============================================================================
+# 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, 257
+
+# 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
+    """
+    
+    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
+        
+        # Create ONE ModelLedger for everything
+        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=(),  # NO LoRAs
+            quantization=quantization,
+        )
+        
+        # Pipeline components
+        self.pipeline_components = PipelineComponents(
+            dtype=self.dtype,
+            device=self.device,
+        )
+        
+        # Storage for cached LoRA state
+        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,
+        num_inference_steps: int,
+        video_guider_params: MultiModalGuiderParams,
+        audio_guider_params: MultiModalGuiderParams,
+        images: list,
+        tiling_config: TilingConfig | None = None,
+        enhance_prompt: bool = False,
+    ):
+        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.components.schedulers import LTX2Scheduler
+        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)
+        
+        # Apply cached LoRA state if available
+        if self._cached_state is not None:
+            print("[LoRA] Applying cached state to transformer...")
+            transformer = self.model_ledger.transformer()
+            with torch.no_grad():
+                transformer.load_state_dict(self._cached_state, strict=False)
+        
+        ctx_p, ctx_n = encode_prompts(
+            [prompt, negative_prompt],
+            self.model_ledger,
+            enhance_first_prompt=enhance_prompt,
+            enhance_prompt_image=images[0][0] if len(images) > 0 else None,
+            enhance_prompt_seed=seed,
+        )
+        
+        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 =====================
+        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()
+        
+        empty_latent = torch.empty(VideoLatentShape.from_pixel_shape(stage_1_output_shape).to_torch_shape())
+        stepper = Res2sDiffusionStep()
+        sigmas = (
+            LTX2Scheduler()
+            .execute(latent=empty_latent, steps=num_inference_steps)
+            .to(dtype=torch.float32, device=device)
+        )
+        
+        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=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 =====================
+        transformer = self.model_ledger.transformer()
+        
+        from ltx_pipelines.utils.constants import STAGE_2_DISTILLED_SIGMA_VALUES
+        distilled_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=distilled_sigmas,
+            stepper=stepper,
+            denoising_loop_fn=second_stage_denoising_loop,
+            components=self.pipeline_components,
+            dtype=dtype,
+            device=device,
+            noise_scale=distilled_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)
+
+checkpoint_path = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-22b-dev.safetensors")
+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
+    
+    progress(0.05, desc="Preparing LoRA cache...")
+    
+    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"
+    
+    if cache_path.exists():
+        progress(0.20, desc="Loading cached LoRA state...")
+        state = load_file(str(cache_path))
+        pipeline.apply_cached_lora_state(state)
+        return f"Loaded cached LoRA state: {cache_path.name}"
+    
+    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 = [
+        LoraPathStrengthAndSDOps(path, strength, LTXV_LORA_COMFY_RENAMING_MAP)
+        for path, strength in entries
+        if path is not None and float(strength) != 0.0
+    ]
+    
+    progress(0.35, desc="Building fused state (CPU)...")
+    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),
+        quantization=None,
+    )
+    transformer = tmp_ledger.transformer()
+    state = {k: v.detach().cpu().contiguous() for k, v in transformer.state_dict().items()}
+    save_file(state, str(cache_path))
+    
+    del transformer, tmp_ledger
+    gc.collect()
+    
+    progress(0.90, desc="Applying LoRA state to pipeline...")
+    pipeline.apply_cached_lora_state(state)
+    
+    progress(1.0, desc="Done!")
+    return f"Built and cached LoRA state: {cache_path.name}"
+
+
+# =============================================================================
+# 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 validate_resolution(height: int, width: int) -> tuple[int, int]:
+    height = round(height / STEP) * STEP
+    width = round(width / STEP) * STEP
+    height = max(MIN_DIM, min(height, MAX_DIM))
+    width = max(MIN_DIM, min(width, MAX_DIM))
+    return height, width
+
+
+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,
+    enhance_prompt: bool = False,
+    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,
+    enhance_prompt: bool = False,
+    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}")
+
+        height, width = validate_resolution(int(height), int(width))
+        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((str(temp_first_path), 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((str(temp_last_path), 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")
+
+        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,
+            num_inference_steps=LTX_2_3_HQ_PARAMS.num_inference_steps,
+            video_guider_params=video_guider_params,
+            audio_guider_params=audio_guider_params,
+            images=images,
+            tiling_config=tiling_config,
+            enhance_prompt=enhance_prompt,
+        )
+
+        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: white}
+"""
+
+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=0.5, maximum=8.0, value=2.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():
+                enhance_prompt = gr.Checkbox(label="Enhance Prompt", value=False)
+                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=True)
+            
+            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, enhance_prompt,
+            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=True)