"""The recognizer port + implementations. The service depends on the `Recognizer` PROTOCOL, never a concrete model library — the same swappable-behind-an-interface principle as the Node device adapters (wiki/concepts/device-adapter-pattern.md). Two impls today: - StubRecognizer: no model weights, deterministic placeholder. Lets the service boot and the tests run offline with nothing downloaded (dev/CI default). - FastAlprRecognizer: the real MIT YOLOv9-detector + CCT-OCR stack on ONNX Runtime (the `alpr` extra). See wiki/entities/opencv-anpr-service.md "Recognizer evaluation". Adding a recognizer (e.g. a fine-tuned YOLO + PaddleOCR) = a new class here, no app change. """ from __future__ import annotations import time from typing import Protocol from .schemas import AnalyzeResponse, BBox, PlateResult from .settings import Settings from .vehicle import VehicleDetector, YoloxVehicleDetector class Recognizer(Protocol): """Reads plates from a JPEG/PNG image. Implementations must be process-local and offline.""" @property def model_version(self) -> str: ... @property def ready(self) -> bool: ... def analyze(self, image_bytes: bytes) -> AnalyzeResponse: ... def _reduce_confidence(raw: object) -> float: """fast-alpr's OCR confidence is a LIST of per-character confidences. Reduce to one plate confidence via the MIN — a plate is only as trustworthy as its weakest character (one misread digit changes the identity). Tolerates a scalar (future models) or junk (→ 0.0). Pure + model-free so it's unit-testable without weights.""" if isinstance(raw, (list, tuple)) and raw: try: return float(min(raw)) except (TypeError, ValueError): return 0.0 if isinstance(raw, (int, float)): return float(raw) return 0.0 def plate_from_alpr_result(r: object) -> PlateResult | None: """Map ONE fast-alpr ALPRResult to our PlateResult, or None if it carries no text. Uses getattr throughout so it's decoupled from the exact fast-alpr classes (and testable with a duck-typed stand-in). See wiki/entities/opencv-anpr-service.md.""" ocr = getattr(r, "ocr", None) det = getattr(r, "detection", None) text = getattr(ocr, "text", None) if ocr is None or not text: return None bbox = None box = getattr(det, "bounding_box", None) if box is not None: bbox = BBox(x1=int(box.x1), y1=int(box.y1), x2=int(box.x2), y2=int(box.y2)) return PlateResult( text=text, confidence=_reduce_confidence(getattr(ocr, "confidence", None)), bbox=bbox, region=getattr(ocr, "region", None), ) class StubRecognizer: """A no-model placeholder. Returns an empty (no-plate) result quickly so the whole HTTP path — Node adapter, contract, error handling — can be exercised without the heavy recognizer stack or any model download.""" model_version = "stub-0" ready = True def __init__(self, settings: Settings) -> None: self._settings = settings def analyze(self, image_bytes: bytes) -> AnalyzeResponse: started = time.perf_counter() # Deliberately recognizes nothing — it is a stub, not a fake "always finds a plate" # (which would be dangerous: recognition must never invent an identity). took_ms = (time.perf_counter() - started) * 1000.0 return AnalyzeResponse( plate=None, plates=[], vehicle=None, low_confidence=False, model_version=self.model_version, took_ms=took_ms, ) class FastAlprRecognizer: """The real recognizer: fast-alpr (YOLOv9 plate detector + CCT OCR, ONNX Runtime). Imported lazily so the service still imports/boots in stub mode when the `alpr` extra (and its model weights) are not installed — a missing recognizer must not crash the process; it degrades to a clear `ready=False`. """ def __init__(self, settings: Settings) -> None: self._settings = settings self._alpr = None self._error: str | None = None try: from fast_alpr import ALPR self._alpr = ALPR( detector_model=settings.detector_model, ocr_model=settings.ocr_model, ) except Exception as exc: # noqa: BLE001 - any failure ⇒ not-ready, surfaced via /health self._error = f"{type(exc).__name__}: {exc}" @property def model_version(self) -> str: return f"fast-alpr:{self._settings.detector_model}+{self._settings.ocr_model}" @property def ready(self) -> bool: return self._alpr is not None @property def error(self) -> str | None: return self._error def analyze(self, image_bytes: bytes) -> AnalyzeResponse: if self._alpr is None: raise RuntimeError(f"fast-alpr not available: {self._error}") # fast-alpr's predict() takes a BGR ndarray; decode the JPEG with cv2 (pulled in # transitively by the alpr extra). Import locally so stub mode needs neither. import cv2 import numpy as np # local import: only needed on the real path started = time.perf_counter() buf = np.frombuffer(image_bytes, dtype=np.uint8) frame = cv2.imdecode(buf, cv2.IMREAD_COLOR) if frame is None: raise ValueError("could not decode image bytes") results = self._alpr.predict(frame) plates: list[PlateResult] = [] for r in results: plate = plate_from_alpr_result(r) if plate is not None: plates.append(plate) plates.sort(key=lambda p: p.confidence, reverse=True) best = plates[0] if plates else None low = best is not None and best.confidence < self._settings.min_confidence took_ms = (time.perf_counter() - started) * 1000.0 return AnalyzeResponse( plate=best, plates=plates, vehicle=None, # Job 2 not built yet low_confidence=low, model_version=self.model_version, took_ms=took_ms, ) class WithVehicle: """Composition: any plate recognizer + the vehicle stage. Runs the plate stage first (its box picks WHICH vehicle), then fills `vehicle`. A failing vehicle stage is logged into `error` and yields null — it must never cost the plate read.""" def __init__(self, inner: Recognizer, detector: VehicleDetector) -> None: self._inner = inner self._detector = detector self.vehicle_error: str | None = None @property def model_version(self) -> str: return f"{self._inner.model_version}+{self._detector.model_version}" @property def ready(self) -> bool: return bool(self._inner.ready) @property def error(self) -> str | None: inner = getattr(self._inner, "error", None) det = getattr(self._detector, "error", None) or self.vehicle_error parts = [p for p in (inner, f"vehicle: {det}" if det else None) if p] return "; ".join(parts) if parts else None def analyze(self, image_bytes: bytes) -> AnalyzeResponse: started = time.perf_counter() res = self._inner.analyze(image_bytes) try: vehicle = self._detector.detect(image_bytes, res.plate.bbox if res.plate else None) except Exception as exc: # noqa: BLE001 - advisory stage, never fatal self.vehicle_error = f"{type(exc).__name__}: {exc}" vehicle = None took_ms = (time.perf_counter() - started) * 1000.0 return res.model_copy( update={"vehicle": vehicle, "model_version": self.model_version, "took_ms": took_ms} ) def build_recognizer(settings: Settings) -> Recognizer: """Factory: pick the recognizer from settings. Falls back to the stub if the real one can't load, so the service always comes up (with ready=False surfaced). The vehicle stage wraps whichever recognizer runs when a model path is configured.""" rec: Recognizer if settings.recognizer == "fast_alpr": rec = FastAlprRecognizer(settings) else: rec = StubRecognizer(settings) if settings.vehicle_model_path: detector = YoloxVehicleDetector( settings.vehicle_model_path, input_size=settings.vehicle_input_size, min_confidence=settings.vehicle_min_confidence, ) return WithVehicle(rec, detector) return rec