Update app.py

app.py

@@ -1,394 +1,6 @@
-import copy
-import os
-import random
+from diffusers import DiffusionPipeline
 
-os.system('pip install dashscope')
-import gradio as gr
-import dashscope
-from dashscope import VideoSynthesis
-from examples import t2v_examples, i2v_examples
-import time
+pipe = DiffusionPipeline.from_pretrained("Wan-AI/Wan2.1-FLF2V-14B-720P")
 
-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 = 10
-LIMIT_HISTORY_RUNING_TASK = 20
-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("""
-               <div style="text-align: center; font-size: 32px; font-weight: bold; margin-bottom: 20px;">
-                   Wan2.1: Open and Advanced Large-Scale Video Generative Models
-               </div>
-               <div style="text-align: center;">
-                   <a href="https://github.com/Wan-Video/Wan2.1">Code</a> |
-                   <a href="https://huggingface.co/Wan-AI">Huggingface</a> |
-                   <a href="https://modelscope.cn/organization/Wan-AI">Modelscope</a>
-               </div>
-                <div style="text-align: center; font-size: 16px; font-weight: bold; margin-bottom: 20px;">
-                   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!
-                   <a href="https://wan.video/wanxiang/creation">Wan Web</a>
-               </div>
-               """)
-    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('<span style="color: blue;">Due to automatic refresh of task status causing significant network congestion, please manually click the "Refresh Generating Status" button to check the task status.</span>')
-    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)
+prompt = "Astronaut in a jungle, cold color palette, muted colors, detailed, 8k"
+image = pipe(prompt).images[0]