Update app_exp.py

app_exp.py

@@ -1,37 +1,50 @@
+import gradio as gr
+import torch
 import os
 import sys
 import subprocess
 import tempfile
 import numpy as np
+import site
+import importlib
 from PIL import Image
+from huggingface_hub import snapshot_download, hf_hub_download
 
-import torch
-import gradio as gr
-from torchvision.io import write_video
+# ============================================================
+# 0️⃣ FlashAttention 3 Setup
+# ============================================================
+try:
+    print("Attempting to download and install FlashAttention wheel...")
+    flash_attention_wheel = hf_hub_download(
+        repo_id="rahul7star/flash-attn-3",
+        repo_type="model",
+        filename="128/flash_attn_3-3.0.0b1-cp39-abi3-linux_x86_64.whl",
+    )
+    subprocess.run(["pip", "install", flash_attention_wheel], check=True)
+    site.addsitedir(site.getsitepackages()[0])
+    importlib.invalidate_caches()
+    print("✅ FlashAttention installed successfully.")
+    enable_fa3 = True
+except Exception as e:
+    print(f"⚠️ Could not install FlashAttention: {e}")
+    print("Continuing without FlashAttention...")
+    enable_fa3 = False
 
-# Define paths
+# ============================================================
+# 1️⃣ Repository Setup
+# ============================================================
 REPO_PATH = "LongCat-Video"
 CHECKPOINT_DIR = os.path.join(REPO_PATH, "weights", "LongCat-Video")
 
-# Clone the repository if it doesn't exist
 if not os.path.exists(REPO_PATH):
     print(f"Cloning LongCat-Video repository to '{REPO_PATH}'...")
-    try:
-        subprocess.run(
-            ["git", "clone", "https://github.com/meituan-longcat/LongCat-Video.git", REPO_PATH],
-            check=True,
-            capture_output=True
-        )
-        print("Repository cloned successfully.")
-    except subprocess.CalledProcessError as e:
-        print(f"Error cloning repository: {e.stderr.decode()}")
-        sys.exit(1)
-
-# Add the cloned repository to the Python path to allow imports
+    subprocess.run(
+        ["git", "clone", "https://github.com/meituan-longcat/LongCat-Video.git", REPO_PATH],
+        check=True
+    )
+
 sys.path.insert(0, os.path.abspath(REPO_PATH))
 
-# Now that the repo is in the path, we can import its modules
-from huggingface_hub import snapshot_download
 from longcat_video.pipeline_longcat_video import LongCatVideoPipeline
 from longcat_video.modules.scheduling_flow_match_euler_discrete import FlowMatchEulerDiscreteScheduler
 from longcat_video.modules.autoencoder_kl_wan import AutoencoderKLWan
@@ -40,136 +53,146 @@ from longcat_video.context_parallel import context_parallel_util
 from transformers import AutoTokenizer, UMT5EncoderModel
 from diffusers.utils import export_to_video
 
-# Download model weights from Hugging Face Hub if they don't exist
+# Download weights if not present
 if not os.path.exists(CHECKPOINT_DIR):
     print(f"Downloading model weights to '{CHECKPOINT_DIR}'...")
-    try:
-        snapshot_download(
-            repo_id="meituan-longcat/LongCat-Video",
-            local_dir=CHECKPOINT_DIR,
-            local_dir_use_symlinks=False,  # Use False for better Windows compatibility
-            ignore_patterns=["*.md", "*.gitattributes", "assets/*"] # ignore non-essential files
-        )
-        print("Model weights downloaded successfully.")
-    except Exception as e:
-        print(f"Error downloading model weights: {e}")
-        sys.exit(1)
-
-# Global placeholder for the pipeline and device configuration
-pipe = None
+    snapshot_download(
+        repo_id="meituan-longcat/LongCat-Video",
+        local_dir=CHECKPOINT_DIR,
+        local_dir_use_symlinks=False,
+        ignore_patterns=["*.md", "*.gitattributes", "assets/*"]
+    )
+
+# ============================================================
+# 2️⃣ Device & Models
+# ============================================================
 device = "cuda" if torch.cuda.is_available() else "cpu"
 torch_dtype = torch.bfloat16 if device == "cuda" else torch.float32
 
