AiDA_Python/app/service/design/items/pipelines/segmentation.py

import os

import numpy as np

from app.core.config import SEG_CACHE_PATH
from ..builder import PIPELINES
from ...utils.design_ensemble import get_seg_result


@PIPELINES.register_module()
class Segmentation(object):

    # @ClassCallRunTime
    def __call__(self, result):
        # 本地查询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_PATH}{image_id}.npy"
        try:
            np.save(file_path, seg_result)
            print("保存成功", os.path.abspath(file_path))
        except Exception as e:
            print(f"保存失败: {e}")

    @staticmethod
    def load_seg_result(image_id):
        file_path = f"{SEG_CACHE_PATH}{image_id}.npy"
        try:
            seg_result = np.load(file_path)
            return True, seg_result
        except FileNotFoundError:
            print("文件不存在")
            return False, None
        except Exception as e:
            print(f"加载失败: {e}")
            return False, None