Spaces:

Opera8
/

LTX-2-3

Paused

Transfer Bot

Moved to Hugging Face automatically

87d4eda 7 days ago

15.8 kB

import importlib.util
import os
import shutil
import subprocess
import sys

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

# --------------------------------------------------------------------------- #
# Bootstrap: xformers + LTX-2 packages
# --------------------------------------------------------------------------- #
# xformers is pinned in requirements.txt (installed at *build* time). This is only
# a safety net so we do NOT re-download ~117 MB on every restart.
if importlib.util.find_spec("xformers") is None:
    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_SHA = "ae855f8538843825f9015a419cf4ba5edaf5eec2"


def ensure_ltx_repo() -> None:
    """Clone LTX-2 at the pinned commit. Re-clones if a previous clone was partial."""
    marker = os.path.join(LTX_REPO_DIR, "packages", "ltx-core", "pyproject.toml")
    if os.path.exists(marker):
        return
    if os.path.exists(LTX_REPO_DIR):
        print(f"Removing incomplete LTX-2 checkout at {LTX_REPO_DIR}")
        shutil.rmtree(LTX_REPO_DIR, ignore_errors=True)

    print(f"Cloning {LTX_REPO_URL}...")
    os.makedirs(LTX_REPO_DIR)
    subprocess.run(["git", "init", LTX_REPO_DIR], check=True)
    subprocess.run(["git", "remote", "add", "origin", LTX_REPO_URL], cwd=LTX_REPO_DIR, check=True)
    subprocess.run(["git", "fetch", "--depth", "1", "origin", LTX_COMMIT_SHA], cwd=LTX_REPO_DIR, check=True)
    subprocess.run(["git", "checkout", LTX_COMMIT_SHA], cwd=LTX_REPO_DIR, check=True)


ensure_ltx_repo()

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 torch
torch._dynamo.config.suppress_errors = True
torch._dynamo.config.disable = True

import spaces
import gradio as gr
import gradio.queueing


# Gradio 6.x bug workaround: Queue.send_message raises KeyError when an event's
# session is no longer in pending_messages_per_session (an LRUCache capped at 2000
# sessions, also cleared on client disconnect). If that happens inside
# Queue.start_processing (via broadcast_estimations), its `finally` sets
# queue.stopped = True permanently and every request gets "Queue is stopped."
# until the Space restarts. Drop the orphaned event instead of killing the queue.
_original_send_message = gradio.queueing.Queue.send_message


def _send_message_tolerating_lost_sessions(self, event, event_message):
    try:
        _original_send_message(self, event, event_message)
    except KeyError:
        event.alive = False


gradio.queueing.Queue.send_message = _send_message_tolerating_lost_sessions

import numpy as np

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

from helpers import (
    DEFAULT_FRAME_RATE,
    compute_num_frames,
    pick_resolution,
    snap_dimension,
)
from model_resolver import ModelSetupError, resolve_model_paths

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

# Disable xformers FA3 dispatch: FA3 kernels are Hopper-only (sm_90a), but
# xformers' dispatcher gates them on `device_capability >= (9, 0)`, which also
# matches Blackwell (RTX PRO 6000, the ZeroGPU fleet hardware since 2026-05-12)
# and crashes at kernel launch with "invalid argument".
try:
    from xformers.ops.fmha import _set_use_fa3
    _set_use_fa3(False)
    print("[ATTN] xformers FA3 dispatch disabled (Blackwell-incompatible)")
except Exception as e:
    print(f"[ATTN] FA3 disable FAILED: {type(e).__name__}: {e}")

# FUSE/mmap workaround: SafetensorsStateDictLoader.load uses safetensors.safe_open
# under the hood, which mmap's the file. On bucket FUSE mounts that triggers a
# page-fault storm and deadlocks loading. Bypass mmap by parsing the safetensors
# header ourselves and reading each tensor's bytes directly.
import json
import struct

from ltx_core.loader.primitives import StateDict
from ltx_core.loader.sft_loader import SafetensorsStateDictLoader

_SAFETENSORS_DTYPE_MAP = {
    "F64": torch.float64,
    "F32": torch.float32,
    "F16": torch.float16,
    "BF16": torch.bfloat16,
    "F8_E5M2": torch.float8_e5m2,
    "F8_E4M3": torch.float8_e4m3fn,
    "I64": torch.int64,
    "I32": torch.int32,
    "I16": torch.int16,
    "I8": torch.int8,
    "U8": torch.uint8,
    "BOOL": torch.bool,
}


