app.py

import os
import subprocess
import sys

# Disable torch.compile / dynamo before any torch import
os.environ["TORCH_COMPILE_DISABLE"] = "1"
os.environ["TORCHDYNAMO_DISABLE"] = "1"

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

# Clone LTX-2 repo and install packages
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 = "a2c3f24078eb918171967f74b6f66b756b29ee45"  # known working commit with decode_video

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)

print(f"Checking out pinned commit {LTX_COMMIT}...")
subprocess.run(["git", "fetch", "--all", "--tags"], cwd=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"))

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 safetensors import safe_open
import json
import requests

from ltx_core.components.diffusion_steps import Res2sDiffusionStep
from ltx_core.components.guiders import MultiModalGuider, MultiModalGuiderParams
from ltx_core.components.noisers import GaussianNoiser
from ltx_core.model.video_vae import TilingConfig, get_video_chunks_number
from ltx_core.types import Audio, VideoLatentShape, VideoPixelShape
from ltx_pipelines.utils.args import ImageConditioningInput, hq_2_stage_arg_parser
from ltx_pipelines.utils.blocks import (
    AudioDecoder,
    DiffusionStage,
    ImageConditioner,
    PromptEncoder,
    VideoDecoder,
    VideoUpsampler,
)
from ltx_pipelines.utils.constants import LTX_2_3_HQ_PARAMS, STAGE_2_DISTILLED_SIGMAS
from ltx_pipelines.utils.denoisers import GuidedDenoiser, SimpleDenoiser
from ltx_pipelines.utils.helpers import (
    assert_resolution,
    combined_image_conditionings,
    get_device,
)
from ltx_pipelines.utils.media_io import encode_video
from ltx_pipelines.utils.samplers import res2s_audio_video_denoising_loop
from ltx_core.loader.primitives import LoraPathStrengthAndSDOps
from ltx_core.loader.sd_ops import LTXV_LORA_COMFY_RENAMING_MAP

from collections.abc import Iterator

from ltx_core.components.schedulers import LTX2Scheduler
from ltx_core.loader.registry import Registry
from ltx_core.quantization import QuantizationPolicy
from ltx_pipelines.utils.types import ModalitySpec

# Force-patch xformers attention into the LTX attention module.
from ltx_core.model.transformer import attention as _attn_mod
print(f"[ATTN] Before patch: memory_efficient_attention={_attn_mod.memory_efficient_attention}")
try:
    from xformers.ops import memory_efficient_attention as _mea
    _attn_mod.memory_efficient_attention = _mea
    print(f"[ATTN] After patch: memory_efficient_attention={_attn_mod.memory_efficient_attention}")
except Exception as e:
    print(f"[ATTN] xformers patch FAILED: {type(e).__name__}: {e}")

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

MAX_SEED = np.iinfo(np.int32).max
DEFAULT_PROMPT = (
    "An astronaut hatches from a fragile egg on the surface of the Moon, "
    "the shell cracking and peeling apart in gentle low-gravity motion. "
    "Fine lunar dust lifts and drifts outward with each movement, floating "
    "in slow arcs before settling back onto the ground."
)
DEFAULT_FRAME_RATE = 24.0

# Resolution presets: (width, height)
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)},
}


