"""
LTX-2.3 Turbo — ZeroGPU Edition
Generates synchronized audio-video content using Lightricks/LTX-2.3 on
free ZeroGPU hardware via Hugging Face Spaces.
UI inspired by alexnasa/ltx-2-TURBO with full feature parity for LTX-2.3:
- Image-to-Video mode (first frame conditioning)
- Interpolate mode (first + last frame)
- Audio input (user provides audio for lip-sync/soundtrack)
- Custom UI components (RadioAnimated, PromptBox, CameraDropdown, AudioDropUpload)
- Duration presets (2s, 3s, 5s, 6s, 8s, 10s, 12s) and resolution selector with SVG icons
Architecture (following alexnasa/ltx-2-TURBO's proven ZeroGPU pattern):
1. Vendored ltx-core and ltx-pipelines added to sys.path before any imports.
2. Model files downloaded at module startup (CPU, no GPU lease).
3. ModelLedger constructed at module level (CPU-only dataclass, no CUDA init).
4. Text encoder loaded at module level (kept in memory for reuse).
5. DistilledPipeline constructed with gemma_root=None (no text encoder in pipeline).
6. Video encoder and transformer pre-loaded at module level via pipeline cache.
7. @spaces.GPU() on encode_prompt — encodes text, returns .detach().cpu() tensors.
8. @spaces.GPU(duration=callable) on generate_video — runs pipeline with pre-encoded
contexts passed as video_context/audio_context kwargs.
9. FP8 quantization fits the 22B transformer on ZeroGPU's A10G (40GB VRAM).
Based on the official LTX-2 codebase: https://github.com/Lightricks/LTX-2
Architecture inspired by alexnasa/ltx-2-TURBO.
"""
# ───────────────────────────────────────────────────────────────────────────
# 0) Add vendored packages to sys.path BEFORE any ltx imports
# ───────────────────────────────────────────────────────────────────────────
import sys
from pathlib import Path
_here = Path(__file__).parent
sys.path.insert(0, str(_here / "packages" / "ltx-pipelines" / "src"))
sys.path.insert(0, str(_here / "packages" / "ltx-core" / "src"))
# ───────────────────────────────────────────────────────────────────────────
# Standard library & third-party imports
# ───────────────────────────────────────────────────────────────────────────
import json
import logging
import os
import random
import subprocess
import tempfile
import time
import traceback
import uuid
from typing import Any
import gradio as gr
import numpy as np
import spaces
import torch
import torch.nn.functional as F
import torchaudio
from huggingface_hub import hf_hub_download, snapshot_download
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ───────────────────────────────────────────────────────────────────────────
# LTX imports (from vendored packages)
# ───────────────────────────────────────────────────────────────────────────
from ltx_core.model.video_vae import TilingConfig
from ltx_core.quantization import QuantizationPolicy
from ltx_pipelines.distilled import DistilledPipeline
from ltx_pipelines.utils import ModelLedger
from ltx_pipelines.utils.args import ImageConditioningInput
from ltx_pipelines.utils.helpers import generate_enhanced_prompt
# ───────────────────────────────────────────────────────────────────────────
# Constants
# ───────────────────────────────────────────────────────────────────────────
MAX_SEED = np.iinfo(np.int32).max
LTX_REPO = "Lightricks/LTX-2.3"
GEMMA_REPO = "google/gemma-3-12b-it-qat-q4_0-unquantized"
CKPT_DISTILLED = "ltx-2.3-22b-distilled.safetensors"
CKPT_UPSCALER = "ltx-2.3-spatial-upscaler-x2-1.0.safetensors"
RESOLUTION_MAP = {
"16:9": (768, 512),
"1:1": (512, 512),
"9:16": (512, 768),
}
# ───────────────────────────────────────────────────────────────────────────
# Audio helper functions (ported from alexnasa/ltx-2-TURBO)
# ───────────────────────────────────────────────────────────────────────────
def _coerce_audio_path(audio_path: Any) -> str:
"""Handle Gradio's various audio path formats (tuple, dict, string)."""
if isinstance(audio_path, tuple) and len(audio_path) > 0:
audio_path = audio_path[0]
if isinstance(audio_path, dict):
audio_path = audio_path.get("name") or audio_path.get("path")
if not isinstance(audio_path, (str, bytes, os.PathLike)):
raise TypeError(
f"audio_path must be a path-like, got {type(audio_path)}: {audio_path}"
)
return os.fspath(audio_path)
def match_audio_to_duration(
audio_path: str,
target_seconds: float,
target_sr: int = 48000,
to_mono: bool = True,
pad_mode: str = "silence",
device: str = "cuda",
):
"""
Load audio, resample, (optionally) mono, then trim/pad to exactly target_seconds.
Returns: (waveform tensor, sample_rate)
"""
audio_path = _coerce_audio_path(audio_path)
wav, sr = torchaudio.load(audio_path) # [C, T] float32 CPU
if sr != target_sr:
wav = torchaudio.functional.resample(wav, sr, target_sr)
sr = target_sr
if to_mono and wav.shape[0] > 1:
wav = wav.mean(dim=0, keepdim=True)
target_len = int(round(target_seconds * sr))
cur_len = wav.shape[-1]
if cur_len > target_len:
wav = wav[..., :target_len]
elif cur_len < target_len:
pad_len = target_len - cur_len
if pad_mode == "repeat" and cur_len > 0:
reps = (target_len + cur_len - 1) // cur_len
wav = wav.repeat(1, reps)[..., :target_len]
else:
wav = F.pad(wav, (0, pad_len))
wav = wav.to(device, non_blocking=True)
return wav, sr
# ───────────────────────────────────────────────────────────────────────────
# 1) Download model files at module startup (CPU, no GPU lease)
# ───────────────────────────────────────────────────────────────────────────
logger.info("Downloading LTX model files...")
checkpoint_path = hf_hub_download(repo_id=LTX_REPO, filename=CKPT_DISTILLED)
logger.info(f" Distilled checkpoint: {checkpoint_path}")
spatial_upsampler_path = hf_hub_download(repo_id=LTX_REPO, filename=CKPT_UPSCALER)
logger.info(f" Upscaler: {spatial_upsampler_path}")
logger.info("Downloading Gemma text encoder...")
HF_TOKEN = os.environ.get("HF_TOKEN")
gemma_root = snapshot_download(repo_id=GEMMA_REPO, token=HF_TOKEN)
logger.info(f" Gemma root: {gemma_root}")
logger.info("All model files ready on disk.")
# ───────────────────────────────────────────────────────────────────────────
# 2) Construct ModelLedger (CPU — no model weights loaded to GPU)
# ───────────────────────────────────────────────────────────────────────────
logger.info("Constructing ModelLedger (with Gemma for text encoding)...")
fp8_quantization = QuantizationPolicy.fp8_cast()
model_ledger = ModelLedger(
dtype=torch.bfloat16,
device="cuda",
checkpoint_path=checkpoint_path,
gemma_root_path=gemma_root,
spatial_upsampler_path=spatial_upsampler_path,
loras=(),
quantization=fp8_quantization,
)
logger.info("ModelLedger constructed.")
# ───────────────────────────────────────────────────────────────────────────
# 3) Load text encoder at module level (kept in memory for reuse)
# ───────────────────────────────────────────────────────────────────────────
logger.info("Loading Gemma text encoder...")
text_encoder = model_ledger.text_encoder()
logger.info("Text encoder loaded and ready!")
# ───────────────────────────────────────────────────────────────────────────
# 4) Construct DistilledPipeline WITHOUT text encoder (gemma_root=None)
# ───────────────────────────────────────────────────────────────────────────
logger.info("Constructing DistilledPipeline (gemma_root=None)...")
pipeline = DistilledPipeline(
device=torch.device("cuda"),
checkpoint_path=checkpoint_path,
spatial_upsampler_path=spatial_upsampler_path,
gemma_root=None,
loras=[],
quantization=fp8_quantization,
)
# ───────────────────────────────────────────────────────────────────────────
# 5) Pre-load video encoder and transformer at module level
# ───────────────────────────────────────────────────────────────────────────
logger.info("Pre-loading video encoder and transformer...")
pipeline._video_encoder = pipeline.model_ledger.video_encoder()
pipeline._transformer = pipeline.model_ledger.transformer()
logger.info("=" * 60)
logger.info("Pipeline fully loaded and ready!")
logger.info("=" * 60)
# ───────────────────────────────────────────────────────────────────────────
# Helpers
# ───────────────────────────────────────────────────────────────────────────
def calc_frames(duration: float, fps: float) -> int:
"""Compute num_frames = 8k + 1, frames >= 9."""
raw = int(duration * fps) + 1
raw = max(raw, 9)
k = (raw - 1 + 7) // 8
return k * 8 + 1
def encode_text_simple(te, prompt: str):
"""Simple text encoding without using pipeline_utils."""
hidden_states, attention_mask = te.encode(prompt)
embeddings_processor = model_ledger.gemma_embeddings_processor()
result = embeddings_processor.process_hidden_states(hidden_states, attention_mask)
del embeddings_processor
return result.video_encoding, result.audio_encoding
def apply_resolution(resolution: str):
w, h = RESOLUTION_MAP.get(resolution, (768, 512))
return int(w), int(h)
def apply_duration(duration_str: str):
return int(duration_str[:-1])
def on_mode_change(selected: str):
is_interpolate = selected == "Interpolate"
return gr.update(visible=is_interpolate)
def get_duration(
first_frame,
end_frame,
prompt,
duration,
generation_mode,
enhance_prompt,
seed,
randomize_seed,
height,
width,
audio_path,
*args,
**kwargs,
):
"""Estimate GPU lease duration for @spaces.GPU(duration=...)."""
extra_time = 0
if audio_path is not None:
extra_time += 10
dur = float(duration)
if dur <= 6:
return 200 + extra_time
elif dur <= 8:
return 250 + extra_time
elif dur <= 10:
return 300 + extra_time
else:
return 350 + extra_time
# ───────────────────────────────────────────────────────────────────────────
# Phase 1: Text Encoding (separate GPU lease)
# ───────────────────────────────────────────────────────────────────────────
@spaces.GPU()
def encode_prompt(
prompt: str,
enhance_prompt: bool = True,
input_image=None,
seed: int = 42,
):
"""
Encode prompt using the module-level text_encoder + embeddings_processor.
Returns a dict with video_context and audio_context tensors on CPU.
"""
logger.info(f"[encode_prompt] prompt='{prompt[:80]}...', enhance={enhance_prompt}")
final_prompt = prompt
if enhance_prompt:
final_prompt = generate_enhanced_prompt(
text_encoder=text_encoder,
prompt=prompt,
image_path=input_image if input_image is not None else None,
seed=seed,
)
logger.info(f"[encode_prompt] Enhanced prompt: '{final_prompt[:120]}...'")
with torch.inference_mode():
video_context, audio_context = encode_text_simple(text_encoder, final_prompt)
embedding_data = {
"video_context": video_context.detach().cpu(),
"audio_context": audio_context.detach().cpu(),
"prompt": final_prompt,
}
logger.info("[encode_prompt] Done.")
return embedding_data, final_prompt
# ───────────────────────────────────────────────────────────────────────────
# Phase 2: Video Generation (separate GPU lease, dynamic duration)
# ───────────────────────────────────────────────────────────────────────────
@spaces.GPU(duration=get_duration)
def generate_video(
first_frame,
end_frame,
prompt: str,
duration: float,
generation_mode: str = "Image-to-Video",
enhance_prompt: bool = True,
seed: int = 42,
randomize_seed: bool = True,
height: int = 512,
width: int = 768,
audio_path=None,
progress=gr.Progress(track_tqdm=True),
):
"""
Full generation: encode prompt then run pipeline with pre-encoded contexts.
Supports Image-to-Video, Interpolate, and audio input modes.
"""
if not prompt or not prompt.strip():
raise gr.Error("Please enter a prompt.")
current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)
num_frames = calc_frames(duration, 24.0)
frame_rate = 24.0
logger.info(
f"[generate_video] mode={generation_mode}, seed={current_seed}, {width}x{height}, "
f"frames={num_frames}, duration={duration}s, enhance={enhance_prompt}, "
f"audio={'yes' if audio_path else 'no'}"
)
# --- Handle input images ---
images = []
image_path_for_enhance = None
if first_frame is not None:
# first_frame is filepath from gr.Image(type="filepath")
if isinstance(first_frame, str):
img_path = first_frame
else:
tmp_dir = tempfile.mkdtemp()
img_path = os.path.join(tmp_dir, f"input_{int(time.time())}.png")
if hasattr(first_frame, "save"):
first_frame.save(img_path)
else:
from PIL import Image as PILImage
PILImage.open(first_frame).save(img_path)
images.append((img_path, 0, 1.0))
image_path_for_enhance = img_path
# Interpolation: add end frame as guiding latent
if generation_mode == "Interpolate" and end_frame is not None:
if isinstance(end_frame, str):
end_path = end_frame
else:
tmp_dir = tempfile.mkdtemp()
end_path = os.path.join(tmp_dir, f"end_{int(time.time())}.png")
if hasattr(end_frame, "save"):
end_frame.save(end_path)
else:
from PIL import Image as PILImage
PILImage.open(end_frame).save(end_path)
end_idx = max(0, num_frames - 1)
images.append((end_path, end_idx, 0.5))
t0 = time.time()
try:
# Phase 1: Encode prompt
embeddings, final_prompt = encode_prompt(
prompt=prompt,
enhance_prompt=enhance_prompt,
input_image=image_path_for_enhance,
seed=current_seed,
)
video_context = embeddings["video_context"].to("cuda", non_blocking=True)
audio_context = embeddings["audio_context"].to("cuda", non_blocking=True)
del embeddings
torch.cuda.empty_cache()
# If user provided audio, use a neutral audio_context (encode empty prompt)
if audio_path is not None:
with torch.inference_mode():
_, neutral_audio_context = encode_text_simple(text_encoder, "")
del audio_context
audio_context = neutral_audio_context
# Prepare audio waveform if provided
input_waveform = None
input_waveform_sample_rate = None
if audio_path is not None:
video_seconds = (num_frames - 1) / frame_rate
input_waveform, input_waveform_sample_rate = match_audio_to_duration(
audio_path=audio_path,
target_seconds=video_seconds,
target_sr=48000,
to_mono=True,
pad_mode="silence",
device="cuda",
)
torch.cuda.empty_cache()
# Phase 2: Run pipeline with pre-encoded contexts
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmpfile:
output_path = tmpfile.name
with torch.inference_mode():
pipeline(
prompt=prompt,
output_path=output_path,
seed=current_seed,
height=height,
width=width,
num_frames=num_frames,
frame_rate=frame_rate,
images=images,
tiling_config=TilingConfig.default(),
video_context=video_context,
audio_context=audio_context,
input_waveform=input_waveform,
input_waveform_sample_rate=input_waveform_sample_rate,
)
del video_context, audio_context
if input_waveform is not None:
del input_waveform
torch.cuda.empty_cache()
elapsed = time.time() - t0
logger.info(f"[generate_video] Done in {elapsed:.1f}s")
except torch.cuda.OutOfMemoryError:
elapsed = time.time() - t0
logger.error(f"OOM after {elapsed:.1f}s")
raise gr.Error("Out of GPU memory. Try a shorter duration or lower resolution.")
except Exception as e:
elapsed = time.time() - t0
tb = traceback.format_exc()
logger.error(f"Generation failed after {elapsed:.1f}s:\n{tb}")
raise gr.Error(f"Generation failed: {type(e).__name__}: {e}")
# Build metadata HTML
meta_parts = [
f'Seed {current_seed}',
f'Resolution {width}×{height}',
f'Duration {duration}s',
f'Time {elapsed:.1f}s',
]
meta_html = '
'
meta_html += '
' + "".join(meta_parts) + "
"
if enhance_prompt and final_prompt and final_prompt != prompt:
escaped = (
final_prompt.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
.replace('"', """)
)
meta_html += (
'
'
'Enhanced Prompt'
f'
{escaped}
'
"
"
)
meta_html += "
"
return str(output_path), meta_html
# ───────────────────────────────────────────────────────────────────────────
# Custom UI Components (ported from alexnasa/ltx-2-TURBO)
# ───────────────────────────────────────────────────────────────────────────
class RadioAnimated(gr.HTML):
"""Animated segmented radio (like iOS pill selector)."""
def __init__(self, choices, value=None, **kwargs):
if not choices or len(choices) < 2:
raise ValueError("RadioAnimated requires at least 2 choices.")
if value is None:
value = choices[0]
uid = uuid.uuid4().hex[:8]
group_name = f"ra-{uid}"
inputs_html = "\n".join(
f''
f''
for i, c in enumerate(choices)
)
html_template = f"""
{inputs_html}
"""
js_on_load = r"""
(() => {
const wrap = element.querySelector('.ra-wrap');
const inner = element.querySelector('.ra-inner');
const highlight = element.querySelector('.ra-highlight');
const inputs = Array.from(element.querySelectorAll('.ra-input'));
const labels = Array.from(element.querySelectorAll('.ra-label'));
if (!inputs.length || !labels.length) return;
const choices = inputs.map(i => i.value);
const PAD = 6;
let currentIdx = 0;
function setHighlightByIndex(idx) {
currentIdx = idx;
const lbl = labels[idx];
if (!lbl) return;
const innerRect = inner.getBoundingClientRect();
const lblRect = lbl.getBoundingClientRect();
highlight.style.width = `${lblRect.width}px`;
const x = (lblRect.left - innerRect.left - PAD);
highlight.style.transform = `translateX(${x}px)`;
}
function setCheckedByValue(val, shouldTrigger=false) {
const idx = Math.max(0, choices.indexOf(val));
inputs.forEach((inp, i) => { inp.checked = (i === idx); });
requestAnimationFrame(() => setHighlightByIndex(idx));
props.value = choices[idx];
if (shouldTrigger) trigger('change', props.value);
}
setCheckedByValue(props.value ?? choices[0], false);
inputs.forEach((inp) => {
inp.addEventListener('change', () => setCheckedByValue(inp.value, true));
});
window.addEventListener('resize', () => setHighlightByIndex(currentIdx));
let last = props.value;
const syncFromProps = () => {
if (props.value !== last) {
last = props.value;
setCheckedByValue(last, false);
}
requestAnimationFrame(syncFromProps);
};
requestAnimationFrame(syncFromProps);
})();
"""
super().__init__(
value=value,
html_template=html_template,
js_on_load=js_on_load,
**kwargs,
)
class PromptBox(gr.HTML):
"""Prompt textarea with an internal footer slot for embedding dropdowns."""
def __init__(self, value="", placeholder="Describe what you want...", **kwargs):
uid = uuid.uuid4().hex[:8]
html_template = f"""
"""
js_on_load = r"""
(() => {
const textarea = element.querySelector(".ds-textarea");
if (!textarea) return;
const autosize = () => {
textarea.style.height = "0px";
textarea.style.height = Math.min(textarea.scrollHeight, 240) + "px";
};
const setValue = (v, triggerChange=false) => {
const val = (v ?? "");
if (textarea.value !== val) textarea.value = val;
autosize();
props.value = textarea.value;
if (triggerChange) trigger("change", props.value);
};
setValue(props.value, false);
textarea.addEventListener("input", () => {
autosize();
props.value = textarea.value;
trigger("change", props.value);
});
const shouldAutoFocus = () => {
const ae = document.activeElement;
if (ae && ae !== document.body && ae !== document.documentElement) return false;
if (window.matchMedia && window.matchMedia("(max-width: 768px)").matches) return false;
return true;
};
const focusWithRetry = (tries = 30) => {
if (!shouldAutoFocus()) return;
if (document.activeElement !== textarea) textarea.focus({ preventScroll: true });
if (document.activeElement === textarea) return;
if (tries > 0) requestAnimationFrame(() => focusWithRetry(tries - 1));
};
requestAnimationFrame(() => focusWithRetry());
let last = props.value;
const syncFromProps = () => {
if (props.value !== last) {
last = props.value;
setValue(last, false);
}
requestAnimationFrame(syncFromProps);
};
requestAnimationFrame(syncFromProps);
})();
"""
super().__init__(
value=value,
html_template=html_template,
js_on_load=js_on_load,
**kwargs,
)
class CameraDropdown(gr.HTML):
"""Custom dropdown with optional icons per item."""
def __init__(self, choices, value="None", title="Dropdown", **kwargs):
if not choices:
raise ValueError("CameraDropdown requires choices.")
norm = []
for c in choices:
if isinstance(c, dict):
label = str(c.get("label", c.get("value", "")))
val = str(c.get("value", label))
icon = c.get("icon", None)
norm.append({"label": label, "value": val, "icon": icon})
else:
s = str(c)
norm.append({"label": s, "value": s, "icon": None})
uid = uuid.uuid4().hex[:8]
def render_item(item):
icon_html = ""
if item["icon"]:
icon_html = f'{item["icon"]}'
return (
f'"
)
items_html = "\n".join(render_item(item) for item in norm)
html_template = f"""