-# --- Initialize models ---
-cp_split_hw = context_parallel_util.get_optimal_split(1)
-tokenizer = AutoTokenizer.from_pretrained(CHECKPOINT_DIR, subfolder="tokenizer", torch_dtype=torch_dtype)
-text_encoder = UMT5EncoderModel.from_pretrained(CHECKPOINT_DIR, subfolder="text_encoder", torch_dtype=torch_dtype)
-vae = AutoencoderKLWan.from_pretrained(CHECKPOINT_DIR, subfolder="vae", torch_dtype=torch_dtype)
-scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(CHECKPOINT_DIR, subfolder="scheduler", torch_dtype=torch_dtype)
-
-dit = LongCatVideoTransformer3DModel.from_pretrained(CHECKPOINT_DIR,
-                                                         enable_flashattn3=True,
-                                                         enable_flashattn2=False,
-                                                         enable_xformers=True,
-                                                         subfolder="dit",
-                                                         cp_split_hw=cp_split_hw,
-                                                         torch_dtype=torch_dtype)
-
-pipe = LongCatVideoPipeline(
-    tokenizer=tokenizer,
-    text_encoder=text_encoder,
-    vae=vae,
-    scheduler=scheduler,
-    dit=dit,
-)
-pipe.to(device)
-
-# --- Load LoRAs ---
-cfg_lora = os.path.join(CHECKPOINT_DIR, 'lora/cfg_step_lora.safetensors')
-refine_lora = os.path.join(CHECKPOINT_DIR, 'lora/refinement_lora.safetensors')
-pipe.dit.load_lora(cfg_lora, 'cfg_step_lora')
-pipe.dit.load_lora(refine_lora, 'refinement_lora')
-
-# --- Enable Cache-DiT for DiT transformer ---
-cache_dit.enable_cache(pipe.dit)
+print(f"Device: {device}, dtype: {torch_dtype}")
+
+pipe = None
+try:
+    cp_split_hw = context_parallel_util.get_optimal_split(1)
+
+    tokenizer = AutoTokenizer.from_pretrained(CHECKPOINT_DIR, subfolder="tokenizer", torch_dtype=torch_dtype)
+    text_encoder = UMT5EncoderModel.from_pretrained(CHECKPOINT_DIR, subfolder="text_encoder", torch_dtype=torch_dtype)
+
+    vae = AutoencoderKLWan.from_pretrained(CHECKPOINT_DIR, subfolder="vae", torch_dtype=torch_dtype)
+    scheduler = FlowMatchEulerDiscreteScheduler.from_pretrained(CHECKPOINT_DIR, subfolder="scheduler", torch_dtype=torch_dtype)
+
+    dit = LongCatVideoTransformer3DModel.from_pretrained(
+        CHECKPOINT_DIR,
+        enable_flashattn3=enable_fa3,
+        enable_flashattn2=False,
+        enable_xformers=True,
+        subfolder="dit",
+        cp_split_hw=cp_split_hw,
+        torch_dtype=torch_dtype
+    )
+
+    pipe = LongCatVideoPipeline(
+        tokenizer=tokenizer,
+        text_encoder=text_encoder,
+        vae=vae,
+        scheduler=scheduler,
+        dit=dit,
+    )
+    pipe.to(device)
 
+    # Load LoRA weights
+    pipe.dit.load_lora(os.path.join(CHECKPOINT_DIR, 'lora/cfg_step_lora.safetensors'), 'cfg_step_lora')
+    pipe.dit.load_lora(os.path.join(CHECKPOINT_DIR, 'lora/refinement_lora.safetensors'), 'refinement_lora')
+
+    print("✅ Models loaded successfully")
+
+except Exception as e:
+    print(f"❌ Failed to load models: {e}")
+    pipe = None
+
+# ============================================================
+# 3️⃣ Generation Helpers
+# ============================================================
 def torch_gc():
     if torch.cuda.is_available():
         torch.cuda.empty_cache()
         torch.cuda.ipc_collect()
 
