init

app.py

@@ -0,0 +1,115 @@
+
+from mtcnn.mtcnn import MTCNN
+from utils import *
+
+
+face_detector = MTCNN()
+
+# Description
+title = r"""
+<h1 align="center">IDM-VTON + Outfit Anyone in the Wild </h1>
+"""
+
+description = r"""
+This demo combines <b>IDM-VTON </b> and <b>Outfit Anyone in the Wild </b>
+1. Human pose detection and reconstruction using large human model from Outfit Anyone in the Wild.
+2. Use IDM-VTON for training-free try-on.
+3. Applying the refine network from Outfit Anyone in the Wild.<br>
+"""
+
+css = """
+.gradio-container {width: 85% !important}
+"""
+
+
+def onClick(cloth_image, pose_image, category, 
+        caption, request: gr.Request):
+    if pose_image is None:
+        yield None, f"no user image found !"
+        return None, "no user image found !"
+    elif cloth_image is None:
+        yield None, f"no cloth image found !"
+        return None, "no cloth image found !"
+    try:
+        faces = face_detector.detect_faces(pose_image[:,:,::-1])
+        if len(faces)==0:
+            print(client_ip, 'faces num is 0! ', flush=True)
+            yield None, "Fatal Error !!! No face detected in pose image !!! "
+            return None, "Fatal Error !!! No face detected in pose image !!! "
+        else:
+            x, y, w, h = faces[0]["box"]
+            H, W = pose_image.shape[:2]
+            max_face_ratio = 1/3.3
+            if w/W>max_face_ratio or h/H>max_face_ratio:
+                yield None, "Fatal Error !!! Headshot is not allowed in pose image!!!"
+                return None, "Fatal Error !!! Headshot is not allowed in pose image!!!"
+        
+        uploads = upload_imgs(ApiUrl, UploadToken, cloth_image, pose_image)
+        if uploads is None:
+            yield None, "fail to upload"
+            return None, "fail to upload"
+
+        infId = publicFastSwap(ApiUrl, OpenId, ApiKey, uploads, category, caption)
+        if not infId:
+            yield None, "fail to public you task"
+            return None, "fail to public you task"
+
+        max_try = 30
+        wait_s = 3
+        yield None, "start to process, please wait..."
+        for i in range(max_try):
+            time.sleep(wait_s)
+            taskStatus = getTaskRes(ApiUrl, infId)
+            if taskStatus is None: continue
+
+            status = taskStatus['status']
+            if status in ['FAILED', 'CANCELLED', 'TIMED_OUT', ]:
+                yield None, f"task failed, query {i}, status {status}"
+                return None, f"task failed, query {i}, status {status}"
+            elif status in ['IN_QUEUE', 'IN_PROGRESS', 'IN_QUEUE', ]:
+                pass
+                yield None, f"task is on processing, query {i}, status {status}, please do not exit !!!"
+            elif status=='COMPLETED':
+                out = taskStatus['output']['job_results']['output1']
+                yield out, f"task is COMPLETED"
+                return out, f"{i} task COMPLETED"
+        yield None, "fail to query task.."
+        return None, "fail to query task.."
+
+
+    except Exception as e:
+        print(e)
+        return None, "fail to create task"
+
+
+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(value=None, type="numpy", label="cloth")
+        with gr.Column():
+            with gr.Column():
+                pose_image = gr.Image(value=None, type="numpy", label="user photo")
+        with gr.Column():
+            with gr.Column():
+                category = gr.Dropdown(value="upper_cloth", choices=["upper_cloth", 
+                    "lower_cloth", "full_body", "dresses"], interactive=True)
+                caption = gr.Textbox(value="", interactive=True, label='cloth caption')
+                
+                info_text = gr.Textbox(value="", interactive=False, label='runtime information')
+                run_button = gr.Button(value="Run")
+                res_image = gr.Image(label="result image", value=None, type="filepath")
+
+    run_button.click(fn=onClick, inputs=[cloth_image, pose_image, 
+        category, caption, ], 
+        outputs=[res_image, info_text, ])
+
+if __name__ == "__main__":
+
+    demo.queue(max_size=50)
+    demo.launch(server_name='0.0.0.0')
+    

requirements.txt

@@ -0,0 +1,9 @@
+opencv-python
+numpy
+requests
+gradio==3.41.2
+gradio-client==0.5.0
+mtcnn
+tensorflow
+func_timeout
+httpx==0.24.1

