feat(trainer): phase-B body-type classifier — trainer job on the collector host + the classifier stage on the booth

apps/trainer (parking-trainer): inspect / train / evaluate / publish. Reads the wash
collector's SQLite + crops read-only off its volume; time split (validation = newest
slice); thin classes dropped; damped class weights; `features` mode (frozen ImageNet
backbone, on-disk feature cache, seconds to retrain) and `finetune` mode (light
augmentation). CPU-only torch from PyTorch's wheel index. ONNX export checked against
the torch model; NO model file below the validation floor (exit 3, report still written);
exit 2 = not enough labels. `evaluate` scores a shipped model on labels reviewed after
training + the unlabelled pile; `publish` PUTs a version folder to a Gitea generic package.
Light core deps; the `train` extra is heavy — CI syncs without it, torch tests skip.

apps/vision: BodyTypeClassifier (bodytype.onnx + sidecar = the preprocessing contract:
crop margin, input size, RGB 0-255, normalisation inside the graph) and
RefinedVehicleDetector over YOLOX — refines only `car` or a class the classifier trained
on, min-confidence, `detector_class` on the result; path set but no file = phase B off
without an error; a broken file is a health detail. models/bodytype.version (tracked,
empty) pins the published version the Dockerfile fetches at build (BuildKit secret;
a pin that cannot be fetched fails the build). Verified: a trainer model gives identical
probabilities inside the vision service; both images built and smoke-tested.

Delivery: parking-trainer image in build-images.yml, the `trainer` compose profile on the
collector stack (CPU, read-only data, TRAINER_OUT), commented TRAINER_OUT/PUBLISH_TOKEN in
the wash-collector stack, .dockerignore for both Python contexts, trainer deps synced in CI.

Wiki: bodytype-classifier-training rewritten as built (+ one fleet model not per site,
secrets/access, where the crops live), opencv-anpr-service §Phase B, vision-review-outbox,
vision-service-packaging, fleet-deployment-komodo, index, log.

Claude-Session: https://claude.ai/code/session_01FWncR69HgGPuei1dLrW3cU

apps/vision/vision_service/recognizer.py

@@ -14,12 +14,16 @@ Adding a recognizer (e.g. a fine-tuned YOLO + PaddleOCR) = a new class here, no
 
 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 VehicleDetector, YoloxVehicleDetector
+from .vehicle import BodyTypeClassifier, RefinedVehicleDetector, VehicleDetector, YoloxVehicleDetector
+
+log = logging.getLogger("vision")
 
 
 class Recognizer(Protocol):
@@ -215,10 +219,21 @@ def build_recognizer(settings: Settings) -> Recognizer:
     else:
         rec = StubRecognizer(settings)
     if settings.vehicle_model_path:
