wan2_fast.py

import spaces
import torch
from diffusers import AutoencoderKLWan, WanPipeline, UniPCMultistepScheduler
from diffusers.utils import export_to_video
import gradio as gr
import tempfile
import spaces
from huggingface_hub import hf_hub_download
import numpy as np
import random
import os



# LIGHT WEIGHT 1.3b
# MODEL_ID = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers"
# LORA_REPO_ID = "Kijai/WanVideo_comfy"
# LORA_FILENAME = "Wan21_CausVid_bidirect2_T2V_1_3B_lora_rank32.safetensors"


## This is working well ##

MODEL_ID = "FastVideo/FastWan2.2-TI2V-5B-FullAttn-Diffusers"

#MODEL_ID = "FastDM/Wan2.2-I2V-A14B-Merge-Lightning-V1.0-Diffusers"

## below model has a blury output but it get loaded 
##MODEL_ID ="Runware/Wan2.2-TI2V-5B"

## all these are exp  and not wokring due to memoerty issue 

#MODEL_ID ="Wan-AI/Wan2.2-T2V-A14B-Diffusers"


#MODEL_ID ="linoyts/Wan2.2-T2V-A14B-Diffusers-BF16"

#MODEL_ID = "Wan-AI/Wan2.2-TI2V-5B-Diffusers"



HF_MODEL = os.environ.get("HF_UPLOAD_REPO", "rahul7star/VideoExplain")
from huggingface_hub import HfApi, upload_file
import os
import uuid
import logging

import os
import uuid
import logging
from datetime import datetime
from huggingface_hub import HfApi, upload_file
import subprocess
import tempfile
import logging
import shutil
import os
from huggingface_hub import HfApi, upload_file
from datetime import datetime
import uuid

HF_MODEL = os.environ.get("HF_UPLOAD_REPO", "rahul7star/VideoExplain")

def upscale_and_upload_4k(input_video_path: str, summary_text: str) -> str:
    """
    Upscale a video to 4K and upload it to Hugging Face Hub without replacing the original file.

    Args:
        input_video_path (str): Path to the original video.
        summary_text (str): Text summary to upload alongside the video.

    Returns:
        str: Hugging Face folder path where the video and summary were uploaded.
    """
    logging.info(f"Upscaling video to 4K for upload: {input_video_path}")

    # Create a temporary file for the upscaled video
    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp_upscaled:
        upscaled_path = tmp_upscaled.name

    # FFmpeg upscale command
    cmd = [
        "ffmpeg",
        "-i", input_video_path,
        "-vf", "scale=3840:2160:flags=lanczos",
        "-c:v", "libx264",
        "-crf", "18",
        "-preset", "slow",
        "-y",
        upscaled_path,
    ]
    try:
        subprocess.run(cmd, check=True, capture_output=True)
        logging.info(f"✅ Upscaled video created at: {upscaled_path}")
    except subprocess.CalledProcessError as e:
        logging.error(f"FFmpeg failed:\n{e.stderr.decode()}")
        raise

    # Create a date-based folder on HF
    today_str = datetime.now().strftime("%Y-%m-%d")
    unique_subfolder = f"Upload-4K-{uuid.uuid4().hex[:8]}"
    hf_folder = f"{today_str}/{unique_subfolder}"

    # Upload video
    video_filename = os.path.basename(input_video_path)
    video_hf_path = f"{hf_folder}/{video_filename}"
    upload_file(
        path_or_fileobj=upscaled_path,
        path_in_repo=video_hf_path,
        repo_id=HF_MODEL,
        repo_type="model",
        token=os.environ.get("HUGGINGFACE_HUB_TOKEN"),
    )
    logging.info(f"✅ Uploaded 4K video to HF: {video_hf_path}")

    # Upload summary.txt
    summary_file = tempfile.NamedTemporaryFile(delete=False, suffix=".txt").name
    with open(summary_file, "w", encoding="utf-8") as f:
        f.write(summary_text)

    summary_hf_path = f"{hf_folder}/summary.txt"
    upload_file(
        path_or_fileobj=summary_file,
        path_in_repo=summary_hf_path,
        repo_id=HF_MODEL,
        repo_type="model",
        token=os.environ.get("HUGGINGFACE_HUB_TOKEN"),
    )
    logging.info(f"✅ Uploaded summary to HF: {summary_hf_path}")

    # Cleanup temporary files
    os.remove(upscaled_path)
    os.remove(summary_file)

    return hf_folder