-# --- Video generation function ---
 def generate_video(
     mode,
     prompt,
     neg_prompt,
     image,
-    num_frames,
+    height, width, resolution,
     seed,
     use_distill,
     use_refine,
+    duration_sec,
+    progress=gr.Progress(track_tqdm=True)
 ):
     if pipe is None:
-        raise gr.Error("Models not loaded.")
+        raise gr.Error("❌ Models failed to load")
 
+    # Adaptive FPS for faster testing
+    fps = 15 if use_distill else 30
+    num_frames = int(duration_sec * fps)
     generator = torch.Generator(device=device).manual_seed(int(seed))
     is_distill = use_distill or use_refine
 
-    if is_distill:
-        pipe.dit.enable_loras(['cfg_step_lora'])
-        num_inference_steps = 16
-        guidance_scale = 1.0
-        current_neg_prompt = ""
-    else:
-        num_inference_steps = 50
-        guidance_scale = 4.0
-        current_neg_prompt = neg_prompt
+    # Stage 1
+    progress(0.2, desc="Stage 1: Base Video Generation")
+    pipe.dit.enable_loras(['cfg_step_lora'] if is_distill else [])
+
+    num_inference_steps = 12 if is_distill else 24
+    guidance_scale = 2.0 if is_distill else 4.0
+    curr_neg_prompt = "" if is_distill else neg_prompt
 
     if mode == "t2v":
         output = pipe.generate_t2v(
             prompt=prompt,
-            negative_prompt=current_neg_prompt,
-            height=480,
-            width=832,
+            negative_prompt=curr_neg_prompt,
+            height=height,
+            width=width,
             num_frames=num_frames,
             num_inference_steps=num_inference_steps,
+            use_distill=is_distill,
             guidance_scale=guidance_scale,
-            generator=generator,
+            generator=generator
         )[0]
-    else:  # i2v
-        pil_image = Image.fromarray(image)
+    else:
+        pil_img = Image.fromarray(image)
         output = pipe.generate_i2v(
-            image=pil_image,
+            image=pil_img,
             prompt=prompt,
-            negative_prompt=current_neg_prompt,
-            resolution="480p",
+            negative_prompt=curr_neg_prompt,
+            resolution=resolution,
             num_frames=num_frames,
             num_inference_steps=num_inference_steps,
+            use_distill=is_distill,
             guidance_scale=guidance_scale,
-            generator=generator,
+            generator=generator
         )[0]
 
-    if is_distill:
-        pipe.dit.disable_all_loras()
-
+    pipe.dit.disable_all_loras()
     torch_gc()
 
-    # Optional refinement
+    # Stage 2: Optional refinement
     if use_refine:
+        progress(0.5, desc="Stage 2: Refinement")
         pipe.dit.enable_loras(['refinement_lora'])
         pipe.dit.enable_bsa()
 
         stage1_video_pil = [(frame * 255).astype(np.uint8) for frame in output]
         stage1_video_pil = [Image.fromarray(img) for img in stage1_video_pil]
-
         refine_image = Image.fromarray(image) if mode == 'i2v' else None
+
         output = pipe.generate_refine(
             image=refine_image,
             prompt=prompt,
             stage1_video=stage1_video_pil,
-            num_cond_frames=1 if mode == 'i2v' else 0,
+            num_cond_frames=1 if mode=='i2v' else 0,
             num_inference_steps=50,
-            generator=generator,
+            generator=generator
         )[0]
 
         pipe.dit.disable_all_loras()
