import random import time from pathlib import Path from typing import Optional import modal import numpy as np import torch from diffusers import AutoencoderKLWan, WanImageToVideoPipeline from diffusers.utils import export_to_video, load_image from transformers import CLIPVisionModel # Create Modal app app = modal.App("wan-image-to-video") # Define the image with git installed image = ( modal.Image.debian_slim() .apt_install( "git", "libgl1-mesa-glx", # Required for OpenCV "libglib2.0-0", # Required for OpenCV "libsm6", # Required for OpenCV "libxext6", # Required for OpenCV "libxrender-dev", # Required for OpenCV ) .pip_install( "accelerate==1.4.0", "fastapi[standard]==0.115.8", "huggingface-hub[hf_transfer]==0.29.1", "hf_transfer", # Added for fast downloads "imageio-ffmpeg==0.6.0", "imageio==2.37.0", "numpy==1.26.4", "opencv-python", "pillow==11.1.0", "safetensors", "torch==2.6.0", "torchvision==0.21.0", "transformers==4.49.0", "ftfy", "git+https://github.com/huggingface/diffusers.git@main", ) ) # Model configuration MODEL_ID = "Wan-AI/Wan2.1-I2V-14B-480P-Diffusers" MODEL_ID_light = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers" # Set up volumes for model weights and outputs model_volume = modal.Volume.from_name("hf-hub-cache", create_if_missing=True) output_volume = modal.Volume.from_name("outputs", create_if_missing=True) MODEL_PATH = "/models" OUTPUT_PATH = "/outputs" image = image.env( { "HF_HUB_ENABLE_HF_TRANSFER": "1", "HF_HUB_CACHE": MODEL_PATH, } ).add_local_python_source("suno_utils", copy=False) @app.cls( image=image, gpu="H100", timeout=100 * 60, # 10 minutes volumes={MODEL_PATH: model_volume, OUTPUT_PATH: output_volume}, ) class WanInference: @modal.enter() def load_pipeline(self): self.image_encoder = CLIPVisionModel.from_pretrained( MODEL_ID, subfolder="image_encoder", torch_dtype=torch.float32 ) self.vae = AutoencoderKLWan.from_pretrained(MODEL_ID, subfolder="vae", torch_dtype=torch.float32) self.pipe = WanImageToVideoPipeline.from_pretrained( MODEL_ID, vae=self.vae, image_encoder=self.image_encoder, torch_dtype=torch.bfloat16 ).to("cuda") # 1.3B model # self.light_vae = AutoModel.from_pretrained( # "Wan-AI/Wan2.1-T2V-1.3B-Diffusers", subfolder="vae", torch_dtype=torch.float32 # ) # self.light_pipe = WanPipeline.from_pretrained( # "Wan-AI/Wan2.1-T2V-1.3B-Diffusers", vae=self.light_vae, torch_dtype=torch.bfloat16 # ) # self.light_pipe.scheduler = UniPCMultistepScheduler.from_config( # self.light_pipe.scheduler.config, flow_shift=5.0 # ) # self.light_pipe.to("cuda") @modal.method() def run( self, image_bytes: bytes, prompt: str, negative_prompt: Optional[str] = None, num_frames: Optional[int] = None, guidance_scale: Optional[float] = None, seed: Optional[int] = None, ) -> str: print("Running Wan inference") # Set default values negative_prompt = ( # negative_prompt or "worst quality, inconsistent motion, blurry, jittery, distorted" "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" ) num_frames = num_frames or 81 guidance_scale = guidance_scale or 5.0 seed = seed or random.randint(0, 2**32 - 1) print(f"Seeding RNG with: {seed}") torch.manual_seed(seed) device = torch.device("cuda") with torch.no_grad(), torch.autocast(str(device), dtype=self.pipe.transformer.dtype): # Load and process image image = load_image( # "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/astronaut.jpg" "https://cdn2.suno.ai/cef61953-86e9-49e3-ab4b-f25d9e63b35e_00b33c14.jpeg" ) # Calculate dimensions max_area = 900 * 600 aspect_ratio = image.height / image.width mod_value = self.pipe.vae_scale_factor_spatial * self.pipe.transformer.config.patch_size[1] # make it smaller height = round(np.sqrt(max_area * aspect_ratio)) // mod_value * mod_value width = round(np.sqrt(max_area / aspect_ratio)) // mod_value * mod_value image = image.resize((width, height)) # check whether gpu is in use with torch print(f"CUDA available: {torch.cuda.is_available()}") if torch.cuda.is_available(): print(f"Current GPU device: {torch.cuda.current_device()}") print(f"GPU device name: {torch.cuda.get_device_name()}") print(f"GPU memory allocated: {torch.cuda.memory_allocated() / 1024**2:.2f} MB") print(f"GPU memory cached: {torch.cuda.memory_reserved() / 1024**2:.2f} MB") else: print("No GPU available, using CPU") prompt = "A person riding on a horse, facing back to the camera, in the desert going further and further away, a line of birds flying in the sky,the camera is zooming out, " # Generate video output = self.pipe( image=image, prompt=prompt, negative_prompt=negative_prompt, height=height, width=width, num_frames=num_frames, num_inference_steps=50, guidance_scale=guidance_scale, ).frames[0] # Save video mp4_name = slugify(prompt) export_to_video(output, f"{Path(OUTPUT_PATH) / mp4_name}", fps=16) output_volume.commit() torch.cuda.empty_cache() # reduce fragmentation return mp4_name @app.local_entrypoint() def entrypoint( image_path: str, prompt: str, negative_prompt: Optional[str] = None, num_frames: Optional[int] = None, guidance_scale: Optional[float] = None, seed: Optional[int] = None, ): import os import urllib.request print(f"�� Generating a video from the image at {image_path}") print(f"🎥 using the prompt {prompt}") if image_path.startswith(("http://", "https://")): image_bytes = urllib.request.urlopen(image_path).read() elif os.path.isfile(image_path): image_bytes = Path(image_path).read_bytes() else: raise ValueError(f"{image_path} is not a valid file or URL.") inference_service = WanInference() start = time.time() print("Running inference") mp4_name = inference_service.run.remote( image_bytes=image_bytes, prompt=prompt, negative_prompt=negative_prompt, num_frames=num_frames, guidance_scale=guidance_scale, seed=seed, ) duration = time.time() - start print(f"🎥 Generated video in {duration:.3f}s") output_dir = Path("/tmp/wan_video") output_dir.mkdir(exist_ok=True, parents=True) output_path = output_dir / mp4_name output_path.write_bytes(b"".join(output_volume.read_file(mp4_name))) print(f"🎥 Video saved to {output_path}") def slugify(s: str) -> str: return f"{time.strftime('%Y%m%d_%H%M%S')}_{''.join(c if c.isalnum() else '-' for c in s[:100]).strip('-')}.mp4" # image = load_image( # "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/astronaut.jpg" # ) # uv run modal run suno_utils/worker/music_video_gen/wan_video_gen.py --image-path "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/diffusers/astronaut.jpg" --prompt "the astronaut is moving the egg shell"