2

backend/txm/app/pipeline.py

@@ -0,0 +1,139 @@
+from typing import Optional, List, Dict
+
+import cv2
+import logging
+
+from .config_loader import load_config
+from .image_processing import (
+    binarize_image,
+    enhance_barcode_stripes,
+    find_barcode_roi,
+    morph_close,
+    resize_keep_aspect,
+    warp_barcode_region,
+)
+from .ean13_decoder import sample_and_decode
+from .io_utils import read_image_bgr
+from .pyzbar_engine import decode_with_pyzbar
+
+
+class EAN13Recognizer:
+    def __init__(self, config_path: str = "config/config.yaml") -> None:
+        self.logger = logging.getLogger(self.__class__.__name__)
+        self.config = load_config(config_path)
+        self.logger.debug("配置加载完成: preprocess=%s, roi=%s, decoder=%s", self.config.get("preprocess"), self.config.get("roi"), self.config.get("decoder"))
+
+    def recognize_from_path(self, image_path: str) -> Optional[str]:
+        image = read_image_bgr(image_path)
+        if image is None:
+            self.logger.warning("读取图像失败: path=%s", image_path)
+            return None
+        return self.recognize_from_image(image)
+
+    def _recognize_with_pyzbar(self, img) -> Optional[str]:
+        decoder_cfg = self.config["decoder"]
+        pyz_res = decode_with_pyzbar(
+            img,
+            try_invert=decoder_cfg.get("try_invert", True),
+            rotations=decoder_cfg.get("rotations", [0, 90, 180, 270]),
+        )
+        if pyz_res:
+            self.logger.debug("pyzbar 识别到 %d 条结果", len(pyz_res))
+        for r in pyz_res:
+            if r.get("type") in ("EAN13", "EAN-13") and r.get("code"):
+                self.logger.info("pyzbar 命中 EAN13: %s", r.get("code"))
+                return r.get("code")
+        return None
+
+    def recognize_from_image(self, image) -> Optional[str]:
+        cfg = self.config
+        preprocess_cfg = cfg["preprocess"]
+        roi_cfg = cfg["roi"]
+        decoder_cfg = cfg["decoder"]
+
+        img = resize_keep_aspect(image, preprocess_cfg.get("resize_width", 0))
+        self.logger.debug("输入尺寸=%s, 预处理后尺寸=%s", getattr(image, 'shape', None), getattr(img, 'shape', None))
+
+        # 先按引擎优先级尝试 Pyzbar
+        engine_order: List[str] = decoder_cfg.get("engine_order", ["pyzbar", "ean13"])
+        if "pyzbar" in engine_order:
+            code = self._recognize_with_pyzbar(img)
+            if code:
+                return code
+            if "ean13" not in engine_order:
+                return None
+
+        # 回退到自研 EAN-13 解码
+        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
+        grad = enhance_barcode_stripes(
+            gray,
+            gaussian_ksize=preprocess_cfg.get("gaussian_blur_ksize", 3),
+            sobel_ksize=preprocess_cfg.get("sobel_ksize", 3),
+        )
+        bin_img = binarize_image(grad, preprocess_cfg.get("binarize", "otsu"))
+        closed = morph_close(bin_img, preprocess_cfg.get("close_kernel", 21))
+
+        roi_result = find_barcode_roi(
+            closed,
+            original_shape=gray.shape[:2],
+            min_area_ratio=roi_cfg.get("min_area_ratio", 0.01),
+            min_wh_ratio=roi_cfg.get("min_wh_ratio", 2.0),
+        )
+        if not roi_result:
+            self.logger.debug("未找到条码 ROI")
+            return None
+        _, bbox = roi_result
+        self.logger.debug("ROI bbox=%s", bbox)
+        warped = warp_barcode_region(
+            gray,
+            bbox,
+            roi_cfg.get("warp_target_height", 120),
+            crop_bottom_ratio=roi_cfg.get("crop_bottom_ratio", 0.0),
+        )
+        self.logger.debug("透视矫正后尺寸=%s", getattr(warped, 'shape', None))
+
+        code = sample_and_decode(
+            warped,
+            decoder_cfg.get("sample_rows", [0.5]),
+            decoder_cfg.get("total_modules", 95),
+        )
+        if code:
+            self.logger.info("自研 EAN13 解码成功: %s", code)
+        else:
+            self.logger.debug("自研 EAN13 解码失败")
+        return code
+
+    def recognize_any_from_image(self, image) -> Dict[str, object]:
+        cfg = self.config
+        decoder_cfg = cfg["decoder"]
+        img = resize_keep_aspect(image, cfg["preprocess"].get("resize_width", 0))
+
+        # Pyzbar 全量识别
+        pyz_res = decode_with_pyzbar(
+            img,
+            try_invert=decoder_cfg.get("try_invert", True),
+            rotations=decoder_cfg.get("rotations", [0, 90, 180, 270]),
+        )
+        self.logger.debug("pyzbar 返回 %d 条结果", len(pyz_res) if isinstance(pyz_res, list) else 0)
+        ean13 = ""
+        for r in pyz_res:
+            if r.get("type") in ("EAN13", "EAN-13") and r.get("code"):
+                ean13 = r.get("code")
+                break
+        others = [r for r in pyz_res if r.get("type") not in ("EAN13", "EAN-13")]
+        if not ean13:
+            e = self.recognize_from_image(img)
+            if e:
+                ean13 = e
+        if ean13:
+            self.logger.info("recognize_any 命中 EAN13: %s, others=%d", ean13, len(others))
+        else:
+            self.logger.debug("recognize_any 未命中 EAN13, others=%d", len(others))
+        return {"ean13": ean13, "others": others}
+
+    def recognize_any_from_path(self, image_path: str) -> Dict[str, object]:
+        image = read_image_bgr(image_path)
+        if image is None:
+            return {"ean13": "", "others": []}
+        return self.recognize_any_from_image(image)
+