init

app.py

@@ -43,13 +43,71 @@ def is_http_resource_accessible(url):
         return False
 
 
+def onPoseChange(prompt_text, source_image, request: gr.Request):
+    """处理姿势变换请求"""
+    if source_image is None:
+        return "请先提供源图片!", None, None, None
+    
+    if not prompt_text or prompt_text.strip() == "":
+        prompt_text = "Change the pose: two hands on hips.#Change the pose: arms extended to show outfit."
+    
+    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
+        
+        # 如果source_image是本地文件路径，需要先上传
+        if isinstance(source_image, str) and not source_image.startswith('http'):
+            timeId = int(str(time.time()).replace(".", "")) + random.randint(1000, 9999)
+            image_url = upload_pose_img(client_ip, timeId, source_image)
+            if not image_url:
+                return "图片上传失败!", None, None, None
+        else:
+            image_url = source_image
+        
+        # 发起姿势变换请求
+        pose_result = public_pose_changer(image_url, prompt_text)
+        if pose_result is None:
+            return "姿势变换请求失败!", None, None, None
+        
+        # 轮询结果
+        max_try = 120
+        wait_s = 1
+        for i in range(max_try):
+            time.sleep(wait_s)
+            result = get_pose_changer_res(pose_result['id'])
+            
+            if result is None:
+                continue
+            elif result['status'] == 'PROCESSING':
+                continue
+            elif result['status'] == 'SUCCEED':
+                # 提取前3个有效的输出图片
+                output_images = [None, None, None]
+                for j in range(1, 4):  # output1 到 output3
+                    output_key = f'output{j}'
+                    if output_key in result and result[output_key] and result[output_key].strip():
+                        timestamp = int(time.time() * 1000)
+                        output_images[j-1] = result[output_key] + f"?t={timestamp}"
+                
+                return f"姿势变换完成! {result.get('msg', '')}", output_images[0], output_images[1], output_images[2]
+            elif result['status'] == 'FAILED':
+                return f"姿势变换失败: {result.get('msg', '')}", None, None, None
+        
+        return "姿势变换超时!", None, None, None
+        
+    except Exception as e:
+        print(f"姿势变换异常: {e}")
+        return f"处理异常: {str(e)}", None, None, None
+
 def onClick(cloth_image, pose_image, high_resolution, request: gr.Request):
     if pose_image is None:
-        yield None, "no pose image found !", ""
-        return None, "no pose image found !", ""
+        yield None, "no pose image found !", "", None
+        return None, "no pose image found !", "", None
     if cloth_image is None:
-        yield None, "no cloth image found !", ""
-        return None, "no cloth image found !", ""
+        yield None, "no cloth image found !", "", None
+        return None, "no cloth image found !", "", None
 
     pose_id = os.path.basename(pose_image).split(".")[0]
     cloth_id = int(os.path.basename(cloth_image).split(".")[0])
@@ -65,45 +123,45 @@ def onClick(cloth_image, pose_image, high_resolution, request: gr.Request):
         faces = face_detector.detect_faces(pose_np[:,:,::-1])
         if len(faces)==0:
             print(client_ip, 'faces num is 0! ', flush=True)
-            yield None, "Fatal Error !!! No face detected !!! You must upload a human photo!!! Not clothing photo!!!", ""
-            return None, "Fatal Error !!! No face detected !!! You must upload a human photo!!! Not clothing photo!!!", ""
+            yield None, "Fatal Error !!! No face detected !!! You must upload a human photo!!! Not clothing photo!!!", "", None
+            return None, "Fatal Error !!! No face detected !!! You must upload a human photo!!! Not clothing photo!!!", "", None
         else:
             x, y, w, h = faces[0]["box"]
             H, W = pose_np.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 !!! You must upload a full-body or half-body photo!!!", ""
-                return None, "Fatal Error !!! Headshot is not allowed !!! You must upload a full-body or half-body photo!!!", ""
+                yield None, "Fatal Error !!! Headshot is not allowed !!! You must upload a full-body or half-body photo!!!", "", None
+                return None, "Fatal Error !!! Headshot is not allowed !!! You must upload a full-body or half-body photo!!!", "", None
 
         if not check_region_warp(client_ip):
