"""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 logging import time from pathlib import Path from typing import Protocol from .schemas import AnalyzeResponse, BBox, PlateResult from .settings import Settings from .vehicle import BodyTypeClassifier, RefinedVehicleDetector, VehicleDetector, YoloxVehicleDetector log = logging.getLogger("vision") 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: VehicleDetector = YoloxVehicleDetector( settings.vehicle_model_path, input_size=settings.vehicle_input_size, min_confidence=settings.vehicle_min_confidence, ) if settings.vehicle_classifier_path: if Path(settings.vehicle_classifier_path).is_file(): detector = RefinedVehicleDetector( detector, BodyTypeClassifier( settings.vehicle_classifier_path, min_confidence=settings.vehicle_classifier_min_confidence, ), ) else: log.info("no body-type classifier at %s — phase B off", settings.vehicle_classifier_path) return WithVehicle(rec, detector) return rec