def _patched_load(self, path, sd_ops, device=None):
    sd = {}
    size = 0
    dtype = set()
    device = device or torch.device("cpu")
    model_paths = path if isinstance(path, list) else [path]
    for shard_path in model_paths:
        with open(shard_path, "rb") as f:
            header_len = struct.unpack("<Q", f.read(8))[0]
            header = json.loads(f.read(header_len).decode("utf-8"))
            data_base = 8 + header_len
            for name, meta in header.items():
                if name == "__metadata__":
                    continue
                expected_name = name if sd_ops is None else sd_ops.apply_to_key(name)
                if expected_name is None:
                    continue
                start, end = meta["data_offsets"]
                f.seek(data_base + start)
                buf = f.read(end - start)
                t = torch.frombuffer(
                    bytearray(buf), dtype=_SAFETENSORS_DTYPE_MAP[meta["dtype"]]
                ).reshape(meta["shape"])
                t = t.to(device=device, non_blocking=True, copy=False)
                kvs = (
                    ((expected_name, t),)
                    if sd_ops is None
                    else sd_ops.apply_to_key_value(expected_name, t)
                )
                for key, v in kvs:
                    size += v.nbytes
                    dtype.add(v.dtype)
                    sd[key] = v
    return StateDict(sd=sd, device=device, size=size, dtype=dtype)


SafetensorsStateDictLoader.load = _patched_load
print("[FUSE-PATCH] SafetensorsStateDictLoader.load replaced (chunked-read)")

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

# --------------------------------------------------------------------------- #
# Locate the models (mounted volumes -> HF cache -> Hub download).
# Raises a descriptive ModelSetupError instead of the cryptic
# "No files matching pattern 'tokenizer.model'" from inside ltx-core.
# --------------------------------------------------------------------------- #
try:
    model_paths = resolve_model_paths()
except ModelSetupError as exc:
    print("=" * 80)
    print("MODEL SETUP ERROR")
    print(exc)
    print("=" * 80)
    raise

# Initialize pipeline WITH text encoder
pipeline = DistilledPipeline(
    distilled_checkpoint_path=model_paths.distilled_checkpoint,
    spatial_upsampler_path=model_paths.spatial_upsampler,
    gemma_root=model_paths.gemma_root,
    loras=[],
    quantization=QuantizationPolicy.fp8_cast(),
)

# Preload all models for ZeroGPU tensor packing.
print("Preloading all models (including Gemma)...")
ledger = pipeline.model_ledger
_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()
_text_encoder = ledger.text_encoder()
_embeddings_processor = ledger.gemma_embeddings_processor()

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
ledger.text_encoder = lambda: _text_encoder
ledger.gemma_embeddings_processor = lambda: _embeddings_processor
print("All models preloaded (including Gemma text encoder)!")

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 update_resolution(image, high_res):
    """Auto-set width/height from the uploaded image's aspect ratio + quality tier."""
    w, h = pick_resolution(image, high_res)
    return gr.update(value=w), gr.update(value=h)


@spaces.GPU(duration=75)
@torch.inference_mode()
def generate_video(
    input_image,
    prompt: str,
    duration: float,
    enhance_prompt: bool = True,
    seed: int = 42,
    randomize_seed: bool = True,
    height: int = 1024,
    width: int = 1536,
    progress=gr.Progress(track_tqdm=True),
):
    # Defined first so the error path can never hit an unbound name.
    current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)
    temp_image_path = None

    try:
        if not prompt or not prompt.strip():
            raise gr.Error("Please enter a prompt.")

        torch.cuda.reset_peak_memory_stats()
        log_memory("start")

        frame_rate = DEFAULT_FRAME_RATE
        num_frames = compute_num_frames(duration, frame_rate)
        # Free-form boxes: snap to multiples of 64 or the two-stage pipeline crashes.
        height = snap_dimension(height)
        width = snap_dimension(width)

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

        images = []
        if input_image is not None:
            output_dir = Path("outputs")
            output_dir.mkdir(exist_ok=True)
            if hasattr(input_image, "save"):
                # PNG + RGB: saving an RGBA image as JPEG raises "cannot write mode RGBA as JPEG".
                fd, tmp_name = tempfile.mkstemp(suffix=".png", prefix=f"input_{current_seed}_", dir=output_dir)
                os.close(fd)
                temp_image_path = Path(tmp_name)
                input_image.convert("RGB").save(temp_image_path)
                image_path = temp_image_path
            else:
                image_path = Path(input_image)
            images = [ImageConditioningInput(path=str(image_path), frame_idx=0, strength=1.0)]

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

        log_memory("before pipeline call")

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

        log_memory("after pipeline call")

        fd, output_path = tempfile.mkstemp(suffix=".mp4")
        os.close(fd)
        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 gr.Error:
        raise
    except Exception as e:
        import traceback
        log_memory("on error")
        print(f"Error: {str(e)}\n{traceback.format_exc()}")
        # Surface the failure in the UI instead of silently returning an empty video.
        raise gr.Error(f"Generation failed: {type(e).__name__}: {e}")
    finally:
        if temp_image_path is not None:
            temp_image_path.unlink(missing_ok=True)