@@ -177,35 +200,74 @@ def generate_video(
         torch_gc()
 
     # Export video
+    progress(1.0, desc="Exporting video")
     with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as temp_video_file:
-        export_to_video(output, temp_video_file.name, fps=15)
+        export_to_video(output, temp_video_file.name, fps=fps)
         return temp_video_file.name
 
-# --- Gradio UI ---
+# ============================================================
+# 4️⃣ Gradio UI
+# ============================================================
 css = ".fillable{max-width: 960px !important}"
+
 with gr.Blocks(css=css) as demo:
-    gr.Markdown("# 🎬 LongCat-Video Optimized")
-    with gr.Row():
-        with gr.Column(scale=2):
-            prompt_input = gr.Textbox(label="Prompt", lines=4)
-            neg_prompt_input = gr.Textbox(label="Negative Prompt", lines=2, value="ugly, blurry, low quality")
-            seed_input = gr.Number(label="Seed", value=42, precision=0)
-            frames_slider = gr.Slider(16, 128, value=48, step=1, label="Number of Frames")
-            distill_checkbox = gr.Checkbox(label="Use Distill Mode", value=True)
-            refine_checkbox = gr.Checkbox(label="Use Refine Mode", value=False)
-            t2v_button = gr.Button("Generate Video")
-        with gr.Column(scale=3):
-            video_output = gr.Video(label="Generated Video", interactive=False)
+    gr.Markdown("# 🎬 LongCat-Video")
+    gr.Markdown("13.6B parameter dense video-generation model by Meituan — [[Model](https://huggingface.co/meituan-longcat/LongCat-Video)]")
+
+    with gr.Tabs() as tabs:
+        with gr.TabItem("Text-to-Video"):
+            mode_t2v = gr.State("t2v")
+            with gr.Row():
+                with gr.Column(scale=2):
+                    prompt_t2v = gr.Textbox(label="Prompt", lines=4)
+                    neg_prompt_t2v = gr.Textbox(label="Negative Prompt", lines=2, value="blurry, low quality")
+                    height_t2v = gr.Slider(256, 1024, step=64, value=480, label="Height")
+                    width_t2v = gr.Slider(256, 1024, step=64, value=832, label="Width")
+                    seed_t2v = gr.Number(value=42, label="Seed")
+                    distill_t2v = gr.Checkbox(value=True, label="Use Distill Mode")
+                    refine_t2v = gr.Checkbox(value=False, label="Use Refine Mode")
+                    duration_t2v = gr.Slider(1, 20, step=1, value=2, label="Video Duration (seconds)")
+
+                    t2v_button = gr.Button("Generate Video")
+                with gr.Column(scale=3):
+                    video_output_t2v = gr.Video(label="Generated Video")
 
+        with gr.TabItem("Image-to-Video"):
+            mode_i2v = gr.State("i2v")
+            with gr.Row():
+                with gr.Column(scale=2):
+                    image_i2v = gr.Image(type="numpy", label="Input Image")
+                    prompt_i2v = gr.Textbox(label="Prompt", lines=4)
+                    neg_prompt_i2v = gr.Textbox(label="Negative Prompt", lines=2, value="blurry, low quality")
+                    resolution_i2v = gr.Dropdown(["480p","720p"], value="480p", label="Resolution")
+                    seed_i2v = gr.Number(value=42, label="Seed")
+                    distill_i2v = gr.Checkbox(value=True, label="Use Distill Mode")
+                    refine_i2v = gr.Checkbox(value=False, label="Use Refine Mode")
+                    duration_i2v = gr.Slider(1, 20, step=1, value=2, label="Video Duration (seconds)")
+
+                    i2v_button = gr.Button("Generate Video")
+                with gr.Column(scale=3):
+                    video_output_i2v = gr.Video(label="Generated Video")
+
+    # Event binding
     t2v_button.click(
-        fn=generate_video,
-        inputs=[
-            gr.State("t2v"), prompt_input, neg_prompt_input,
-            gr.State(None), frames_slider, seed_input,
-            distill_checkbox, refine_checkbox
-        ],
-        outputs=video_output
+        generate_video,
+        inputs=[mode_t2v, prompt_t2v, neg_prompt_t2v, gr.State(None),
+                height_t2v, width_t2v, gr.State("480p"),
+                seed_t2v, distill_t2v, refine_t2v, duration_t2v],
+        outputs=video_output_t2v
+    )
+
+    i2v_button.click(
+        generate_video,
+        inputs=[mode_i2v, prompt_i2v, neg_prompt_i2v, image_i2v,
+                gr.State(None), gr.State(None), resolution_i2v,
+                seed_i2v, distill_i2v, refine_i2v, duration_i2v],
+        outputs=video_output_i2v
     )
 
+# ============================================================
+# 5️⃣ Launch
+# ============================================================
 if __name__ == "__main__":
-    demo.launch()
+    demo.launch()