| 1 | """ALPR recognition engine: image/video frame -> license plate number(s). |
| 2 | |
| 3 | Uses fast-alpr (open-source): an ONNX YOLO plate detector + an ONNX plate-OCR model, CPU-only. |
| 4 | Models download once on first use. This is the 'algorithm' - it turns pixels into plate strings; |
| 5 | the existing normalize/dedup/watchlist pipeline turns those strings into intelligence. |
| 6 | |
| 7 | Recognized plates from real images are REAL reads. Provide images you are authorized to process. |
| 8 | """ |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import os |
| 12 | import threading |
| 13 | from dataclasses import dataclass |
| 14 | |
| 15 | os.environ.setdefault("OMP_NUM_THREADS", "2") |
| 16 | |
| 17 | import numpy as np |
| 18 | |
| 19 | _alpr = None |
| 20 | _lock = threading.Lock() |
| 21 | |
| 22 | |
| 23 | def _engine(): |
| 24 | """Lazily construct the ALPR engine (downloads ONNX models on first call).""" |
| 25 | global _alpr |
| 26 | if _alpr is None: |
| 27 | with _lock: |
| 28 | if _alpr is None: |
| 29 | from fast_alpr import ALPR |
| 30 | _alpr = ALPR( |
| 31 | detector_model="yolo-v9-t-384-license-plate-end2end", |
| 32 | ocr_model="global-plates-mobile-vit-v2-model", |
| 33 | ) |
| 34 | return _alpr |
| 35 | |
| 36 | |
| 37 | @dataclass |
| 38 | class PlateDetection: |
| 39 | plate: str |
| 40 | plate_norm: str |
| 41 | detection_confidence: float |
| 42 | ocr_confidence: float |
| 43 | bbox: list[int] |
| 44 | |
| 45 | |
| 46 | def _norm(s: str) -> str: |
| 47 | return "".join(ch for ch in (s or "").upper() if ch.isalnum()) |
| 48 | |
| 49 | |
| 50 | def _scalar_conf(c) -> float: |
| 51 | """OCR confidence may be a per-character list; collapse to a single mean score.""" |
| 52 | try: |
| 53 | if isinstance(c, (list, tuple, np.ndarray)): |
| 54 | vals = [float(x) for x in c if x is not None] |
| 55 | return float(sum(vals) / len(vals)) if vals else 0.0 |
| 56 | return float(c) |
| 57 | except (TypeError, ValueError): |
| 58 | return 0.0 |
| 59 | |
| 60 | |
| 61 | def recognize_array(img_bgr: np.ndarray, min_conf: float = 0.4) -> list[PlateDetection]: |
| 62 | results = _engine().predict(img_bgr) |
| 63 | out: list[PlateDetection] = [] |
| 64 | for r in results: |
| 65 | det = getattr(r, "detection", None) |
| 66 | ocr = getattr(r, "ocr", None) |
| 67 | text = (ocr.text if ocr else "") or "" |
| 68 | oconf = _scalar_conf(ocr.confidence) if (ocr and ocr.confidence is not None) else 0.0 |
| 69 | dconf = float(det.confidence) if det and det.confidence is not None else 0.0 |
| 70 | bb = det.bounding_box if det else None |
| 71 | bbox = [int(bb.x1), int(bb.y1), int(bb.x2), int(bb.y2)] if bb else [] |
| 72 | norm = _norm(text) |
| 73 | if not norm or oconf < min_conf: |
| 74 | continue |
| 75 | out.append(PlateDetection(plate=text.strip(), plate_norm=norm, |
| 76 | detection_confidence=dconf, ocr_confidence=oconf, bbox=bbox)) |
| 77 | return out |
| 78 | |
| 79 | |
| 80 | def recognize_bytes(data: bytes, min_conf: float = 0.4) -> list[PlateDetection]: |
| 81 | import cv2 |
| 82 | arr = np.frombuffer(data, dtype=np.uint8) |
| 83 | img = cv2.imdecode(arr, cv2.IMREAD_COLOR) |
| 84 | if img is None: |
| 85 | raise ValueError("could not decode image") |
| 86 | return recognize_array(img, min_conf=min_conf) |