xsl/change_cloth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

save code

Browse Source
This commit is contained in:
xsl
2025-10-04 12:53:01 +08:00
parent 20e6b294f1
commit 1fbb91eadb
2 changed files with 185 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
change_app.py
+90 -92
View File
@@ -236,7 +236,7 @@ def generate_from_face(human_name, sex_girl):
return out_img_file_name
return None
def change(human_name, cloth_name, c_width, c_height, cloth_url, human_url, no2, tuodi, kuzi):
def change(human_name, cloth_name, c_width, c_height, cloth_url, human_url, no2, tuodi):
human_json_str = GetHumanDesDesc(human_url)
print(f"human_json_str {human_json_str}")
human_json_data = json.loads(human_json_str)
@@ -247,10 +247,6 @@ def change(human_name, cloth_name, c_width, c_height, cloth_url, human_url, no2,
sgwzd = human_json_data['身高完整度']
if kuzi:
cloth_len = "裤子"
cloth_short = False
else:
json_str = GetPicDesc(cloth_url)
json_data = json.loads(json_str)
cloth_len = json_data['服装长度']
@@ -260,19 +256,13 @@ def change(human_name, cloth_name, c_width, c_height, cloth_url, human_url, no2,
print(f"GetPicDesc:{json_str}")
print(f"tuodi:{tuodi}")
print(f"kuzi:{kuzi}")
if tuodi == True or tuodi == "true":
cloth_len = '拖地'
if kuzi == True:
cloth_len = "裤子"
if no2 == True:
print(f'不用第二部 no2{no2}')
else:
if kuzi == True:
print(f'有裤子，也不用写真 kuzi{kuzi}')
else:
print(f'要第二部 no2{no2} 先搞第一步')
if sgwzd >= 0.55:
@@ -334,74 +324,6 @@ def change(human_name, cloth_name, c_width, c_height, cloth_url, human_url, no2,
response = requests.post("http://localhost:8188/prompt", data=data)
prompt_id = response.json()["prompt_id"]
async def listen_node_output(prompt_id, target_node_id):
uri = "ws://localhost:8188/ws"
async with websockets.connect(uri) as websocket:
await websocket.send(json.dumps({"prompt_id": prompt_id}))
while True:
message = await websocket.recv()
# 1. 处理二进制数据
if isinstance(message, bytes):
try:
message = message.decode("utf-8")
except UnicodeDecodeError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] 收到二进制数据 (长度: {len(message)} bytes)")
continue
# 2. 解析JSON
try:
data = json.loads(message)
print(f"cur data {data}")
except json.JSONDecodeError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] 非JSON数据: {message[:100]}...")
continue
if data.get("type") == "progress":
pdata = data.get('data', {})
print(f"cur progress {pdata}")
# if value >= max:
# return f"finished value:{value} max:{max}"
if data.get("type") == "progress_state":
nodes = data.get('data', {}).get('nodes', {})
for node in nodes:
if node.get("state", None) != 'finished':
print(f"progress_state node:{node}")
# 3. 输出运行状态
if data.get("type") == "status":
print(f"[{datetime.now().strftime('%H:%M:%S')}] 系统状态: {data.get('data', {}).get('status', {})}")
status = data.get('data', {}).get('status', None)
print(f"statue:{status}")
if status:
exec_info = status.get('exec_info', None)
print(f"exec_info:{exec_info}")
if exec_info:
queue_remaining = exec_info.get('queue_remaining', None)
print(f"queue_remaining:{queue_remaining}")
if queue_remaining == 0:
return "process end"
continue
if data.get("type") == "executing":
node_id = data.get("data", {}).get("node")
progress = data.get("data", {}).get("progress", 0)
print(f"[{datetime.now().strftime('%H:%M:%S')}] 正在执行节点 {node_id} (进度: {progress:.0%})")
# 4. 目标节点完成
if data.get("type") == "executed" and data.get("data", {}).get("node") == target_node_id:
output = data.get("data", {}).get("output")
print(f"[{datetime.now().strftime('%H:%M:%S')}] 节点 {target_node_id} 完成!")
return output
# output = asyncio.run(listen_node_output(prompt_id, "104"))
# print("最终输出:", output)
while True:
# response = requests.post("http://localhost:8188/interrupt")
queue = requests.get("http://localhost:8188/queue").json()
@@ -423,6 +345,61 @@ def change(human_name, cloth_name, c_width, c_height, cloth_url, human_url, no2,
return out_img_file_name, human_name, sex_girl,"success"
return None, human_name, sex_girl, "can not find output image"
def change_kuzi(human_name, kuzi_name):
queue = requests.get("http://localhost:8188/queue").json()
if queue["queue_running"] or queue["queue_pending"]:
return None, "cur gpu is busy, Shou not happen"
print(f'开始换裤子 {change_only_name}')
with open(f'{APP_ROOT}/{change_only_name}', 'r', encoding='utf-8') as file:
prompt_text = file.read()
prompt = json.loads(prompt_text)
prompt["96"]["inputs"]["cloth_len"] = "裤子"
prompt["96"]["inputs"]["cloth_short"] = False
# prompt["99"]["inputs"]["width"] = int((c_width/c_height) * 1024)
#input cloth img
prompt["22"]["inputs"]["image"] = kuzi_name
#input human img
prompt["61"]["inputs"]["image"] = human_name
#out put name
out_img_name = str(uuid.uuid4())[:8]
prompt["102"]["inputs"]["filename_prefix"] = out_img_name
p = {"prompt": prompt}
data = json.dumps(p).encode('utf-8')
response = requests.post("http://localhost:8188/prompt", data=data)
prompt_id = response.json()["prompt_id"]
while True:
# response = requests.post("http://localhost:8188/interrupt")
queue = requests.get("http://localhost:8188/queue").json()
# print(queue)
if not queue["queue_running"] and not queue["queue_pending"]:
break # 队列为空，任务已完成
time.sleep(0.5) # 避免频繁请求
# 3. 从历史记录中获取结果
history = requests.get("http://localhost:8188/history").json()
# print("History:", history)
outputs = history[prompt_id]["outputs"]
for out in outputs:
if 'images' in outputs[out]:
for out_img in outputs[out]['images']:
if 'filename' in out_img:
out_img_file_name = out_img['filename']
if out_img_name in out_img_file_name:
return out_img_file_name,"success"
return None, "can not find output image"
def save_base64_image(base64_str, prefix):
"""
将base64字符串保存为图片文件
@@ -529,10 +506,10 @@ def upload_to_oss(image_path, object_name=None):
return None
def process_change_cloth(human_filename, cloth_filename, output_format, img_url, human_url, no2, tuodi, kuzi):
def process_change_cloth(human_filename, cloth_filename, output_format, img_url, human_url, no2, tuodi, kuzi_url):
w,h = get_image_dimensions(f'{APP_ROOT}/../input/{human_filename}')
out_put_name, out_human_name, is_girl, msg = change(human_filename, cloth_filename, w, h, img_url, human_url, no2, tuodi, kuzi)
out_put_name, out_human_name, is_girl, msg = change(human_filename, cloth_filename, w, h, img_url, human_url, no2, tuodi, kuzi_url)
if out_put_name == None:
print(f'Failed to change cloth {msg}')
return jsonify({"ret":-1, 'msg': f'Failed to change cloth {msg}'}), 200
@@ -551,7 +528,6 @@ def process_change_cloth(human_filename, cloth_filename, output_format, img_url,
out_human_https_url = f'https://llyz.oss-cn-beijing.aliyuncs.com/{out_human_jpg_name}'
print(f"生成的第一步图片 HTTPS URL: {out_human_https_url}")
image = Image.open(f'{APP_ROOT}/../output/{out_put_name}')
jpg_name = out_put_name.replace(".png", ".jpg")
jpg_path_name = f'{APP_ROOT}/static/imgs/{jpg_name}'
@@ -562,11 +538,27 @@ def process_change_cloth(human_filename, cloth_filename, output_format, img_url,
https_url = f'https://llyz.oss-cn-beijing.aliyuncs.com/{jpg_name}'
print(f"生成的HTTPS URL: {https_url}")
out_befor_kuzi_url = https_url
if kuzi_url:
kuzi_filename = save_image_from_url(kuzi_url)
kuzi_out_put_name = change_kuzi(human_filename, kuzi_filename)
image = Image.open(f'{APP_ROOT}/../output/{kuzi_out_put_name}')
jpg_name = kuzi_out_put_name.replace(".png", ".jpg")
jpg_path_name = f'{APP_ROOT}/static/imgs/{jpg_name}'
image.save(jpg_path_name, quality=90)
upload_to_oss(jpg_path_name, jpg_name) # 第二个参数可选，指定OSS上的路径
https_url = f'https://llyz.oss-cn-beijing.aliyuncs.com/{jpg_name}'
print(f"生成的加上换裤子的 HTTPS URL: {https_url}")
if 'base64' in output_format:
return jsonify({
"ret":0,
"state": 0,
"msg":"success",
"second_step_datea":image_to_base64(out_befor_kuzi_url),
"first_step_data":image_to_base64(out_human_jpg_path_name),
"is_girl":is_girl,
"data":image_to_base64(jpg_path_name)
@@ -576,6 +568,7 @@ def process_change_cloth(human_filename, cloth_filename, output_format, img_url,
"ret":0,
"state": 0,
"msg":"success",
"second_url":out_befor_kuzi_url,
"first_url":out_human_https_url,
"is_girl":is_girl,
"url":https_url
@@ -651,13 +644,15 @@ def do_change_cloth():
tuodi = data['tuodi']
print(f"tuodi:{tuodi}")
kuzi = False
if 'kuzi' in data:
kuzi = data['kuzi']
print(f"kuzi:{kuzi}")
if 'kuzi_url' in data:
kuzi_url = data.get("kuzi_url")
kuzi_filename = save_image_from_url(kuzi_url)
print(f"要弄裤子 kuzi:{kuzi_url}")
try:
return process_change_cloth(human_filename, cloth_filename, output_format, cloth_url, human_url, no2, tuodi, kuzi)
return process_change_cloth(human_filename, cloth_filename, output_format, cloth_url, human_url, no2, tuodi, kuzi_url)
except Exception as e:
traceback.print_exc()
print(f"错误详情：{e}")
@@ -780,6 +775,13 @@ def change_cloth_base64():
data["human_url"] = human_url
data["cloth_url"] = cloth_url
kuzi_img = data.get('kuzi_img')
if kuzi_img:
kuzi_filename = save_base64_image(kuzi_img, 'kuzi')
data['kuzi_img'] = None
kuzi_url = f"http://112.126.94.241:{base64_test_port}/static/imgs/{kuzi_filename}"
data['kuzi_url'] = kuzi_url
no2 = data.get('no2')
if not no2:
data['no2'] = False
@@ -788,10 +790,6 @@ def change_cloth_base64():
if not tuodi:
data['tuodi'] = False
kuzi = data.get('kuzi')
if not kuzi:
data['kuzi'] = False
# 在内部调用第二个HTTP请求
try:
# 调用第二个API（可以是外部服务或自己的另一个端点）
check_img_body.py
+67
View File
@@ -1,6 +1,73 @@
import cv2
import mediapipe as mp
async def listen_node_output(prompt_id, target_node_id):
uri = "ws://localhost:8188/ws"
async with websockets.connect(uri) as websocket:
await websocket.send(json.dumps({"prompt_id": prompt_id}))
while True:
message = await websocket.recv()
# 1. 处理二进制数据
if isinstance(message, bytes):
try:
message = message.decode("utf-8")
except UnicodeDecodeError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] 收到二进制数据 (长度: {len(message)} bytes)")
continue
# 2. 解析JSON
try:
data = json.loads(message)
print(f"cur data {data}")
except json.JSONDecodeError:
print(f"[{datetime.now().strftime('%H:%M:%S')}] 非JSON数据: {message[:100]}...")
continue
if data.get("type") == "progress":
pdata = data.get('data', {})
print(f"cur progress {pdata}")
# if value >= max:
# return f"finished value:{value} max:{max}"
if data.get("type") == "progress_state":
nodes = data.get('data', {}).get('nodes', {})
for node in nodes:
if node.get("state", None) != 'finished':
print(f"progress_state node:{node}")
# 3. 输出运行状态
if data.get("type") == "status":
print(f"[{datetime.now().strftime('%H:%M:%S')}] 系统状态: {data.get('data', {}).get('status', {})}")
status = data.get('data', {}).get('status', None)
print(f"statue:{status}")
if status:
exec_info = status.get('exec_info', None)
print(f"exec_info:{exec_info}")
if exec_info:
queue_remaining = exec_info.get('queue_remaining', None)
print(f"queue_remaining:{queue_remaining}")
if queue_remaining == 0:
return "process end"
continue
if data.get("type") == "executing":
node_id = data.get("data", {}).get("node")
progress = data.get("data", {}).get("progress", 0)
print(f"[{datetime.now().strftime('%H:%M:%S')}] 正在执行节点 {node_id} (进度: {progress:.0%})")
# 4. 目标节点完成
if data.get("type") == "executed" and data.get("data", {}).get("node") == target_node_id:
output = data.get("data", {}).get("output")
print(f"[{datetime.now().strftime('%H:%M:%S')}] 节点 {target_node_id} 完成!")
return output
# output = asyncio.run(listen_node_output(prompt_id, "104"))
# print("最终输出:", output)
# def is_thigh_visible(image_path):
# # 初始化MediaPipe Pose模型
# mp_pose = mp.solutions.pose