utils.py

@@ -0,0 +1,113 @@
+
+import os
+import sys
+import cv2
+import json
+import random
+import time
+import requests
+import func_timeout
+import numpy as np
+import gradio as gr
+
+
+ApiUrl = os.environ['ApiUrl']
+OpenId = os.environ['OpenId']
+ApiKey = os.environ['ApiKey']
+UploadToken = os.environ['UploadToken']
+
+
+proj_dir = os.path.dirname(os.path.abspath(__file__))
+data_dir = os.path.join(proj_dir, 'Datas')
+# data_dir = "Datas"
+tmpFolder = "tmp"
+os.makedirs(tmpFolder, exist_ok=True)
+
+
+
+def upload_imgs(apiUrl, UploadToken, cloth_image, pose_image):
+    folder = os.path.join(tmpFolder, str(random.randint(0, 100)))
+    os.makedirs(folder, exist_ok=True)
+    pose_path = os.path.join(folder, 'pose.jpg')
+    cloth_path = os.path.join(folder, 'cloth.jpg')
+    cv2.imwrite(pose_path, pose_image[:,:,::-1])
+    cv2.imwrite(cloth_path, cloth_image[:,:,::-1])
+
+    params = {'token':UploadToken,
+        'input1':'pose.jpg', 
+        'input2':'cloth.jpg', 
+        'protocol':'https'}
+    session = requests.session()
+    ret = requests.post(f"{apiUrl}/upload", data=json.dumps(params))
+    if ret.status_code==200:
+        if 'upload1' in ret.json():
+            data = ret.json()
+            with open(cloth_path, 'rb') as file:
+                headers = {"Content-Type": 'image/jpeg'}
+                response = requests.put(data['upload2'], data=file, headers=headers)
+                if response.status_code == 200:
+                    print("上传成功")
+                else:
+                    print(f"上传失败，状态码: {response.status_code}, 响应内容: {response.text}")
+                    return 
+            with open(pose_path, 'rb') as file:
+                response = requests.put(data['upload1'], data=file, headers=headers)
+                if response.status_code == 200:
+                    print("上传成功")
+                else:
+                    print(f"上传失败，状态码: {response.status_code}, 响应内容: {response.text}")
+                    return 
+            if os.path.exists(pose_path): os.remove(pose_path)
+            if os.path.exists(cloth_path): os.remove(cloth_path)
+            return {'pose':data['upload1'], 'cloth':data['upload2']}
+
+def publicFastSwap(apiUrl, openId, apiKey, uploads, category, caption):
+    if category=="upper_cloth":
+        category = 1
+    elif category=="lower_cloth":
+        category = 2
+    elif category=="dresses":
+        category = 3
+    elif category=="full_body":
+        category = 4
+    params = {'openId':OpenId, 'apiKey':ApiKey, 
+        'task_type':"10", 'image':str(uploads['pose']), 
+        'mask':str(uploads['cloth']), 
+        'param1':str(category), 'param2':str(caption), 
+        'param3':'', 'param4':'', 'param5':'', }
+    session = requests.session()
+    ret = requests.post(f"{ApiUrl}/public", data=json.dumps(params))
+    if ret.status_code==200:
+        if 'id' in ret.json():
+            print('public task successfully!')
+            return ret.json()['id']
+
+def getTaskRes(apiUrl, taskId):
+    params = {'id':taskId, 'task_type':"10"}
+    session = requests.session()
+    ret = requests.post(f"{apiUrl}/status", data=json.dumps(params))
+    if ret.status_code==200:
+        if 'status' in ret.json():
+            return ret.json()
+    else:
+        print(ret.json(), ret.status_code, 'call status failed')
+        return None
+
+@func_timeout.func_set_timeout(10)
+def check_func(ip):
+    session = requests.session()
+    ret = requests.get(f"https://webapi-pc.meitu.com/common/ip_location?ip={ip}")
+    for k in ret.json()['data']:
+        nat = ret.json()['data'][k]['nation']
+        if nat.lower() in Regions.lower():
+            print(nat, 'invalid')
+            return False
+        else:
+            print(nat, 'valid')
+    return True
+def check_warp(ip):
+    try:
+        return check_func(ip)
+    except Exception as e:
+        print(e)
+        return True