class LTX23NegativePromptTwoStagePipeline:
    def __init__(
        self,
        checkpoint_path: str,
        spatial_upsampler_path: str,
        gemma_root: str,
        loras: tuple[LoraPathStrengthAndSDOps, ...],
        device: torch.device | None = None,
        quantization: QuantizationPolicy | None = None,
        registry: Registry | None = None,
        torch_compile: bool = False,
    ):
        self.device = device or get_device()
        self.dtype = torch.bfloat16
        self._scheduler = LTX2Scheduler()

        self.prompt_encoder = PromptEncoder(checkpoint_path, gemma_root, self.dtype, self.device, registry=registry)
        self.image_conditioner = ImageConditioner(checkpoint_path, self.dtype, self.device, registry=registry)
        self.upsampler = VideoUpsampler(checkpoint_path, spatial_upsampler_path, self.dtype, self.device, registry=registry)
        self.video_decoder = VideoDecoder(checkpoint_path, self.dtype, self.device, registry=registry)
        self.audio_decoder = AudioDecoder(checkpoint_path, self.dtype, self.device, registry=registry)

        self.stage_1 = DiffusionStage(
            checkpoint_path,
            self.dtype,
            self.device,
            loras=tuple(loras),
            quantization=quantization,
            registry=registry,
            torch_compile=torch_compile,
        )
        self.stage_2 = DiffusionStage(
            checkpoint_path,
            self.dtype,
            self.device,
            loras=tuple(loras),
            quantization=quantization,
            registry=registry,
            torch_compile=torch_compile,
        )

    def __call__(
        self,
        prompt: str,
        negative_prompt: str,
        seed: int,
        height: int,
        width: int,
        num_frames: int,
        frame_rate: float,
        images: list[ImageConditioningInput],
        tiling_config: TilingConfig | None = None,
        enhance_prompt: bool = False,
        streaming_prefetch_count: int | None = None,
        max_batch_size: int = 1,
        num_inference_steps: int = 8,
        stage_1_sigmas: torch.Tensor | None = None,
        stage_2_sigmas: torch.Tensor = STAGE_2_DISTILLED_SIGMAS,
        video_guider_params: MultiModalGuiderParams | None = None,
        audio_guider_params: MultiModalGuiderParams | None = None,
    ) -> tuple[Iterator[torch.Tensor], Audio]:
        assert_resolution(height=height, width=width, is_two_stage=True)

        generator = torch.Generator(device=self.device).manual_seed(seed)
        noiser = GaussianNoiser(generator=generator)
        dtype = torch.bfloat16

        ctx_p, ctx_n = self.prompt_encoder(
            [prompt, negative_prompt],
            enhance_first_prompt=enhance_prompt,
            enhance_prompt_image=images[0][0] if len(images) > 0 else None,
            enhance_prompt_seed=seed,
            streaming_prefetch_count=streaming_prefetch_count,
        )
        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

        if video_guider_params is None:
            video_guider_params = LTX_2_3_HQ_PARAMS.video_guider_params
        
        if audio_guider_params is None:
            audio_guider_params = LTX_2_3_HQ_PARAMS.audio_guider_params
        
        stage_1_output_shape = VideoPixelShape(
            batch=1, frames=num_frames, width=width // 2, height=height // 2, fps=frame_rate
        )
        stage_1_conditionings = self.image_conditioner(
            lambda enc: combined_image_conditionings(
                images=images,
                height=stage_1_output_shape.height,
                width=stage_1_output_shape.width,
                video_encoder=enc,
                dtype=dtype,
                device=self.device,
            )
        )

        stepper = Res2sDiffusionStep()
        if stage_1_sigmas is None:
            empty_latent = torch.empty(VideoLatentShape.from_pixel_shape(stage_1_output_shape).to_torch_shape())
            stage_1_sigmas = self._scheduler.execute(latent=empty_latent, steps=num_inference_steps)
        sigmas = stage_1_sigmas.to(dtype=torch.float32, device=self.device)

        video_state, audio_state = self.stage_1(
            denoiser=GuidedDenoiser(
                v_context=v_context_p,
                a_context=a_context_p,
                video_guider=MultiModalGuider(
                    params=video_guider_params,
                    negative_context=v_context_n,
                ),
                audio_guider=MultiModalGuider(
                    params=audio_guider_params,
                    negative_context=a_context_n,
                ),
            ),
            sigmas=sigmas,
            noiser=noiser,
            stepper=stepper,
            width=stage_1_output_shape.width,
            height=stage_1_output_shape.height,
            frames=num_frames,
            fps=frame_rate,
            video=ModalitySpec(context=v_context_p, conditionings=stage_1_conditionings),
            audio=ModalitySpec(context=a_context_p),
            loop=res2s_audio_video_denoising_loop,
            streaming_prefetch_count=streaming_prefetch_count,
            max_batch_size=max_batch_size,
        )

        upscaled_video_latent = self.upsampler(video_state.latent[:1])

        stage_2_conditionings = self.image_conditioner(
            lambda enc: combined_image_conditionings(
                images=images,
                height=height,
                width=width,
                video_encoder=enc,
                dtype=dtype,
                device=self.device,
            )
        )

        video_state, audio_state = self.stage_2(
            denoiser=SimpleDenoiser(v_context=v_context_p, a_context=a_context_p),
            sigmas=stage_2_sigmas.to(dtype=torch.float32, device=self.device),
            noiser=noiser,
            stepper=stepper,
            width=width,
            height=height,
            frames=num_frames,
            fps=frame_rate,
            video=ModalitySpec(
                context=v_context_p,
                conditionings=stage_2_conditionings,
                noise_scale=stage_2_sigmas[0].item(),
                initial_latent=upscaled_video_latent,
            ),
            audio=ModalitySpec(
                context=a_context_p,
                noise_scale=stage_2_sigmas[0].item(),
                initial_latent=audio_state.latent,
            ),
            loop=res2s_audio_video_denoising_loop,
            streaming_prefetch_count=streaming_prefetch_count,
        )

        decoded_video = self.video_decoder(video_state.latent, tiling_config, generator)
        decoded_audio = self.audio_decoder(audio_state.latent)
        return decoded_video, decoded_audio


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

