Spaces:
Runtime error
Runtime error
| from utils import * | |
| cloth_examples = get_cloth_examples() | |
| pose_examples = get_pose_examples() | |
| # Description | |
| title = r""" | |
| <h1 align="center">Outfit Anyone in the Wild: Get rid of Annoying Restrictions for Virtual Try-on Task</h1> | |
| """ | |
| description = r""" | |
| <b>Official 🤗 Gradio demo</b> for <a href='https://github.com/selfitcamera/Outfit-Anyone-in-the-Wild' target='_blank'><b>Outfit Anyone in the Wild: Get rid of Annoying Restrictions for Virtual Try-on Task</b></a>.<br> | |
| 1. Clothing models are fixed in this demo, but you can create your own in SelfitCamera WeChat applet (for Chainese users). | |
| 2. You can upload your own pose photo, then click the run button and wait for 3~5 minutes to see the results. | |
| 3. After submitting the task, feel free to leave this page. Everytime you refresh this page, completed tasks will be displayed on the history tab (bind with your ip address). | |
| 4. Share your try-on photo with your friends and enjoy! 😊""" | |
| css = """ | |
| .gradio-container {width: 85% !important} | |
| """ | |
| mk_guide = "If image does not display successfully after button clicked in your browser(mostly Mac+Chrome), try [this demo](https://openxlab.org.cn/apps/detail/jiangxiaoguo/OutfitAnyone-in-the-Wild) please" | |
| def onUpload(): | |
| return "" | |
| def onClick(cloth_id, pose_image, pose_id, size, request: gr.Request): | |
| if pose_image is None: | |
| return None, "no pose image found !", "" | |
| # pose_id, cloth_id = pose_id['label'], cloth_id['label'] | |
| # print(pose_id, cloth_id, size, (pose_image is None), len(pose_id)>0) | |
| if len(pose_id)>0: | |
| res = get_result_example(cloth_id, pose_id) | |
| # print(res) | |
| assert os.path.exists(res), res | |
| # res = cv2.imread(res) | |
| return res, "Done! Use the pre-run results directly, the cloth size does not take effect ", mk_guide | |
| else: | |
| try: | |
| client_ip = request.client.host | |
| x_forwarded_for = dict(request.headers).get('x-forwarded-for') | |
| if x_forwarded_for: | |
| client_ip = x_forwarded_for | |
| timeId = int( str(time.time()).replace(".", "") )+random.randint(1000, 9999) | |
| isUpload = upload_pose_img(ApiUrl, OpenId, ApiKey, client_ip, timeId, pose_image) | |
| if isUpload==0: | |
| return None, "fail to upload", "" | |
| elif isUpload==-1: | |
| return None, "There is a running task already, please wait and check the history tab", "" | |
| elif isUpload==-2: | |
| return None, "can not creat task, you have exhausted free trial quota", "" | |
| taskId = publicClothSwap(ApiUrl, OpenId, ApiKey, client_ip, cloth_id, timeId, size) | |
| if taskId==0: | |
| return None, "fail to public you task", "" | |
| max_try = 1 | |
| wait_s = 30 | |
| for i in range(max_try): | |
| time.sleep(wait_s) | |
| state = getInfRes(ApiUrl, OpenId, ApiKey, client_ip, timeId) | |
| if state=='stateIs-1': | |
| return None, "task failed, it may be that no human was detected, or there may be illegal content, etc. ", "" | |
| elif state=='stateIs0': | |
| return None, "task not public success", "" | |
| elif len(state)>20: | |
| return state, "task finished", "" | |
| elif (not state.startswith('stateIs')): | |
| # return None, 'task is in queue, position is '+str(state) | |
| pass | |
| else: | |
| return None, state, "" | |
| return None, "task has been created successfully, you can refresh the page 5~15 mins latter, and check the following history tab", "" | |
| except Exception as e: | |
| print(e) | |
| return None, "fail to create task", "" | |
| def onLoad(request: gr.Request): | |
| client_ip = request.client.host | |
| x_forwarded_for = dict(request.headers).get('x-forwarded-for') | |
| if x_forwarded_for: | |
| client_ip = x_forwarded_for | |
| his_datas = [None for _ in range(10)] | |
| info = '' | |
| try: | |
| infs = getAllInfs(ApiUrl, OpenId, ApiKey, client_ip) | |
| print(client_ip, 'history infs: ', len(infs)) | |
| cnt = 0 | |
| finish_n, fail_n, queue_n = 0, 0, 0 | |
| for i, inf in enumerate(infs): | |
| if inf['state']==2: | |
| if cnt>4: continue | |
| pose, res = inf['pose'], inf['res'] | |
| his_datas[cnt*2] = f"<img src=\"{pose}\" >" | |
| his_datas[cnt*2+1] = f"<img src=\"{res}\" >" | |
| finish_n += 1 | |
| cnt += 1 | |
| elif inf['state'] in [-1, -2, 0]: | |
| fail_n += 1 | |
| elif inf['state'] in [1]: | |
| queue_n += 1 | |
| info = f"{client_ip}, you have {finish_n} successed tasks, {queue_n} running tasks, {fail_n} failed tasks." | |
| if queue_n>0: | |
| info = info+" Wait for 3~10 mins and refresh this page, successed results will display in the history tab at the bottom" | |
| time.sleep(3) | |
| except Exception as e: | |
| print(e) | |
| his_datas = his_datas + [info] | |
| return his_datas | |
| with gr.Blocks(css=css) as demo: | |
| # description | |
| gr.Markdown(title) | |
| gr.Markdown(description) | |
| with gr.Row(): | |
| with gr.Column(): | |
| with gr.Column(): | |
| # cloth_image = gr.Image(type="numpy", value=cloth_examples[0][1], label="") | |
| cloth_image = gr.Image(sources='clipboard', type="filepath", label="", | |
| value=None) | |
| cloth_id = gr.Label(value=cloth_examples[0][0], label="Clothing 3D Model", visible=False) | |
| example = gr.Examples(inputs=[cloth_id, cloth_image], | |
| examples_per_page=3, | |
| examples = cloth_examples) | |
| with gr.Column(): | |
| with gr.Column(): | |
| # pose_image = gr.Image(source='upload', value=pose_examples[0][1], | |
| # type="numpy", label="") | |
| pose_image = gr.Image(value=None, | |
| type="numpy", label="") | |
| pose_id = gr.Label(value=pose_examples[0][0], label="Pose Image", visible=False) | |
| example_pose = gr.Examples(inputs=[pose_id, pose_image], | |
| examples_per_page=3, | |
| examples=pose_examples) | |
| size_slider = gr.Slider(-2.5, 2.5, value=1, interactive=True, label="clothes size") | |
| with gr.Column(): | |
| with gr.Column(): | |
| run_button = gr.Button(value="Run") | |
| init_res = get_result_example(cloth_examples[0][0], pose_examples[0][0]) | |
| res_image = gr.Image(label="result image", value=None, type="filepath") | |
| # res_image = gr.Image(label="result image", value=None, type="numpy") | |
| # res_image = gr.Image(label="result image", value=cv2.imread(init_res), | |
| # type="numpy") | |
| MK01 = gr.Markdown() | |
| info_text = gr.Textbox(value="", interactive=False, | |
| label='runtime information') | |
| with gr.Tab('history'): | |
| with gr.Row(): | |
| MK02 = gr.Markdown() | |
| with gr.Row(): | |
| his_pose_image1 = gr.HTML() | |
| his_res_image1 = gr.HTML() | |
| with gr.Row(): | |
| his_pose_image2 = gr.HTML() | |
| his_res_image2 = gr.HTML() | |
| with gr.Row(): | |
| his_pose_image3 = gr.HTML() | |
| his_res_image3 = gr.HTML() | |
| with gr.Row(): | |
| his_pose_image4 = gr.HTML() | |
| his_res_image4 = gr.HTML() | |
| with gr.Row(): | |
| his_pose_image5 = gr.HTML() | |
| his_res_image5 = gr.HTML() | |
| run_button.click(fn=onClick, inputs=[cloth_id, pose_image, pose_id, size_slider], | |
| outputs=[res_image, info_text, MK01], concurrency_limit=50) | |
| pose_image.upload(fn=onUpload, inputs=[], outputs=[pose_id],) | |
| demo.load(onLoad, inputs=[], outputs=[his_pose_image1, his_res_image1, | |
| his_pose_image2, his_res_image2, his_pose_image3, his_res_image3, | |
| his_pose_image4, his_res_image4, his_pose_image5, his_res_image5, | |
| MK02]) | |
| if __name__ == "__main__": | |
| demo.queue(max_size=50) | |
| # demo.queue(concurrency_count=60) | |
| # demo.launch(server_name='0.0.0.0', server_port=225) | |
| demo.launch(server_name='0.0.0.0') | |