parking_solution/apps/vision/vision_service/recognizer.py

"""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


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,
        )


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)."""
    if settings.recognizer == "fast_alpr":
        rec = FastAlprRecognizer(settings)
        return rec
    return StubRecognizer(settings)