app.py

import os
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor

os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = "1"
os.environ["HF_XET_HIGH_PERFORMANCE"] = "1"
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 os.path.exists(LTX_REPO_DIR):
    subprocess.run(["rm", "-rf", LTX_REPO_DIR], check=True)

subprocess.run(["git", "clone", LTX_REPO_URL, LTX_REPO_DIR], check=True)
subprocess.run(["git", "-C", LTX_REPO_DIR, "checkout", LTX_COMMIT], check=True)
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"))

# Patch sft_loader: remove non_blocking=True so ZeroGPU CUDA emulation can handle
# module-level tensor loading (emulation mode doesn't support async CUDA streams)
_sft_path = os.path.join(LTX_REPO_DIR, "packages", "ltx-core", "src", "ltx_core", "loader", "sft_loader.py")
with open(_sft_path, "r") as _f:
    _sft = _f.read()
_sft = _sft.replace(
    'with safetensors.safe_open(shard_path, framework="pt", device=str(device)) as f:',
    'with safetensors.safe_open(shard_path, framework="pt", device="cpu") as f:',
)
with open(_sft_path, "w") as _f:
    _f.write(_sft)

import logging
import random
import tempfile
from pathlib import Path

import torch
torch._dynamo.config.suppress_errors = True
torch._dynamo.config.disable = True
torch.inference_mode = torch.no_grad

import spaces
import gradio as gr
import numpy as np
from huggingface_hub import hf_hub_download, snapshot_download

# --- Core LTX imports ---
from ltx_core.components.diffusion_steps import EulerDiffusionStep
from ltx_core.components.noisers import GaussianNoiser
from ltx_core.model.audio_vae import encode_audio as vae_encode_audio
from ltx_core.model.upsampler import upsample_video
from ltx_core.model.video_vae import TilingConfig, get_video_chunks_number
from ltx_core.model.video_vae import decode_video as vae_decode_video
from ltx_core.types import Audio, AudioLatentShape, VideoPixelShape
from ltx_pipelines.utils import ModelLedger, euler_denoising_loop
from ltx_pipelines.utils.args import ImageConditioningInput
from ltx_pipelines.utils.constants import DISTILLED_SIGMA_VALUES, STAGE_2_DISTILLED_SIGMA_VALUES
from ltx_pipelines.utils.helpers import (
    assert_resolution,
    cleanup_memory,
    combined_image_conditionings,
    denoise_video_only,
    encode_prompts,
    get_device,
    simple_denoising_func,
)
from ltx_pipelines.utils.media_io import decode_audio_from_file, encode_video
from ltx_pipelines.utils.types import PipelineComponents
from ltx_core.loader.primitives import LoraPathStrengthAndSDOps
from ltx_core.loader.sd_ops import LTXV_LORA_COMFY_RENAMING_MAP

# --- Attention backend patch (same as Element-16) ---
import torch.nn.functional as F
from ltx_core.model.transformer import attention as _attn_mod

def _sdpa_as_mea(query, key, value, attn_bias=None, scale=None, **kwargs):
    q, k, v = query.transpose(1, 2), key.transpose(1, 2), value.transpose(1, 2)
    return F.scaled_dot_product_attention(q, k, v, scale=scale).transpose(1, 2)

_cap = torch.cuda.get_device_capability() if torch.cuda.is_available() else (0, 0)
_use_xformers = False
if _cap < (12, 0):
    try:
        from xformers.ops import memory_efficient_attention as _mea
        _attn_mod.memory_efficient_attention = _mea
        _use_xformers = True
        print(f"[ATTN] Using xformers memory_efficient_attention")
    except Exception as e:
        print(f"[ATTN] xformers unavailable ({e}), falling back to SDPA")
if not _use_xformers:
    _attn_mod.memory_efficient_attention = _sdpa_as_mea
    print(f"[ATTN] Using SDPA fallback (sm_{_cap[0]}{_cap[1]})")

logging.getLogger().setLevel(logging.INFO)
device = get_device()

# ---------------------------------------------------------------------------
# Custom pipeline: DistilledPipeline quality + audio guidance conditioning
# Uses sulphur_distil_bf16 (same checkpoint as Element-16) with frozen audio
# input latent to guide video generation — no mismatched LoRA.
# ---------------------------------------------------------------------------
class DistilledAudioGuidancePipeline:
    """
    Two-stage distilled pipeline with audio guidance.
    Identical to DistilledPipeline but replaces joint audio/video denoising
    with video-only denoising conditioned on a frozen input audio latent.
    This gives Element-16 quality + audio reactivity.
    """

    def __init__(self, distilled_checkpoint_path, spatial_upsampler_path, gemma_root,
                 loras=(), device=device, quantization=None):
        self.device = device
        self.dtype = torch.bfloat16
        self.model_ledger = ModelLedger(
            dtype=self.dtype,
            device=device,
            checkpoint_path=distilled_checkpoint_path,
            spatial_upsampler_path=spatial_upsampler_path,
            gemma_root_path=gemma_root,
            loras=loras,
            quantization=quantization,
        )
        self.pipeline_components = PipelineComponents(dtype=self.dtype, device=device)

    def __call__(
        self,
        prompt: str,
        seed: int,
        height: int,
        width: int,
        num_frames: int,
        frame_rate: float,
        images: list,
        audio_path: str,
        audio_start_time: float = 0.0,
        audio_max_duration: float | None = None,
        tiling_config: TilingConfig | None = None,
        enhance_prompt: bool = False,
    ):
        assert_resolution(height=height, width=width, is_two_stage=True)

        generator = torch.Generator(device=self.device).manual_seed(seed)
        noiser = GaussianNoiser(generator=generator)
        stepper = EulerDiffusionStep()
        dtype = torch.bfloat16
        duration = audio_max_duration or num_frames / frame_rate

        # 1. Encode prompts
        (ctx_p,) = encode_prompts(
            [prompt], self.model_ledger,
            enhance_first_prompt=enhance_prompt,
            enhance_prompt_image=images[0][0] if len(images) > 0 else None,
        )
        video_context, audio_context = ctx_p.video_encoding, ctx_p.audio_encoding

        # 2. Encode input audio — freeze as conditioning latent
        audio_encoder = self.model_ledger.audio_encoder()
        decoded_audio = decode_audio_from_file(audio_path, self.device, audio_start_time, duration)
        encoded_audio_latent = vae_encode_audio(decoded_audio, audio_encoder)
        del audio_encoder
        cleanup_memory()

        audio_shape = AudioLatentShape.from_duration(
            batch=1, duration=num_frames / frame_rate, channels=8, mel_bins=16
        )
        target_frames = audio_shape.frames
        current_frames = encoded_audio_latent.shape[2]
        if current_frames < target_frames:
            # Audio shorter than video — pad with zeros
            pad = target_frames - current_frames
            encoded_audio_latent = torch.nn.functional.pad(encoded_audio_latent, (0, 0, 0, pad))
        else:
            encoded_audio_latent = encoded_audio_latent[:, :, :target_frames]

        # 3. Load transformer + video_encoder ONCE — reused across both stages
        video_encoder = self.model_ledger.video_encoder()
        transformer   = self.model_ledger.transformer()
        stage_1_sigmas = torch.Tensor(DISTILLED_SIGMA_VALUES).to(self.device)

        def make_denoising_loop(t):
            def loop(sigmas, video_state, audio_state, stepper):
                return euler_denoising_loop(
                    sigmas=sigmas, video_state=video_state, audio_state=audio_state,
                    stepper=stepper,
                    denoise_fn=simple_denoising_func(
                        video_context=video_context,
                        audio_context=audio_context,
                        transformer=t,
                    ),
                )
            return loop

        stage_1_shape = VideoPixelShape(
            batch=1, frames=num_frames, width=width // 2, height=height // 2, fps=frame_rate
        )
        stage_1_conditionings = combined_image_conditionings(
            images=images, height=stage_1_shape.height, width=stage_1_shape.width,
            video_encoder=video_encoder, dtype=dtype, device=self.device,
        )

        video_state = denoise_video_only(
            output_shape=stage_1_shape,
            conditionings=stage_1_conditionings,
            noiser=noiser,
            sigmas=stage_1_sigmas,
            stepper=stepper,
            denoising_loop_fn=make_denoising_loop(transformer),
            components=self.pipeline_components,
            dtype=dtype,
            device=self.device,
            initial_audio_latent=encoded_audio_latent,
        )

        # 4. Upsample — video_encoder stays loaded
        upscaled_video_latent = upsample_video(
            latent=video_state.latent[:1],
            video_encoder=video_encoder,
            upsampler=self.model_ledger.spatial_upsampler(),
        )
        torch.cuda.synchronize()

        # 5. Stage 2 — reuse same transformer + video_encoder, no reload
        stage_2_sigmas = torch.Tensor(STAGE_2_DISTILLED_SIGMA_VALUES).to(self.device)
        stage_2_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_shape.height, width=stage_2_shape.width,
            video_encoder=video_encoder, dtype=dtype, device=self.device,
        )

        video_state = denoise_video_only(
            output_shape=stage_2_shape,
            conditionings=stage_2_conditionings,
            noiser=noiser,
            sigmas=stage_2_sigmas,
            stepper=stepper,
            denoising_loop_fn=make_denoising_loop(transformer),
            components=self.pipeline_components,
            dtype=dtype,
            device=self.device,
            noise_scale=stage_2_sigmas[0],
            initial_video_latent=upscaled_video_latent,
            initial_audio_latent=encoded_audio_latent,
        )

        torch.cuda.synchronize()
        del transformer, video_encoder
        cleanup_memory()

        # 6. Decode video, return original input audio (preserve fidelity)
        decoded_video = vae_decode_video(
            video_state.latent, self.model_ledger.video_decoder(), tiling_config, generator
        )
        original_audio = Audio(
            waveform=decoded_audio.waveform.squeeze(0),
            sampling_rate=decoded_audio.sampling_rate,
        )

        return decoded_video, original_audio


# ---------------------------------------------------------------------------
# Model setup
# ---------------------------------------------------------------------------
MAX_SEED = np.iinfo(np.int32).max
DEFAULT_PROMPT = (
    "A person speaking naturally, lips moving in perfect sync with their voice, "
    "cinematic lighting, sharp focus, smooth motion."
)
DEFAULT_FRAME_RATE = 24.0

CHECKPOINT_REPO = "SulphurAI/Sulphur-2-base"
LTX_MODEL_REPO  = "Lightricks/LTX-2.3"
GEMMA_REPO      = "Lightricks/gemma-3-12b-it-qat-q4_0-unquantized"

RESOLUTIONS = {
    "9:16": (512, 896),
    "16:9": (896, 512),
    "1:1":  (640, 640),
}

print("=" * 80)
print("Downloading Sulphur distilled checkpoint + upsampler + Gemma (parallel)...")
print("=" * 80)

def download_checkpoint():
    return hf_hub_download(repo_id=CHECKPOINT_REPO, filename="sulphur_distil_bf16.safetensors")

def download_upsampler():
    return hf_hub_download(repo_id=LTX_MODEL_REPO, filename="ltx-2.3-spatial-upscaler-x2-1.0.safetensors")

def download_gemma():
    return snapshot_download(repo_id=GEMMA_REPO)

def download_talking_head_lora():
    return hf_hub_download(
        repo_id="elix3r/LTX-2.3-22b-AV-LoRA-talking-head",
        filename="LTX-2.3-22b-AV-LoRA-talking-head-v1.safetensors"
    )

with ThreadPoolExecutor(max_workers=4) as executor:
    f_ckpt      = executor.submit(download_checkpoint)
    f_upsampler = executor.submit(download_upsampler)
    f_gemma     = executor.submit(download_gemma)
    f_lora      = executor.submit(download_talking_head_lora)

    checkpoint_path   = f_ckpt.result()
    upsampler_path    = f_upsampler.result()
    gemma_root        = f_gemma.result()
    talking_head_lora = f_lora.result()

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

pipeline = DistilledAudioGuidancePipeline(
    distilled_checkpoint_path=checkpoint_path,
    spatial_upsampler_path=upsampler_path,
    gemma_root=gemma_root,
    loras=(LoraPathStrengthAndSDOps(
        path=talking_head_lora,
        strength=0.8,
        sd_ops=LTXV_LORA_COMFY_RENAMING_MAP
    ),),
)

# Preload all models for ZeroGPU tensor packing
# Module-level CUDA works via ZeroGPU's CUDA emulation mode (same pattern as Element-16)
print("Preloading all pipeline components via model_ledger...")
ledger = pipeline.model_ledger
_transformer          = ledger.transformer()
_video_encoder        = ledger.video_encoder()
_audio_encoder        = ledger.audio_encoder()
_video_decoder        = ledger.video_decoder()
_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.audio_encoder            = lambda: _audio_encoder
ledger.video_decoder            = lambda: _video_decoder
ledger.spatial_upsampler        = lambda: _spatial_upsampler
ledger.text_encoder             = lambda: _text_encoder
ledger.gemma_embeddings_processor = lambda: _embeddings_processor
print("All models preloaded!")

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


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def log_memory(tag: str):
    if torch.cuda.is_available():
        alloc = 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}] alloc={alloc:.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 "9:16"
    w, h = (image.size if hasattr(image, "size") else (image.shape[1], image.shape[0]))
    ratio = w / h
    candidates = {"9:16": 9 / 16, "16:9": 16 / 9, "1:1": 1.0}
    return min(candidates, key=lambda k: abs(ratio - candidates[k]))


def on_image_upload(image):
    aspect = detect_aspect_ratio(image)
    w, h = RESOLUTIONS[aspect]
    return gr.update(value=w), gr.update(value=h)


def ensure_stereo(audio_path: str) -> str:
    """Convert audio to stereo WAV — audio VAE expects 2 channels."""
    out_path = tempfile.mktemp(suffix="_stereo.wav")
    subprocess.run(
        ["ffmpeg", "-y", "-i", audio_path, "-ac", "2", "-ar", "44100", out_path],
        check=True, capture_output=True,
    )
    return out_path


DEFAULT_NEGATIVE_PROMPT = (
    "色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，"
    "JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，"
    "形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走, "
    "blurry, glasses, deformed, subtitles, text, captions, worst quality, low quality, "
    "inconsistent motion, jittery, distorted"
)


# ---------------------------------------------------------------------------
# Inference
# ---------------------------------------------------------------------------
@spaces.GPU(duration=90)
@torch.inference_mode()
def generate_video(
    first_image,
    audio_input,
    prompt: str,
    duration: float,
    enhance_prompt: bool,
    seed: int,
    randomize_seed: bool,
    height: int,
    width: int,
    negative_prompt: str,
    progress=gr.Progress(track_tqdm=True),
):
    if audio_input is None:
        raise gr.Error("Please provide an audio file.")

    audio_path = ensure_stereo(audio_input)

    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: {width}x{height}, {num_frames} frames ({duration}s), seed={current_seed}")

        output_dir = Path("outputs")
        output_dir.mkdir(exist_ok=True)

        images = []
        if first_image is not None:
            temp_path = output_dir / f"temp_first_{current_seed}.jpg"
            if hasattr(first_image, "save"):
                first_image.save(temp_path)
            else:
                temp_path = Path(first_image)
            images.append(ImageConditioningInput(path=str(temp_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_frames_iter, audio = pipeline(
            prompt=prompt,
            seed=current_seed,
            height=int(height),
            width=int(width),
            num_frames=num_frames,
            frame_rate=frame_rate,
            images=images,
            audio_path=audio_path,
            audio_max_duration=num_frames / frame_rate,
            tiling_config=tiling_config,
            enhance_prompt=enhance_prompt,
        )

        frames = [frame for frame in video_frames_iter]
        video_tensor = torch.cat(frames, dim=0) if len(frames) > 1 else frames[0]

        log_memory("after pipeline call")

        output_path = tempfile.mktemp(suffix=".mp4")
        encode_video(
            video=video_tensor,
            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()}")
        raise gr.Error(str(e))


# ---------------------------------------------------------------------------
# Gradio UI
# ---------------------------------------------------------------------------
with gr.Blocks(title="Element-16 Audio Guidance", delete_cache=(3600, 7200)) as demo:
    gr.Markdown("# Element-16: Audio-Guided Video Generation")
    gr.Markdown(
        "Generate audio-reactive video from an audio file and an optional image. "
        "Powered by Sulphur distilled checkpoint with native audio conditioning. "
        "[[code]](https://github.com/Lightricks/LTX-2)"
    )

    with gr.Row():
        with gr.Column():
            first_image  = gr.Image(label="First Frame (Optional)", type="pil")
            audio_input  = gr.Audio(label="Audio Input", type="filepath")
            prompt       = gr.Textbox(
                label="Prompt",
                info="Describe the scene and motion. Be specific for best results.",
                value=DEFAULT_PROMPT,
                lines=3,
            )
            duration     = gr.Slider(label="Duration (seconds)", minimum=1.0, maximum=20.0, value=5.0, step=0.5)
            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=42, step=1)
                randomize_seed  = gr.Checkbox(label="Randomize Seed", value=True)
                with gr.Row():
                    width  = gr.Number(label="Width",  value=512, precision=0)
                    height = gr.Number(label="Height", value=896, precision=0)
                enhance_prompt  = gr.Checkbox(label="Enhance Prompt", value=False)
                negative_prompt = gr.Textbox(
                    label="Negative Prompt",
                    value=DEFAULT_NEGATIVE_PROMPT,
                    lines=3,
                )

        with gr.Column():
            output_video = gr.Video(label="Generated Video", autoplay=True)
            used_seed    = gr.Number(label="Used Seed", interactive=False)

    first_image.change(fn=on_image_upload, inputs=[first_image], outputs=[width, height])

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

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

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