def upload_to_hf(video_path, summary_text):
    api = HfApi()
    
    # Create a date-based folder (YYYY-MM-DD)
    today_str = datetime.now().strftime("%Y-%m-%d")
    date_folder = today_str
    
    # Generate a unique subfolder for this upload
    unique_subfolder = f"Wan22-Insta-upload_{uuid.uuid4().hex[:8]}"
    hf_folder = f"{date_folder}/{unique_subfolder}"
    logging.info(f"Uploading files to HF folder: {hf_folder} in repo {HF_MODEL}")

    # Upload video
    video_filename = os.path.basename(video_path)
    video_hf_path = f"{hf_folder}/{video_filename}"
    upload_file(
        path_or_fileobj=video_path,
        path_in_repo=video_hf_path,
        repo_id=HF_MODEL,
        repo_type="model",
        token=os.environ.get("HUGGINGFACE_HUB_TOKEN"),
    )
    logging.info(f"✅ Uploaded video to HF: {video_hf_path}")

    # Upload summary.txt
    summary_file = "/tmp/summary.txt"
    with open(summary_file, "w", encoding="utf-8") as f:
        f.write(summary_text)

    summary_hf_path = f"{hf_folder}/summary.txt"
    upload_file(
        path_or_fileobj=summary_file,
        path_in_repo=summary_hf_path,
        repo_id=HF_MODEL,
        repo_type="model",
        token=os.environ.get("HUGGINGFACE_HUB_TOKEN"),
    )
    logging.info(f"✅ Uploaded summary to HF: {summary_hf_path}")

    return hf_folder







vae = AutoencoderKLWan.from_pretrained(MODEL_ID, subfolder="vae", torch_dtype=torch.float32)
pipe = WanPipeline.from_pretrained(
    MODEL_ID, vae=vae, torch_dtype=torch.bfloat16
)
pipe.scheduler = UniPCMultistepScheduler.from_config(pipe.scheduler.config, flow_shift=8.0)
pipe.to("cuda")

torch.cuda.empty_cache()

# hold lora for now



# pipe.load_lora_weights(
#    "vrgamedevgirl84/Wan14BT2VFusioniX", 
#    weight_name="FusionX_LoRa/Phantom_Wan_14B_FusionX_LoRA.safetensors", 
#     adapter_name="phantom"
# )





# LORA_REPO_ID = "rahul7star/wan2.2Lora"
# LORA_FILENAME = "wan2.2/Wan2.2-Lightning_I2V-A14B-4steps-lora_LOW_fp16.safetensors"
# causvid_path = hf_hub_download(repo_id=LORA_REPO_ID, filename=LORA_FILENAME)
# pipe.load_lora_weights(causvid_path, adapter_name="causvid_lora")
# pipe.set_adapters(["causvid_lora"], adapter_weights=[0.95])
# pipe.fuse_lora()






# LORA_REPO_ID = "yeqiu168182/NSFW-22-H-e8"
# LORA_FILENAME = "NSFW-22-H-e8.safetensors"
# causvid_path = hf_hub_download(repo_id=LORA_REPO_ID, filename=LORA_FILENAME)
# pipe.load_lora_weights(causvid_path, adapter_name="causvid_lora")
# pipe.set_adapters(["causvid_lora"], adapter_weights=[0.95])
# pipe.fuse_lora()


# LORA_REPO_ID = "Kijai/WanVideo_comfy"
# LORA_FILENAME = "Wan22-Lightning/Wan2.2-Lightning_T2V-A14B-4steps-lora_LOW_fp16.safetensors"
# causvid_path = hf_hub_download(repo_id=LORA_REPO_ID, filename=LORA_FILENAME)
# pipe.load_lora_weights(causvid_path, adapter_name="causvid_lora")
# pipe.set_adapters(["causvid_lora"], adapter_weights=[0.95])
# pipe.fuse_lora()
#####################################################




# MOD_VALUE = 32
# DEFAULT_H_SLIDER_VALUE =  512
# DEFAULT_W_SLIDER_VALUE =  896

# # Environment variable check
# IS_ORIGINAL_SPACE = os.environ.get("IS_ORIGINAL_SPACE", "True") == "True"

# # Original limits
# ORIGINAL_SLIDER_MIN_H, ORIGINAL_SLIDER_MAX_H = 128, 1280
# ORIGINAL_SLIDER_MIN_W, ORIGINAL_SLIDER_MAX_W = 128, 1280
# ORIGINAL_MAX_DURATION = round(81/24, 1)  # MAX_FRAMES_MODEL/FIXED_FPS

# # Limited space constants
# LIMITED_MAX_RESOLUTION = 640
# LIMITED_MAX_DURATION = 2.0
# LIMITED_MAX_STEPS = 4