# Download model checkpoints
print("=" * 80)
print("Downloading LTX-2.3 distilled model + Gemma...")
print("=" * 80)

# LoRA cache directory and currently-applied key
LORA_CACHE_DIR = Path("lora_cache")
LORA_CACHE_DIR.mkdir(exist_ok=True)

current_lora_key: str | None = None
PENDING_LORA_KEY: str | None = None
PENDING_LORA_LORAS: tuple[LoraPathStrengthAndSDOps, ...] | None = None
PENDING_LORA_STATUS: str = "No LoRA config prepared yet."

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,
)
spatial_upsampler_path = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-spatial-upscaler-x2-1.1.safetensors")
gemma_root = snapshot_download(repo_id=GEMMA_REPO)

# ---- Insert block (LoRA downloads) between lines 268 and 269 ----
# LoRA repo + download the requested LoRA adapters
LORA_REPO = "dagloop5/LoRA"

print("=" * 80)
print("Downloading LoRA adapters from dagloop5/LoRA...")
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") # m15510n4ry, bl0wj0b, d0ubl3_bj, d0gg1e, c0wg1rl
mself_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="Furry Hyper Masturbation - LTX-2 I2V v1.safetensors") # Hyperfap
dramatic_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="LTX-2.3 - Orgasm.safetensors") # "[He | She] is having am orgasm." (am or an?)
fluid_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="cr3ampi3_animation_i2v_ltx2_v1.0.safetensors") # cr3ampi3 animation., missionary animation, doggystyle bouncy animation, double penetration animation
liquid_lora_path = hf_hub_download(repo_id=LORA_REPO, filename="liquid_wet_dr1pp_ltx2_v1.0_scaled.safetensors") # wet dr1pp
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="valiantcat/LTX-2.3-Transition-LORA", filename="ltx2.3-transition.safetensors")

print(f"Pose LoRA: {pose_lora_path}")
print(f"General LoRA: {general_lora_path}")
print(f"Motion LoRA: {motion_lora_path}")
print(f"Dreamlay LoRA: {dreamlay_lora_path}")
print(f"Mself LoRA: {mself_lora_path}")
print(f"Dramatic LoRA: {dramatic_lora_path}")
print(f"Fluid LoRA: {fluid_lora_path}")
print(f"Liquid LoRA: {liquid_lora_path}")
print(f"Demopose LoRA: {demopose_lora_path}")
print(f"Voice LoRA: {voice_lora_path}")
print(f"Realism LoRA: {realism_lora_path}")
print(f"Transition LoRA: {transition_lora_path}")
# ----------------------------------------------------------------

print(f"Checkpoint: {checkpoint_path}")
print(f"Spatial upsampler: {spatial_upsampler_path}")
print(f"Gemma root: {gemma_root}")

# Initialize pipeline WITH text encoder and optional audio support
# ---- Replace block (pipeline init) lines 275-281 ----
pipeline = LTX23NegativePromptTwoStagePipeline(
    checkpoint_path=str(checkpoint_path),
    spatial_upsampler_path=str(spatial_upsampler_path),
    gemma_root=str(gemma_root),
    loras=[],
    quantization=QuantizationPolicy.fp8_cast(),
)
# ----------------------------------------------------------------

def _make_lora_key(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) -> tuple[str, str]:
    rp = round(float(pose_strength), 2)
    rg = round(float(general_strength), 2)
    rm = round(float(motion_strength), 2)
    rd = round(float(dreamlay_strength), 2)
    rs = round(float(mself_strength), 2)
    rr = round(float(dramatic_strength), 2)
    rf = round(float(fluid_strength), 2)
    rl = round(float(liquid_strength), 2)
    ro = round(float(demopose_strength), 2)
    rv = round(float(voice_strength), 2)
    re = round(float(realism_strength), 2)
    rt = round(float(transition_strength), 2)
    key_str = f"{pose_lora_path}:{rp}|{general_lora_path}:{rg}|{motion_lora_path}:{rm}|{dreamlay_lora_path}:{rd}|{mself_lora_path}:{rs}|{dramatic_lora_path}:{rr}|{fluid_lora_path}:{rf}|{liquid_lora_path}:{rl}|{demopose_lora_path}:{ro}|{voice_lora_path}:{rv}|{realism_lora_path}:{re}|{transition_lora_path}:{rt}"
    key = hashlib.sha256(key_str.encode("utf-8")).hexdigest()
    return key, key_str


