""" Generate carpet cleaning reel clips via Wan 2.2 TI2V (text+image to video) through ComfyUI. Uses FLUX-generated hero stills as input frames, animates each into a 3-5 second clip. Run after gen-images-flux.py completes and images are converted to webp. Usage: python3 tools/gen-video-wan.py Output: assets/videos/clips/*.mp4 — stitch with ffmpeg into final reel. """ import json, time, urllib.request, os, random, io COMFY = "http://localhost:8188" HERO_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "assets", "images", "hero") OUT_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "assets", "videos", "clips") os.makedirs(OUT_DIR, exist_ok=True) WAN_MODEL = "Wan2.2-TI2V-5B-Q4_K_M.gguf" # Each clip: input still + motion prompt → 3-5 sec animated clip # Order matches the reel sequence CLIPS = [ { "filename": "clip-01-carpet.mp4", "image": "hero-carpet-cleaning.webp", "prompt": "slow dolly forward across clean plush carpet, gentle camera push toward the far wall, warm afternoon light, cinematic, smooth motion", "frames": 49, # ~4 seconds at ~12fps }, { "filename": "clip-02-stairs.mp4", "image": "hero-stairs.webp", "prompt": "slow pan upward along clean carpeted staircase, camera tilts up following the banister, soft natural light, cinematic motion", "frames": 49, }, { "filename": "clip-03-upholstery.mp4", "image": "hero-upholstery.webp", "prompt": "gentle push in toward clean linen sofa, shallow depth of field, warm light, slow cinematic camera movement", "frames": 49, }, { "filename": "clip-04-commercial.mp4", "image": "hero-commercial.webp", "prompt": "slow tracking shot moving forward down a clean corporate lobby, receding vanishing point, professional lighting, cinematic", "frames": 49, }, { "filename": "clip-05-floors.mp4", "image": "hero-floors.webp", "prompt": "floor-level drift forward along gleaming hardwood, camera slides smoothly down the hallway, natural light", "frames": 49, }, { "filename": "clip-06-clean-result.mp4", "image": "hero-clean-result.webp", "prompt": "slow rack focus across clean carpet fibers, foreground to background, raking natural light, macro detail, cinematic", "frames": 49, }, ] def load_image_as_base64(image_path): import base64 with open(image_path, "rb") as f: return base64.b64encode(f.read()).decode("utf-8") def upload_image(image_path): """Upload image to ComfyUI and return the filename it assigned.""" import base64 fname = os.path.basename(image_path) with open(image_path, "rb") as f: img_data = f.read() boundary = "----FormBoundary" + str(random.randint(100000, 999999)) body = ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="image"; filename="{fname}"\r\n' f"Content-Type: image/webp\r\n\r\n" ).encode() + img_data + f"\r\n--{boundary}--\r\n".encode() req = urllib.request.Request( f"{COMFY}/upload/image", data=body, headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}, ) with urllib.request.urlopen(req) as resp: result = json.loads(resp.read()) return result["name"] def build_workflow(image_name, prompt, frames, seed=None): if seed is None: seed = random.randint(0, 2**32) return { "1": { "class_type": "UnetLoaderGGUF", "inputs": {"unet_name": WAN_MODEL}, }, "2": { "class_type": "CLIPLoader", "inputs": { "clip_name": "umt5_xxl_fp8_e4m3fn_scaled.safetensors", "type": "wan", }, }, "3": { "class_type": "VAELoader", "inputs": {"vae_name": "wan_2.1_vae.safetensors"}, }, "4": { "class_type": "LoadImage", "inputs": {"image": image_name}, }, "5": { "class_type": "CLIPTextEncode", "inputs": {"clip": ["2", 0], "text": prompt}, }, "6": { "class_type": "CLIPTextEncode", "inputs": {"clip": ["2", 0], "text": "blur, low quality, distortion, text, watermark, people, faces"}, }, "7": { "class_type": "WanImageToVideo", "inputs": { "model": ["1", 0], "clip": ["2", 0], "vae": ["3", 0], "image": ["4", 0], "positive": ["5", 0], "negative": ["6", 0], "width": 832, "height": 480, "length": frames, "batch_size": 1, "seed": seed, "steps": 20, "cfg": 6.0, "sampler_name": "uni_pc", "scheduler": "simple", "denoise": 1.0, }, }, "8": { "class_type": "SaveAnimatedWEBP", "inputs": { "images": ["7", 0], "filename_prefix": "wan_lahr", "fps": 16, "lossless": False, "quality": 85, "method": "default", }, }, } def queue_prompt(workflow): data = json.dumps({"prompt": workflow}).encode() req = urllib.request.Request( f"{COMFY}/prompt", data=data, headers={"Content-Type": "application/json"} ) with urllib.request.urlopen(req) as resp: return json.loads(resp.read())["prompt_id"] def wait_for_result(prompt_id, timeout=1800): start = time.time() while time.time() - start < timeout: try: with urllib.request.urlopen(f"{COMFY}/history/{prompt_id}") as resp: hist = json.loads(resp.read()) if prompt_id in hist: entry = hist[prompt_id] if entry.get("status", {}).get("status_str") == "error": print(f" ERROR: {entry['status'].get('messages', '')}", flush=True) return None for node_out in entry.get("outputs", {}).values(): if "images" in node_out: return node_out["images"] if "gifs" in node_out: return node_out["gifs"] except Exception: pass time.sleep(8) print(" waiting...", flush=True) return None def download_video(vid_info, out_path): fname = vid_info["filename"] subfolder = vid_info.get("subfolder", "") img_type = vid_info.get("type", "output") url = f"{COMFY}/view?filename={fname}&subfolder={subfolder}&type={img_type}" with urllib.request.urlopen(url) as resp: data = resp.read() with open(out_path, "wb") as f: f.write(data) print(f" saved: {os.path.basename(out_path)} ({len(data)//1024}KB)", flush=True) total = len(CLIPS) for i, clip in enumerate(CLIPS): print(f"\n[{i+1}/{total}] {clip['filename']}", flush=True) image_path = os.path.join(HERO_DIR, clip["image"]) if not os.path.exists(image_path): print(f" SKIP: {image_path} not found", flush=True) continue print(f" uploading {clip['image']}...", flush=True) image_name = upload_image(image_path) workflow = build_workflow(image_name, clip["prompt"], clip["frames"]) prompt_id = queue_prompt(workflow) print(f" queued {prompt_id[:8]}...", flush=True) results = wait_for_result(prompt_id) if results: out_path = os.path.join(OUT_DIR, clip["filename"]) # rename .webp output to .mp4 for compatibility — or save as webp animation out_path_webp = out_path.replace(".mp4", ".webp") download_video(results[0], out_path_webp) else: print(f" FAILED", flush=True) print("\nAll clips done. Stitch with:") print(f" ffmpeg -f concat -safe 0 -i tools/clip-list.txt -c copy assets/videos/hero-reel-flux.mp4")