# # Set limits based on environment variable
# if IS_ORIGINAL_SPACE:
#     SLIDER_MIN_H, SLIDER_MAX_H = 128, LIMITED_MAX_RESOLUTION
#     SLIDER_MIN_W, SLIDER_MAX_W = 128, LIMITED_MAX_RESOLUTION
#     MAX_DURATION = LIMITED_MAX_DURATION
#     MAX_STEPS = LIMITED_MAX_STEPS
# else:
#     SLIDER_MIN_H, SLIDER_MAX_H = ORIGINAL_SLIDER_MIN_H, ORIGINAL_SLIDER_MAX_H
#     SLIDER_MIN_W, SLIDER_MAX_W = ORIGINAL_SLIDER_MIN_W, ORIGINAL_SLIDER_MAX_W
#     MAX_DURATION = ORIGINAL_MAX_DURATION
#     MAX_STEPS = 8

# MAX_SEED = np.iinfo(np.int32).max

# FIXED_FPS = 24
# FIXED_OUTPUT_FPS = 18 # we downspeed the output video as a temporary "trick"
# MIN_FRAMES_MODEL = 8
# MAX_FRAMES_MODEL = 81 


#New math to make it High Res

# MOD_VALUE = 32

# # Defaults for higher-res generation
# DEFAULT_H_SLIDER_VALUE = 768
# DEFAULT_W_SLIDER_VALUE = 1344  # 16:9 friendly and divisible by MOD_VALUE

# # Original Space = Hugging Face space with compute limits
# IS_ORIGINAL_SPACE = os.environ.get("IS_ORIGINAL_SPACE", "True") == "True"

# # Conservative limits for low-end environments
# LIMITED_MAX_RESOLUTION = 640
# LIMITED_MAX_DURATION = 2.0
# LIMITED_MAX_STEPS = 4

# # Generous limits for local or Pro spaces
# ORIGINAL_SLIDER_MIN_H, ORIGINAL_SLIDER_MAX_H = 128, 1536
# ORIGINAL_SLIDER_MIN_W, ORIGINAL_SLIDER_MAX_W = 128, 1536
# ORIGINAL_MAX_DURATION = round(81 / 24, 1)  # 3.4 seconds
# ORIGINAL_MAX_STEPS = 8

# # Use limited or original (generous) settings
# if IS_ORIGINAL_SPACE:
#     SLIDER_MIN_H, SLIDER_MAX_H = 128, LIMITED_MAX_RESOLUTION
#     SLIDER_MIN_W, SLIDER_MAX_W = 128, LIMITED_MAX_RESOLUTION
#     MAX_DURATION = LIMITED_MAX_DURATION
#     MAX_STEPS = LIMITED_MAX_STEPS
# else:
#     SLIDER_MIN_H, SLIDER_MAX_H = ORIGINAL_SLIDER_MIN_H, ORIGINAL_SLIDER_MAX_H
#     SLIDER_MIN_W, SLIDER_MAX_W = ORIGINAL_SLIDER_MIN_W, ORIGINAL_SLIDER_MAX_W
#     MAX_DURATION = ORIGINAL_MAX_DURATION
#     MAX_STEPS = ORIGINAL_MAX_STEPS

# MAX_SEED = np.iinfo(np.int32).max

# FIXED_FPS = 24
# FIXED_OUTPUT_FPS = 18  # reduce final video FPS to save space
# MIN_FRAMES_MODEL = 8
# MAX_FRAMES_MODEL = 81




# LORA_REPO_ID = "rahul7star/wan2.2Lora"
# LORA_FILENAME = "c0wg1rl.3_wan22-5b-ti2v - e380.safetensors"
# causvid_path = hf_hub_download(repo_id=LORA_REPO_ID, filename=LORA_FILENAME)
# pipe.load_lora_weights(causvid_path, adapter_name="causvid_lora")
# pipe.set_adapters(["causvid_lora"], adapter_weights=[0.95])
# pipe.fuse_lora()




LORA_REPO_ID = "UnifiedHorusRA/Missionary_POV_Wan_2.2_5B_LoRA"
LORA_FILENAME = "missionary-pov-wan2.2_5b-v1.0-vfxai.safetensors"
causvid_path = hf_hub_download(repo_id=LORA_REPO_ID, filename=LORA_FILENAME)
pipe.load_lora_weights(causvid_path, adapter_name="causvid_lora")
pipe.set_adapters(["causvid_lora"], adapter_weights=[0.95])
pipe.fuse_lora()


