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