def prepare_lora_cache(
    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),
):
    """
    Prepare the LoRA selection for the guided pipeline.
    This caches the LoRA config, not fused weights.
    """
    global PENDING_LORA_KEY, PENDING_LORA_LORAS, PENDING_LORA_STATUS

    key = _make_lora_key(
        pose_strength, general_strength, motion_strength, dreamlay_strength,
        mself_strength, dramatic_strength, fluid_strength, liquid_strength,
        demopose_strength, voice_strength, realism_strength, transition_strength
    )
    cache_path = LORA_CACHE_DIR / f"{key}.json"

    progress(0.05, desc="Preparing LoRA config")

    entries = [
        (pose_lora_path, round(float(pose_strength), 2)),
        (general_lora_path, round(float(general_strength), 2)),
        (motion_lora_path, round(float(motion_strength), 2)),
        (dreamlay_lora_path, round(float(dreamlay_strength), 2)),
        (mself_lora_path, round(float(mself_strength), 2)),
        (dramatic_lora_path, round(float(dramatic_strength), 2)),
        (fluid_lora_path, round(float(fluid_strength), 2)),
        (liquid_lora_path, round(float(liquid_strength), 2)),
        (demopose_lora_path, round(float(demopose_strength), 2)),
        (voice_lora_path, round(float(voice_strength), 2)),
        (realism_lora_path, round(float(realism_strength), 2)),
        (transition_lora_path, round(float(transition_strength), 2)),
    ]

    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
    ]

    if not loras_for_builder:
        PENDING_LORA_KEY = None
        PENDING_LORA_LORAS = None
        PENDING_LORA_STATUS = "No non-zero LoRA strengths selected; nothing to prepare."
        return PENDING_LORA_STATUS

    try:
        if cache_path.exists():
            progress(0.20, desc="Loading cached LoRA config")
            data = json.loads(cache_path.read_text())
            loras_for_builder = [
                LoraPathStrengthAndSDOps(item["path"], item["strength"], LTXV_LORA_COMFY_RENAMING_MAP)
                for item in data
                if float(item["strength"]) != 0.0
            ]
        else:
            progress(0.30, desc="Saving LoRA config cache")
            cache_path.write_text(
                json.dumps(
                    [{"path": path, "strength": strength} for path, strength in entries if float(strength) != 0.0],
                    indent=2,
                )
            )

        PENDING_LORA_KEY = key
        PENDING_LORA_LORAS = tuple(loras_for_builder)
        PENDING_LORA_STATUS = f"Prepared LoRA config: {cache_path.name}"
        return PENDING_LORA_STATUS

    except Exception as e:
        import traceback
        print(f"[LoRA] Prepare failed: {type(e).__name__}: {e}")
        print(traceback.format_exc())
        PENDING_LORA_KEY = None
        PENDING_LORA_LORAS = None
        PENDING_LORA_STATUS = f"LoRA prepare failed: {type(e).__name__}: {e}"
        return PENDING_LORA_STATUS


