aidlab/AiDA_Python
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: aidlab:8ccbbe41b1a426753725bca04d8ff384ba79d554
Branches Tags
aidlab:master aidlab:develop aidlab:dev-ltx aidlab:fix aidlab:local
..
compare: aidlab:local
Branches Tags
aidlab:develop aidlab:master aidlab:dev-ltx aidlab:fix aidlab:local

1 Commits
8ccbbe41b1 ... local

Author SHA1 Message Date
zhh
94c3d1e30d feat（新功能）: 
fix（修复bug）:  修复配饰位置异常
docs（文档变更）:
refactor（重构）:
test(增加测试):
 2025-09-16 16:57:33 +08:00
14 changed files with 40 additions and 1289 deletions
Expand all files Collapse all files

2
README.md
View File

@@ -42,4 +42,4 @@ Docker 部署
3. 查看日志 3. 查看日志
$ docker-compose logs -f $ docker-compose logs -f

49
app/api/api_pose_transform.py
View File

@@ -1,13 +1,10 @@
import json import json
import logging import logging
import requests
from fastapi import APIRouter, BackgroundTasks, HTTPException from fastapi import APIRouter, BackgroundTasks, HTTPException
from app.core.config import COMFYUI_SERVER_ADDRESS
from app.schemas.pose_transform import PoseTransformModel from app.schemas.pose_transform import PoseTransformModel
from app.schemas.response_template import ResponseModel from app.schemas.response_template import ResponseModel
from app.service.comfyui_I2V.server import ComfyUIServerI2V
from app.service.generate_image.service_pose_transform import PoseTransformService, infer_cancel as pose_transform_infer_cancel from app.service.generate_image.service_pose_transform import PoseTransformService, infer_cancel as pose_transform_infer_cancel
router = APIRouter() router = APIRouter()
@@ -50,49 +47,3 @@ def pose_transform_cancel(tasks_id: str):
logger.warning(f"pose_transform_cancel Run Exception @@@@@@:{e}") logger.warning(f"pose_transform_cancel Run Exception @@@@@@:{e}")
raise HTTPException(status_code=404, detail=str(e)) raise HTTPException(status_code=404, detail=str(e))
return ResponseModel(data=data['data']) return ResponseModel(data=data['data'])
@router.post("/comfyui_pose_transform")
def comfyui_pose_transform(request_item: PoseTransformModel, background_tasks: BackgroundTasks):
"""
创建一个具有以下参数的请求体:
- **tasks_id**: 任务id 用于取消生成任务和获取生成结果
- **image_url**: 被生成图片的S3或minio url地址
- **pose_id**: 1
示例参数：
{
"tasks_id": "123-89",
"image_url": "aida-results/result_0000b606-1902-11ef-9424-0242ac180002.png",
"pose_id": "1"
}
"""
try:
logger.info(f"pose_transform request item is : @@@@@@:{json.dumps(request_item.dict())}")
service = ComfyUIServerI2V(request_item)
background_tasks.add_task(service.get_result)
except Exception as e:
logger.warning(f"pose_transform Run Exception @@@@@@:{e}")
raise HTTPException(status_code=404, detail=str(e))
return ResponseModel()
@router.get("/comfyui_pose_transform_cancel/{tasks_id}")
def pose_transform_cancel(tasks_id: str):
try:
logger.info(f"pose_transform_cancel request item is : @@@@@@:{tasks_id}")
response = requests.post(
f"http://{COMFYUI_SERVER_ADDRESS}/interrupt",
json={"prompt_id": tasks_id}
)
data = {}
if response.status_code == 200:
data['data']['message'] = "任务已成功中断"
else:
data['data']['message'] = f"中断失败：{response.text}"
logger.info(f"pose_transform_cancel response @@@@@@:{data}")
except Exception as e:
logger.warning(f"pose_transform_cancel Run Exception @@@@@@:{e}")
raise HTTPException(status_code=404, detail=str(e))
return ResponseModel(data=data['data'])

3
app/core/config.py
View File

@@ -230,6 +230,3 @@ TABLE_CATEGORIES = {
"male_bottoms": "male/bottoms", "male_bottoms": "male/bottoms",
"male_outwear": "male/outwear" "male_outwear": "male/outwear"
} }
# --- ComfyUI 配置信息 ---
COMFYUI_SERVER_ADDRESS = "10.1.2.227:8080" # 替换为您的 ComfyUI 服务器地址

699
app/service/comfyui_I2V/server.py
View File

