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: ...


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:
            ocr = getattr(r, "ocr", None)
            det = getattr(r, "detection", None)
            text = getattr(ocr, "text", None)
            if not text:
                continue
            conf = float(getattr(ocr, "confidence", 0.0) or 0.0)
            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)
                )
            plates.append(PlateResult(text=text, confidence=conf, bbox=bbox))

        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)