-            yield None, "Failed !!! Our server is under maintenance, please try again later", ""
-            return None, "Failed !!! Our server is under maintenance, please try again later", ""
+            yield None, "Failed !!! Our server is under maintenance, please try again later", "", None
+            return None, "Failed !!! Our server is under maintenance, please try again later", "", None
             
         # client_ip = '8.8.8.8'
-        yield None, "begin to upload ", ""
+        yield None, "begin to upload ", "", None
 
         timeId = int(  str(time.time()).replace(".", "")  )+random.randint(1000, 9999)
         upload_url = upload_pose_img(client_ip, timeId, pose_image)
         # exit(0)
-        yield None, "begin to public task ", ""
+        yield None, "begin to public task ", "", None
         # return None, "begin to public task ", ""
         
         if len(upload_url)==0:
-            yield None, "fail to upload", ""
-            return None, "fail to upload", ""
+            yield None, "fail to upload", "", None
+            return None, "fail to upload", "", None
 
         if high_resolution:
             public_res = publicClothSwap(upload_url, cloth_id, is_hr=1)
         else:
             public_res = publicClothSwap(upload_url, cloth_id, is_hr=0)
         if public_res is None:
-            yield None, "fail to public you task", ""
-            return None, "fail to public you task", ""
+            yield None, "fail to public you task", "", None
+            return None, "fail to public you task", "", None
 
         print(client_ip, public_res['mid_result'])
         # Check if mid_result resource is accessible
         mid_result = public_res['mid_result'] if is_http_resource_accessible(public_res['mid_result']) else None
-        yield mid_result, f"task is processing, task id: {public_res['id']}, {public_res['msg']}", ""
+        yield mid_result, f"task is processing, task id: {public_res['id']}, {public_res['msg']}", "", mid_result
 
         max_try = 120*3
         wait_s = 0.5
@@ -114,26 +172,27 @@ def onClick(cloth_image, pose_image, high_resolution, request: gr.Request):
             if state is None:
                 mid_result = public_res['mid_result'] if is_http_resource_accessible(public_res['mid_result']) else None
                 result_url = mid_result + f"?t={timestamp}" if mid_result else None
-                yield result_url, "task query failed,", ""
+                yield result_url, "task query failed,", "", result_url
             elif state['status']=='PROCESSING':
                 mid_result = public_res['mid_result'] if is_http_resource_accessible(public_res['mid_result']) else None
                 result_url = mid_result + f"?t={timestamp}" if mid_result else None
-                yield result_url, f"task is processing, query {i}", ""
+                yield result_url, f"task is processing, query {i}", "", result_url
             elif state['status']=='SUCCEED':
-                yield state['output1'] + f"?t={timestamp}", f"task finished, {state['msg']}", ""
-                return state['output1'] + f"?t={timestamp}", f"task finished, {state['msg']}", ""
+                result_image = state['output1'] + f"?t={timestamp}"
+                yield result_image, f"task finished, {state['msg']}", "", result_image
+                return result_image, f"task finished, {state['msg']}", "", result_image
             elif state['status']=='FAILED':
-                yield None, f"task failed, {state['msg']}", ""
-                return None, f"task failed, {state['msg']}", ""
+                yield None, f"task failed, {state['msg']}", "", None
+                return None, f"task failed, {state['msg']}", "", None
             else:
                 mid_result = public_res['mid_result'] if is_http_resource_accessible(public_res['mid_result']) else None
                 result_url = mid_result + f"?t={timestamp}" if mid_result else None
-                yield result_url, f"task is on processing, query {i}", ""
-        return None, "no machine...", ""
+                yield result_url, f"task is on processing, query {i}", "", result_url
+        return None, "no machine...", "", None
     except Exception as e:
         print(e)
         raise e
-        return None, "fail to create task", ""
+        return None, "fail to create task", "", None
 
 with gr.Blocks() as demo:
     gr.Markdown(title)
