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