design batch 代码整理

app/service/design_batch/pipeline/__init__.py

@@ -0,0 +1,20 @@
+from .color import Color
+from .contour_detection import ContourDetection
+from .keypoint import KeyPoint
+from .keypoint import KeyPoint
+from .loading import LoadImage, LoadBodyImage
+from .print_painting import PrintPainting
+from .scale import Scaling
+from .segmentation import Segmentation
+from .split import Split
+
+__all__ = [
+    'LoadBodyImage', 'LoadImage',
+    'KeyPoint',
+    'ContourDetection',
+    'Segmentation',
+    'Color',
+    'PrintPainting',
+    'Scaling',
+    'Split'
+]

app/service/design_batch/pipeline/color.py

@@ -0,0 +1,62 @@
+import logging
+
+import cv2
+import numpy as np
+
+from app.service.utils.new_oss_client import oss_get_image
+
+logger = logging.getLogger()
+
+
+class Color:
+    def __init__(self, minio_client):
+        self.minio_client = minio_client
+
+    def __call__(self, result):
+        dim_image_h, dim_image_w = result['image'].shape[0:2]
+        if "gradient" in result.keys() and result['gradient'] != "":
+            bucket_name = result['gradient'].split('/')[0]
+            object_name = result['gradient'][result['gradient'].find('/') + 1:]
+            pattern = self.get_gradient(bucket_name=bucket_name, object_name=object_name)
+            resize_pattern = cv2.resize(pattern, (dim_image_w, dim_image_h), interpolation=cv2.INTER_AREA)
+        else:
+            pattern = self.get_pattern(result['color'])
+            resize_pattern = cv2.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)
+        gray_mo = np.expand_dims(result['gray'], axis=2).repeat(3, axis=2)
+        get_image_fir = resize_pattern * (closed_mo / 255) * (gray_mo / 255)
+        result['pattern_image'] = get_image_fir.astype(np.uint8)
+        result['final_image'] = result['pattern_image']
+        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['single_image'] = cv2.add(tmp1, tmp2)
+        result['alpha'] = 100 / 255.0
+        return result
+
+    def get_gradient(self, bucket_name, object_name):
+        # 获取渐变色图案
+        image = oss_get_image(oss_client=self.minio_client, bucket=bucket_name, object_name=object_name, data_type="cv2")
+        if image.shape[2] == 4:
+            image = cv2.cvtColor(image, cv2.COLOR_BGRA2BGR)
+        return image
+
+    @staticmethod
+    def crop_image(image, image_size_h, image_size_w):
+        x_offset = np.random.randint(low=0, high=int(image_size_h / 5) - 6)
+        y_offset = np.random.randint(low=0, high=int(image_size_w / 5) - 6)
+        image = image[x_offset: x_offset + image_size_h, y_offset: y_offset + image_size_w, :]
+        return image
+
+    @staticmethod
+    def get_pattern(single_color):
+        if single_color is None:
+            raise False
+        R, G, B = single_color.split(' ')
+        pattern = np.zeros([1, 1, 3], np.uint8)
+        pattern[0, 0, 0] = int(B)
+        pattern[0, 0, 1] = int(G)
+        pattern[0, 0, 2] = int(R)
+        return pattern

app/service/design_batch/pipeline/contour_detection.py

