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

feat 更新响应模板

Browse Source 
fix
This commit is contained in:
zhouchengrong
2024-06-17 13:10:46 +08:00
parent 0a868d6817
commit f2bb7a11f9
4 changed files with 112 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
app/service/design/service.py
View File

@@ -5,6 +5,8 @@ from app.service.design.utils.redis_utils import Redis
from app.service.design.utils.synthesis_item import synthesis, synthesis_single from app.service.design.utils.synthesis_item import synthesis, synthesis_single
import concurrent.futures import concurrent.futures
from app.service.utils.decorator import RunTime
def process_item(item, layers): def process_item(item, layers):
# logging.info("process running.........") # logging.info("process running.........")
@@ -38,6 +40,7 @@ def final_progress(process_id):
return progress return progress
@RunTime
def generate(request_data): def generate(request_data):
return_response = {} return_response = {}
request_data = request_data.dict() request_data = request_data.dict()

2
app/service/design/utils/synthesis_item.py
View File

@@ -75,7 +75,7 @@ def positioning(all_mask_shape, mask_shape, offset):
return all_start, all_end, mask_start, mask_end return all_start, all_end, mask_start, mask_end
@RunTime # @RunTime
def synthesis(data, size): def synthesis(data, size):
# 创建底图 # 创建底图
base_image = Image.new('RGBA', size, (0, 0, 0, 0)) base_image = Image.new('RGBA', size, (0, 0, 0, 0))

189
app/service/design/utils/upload_image.py
View File

