import copy import os import random os.system('pip install dashscope') import gradio as gr import dashscope from dashscope import VideoSynthesis from examples import t2v_examples, i2v_examples import time DASHSCOPE_API_KEY = os.getenv('DASHSCOPE_API_KEY') dashscope.api_key = DASHSCOPE_API_KEY KEEP_SUCCESS_TASK = 3600 * 10 KEEP_RUNING_TASK = 3600 * 1 # the total running task number in 1800 seconds LIMIT_RUNING_TASK = 30 LIMIT_HISTORY_RUNING_TASK = 60 FRESH_TIME = None task_status = {} total_task_info = { "total_process_cost": 0, "total_complete_task": 0, "total_submit": 0, "latest_1h_submit_status": { } } def get_submit_code(): submit_code = random.randint(0, 2147483647) # for sub_c, sub_info in copy.deepcopy(total_task_info["latest_1h_submit_status"]).items(): if time.time() - sub_info > 3600: total_task_info["latest_1h_submit_status"].pop(sub_c) return submit_code def t2v_generation(prompt, resolution, watermark_wan, seed=-1): seed = seed if seed >= 0 else random.randint(0, 2147483647) total_task_info["latest_1h_submit_status"][get_submit_code()] = time.time() total_task_info["total_submit"] += 1 if not allow_task_num(): gr.Info(f"Warning: The number of running tasks is too large, the estimate waiting time is {get_waiting_time('-1')} s.") return None, gr.Button(visible=True) try: rsp = VideoSynthesis.call(model="wanx2.1-t2v-plus", prompt=prompt, seed=seed, watermark_wanx=watermark_wan, size=resolution) video_url = rsp.output.video_url return video_url, gr.Button(visible=True) except Exception as e: gr.Warning(f"Warning: {e}") return None, gr.Button(visible=True) def t2v_generation_async(prompt, size, watermark_wan, seed=-1): print(seed) seed = seed if seed >= 0 else random.randint(0, 2147483647) total_task_info["latest_1h_submit_status"][get_submit_code()] = time.time() total_task_info["total_submit"] += 1 print(seed) if not allow_task_num(): gr.Info(f"Warning: The number of running tasks is too large, the estimate waiting time is {get_waiting_time('-1')} s.") return None, False, gr.Button(visible=True), gr.Button(visible=False), gr.Slider(), gr.Slider() try: rsp = VideoSynthesis.async_call(model="wanx2.1-t2v-plus", prompt=prompt, size=size, seed=seed, watermark_wanx=watermark_wan) task_id = rsp.output.task_id status = False return task_id, status, gr.Button(visible=False), gr.Button(visible=True), get_cost_time(task_id), get_waiting_time(task_id) except Exception as e: gr.Warning(f"Warning: {e}") return None, True, gr.Button(), gr.Button(), gr.Slider(), gr.Slider() def i2v_generation(prompt, image, watermark_wan, seed=-1): seed = seed if seed >= 0 else random.randint(0, 2147483647) video_url = None try: rsp = VideoSynthesis.call(model="wanx2.1-i2v-plus", prompt=prompt, img_url=image, seed=seed, watermark_wanx=watermark_wan ) video_url = rsp.output.video_url except Exception as e: gr.Warning(f"Warning: {e}") return video_url def i2v_generation_async(prompt, image, watermark_wan, seed=-1): seed = seed if seed >= 0 else random.randint(0, 2147483647) total_task_info["latest_1h_submit_status"][get_submit_code()] = time.time() total_task_info["total_submit"] += 1 if not allow_task_num(): gr.Info(f"Warning: The number of running tasks is too large, the estimate waiting time is {get_waiting_time('-1')} s.") return "", None, gr.Button(visible=True), gr.Button(visible=False), gr.Slider(), gr.Slider() try: rsp = VideoSynthesis.async_call(model="wanx2.1-i2v-plus", prompt=prompt, seed=seed, img_url=image, watermark_wanx=watermark_wan) print(rsp) task_id = rsp.output.task_id status = False return task_id, status, gr.Button(visible=False), gr.Button(visible=True), get_cost_time(task_id), get_waiting_time(task_id) except Exception as e: gr.Warning(f"Warning: {e}") return "", None, gr.Button(), gr.Button(), gr.Slider(), gr.Slider() def get_result_with_task_id(task_id): if task_id == "": return True, None try: rsp = VideoSynthesis.fetch(task=task_id) print(rsp) if rsp.output.task_status == "FAILED": gr.Info(f"Warning: task running {rsp.output.task_status}") status = True video_url = None else: video_url = rsp.output.video_url video_url = video_url if video_url != "" else None status = video_url is not None if status: total_task_info["total_complete_task"] += 1 total_task_info["total_process_cost"] += time.time() - task_status[task_id]["time"] print(total_task_info["total_complete_task"], total_task_info["total_process_cost"]) except: video_url = None status = False return status, None if video_url == "" else video_url # return True, "https://dashscope-result-wlcb.oss-cn-wulanchabu.aliyuncs.com/1d/f8/20250220/e7d3f375/ccc590a2-7e90-4d92-84bc-22668db42979.mp4?Expires=1740137152&OSSAccessKeyId=LTAI5tQZd8AEcZX6KZV4G8qL&Signature=i3S3jA5FY6XYfvzZNHnvQiPzZSw%3D" def allow_task_num(): num = 0 total_num = 0 for task_id in task_status: if not task_status[task_id]["status"] and task_status[task_id]["time"] + 1800 > time.time(): num += 1 if not task_status[task_id]["status"]: total_num += 1 return num < LIMIT_RUNING_TASK or total_num < LIMIT_HISTORY_RUNING_TASK def get_waiting_time(task_id): # if the num of running task < Limit # waiting time = num * 480s # task_id not in task_status, return a large number # prediction the waiting time # avg_cost * latest submit time num = 0 for task_id in task_status: if not task_status[task_id]["status"]: num += 1 latest_submit_tasks = len(total_task_info["latest_1h_submit_status"]) print("latest submit tasks", latest_submit_tasks) if task_id in task_status: return int(640 - (time.time() - task_status[task_id]["time"])) else: return int(latest_submit_tasks * (total_task_info["total_process_cost"]/(total_task_info["total_complete_task"]+1))) def online_get_waiting_time(task, t2v_task_id, i2v_task_id): task_id = t2v_task_id if task == "t2v" else i2v_task_id return get_waiting_time(task_id) def clean_task_status(): # clean the task over 1800 seconds for task_id in copy.deepcopy(task_status): if task_id == "": continue # finished task, keep 3600 seconds if task_status[task_id]["status"]: if task_status[task_id]["time"] + KEEP_SUCCESS_TASK < time.time(): task_status.pop(task_id) else: # clean the task over 3600 * 2 seconds if task_status[task_id]["time"] + KEEP_RUNING_TASK < time.time(): task_status.pop(task_id) def get_cost_time(task_id): if task_id in task_status and not task_status[task_id]["status"]: et = int(time.time() - task_status[task_id]["time"]) return f"{et:.2f}" else: return gr.Textbox() def online_get_cost_time(task, t2v_task_id, i2v_task_id): task_id = t2v_task_id if task == "t2v" else i2v_task_id return get_cost_time(task_id) def get_process_bar(task, t2v_task_id, i2v_task_id, status): task_id = t2v_task_id if task == "t2v" else i2v_task_id clean_task_status() if task_id not in task_status: task_status[task_id] = { "value": 0 if not task_id == "" else 100, "status": status if not task_id == "" else True, "time": time.time(), "url": None } if not task_status[task_id]["status"]: # only when > 50% do check status if task_status[task_id]["value"] >= 5 and task_status[task_id]["value"] % 5 == 0: status, video_url = get_result_with_task_id(task_id) else: status, video_url = False, None task_status[task_id]["status"] = status task_status[task_id]["url"] = video_url if task_status[task_id]["status"]: task_status[task_id]["value"] = 100 else: task_status[task_id]["value"] += 5 if task_status[task_id]["value"] >= 100 and not task_status[task_id]["status"]: task_status[task_id]["value"] = 95 # print(task_id, task_status[task_id], task_status) value = task_status[task_id]["value"] return gr.Slider(label=f"({value}%)Generating" if value % 2 == 1 else f"({value}%)Generating.....", value=value) with gr.Blocks() as demo: gr.HTML("""
Wan2.1: Open and Advanced Large-Scale Video Generative Models
Code | Huggingface | Modelscope
We are excited to announce that Wan's international experience page is officially live, supporting image and video generation, and it's completely free. We welcome you to try it out! Wan Web
""") t2v_task_id = gr.State(value="") i2v_task_id = gr.State(value="") status = gr.State(value=False) task = gr.State(value="t2v") with gr.Row(): with gr.Column(): with gr.Row(): with gr.Tabs(): # Text to Video Tab with gr.TabItem("Text to Video") as t2v_tab: with gr.Row(): txt2vid_prompt = gr.Textbox( label="Prompt", placeholder="Describe the video you want to generate", lines=19, ) with gr.Row(): resolution = gr.Dropdown( label="Resolution", choices=["1280*720", "960*960", "720*1280", "1088*832", "832*1088"], value="1280*720", ) with gr.Row(): run_t2v_button = gr.Button("Generate Video") t2v_refresh_status = gr.Button("Refresh Generating Status", visible=False) # Image to Video Tab with gr.TabItem("Image to Video") as i2v_tab: with gr.Row(): with gr.Column(): img2vid_image = gr.Image( type="filepath", label="Upload Input Image", elem_id="image_upload", ) img2vid_prompt = gr.Textbox( label="Prompt", placeholder="Describe the video you want to generate", value="", lines=5, ) with gr.Row(): run_i2v_button = gr.Button("Generate Video") i2v_refresh_status = gr.Button("Refresh Generating Status", visible=False) with gr.Column(): with gr.Row(): result_gallery = gr.Video(label='Generated Video', interactive=False, height=500) with gr.Row(): watermark_wan = gr.Checkbox(label="Watermark", value=True, visible=True, container=False) seed = gr.Number(label="Seed", value=-1, container=True) # cost_time = gr.Number(label="Cost Time(secs)", value=online_get_cost_time, interactive=False, # every=FRESH_TIME, inputs=[task, t2v_task_id, i2v_task_id], container=True) cost_time = gr.Number(label="Cost Time(secs)", value=0, interactive=False, container=True) # waiting_time = gr.Number(label="Estimated Waiting Time(secs)", value=online_get_waiting_time, interactive=False, # every=FRESH_TIME, inputs=[task, t2v_task_id, i2v_task_id], container=True) waiting_time = gr.Number(label="Estimated Waiting Time(secs)", value=0, interactive=False, container=True) # process_bar = gr.Slider(show_label=True, label="", value=get_process_bar, maximum=100, # interactive=True, every=FRESH_TIME, inputs=[task, t2v_task_id, i2v_task_id, status], container=True) process_bar = gr.Slider(show_label=True, label="", value=100, maximum=100, interactive=True, container=True) with gr.Row(): gr.Markdown('Due to automatic refresh of task status causing significant network congestion, please manually click the "Refresh Generating Status" button to check the task status.') fake_video = gr.Video(label='Examples', visible=False, interactive=False) with gr.Row(visible=True) as t2v_eg: gr.Examples(t2v_examples, inputs=[txt2vid_prompt, result_gallery], outputs=[result_gallery]) with gr.Row(visible=False) as i2v_eg: gr.Examples(i2v_examples, inputs=[img2vid_prompt, img2vid_image, result_gallery], outputs=[result_gallery]) def process_change(task_id, task): status = task_status.get(task_id, {"status":False})["status"] if status: video_url = task_status[task_id]["url"] ret_t2v_btn = gr.Button(visible=True) if task == 't2v' else gr.Button() ret_t2v_status_btn = gr.Button(visible=False) if task == 't2v' else gr.Button() ret_i2v_btn = gr.Button(visible=True) if task == 'i2v' else gr.Button() ret_i2v_status_btn = gr.Button(visible=False) if task == 'i2v' else gr.Button() return gr.Video(value=video_url), ret_t2v_btn, ret_i2v_btn, ret_t2v_status_btn, ret_i2v_status_btn return gr.Video(value=None), gr.Button(), gr.Button(), gr.Button(), gr.Button() def online_process_change(task, t2v_task_id, i2v_task_id): task_id = t2v_task_id if task == 't2v' else i2v_task_id return process_change(task_id, task) process_bar.change(online_process_change, inputs=[task, t2v_task_id, i2v_task_id], outputs=[result_gallery, run_t2v_button, run_i2v_button, t2v_refresh_status, i2v_refresh_status]) def switch_i2v_tab(): return gr.Row(visible=False), gr.Row(visible=True), "i2v" def switch_t2v_tab(): return gr.Row(visible=True), gr.Row(visible=False), "t2v" i2v_tab.select(switch_i2v_tab, outputs=[t2v_eg, i2v_eg, task]) t2v_tab.select(switch_t2v_tab, outputs=[t2v_eg, i2v_eg, task]) run_t2v_button.click( fn=t2v_generation_async, inputs=[txt2vid_prompt, resolution, watermark_wan, seed], outputs=[t2v_task_id, status, run_t2v_button, t2v_refresh_status, cost_time, waiting_time], ) run_i2v_button.click( fn=i2v_generation_async, inputs=[img2vid_prompt, img2vid_image, watermark_wan, seed], outputs=[i2v_task_id, status, run_i2v_button, i2v_refresh_status, cost_time, waiting_time], ) def status_refresh(task_id, task, status): if task_id in task_status and not task_status[task_id]["status"]: cost_time = int(time.time() - task_status[task_id]["time"]) else: cost_time = 0 status, video_url = get_result_with_task_id(task_id) if task_id not in task_status: task_status[task_id] = {"status": status, "url": video_url, "time": time.time(), "value": 100 if status else 0} else: task_status[task_id]["status"] = status task_status[task_id]["url"] = video_url waiting_time = get_waiting_time(task_id) value = task_status.get(task_id, {"value": 100})["value"] value = max(value, int(cost_time*100/waiting_time)) task_status[task_id]["value"] = value if value < 100 else 100 if not video_url == "" and status: value = 100 process_bar = gr.Slider(label=f"({value}%)Generating" if value % 2 == 1 else f"({value}%)Generating.....", value=value) process_change_ret = process_change(task_id, task) return *process_change_ret, cost_time, waiting_time, process_bar t2v_refresh_status.click( fn=status_refresh, inputs=[t2v_task_id, task, status], outputs=[result_gallery, run_t2v_button, run_i2v_button, t2v_refresh_status, i2v_refresh_status, cost_time, waiting_time, process_bar] ) i2v_refresh_status.click( fn=status_refresh, inputs=[i2v_task_id, task, status], outputs=[result_gallery, run_t2v_button, run_i2v_button, t2v_refresh_status, i2v_refresh_status, cost_time, waiting_time, process_bar] ) #demo.queue(max_size=10) demo.launch(ssr_mode=False)