@@ -167,8 +226,40 @@ with gr.Blocks() as demo:
                 res_image = gr.Image(label="result image", value=None, type="filepath")
                 MK01 = gr.Markdown()
 
+    # 添加pose changer模块
+    with gr.Accordion('pose changer', open=False):
+        # 上方：文本框和按钮
+        with gr.Row():
+            pose_prompt = gr.Textbox(
+                value="Change the pose: two hands on hips.#Change the pose: arms extended to show outfit.",
+                label="姿势变换提示词",
+                placeholder="输入姿势变换的描述...",
+                lines=2,
+                scale=4
+            )
+            change_button = gr.Button(value="Change", scale=1)
+        
+        # 下方：左边源图片，右边结果图片
+        with gr.Row():
+            with gr.Column(scale=1):
+                pose_changer_image = gr.Image(value=None, type="filepath", label="源图片", interactive=True)
+                pose_change_info = gr.Textbox(value="", interactive=False, label="处理信息")
+            
+            with gr.Column(scale=2):
+                with gr.Row():
+                    pose_result_1 = gr.Image(label="结果1", value=None, type="filepath")
+                    pose_result_2 = gr.Image(label="结果2", value=None, type="filepath")
+                    pose_result_3 = gr.Image(label="结果3", value=None, type="filepath")
+    
     run_button.click(fn=onClick, inputs=[cloth_image, pose_image, high_resolution], 
-                     outputs=[res_image, info_text, MK01])
+                     outputs=[res_image, info_text, MK01, pose_changer_image])
+    
+    # 绑定pose changer的change按钮
+    change_button.click(
+        fn=onPoseChange,
+        inputs=[pose_prompt, pose_changer_image],
+        outputs=[pose_change_info, pose_result_1, pose_result_2, pose_result_3]
+    )
 
 
 if __name__ == "__main__":

utils.py

@@ -174,3 +174,75 @@ def check_region_warp(ip):
     except Exception as e:
         print(e)
         return True
+
+def public_pose_changer(image_url, prompt="Change the pose: two hands on hips.#Change the pose: arms extended to show outfit."):
+    """
+    发送姿势变换请求到API
+    """
+    headers = {
+        'Content-Type': 'application/json',
+        'Authorization': f'Bearer {TOKEN}'
+    }
+    
+    json_data = {
+        'image': image_url,
+        'param1': 'pose_change',
+        'param2': prompt,
+        'param3': '',
+        'param4': '',
+        'param5': '',
+        'is_private': '0',
+        'delete_if_complete': '0'
+    }
+    
+    try:
+        ret = requests.post(
+            f'{UKAPIURL}/public_comfyui',
+            headers=headers,
+            json=json_data
+        )
+        
+        if ret.status_code == 200:
+            response = ret.json()
+            if 'id' in response:
+                return {'id': response['id'], 'msg': response.get('msg', '')}
+            return None
+        else:
+            print(f"姿势变换请求失败，状态码: {ret.status_code}")
+            return None
+    except Exception as e:
+        print(f"姿势变换请求异常: {e}")
+        return None
+    
+def get_pose_changer_res(task_id):
+    """
+    获取姿势变换任务的结果
+    """
+    headers = {
+        'Content-Type': 'application/json'
+    }
+    try:
+        ret = requests.post(
+            f'{UKAPIURL}/status_comfyui',
+            headers=headers,
+            json={ 'id': task_id}
+        )
+        if ret.status_code == 200:
+            response = ret.json()
+            return response
+        else:
+            print(f"获取姿势变换结果失败，状态码: {ret.status_code}")
+            return None
+    except Exception as e:
+        print(f"获取姿势变换结果异常: {e}")
+        return None
+
+if __name__ == "__main__":
+    
+    res = public_pose_changer(image_url="https://www.selfitcamera.site/SelfitUpload/2024-10-11/p2.jpg", 
+        prompt="Change the pose: two hands on hips.#Change the pose: arms extended to show outfit.")
+    print(res)
+    for _ in range(100):
+        result = get_pose_changer_res(res['id'])
+        print(result)
+        time.sleep(1)