CheckCardBoxAndDefectServer/app/services/defect_service.py

import cv2
import numpy as np
from ..core.model_loader import get_predictor
from app.utils.CardDefectAggregator import CardDefectAggregator
from app.utils.arean_anylize_draw import DefectProcessor, DrawingParams
from app.utils.AnalyzeCenter import analyze_centering_rotated
from app.utils.json_data_formate import formate_center_data
from app.core.config import settings
from app.core.logger import logger
import json


class DefectInferenceService:
    def defect_inference(self, inference_type: str, image_bytes: bytes,
                         is_draw_image=False) -> dict:
        """
        执行卡片识别推理。

        Args:
            inference_type: 模型类型 (e.g., 'outer_box').
            image_bytes: 从API请求中获得的原始图像字节。

        Returns:
            一个包含推理结果的字典。
        """
        # 2. 将字节流解码为OpenCV图像
        # 将字节数据转换为numpy数组
        np_arr = np.frombuffer(image_bytes, np.uint8)
        # 从numpy数组中解码图像
        img_bgr = cv2.imdecode(np_arr, cv2.IMREAD_COLOR)

        if img_bgr is None:
            logger.error("无法解码图像，请确保上传的是有效的图片格式 (JPG, PNG, etc.)")
            return {}

        # 面
        if (inference_type == "pokemon_front_face_no_reflect_defect"
                or inference_type == "pokemon_front_face_reflect_defect"
                or inference_type == "pokemon_back_face_defect"):
            # 1. 获取对应的预测器实例
            predictor = get_predictor(inference_type)

            # 3. 调用我们新加的 predict_from_image 方法进行推理
            # result = predictor.predict_from_image(img_bgr)

            # 3. 实例化我们聚合器，传入预测器
            aggregator = CardDefectAggregator(
                predictor=predictor,
                tile_size=512,
                overlap_ratio=0.1,  # 10% 重叠
            )

            json_data = aggregator.process_image(
                image=img_bgr,
                mode='face'
            )
            # merge_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-merge.json'
            # with open(merge_json_path, 'w', encoding='utf-8') as f:
            #     json.dump(json_data, f, ensure_ascii=False, indent=4)
            # logger.info(f"合并结束")

            processor = DefectProcessor(pixel_resolution=settings.pixel_resolution)
            area_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-area_result.json'
            if is_draw_image:
                drawing_params_with_rect = DrawingParams(draw_min_rect=True)
                drawn_image, area_json = processor.analyze_and_draw(img_bgr, json_data,
                                                                    drawing_params_with_rect)
                temp_img_path = settings.TEMP_WORK_DIR / f'{inference_type}-area_result.jpg'
                cv2.imwrite(temp_img_path, drawn_image)
            else:
                area_json = processor.analyze_from_json(json_data)

            with open(area_json_path, 'w', encoding='utf-8') as f:
                json.dump(area_json, f, ensure_ascii=False, indent=4)
            logger.info("面积计算结束")

            return area_json

        # 边角
        elif (inference_type == "pokemon_front_corner_no_reflect_defect"
              or inference_type == "pokemon_front_corner_reflect_defect"
              or inference_type == "pokemon_back_corner_defect"):
            predictor = get_predictor(inference_type)

            aggregator = CardDefectAggregator(
                predictor=predictor,
                tile_size=512,
                overlap_ratio=0.1,  # 10% 重叠
            )

            json_data = aggregator.process_image(
                image=img_bgr,
                mode='edge'
            )

            # merge_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-merge.json'
            # with open(merge_json_path, 'w', encoding='utf-8') as f:
            #     json.dump(json_data, f, ensure_ascii=False, indent=4)
            # logger.info(f"合并结束")

            processor = DefectProcessor(pixel_resolution=settings.pixel_resolution)
            area_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-area_result.json'
            if is_draw_image:
                drawing_params_with_rect = DrawingParams(draw_min_rect=True)
                drawn_image, area_json = processor.analyze_and_draw(img_bgr, json_data,
                                                                    drawing_params_with_rect)
                temp_img_path = settings.TEMP_WORK_DIR / f'{inference_type}-area_result.jpg'
                cv2.imwrite(temp_img_path, drawn_image)
            else:
                area_json = processor.analyze_from_json(json_data)

            with open(area_json_path, 'w', encoding='utf-8') as f:
                json.dump(area_json, f, ensure_ascii=False, indent=4)
            logger.info("面积计算结束")

            return area_json

        elif inference_type == "pokemon_card_center":
            predictor_inner = get_predictor(settings.DEFECT_TYPE[inference_type]['inner_box'])
            predictor_outer = get_predictor(settings.DEFECT_TYPE[inference_type]['outer_box'])

            inner_result = predictor_inner.predict_from_image(img_bgr)
            outer_result = predictor_outer.predict_from_image(img_bgr)

            # temp_inner_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-inner_result.json'
            # temp_outer_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-outer_result.json'
            # with open(temp_inner_json_path, 'w', encoding='utf-8') as f:
            #     json.dump(inner_result, f, ensure_ascii=False, indent=4)
            # with open(temp_outer_json_path, 'w', encoding='utf-8') as f:
            #     json.dump(outer_result, f, ensure_ascii=False, indent=4)

            inner_points = inner_result['shapes'][0]['points']
            outer_points = outer_result['shapes'][0]['points']
            center_result = analyze_centering_rotated(inner_points, outer_points)
            logger.info("格式化居中数据")
            center_result = formate_center_data(center_result, inner_result, outer_result)

            temp_center_json_path = settings.TEMP_WORK_DIR / f'{inference_type}-center_result.json'
            with open(temp_center_json_path, 'w', encoding='utf-8') as f:
                json.dump(center_result, f, ensure_ascii=False, indent=2)
            return center_result

        else:
            return {}


# 创建一个单例服务
defect_service = DefectInferenceService()