@@ -0,0 +1,37 @@
+import cv2
+import numpy as np
+
+
+class ContourDetection:
+    def __call__(self, result):
+        Contour = self.get_contours(result['image'])
+        Mask = np.zeros(result['image'].shape[:2], np.uint8)
+        if len(Contour):
+            Max_contour = Contour[0]
+            Epsilon = 0.001 * cv2.arcLength(Max_contour, True)
+            Approx = cv2.approxPolyDP(Max_contour, Epsilon, True)
+            cv2.drawContours(Mask, [Approx], -1, 255, -1)
+        else:
+            Mask = np.ones(result['image'].shape[:2], np.uint8) * 255
+        # TODO 修复部分图片出现透明的情况 下版本上线
+        # img2gray = cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY)
+        # ret, Mask = cv2.threshold(img2gray, 126, 255, cv2.THRESH_BINARY)
+        # Mask = cv2.bitwise_not(Mask)
+        if result['pre_mask'] is None:
+            result['mask'] = Mask
+        else:
+            result['mask'] = cv2.bitwise_and(Mask, result['pre_mask'])
+        result['front_mask'] = result['mask']
+        result['back_mask'] = result['mask']
+        return result
+
+    @staticmethod
+    def get_contours(image):
+        gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
+        Edge = cv2.Canny(gray, 10, 150)
+        kernel = np.ones((5, 5), np.uint8)
+        Edge = cv2.dilate(Edge, kernel=kernel, iterations=1)
+        Edge = cv2.erode(Edge, kernel=kernel, iterations=1)
+        Contour, _ = cv2.findContours(Edge, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+        Contour = sorted(Contour, key=cv2.contourArea, reverse=True)
+        return Contour

app/service/design_batch/pipeline/keypoint.py

@@ -0,0 +1,114 @@
+import logging
+
+import numpy as np
+from pymilvus import MilvusClient
+
+from app.core.config import *
+from app.service.design_batch.utils.design_ensemble import get_keypoint_result
+
+logger = logging.getLogger(__name__)
+
+
+class KeyPoint:
+    name = "KeyPoint"
+
+    @classmethod
+    def get_name(cls):
+        return cls.name
+
+    def __call__(self, result):
+        if result['name'] in ['blouse', 'skirt', 'dress', 'outwear', 'trousers', 'tops', 'bottoms']:  # 查询是否有数据 且类别相同 相同则直接读 不同则推理后更新
+            # result['clothes_keypoint'] = self.infer_keypoint_result(result)
+            site = 'up' if result['name'] in ['blouse', 'outwear', 'dress', 'tops'] else 'down'
+            # keypoint_cache = search_keypoint_cache(result["image_id"], site)
+            keypoint_cache = self.keypoint_cache(result, site)
+            # 取消向量查询 直接过模型推理
+            # keypoint_cache = False
+            if keypoint_cache is False:
+                keypoint_infer_result, site = self.infer_keypoint_result(result)
+                result['clothes_keypoint'] = self.save_keypoint_cache(result["image_id"], keypoint_infer_result, site)
+            else:
+                result['clothes_keypoint'] = keypoint_cache
+        return result
+
+    @staticmethod
+    def infer_keypoint_result(result):
+        site = 'up' if result['name'] in ['blouse', 'outwear', 'dress', 'tops'] else 'down'
+        keypoint_infer_result = get_keypoint_result(result["image"], site)  # 推理结果
+        return keypoint_infer_result, site
+
+    @staticmethod
+    def save_keypoint_cache(keypoint_id, cache, site):
+        if site == "down":
+            zeros = np.zeros(20, dtype=int)
+            result = np.concatenate([zeros, cache.flatten()])
+        else:
+            zeros = np.zeros(4, dtype=int)
+            result = np.concatenate([cache.flatten(), zeros])
+        # 取消向量保存 直接拿结果
+        data = [
+            {"keypoint_id": keypoint_id,
+             "keypoint_site": site,
+             "keypoint_vector": result.tolist()
+             }
+        ]
+        try:
+            client = MilvusClient(uri=MILVUS_URL, token=MILVUS_TOKEN, db_name=MILVUS_ALIAS)
+            res = client.upsert(collection_name=MILVUS_TABLE_KEYPOINT, data=data)
+            client.close()
+            return dict(zip(KEYPOINT_RESULT_TABLE_FIELD_SET, result.reshape(12, 2).astype(int).tolist()))
+        except Exception as e:
+            logger.info(f"save keypoint cache milvus error : {e}")
+            return dict(zip(KEYPOINT_RESULT_TABLE_FIELD_SET, result.reshape(12, 2).astype(int).tolist()))
+
+    @staticmethod
+    def update_keypoint_cache(keypoint_id, infer_result, search_result, site):
+        if site == "up":
+            # 需要的是up 即推理出来的是up 那么查询的就是down
+            result = np.concatenate([infer_result.flatten(), search_result[-4:]])
+        else:
+            # 需要的是down 即推理出来的是down 那么查询的就是up
+            result = np.concatenate([search_result[:20], infer_result.flatten()])
+        data = [
+            {"keypoint_id": keypoint_id,
+             "keypoint_site": "all",
+             "keypoint_vector": result.tolist()
+             }
+        ]
+
+        try:
+            client = MilvusClient(uri=MILVUS_URL, token=MILVUS_TOKEN, db_name=MILVUS_ALIAS)
+            client.upsert(
+                collection_name=MILVUS_TABLE_KEYPOINT,
+                data=data
+            )
+            return dict(zip(KEYPOINT_RESULT_TABLE_FIELD_SET, result.reshape(12, 2).astype(int).tolist()))
+        except Exception as e:
+            logger.info(f"save keypoint cache milvus error : {e}")
+            return dict(zip(KEYPOINT_RESULT_TABLE_FIELD_SET, result.reshape(12, 2).astype(int).tolist()))
+
+    # @ RunTime
+    def keypoint_cache(self, result, site):
+        try:
+            client = MilvusClient(uri=MILVUS_URL, token=MILVUS_TOKEN, db_name=MILVUS_ALIAS)
+            keypoint_id = result['image_id']
+            res = client.query(
+                collection_name=MILVUS_TABLE_KEYPOINT,
+                # ids=[keypoint_id],
+                filter=f"keypoint_id == {keypoint_id}",
+                output_fields=['keypoint_vector', 'keypoint_site']
+            )
+            if len(res) == 0:
+                # 没有结果 直接推理拿结果 并保存
+                keypoint_infer_result, site = self.infer_keypoint_result(result)
+                return self.save_keypoint_cache(result['image_id'], keypoint_infer_result, site)
+            elif res[0]["keypoint_site"] == "all" or res[0]["keypoint_site"] == site:
+                # 需要的类型和查询的类型一致，或者查询的类型为all 则直接返回查询的结果
+                return dict(zip(KEYPOINT_RESULT_TABLE_FIELD_SET, np.array(res[0]['keypoint_vector']).astype(int).reshape(12, 2).tolist()))
+            elif res[0]["keypoint_site"] != site:
+                # 需要的类型和查询到的不一致，则更新类型为all
+                keypoint_infer_result, site = self.infer_keypoint_result(result)
+                return self.update_keypoint_cache(result["image_id"], keypoint_infer_result, res[0]['keypoint_vector'], site)
+        except Exception as e:
+            logger.info(f"search keypoint cache milvus error {e}")
+            return False

app/service/design_batch/pipeline/loading.py

@@ -0,0 +1,80 @@
+import io
+import logging
+
+import cv2
+import numpy as np
+from PIL import Image
+
+from app.service.utils.new_oss_client import oss_get_image
+
+logger = logging.getLogger()
+
+
+class LoadBodyImage:
+    name = "LoadBodyImage"
+
+    def __init__(self, minio_client):
+        self.minio_client = minio_client
+
+    @classmethod
+    def get_name(cls):
+        return cls.name
+
+    def __call__(self, result):
+        result["name"] = "mannequin"
+        result['body_image'] = oss_get_image(oss_client=self.minio_client, bucket=result['body_path'].split("/", 1)[0], object_name=result['body_path'].split("/", 1)[1], data_type="PIL")
+        return result
+
+
+class LoadImage:
+    name = "LoadImage"
+
+    def __init__(self, minio_client):
+        self.minio_client = minio_client
+
+    @classmethod
+    def get_name(cls):
+        return cls.name
+
+    def __call__(self, result):
+        result['image'], result['pre_mask'] = self.read_image(result['path'])
+        result['gray'] = cv2.cvtColor(result['image'], cv2.COLOR_BGR2GRAY)
+        result['keypoint'] = self.get_keypoint(result['name'])
+        result['img_shape'] = result['image'].shape
+        result['ori_shape'] = result['image'].shape
+        return result
+
+    def read_image(self, image_path):
+        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")
+        if len(image.shape) == 2:
+            image = cv2.cvtColor(image, cv2.COLOR_GRAY2RGB)
+        if image.shape[2] == 4:  # 如果是四通道 mask
+            image_mask = image[:, :, 3]
+            image = image[:, :, :3]
+
+        if image.shape[:2] <= (50, 50):
+            # 计算新尺寸
+            new_size = (image.shape[1] * 2, image.shape[0] * 2)
+            # 调整大小
+            image = cv2.resize(image, new_size, interpolation=cv2.INTER_LINEAR)
+        return image, image_mask
+
+    @staticmethod
+    def get_keypoint(name):
+        if name == 'blouse' or name == 'outwear' or name == 'dress' or name == 'tops':
+            keypoint = 'shoulder'
+        elif name == 'trousers' or name == 'skirt' or name == 'bottoms':
+            keypoint = 'waistband'
+        elif name == 'bag':
+            keypoint = 'hand_point'
+        elif name == 'shoes':
+            keypoint = 'toe'
+        elif name == 'hairstyle':
+            keypoint = 'head_point'
+        elif name == 'earring':
+            keypoint = 'ear_point'
+        else:
+            raise KeyError(f"{name} does not belong to item category list: blouse, outwear, dress, trousers, skirt, "
+                           f"bag, shoes, hairstyle, earring.")
+        return keypoint

app/service/design_batch/pipeline/print_painting.py

@@ -0,0 +1,524 @@
+import random
+
+import cv2
+import numpy as np
+from PIL import Image
+
+from app.service.utils.new_oss_client import oss_get_image
+
+
+class PrintPainting:
+    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]}
+            result['print_image'] = result['pattern_image']
+            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['print_image'] = self.printpaint(result, painting_dict, print_=True)
+            result['single_image'] = result['final_image'] = result['pattern_image'] = result['print_image']
+
+        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 == "RGBA":
+                    new_size = (int(image.width * single_print['print_scale_list'][i]), int(image.height * single_print['print_scale_list'][i]))
+
+                    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)
+                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)
+                    # 旋转后的坐标需要重新算
+                    rotate_mask, _ = self.img_rotate(mask, single_print['print_angle_list'][i], single_print['print_scale_list'][i])
+                    rotate_image, rotated_new_size = self.img_rotate(image, single_print['print_angle_list'][i], single_print['print_scale_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(single_print['location'][i][0] - rotated_new_size[0]), int(single_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]
+
+                    #  有bug
+                    # if x + print_x > image_x:
+                    #     rotate_image = rotate_image[:, :x + print_x - image_x]
+                    #     rotate_mask = rotate_mask[:, :x + print_x - image_x]
+                    # #
+                    # if y + print_y > image_y:
+                    #     rotate_image = rotate_image[:y + print_y - image_y]
+                    #     rotate_mask = rotate_mask[:y + print_y - image_y]
+
+                    #  不能是并行
+                    # 当前第一轮的if （108以及115）是判断有没有过下界和右界。第二轮的是判断左上有没有超出。 如果这个样子的话，先裁了右边，再左移，region就会有问题
+                    # 先挪 再判断 最后裁剪
+
+                    # 如果print旋转了 或者 print贴边了 则需要判断 判断左界和上界是否小于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
+
+                    # ------------------
+                    # 如果print-size大于image-size 则需要裁剪print
+
+                    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[start_y:y + rotate_mask.shape[0], start_x:x + rotate_mask.shape[1]] = cv2.bitwise_xor(mask_background[start_y:y + rotate_mask.shape[0], start_x:x + rotate_mask.shape[1]], rotate_mask)
+                    # print_background[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]] = cv2.add(print_background[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]], rotate_image)
+
+                    # mask_background[start_y:y + rotate_mask.shape[0], start_x:x + rotate_mask.shape[1]] = rotate_mask
+                    # print_background[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]] = rotate_image
+                    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)
+
+                # gray_image = cv2.cvtColor(mask_background, cv2.COLOR_BGR2GRAY)
+                # print_background = cv2.bitwise_and(print_background, print_background, mask=gray_image)
+
+                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)
+                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['single_image'] = cv2.add(tmp1, tmp2)
+
+        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(image.width * element_print['element_scale_list'][i]), int(image.height * element_print['element_scale_list'][i]))
+
+                    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)
+                    # 旋转后的坐标需要重新算
+                    rotate_mask, _ = self.img_rotate(mask, element_print['element_angle_list'][i], element_print['element_scale_list'][i])
+                    rotate_image, rotated_new_size = self.img_rotate(image, element_print['element_angle_list'][i], element_print['element_scale_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]
+
+                    #  有bug
+                    # if x + print_x > image_x:
+                    #     rotate_image = rotate_image[:, :x + print_x - image_x]
+                    #     rotate_mask = rotate_mask[:, :x + print_x - image_x]
+                    # #
+                    # if y + print_y > image_y:
+                    #     rotate_image = rotate_image[:y + print_y - image_y]
+                    #     rotate_mask = rotate_mask[:y + print_y - image_y]
+
+                    #  不能是并行
+                    # 当前第一轮的if （108以及115）是判断有没有过下界和右界。第二轮的是判断左上有没有超出。 如果这个样子的话，先裁了右边，再左移，region就会有问题
+                    # 先挪 再判断 最后裁剪
+
+                    # 如果print旋转了 或者 print贴边了 则需要判断 判断左界和上界是否小于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
+
+                    # ------------------
+                    # 如果print-size大于image-size 则需要裁剪print
+
+                    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[start_y:y + rotate_mask.shape[0], start_x:x + rotate_mask.shape[1]] = cv2.bitwise_xor(mask_background[start_y:y + rotate_mask.shape[0], start_x:x + rotate_mask.shape[1]], rotate_mask)
+                    # print_background[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]] = cv2.add(print_background[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]], rotate_image)
+
+                    # mask_background[start_y:y + rotate_mask.shape[0], start_x:x + rotate_mask.shape[1]] = rotate_mask
+                    # print_background[start_y:y + rotate_image.shape[0], start_x:x + rotate_image.shape[1]] = rotate_image
+                    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)
+
+            # gray_image = cv2.cvtColor(mask_background, cv2.COLOR_BGR2GRAY)
+            # print_background = cv2.bitwise_and(print_background, print_background, mask=gray_image)
+
+            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)
+            # TODO element  丢失信息
+            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)
+            # 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)
+            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['single_image'] = 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:
+            # 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])
+
+            # print_tile = cv2.cvtColor(print_, cv2.COLOR_BGR2LAB)
+            # mask_inv = cv2.cvtColor(print_tile, cv2.COLOR_BGR2GRAY)
+
+            # mask_inv = cv2.cvtColor(print_, cv2.COLOR_BGR2GRAY)
+            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.3:
+            print_dict['scale'] = 0.3
+        else:
+            print_dict['scale'] = print_dict['print_scale_list'][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)
+        # logging.info(f'overall print location : {location}')
+        # x_offset = random.randint(0, image.shape[0] - image_size_h)
+        # y_offset = random.randint(0, image.shape[1] - image_size_w)
+
+        # 1.拿到偏移量后和resize后的print宽高取余 得到真正偏移量
+        x_offset = print_w - int(location[0][1] % print_w)
+        y_offset = print_w - int(location[0][0] % print_h)
+
+        # 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, scale):
+        """顺时针旋转图像任意角度
+
+        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, scale)
+        # 调整旋转后的图像长宽
+        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] * scale) // 2, (rotated_img.shape[0] - image.shape[0] * scale) // 2)
+        # return rotated_img, (0, 0)
+
+    @staticmethod
+    def rotate_crop_image(img, angle, crop):
+        """
+        angle: 旋转的角度
+        crop: 是否需要进行裁剪，布尔向量
+        """
+        crop_image = lambda img, x0, y0, w, h: img[y0:y0 + h, x0:x0 + w]
+        w, h = 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 > 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

app/service/design_batch/pipeline/scale.py

@@ -0,0 +1,49 @@
+import math
+
+import cv2
+
+
+class Scaling:
+    def __call__(self, result):
+        if result['keypoint'] in ['waistband', 'shoulder', 'head_point']:
+            # milvus_db_keypoint_cache
+            distance_clo = math.sqrt(
+                (int(result['clothes_keypoint'][result['keypoint'] + '_left'][0]) - int(result['clothes_keypoint'][result['keypoint'] + '_right'][0])) ** 2
+                +
+                (int(result['clothes_keypoint'][result['keypoint'] + '_left'][1]) - int(result['clothes_keypoint'][result['keypoint'] + '_right'][1])) ** 2
+            )
+
+            distance_bdy = math.sqrt(
+                (int(result['body_point_test'][result['keypoint'] + '_left'][0])
+                 -
+                 int(result['body_point_test'][result['keypoint'] + '_right'][0])) ** 2 + 1
+            )
+            
+            if distance_clo == 0:
+                result['scale'] = 1
+            else:
+                result['scale'] = distance_bdy / distance_clo
+        elif result['keypoint'] == 'toe':
+            distance_bdy = math.sqrt(
+                (int(result['body_point_test']['foot_length'][0]) - int(result['body_point_test']['foot_length'][2])) ** 2
+                +
+                (int(result['body_point_test']['foot_length'][1]) - int(result['body_point_test']['foot_length'][3])) ** 2
+            )
+
+            Blur = cv2.GaussianBlur(result['gray'], (3, 3), 0)
+            Edge = cv2.Canny(Blur, 10, 200)
+            Edge = cv2.dilate(Edge, None)
+            Edge = cv2.erode(Edge, None)
+            Contour, _ = cv2.findContours(Edge, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+            Contours = sorted(Contour, key=cv2.contourArea, reverse=True)
+
+            Max_contour = Contours[0]
+            x, y, w, h = cv2.boundingRect(Max_contour)
+            width = w
+            distance_clo = width
+            result['scale'] = distance_bdy / distance_clo
+        elif result['keypoint'] == 'hand_point':
+            result['scale'] = result['scale_bag']
+        elif result['keypoint'] == 'ear_point':
+            result['scale'] = result['scale_earrings']
+        return result

app/service/design_batch/pipeline/segmentation.py

@@ -0,0 +1,70 @@
+import logging
+import os
+
+import cv2
+import numpy as np
+
+from app.core.config import SEG_CACHE_PATH
+from app.service.design_batch.utils.design_ensemble import get_seg_result
+from app.service.utils.new_oss_client import oss_get_image
+
+logger = logging.getLogger()
+
+
+class Segmentation:
+    def __init__(self, minio_client):
+        self.minio_client = minio_client
+
+    def __call__(self, result):
+        if "seg_mask_url" in result.keys() and result['seg_mask_url'] != "":
+            seg_mask = oss_get_image(oss_client=self.minio_client, bucket=result['seg_mask_url'].split('/')[0], object_name=result['seg_mask_url'][result['seg_mask_url'].find('/') + 1:], data_type="cv2")
+            seg_mask = cv2.resize(seg_mask, (result['img_shape'][1], result['img_shape'][0]), interpolation=cv2.INTER_NEAREST)
+            # 转换颜色空间为 RGB（OpenCV 默认是 BGR）
+            image_rgb = cv2.cvtColor(seg_mask, cv2.COLOR_BGR2RGB)
+
+            r, g, b = cv2.split(image_rgb)
+            red_mask = r > g
+            green_mask = g > r
+
+            # 创建红色和绿色掩码
+            result['front_mask'] = np.array(red_mask, dtype=np.uint8) * 255
+            result['back_mask'] = np.array(green_mask, dtype=np.uint8) * 255
+            result['mask'] = result['front_mask'] + result['back_mask']
+        else:
+            # 本地查询seg 缓存是否存在
+            _, seg_result = self.load_seg_result(result["image_id"])
+            result['seg_result'] = seg_result
+            if not _:
+                # 推理获得seg 结果
+                seg_result = get_seg_result(result["image_id"], result['image'])[0]
+                self.save_seg_result(seg_result, result['image_id'])
+            # 处理前片后片
+            temp_front = seg_result == 1.0
+            result['front_mask'] = (255 * (temp_front + 0).astype(np.uint8))
+            temp_back = seg_result == 2.0
+            result['back_mask'] = (255 * (temp_back + 0).astype(np.uint8))
+            result['mask'] = result['front_mask'] + result['back_mask']
+        return result
+
+    @staticmethod
+    def save_seg_result(seg_result, image_id):
+        file_path = f"seg_cache/{image_id}.npy"
+        try:
+            np.save(file_path, seg_result)
+            logger.info(f"保存成功 ：{os.path.abspath(file_path)}")
+        except Exception as e:
+            logger.error(f"保存失败: {e}")
+
+    @staticmethod
+    def load_seg_result(image_id):
+        file_path = f"seg_cache/{image_id}.npy"
+        logger.info(f"load seg file name is :{SEG_CACHE_PATH}{image_id}.npy")
+        try:
+            seg_result = np.load(file_path)
+            return True, seg_result
+        except FileNotFoundError:
+            logger.warning("文件不存在")
+            return False, None
+        except Exception as e:
+            logger.error(f"加载失败: {e}")
+            return False, None

app/service/design_batch/pipeline/split.py

@@ -0,0 +1,74 @@
+import io
+import logging
+
+import cv2
+import numpy as np
+from PIL import Image
+from cv2 import cvtColor, COLOR_BGR2RGBA
+
+from app.core.config import AIDA_CLOTHING
+from app.service.design_batch.utils.conversion_image import rgb_to_rgba
+from app.service.design_batch.utils.upload_image import upload_png_mask
+from app.service.utils.generate_uuid import generate_uuid
+from app.service.utils.new_oss_client import oss_upload_image
+
+
+class Split(object):
+    def __init__(self, minio_client):
+        self.minio_client = minio_client
+
+    def __call__(self, result):
+        try:
+
+            if result['name'] in ('outwear', 'dress', 'blouse', 'skirt', 'trousers', 'tops', 'bottoms'):
+                front_mask = result['front_mask']
+                back_mask = result['back_mask']
+                rgba_image = rgb_to_rgba(result['final_image'], front_mask + back_mask)
+                new_size = (int(rgba_image.shape[1] * result["scale"] * result["resize_scale"][0]), int(rgba_image.shape[0] * result["scale"] * result["resize_scale"][1]))
+                rgba_image = cv2.resize(rgba_image, new_size)
+                result_front_image = np.zeros_like(rgba_image)
+                front_mask = cv2.resize(front_mask, new_size)
+                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'], result["front_image_url"], _ = upload_png_mask(self.minio_client, result_front_image_pil, f'{generate_uuid()}', mask=None)
+
+                height, width = front_mask.shape
+                mask_image = np.zeros((height, width, 3))
+                mask_image[front_mask != 0] = [0, 0, 255]
+
+                if result["name"] in ('blouse', 'dress', 'outwear', 'tops'):
+                    result_back_image = np.zeros_like(rgba_image)
+                    back_mask = cv2.resize(back_mask, new_size)
+                    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'], result["back_image_url"], _ = upload_png_mask(self.minio_client, result_back_image_pil, f'{generate_uuid()}', mask=None)
+                    mask_image[back_mask != 0] = [0, 255, 0]
+
+                    rbga_mask = rgb_to_rgba(mask_image, front_mask + back_mask)
+                    mask_pil = Image.fromarray(cvtColor(rbga_mask.astype(np.uint8), COLOR_BGR2RGBA))
+                    image_data = io.BytesIO()
+                    mask_pil.save(image_data, format='PNG')
+                    image_data.seek(0)
+                    image_bytes = image_data.read()
+                    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
+                else:
+                    rbga_mask = rgb_to_rgba(mask_image, front_mask)
+                    mask_pil = Image.fromarray(cvtColor(rbga_mask.astype(np.uint8), COLOR_BGR2RGBA))
+                    image_data = io.BytesIO()
+                    mask_pil.save(image_data, format='PNG')
+                    image_data.seek(0)
+                    image_bytes = image_data.read()
+                    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['back_image'] = None
+                    result["back_image_url"] = None
+                    # result["back_mask_url"] = None
+                    # result['back_mask_image'] = None
+            # 创建中间图层
+            result_pattern_image_rgba = rgb_to_rgba(result['pattern_image'], result['mask'])
+            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()}')
+            return result
+        except Exception as e:
+            logging.warning(f"split runtime exception : {e} image_id : {result['image_id']}")