with gr.Blocks(title="LTX-2.3 Distilled") as demo:
    gr.Markdown("# LTX-2.3 Distilled (22B): Fast Audio-Video Generation")
    gr.Markdown(
        "Fast and high quality video + audio generation "
        "[[model]](https://huggingface.co/Lightricks/LTX-2.3) "
        "[[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",
                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...",
            )

            with gr.Row():
                duration = gr.Slider(label="Duration (seconds)", minimum=1.0, maximum=10.0, value=3.0, step=0.1)
                with gr.Column():
                    enhance_prompt = gr.Checkbox(label="Enhance Prompt", value=False)
                    high_res = gr.Checkbox(label="High Resolution", value=True)

            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 (rounded to a multiple of 64)", value=1536, precision=0)
                    height = gr.Number(label="Height (rounded to a multiple of 64)", value=1024, precision=0)

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

    # Auto-detect aspect ratio from uploaded image and set resolution
    input_image.change(
        fn=update_resolution,
        inputs=[input_image, high_res],
        outputs=[width, height],
    )

    # Update resolution when high-res toggle changes
    high_res.change(
        fn=update_resolution,
        inputs=[input_image, high_res],
        outputs=[width, height],
    )

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


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

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

	import importlib.util
	import os
	import shutil
	import subprocess
	import sys

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

	# --------------------------------------------------------------------------- #
	# Bootstrap: xformers + LTX-2 packages
	# --------------------------------------------------------------------------- #
	# xformers is pinned in requirements.txt (installed at build time). This is only
	# a safety net so we do NOT re-download ~117 MB on every restart.
	if importlib.util.find_spec("xformers") is None:
	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_SHA = "ae855f8538843825f9015a419cf4ba5edaf5eec2"


	def ensure_ltx_repo() -> None:
	"""Clone LTX-2 at the pinned commit. Re-clones if a previous clone was partial."""
	marker = os.path.join(LTX_REPO_DIR, "packages", "ltx-core", "pyproject.toml")
	if os.path.exists(marker):
	return
	if os.path.exists(LTX_REPO_DIR):
	print(f"Removing incomplete LTX-2 checkout at {LTX_REPO_DIR}")
	shutil.rmtree(LTX_REPO_DIR, ignore_errors=True)

	print(f"Cloning {LTX_REPO_URL}...")
	os.makedirs(LTX_REPO_DIR)
	subprocess.run(["git", "init", LTX_REPO_DIR], check=True)
	subprocess.run(["git", "remote", "add", "origin", LTX_REPO_URL], cwd=LTX_REPO_DIR, check=True)
	subprocess.run(["git", "fetch", "--depth", "1", "origin", LTX_COMMIT_SHA], cwd=LTX_REPO_DIR, check=True)
	subprocess.run(["git", "checkout", LTX_COMMIT_SHA], cwd=LTX_REPO_DIR, check=True)


	ensure_ltx_repo()

	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 torch
	torch._dynamo.config.suppress_errors = True
	torch._dynamo.config.disable = True

	import spaces
	import gradio as gr
	import gradio.queueing


	# Gradio 6.x bug workaround: Queue.send_message raises KeyError when an event's
	# session is no longer in pending_messages_per_session (an LRUCache capped at 2000
	# sessions, also cleared on client disconnect). If that happens inside
	# Queue.start_processing (via broadcast_estimations), its `finally` sets
	# queue.stopped = True permanently and every request gets "Queue is stopped."
	# until the Space restarts. Drop the orphaned event instead of killing the queue.
	_original_send_message = gradio.queueing.Queue.send_message


	def _send_message_tolerating_lost_sessions(self, event, event_message):
	try:
	_original_send_message(self, event, event_message)
	except KeyError:
	event.alive = False


	gradio.queueing.Queue.send_message = _send_message_tolerating_lost_sessions

	import numpy as np

	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

	from helpers import (
	DEFAULT_FRAME_RATE,
	compute_num_frames,
	pick_resolution,
	snap_dimension,
	)
	from model_resolver import ModelSetupError, resolve_model_paths

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

	# Disable xformers FA3 dispatch: FA3 kernels are Hopper-only (sm_90a), but
	# xformers' dispatcher gates them on `device_capability >= (9, 0)`, which also
	# matches Blackwell (RTX PRO 6000, the ZeroGPU fleet hardware since 2026-05-12)
	# and crashes at kernel launch with "invalid argument".
	try:
	from xformers.ops.fmha import _set_use_fa3
	_set_use_fa3(False)
	print("[ATTN] xformers FA3 dispatch disabled (Blackwell-incompatible)")
	except Exception as e:
	print(f"[ATTN] FA3 disable FAILED: {type(e).__name__}: {e}")

	# FUSE/mmap workaround: SafetensorsStateDictLoader.load uses safetensors.safe_open
	# under the hood, which mmap's the file. On bucket FUSE mounts that triggers a
	# page-fault storm and deadlocks loading. Bypass mmap by parsing the safetensors
	# header ourselves and reading each tensor's bytes directly.
	import json
	import struct

	from ltx_core.loader.primitives import StateDict
	from ltx_core.loader.sft_loader import SafetensorsStateDictLoader

	_SAFETENSORS_DTYPE_MAP = {
	"F64": torch.float64,
	"F32": torch.float32,
	"F16": torch.float16,
	"BF16": torch.bfloat16,
	"F8_E5M2": torch.float8_e5m2,
	"F8_E4M3": torch.float8_e4m3fn,
	"I64": torch.int64,
	"I32": torch.int32,
	"I16": torch.int16,
	"I8": torch.int8,
	"U8": torch.uint8,
	"BOOL": torch.bool,
	}


	def _patched_load(self, path, sd_ops, device=None):
	sd = {}
	size = 0
	dtype = set()
	device = device or torch.device("cpu")
	model_paths = path if isinstance(path, list) else [path]
	for shard_path in model_paths:
	with open(shard_path, "rb") as f:
	header_len = struct.unpack("<Q", f.read(8))[0]
	header = json.loads(f.read(header_len).decode("utf-8"))
	data_base = 8 + header_len
	for name, meta in header.items():
	if name == "__metadata__":
	continue
	expected_name = name if sd_ops is None else sd_ops.apply_to_key(name)
	if expected_name is None:
	continue
	start, end = meta["data_offsets"]
	f.seek(data_base + start)
	buf = f.read(end - start)
	t = torch.frombuffer(
	bytearray(buf), dtype=_SAFETENSORS_DTYPE_MAP[meta["dtype"]]
	).reshape(meta["shape"])
	t = t.to(device=device, non_blocking=True, copy=False)
	kvs = (
	((expected_name, t),)
	if sd_ops is None
	else sd_ops.apply_to_key_value(expected_name, t)
	)
	for key, v in kvs:
	size += v.nbytes
	dtype.add(v.dtype)
	sd[key] = v
	return StateDict(sd=sd, device=device, size=size, dtype=dtype)


	SafetensorsStateDictLoader.load = _patched_load
	print("[FUSE-PATCH] SafetensorsStateDictLoader.load replaced (chunked-read)")

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

	# --------------------------------------------------------------------------- #
	# Locate the models (mounted volumes -> HF cache -> Hub download).
	# Raises a descriptive ModelSetupError instead of the cryptic
	# "No files matching pattern 'tokenizer.model'" from inside ltx-core.
	# --------------------------------------------------------------------------- #
	try:
	model_paths = resolve_model_paths()
	except ModelSetupError as exc:
	print("=" * 80)
	print("MODEL SETUP ERROR")
	print(exc)
	print("=" * 80)
	raise

	# Initialize pipeline WITH text encoder
	pipeline = DistilledPipeline(
	distilled_checkpoint_path=model_paths.distilled_checkpoint,
	spatial_upsampler_path=model_paths.spatial_upsampler,
	gemma_root=model_paths.gemma_root,
	loras=[],
	quantization=QuantizationPolicy.fp8_cast(),
	)

	# Preload all models for ZeroGPU tensor packing.
	print("Preloading all models (including Gemma)...")
	ledger = pipeline.model_ledger
	_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()
	_text_encoder = ledger.text_encoder()
	_embeddings_processor = ledger.gemma_embeddings_processor()

	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
	ledger.text_encoder = lambda: _text_encoder
	ledger.gemma_embeddings_processor = lambda: _embeddings_processor
	print("All models preloaded (including Gemma text encoder)!")

	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 / 10243:.2f}GB total={total / 10243:.2f}GB")


	def update_resolution(image, high_res):
	"""Auto-set width/height from the uploaded image's aspect ratio + quality tier."""
	w, h = pick_resolution(image, high_res)
	return gr.update(value=w), gr.update(value=h)


	@spaces.GPU(duration=75)
	@torch.inference_mode()
	def generate_video(
	input_image,
	prompt: str,
	duration: float,
	enhance_prompt: bool = True,
	seed: int = 42,
	randomize_seed: bool = True,
	height: int = 1024,
	width: int = 1536,
	progress=gr.Progress(track_tqdm=True),
	):
	# Defined first so the error path can never hit an unbound name.
	current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)
	temp_image_path = None

	try:
	if not prompt or not prompt.strip():
	raise gr.Error("Please enter a prompt.")

	torch.cuda.reset_peak_memory_stats()
	log_memory("start")

	frame_rate = DEFAULT_FRAME_RATE
	num_frames = compute_num_frames(duration, frame_rate)
	# Free-form boxes: snap to multiples of 64 or the two-stage pipeline crashes.
	height = snap_dimension(height)
	width = snap_dimension(width)

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

	images = []
	if input_image is not None:
	output_dir = Path("outputs")
	output_dir.mkdir(exist_ok=True)
	if hasattr(input_image, "save"):
	# PNG + RGB: saving an RGBA image as JPEG raises "cannot write mode RGBA as JPEG".
	fd, tmp_name = tempfile.mkstemp(suffix=".png", prefix=f"input_{current_seed}_", dir=output_dir)
	os.close(fd)
	temp_image_path = Path(tmp_name)
	input_image.convert("RGB").save(temp_image_path)
	image_path = temp_image_path
	else:
	image_path = Path(input_image)
	images = [ImageConditioningInput(path=str(image_path), frame_idx=0, strength=1.0)]

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

	log_memory("before pipeline call")

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

	log_memory("after pipeline call")

	fd, output_path = tempfile.mkstemp(suffix=".mp4")
	os.close(fd)
	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 gr.Error:
	raise
	except Exception as e:
	import traceback
	log_memory("on error")
	print(f"Error: {str(e)}\n{traceback.format_exc()}")
	# Surface the failure in the UI instead of silently returning an empty video.
	raise gr.Error(f"Generation failed: {type(e).__name__}: {e}")
	finally:
	if temp_image_path is not None:
	temp_image_path.unlink(missing_ok=True)


	with gr.Blocks(title="LTX-2.3 Distilled") as demo:
	gr.Markdown("# LTX-2.3 Distilled (22B): Fast Audio-Video Generation")
	gr.Markdown(
	"Fast and high quality video + audio generation "
	"[[model]](https://huggingface.co/Lightricks/LTX-2.3) "
	"[[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",
	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...",
	)

	with gr.Row():
	duration = gr.Slider(label="Duration (seconds)", minimum=1.0, maximum=10.0, value=3.0, step=0.1)
	with gr.Column():
	enhance_prompt = gr.Checkbox(label="Enhance Prompt", value=False)
	high_res = gr.Checkbox(label="High Resolution", value=True)

	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 (rounded to a multiple of 64)", value=1536, precision=0)
	height = gr.Number(label="Height (rounded to a multiple of 64)", value=1024, precision=0)

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

	# Auto-detect aspect ratio from uploaded image and set resolution
	input_image.change(
	fn=update_resolution,
	inputs=[input_image, high_res],
	outputs=[width, height],
	)

	# Update resolution when high-res toggle changes
	high_res.change(
	fn=update_resolution,
	inputs=[input_image, high_res],
	outputs=[width, height],
	)

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


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

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