@@ -28,128 +28,129 @@ minio_client = Minio(
# s3 = boto3.client('s3', aws_access_key_id=S3_ACCESS_KEY, aws_secret_access_key=S3_AWS_SECRET_ACCESS_KEY, region_name=S3_REGION_NAME) # s3 = boto3.client('s3', aws_access_key_id=S3_ACCESS_KEY, aws_secret_access_key=S3_AWS_SECRET_ACCESS_KEY, region_name=S3_REGION_NAME)
#
@RunTime
def upload_png_mask(front_image, object_name, mask=None):
mask_url = None
if mask is not None:
# 反转掩模
mask_inverted = cv2.bitwise_not(mask)
# 将掩模转换为 RGBA 格式
rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
# 将图像数据保存到内存中的 BytesIO 对象中
image_bytes = io.BytesIO()
image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
image_bytes.seek(0)
try:
key = f"mask/mask_{object_name}.png"
mask_url = f"{AIDA_CLOTHING}/{key}"
s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
except Exception as e:
print(f'上传到 S3 失败: {e}')
with io.BytesIO() as output:
front_image.save(output, format='PNG')
data = output.getvalue()
# 创建一个 S3 客户端
try:
key = f"image/image_{object_name}.png"
image_url = f"{AIDA_CLOTHING}/{key}"
s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
return front_image, image_url, mask_url
except Exception as e:
print(f'上传到 S3 失败: {e}')
@RunTime
def upload_layer_image(image, object_name):
with io.BytesIO() as output:
image.save(output, format='PNG')
data = output.getvalue()
# 创建一个 S3 客户端
try:
key = f"image/image_{object_name}.png"
image_url = f"{AIDA_CLOTHING}/{key}"
s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
return image_url
except Exception as e:
print(f'上传到 S3 失败: {e}')
@RunTime
def upload_mask_image(mask, object_name):
# 反转掩模
mask_inverted = cv2.bitwise_not(mask)
# 将掩模转换为 RGBA 格式
rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
# 将图像数据保存到内存中的 BytesIO 对象中
image_bytes = io.BytesIO()
image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
image_bytes.seek(0)
try:
key = f"mask/mask_{object_name}.png"
mask_url = f"{AIDA_CLOTHING}/{key}"
s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
return mask_url
except Exception as e:
print(f'上传到 S3 失败: {e}')
"""minio 上传"""
# @RunTime # @RunTime
# def upload_png_mask(front_image, object_name, mask=None): # def upload_png_mask(front_image, object_name, mask=None):
# start_time = time.time()
# try:
# mask_url = None # mask_url = None
# if mask is not None: # if mask is not None:
# # 反转掩模
# mask_inverted = cv2.bitwise_not(mask) # mask_inverted = cv2.bitwise_not(mask)
# # 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明 # # 将掩模转换为 RGBA 格式
# rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA) # rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
# rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0] # rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
# # 将图像数据保存到内存中的 BytesIO 对象中
# image_bytes = io.BytesIO() # image_bytes = io.BytesIO()
# image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes()) # image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
#
# image_bytes.seek(0) # image_bytes.seek(0)
# mask_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'mask/mask_{object_name}.png', image_bytes, len(image_bytes.getvalue()), content_type='image/png').object_name}" # try:
# # key = f"mask/mask_{object_name}.png"
# image_data = io.BytesIO() # mask_url = f"{AIDA_CLOTHING}/{key}"
# front_image.save(image_data, format='PNG') # s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
# image_data.seek(0) # except Exception as e:
# image_bytes = image_data.read() # print(f'上传到 S3 失败: {e}')
# image_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'image/image_{object_name}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}" # with io.BytesIO() as output:
# # print(f"upload_png_mask {object_name} = {time.time() - start_time}") # front_image.save(output, format='PNG')
# data = output.getvalue()
# # 创建一个 S3 客户端
# try:
# key = f"image/image_{object_name}.png"
# image_url = f"{AIDA_CLOTHING}/{key}"
# s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
# return front_image, image_url, mask_url # return front_image, image_url, mask_url
# except Exception as e: # except Exception as e:
# logging.warning(f"upload_png_mask runtime exception : {e}") # print(f'上传到 S3 失败: {e}')
# #
# #
# @RunTime # @RunTime
# def upload_layer_image(image, object_name): # def upload_layer_image(image, object_name):
# with io.BytesIO() as output:
# image.save(output, format='PNG')
# data = output.getvalue()
# # 创建一个 S3 客户端
# try: # try:
# image_data = io.BytesIO() # key = f"image/image_{object_name}.png"
# image.save(image_data, format='PNG') # image_url = f"{AIDA_CLOTHING}/{key}"
# image_data.seek(0) # s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
# image_bytes = image_data.read()
# image_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'image/image_{object_name}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
# return image_url # return image_url
# except Exception as e: # except Exception as e:
# logging.warning(f"upload_png_mask runtime exception : {e}") # print(f'上传到 S3 失败: {e}')
# #
# #
# @RunTime # @RunTime
# def upload_mask_image(mask, object_name): # def upload_mask_image(mask, object_name):
# try: # # 反转掩模
# mask_inverted = cv2.bitwise_not(mask) # mask_inverted = cv2.bitwise_not(mask)
# # 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明 # # 将掩模转换为 RGBA 格式
# rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA) # rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
# rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0] # rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
# # 将图像数据保存到内存中的 BytesIO 对象中
# image_bytes = io.BytesIO() # image_bytes = io.BytesIO()
# image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes()) # image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
#
# image_bytes.seek(0) # image_bytes.seek(0)
# mask_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'mask/mask_{object_name}.png', image_bytes, len(image_bytes.getvalue()), content_type='image/png').object_name}" # try:
# key = f"mask/mask_{object_name}.png"
# mask_url = f"{AIDA_CLOTHING}/{key}"
# s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
# return mask_url # return mask_url
# except Exception as e: # except Exception as e:
# logging.warning(f"upload_png_mask runtime exception : {e}") # print(f'上传到 S3 失败: {e}')
"""minio 上传"""
# @RunTime
def upload_png_mask(front_image, object_name, mask=None):
start_time = time.time()
try:
mask_url = None
if mask is not None:
mask_inverted = cv2.bitwise_not(mask)
# 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明
rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
image_bytes = io.BytesIO()
image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
image_bytes.seek(0)
mask_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'mask/mask_{object_name}.png', image_bytes, len(image_bytes.getvalue()), content_type='image/png').object_name}"
image_data = io.BytesIO()
front_image.save(image_data, format='PNG')
image_data.seek(0)
image_bytes = image_data.read()
image_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'image/image_{object_name}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
# print(f"upload_png_mask {object_name} = {time.time() - start_time}")
return front_image, image_url, mask_url
except Exception as e:
logging.warning(f"upload_png_mask runtime exception : {e}")
@RunTime
def upload_layer_image(image, object_name):
try:
image_data = io.BytesIO()
image.save(image_data, format='PNG')
image_data.seek(0)
image_bytes = image_data.read()
image_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'image/image_{object_name}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
return image_url
except Exception as e:
logging.warning(f"upload_png_mask runtime exception : {e}")
@RunTime
def upload_mask_image(mask, object_name):
try:
mask_inverted = cv2.bitwise_not(mask)
# 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明
rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
image_bytes = io.BytesIO()
image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
image_bytes.seek(0)
mask_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'mask/mask_{object_name}.png', image_bytes, len(image_bytes.getvalue()), content_type='image/png').object_name}"
return mask_url
except Exception as e:
logging.warning(f"upload_png_mask runtime exception : {e}")

2
logging_env.py
View File

@@ -9,7 +9,7 @@ LOGGER_CONFIG_DICT = {
"handlers": { "handlers": {
"console": { "console": {
"class": "logging.StreamHandler", "class": "logging.StreamHandler",
"level": "DEBUG", "level": "INFO",
"formatter": "simple", "formatter": "simple",
"stream": "ext://sys.stdout", "stream": "ext://sys.stdout",
}, },