""" pipeline.py — Two-Stage AI Pipeline Stage 1 : YOLOv8 (ultralytics) → หา Bounding Box ของ "แผ่นป้ายทะเบียน" แล้ว Crop Stage 2 : EasyOCR (th + en) → อ่านตัวอักษร/ตัวเลขจากภาพที่ Crop มา Stage 3 : Post-Processing → Regex กรองเฉพาะรูปแบบป้ายรถบรรทุก (ช่วงหมวดตาม PLATE_PREFIX_MIN/MAX ใน .env) หลักการสำคัญ: * โมเดลถูกโหลด "ครั้งเดียว" ตอนสตาร์ทเซิร์ฟเวอร์ (warm-up) ไม่ใช่โหลดทุก request * ทุกขั้นตอนห่อด้วย try/except — error ใดๆ กลายเป็นผลลัพธ์ที่อธิบายได้ ไม่ใช่ 500 * งาน AI เป็น CPU-bound จึงถูกเรียกผ่าน thread pool ในชั้น API (ไม่บล็อก event loop) """ from __future__ import annotations import logging import os import time from collections import Counter from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Tuple import cv2 import numpy as np from .config import settings from . import plate_utils as pu log = logging.getLogger("pipeline") # ===================================================================== # ผลลัพธ์ของ pipeline # ===================================================================== @dataclass class ScanResult: ok: bool code: str # OK / PLATE_NOT_DETECTED / OCR_FAILED / INVALID_FORMAT / ENGINE_ERROR message: str plate: Optional[str] = None confidence: float = 0.0 votes: int = 0 # จำนวน variant ที่อ่านได้ตรงกัน needs_confirm: bool = False # True = ได้เสียงเดียว ให้พนักงานตรวจก่อนปล่อยคิว province: Optional[str] = None raw_texts: List[str] = field(default_factory=list) detector: str = "" # 'plate-model' / 'vehicle-fallback' / 'whole-image' elapsed_ms: int = 0 debug_path: Optional[str] = None # ===================================================================== # ตัว Pipeline # ===================================================================== class PlatePipeline: """ห่อโมเดล YOLO + EasyOCR ไว้ในคลาสเดียว โหลดครั้งเดียวใช้ตลอดอายุโปรเซส""" # class id ของ COCO ที่นับว่าเป็น "ยานพาหนะ" (ใช้เฉพาะตอน fallback yolov8n.pt) _COCO_VEHICLE_IDS = {2, 3, 5, 7} # car, motorcycle, bus, truck # ต้องอ่านได้ตรงกันกี่ครั้งจึงจะถือว่า "มั่นใจ" — ดูเหตุผลละเอียดใน scan() MIN_VOTES = max(1, settings.MIN_VOTES) def __init__(self) -> None: self.detector = None self.reader = None self.detector_kind = "none" # 'plate' = โมเดลป้ายทะเบียนโดยตรง, 'vehicle' = yolov8n self.ready = False self.load_error: Optional[str] = None # ----------------------------------------------------------------- # โหลดโมเดล (เรียกครั้งเดียวตอน startup) # ----------------------------------------------------------------- def load(self) -> None: t0 = time.time() # ---------- Stage 1 : YOLO ---------- try: from ultralytics import YOLO # import ตรงนี้เพื่อให้ error ชัดเจนถ้ายังไม่ได้ pip install model_path = self._pick_model_path() log.info("กำลังโหลดโมเดล YOLO: %s", model_path) self.detector = YOLO(model_path) # ตัดสินว่าเป็นโมเดลป้ายทะเบียนเฉพาะทาง หรือโมเดล COCO ทั่วไป names = getattr(self.detector, "names", {}) or {} label_blob = " ".join(str(v).lower() for v in names.values()) is_plate_model = ("plate" in label_blob or "licence" in label_blob or "license" in label_blob or len(names) <= 3) self.detector_kind = "plate" if is_plate_model else "vehicle" log.info("ชนิดโมเดลที่ตรวจพบ: %s (classes=%s)", self.detector_kind, list(names.values())[:5]) except Exception as exc: # noqa: BLE001 self.load_error = f"โหลดโมเดล YOLO ไม่สำเร็จ: {exc}" log.exception(self.load_error) # ---------- Stage 2 : EasyOCR ---------- try: import easyocr log.info("กำลังโหลด EasyOCR langs=%s gpu=%s (ครั้งแรกจะดาวน์โหลดโมเดล ~100MB)", settings.OCR_LANGS, settings.OCR_GPU) self.reader = easyocr.Reader(settings.OCR_LANGS, gpu=settings.OCR_GPU, verbose=False) except Exception as exc: # noqa: BLE001 msg = f"โหลด EasyOCR ไม่สำเร็จ: {exc}" self.load_error = (self.load_error + " | " + msg) if self.load_error else msg log.exception(msg) self.ready = self.reader is not None log.info("โหลดโมเดลเสร็จใน %.1f วินาที (ready=%s)", time.time() - t0, self.ready) # warm-up: ยิงภาพดำ 1 ครั้ง ให้ lazy-init ภายในไลบรารีทำงานจบก่อนรับ request จริง if self.ready: try: dummy = np.zeros((120, 360, 3), dtype=np.uint8) if self.detector is not None: self.detector.predict(dummy, verbose=False, device=settings.YOLO_DEVICE) self.reader.readtext(dummy, detail=1) log.info("Warm-up สำเร็จ พร้อมรับงาน") except Exception as exc: # noqa: BLE001 log.warning("Warm-up ไม่ผ่าน (ไม่ร้ายแรง): %s", exc) def _pick_model_path(self) -> str: """เลือกไฟล์โมเดลตัวแรกที่มีอยู่จริง ถ้าไม่มีเลยให้ใช้ yolov8n.pt (ultralytics โหลดเอง)""" settings.MODEL_DIR.mkdir(parents=True, exist_ok=True) for cand in settings.PLATE_MODEL_CANDIDATES: if cand and Path(cand).is_file(): return cand fallback = settings.FALLBACK_MODEL if Path(fallback).is_file(): return fallback # ส่งชื่อเปล่าๆ ให้ ultralytics ดาวน์โหลดเองครั้งแรก return "yolov8n.pt" # ----------------------------------------------------------------- # Stage 1 : ตรวจจับ + Crop # ----------------------------------------------------------------- def detect_plate_regions(self, bgr: np.ndarray) -> Tuple[List[np.ndarray], str]: """ คืน (รายการภาพ crop เรียงตามความมั่นใจมาก→น้อย, ชื่อวิธีที่ใช้) กรณีใช้โมเดลป้ายทะเบียนโดยตรง → crop เฉพาะกรอบป้าย กรณีใช้ yolov8n (COCO) → crop ตัวรถก่อน แล้วโซนล่างของรถ (ป้ายมักอยู่ครึ่งล่าง) ถ้าตรวจไม่เจออะไรเลย → คืนทั้งภาพเป็นทางเลือกสุดท้าย """ crops: List[np.ndarray] = [] if self.detector is None: return [bgr], "whole-image" try: res = self.detector.predict( bgr, conf=settings.YOLO_CONF, imgsz=settings.YOLO_IMGSZ, device=settings.YOLO_DEVICE, verbose=False, )[0] except Exception as exc: # noqa: BLE001 log.warning("YOLO predict ล้มเหลว ใช้ทั้งภาพแทน: %s", exc) return [bgr], "whole-image" boxes = getattr(res, "boxes", None) if boxes is None or len(boxes) == 0: return [bgr], "whole-image" # เรียงตาม confidence มาก → น้อย items = [] for b in boxes: conf = float(b.conf[0]) cls = int(b.cls[0]) x1, y1, x2, y2 = [int(v) for v in b.xyxy[0].tolist()] items.append((conf, cls, x1, y1, x2, y2)) items.sort(key=lambda t: -t[0]) if self.detector_kind == "plate": for conf, _cls, x1, y1, x2, y2 in items[:3]: x1, y1, x2, y2 = pu.expand_box(x1, y1, x2, y2, bgr.shape, pad_ratio=0.08) if x2 > x1 and y2 > y1: crops.append(bgr[y1:y2, x1:x2].copy()) if crops: return crops, "plate-model" return [bgr], "whole-image" # ----- fallback: โมเดล COCO ตรวจได้แค่ "รถ" ----- for conf, cls, x1, y1, x2, y2 in items: if cls not in self._COCO_VEHICLE_IDS: continue x1, y1, x2, y2 = pu.expand_box(x1, y1, x2, y2, bgr.shape, pad_ratio=0.02) if x2 <= x1 or y2 <= y1: continue vehicle = bgr[y1:y2, x1:x2] vh = vehicle.shape[0] # ป้ายทะเบียนรถบรรทุกอยู่ช่วงล่างของตัวรถ → ตัดเฉพาะ 45% ล่าง ลดพื้นที่รบกวน # # เดิมใส่ "ภาพรถทั้งคัน" ต่อท้ายด้วย แต่วัดจากภาพหน้างานจริงแล้วพบว่า # crop ทั้งคันมีขนาดราว 1000x900 px ซึ่ง EasyOCR ใช้เวลาอ่านหลายวินาทีต่อรอบ # ทั้งที่แถบล่างอ่านเจอป้ายอยู่แล้ว จึงตัดทิ้งเพื่อลดเวลารวมลงครึ่งหนึ่ง crops.append(vehicle[int(vh * 0.55):, :].copy()) if len(crops) >= 2: break if crops: return crops, "vehicle-fallback" return [bgr], "whole-image" # ----------------------------------------------------------------- # Stage 2 : OCR # ----------------------------------------------------------------- def ocr_texts(self, img: np.ndarray) -> List[Tuple[str, float]]: """อ่านข้อความจากภาพ คืน [(ข้อความ, ความมั่นใจ), ...]""" if self.reader is None: return [] try: # allowlist ไม่ถูกกำหนด เพราะต้องอ่านชื่อจังหวัดภาษาไทยด้วย raw = self.reader.readtext(img, detail=1, paragraph=False) except Exception as exc: # noqa: BLE001 log.warning("EasyOCR readtext ล้มเหลว: %s", exc) return [] out: List[Tuple[str, float]] = [] for item in raw: try: _box, text, conf = item[0], item[1], float(item[2]) except (IndexError, TypeError, ValueError): continue if conf >= settings.OCR_MIN_CONF and text and text.strip(): out.append((text.strip(), conf)) return out # ----------------------------------------------------------------- # รวมทุกขั้นตอน # ----------------------------------------------------------------- def scan(self, bgr: np.ndarray) -> ScanResult: t0 = time.perf_counter() if not self.ready: return ScanResult(False, "ENGINE_ERROR", self.load_error or "ระบบ AI ยังไม่พร้อมใช้งาน", elapsed_ms=0) try: # ---------- Stage 1 ---------- bgr = pu.reduce_glare(bgr) crops, how = self.detect_plate_regions(bgr) # ---------- Stage 2 + 3 : อ่านหลายเวอร์ชันแล้ว "โหวต" ---------- # # เหตุผลที่ต้องโหวต (สำคัญมาก): # จากการทดสอบกับภาพป้ายจริง OCR สามารถอ่านผิดเป็นเลขที่ "ถูกรูปแบบ" ได้ # เช่นป้ายจริง 70-8511 แต่อ่านออกมาเป็น 70-8514 หรือ 70-0511 # ซึ่งผ่าน Regex ทุกด่านและจะไปเรียก Order ของรถคันอื่นแบบเงียบๆ # # วิธีแก้: อ่านหลาย variant แล้วรับเฉพาะเลขที่ "อ่านได้ตรงกันอย่างน้อย 2 ครั้ง" # ถ้าตรงกันครั้งเดียว → ยังส่งผลกลับไป แต่ตั้งธง needs_confirm ให้พนักงานตรวจก่อน all_texts: List[str] = [] votes: Counter = Counter() conf_of: Dict[str, float] = {} plate: Optional[str] = None # จำกัดเวลา/ปริมาณงาน OCR ไม่ให้พนักงานยืนรอนานเกินไป # # เหตุผล (วัดจากการรันจริงบน CPU): # - อ่านภาพป้ายที่ crop มาแล้ว ~1.3 วินาที/variant # - อ่านภาพเต็ม 1280px ทั้งภาพ ~15 วินาที/variant ← ช้ากว่า 10 เท่า # ถ้าโมเดลตรวจจับป้ายใช้ไม่ได้ ระบบจะตกมาโหมด "ทั้งภาพ" ซึ่งถ้าไล่ครบ # 4 variant จะกินเวลาถึง 1 นาที — รับไม่ได้หน้างาน จึงจำกัดไว้ # โหมด whole-image อ่านได้แค่รอบเดียว จึงได้เสียงโหวตสูงสุด 1 # แปลว่าจะขึ้น "ให้พนักงานตรวจ" เสมอ — ซึ่งถูกต้องแล้ว # เพราะโหมดนี้เชื่อถือได้น้อยที่สุด ไม่ควรปล่อยผ่านอัตโนมัติ max_variants = {"whole-image": 1, "vehicle-fallback": 3}.get(how, 4) deadline = t0 + settings.MAX_SCAN_SECONDS for crop in crops: if crop is None or crop.size == 0: continue for variant in pu.variant_images(crop)[:max_variants]: if time.perf_counter() > deadline: log.warning("หมดเวลางบประมาณ OCR (%.0fs) — ใช้ผลเท่าที่ได้", settings.MAX_SCAN_SECONDS) break pairs = self.ocr_texts(variant) if not pairs: continue texts = [p[0] for p in pairs] all_texts.extend(texts) for cand in set(pu.collect_truck_plates(texts)): votes[cand] += 1 # ความมั่นใจ = ค่าสูงสุดของก้อนข้อความที่มีตัวเลขของป้ายอยู่ digits = cand.replace("-", "") best = conf_of.get(cand, 0.0) for txt, conf in pairs: squashed = pu.normalize_digits(txt).replace("-", "").replace(" ", "") if digits[:2] in squashed or digits[2:] in squashed: best = max(best, conf) conf_of[cand] = best or max(c for _, c in pairs) # หยุดทันทีเมื่อมีตัวใดได้ 2 เสียงแล้ว (ประหยัดเวลา ไม่ต้องอ่านครบทุก variant) if votes and votes.most_common(1)[0][1] >= self.MIN_VOTES: break if time.perf_counter() > deadline: break if votes and votes.most_common(1)[0][1] >= self.MIN_VOTES: break elapsed = int((time.perf_counter() - t0) * 1000) uniq_texts = list(dict.fromkeys(all_texts))[:20] # ---------- สรุปผล ---------- if votes: plate, n_votes = votes.most_common(1)[0] needs_confirm = n_votes < self.MIN_VOTES msg = ("อ่านป้ายทะเบียนสำเร็จ" if not needs_confirm else "อ่านได้แต่ยังไม่มั่นใจ — กรุณาตรวจสอบเลขกับตัวรถก่อนปล่อยคิว") return ScanResult( True, "OK", msg, plate=plate, confidence=round(conf_of.get(plate, 0.0), 3), votes=n_votes, needs_confirm=needs_confirm, province=pu.extract_province(uniq_texts), raw_texts=uniq_texts, detector=how, elapsed_ms=elapsed, ) if not all_texts: code, msg = ("PLATE_NOT_DETECTED", "ไม่พบตัวอักษรบนภาพ — ป้ายอาจอยู่ไกลเกินไป มีโคลนบัง หรือแสงสะท้อนจนขาว") else: car = pu.find_car_plate(uniq_texts) if car: code, msg = ("INVALID_FORMAT", f"อ่านได้เป็นป้ายรถยนต์ทั่วไป ({car}) ไม่ใช่ป้ายรถบรรทุกหมวด " f"{settings.PLATE_PREFIX_MIN}-{settings.PLATE_PREFIX_MAX}") else: code, msg = ("INVALID_FORMAT", "อ่านตัวอักษรได้ แต่ไม่พบรูปแบบเลขทะเบียนรถบรรทุกที่ถูกต้อง") return ScanResult(False, code, msg, raw_texts=uniq_texts, detector=how, elapsed_ms=elapsed) except Exception as exc: # noqa: BLE001 # กันทุกกรณีที่คาดไม่ถึง ไม่ให้ worker ตาย log.exception("pipeline.scan เกิดข้อผิดพลาดที่ไม่คาดคิด") return ScanResult(False, "ENGINE_ERROR", f"ประมวลผลภาพไม่สำเร็จ: {type(exc).__name__}", elapsed_ms=int((time.perf_counter() - t0) * 1000)) # ----------------------------------------------------------------- # เก็บภาพไว้ตรวจสอบย้อนหลัง (ช่วยเก็บ dataset ไปเทรนโมเดลเองภายหลัง) # ----------------------------------------------------------------- @staticmethod def save_debug(bgr: np.ndarray, tag: str) -> Optional[str]: if not settings.SAVE_DEBUG_IMAGES: return None try: settings.DEBUG_DIR.mkdir(parents=True, exist_ok=True) name = f"{time.strftime('%Y%m%d_%H%M%S')}_{tag}.jpg" path = settings.DEBUG_DIR / name cv2.imwrite(str(path), bgr, [int(cv2.IMWRITE_JPEG_QUALITY), 70]) return str(path) except Exception as exc: # noqa: BLE001 log.warning("บันทึกภาพ debug ไม่สำเร็จ: %s", exc) return None # instance เดียวใช้ทั้งแอป pipeline = PlatePipeline()