# Constants
MOD_VALUE = 32
DEFAULT_H_SLIDER_VALUE = 896
DEFAULT_W_SLIDER_VALUE = 896
NEW_FORMULA_MAX_AREA = 720 * 1024
SLIDER_MIN_H, SLIDER_MAX_H = 256, 1024
SLIDER_MIN_W, SLIDER_MAX_W = 256, 1024
MAX_SEED = np.iinfo(np.int32).max
FIXED_FPS = 24
MIN_FRAMES_MODEL = 25
MAX_FRAMES_MODEL = 193

FIXED_OUTPUT_FPS = 24


IS_ORIGINAL_SPACE = os.environ.get("IS_ORIGINAL_SPACE", "True") == "True"


# Conservative limits for low-end environments
LIMITED_MAX_RESOLUTION = 640
LIMITED_MAX_DURATION = 2.0
LIMITED_MAX_STEPS = 50 #4

# Generous limits for local or Pro spaces
ORIGINAL_SLIDER_MIN_H, ORIGINAL_SLIDER_MAX_H = 128, 1536
ORIGINAL_SLIDER_MIN_W, ORIGINAL_SLIDER_MAX_W = 128, 1536
ORIGINAL_MAX_DURATION = round(81 / 24, 1)  # 3.4 seconds
ORIGINAL_MAX_STEPS = 30 #8

if IS_ORIGINAL_SPACE:
    SLIDER_MIN_H, SLIDER_MAX_H = 128, LIMITED_MAX_RESOLUTION
    SLIDER_MIN_W, SLIDER_MAX_W = 128, LIMITED_MAX_RESOLUTION
    MAX_DURATION = LIMITED_MAX_DURATION
    MAX_STEPS = LIMITED_MAX_STEPS
else:
    SLIDER_MIN_H, SLIDER_MAX_H = ORIGINAL_SLIDER_MIN_H, ORIGINAL_SLIDER_MAX_H
    SLIDER_MIN_W, SLIDER_MAX_W = ORIGINAL_SLIDER_MIN_W, ORIGINAL_SLIDER_MAX_W
    MAX_DURATION = ORIGINAL_MAX_DURATION
    MAX_STEPS = ORIGINAL_MAX_STEPS

default_prompt_t2v = "cinematic footage, group of pedestrians dancing in the streets of NYC, high quality breakdance, 4K, tiktok video, intricate details, instagram feel, dynamic camera, smooth dance motion, dimly lit, stylish, beautiful faces, smiling, music video"
default_negative_prompt = "Bright tones, overexposed, static, blurred details, subtitles, style, works, paintings, images, static, overall gray, worst quality, low quality, JPEG compression residue, ugly, incomplete, extra fingers, poorly drawn hands, poorly drawn faces, deformed, disfigured, misshapen limbs, fused fingers, still picture, messy background, three legs, many people in the background, walking backwards, watermark, text, signature"

def get_durationold(prompt, height, width, 
                   negative_prompt, duration_seconds,
                   guidance_scale, steps,
                   seed, randomize_seed, 
                   progress):
    if steps > 4 and duration_seconds > 2:
        return 90
    elif steps > 4 or duration_seconds > 2:
        return 75
    else:
        return 60


def get_duration(
       prompt, height, width, 
                   negative_prompt, duration_seconds,
                   guidance_scale, steps,
                   seed, randomize_seed, 
                   progress,
):
    duration = int(duration_seconds) * int(steps) * 2.25 + 5
    
    return duration