-        detector = YoloxVehicleDetector(
+        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

apps/vision/vision_service/schemas.py

@@ -45,6 +45,9 @@ class VehicleResult(BaseModel):
     confidence: float | None = Field(default=None, ge=0.0, le=1.0)
     # The vehicle's box in frame pixels — the crop a reviewer sees / a classifier eats.
     bbox: BBox | None = None
+    # Phase B: the detector's coarse class when the body-type classifier ran on this crop
+    # (body_type is then the classifier's answer if confident, else the detector's).
+    detector_class: str | None = None
     make: str | None = None
     model: str | None = None
 

apps/vision/vision_service/settings.py

@@ -42,6 +42,14 @@ class Settings(BaseSettings):
     # site's own, stricter threshold before it FLAGS anything).
     vehicle_min_confidence: float = 0.4
 
+    # Phase B — the body-type classifier on the detector's crop (bodytype.onnx + its .json
+    # sidecar, produced by apps/trainer, baked into the image when
+    # models/bodytype.version pins a published version). Path set but NO file = the normal
+    # state before the first model ships: the stage is simply off (logged, not an error).
+    vehicle_classifier_path: str | None = None
+    # Below this probability the classifier's answer is dropped and the detector's stands.
+    vehicle_classifier_min_confidence: float = 0.6
+
 
 def get_settings() -> Settings:
     return Settings()

apps/vision/vision_service/vehicle.py

@@ -229,6 +229,13 @@ class YoloxVehicleDetector:
         frame = cv2.imdecode(np.frombuffer(image_bytes, dtype=np.uint8), cv2.IMREAD_COLOR)
         if frame is None:
             return None
+        return self.detect_frame(frame, plate)
+
+    def detect_frame(self, frame: Any, plate: BBox | None) -> VehicleResult | None:
+        """Same as detect() on an already-decoded BGR frame (the classifier stage decodes
+        once and shares it)."""
+        if self._session is None:
+            return None
         tensor, scale = letterbox(frame, self._size)
         raw = self._session.run(None, {self._input_name: tensor})[0][0]
         found = vehicles_from_output(raw, self._size, scale, self._min_confidence)
@@ -242,6 +249,169 @@ class YoloxVehicleDetector:
         return VehicleResult(body_type=best.body_type, confidence=round(best.confidence, 4), bbox=box)
 
 
+# ----------------------------------------------------------------------------------
+# Phase B: the body-type classifier on the detector's crop
+# ----------------------------------------------------------------------------------
+
+SIDECAR_FORMAT = "parking-bodytype/1"
+
+
+def crop_vehicle(frame: Any, box: BBox, plate: BBox | None, margin: float) -> Any:
+    """The detector's box + margin, plate blurred — the SAME cut the collector stores
+    (apps/server review-outbox.ts makeReviewCrop), so the classifier sees at the booth
+    what it was trained on. Returns a BGR array, or None when the box is degenerate."""
+    import cv2
+
+    h, w = frame.shape[:2]
+    mw = round((box.x2 - box.x1) * margin)
+    mh = round((box.y2 - box.y1) * margin)
+    left, top = max(0, box.x1 - mw), max(0, box.y1 - mh)
+    right, bottom = min(w, box.x2 + mw), min(h, box.y2 + mh)
+    if right - left < 8 or bottom - top < 8:
+        return None
+    crop = frame[top:bottom, left:right].copy()
+    if plate is not None:
+        pad = round(max(plate.x2 - plate.x1, plate.y2 - plate.y1) * 0.25)
+        pl, pt = max(0, plate.x1 - pad - left), max(0, plate.y1 - pad - top)
+        pr, pb = min(right - left, plate.x2 + pad - left), min(bottom - top, plate.y2 + pad - top)
+        if pr - pl >= 2 and pb - pt >= 2:
+            sigma = max(6, round((pr - pl) / 6))
+            crop[pt:pb, pl:pr] = cv2.GaussianBlur(crop[pt:pb, pl:pr], (0, 0), sigma)
+    return crop
+
+
+class BodyTypeClassifier:
+    """`bodytype.onnx` + its `bodytype.json` sidecar (written by apps/trainer). The sidecar
+    carries the preprocessing contract — class list, input size, crop margin — and the graph
+    normalises internally, so this side only cuts, resizes (INTER_AREA, like the trainer)
+    and feeds raw RGB 0–255. Load failure → `error`, the stage yields nothing."""
+
+    def __init__(self, model_path: str, min_confidence: float = 0.6) -> None:
+        import json
+
+        self._path = Path(model_path)
+        self.min_confidence = min_confidence
+        self._session = None
+        self._input_name = "image"
+        self._error: str | None = None
+        self.classes: list[str] = []
+        self.version = "?"
+        self.input_size = 224
+        self.crop_margin = 0.08
+        try:
+            side = json.loads(self._path.with_suffix(".json").read_text())
+            if side.get("format") != SIDECAR_FORMAT:
+                raise ValueError(f"unknown sidecar format {side.get('format')!r}")
+            self.classes = [str(c) for c in side["classes"]]
+            self.version = str(side.get("version", "?"))
+            self.input_size = int(side.get("input_size", 224))
+            self.crop_margin = float(side.get("crop_margin", 0.08))
+            import onnxruntime as ort
+
+            opts = ort.SessionOptions()
+            opts.intra_op_num_threads = 2
+            self._session = ort.InferenceSession(
+                str(self._path), sess_options=opts, providers=["CPUExecutionProvider"]
+            )
+            self._input_name = self._session.get_inputs()[0].name
+        except Exception as exc:  # noqa: BLE001 - not-ready, never fatal
+            self._error = f"{type(exc).__name__}: {exc}"
+
+    @property
+    def model_version(self) -> str:
+        return f"bodytype:{self.version}"
+
+    @property
+    def ready(self) -> bool:
+        return self._session is not None
+
+    @property
+    def error(self) -> str | None:
+        return self._error
+
+    def classify(self, frame: Any, box: BBox, plate: BBox | None) -> tuple[str, float] | None:
+        """(class, probability) for the vehicle in `box`, or None when nothing could be cut."""
+        if self._session is None:
+            return None
+        import cv2
+        import numpy as np
+
+        crop = crop_vehicle(frame, box, plate, self.crop_margin)
+        if crop is None:
+            return None
+        resized = cv2.resize(crop, (self.input_size, self.input_size), interpolation=cv2.INTER_AREA)
+        rgb = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB)
+        x = np.ascontiguousarray(rgb.transpose(2, 0, 1)[None].astype(np.float32))
+        logits = self._session.run(None, {self._input_name: x})[0][0]
+        z = logits - logits.max()
+        p = np.exp(z) / np.exp(z).sum()
+        i = int(p.argmax())
+        return self.classes[i], float(p[i])
+
+
+class RefinedVehicleDetector:
+    """Detector + classifier. The detector finds the vehicle (and picks WHICH one); when its
+    class is `car` — or one the classifier was trained on — the classifier's answer replaces
+    it if confident enough, else the detector's stands. A truck or bus the classifier has
+    never seen is left alone: its softmax on an unknown thing means nothing."""
+
+    def __init__(self, detector: Any, classifier: BodyTypeClassifier) -> None:
+        self._detector = detector
+        self._classifier = classifier
+        self.stage_error: str | None = None
+
+    @property
+    def model_version(self) -> str:
+        return f"{self._detector.model_version}+{self._classifier.model_version}"
+
+    @property
+    def ready(self) -> bool:
+        return bool(getattr(self._detector, "ready", True))
+
+    @property
+    def error(self) -> str | None:
+        parts = [
+            getattr(self._detector, "error", None),
+            f"classifier: {self._classifier.error}" if self._classifier.error else None,
+            f"classifier: {self.stage_error}" if self.stage_error else None,
+        ]
+        kept = [p for p in parts if p]
+        return "; ".join(kept) if kept else None
+
+    def detect(self, image_bytes: bytes, plate: BBox | None) -> VehicleResult | None:
+        import cv2
+        import numpy as np
+
+        frame = cv2.imdecode(np.frombuffer(image_bytes, dtype=np.uint8), cv2.IMREAD_COLOR)
+        if frame is None:
+            return None
+        detect_frame = getattr(self._detector, "detect_frame", None)
+        base: VehicleResult | None = (
+            detect_frame(frame, plate) if detect_frame else self._detector.detect(image_bytes, plate)
+        )
+        if base is None or base.bbox is None or not self._classifier.ready:
+            return base
+        if not (base.body_type == "car" or base.body_type in self._classifier.classes):
+            return base
+        try:
+            out = self._classifier.classify(frame, base.bbox, plate)
+        except Exception as exc:  # noqa: BLE001 - advisory stage, never fatal
+            self.stage_error = f"{type(exc).__name__}: {exc}"
+            return base
+        if out is None:
+            return base
+        body_type, confidence = out
+        if confidence < self._classifier.min_confidence:
+            return base.model_copy(update={"detector_class": base.body_type})
+        return base.model_copy(
+            update={
+                "body_type": body_type,
+                "confidence": round(confidence, 4),
+                "detector_class": base.body_type,
+            }
+        )
+
+
 def time_detect(
     detector: VehicleDetector, image_bytes: bytes, plate: BBox | None
 ) -> tuple[VehicleResult | None, float]: