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)