@@ -1,699 +0,0 @@
import io
import json
import logging
import os
import random
import tempfile
import time
import uuid
import redis
import requests
from PIL import Image
from minio import Minio, S3Error
from moviepy.video.io.VideoFileClip import VideoFileClip
from app.core.config import REDIS_HOST, REDIS_PORT, REDIS_DB, MINIO_URL, MINIO_ACCESS, MINIO_SECRET, MINIO_SECURE, COMFYUI_SERVER_ADDRESS, PS_RABBITMQ_QUEUES, DEBUG
from app.schemas.pose_transform import PoseTransformModel
from app.service.generate_image.utils.mq import publish_status
logger = logging.getLogger()
workflow_json = {
"162": {
"inputs": {
"text": "色调艳丽，过曝，静态，细节模糊不清，字幕，风格，作品，画作，画面，静止，整体发灰，最差质量，低质量，JPEG压缩残留，丑陋的，残缺的，多余的手指，画得不好的手部，画得不好的脸部，畸形的，毁容的，形态畸形的肢体，手指融合，静止不动的画面，杂乱的背景，三条腿，背景人很多，倒着走",
"clip": [
"167",
0
]
},
"class_type": "CLIPTextEncode",
"_meta": {
"title": "CLIP Text Encode (Negative Prompt)"
}
},
"163": {
"inputs": {
"fps": 24,
"images": [
"164",
0
]
},
"class_type": "CreateVideo",
"_meta": {
"title": "创建视频"
}
},
"164": {
"inputs": {
"samples": [
"175",
0
],
"vae": [
"168",
0
]
},
"class_type": "VAEDecode",
"_meta": {
"title": "VAE解码"
}
},
"165": {
"inputs": {
"unet_name": "wan2.2_fun_control_high_noise_14B_fp8_scaled.safetensors",
"weight_dtype": "default"
},
"class_type": "UNETLoader",
"_meta": {
"title": "UNet加载器"
}
},
"166": {
"inputs": {
"unet_name": "wan2.2_fun_control_low_noise_14B_fp8_scaled.safetensors",
"weight_dtype": "default"
},
"class_type": "UNETLoader",
"_meta": {
"title": "UNet加载器"
}
},
"167": {
"inputs": {
"clip_name": "umt5_xxl_fp8_e4m3fn_scaled.safetensors",
"type": "wan",
"device": "default"
},
"class_type": "CLIPLoader",
"_meta": {
"title": "加载CLIP"
}
},
"168": {
"inputs": {
"vae_name": "wan_2.1_vae.safetensors"
},
"class_type": "VAELoader",
"_meta": {
"title": "加载VAE"
}
},
"169": {
"inputs": {
"add_noise": "enable",
"noise_seed": 8860422635573,
"steps": 4,
"cfg": 1,
"sampler_name": "euler",
"scheduler": "simple",
"start_at_step": 0,
"end_at_step": 2,
"return_with_leftover_noise": "enable",
"model": [
"176",
0
],
"positive": [
"180",
0
],
"negative": [
"180",
1
],
"latent_image": [
"180",
2
]
},
"class_type": "KSamplerAdvanced",
"_meta": {
"title": "K采样器（高级）"
}
},
"170": {
"inputs": {
"filename_prefix": "video/wan2.2_fun_control",
"format": "auto",
"codec": "auto",
"video-preview": "",
"video": [
"163",
0
]
},
"class_type": "SaveVideo",
"_meta": {
"title": "保存视频"
}
},
"171": {
"inputs": {
"video": [
"174",
0
]
},
"class_type": "GetVideoComponents",
"_meta": {
"title": "获取视频组件"
}
},
"174": {
"inputs": {
"file": "skeleton_3.mp4"
},
"class_type": "LoadVideo",
"_meta": {
"title": "加载视频"
}
},
"175": {
"inputs": {
"add_noise": "disable",
"noise_seed": 0,
"steps": 4,
"cfg": 1,
"sampler_name": "euler",
"scheduler": "simple",
"start_at_step": 2,
"end_at_step": 4,
"return_with_leftover_noise": "disable",
"model": [
"177",
0
],
"positive": [
"180",
0
],
"negative": [
"180",
1
],
"latent_image": [
"169",
0
]
},
"class_type": "KSamplerAdvanced",
"_meta": {
"title": "K采样器（高级）"
}
},
"176": {
"inputs": {
"shift": 8.000000000000002,
"model": [
"181",
0
]
},
"class_type": "ModelSamplingSD3",
"_meta": {
"title": "采样算法（SD3）"
}
},
"177": {
"inputs": {
"shift": 8.000000000000002,
"model": [
"182",
0
]
},
"class_type": "ModelSamplingSD3",
"_meta": {
"title": "采样算法（SD3）"
}
},
"178": {
"inputs": {
"image": "296f5fd6-c5e4-4003-9798-f378a4f08411-0-89.png"
},
"class_type": "LoadImage",
"_meta": {
"title": "加载图像"
}
},
"179": {
"inputs": {
"text": "On a sunny summer day, there are marshmallow - like clouds, and the sunlight is bright and warm. A girl with white curly double - ponytails is wearing unique sunglasses, distinctive clothes and shoes. Her posture is natural and full of dynamic tension. The background is the scene of the Leaning Tower of Pisa in Italy, emphasizing the realistic contrast of details in reality. The whole picture is in a realistic 3D style, rich in details and with a relaxed atmosphere. She is dancing slowly, waving her hands.",
"clip": [
"167",
0
]
},
"class_type": "CLIPTextEncode",
"_meta": {
"title": "CLIP Text Encode (Positive Prompt)"
}
},
"180": {
"inputs": {
"width": 512,
"height": 768,
"length": 121,
"batch_size": 1,
"positive": [
"179",
0
],
"negative": [
"162",
0
],
"vae": [
"168",
0
],
"ref_image": [
"178",
0
],
"control_video": [
"171",
0
]
},
"class_type": "Wan22FunControlToVideo",
"_meta": {
"title": "Wan22FunControlToVideo"
}
},
"181": {
"inputs": {
"lora_name": "wan2.2_i2v_lightx2v_4steps_lora_v1_high_noise.safetensors",
"strength_model": 1,
"model": [
"165",
0
]
},
"class_type": "LoraLoaderModelOnly",
"_meta": {
"title": "LoRA加载器（仅模型）"
}
},
"182": {
"inputs": {
"lora_name": "wan2.2_i2v_lightx2v_4steps_lora_v1_low_noise.safetensors",
"strength_model": 1,
"model": [
"166",
0
]
},
"class_type": "LoraLoaderModelOnly",
"_meta": {
"title": "LoRA加载器（仅模型）"
}
}
}
pose_video_map = {
"1": "input_pose_video/1.mp4",
"2": "input_pose_video/2.mp4",
"3": "input_pose_video/3.mp4",
"4": "input_pose_video/4.mp4",
"5": "input_pose_video/5.mp4",
"6": "input_pose_video/6.mp4"
}
class ComfyUIServerI2V:
def __init__(self, request_data):
self.image_url = request_data.image_url
self.pose_num = request_data.pose_id
self.tasks_id = request_data.tasks_id
self.user_id = self.tasks_id[self.tasks_id.rfind('-') + 1:]
self.redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True)
self.pose_transform_data = {'tasks_id': self.tasks_id, 'status': 'PENDING', 'message': "pending", 'gif_url': '', 'video_url': '', 'image_url': ''}
self.redis_client.set(self.tasks_id, json.dumps(self.pose_transform_data))
self.redis_client.expire(self.tasks_id, 600)
self.minio_client = Minio(MINIO_URL, access_key=MINIO_ACCESS, secret_key=MINIO_SECRET, secure=MINIO_SECURE)
def get_result(self):
workflow_json['174']['inputs']['file'] = pose_video_map[self.pose_num]
workflow_json['169']['inputs']['noise_seed'] = random.randint(0, 10 ** 18)
# 下载图片 上传 comfyui server
in_memory_file, object_name = self.download_from_minio_in_memory()
if in_memory_file and object_name:
uploaded_filename = self.upload_in_memory_file_to_comfyui(in_memory_file, object_name)
workflow_json['178']['inputs']['image'] = uploaded_filename
# 1. 提交任务
prompt_response = self.queue_prompt(workflow_json, self.tasks_id)
if not prompt_response:
return
prompt_id = prompt_response.get("prompt_id")
logger.info(f" 任务已提交，Prompt ID: {prompt_id}")
outputs = self.poll_history(prompt_id)
file_list = {}
for node_id, node_output in outputs.items():
# 检查当前节点输出中是否包含 'images' 列表
if 'images' in node_output and isinstance(node_output['images'], list):
# 'images' 列表中的每个元素都是一个文件对象
for file_info in node_output['images']:
# 确保关键字段存在
if all(key in file_info for key in ['filename', 'subfolder', 'type']):
file_list = {
'filename': file_info['filename'],
'subfolder': file_info['subfolder'],
'type': file_info['type']
}
logger.info(file_list)
return self.process_and_upload_comfyui_video(filename=file_list['filename'], subfolder=file_list['subfolder'], prompt_id=prompt_response['prompt_id']), prompt_id
def read_tasks_status(self):
status_data = self.redis_client.get(self.tasks_id)
return json.loads(status_data), status_data
def download_from_minio_in_memory(self):
bucket = self.image_url.split('/')[0]
object_name = self.image_url[self.image_url.find('/') + 1:]
# print("🚀 正在连接 MinIO 客户端...")
try:
# get_object 返回一个 ResponseStream 对象
response_stream = self.minio_client.get_object(
bucket,
object_name,
)
# 读取整个流到内存 (BytesIO)，避免写入本地文件
image_bytes = response_stream.read()
response_stream.close()
response_stream.release_conn()
in_memory_file = io.BytesIO(image_bytes)
# print(f"✅ 图片已下载到内存 ({len(image_bytes)} 字节)。")
return in_memory_file, object_name.rsplit('/')[-1]
except S3Error as e:
logger.error(f"❌ MinIO S3 错误 (例如，对象不存在): {e}")
return None, None
except Exception as e:
logger.error(f"❌ MinIO 下载过程中发生未知错误: {e}")
return None, None
def upload_video_to_minio(self, BUCKET_NAME, OBJECT_NAME, LOCAL_FILE_PATH):
"""使用 fput_object 从本地路径上传 MP4 文件"""
try:
# 使用 fput_object 上传文件
# content_type 对于视频流播放非常重要，MP4 文件应使用 'video/mp4'
result = self.minio_client.fput_object(
bucket_name=BUCKET_NAME,
object_name=OBJECT_NAME,
file_path=LOCAL_FILE_PATH,
content_type="video/mp4" # 设置正确的内容类型
)
# print(f"✅ 文件 '{LOCAL_FILE_PATH}' 已成功上传至 MinIO:")
# print(f" 对象名: {result.object_name}")
# print(f" Etag: {result.etag}")
except S3Error as e:
logger.error(f"❌ MinIO 操作失败: {e}")
except FileNotFoundError:
logger.error(f"❌ 找不到本地文件: {LOCAL_FILE_PATH}")
except Exception as e:
logger.error(f"❌ 发生未知错误: {e}")
def upload_gif_to_minio(self, BUCKET_NAME, OBJECT_NAME, LOCAL_FILE_PATH):
"""使用 fput_object 从本地路径上传 MP4 文件"""
try:
# 使用 fput_object 上传文件
# content_type 对于视频流播放非常重要，MP4 文件应使用 'video/mp4'
result = self.minio_client.fput_object(
bucket_name=BUCKET_NAME,
object_name=OBJECT_NAME,
file_path=LOCAL_FILE_PATH,
content_type="video/mp4" # 设置正确的内容类型
)
# print(f"✅ 文件 '{LOCAL_FILE_PATH}' 已成功上传至 MinIO:")
# print(f" 对象名: {result.object_name}")
# print(f" Etag: {result.etag}")
except S3Error as e:
logger.error(f"❌ MinIO 操作失败: {e}")
except FileNotFoundError:
logger.error(f"❌ 找不到本地文件: {LOCAL_FILE_PATH}")
except Exception as e:
logger.error(f"❌ 发生未知错误: {e}")
def upload_in_memory_file_to_comfyui(self, in_memory_file, filename):
upload_url = f"http://{COMFYUI_SERVER_ADDRESS}/upload/image"
data = {
"overwrite": "true",
"type": "input"
}
# 构建 multipart/form-data: (文件名, 内存文件对象, MIME 类型)
# MIME 类型可以根据实际图片类型修改，这里使用常见的 png/jpeg
mime_type = 'image/png' if filename.lower().endswith('.png') else 'image/jpeg'
files = {
'image': (filename, in_memory_file, mime_type)
}
# print(f"⬆️ 正在上传图片 ({filename}) 到 ComfyUI...")
try:
comfyui_response = requests.post(upload_url, data=data, files=files)
comfyui_response.raise_for_status()
result = comfyui_response.json()
uploaded_name = result.get('name')
# print(f"🎉 ComfyUI 上传成功! 服务器文件名: {uploaded_name}")
return uploaded_name
except requests.exceptions.RequestException as e:
logger.error(f"❌ ComfyUI 上传失败: {e}")
logger.error(f" 响应内容: {comfyui_response.text}")
return None
def process_and_upload_comfyui_video(self, filename: str, subfolder: str, prompt_id: str, ):
"""
完整的自动化流程：获取 ComfyUI 视频 -> 转换 GIF 并提取帧 -> 上传所有结果到 MinIO。
"""
# 1. 从 ComfyUI 获取视频二进制数据
mp4_bytes = self.get_comfyui_video_bytes(filename, subfolder)
if not mp4_bytes:
return
# 2. 准备进行视频处理
# moviepy 不支持直接使用 bytes，需要将 bytes 写入一个 BytesIO 或临时文件
# 为了避免写磁盘，我们将使用 BytesIO，但 MoviePy 内部依赖 FFmpeg，有时需要一个可寻址的本地文件路径。
# 最可靠且避免写本地的方案是在内存中操作，然后将结果上传。
# ⚠️ 关键点：将 mp4_bytes 写入 BytesIO 以模拟文件，供 moviepy 读取
# 定义输出对象名
output_base_name = uuid.uuid4().hex
MP4_OBJECT = f"{self.user_id}/pose_transform_video/{prompt_id}/{output_base_name}.mp4"
GIF_OBJECT = f"{self.user_id}/pose_transform_gif/{prompt_id}/{output_base_name}.gif"
FRAME_OBJECT = f"{self.user_id}/pose_transform_first_img/{prompt_id}/{output_base_name}_frame.jpg"
# --- 视频处理和帧提取 ---
try:
# 1. 创建一个临时的 MP4 文件路径
# delete=False 确保文件在关闭后仍然存在，直到我们手动删除
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp_file:
tmp_file.write(mp4_bytes) # 将内存数据写入磁盘
temp_mp4_path = tmp_file.name # 记录文件路径
# print(f"临时文件已写入: {temp_mp4_path}")
# 2. 使用 moviepy 打开临时文件 (传入文件路径字符串)
clip = VideoFileClip(temp_mp4_path)
# --- 在这里进行所有的视频处理和提取操作 ---
# 提取第一帧 (保持原尺寸)
frame_array = clip.get_frame(t=0.0)
image = Image.fromarray(frame_array)
frame_stream = io.BytesIO()
image.save(frame_stream, 'JPEG')
frame_bytes = frame_stream.getvalue()
logger.info("✅ 成功提取第一帧图片。")
# 视频转 GIF (使用另一个临时文件来保存 GIF)
temp_gif_path = ""
with tempfile.NamedTemporaryFile(suffix=".gif", delete=False) as tmp_file:
temp_gif_path = tmp_file.name
target_fps = int(round(clip.fps)) if clip.fps else 24
clip.write_gif(temp_gif_path, fps=target_fps)
with open(temp_gif_path, 'rb') as f:
gif_bytes = f.read()
logger.info("✅ 成功生成 GIF。")
# 返回结果 (例如: 上传到 MinIO)
# return mp4_bytes, gif_bytes, frame_bytes
# -----------------------------------------------
except Exception as e:
logger.error(f"❌ 视频处理或文件操作失败: {e}")
# 在失败时，也尝试清理文件
finally:
# 3. 清理临时文件 (非常重要!)
if os.path.exists(temp_mp4_path):
os.remove(temp_mp4_path)
logger.info(f"🗑️ 已删除临时 MP4 文件: {temp_mp4_path}")
if 'temp_gif_path' in locals() and os.path.exists(temp_gif_path):
os.remove(temp_gif_path)
logger.info(f"🗑️ 已删除临时 GIF 文件: {temp_gif_path}")
# 3. 上传所有结果到 MinIO
try:
# 上传原始 MP4
self.upload_stream_to_minio(mp4_bytes, MP4_OBJECT, "video/mp4")
# 上传生成的 GIF
self.upload_stream_to_minio(gif_bytes, GIF_OBJECT, "image/gif")
# 上传第一帧图片
self.upload_stream_to_minio(frame_bytes, FRAME_OBJECT, "image/jpeg")
self.pose_transform_data = {'tasks_id': self.tasks_id, 'status': 'SUCCESS', 'message': "success", 'gif_url': f'aida-users/{GIF_OBJECT}', 'video_url': f'aida-users/{MP4_OBJECT}', 'image_url': f'aida-users/{FRAME_OBJECT}'}
# 推送消息
if not DEBUG:
publish_status(json.dumps(self.pose_transform_data), PS_RABBITMQ_QUEUES)
logger.info(
f" [x] Sent to： {PS_RABBITMQ_QUEUES} data：@@@@ {json.dumps(self.pose_transform_data, indent=4)}")
return "\n🎉 所有任务完成！"
except Exception as e:
logger.error(e)
return None
# --- 辅助函数：提交任务到队列 ---
def queue_prompt(self, prompt, client_id):
"""向 ComfyUI 提交工作流提示。"""
p = {"prompt": prompt, "client_id": client_id, "prompt_id": client_id}
data = json.dumps(p).encode('utf-8')
# 提交任务到 /prompt 端点
response = requests.post(f"http://{COMFYUI_SERVER_ADDRESS}/prompt", data=data)
# print(f"-------------{response.text}")
# print(f"------------{client_id}")
if response.status_code == 200:
return response.json()
else:
logger.warning(f"提交任务失败，状态码: {response.status_code}")
logger.warning(response.text)
return None
def poll_history(self, prompt_id, interval_seconds=5):
"""步骤 2: 轮询 /history/{prompt_id} 检查任务是否完成"""
url = f"http://{COMFYUI_SERVER_ADDRESS}/history/{prompt_id}"
logger.info(f"⏳ 开始轮询状态 (间隔 {interval_seconds} 秒)...")
while True:
time.sleep(interval_seconds)
try:
response = requests.get(url)
# 任务未完成时，ComfyUI可能会返回404或空响应，我们只关注成功响应
if response.status_code == 200:
history_data = response.json()
# ComfyUI 返回的历史记录结构是 {prompt_id: {outputs: ...}}
if prompt_id in history_data:
logger.info("🎉 任务已完成!")
return history_data[prompt_id]['outputs']
logger.info("⏳ 任务仍在执行或等待中...")
except requests.exceptions.RequestException as e:
# 处理可能的连接错误，但通常不会在内部轮询中发生
logger.info(f"⚠️ 轮询时发生错误: {e}")
pass
def get_comfyui_video_bytes(self, filename: str, subfolder: str, file_type: str = "output"):
"""
从 ComfyUI 的 /view 端点获取视频文件的二进制数据。
参数:
- filename: 视频文件名 (例如: 'ComfyUI_00002_.mp4')
- subfolder: 存储子文件夹 (例如: 'ComfyUI_2025-10-31')
- file_type: 文件类型 (通常是 'output')
返回:
- 视频文件的二进制内容 (bytes) 或 None。
"""
url = f"http://{COMFYUI_SERVER_ADDRESS}/view"
params = {
"filename": filename,
"subfolder": subfolder,
"type": file_type
}
logger.info(f"📡 正在从 ComfyUI 下载视频: {filename}")
try:
# 使用 requests.get 下载文件
response = requests.get(url, params=params, stream=True)
response.raise_for_status() # 检查 HTTP 错误
# 返回文件的完整二进制内容
return response.content
except requests.exceptions.RequestException as e:
logger.error(f"❌ 从 ComfyUI 获取视频失败: {e}")
return None
def upload_stream_to_minio(self, video_bytes: bytes, object_name: str, content_type: str):
"""从内存流上传数据到 MinIO。"""
logger.info(f"☁️ 正在上传对象到 MinIO: {object_name}")
try:
data_stream = io.BytesIO(video_bytes)
result = self.minio_client.put_object(
bucket_name='aida-users',
object_name=object_name,
data=data_stream,
length=len(video_bytes),
content_type=content_type
)
logger.info(f"✅ MinIO 上传成功: {result.object_name}")
return True
except S3Error as e:
logger.error(f"❌ MinIO 上传失败: {e}")
return False
if __name__ == '__main__':
request_data = PoseTransformModel(
tasks_id="12222515151123-89111",
image_url="aida-users/89/product_image/a6949500-2393-42ac-8723-440b5d5da2b2-0-89.png",
pose_id="6"
)
server = ComfyUIServerI2V(request_data)
print(server.get_result())