@spaces.GPU(duration=get_duration)
def generate_video(prompt, height, width, 
                   negative_prompt=default_negative_prompt, duration_seconds = 2,
                   guidance_scale = 0, steps = 50,
                   seed = 42, randomize_seed = False, 
                   progress=gr.Progress(track_tqdm=True)):
  
    if not prompt or prompt.strip() == "":
        raise gr.Error("Please enter a text prompt. Try to use long and precise descriptions.")

    # Apply limits based on environment variable
    if IS_ORIGINAL_SPACE:
        height = min(height, LIMITED_MAX_RESOLUTION)
        width = min(width, LIMITED_MAX_RESOLUTION)
        duration_seconds = min(duration_seconds, LIMITED_MAX_DURATION)
        steps = min(steps, LIMITED_MAX_STEPS)

    target_h = max(MOD_VALUE, (int(height) // MOD_VALUE) * MOD_VALUE)
    target_w = max(MOD_VALUE, (int(width) // MOD_VALUE) * MOD_VALUE)
    
    num_frames = np.clip(int(round(duration_seconds * FIXED_FPS)), MIN_FRAMES_MODEL, MAX_FRAMES_MODEL)
    
    current_seed = random.randint(0, MAX_SEED) if randomize_seed else int(seed)

    with torch.inference_mode():
        output_frames_list = pipe(
            prompt=prompt, negative_prompt=negative_prompt,
            height=target_h, width=target_w, num_frames=num_frames,
            guidance_scale=float(guidance_scale), num_inference_steps=int(steps),
            generator=torch.Generator(device="cuda").manual_seed(current_seed)
        ).frames[0]

    with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmpfile:
        video_path = tmpfile.name
    export_to_video(output_frames_list, video_path, fps=FIXED_OUTPUT_FPS)
    #upscale_and_upload_4k(video_path, prompt)
    return video_path, current_seed
    
    

with gr.Blocks(css="body { max-width: 100vw; overflow-x: hidden; }") as demo:
    gr.HTML('<meta name="viewport" content="width=device-width, initial-scale=1">')
    # ... your other components here ...
    gr.Markdown("# ⚡ InstaVideo - FastWan2.2 Demo")
  
    # Add notice for limited spaces
    if IS_ORIGINAL_SPACE:
        gr.Markdown("⚠️ **This free public demo limits the resolution to 640px, duration to 2s, and inference steps to 4. For full capabilities please duplicate this space.**")
    
    with gr.Row():
        with gr.Column():
            prompt_input = gr.Textbox(label="Prompt", value=default_prompt_t2v, placeholder="Describe the video you want to generate...")
            
            with gr.Accordion("Advanced Settings", open=False):
                negative_prompt_input = gr.Textbox(label="Negative Prompt", value=default_negative_prompt, lines=3)
                seed_input = gr.Slider(label="Seed", minimum=0, maximum=MAX_SEED, step=1, value=42, interactive=True)
                randomize_seed_checkbox = gr.Checkbox(label="Randomize seed", value=True, interactive=True)
                with gr.Row():
                    height_input = gr.Slider(
                        minimum=SLIDER_MIN_H, 
                        maximum=SLIDER_MAX_H, 
                        step=MOD_VALUE, 
                        value=min(DEFAULT_H_SLIDER_VALUE, SLIDER_MAX_H), 
                        label=f"Output Height (multiple of {MOD_VALUE})"
                    )
                    width_input = gr.Slider(
                        minimum=SLIDER_MIN_W, 
                        maximum=SLIDER_MAX_W, 
                        step=MOD_VALUE, 
                        value=min(DEFAULT_W_SLIDER_VALUE, SLIDER_MAX_W), 
                        label=f"Output Width (multiple of {MOD_VALUE})"
                    )
                duration_seconds_input = gr.Slider(
                    minimum=round(MIN_FRAMES_MODEL/FIXED_FPS,1), 
                    maximum=MAX_DURATION, 
                    step=0.1, 
                    value=2, 
                    label="Duration (seconds)", 
                    info=f"Clamped to model's {MIN_FRAMES_MODEL}-{MAX_FRAMES_MODEL} frames at {FIXED_FPS}fps."
                )
                steps_slider = gr.Slider(minimum=1, maximum=MAX_STEPS, step=1, value=4, label="Inference Steps") 
                guidance_scale_input = gr.Slider(minimum=0.0, maximum=20.0, step=0.5, value=1.0, label="Guidance Scale", visible=False)

            generate_button = gr.Button("Generate Video", variant="primary")
        with gr.Column():
            video_output = gr.Video(label="Generated Video", autoplay=True, interactive=False)
    
    ui_inputs = [
        prompt_input, height_input, width_input,
        negative_prompt_input, duration_seconds_input,
        guidance_scale_input, steps_slider, seed_input, randomize_seed_checkbox
    ]
    generate_button.click(fn=generate_video, inputs=ui_inputs, outputs=[video_output, seed_input])

    # Adjust examples based on space limits
    example_configs = [
        ["a majestic eagle soaring through mountain peaks, cinematic aerial view", 896, 512],
        ["a serene ocean wave crashing on a sandy beach at sunset", 448, 832],
        ["a field of flowers swaying in the wind, spring morning light", 512, 896],
    ]
    
    if IS_ORIGINAL_SPACE:
        # Limit example resolutions for limited spaces
        example_configs = [
            [example[0], min(example[1], LIMITED_MAX_RESOLUTION), min(example[2], LIMITED_MAX_RESOLUTION)]
            for example in example_configs
        ]
    
    gr.Examples(
        examples=example_configs,
        inputs=[prompt_input, height_input, width_input], 
        outputs=[video_output, seed_input], 
        fn=generate_video, 
        cache_examples="lazy"
    )

if __name__ == "__main__":
    demo.queue().launch()