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"

# 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")

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

# Install ltx-core and ltx-pipelines if not already installed
try:
    import ltx_pipelines  # noqa: F401
except ImportError:
    print("Installing ltx-core and ltx-pipelines...")
    subprocess.run(
        [sys.executable, "-m", "pip", "install", "-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

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 ltx_core.model.video_vae import TilingConfig, get_video_chunks_number
from ltx_core.quantization import QuantizationPolicy
from ltx_pipelines.distilled import DistilledPipeline
from ltx_pipelines.utils.args import ImageConditioningInput
from ltx_pipelines.utils.media_io import encode_video

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."
)
DEFAULT_HEIGHT = 1024
DEFAULT_WIDTH = 1536
DEFAULT_FRAME_RATE = 24.0

# Download models from Hugging Face
LTX_MODEL_REPO = "diffusers-internal-dev/ltx-23"
GEMMA_MODEL_REPO = "google/gemma-3-12b-it-qat-q4_0-unquantized"

print("=" * 80)
print("Downloading models from Hugging Face...")
print("=" * 80)

DISTILLED_CHECKPOINT = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-22b-distilled.safetensors")
SPATIAL_UPSAMPLER = hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-spatial-upscaler-x2-1.0.safetensors")
GEMMA_ROOT = snapshot_download(repo_id=GEMMA_MODEL_REPO)

print(f"Distilled checkpoint: {DISTILLED_CHECKPOINT}")
print(f"Spatial upsampler: {SPATIAL_UPSAMPLER}")
print(f"Gemma root: {GEMMA_ROOT}")

# Initialize pipeline
print("=" * 80)
print("Loading LTX-2.3 Distilled pipeline...")
print("=" * 80)

pipeline = DistilledPipeline(
    distilled_checkpoint_path=DISTILLED_CHECKPOINT,
    spatial_upsampler_path=SPATIAL_UPSAMPLER,
    gemma_root=GEMMA_ROOT,
    loras=[],
    quantization=QuantizationPolicy.fp8_cast(),
)

# Preload all models so first request is fast.
# On ZeroGPU, .to('cuda') is intercepted and actual GPU allocation
# happens inside the @spaces.GPU decorated function.
print("Preloading models...")
ledger = pipeline.model_ledger
_text_encoder = ledger.text_encoder()
_transformer = ledger.transformer()
_video_encoder = ledger.video_encoder()
_video_decoder = ledger.video_decoder()
_audio_decoder = ledger.audio_decoder()
_vocoder = ledger.vocoder()
_spatial_upsampler = ledger.spatial_upsampler()

ledger.text_encoder = lambda: _text_encoder
ledger.transformer = lambda: _transformer
ledger.video_encoder = lambda: _video_encoder
ledger.video_decoder = lambda: _video_decoder
ledger.audio_decoder = lambda: _audio_decoder
ledger.vocoder = lambda: _vocoder
ledger.spatial_upsampler = lambda: _spatial_upsampler

print("All models preloaded!")


@spaces.GPU(duration=300)
@torch.inference_mode()
def generate_video(
    input_image,
    prompt: str,
    duration: float,
    enhance_prompt: bool,
    seed: int,
    randomize_seed: bool,
    height: int,
    width: int,
    progress=gr.Progress(track_tqdm=True),
):
    current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)
    num_frames = int(duration * DEFAULT_FRAME_RATE) + 1
    num_frames = ((num_frames - 1 + 7) // 8) * 8 + 1

    images = []
    if input_image is not None:
        with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as f:
            temp_path = f.name
            if hasattr(input_image, "save"):
                input_image.save(temp_path)
            else:
                from shutil import copy2
                copy2(str(input_image), temp_path)
            images = [ImageConditioningInput(path=temp_path, frame_idx=0, strength=1.0)]

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

    video, audio = pipeline(
        prompt=prompt,
        seed=current_seed,
        height=int(height),
        width=int(width),
        num_frames=num_frames,
        frame_rate=DEFAULT_FRAME_RATE,
        images=images,
        tiling_config=tiling_config,
        enhance_prompt=enhance_prompt,
    )

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

    return output_path, current_seed


with gr.Blocks(title="LTX-2.3 Distilled") as demo:
    gr.Markdown("# LTX-2.3 Distilled (22B): Fast Audio-Video Generation")
    gr.Markdown(
        "Fast video + audio generation using the distilled model (8 steps stage 1, 4 steps stage 2). "
        "[[model]](https://huggingface.co/Lightricks/LTX-2) "
        "[[code]](https://github.com/Lightricks/LTX-2)"
    )

    with gr.Row():
        with gr.Column():
            input_image = gr.Image(label="Input Image (Optional)", type="pil")
            prompt = gr.Textbox(
                label="Prompt",
                value=DEFAULT_PROMPT,
                lines=3,
                placeholder="Describe the video you want to generate...",
            )
            with gr.Row():
                duration = gr.Slider(label="Duration (seconds)", minimum=1.0, maximum=10.0, value=5.0, step=0.5)
                enhance_prompt = gr.Checkbox(label="Enhance Prompt", value=True)

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

            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=DEFAULT_WIDTH, precision=0)
                    height = gr.Number(label="Height", value=DEFAULT_HEIGHT, precision=0)

        with gr.Column():
            output_video = gr.Video(label="Generated Video", autoplay=True)

    generate_btn.click(
        fn=generate_video,
        inputs=[
            input_image, prompt, duration, enhance_prompt,
            seed, randomize_seed, height, width,
        ],
        outputs=[output_video, seed],
    )


if __name__ == "__main__":
    demo.launch(share=True)