35
app/service/design_fast/design_generate.py
View File

@@ -79,7 +79,7 @@ def design_generate(request_data):
layers = sorted(layers, key=lambda s: s.get("priority", float('inf'))) layers = sorted(layers, key=lambda s: s.get("priority", float('inf')))
layers, new_size = update_base_size_priority(layers, body_size) layers, new_size = update_base_size_priority(layers, body_size)
# pattern_overall_image_url 、 pattern_print_image_url
for lay in layers: for lay in layers:
items_response['layers'].append({ items_response['layers'].append({
'image_category': "body" if lay['name'] == 'mannequin' else lay['name'], 'image_category': "body" if lay['name'] == 'mannequin' else lay['name'],
@@ -90,9 +90,7 @@ def design_generate(request_data):
'gradient_string': lay['gradient_string'] if 'gradient_string' in lay.keys() else "", 'gradient_string': lay['gradient_string'] if 'gradient_string' in lay.keys() else "",
'mask_url': lay['mask_url'], 'mask_url': lay['mask_url'],
'image_url': lay['image_url'] if 'image_url' in lay.keys() else None, 'image_url': lay['image_url'] if 'image_url' in lay.keys() else None,
'pattern_overall_image_url': lay['pattern_overall_image_url'] if 'pattern_overall_image_url' in lay.keys() else None, 'pattern_image_url': lay['pattern_image_url'] if 'pattern_image_url' in lay.keys() else None,
'pattern_print_image_url': lay['pattern_print_image_url'] if 'pattern_print_image_url' in lay.keys() else None,
# 'back_perspective_url': lay['back_perspective_url'] if 'back_perspective_url' in lay.keys() else None, # 'back_perspective_url': lay['back_perspective_url'] if 'back_perspective_url' in lay.keys() else None,
}) })
items_response['synthesis_url'] = synthesis(layers, new_size, basic) items_response['synthesis_url'] = synthesis(layers, new_size, basic)
@@ -106,9 +104,7 @@ def design_generate(request_data):
'image_url': item_result['front_image_url'], 'image_url': item_result['front_image_url'],
'mask_url': item_result['mask_url'], 'mask_url': item_result['mask_url'],
"gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "", "gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "",
'pattern_overall_image_url': item_result['pattern_overall_image_url'] if 'pattern_overall_image_url' in item_result.keys() else None, 'pattern_image_url': item_result['pattern_image_url'] if 'pattern_image_url' in item_result.keys() else None,
'pattern_print_image_url': item_result['pattern_print_image_url'] if 'pattern_print_image_url' in item_result.keys() else None,
}) })
items_response['layers'].append({ items_response['layers'].append({
'image_category': f"{item_result['name']}_back", 'image_category': f"{item_result['name']}_back",
@@ -118,9 +114,7 @@ def design_generate(request_data):
'image_url': item_result['back_image_url'], 'image_url': item_result['back_image_url'],
'mask_url': item_result['mask_url'], 'mask_url': item_result['mask_url'],
"gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "", "gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "",
'pattern_overall_image_url': item_result['pattern_overall_image_url'] if 'pattern_overall_image_url' in item_result.keys() else None, 'pattern_image_url': item_result['pattern_image_url'] if 'pattern_image_url' in item_result.keys() else None,
'pattern_print_image_url': item_result['pattern_print_image_url'] if 'pattern_print_image_url' in item_result.keys() else None,
}) })
items_response['synthesis_url'] = synthesis_single(item_result['front_image'], item_result['back_image']) items_response['synthesis_url'] = synthesis_single(item_result['front_image'], item_result['back_image'])
update_progress(process_id, total) update_progress(process_id, total)
@@ -177,9 +171,7 @@ def design_generate_v2(request_data):
'gradient_string': lay['gradient_string'] if 'gradient_string' in lay.keys() else "", 'gradient_string': lay['gradient_string'] if 'gradient_string' in lay.keys() else "",
'mask_url': lay['mask_url'], 'mask_url': lay['mask_url'],
'image_url': lay['image_url'] if 'image_url' in lay.keys() else None, 'image_url': lay['image_url'] if 'image_url' in lay.keys() else None,
'pattern_overall_image_url': lay['pattern_overall_image_url'] if 'pattern_overall_image_url' in lay.keys() else None, 'pattern_image_url': lay['pattern_image_url'] if 'pattern_image_url' in lay.keys() else None,
'pattern_print_image_url': lay['pattern_print_image_url'] if 'pattern_print_image_url' in lay.keys() else None,
# 'back_perspective_url': lay['back_perspective_url'] if 'back_perspective_url' in lay.keys() else None, # 'back_perspective_url': lay['back_perspective_url'] if 'back_perspective_url' in lay.keys() else None,
}) })
items_response['synthesis_url'] = synthesis(layers, new_size, basic) items_response['synthesis_url'] = synthesis(layers, new_size, basic)
@@ -193,9 +185,7 @@ def design_generate_v2(request_data):
'image_url': item_result['front_image_url'], 'image_url': item_result['front_image_url'],
'mask_url': item_result['mask_url'], 'mask_url': item_result['mask_url'],
"gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "", "gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "",
'pattern_overall_image_url': item_result['pattern_overall_image_url'] if 'pattern_overall_image_url' in item_result.keys() else None, 'pattern_image_url': item_result['pattern_image_url'] if 'pattern_image_url' in item_result.keys() else None,
'pattern_print_image_url': item_result['pattern_print_image_url'] if 'pattern_print_image_url' in item_result.keys() else None,
}) })
items_response['layers'].append({ items_response['layers'].append({
'image_category': f"{item_result['name']}_back", 'image_category': f"{item_result['name']}_back",
@@ -205,17 +195,15 @@ def design_generate_v2(request_data):
'image_url': item_result['back_image_url'], 'image_url': item_result['back_image_url'],
'mask_url': item_result['mask_url'], 'mask_url': item_result['mask_url'],
"gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "", "gradient_string": item_result['gradient_string'] if 'gradient_string' in item_result.keys() else "",
'pattern_overall_image_url': item_result['pattern_overall_image_url'] if 'pattern_overall_image_url' in item_result.keys() else None, 'pattern_image_url': item_result['pattern_image_url'] if 'pattern_image_url' in item_result.keys() else None,
'pattern_print_image_url': item_result['pattern_print_image_url'] if 'pattern_print_image_url' in item_result.keys() else None,
}) })
items_response['synthesis_url'] = synthesis_single(item_result['front_image'], item_result['back_image']) items_response['synthesis_url'] = synthesis_single(item_result['front_image'], item_result['back_image'])
# 发送结果给java端 # 发送结果给java端
url = JAVA_STREAM_API_URL url = JAVA_STREAM_API_URL
xu_pei_test_url = "https://137f6b5c3490.ngrok-free.app/api/third/party/receiveDesignResults" # xu_pei_test_url = "https://cd21b9110505.ngrok-free.app/api/third/party/receiveDesignResults"
tianxaing_test_url = "https://c2ae520723c9.ngrok-free.app/api/third/party/receiveDesignResults" tianxaing_test_url = "https://c2ae520723c9.ngrok-free.app/api/third/party/receiveDesignResults"
logger.info(f"java 回调 -> {url}") logger.info(f"java 回调 -> {url}")
logger.info(f"xupei java 回调 -> {xu_pei_test_url}") # logger.info(f"xupei java 回调 -> {xu_pei_test_url}")
logger.info(f"tianxiang java 回调 -> {tianxaing_test_url}") logger.info(f"tianxiang java 回调 -> {tianxaing_test_url}")
headers = { headers = {
@@ -231,14 +219,13 @@ def design_generate_v2(request_data):
# 打印结果 # 打印结果
logger.info(response.text) logger.info(response.text)
test_xp_response = post_request(xu_pei_test_url, json_data=items_response, headers=headers) # test_response = post_request(xu_pei_test_url, json_data=items_response, headers=headers)
test_response = post_request(tianxaing_test_url, json_data=items_response, headers=headers) test_response = post_request(tianxaing_test_url, json_data=items_response, headers=headers)
if test_response: if test_response:
# 打印结果 # 打印结果
# logger.info(f"xupei test response : {test_response.text}")
logger.info(f"tianxiang test response : {test_response.text}") logger.info(f"tianxiang test response : {test_response.text}")
if test_xp_response:
logger.info(f"xupei test response : {test_xp_response.text}")
for step, object in enumerate(objects_data): for step, object in enumerate(objects_data):
t = threading.Thread(target=process_object, args=(step, object)) t = threading.Thread(target=process_object, args=(step, object))

5
app/service/design_fast/item.py
View File

@@ -1,4 +1,4 @@
from app.service.design_fast.pipeline import LoadImage, KeyPoint, Segmentation, Color, PrintPainting, Scaling, Split, LoadBodyImage, ContourDetection, NoSegPrintPainting from app.service.design_fast.pipeline import LoadImage, KeyPoint, Segmentation, Color, PrintPainting, Scaling, Split, LoadBodyImage, ContourDetection
class BaseItem: class BaseItem:
@@ -19,7 +19,6 @@ class AccessoriesItem(BaseItem):
Segmentation(minio_client), Segmentation(minio_client),
# BackPerspective(minio_client), # BackPerspective(minio_client),
Color(minio_client), Color(minio_client),
NoSegPrintPainting(minio_client),
PrintPainting(minio_client), PrintPainting(minio_client),
Scaling(), Scaling(),
Split(minio_client) Split(minio_client)
@@ -40,7 +39,6 @@ class TopItem(BaseItem):
Segmentation(minio_client), Segmentation(minio_client),
# BackPerspective(minio_client), # BackPerspective(minio_client),
Color(minio_client), Color(minio_client),
NoSegPrintPainting(minio_client),
PrintPainting(minio_client), PrintPainting(minio_client),
Scaling(), Scaling(),
Split(minio_client) Split(minio_client)
@@ -62,7 +60,6 @@ class BottomItem(BaseItem):
Segmentation(minio_client), Segmentation(minio_client),
# BackPerspective(minio_client), # BackPerspective(minio_client),
Color(minio_client), Color(minio_client),
NoSegPrintPainting(minio_client),
PrintPainting(minio_client), PrintPainting(minio_client),
Scaling(), Scaling(),
Split(minio_client) Split(minio_client)

2
app/service/design_fast/pipeline/__init__.py
View File

@@ -5,7 +5,6 @@ from .keypoint import KeyPoint
from .keypoint import KeyPoint from .keypoint import KeyPoint
from .loading import LoadImage, LoadBodyImage from .loading import LoadImage, LoadBodyImage
from .print_painting import PrintPainting from .print_painting import PrintPainting
from .no_seg_print_painting import NoSegPrintPainting
from .scale import Scaling from .scale import Scaling
from .segmentation import Segmentation from .segmentation import Segmentation
from .split import Split from .split import Split
@@ -17,7 +16,6 @@ __all__ = [
'Segmentation', 'Segmentation',
'BackPerspective', 'BackPerspective',
'Color', 'Color',
'NoSegPrintPainting',
'PrintPainting', 'PrintPainting',
'Scaling', 'Scaling',
'Split' 'Split'

6
app/service/design_fast/pipeline/color.py
View File

@@ -22,7 +22,7 @@ class Color:
resize_pattern = cv2.resize(pattern, (dim_image_w, dim_image_h), interpolation=cv2.INTER_AREA) resize_pattern = cv2.resize(pattern, (dim_image_w, dim_image_h), interpolation=cv2.INTER_AREA)
# 无色 # 无色
elif "color" not in result.keys() or result['color'] == "": elif "color" not in result.keys() or result['color'] == "":
result['no_seg_sketch_overall'] = result['no_seg_sketch_print'] = result['final_image'] = result['pattern_image'] = result['single_image'] = result['image'] result['final_image'] = result['pattern_image'] = result['single_image'] = result['image']
result['alpha'] = 100 / 255.0 result['alpha'] = 100 / 255.0
return result return result
# 正常颜色 # 正常颜色
@@ -48,7 +48,7 @@ class Color:
resize_pattern[mask_3ch] = png_rgb[mask_3ch] resize_pattern[mask_3ch] = png_rgb[mask_3ch]
resize_pattern = cv2.resize(resize_pattern, (dim_image_w, dim_image_h), interpolation=cv2.INTER_AREA) resize_pattern = cv2.resize(resize_pattern, (dim_image_w, dim_image_h), interpolation=cv2.INTER_AREA)
closed_mo = np.expand_dims(result['mask'], axis=2).repeat(3, axis=2) closed_mo = np.expand_dims(result['mask'], axis=2).repeat(3, axis=2)
gray_mo = np.expand_dims(cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY), axis=2).repeat(3, axis=2) gray_mo = np.expand_dims(result['gray'], axis=2).repeat(3, axis=2)
get_image_fir = resize_pattern * (closed_mo / 255) * (gray_mo / 255) get_image_fir = resize_pattern * (closed_mo / 255) * (gray_mo / 255)
result['pattern_image'] = get_image_fir.astype(np.uint8) result['pattern_image'] = get_image_fir.astype(np.uint8)
result['final_image'] = result['pattern_image'] result['final_image'] = result['pattern_image']
@@ -59,8 +59,6 @@ class Color:
tmp2 = (result['final_image'] * (temp_fg / 255)).astype(np.uint8) tmp2 = (result['final_image'] * (temp_fg / 255)).astype(np.uint8)
result['single_image'] = cv2.add(tmp1, tmp2) result['single_image'] = cv2.add(tmp1, tmp2)
result['alpha'] = 100 / 255.0 result['alpha'] = 100 / 255.0
result['no_seg_sketch_overall'] = result['no_seg_sketch_print'] = result['final_image'].copy()
return result return result
def get_gradient(self, bucket_name, object_name): def get_gradient(self, bucket_name, object_name):

33
app/service/design_fast/pipeline/loading.py
View File

@@ -1,6 +1,5 @@
import io import io
import logging import logging
import os
import cv2 import cv2
import numpy as np import numpy as np
@@ -39,42 +38,12 @@ class LoadImage:
def __call__(self, result): def __call__(self, result):
result['image'], result['pre_mask'] = self.read_image(result['path']) result['image'], result['pre_mask'] = self.read_image(result['path'])
# if 'extract_lines' in result.keys(): result['gray'] = cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY)
# if result['extract_lines']:
# result['gray'] = self.get_lines(cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY), result['path'])
# else:
# result['gray'] = cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY)
# else:
# result['gray'] = cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY)
result['gray'] = self.get_lines(cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY), result['path'])
result['keypoint'] = self.get_keypoint(result['name']) result['keypoint'] = self.get_keypoint(result['name'])
result['img_shape'] = result['image'].shape result['img_shape'] = result['image'].shape
result['ori_shape'] = result['image'].shape result['ori_shape'] = result['image'].shape
return result return result
def get_lines(self, img, path):
binary = cv2.adaptiveThreshold(img, 255,
cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
cv2.THRESH_BINARY_INV,
25, 10)
# 步骤2：细化边缘（可选，让线条更干净）
# kernel = np.ones((1, 1), np.uint8)
# clean = cv2.morphologyEx(binary, cv2.MORPH_OPEN, kernel)
thinned = cv2.ximgproc.thinning(binary, thinningType=cv2.ximgproc.THINNING_ZHANGSUEN) # thinning算法细化线条
mask = thinned > 0
result = np.ones_like(img) * 255
result[mask] = img[mask]
# 步骤3：反转回 白底黑线
# lines = cv2.bitwise_not(thinned)
# cv2.imwrite(os.path.join('/home/user/PycharmProjects/trinity_client_aida/test/lines_original_result_5', f"Original_{path.replace('/', '-')}.png"), img)
# cv2.imwrite(os.path.join('/home/user/PycharmProjects/trinity_client_aida/test/lines_original_result_5', f"Line_{path.replace('/', '-')}.png"), result)
return result
def read_image(self, image_path): def read_image(self, image_path):
image_mask = None image_mask = None
image = oss_get_image(oss_client=self.minio_client, bucket=image_path.split("/", 1)[0], object_name=image_path.split("/", 1)[1], data_type="cv2") image = oss_get_image(oss_client=self.minio_client, bucket=image_path.split("/", 1)[0], object_name=image_path.split("/", 1)[1], data_type="cv2")

422
app/service/design_fast/pipeline/no_seg_print_painting.py
View File

@@ -1,422 +0,0 @@
import random
import cv2
import numpy as np
from PIL import Image
from app.service.utils.new_oss_client import oss_get_image
class NoSegPrintPainting:
def __init__(self, minio_client):
self.minio_client = minio_client
def __call__(self, result):
single_print = result['print']['single']
overall_print = result['print']['overall']
element_print = result['print']['element']
result['single_image'] = None
result['print_image'] = None
if overall_print['print_path_list']:
painting_dict = {'dim_image_h': result['pattern_image'].shape[0], 'dim_image_w': result['pattern_image'].shape[1]}
if "print_angle_list" in overall_print.keys() and overall_print['print_angle_list'][0] != 0:
painting_dict = self.painting_collection(painting_dict, overall_print, print_trigger=True)
painting_dict['tile_print'] = self.rotate_crop_image(img=painting_dict['tile_print'], angle=-overall_print['print_angle_list'][0], crop=True)
painting_dict['mask_inv_print'] = self.rotate_crop_image(img=painting_dict['mask_inv_print'], angle=-overall_print['print_angle_list'][0], crop=True)
# resize 到sketch大小
painting_dict['tile_print'] = self.resize_and_crop(img=painting_dict['tile_print'], target_width=painting_dict['dim_image_w'], target_height=painting_dict['dim_image_h'])
painting_dict['mask_inv_print'] = self.resize_and_crop(img=painting_dict['mask_inv_print'], target_width=painting_dict['dim_image_w'], target_height=painting_dict['dim_image_h'])
else:
painting_dict = self.painting_collection(painting_dict, overall_print, print_trigger=True, is_single=False)
result['no_seg_sketch_overall'] = result['no_seg_sketch_print'] = self.printpaint(result, painting_dict, print_=True)
result['pattern_image'] = result['no_seg_sketch_overall']
if single_print['print_path_list']:
print_background = np.zeros((result['pattern_image'].shape[0], result['pattern_image'].shape[1], 3), dtype=np.uint8)
mask_background = np.zeros((result['pattern_image'].shape[0], result['pattern_image'].shape[1], 3), dtype=np.uint8)
for i in range(len(single_print['print_path_list'])):
image, image_mode = self.read_image(single_print['print_path_list'][i])
if image_mode == "RGB":
image_rgba = cv2.cvtColor(image, cv2.COLOR_BGR2RGBA)
image = Image.fromarray(image_rgba)
new_size = (int(result['pattern_image'].shape[1] * single_print['print_scale_list'][i][0]), int(result['pattern_image'].shape[0] * single_print['print_scale_list'][i][1]))
mask = image.split()[3]
resized_source = image.resize(new_size)
resized_source_mask = mask.resize(new_size)
rotated_resized_source = resized_source.rotate(-single_print['print_angle_list'][i])
rotated_resized_source_mask = resized_source_mask.rotate(-single_print['print_angle_list'][i])
source_image_pil = Image.fromarray(cv2.cvtColor(print_background, cv2.COLOR_BGR2RGB))
source_image_pil_mask = Image.fromarray(cv2.cvtColor(mask_background, cv2.COLOR_BGR2RGB))
source_image_pil.paste(rotated_resized_source, (int(single_print['location'][i][0]), int(single_print['location'][i][1])), rotated_resized_source)
source_image_pil_mask.paste(rotated_resized_source_mask, (int(single_print['location'][i][0]), int(single_print['location'][i][1])), rotated_resized_source_mask)
print_background = cv2.cvtColor(np.array(source_image_pil), cv2.COLOR_RGBA2BGR)
mask_background = cv2.cvtColor(np.array(source_image_pil_mask), cv2.COLOR_RGBA2BGR)
ret, mask_background = cv2.threshold(mask_background, 124, 255, cv2.THRESH_BINARY)
print_mask = cv2.bitwise_and(result['mask'], cv2.cvtColor(mask_background, cv2.COLOR_BGR2GRAY))
img_fg = cv2.bitwise_or(print_background, print_background, mask=print_mask)
img_bg = cv2.bitwise_and(result['pattern_image'], result['pattern_image'], mask=cv2.bitwise_not(print_mask))
mask_mo = np.expand_dims(print_mask, axis=2).repeat(3, axis=2)
gray_mo = np.expand_dims(result['gray'], axis=2).repeat(3, axis=2)
img_fg = (img_fg * (mask_mo / 255) * (gray_mo / 255)).astype(np.uint8) # 当sketch 图像为灰色时(非纯白) ， 印花*灰度图像会导致印花在sketch上颜色变暗
# img_fg = (img_fg * (mask_mo / 255) ).astype(np.uint8) # 不过灰度图像
final_image = cv2.add(img_bg, img_fg)
canvas = np.full_like(final_image, 255)
temp_bg = np.expand_dims(cv2.bitwise_not(result['mask']), axis=2).repeat(3, axis=2)
tmp1 = (canvas * (temp_bg / 255)).astype(np.uint8)
temp_fg = np.expand_dims(result['mask'], axis=2).repeat(3, axis=2)
tmp2 = (final_image * (temp_fg / 255)).astype(np.uint8)
single_image = cv2.add(tmp1, tmp2)
result['no_seg_sketch_print'] = single_image
if element_print['element_path_list']:
print_background = np.zeros((result['final_image'].shape[0], result['final_image'].shape[1], 3), dtype=np.uint8)
mask_background = np.zeros((result['final_image'].shape[0], result['final_image'].shape[1], 3), dtype=np.uint8)
for i in range(len(element_print['element_path_list'])):
image, image_mode = self.read_image(element_print['element_path_list'][i])
if image_mode == "RGBA":
new_size = (int(result['final_image'].shape[1] * element_print['element_scale_list'][i][0]), int(result['final_image'].shape[0] * element_print['element_scale_list'][i][1]))
mask = image.split()[3]
resized_source = image.resize(new_size)
resized_source_mask = mask.resize(new_size)
rotated_resized_source = resized_source.rotate(-element_print['element_angle_list'][i])
rotated_resized_source_mask = resized_source_mask.rotate(-element_print['element_angle_list'][i])
source_image_pil = Image.fromarray(cv2.cvtColor(print_background, cv2.COLOR_BGR2RGB))
source_image_pil_mask = Image.fromarray(cv2.cvtColor(mask_background, cv2.COLOR_BGR2RGB))
source_image_pil.paste(rotated_resized_source, (int(element_print['location'][i][0]), int(element_print['location'][i][1])), rotated_resized_source)
source_image_pil_mask.paste(rotated_resized_source_mask, (int(element_print['location'][i][0]), int(element_print['location'][i][1])), rotated_resized_source_mask)
print_background = cv2.cvtColor(np.array(source_image_pil), cv2.COLOR_RGBA2BGR)
mask_background = cv2.cvtColor(np.array(source_image_pil_mask), cv2.COLOR_RGBA2BGR)
else:
mask = self.get_mask_inv(image)
mask = np.expand_dims(mask, axis=2)
mask = cv2.cvtColor(mask, cv2.COLOR_GRAY2BGR)
mask = cv2.bitwise_not(mask)
mask = cv2.resize(mask, (int(result['final_image'].shape[1] * single_print['print_scale_list'][i][0]), int(result['final_image'].shape[0] * single_print['print_scale_list'][i][1])))
image = cv2.resize(image, (int(result['final_image'].shape[1] * single_print['print_scale_list'][i][0]), int(result['final_image'].shape[0] * single_print['print_scale_list'][i][1])))
# 旋转后的坐标需要重新算
rotate_mask, _ = self.img_rotate(mask, element_print['element_angle_list'][i])
rotate_image, rotated_new_size = self.img_rotate(image, element_print['element_angle_list'][i])
# x, y = int(result['print']['location'][i][0] - rotated_new_size[0] - (rotate_mask.shape[0] - image.shape[0]) / 2), int(result['print']['location'][i][1] - rotated_new_size[1] - (rotate_mask.shape[1] - image.shape[1]) / 2)
x, y = int(element_print['location'][i][0] - rotated_new_size[0]), int(element_print['location'][i][1] - rotated_new_size[1])
image_x = print_background.shape[1]
image_y = print_background.shape[0]
print_x = rotate_image.shape[1]
print_y = rotate_image.shape[0]
if x <= 0:
rotate_image = rotate_image[:, -x:]
rotate_mask = rotate_mask[:, -x:]
start_x = x = 0
else:
start_x = x
if y <= 0:
rotate_image = rotate_image[-y:, :]
rotate_mask = rotate_mask[-y:, :]
start_y = y = 0
else:
start_y = y
if x + print_x > image_x:
rotate_image = rotate_image[:, :image_x - x]
rotate_mask = rotate_mask[:, :image_x - x]
if y + print_y > image_y:
rotate_image = rotate_image[:image_y - y, :]
rotate_mask = rotate_mask[:image_y - y, :]
mask_background = self.stack_prin(mask_background, result['pattern_image'], rotate_mask, start_y, y, start_x, x)
print_background = self.stack_prin(print_background, result['pattern_image'], rotate_image, start_y, y, start_x, x)
print_mask = cv2.bitwise_and(result['mask'], cv2.cvtColor(mask_background, cv2.COLOR_BGR2GRAY))
img_fg = cv2.bitwise_or(print_background, print_background, mask=print_mask)
three_channel_image = cv2.merge([cv2.bitwise_not(print_mask), cv2.bitwise_not(print_mask), cv2.bitwise_not(print_mask)])
img_bg = cv2.bitwise_and(result['final_image'], three_channel_image)
result['final_image'] = cv2.add(img_bg, img_fg)
canvas = np.full_like(result['final_image'], 255)
temp_bg = np.expand_dims(cv2.bitwise_not(result['mask']), axis=2).repeat(3, axis=2)
tmp1 = (canvas * (temp_bg / 255)).astype(np.uint8)
temp_fg = np.expand_dims(result['mask'], axis=2).repeat(3, axis=2)
tmp2 = (result['final_image'] * (temp_fg / 255)).astype(np.uint8)
result['no_seg_sketch_print'] = cv2.add(tmp1, tmp2)
return result
@staticmethod
def stack_prin(print_background, pattern_image, rotate_image, start_y, y, start_x, x):
temp_print = np.zeros((pattern_image.shape[0], pattern_image.shape[1], 3), dtype=np.uint8)
temp_print[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]] = rotate_image
img2gray = cv2.cvtColor(temp_print, cv2.COLOR_BGR2GRAY)
ret, mask_ = cv2.threshold(img2gray, 1, 255, cv2.THRESH_BINARY)
mask_inv = cv2.bitwise_not(mask_)
img1_bg = cv2.bitwise_and(print_background, print_background, mask=mask_inv)
img2_fg = cv2.bitwise_and(temp_print, temp_print, mask=mask_)
print_background = img1_bg + img2_fg
return print_background
def painting_collection(self, painting_dict, print_dict, print_trigger=False, is_single=False):
if print_trigger:
print_ = self.get_print(print_dict)
painting_dict['Trigger'] = not is_single
painting_dict['location'] = print_['location']
single_mask_inv_print = self.get_mask_inv(print_['image'])
dim_max = max(painting_dict['dim_image_h'], painting_dict['dim_image_w'])
dim_pattern = (int(dim_max * print_['scale'] / 5), int(dim_max * print_['scale'] / 5))
if not is_single:
self.random_seed = random.randint(0, 1000)
# 如果print 模式为overall 且 有角度的话 ， 组合的print为正方形，方便裁剪
if "print_angle_list" in print_dict.keys() and print_dict['print_angle_list'][0] != 0:
painting_dict['mask_inv_print'] = self.tile_image(single_mask_inv_print, dim_pattern, print_['scale'], dim_max, dim_max, painting_dict['location'], trigger=True)
painting_dict['tile_print'] = self.tile_image(print_['image'], dim_pattern, print_['scale'], dim_max, dim_max, painting_dict['location'], trigger=True)
else:
painting_dict['mask_inv_print'] = self.tile_image(single_mask_inv_print, dim_pattern, print_['scale'], painting_dict['dim_image_h'], painting_dict['dim_image_w'], painting_dict['location'], trigger=True)
painting_dict['tile_print'] = self.tile_image(print_['image'], dim_pattern, print_['scale'], painting_dict['dim_image_h'], painting_dict['dim_image_w'], painting_dict['location'], trigger=True)
else:
painting_dict['mask_inv_print'] = self.tile_image(single_mask_inv_print, dim_pattern, print_['scale'], painting_dict['dim_image_h'], painting_dict['dim_image_w'], painting_dict['location'])
painting_dict['tile_print'] = self.tile_image(print_['image'], dim_pattern, print_['scale'], painting_dict['dim_image_h'], painting_dict['dim_image_w'], painting_dict['location'])
painting_dict['dim_print_h'], painting_dict['dim_print_w'] = dim_pattern
return painting_dict
def tile_image(self, pattern, dim, scale, dim_image_h, dim_image_w, location, trigger=False):
tile = None
if not trigger:
tile = cv2.resize(pattern, dim, interpolation=cv2.INTER_AREA)
else:
resize_pattern = cv2.resize(pattern, dim, interpolation=cv2.INTER_AREA)
if len(pattern.shape) == 2:
tile = np.tile(resize_pattern, (int((5 + 1) / scale) + 4, int((5 + 1) / scale) + 4))
if len(pattern.shape) == 3:
tile = np.tile(resize_pattern, (int((5 + 1) / scale) + 4, int((5 + 1) / scale) + 4, 1))
tile = self.crop_image(tile, dim_image_h, dim_image_w, location, resize_pattern.shape)
return tile
def get_mask_inv(self, print_):
if print_[0][0][0] == 255 and print_[0][0][1] == 255 and print_[0][0][2] == 255:
bg_color = cv2.cvtColor(print_, cv2.COLOR_BGR2LAB)[0][0]
print_tile = cv2.cvtColor(print_, cv2.COLOR_BGR2LAB)
bg_l, bg_a, bg_b = bg_color[0], bg_color[1], bg_color[2]
bg_L_high, bg_L_low = self.get_low_high_lab(bg_l, L=True)
bg_a_high, bg_a_low = self.get_low_high_lab(bg_a)
bg_b_high, bg_b_low = self.get_low_high_lab(bg_b)
lower = np.array([bg_L_low, bg_a_low, bg_b_low])
upper = np.array([bg_L_high, bg_a_high, bg_b_high])
mask_inv = cv2.inRange(print_tile, lower, upper)
return mask_inv
else:
mask_inv = np.zeros(print_.shape[:2], dtype=np.uint8)
return mask_inv
@staticmethod
def printpaint(result, painting_dict, print_=False):
if print_ and painting_dict['Trigger']:
print_mask = cv2.bitwise_and(result['mask'], cv2.bitwise_not(painting_dict['mask_inv_print']))
img_fg = cv2.bitwise_and(painting_dict['tile_print'], painting_dict['tile_print'], mask=print_mask)
else:
print_mask = result['mask']
img_fg = result['final_image']
if print_ and not painting_dict['Trigger']:
index_ = None
try:
index_ = len(painting_dict['location'])
except:
assert f'there must be parameter of location if choose IfSingle'
for i in range(index_):
start_h, start_w = int(painting_dict['location'][i][1]), int(painting_dict['location'][i][0])
length_h = min(start_h + painting_dict['dim_print_h'], img_fg.shape[0])
length_w = min(start_w + painting_dict['dim_print_w'], img_fg.shape[1])
change_region = img_fg[start_h: length_h, start_w: length_w, :]
# problem in change_mask
change_mask = print_mask[start_h: length_h, start_w: length_w]
# get real part into change mask
_, change_mask = cv2.threshold(change_mask, 220, 255, cv2.THRESH_BINARY)
mask = cv2.bitwise_not(painting_dict['mask_inv_print'])
img_fg[start_h:start_h + painting_dict['dim_print_h'], start_w:start_w + painting_dict['dim_print_w'], :] = change_region
clothes_mask_print = cv2.bitwise_not(print_mask)
img_bg = cv2.bitwise_and(result['pattern_image'], result['pattern_image'], mask=clothes_mask_print)
mask_mo = np.expand_dims(print_mask, axis=2).repeat(3, axis=2)
gray_mo = np.expand_dims(result['gray'], axis=2).repeat(3, axis=2)
img_fg = (img_fg * (mask_mo / 255) * (gray_mo / 255)).astype(np.uint8)
print_image = cv2.add(img_bg, img_fg)
return print_image
def get_print(self, print_dict):
if 'print_scale_list' not in print_dict.keys() or print_dict['print_scale_list'][0][0] < 0.3:
print_dict['scale'] = 0.3
else:
print_dict['scale'] = print_dict['print_scale_list'][0][0]
bucket_name = print_dict['print_path_list'][0].split("/", 1)[0]
object_name = print_dict['print_path_list'][0].split("/", 1)[1]
image = oss_get_image(oss_client=self.minio_client, bucket=bucket_name, object_name=object_name, data_type="PIL")
# 判断图片格式，如果是RGBA 则贴在一张纯白图片上 防止透明转黑
if image.mode == "RGBA":
new_background = Image.new('RGB', image.size, (255, 255, 255))
new_background.paste(image, mask=image.split()[3])
image = new_background
print_dict['image'] = cv2.cvtColor(np.asarray(image), cv2.COLOR_RGB2BGR)
return print_dict
def crop_image(self, image, image_size_h, image_size_w, location, print_shape):
print_w = print_shape[1]
print_h = print_shape[0]
random.seed(self.random_seed)
# 1.拿到偏移量后和resize后的print宽高取余 得到真正偏移量
# 偏移量增加2分之print.w 使坐标位于图中间 如果要位于左上角删除+ print_w // 2 即可
x_offset = print_w - int(location[0][1] % print_w) + print_w // 2
y_offset = print_h - int(location[0][0] % print_h) + print_h // 2
# y_offset = int(location[0][0])
# x_offset = int(location[0][1])
if len(image.shape) == 2:
image = image[x_offset: x_offset + image_size_h, y_offset: y_offset + image_size_w]
elif len(image.shape) == 3:
image = image[x_offset: x_offset + image_size_h, y_offset: y_offset + image_size_w, :]
return image
@staticmethod
def get_low_high_lab(Lab_value, L=False):
if L:
high = Lab_value + 30 if Lab_value + 30 < 255 else 255
low = Lab_value - 30 if Lab_value - 30 > 0 else 0
else:
high = Lab_value + 30 if Lab_value + 30 < 255 else 255
low = Lab_value - 30 if Lab_value - 30 > 0 else 0
return high, low
@staticmethod
def img_rotate(image, angel):
"""顺时针旋转图像任意角度
Args:
image (np.array): [原始图像]
angel (float): [逆时针旋转的角度]
Returns:
[array]: [旋转后的图像]
"""
h, w = image.shape[:2]
center = (w // 2, h // 2)
# if type(angel) is not int:
# angel = 0
M = cv2.getRotationMatrix2D(center, -angel, 1)
# 调整旋转后的图像长宽
rotated_h = int((w * np.abs(M[0, 1]) + (h * np.abs(M[0, 0]))))
rotated_w = int((h * np.abs(M[0, 1]) + (w * np.abs(M[0, 0]))))
M[0, 2] += (rotated_w - w) // 2
M[1, 2] += (rotated_h - h) // 2
# 旋转图像
rotated_img = cv2.warpAffine(image, M, (rotated_w, rotated_h))
return rotated_img, ((rotated_img.shape[1] - image.shape[1]) // 2, (rotated_img.shape[0] - image.shape[0]) // 2)
# return rotated_img, (0, 0)
@staticmethod
def rotate_crop_image(img, angle, crop):
"""
angle: 旋转的角度
crop: 是否需要进行裁剪，布尔向量
"""
if not isinstance(crop, bool):
raise ValueError("The 'crop' parameter must be a boolean.")
crop_image = lambda img, x0, y0, w, h: img[y0:y0 + h, x0:x0 + w]
h, w = img.shape[:2]
# 旋转角度的周期是360°
angle %= 360
# 计算仿射变换矩阵
M_rotation = cv2.getRotationMatrix2D((w / 2, h / 2), angle, 1)
# 得到旋转后的图像
img_rotated = cv2.warpAffine(img, M_rotation, (w, h))
# 如果需要去除黑边
if crop:
# 裁剪角度的等效周期是180°
angle_crop = angle % 180
if angle_crop > 90:
angle_crop = 180 - angle_crop
# 转化角度为弧度
theta = angle_crop * np.pi / 180
# 计算高宽比
hw_ratio = float(h) / float(w)
# 计算裁剪边长系数的分子项
tan_theta = np.tan(theta)
numerator = np.cos(theta) + np.sin(theta) * np.tan(theta)
# 计算分母中和高宽比相关的项
r = hw_ratio if h > w else 1 / hw_ratio
# 计算分母项
denominator = r * tan_theta + 1
# 最终的边长系数
crop_mult = numerator / denominator
# 得到裁剪区域
w_crop = int(crop_mult * w)
h_crop = int(crop_mult * h)
x0 = int((w - w_crop) / 2)
y0 = int((h - h_crop) / 2)
img_rotated = crop_image(img_rotated, x0, y0, w_crop, h_crop)
return img_rotated
def read_image(self, image_url):
image = oss_get_image(oss_client=self.minio_client, bucket=image_url.split("/", 1)[0], object_name=image_url.split("/", 1)[1], data_type="cv2")
if image.shape[2] == 4:
image_rgb = cv2.cvtColor(image, cv2.COLOR_BGRA2RGBA)
image = Image.fromarray(image_rgb)
image_mode = "RGBA"
else:
image_mode = "RGB"
return image, image_mode
@staticmethod
def resize_and_crop(img, target_width, target_height):
# 获取原始图像的尺寸
original_height, original_width = img.shape[:2]
# 计算目标尺寸的宽高比
target_ratio = target_width / target_height
# 计算原始图像的宽高比
original_ratio = original_width / original_height
# 调整尺寸
if original_ratio > target_ratio:
# 原始图像更宽，按高度resize，然后裁剪宽度
new_height = target_height
new_width = int(original_width * (target_height / original_height))
resized_img = cv2.resize(img, (new_width, new_height))
# 裁剪宽度
start_x = (new_width - target_width) // 2
cropped_img = resized_img[:, start_x:start_x + target_width]
else:
# 原始图像更高，按宽度resize，然后裁剪高度
new_width = target_width
new_height = int(original_height * (target_width / original_width))
resized_img = cv2.resize(img, (new_width, new_height))
# 裁剪高度
start_y = (new_height - target_height) // 2
cropped_img = resized_img[start_y:start_y + target_height, :]
return cropped_img

20
app/service/design_fast/pipeline/print_painting.py
View File

@@ -24,7 +24,6 @@ class PrintPainting:
if result['resize_scale'][0] == 1.0 and result['resize_scale'][1] == 1.0: if result['resize_scale'][0] == 1.0 and result['resize_scale'][1] == 1.0:
pass pass
else: else:
# 2025-9-19 印花调整 印花坐标按照sketch的缩放比调整
height, width = result['pattern_image'].shape[:2] height, width = result['pattern_image'].shape[:2]
new_width = int(width * result['resize_scale'][0]) new_width = int(width * result['resize_scale'][0])
new_height = int(height * result['resize_scale'][1]) new_height = int(height * result['resize_scale'][1])
@@ -35,7 +34,6 @@ class PrintPainting:
result['gray'] = cv2.resize(result['gray'], (new_width, new_height)) result['gray'] = cv2.resize(result['gray'], (new_width, new_height))
if overall_print['print_path_list']: if overall_print['print_path_list']:
overall_print['location'][0] = [x * y for x, y in zip(overall_print['location'][0], result['resize_scale'])]
painting_dict = {'dim_image_h': result['pattern_image'].shape[0], 'dim_image_w': result['pattern_image'].shape[1]} painting_dict = {'dim_image_h': result['pattern_image'].shape[0], 'dim_image_w': result['pattern_image'].shape[1]}
result['print_image'] = result['pattern_image'] result['print_image'] = result['pattern_image']
if "print_angle_list" in overall_print.keys() and overall_print['print_angle_list'][0] != 0: if "print_angle_list" in overall_print.keys() and overall_print['print_angle_list'][0] != 0:
@@ -52,9 +50,6 @@ class PrintPainting:
result['single_image'] = result['final_image'] = result['pattern_image'] = result['print_image'] result['single_image'] = result['final_image'] = result['pattern_image'] = result['print_image']
if single_print['print_path_list']: if single_print['print_path_list']:
# 2025-9-19 印花调整 印花坐标按照sketch的缩放比调整
sketch_resize_scale = result['resize_scale']
print_background = np.zeros((result['pattern_image'].shape[0], result['pattern_image'].shape[1], 3), dtype=np.uint8) print_background = np.zeros((result['pattern_image'].shape[0], result['pattern_image'].shape[1], 3), dtype=np.uint8)
mask_background = np.zeros((result['pattern_image'].shape[0], result['pattern_image'].shape[1], 3), dtype=np.uint8) mask_background = np.zeros((result['pattern_image'].shape[0], result['pattern_image'].shape[1], 3), dtype=np.uint8)
for i in range(len(single_print['print_path_list'])): for i in range(len(single_print['print_path_list'])):
@@ -72,8 +67,8 @@ class PrintPainting:
rotated_resized_source_mask = resized_source_mask.rotate(-single_print['print_angle_list'][i]) rotated_resized_source_mask = resized_source_mask.rotate(-single_print['print_angle_list'][i])
source_image_pil = Image.fromarray(cv2.cvtColor(print_background, cv2.COLOR_BGR2RGB)) source_image_pil = Image.fromarray(cv2.cvtColor(print_background, cv2.COLOR_BGR2RGB))
source_image_pil_mask = Image.fromarray(cv2.cvtColor(mask_background, cv2.COLOR_BGR2RGB)) source_image_pil_mask = Image.fromarray(cv2.cvtColor(mask_background, cv2.COLOR_BGR2RGB))
source_image_pil.paste(rotated_resized_source, (int(single_print['location'][i][0] * sketch_resize_scale[0]), int(single_print['location'][i][1] * sketch_resize_scale[1])), rotated_resized_source) source_image_pil.paste(rotated_resized_source, (int(single_print['location'][i][0]), int(single_print['location'][i][1])), rotated_resized_source)
source_image_pil_mask.paste(rotated_resized_source_mask, (int(single_print['location'][i][0] * sketch_resize_scale[0]), int(single_print['location'][i][1] * sketch_resize_scale[1])), rotated_resized_source_mask) source_image_pil_mask.paste(rotated_resized_source_mask, (int(single_print['location'][i][0]), int(single_print['location'][i][1])), rotated_resized_source_mask)
print_background = cv2.cvtColor(np.array(source_image_pil), cv2.COLOR_RGBA2BGR) print_background = cv2.cvtColor(np.array(source_image_pil), cv2.COLOR_RGBA2BGR)
mask_background = cv2.cvtColor(np.array(source_image_pil_mask), cv2.COLOR_RGBA2BGR) mask_background = cv2.cvtColor(np.array(source_image_pil_mask), cv2.COLOR_RGBA2BGR)
ret, mask_background = cv2.threshold(mask_background, 124, 255, cv2.THRESH_BINARY) ret, mask_background = cv2.threshold(mask_background, 124, 255, cv2.THRESH_BINARY)
@@ -151,9 +146,7 @@ class PrintPainting:
img_bg = cv2.bitwise_and(result['pattern_image'], result['pattern_image'], mask=cv2.bitwise_not(print_mask)) img_bg = cv2.bitwise_and(result['pattern_image'], result['pattern_image'], mask=cv2.bitwise_not(print_mask))
mask_mo = np.expand_dims(print_mask, axis=2).repeat(3, axis=2) mask_mo = np.expand_dims(print_mask, axis=2).repeat(3, axis=2)
gray_mo = np.expand_dims(result['gray'], axis=2).repeat(3, axis=2) gray_mo = np.expand_dims(result['gray'], axis=2).repeat(3, axis=2)
img_fg = (img_fg * (mask_mo / 255) * (gray_mo / 255)).astype(np.uint8) # 当sketch 图像为灰色时(非纯白) ， 印花*灰度图像会导致印花在sketch上颜色变暗 img_fg = (img_fg * (mask_mo / 255) * (gray_mo / 255)).astype(np.uint8)
# img_fg = (img_fg * (mask_mo / 255) ).astype(np.uint8) # 不过灰度图像
result['final_image'] = cv2.add(img_bg, img_fg) result['final_image'] = cv2.add(img_bg, img_fg)
canvas = np.full_like(result['final_image'], 255) canvas = np.full_like(result['final_image'], 255)
temp_bg = np.expand_dims(cv2.bitwise_not(result['mask']), axis=2).repeat(3, axis=2) temp_bg = np.expand_dims(cv2.bitwise_not(result['mask']), axis=2).repeat(3, axis=2)
@@ -163,9 +156,6 @@ class PrintPainting:
result['single_image'] = cv2.add(tmp1, tmp2) result['single_image'] = cv2.add(tmp1, tmp2)
if element_print['element_path_list']: if element_print['element_path_list']:
# 2025-9-19 印花调整 印花坐标按照sketch的缩放比调整
sketch_resize_scale = result['resize_scale']
print_background = np.zeros((result['final_image'].shape[0], result['final_image'].shape[1], 3), dtype=np.uint8) print_background = np.zeros((result['final_image'].shape[0], result['final_image'].shape[1], 3), dtype=np.uint8)
mask_background = np.zeros((result['final_image'].shape[0], result['final_image'].shape[1], 3), dtype=np.uint8) mask_background = np.zeros((result['final_image'].shape[0], result['final_image'].shape[1], 3), dtype=np.uint8)
for i in range(len(element_print['element_path_list'])): for i in range(len(element_print['element_path_list'])):
@@ -183,8 +173,8 @@ class PrintPainting:
source_image_pil = Image.fromarray(cv2.cvtColor(print_background, cv2.COLOR_BGR2RGB)) source_image_pil = Image.fromarray(cv2.cvtColor(print_background, cv2.COLOR_BGR2RGB))
source_image_pil_mask = Image.fromarray(cv2.cvtColor(mask_background, cv2.COLOR_BGR2RGB)) source_image_pil_mask = Image.fromarray(cv2.cvtColor(mask_background, cv2.COLOR_BGR2RGB))
source_image_pil.paste(rotated_resized_source, (int(element_print['location'][i][0] * sketch_resize_scale[0]), int(element_print['location'][i][1] * sketch_resize_scale[1])), rotated_resized_source) source_image_pil.paste(rotated_resized_source, (int(element_print['location'][i][0]), int(element_print['location'][i][1])), rotated_resized_source)
source_image_pil_mask.paste(rotated_resized_source_mask, (int(element_print['location'][i][0] * sketch_resize_scale[0]), int(element_print['location'][i][1] * sketch_resize_scale[1])), rotated_resized_source_mask) source_image_pil_mask.paste(rotated_resized_source_mask, (int(element_print['location'][i][0]), int(element_print['location'][i][1])), rotated_resized_source_mask)
print_background = cv2.cvtColor(np.array(source_image_pil), cv2.COLOR_RGBA2BGR) print_background = cv2.cvtColor(np.array(source_image_pil), cv2.COLOR_RGBA2BGR)
mask_background = cv2.cvtColor(np.array(source_image_pil_mask), cv2.COLOR_RGBA2BGR) mask_background = cv2.cvtColor(np.array(source_image_pil_mask), cv2.COLOR_RGBA2BGR)

26
app/service/design_fast/pipeline/split.py
View File

@@ -32,14 +32,14 @@ class Split(object):
new_width = int(width * result['resize_scale'][0]) new_width = int(width * result['resize_scale'][0])
new_height = int(height * result['resize_scale'][1]) new_height = int(height * result['resize_scale'][1])
front_mask = cv2.resize(result['front_mask'], (new_width, new_height), interpolation=cv2.INTER_AREA) front_mask = cv2.resize(result['front_mask'], (new_width, new_height))
back_mask = cv2.resize(result['back_mask'], (new_width, new_height), interpolation=cv2.INTER_AREA) back_mask = cv2.resize(result['back_mask'], (new_width, new_height))
rgba_image = rgb_to_rgba(result['final_image'], front_mask + back_mask) rgba_image = rgb_to_rgba(result['final_image'], front_mask + back_mask)
new_size = (int(rgba_image.shape[1] * result["scale"]), int(rgba_image.shape[0] * result["scale"])) new_size = (int(rgba_image.shape[1] * result["scale"]), int(rgba_image.shape[0] * result["scale"]))
rgba_image = cv2.resize(rgba_image, new_size, interpolation=cv2.INTER_AREA) rgba_image = cv2.resize(rgba_image, new_size)
result_front_image = np.zeros_like(rgba_image) result_front_image = np.zeros_like(rgba_image)
front_mask = cv2.resize(front_mask, new_size, interpolation=cv2.INTER_AREA) front_mask = cv2.resize(front_mask, new_size)
result_front_image[front_mask != 0] = rgba_image[front_mask != 0] result_front_image[front_mask != 0] = rgba_image[front_mask != 0]
result_front_image_pil = Image.fromarray(cvtColor(result_front_image, COLOR_BGR2RGBA)) result_front_image_pil = Image.fromarray(cvtColor(result_front_image, COLOR_BGR2RGBA))
if 'transparent' in result.keys(): if 'transparent' in result.keys():
@@ -48,7 +48,7 @@ class Split(object):
if transparent['mask_url'] is not None and transparent['mask_url'] != "": if transparent['mask_url'] is not None and transparent['mask_url'] != "":
# 预处理用户自选区mask # 预处理用户自选区mask
seg_mask = oss_get_image(oss_client=self.minio_client, bucket=transparent['mask_url'].split('/')[0], object_name=transparent['mask_url'][transparent['mask_url'].find('/') + 1:], data_type="cv2") seg_mask = oss_get_image(oss_client=self.minio_client, bucket=transparent['mask_url'].split('/')[0], object_name=transparent['mask_url'][transparent['mask_url'].find('/') + 1:], data_type="cv2")
seg_mask = cv2.resize(seg_mask, new_size, interpolation=cv2.INTER_AREA) seg_mask = cv2.resize(seg_mask, new_size, interpolation=cv2.INTER_NEAREST)
# 转换颜色空间为 RGB（OpenCV 默认是 BGR） # 转换颜色空间为 RGB（OpenCV 默认是 BGR）
image_rgb = cv2.cvtColor(seg_mask, cv2.COLOR_BGR2RGB) image_rgb = cv2.cvtColor(seg_mask, cv2.COLOR_BGR2RGB)
@@ -67,6 +67,7 @@ class Split(object):
# mask_image = np.zeros((height, width, 3)) # mask_image = np.zeros((height, width, 3))
# mask_image[front_mask != 0] = [0, 0, 255] # mask_image[front_mask != 0] = [0, 0, 255]
# 切换为原始图片尺寸------------------------------- # 切换为原始图片尺寸-------------------------------
height, width = ori_front_mask.shape height, width = ori_front_mask.shape
mask_image = np.zeros((height, width, 3)) mask_image = np.zeros((height, width, 3))
@@ -75,7 +76,7 @@ class Split(object):
# if result["name"] in ('blouse', 'dress', 'outwear', 'tops'): # if result["name"] in ('blouse', 'dress', 'outwear', 'tops'):
# result_back_image = np.zeros_like(rgba_image) # result_back_image = np.zeros_like(rgba_image)
# back_mask = cv2.resize(back_mask, new_size, interpolation=cv2.INTER_AREA) # back_mask = cv2.resize(back_mask, new_size)
# result_back_image[back_mask != 0] = rgba_image[back_mask != 0] # result_back_image[back_mask != 0] = rgba_image[back_mask != 0]
# result_back_image_pil = Image.fromarray(cvtColor(result_back_image, COLOR_BGR2RGBA)) # result_back_image_pil = Image.fromarray(cvtColor(result_back_image, COLOR_BGR2RGBA))
# result['back_image'], result["back_image_url"], _ = upload_png_mask(self.minio_client, result_back_image_pil, f'{generate_uuid()}', mask=None) # result['back_image'], result["back_image_url"], _ = upload_png_mask(self.minio_client, result_back_image_pil, f'{generate_uuid()}', mask=None)
@@ -104,7 +105,7 @@ class Split(object):
# # result['back_mask_image'] = None # # result['back_mask_image'] = None
result_back_image = np.zeros_like(rgba_image) result_back_image = np.zeros_like(rgba_image)
back_mask = cv2.resize(back_mask, new_size, interpolation=cv2.INTER_AREA) back_mask = cv2.resize(back_mask, new_size)
result_back_image[back_mask != 0] = rgba_image[back_mask != 0] result_back_image[back_mask != 0] = rgba_image[back_mask != 0]
result_back_image_pil = Image.fromarray(cvtColor(result_back_image, COLOR_BGR2RGBA)) result_back_image_pil = Image.fromarray(cvtColor(result_back_image, COLOR_BGR2RGBA))
result['back_image'], result["back_image_url"], _ = upload_png_mask(self.minio_client, result_back_image_pil, f'{generate_uuid()}', mask=None) result['back_image'], result["back_image_url"], _ = upload_png_mask(self.minio_client, result_back_image_pil, f'{generate_uuid()}', mask=None)
@@ -112,6 +113,7 @@ class Split(object):
# mask_image[back_mask != 0] = [0, 255, 0] # mask_image[back_mask != 0] = [0, 255, 0]
mask_image[ori_back_mask != 0] = [0, 255, 0] mask_image[ori_back_mask != 0] = [0, 255, 0]
rbga_mask = rgb_to_rgba(mask_image, ori_front_mask + ori_back_mask) rbga_mask = rgb_to_rgba(mask_image, ori_front_mask + ori_back_mask)
mask_pil = Image.fromarray(cvtColor(rbga_mask.astype(np.uint8), COLOR_BGR2RGBA)) mask_pil = Image.fromarray(cvtColor(rbga_mask.astype(np.uint8), COLOR_BGR2RGBA))
image_data = io.BytesIO() image_data = io.BytesIO()
@@ -121,12 +123,10 @@ class Split(object):
req = oss_upload_image(oss_client=self.minio_client, bucket=AIDA_CLOTHING, object_name=f"mask/mask_{generate_uuid()}.png", image_bytes=image_bytes) req = oss_upload_image(oss_client=self.minio_client, bucket=AIDA_CLOTHING, object_name=f"mask/mask_{generate_uuid()}.png", image_bytes=image_bytes)
result['mask_url'] = req.bucket_name + "/" + req.object_name result['mask_url'] = req.bucket_name + "/" + req.object_name
# 创建中间图层(未分割图层) 1.color + overall_print 2.color + overall_print + print # 创建中间图层
result_pattern_overall_image_pil = Image.fromarray(cvtColor(rgb_to_rgba(result['no_seg_sketch_overall'], ori_front_mask + ori_back_mask), COLOR_BGR2RGBA)) result_pattern_image_rgba = rgb_to_rgba(result['pattern_image'], result['mask'])
result['pattern_overall_image'], result['pattern_overall_image_url'], _ = upload_png_mask(self.minio_client, result_pattern_overall_image_pil, f'{generate_uuid()}') result_pattern_image_pil = Image.fromarray(cvtColor(result_pattern_image_rgba, COLOR_BGR2RGBA))
result['pattern_image'], result['pattern_image_url'], _ = upload_png_mask(self.minio_client, result_pattern_image_pil, f'{generate_uuid()}')
result_pattern_print_image_pil = Image.fromarray(cvtColor(rgb_to_rgba(result['no_seg_sketch_print'], ori_front_mask + ori_back_mask), COLOR_BGR2RGBA))
result['pattern_print_image'], result['pattern_print_image_url'], _ = upload_png_mask(self.minio_client, result_pattern_print_image_pil, f'{generate_uuid()}')
return result return result
except Exception as e: except Exception as e:
logging.warning(f"split runtime exception : {e} image_id : {result['image_id']}") logging.warning(f"split runtime exception : {e} image_id : {result['image_id']}")

23
app/service/design_fast/utils/organize.py
View File

@@ -32,9 +32,7 @@ def organize_clothing(layer):
resize_scale=layer["resize_scale"], resize_scale=layer["resize_scale"],
mask=cv2.resize(layer['mask'], layer["front_image"].size), mask=cv2.resize(layer['mask'], layer["front_image"].size),
gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "", gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "",
pattern_overall_image_url=layer['pattern_overall_image_url'], pattern_image_url=layer['pattern_image_url'],
pattern_print_image_url=layer['pattern_print_image_url'],
pattern_image=layer['pattern_image'], pattern_image=layer['pattern_image'],
# back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else "" # back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else ""
) )
@@ -51,8 +49,7 @@ def organize_clothing(layer):
resize_scale=layer["resize_scale"], resize_scale=layer["resize_scale"],
mask=cv2.resize(layer['mask'], layer["front_image"].size), mask=cv2.resize(layer['mask'], layer["front_image"].size),
gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "", gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "",
pattern_overall_image_url=layer['pattern_overall_image_url'], pattern_image_url=layer['pattern_image_url'],
pattern_print_image_url=layer['pattern_print_image_url'],
# back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else "" # back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else ""
) )
return front_layer, back_layer return front_layer, back_layer
@@ -60,15 +57,7 @@ def organize_clothing(layer):
def organize_accessories(layer): def organize_accessories(layer):
# 起始坐标 # 起始坐标
start_point = (0, 0) start_point = calculate_start_point(layer['keypoint'], layer['scale'], {'accessories_left': [0, 0]}, {'accessories_left': [0, 0]}, layer["offset"], layer["resize_scale"])
layer['clothes_keypoint'] = {
'accessories_left': [0, 0]
}
layer['body_point_test'] = {
'accessories_left': [0, 0]
}
start_point = calculate_start_point(layer['keypoint'], layer['scale'], layer['clothes_keypoint'], layer['body_point_test'], layer["offset"], layer["resize_scale"])
# 前片数据 # 前片数据
front_layer = dict(priority=layer['priority'] if layer.get("layer_order", False) else PRIORITY_DICT.get(f'{layer["name"].lower()}_front', None), front_layer = dict(priority=layer['priority'] if layer.get("layer_order", False) else PRIORITY_DICT.get(f'{layer["name"].lower()}_front', None),
@@ -83,8 +72,7 @@ def organize_accessories(layer):
resize_scale=layer["resize_scale"], resize_scale=layer["resize_scale"],
mask=cv2.resize(layer['mask'], layer["front_image"].size), mask=cv2.resize(layer['mask'], layer["front_image"].size),
gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "", gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "",
pattern_overall_image_url=layer['pattern_overall_image_url'], pattern_image_url=layer['pattern_image_url'],
pattern_print_image_url=layer['pattern_print_image_url'],
pattern_image=layer['pattern_image'], pattern_image=layer['pattern_image'],
# back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else "" # back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else ""
) )
@@ -101,8 +89,7 @@ def organize_accessories(layer):
resize_scale=layer["resize_scale"], resize_scale=layer["resize_scale"],
mask=cv2.resize(layer['mask'], layer["front_image"].size), mask=cv2.resize(layer['mask'], layer["front_image"].size),
gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "", gradient_string=layer['gradient_string'] if 'gradient_string' in layer.keys() else "",
pattern_overall_image_url=layer['pattern_overall_image_url'], pattern_image_url=layer['pattern_image_url'],
pattern_print_image_url=layer['pattern_print_image_url'],
# back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else "" # back_perspective_url=layer['back_perspective_url'] if 'back_perspective_url' in layer.keys() else ""
) )
return front_layer, back_layer return front_layer, back_layer

4
app/service/design_fast/utils/synthesis_item.py
View File

@@ -207,9 +207,7 @@ def update_base_size_priority(layers, size):
if info['name'] == 'mannequin': if info['name'] == 'mannequin':
new_height = info['image'].height new_height = info['image'].height
max_x = max(x_list) max_x = max(x_list)
new_width = max_x - min_x * 2
# x坐标中最小偏移量的绝对值 + 最大偏移量
new_width = max_x + abs(min_x)
# 更新坐标 # 更新坐标
for info in layers: for info in layers:
info['adaptive_position'] = (info['position'][0], info['position'][1] - min_x) info['adaptive_position'] = (info['position'][0], info['position'][1] - min_x)