def apply_prepared_lora_config_to_pipeline():
    global current_lora_key, PENDING_LORA_KEY, PENDING_LORA_LORAS, pipeline

    if PENDING_LORA_LORAS is None or PENDING_LORA_KEY is None:
        print("[LoRA] No prepared LoRA config available; skipping.")
        return False

    if current_lora_key == PENDING_LORA_KEY:
        print("[LoRA] Prepared LoRA config already active; skipping.")
        return True

    pipeline = LTX23NegativePromptTwoStagePipeline(
        checkpoint_path=str(checkpoint_path),
        spatial_upsampler_path=str(spatial_upsampler_path),
        gemma_root=str(gemma_root),
        loras=PENDING_LORA_LORAS,
        quantization=QuantizationPolicy.fp8_cast(),
    )

    current_lora_key = PENDING_LORA_KEY
    print("[LoRA] Prepared LoRA config applied by rebuilding the pipeline.")
    return True

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


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 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 = True,
    seed: int = 42,
    randomize_seed: bool = True,
    height: int = 1024,
    width: int = 1536,
    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,
):
    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 = True,
    seed: int = 42,
    randomize_seed: bool = True,
    height: int = 1024,
    width: int = 1536,
    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)

        frame_rate = DEFAULT_FRAME_RATE
        num_frames = int(duration * frame_rate) + 1
        num_frames = ((num_frames - 1 + 7) // 8) * 8 + 1

        print(f"Generating: {height}x{width}, {num_frames} frames ({duration}s), seed={current_seed}")

        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:
                temp_first_path = Path(first_image)
            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:
                temp_last_path = Path(last_image)
            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)

        log_memory("before pipeline call")

        apply_prepared_lora_config_to_pipeline()

        video, audio = pipeline(
            prompt=prompt,
            negative_prompt=negative_prompt,
            seed=current_seed,
            height=int(height),
            width=int(width),
            num_frames=num_frames,
            frame_rate=frame_rate,
            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=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


with gr.Blocks(title="LTX-2.3 Distilled") as demo:
    gr.Markdown("# LTX-2.3 F2LF with Fast Audio-Video Generation with Frame Conditioning")
    

    with gr.Row():
        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",
                info="for best results - make it as elaborate as possible",
                value="Make this image come alive with cinematic motion, smooth animation",
                lines=3,
                placeholder="Describe the motion and animation you want...",
            )
            negative_prompt = gr.Textbox(
                label="Negative Prompt",
                value="",
                lines=2,
                placeholder="Describe what you want to avoid...",
            )
            duration = gr.Slider(label="Duration (seconds)", minimum=1.0, maximum=30.0, value=10.0, step=0.1)
                

            generate_btn = gr.Button("Generate Video", variant="primary", size="lg")

            with gr.Accordion("Advanced Settings", open=False):
                seed = gr.Slider(label="Seed", minimum=0, maximum=MAX_SEED, value=10, step=1)
                randomize_seed = gr.Checkbox(label="Randomize Seed", value=True)
                with gr.Row():
                    width = gr.Number(label="Width", value=1536, precision=0)
                    height = gr.Number(label="Height", value=1024, precision=0)
                with gr.Row():
                    enhance_prompt = gr.Checkbox(label="Enhance Prompt", value=False)
                    high_res = gr.Checkbox(label="High Resolution", value=True)
                with gr.Column():
                    gr.Markdown("### LoRA adapter strengths (set to 0 to disable; slow and WIP)")
                    pose_strength = gr.Slider(
                        label="Anthro Enhancer strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    general_strength = gr.Slider(
                        label="Reasoning Enhancer strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    motion_strength = gr.Slider(
                        label="Anthro Posing Helper strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    dreamlay_strength = gr.Slider(
                        label="Dreamlay strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    mself_strength = gr.Slider(
                        label="Mself strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    dramatic_strength = gr.Slider(
                        label="Dramatic strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    fluid_strength = gr.Slider(
                        label="Fluid Helper strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    liquid_strength = gr.Slider(
                        label="Liquid Helper strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    demopose_strength = gr.Slider(
                        label="Audio Helper strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    voice_strength = gr.Slider(
                        label="Voice Helper strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    realism_strength = gr.Slider(
                        label="Anthro Realism strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                    transition_strength = gr.Slider(
                        label="Transition strength",
                        minimum=0.0, maximum=2.0, value=0.0, step=0.01
                    )
                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,
                )

        with gr.Column():
            output_video = gr.Video(label="Generated Video", autoplay=False)
            gpu_duration = gr.Slider(
                label="ZeroGPU duration (seconds; 10 second Img2Vid with 1024x1024 and LoRAs = ~70)",
                minimum=30.0,
                maximum=240.0,
                value=75.0,
                step=1.0,
            )

    gr.Examples(
        examples=[
            [
                None,
                "pinkknit.jpg",
                "The camera falls downward through darkness as if dropped into a tunnel. "
                "As it slows, five friends wearing pink knitted hats and sunglasses lean "
                "over and look down toward the camera with curious expressions. The lens "
                "has a strong fisheye effect, creating a circular frame around them. They "
                "crowd together closely, forming a symmetrical cluster while staring "
                "directly into the lens.",
                "",
                3.0,
                80.0,
                False,
                42,
                True,
                1024,
                1024,
                0.0,  # pose_strength (example)
                0.0,  # general_strength (example)
                0.0,  # motion_strength (example)
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
            ],
        ],
        inputs=[
            first_image, last_image, prompt, negative_prompt, duration, gpu_duration,
            enhance_prompt, seed, randomize_seed, height, width,
            pose_strength, general_strength, motion_strength, dreamlay_strength, mself_strength, dramatic_strength, fluid_strength, liquid_strength, demopose_strength, voice_strength, realism_strength, transition_strength,
        ],
    )

    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=[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, enhance_prompt,
            seed, randomize_seed, height, width,
            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],
    )


css = """
.fillable{max-width: 1200px !important}
"""

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