From 1089ee8c21e290909f2cc531f60d4d4092aee1c0 Mon Sep 17 00:00:00 2001 From: Nathan V Date: Wed, 15 Jul 2026 23:30:00 -0700 Subject: [PATCH] Remote detection backend improvements: Roboflow fix, resize, CPU fallback, coralapi - Fix RoboflowInferenceDetector request format for Inference Server 1.x (model_id + image belong in the JSON body; the old query-param form gets HTTP 422, which previously surfaced as silent zero-detection results) - Add opt-in client-side downscaling (roboflow.resize_max_px / coralapi.resize_max_px): the server resizes to model input anyway, so smaller uploads cut per-frame latency 3-4x at identical accuracy; predictions are rescaled to original frame coordinates - Remote detectors now raise DetectorUnavailableError on transport failures instead of returning empty detections; new FallbackDetector optionally falls back to local CPU inference (detection_fallback: local) - immediate switch when the backend never succeeded this job, after 3 consecutive failures otherwise, sticky per job, local model loaded lazily - ProcessingResult now records detection_backend / detection_model / detection_fallback (what actually ran), surfaced in API responses and as backend/model/fallback tags on InfluxDB metrics for per-model dashboards and fallback-rate alerting - New coralapi backend (https://github.com/nathan-v/coralapi) for Coral Edge TPU inference: multipart /v1/vision/detect, normalized-box scaling, label mapping with COCO-90 index fallback for label-less models - Docs: backend comparison/config for coralapi, detection_fallback setting; gitignore local benchmark artifacts --- .gitignore | 5 + README.md | 21 ++- configs/server.yaml | 17 ++ src/clockd/config.py | 34 +++- src/clockd/models.py | 4 + src/clockd/services/detector.py | 232 ++++++++++++++++++++++++--- src/clockd/services/metrics.py | 14 +- src/clockd/services/pipeline.py | 29 +++- tests/test_detector.py | 270 ++++++++++++++++++++++++++++++-- tests/test_detector_fallback.py | 121 ++++++++++++++ tests/test_metrics.py | 23 +++ 11 files changed, 733 insertions(+), 37 deletions(-) create mode 100644 tests/test_detector_fallback.py diff --git a/.gitignore b/.gitignore index 1265355..74b9b2a 100644 --- a/.gitignore +++ b/.gitignore @@ -231,3 +231,8 @@ CODEX.md .copilot/ graphify-out/ *.pt + +# Local working docs and benchmark artifacts (kept out of the repo) +*.local.md +*.local.py +sweep-results/ diff --git a/README.md b/README.md index b5beb73..f143de6 100644 --- a/README.md +++ b/README.md @@ -235,7 +235,8 @@ Server settings in `configs/server.yaml` or via `CLOCKD_` environment variables: |---------|---------|-------------| | `host` | `0.0.0.0` | Bind address | | `port` | `8000` | Bind port | -| `detection_backend` | `local` | `"local"`, `"roboflow"`, `"localai"`, or `"codeproject_ai"` | +| `detection_backend` | `local` | `"local"`, `"roboflow"`, `"localai"`, `"codeproject_ai"`, or `"coralapi"` | +| `detection_fallback` | `none` | `"local"` to fall back to local CPU inference when a remote backend is unreachable | | `model` | `yolo26n.pt` | Default YOLO model (local backend only) | | `confidence` | `0.3` | Detection confidence threshold | | `default_unit` | `mph` | Speed unit (`mph` or `kmh`) | @@ -286,11 +287,13 @@ All remote backends offload detection to an external server. Clockd sends each f | **[Roboflow Inference](https://github.com/roboflow/inference)** | YOLO v8/v10/v11/26, RF-DETR, YOLO-NAS | CUDA + TensorRT | No | Linux, macOS | Very active (daily commits) | GPU server, best model selection | | **[LocalAI](https://github.com/mudler/localai)** | RF-DETR | CUDA, ROCm, Vulkan | No | Linux, macOS, Windows | Very active (monthly releases) | Already running LocalAI for LLMs | | **[CodeProject.AI](https://github.com/codeproject/CodeProject.AI-Server)** | YOLOv5, v8, v11 | CUDA | Yes (YOLOv5 TFLite) | Windows, Linux, macOS | Slow (last release Dec 2024) | Raspberry Pi + Coral TPU | +| **[coralapi](https://github.com/nathan-v/coralapi)** | Edge TPU `.tflite` (SSD MobileNet, EfficientDet) | No (TPU only) | Yes (native) | Linux | Active | Dedicated Coral TPU, lowest power | **Notes:** - **Roboflow Inference** is the recommended remote backend. Broadest model support, actively maintained, no API key needed for pre-trained COCO models, and supports TensorRT quantization for maximum GPU throughput. Runs on Linux and macOS only (Docker-based). For GPU use, Linux with the NVIDIA Container Toolkit is recommended. - **LocalAI** only supports RF-DETR for object detection (not YOLO). It's a good choice if you already run LocalAI for other AI tasks. Supports the widest range of GPU vendors (NVIDIA, AMD, Intel). -- **CodeProject.AI** is the only backend with Coral TPU support, via its YOLOv5 TFLite module. This makes it a viable option for low-power deployments like a Raspberry Pi with a Coral accelerator. The project has slowed (volunteer-maintained, 18+ months since last release) but remains functional. +- **CodeProject.AI** supports Coral TPUs via its YOLOv5 TFLite module. The project has slowed (volunteer-maintained, 18+ months since last release) but remains functional. +- **coralapi** is a purpose-built Edge TPU inference server (sync/async API, per-request model selection, automatic model download, no GPU required). It serves quantized Edge TPU `.tflite` models — expect SSD-MobileNet-class accuracy, below the larger YOLO models, at single-digit-watt power draw. Detections return normalized boxes, so `coralapi.resize_max_px` costs no accuracy on this backend. - For most users, **local + `yolo26n.pt` on CPU** is fast enough. Remote backends are worth it for larger models on GPU, high-volume processing, or shared inference across multiple services. ##### Backend Configuration @@ -329,6 +332,20 @@ codeproject_ai: timeout: 30 ``` +**coralapi:** + +```yaml +detection_backend: "coralapi" + +coralapi: + url: "http://localhost:8000" + model: "ssd_mobilenet_v2_coco_quant_postprocess_edgetpu" + timeout: 30 + resize_max_px: 640 # optional; Edge TPU models run at ~300px anyway +``` + +Start the server (Coral Edge TPU required): see the [coralapi quickstart](https://github.com/nathan-v/coralapi#quickstart-docker). + ## Event Sources (NVR Integration) Clockd can automatically poll your NVR for vehicle detection events, download clips, and process them — no external automation needed. diff --git a/configs/server.yaml b/configs/server.yaml index 389758d..6fec046 100644 --- a/configs/server.yaml +++ b/configs/server.yaml @@ -8,6 +8,10 @@ host: "0.0.0.0" port: 8000 verbose: false detection_backend: "local" # "local", "roboflow", "localai", or "codeproject_ai" +# Optional: fall back to local CPU inference (using `model`) when a remote +# detection backend is unreachable. "none" (default) keeps the old behavior +# of logging and yielding empty detections. +# detection_fallback: "local" model: "yolo26n.pt" confidence: 0.3 default_unit: "mph" @@ -26,6 +30,19 @@ roboflow: url: "http://localhost:9001" model_id: "yolo11n-640" timeout: 30 + # Optional: downscale frames to this longest-side size before sending. + # The server resizes to the model input anyway; enabling this trades a + # little local CPU for much smaller uploads. Omit to send full frames. + # resize_max_px: 640 + +# coralapi Edge TPU server (https://github.com/nathan-v/coralapi) +coralapi: + url: "http://localhost:8000" + model: "ssd_mobilenet_v2_coco_quant_postprocess_edgetpu" + timeout: 30 + # Edge TPU SSD models run at 300x300; downscaling before upload saves + # bandwidth with no accuracy cost (boxes are returned normalized). + # resize_max_px: 640 localai: url: "http://localhost:8080" diff --git a/src/clockd/config.py b/src/clockd/config.py index 8b74de3..020ab89 100644 --- a/src/clockd/config.py +++ b/src/clockd/config.py @@ -7,7 +7,7 @@ import re import yaml -from pydantic import BaseModel, field_validator +from pydantic import BaseModel, Field, field_validator from pydantic_settings import BaseSettings, SettingsConfigDict SAFE_ID_RE = re.compile(r"^[a-zA-Z0-9_-]+$") @@ -148,6 +148,11 @@ class RoboflowInferenceConfig(BaseModel): url: str = "http://localhost:9001" model_id: str = "yolo11n-640" timeout: int = 30 + # Downscale frames so their longest side is at most this many pixels before + # sending to the inference server. The server resizes to the model input size + # anyway, so this trades local resize CPU for much smaller encode/transfer + # payloads. None sends full-resolution frames (pushes all work to the server). + resize_max_px: Optional[int] = Field(default=None, ge=64, le=8192) class LocalAIConfig(BaseModel): @@ -156,6 +161,18 @@ class LocalAIConfig(BaseModel): timeout: int = 30 +class CoralAPIConfig(BaseModel): + """coralapi Edge TPU inference server (https://github.com/nathan-v/coralapi).""" + + url: str = "http://localhost:8000" + model: str = "ssd_mobilenet_v2_coco_quant_postprocess_edgetpu" + timeout: int = 30 + # Same semantics as roboflow.resize_max_px: downscale before upload. + # Edge TPU SSD models run at 300x300, so full-res frames are pure + # transfer overhead; boxes come back normalized, so no accuracy cost. + resize_max_px: Optional[int] = Field(default=None, ge=64, le=8192) + + class UnifiProtectConfig(BaseModel): host: str = "" username: str = "" @@ -232,7 +249,19 @@ def settings_customise_sources( host: str = "0.0.0.0" port: int = 8000 verbose: bool = False # enable detailed logging of processing results - detection_backend: str = "local" # "local", "roboflow", "localai", or "codeproject_ai" + detection_backend: str = "local" # "local", "roboflow", "localai", "codeproject_ai", "coralapi" + # "local": if a remote detection_backend is unreachable, fall back to + # local CPU inference (using `model`) for the rest of the job. "none": + # unreachable backends yield empty detections (logged) as before. + detection_fallback: str = "none" + + @field_validator("detection_fallback") + @classmethod + def _validate_detection_fallback(cls, v: str) -> str: + if v not in ("none", "local"): + raise ValueError("detection_fallback must be 'none' or 'local'") + return v + model: str = "yolo26n.pt" # validated at startup, must be in ALLOWED_MODELS @field_validator("model") @@ -253,6 +282,7 @@ def validate_model(cls, v: str) -> str: codeproject_ai: CodeProjectAIConfig = CodeProjectAIConfig() roboflow: RoboflowInferenceConfig = RoboflowInferenceConfig() localai: LocalAIConfig = LocalAIConfig() + coralapi: CoralAPIConfig = CoralAPIConfig() metrics: MetricsConfig = MetricsConfig() event_sources: dict[str, UnifiEventSourceConfig] = {} diff --git a/src/clockd/models.py b/src/clockd/models.py index 1f3356c..aa3ded7 100644 --- a/src/clockd/models.py +++ b/src/clockd/models.py @@ -36,6 +36,10 @@ class ProcessingResult(BaseModel): vehicles_filtered: int = 0 # tracks excluded by min_detections or speed_range processing_time_s: float warnings: list[str] = [] + # what actually performed detection (reflects fallback, not just config) + detection_backend: str = "" + detection_model: str = "" + detection_fallback: bool = False class JobStatus(str, Enum): diff --git a/src/clockd/services/detector.py b/src/clockd/services/detector.py index 175bee6..b30fb13 100644 --- a/src/clockd/services/detector.py +++ b/src/clockd/services/detector.py @@ -3,11 +3,12 @@ import base64 import logging import urllib.error +import urllib.parse import urllib.request from abc import ABC, abstractmethod from json import dumps as json_dumps from json import loads as json_loads -from typing import Optional +from typing import Callable, Optional import cv2 import numpy as np @@ -25,6 +26,10 @@ def close(self) -> None: pass +class DetectorUnavailableError(RuntimeError): + """The remote detection backend could not be reached.""" + + class LocalDetector(Detector): """Detection using a local YOLO model via ultralytics.""" @@ -77,8 +82,7 @@ def detect(self, frame: np.ndarray) -> sv.Detections: with urllib.request.urlopen(req, timeout=self._timeout) as resp: data = json_loads(resp.read()) except (urllib.error.URLError, OSError) as exc: - logger.warning("CodeProject.AI detection failed: %s", exc) - return sv.Detections.empty() + raise DetectorUnavailableError(f"CodeProject.AI detection failed: {exc}") from exc if not data.get("success") or not data.get("predictions"): return sv.Detections.empty() @@ -131,13 +135,28 @@ def __init__( model_id: str = "yolo11n-640", confidence: float = 0.3, timeout: int = 30, + resize_max_px: Optional[int] = None, ) -> None: self._url = url.rstrip("/") self._model_id = model_id self._confidence = confidence self._timeout = timeout + self._resize_max_px = resize_max_px def detect(self, frame: np.ndarray) -> sv.Detections: + # Optionally downscale before encoding; predictions come back in the + # sent image's pixel space, so scale them back up afterwards. + scale = 1.0 + if self._resize_max_px: + long_side = max(frame.shape[0], frame.shape[1]) + if long_side > self._resize_max_px: + scale = self._resize_max_px / long_side + frame = cv2.resize( + frame, + (round(frame.shape[1] * scale), round(frame.shape[0] * scale)), + interpolation=cv2.INTER_AREA, + ) + ok, buf = cv2.imencode(".jpg", frame) if not ok: return sv.Detections.empty() @@ -146,15 +165,13 @@ def detect(self, frame: np.ndarray) -> sv.Detections: payload = json_dumps( { - "type": "base64", - "value": img_b64, + "model_id": self._model_id, + "image": {"type": "base64", "value": img_b64}, "confidence": self._confidence, } ).encode() url = f"{self._url}/infer/object_detection" - # Model ID goes in query param - url += f"?model_id={self._model_id}" req = urllib.request.Request(url, data=payload, method="POST") req.add_header("Content-Type", "application/json") @@ -162,8 +179,7 @@ def detect(self, frame: np.ndarray) -> sv.Detections: with urllib.request.urlopen(req, timeout=self._timeout) as resp: data = json_loads(resp.read()) except (urllib.error.URLError, OSError) as exc: - logger.warning("Roboflow Inference detection failed: %s", exc) - return sv.Detections.empty() + raise DetectorUnavailableError(f"Roboflow Inference detection failed: {exc}") from exc predictions = data.get("predictions", []) if not predictions: @@ -179,10 +195,10 @@ def detect(self, frame: np.ndarray) -> sv.Detections: if coco_id is None: continue conf = pred.get("confidence", 0.0) - cx = pred.get("x", 0) - cy = pred.get("y", 0) - w = pred.get("width", 0) - h = pred.get("height", 0) + cx = pred.get("x", 0) / scale + cy = pred.get("y", 0) / scale + w = pred.get("width", 0) / scale + h = pred.get("height", 0) / scale boxes.append([cx - w / 2, cy - h / 2, cx + w / 2, cy + h / 2]) class_ids.append(coco_id) confidences.append(conf) @@ -235,8 +251,7 @@ def detect(self, frame: np.ndarray) -> sv.Detections: with urllib.request.urlopen(req, timeout=self._timeout) as resp: data = json_loads(resp.read()) except (urllib.error.URLError, OSError) as exc: - logger.warning("LocalAI detection failed: %s", exc) - return sv.Detections.empty() + raise DetectorUnavailableError(f"LocalAI detection failed: {exc}") from exc detections = data if isinstance(data, list) else data.get("detections", []) if not detections: @@ -270,6 +285,158 @@ def detect(self, frame: np.ndarray) -> sv.Detections: ) +# Vehicle classes in the 0-indexed COCO-90 scheme used by Coral zoo detection +# models (car, motorcycle, bus, truck) — values equal our COCO-80 ids. +CORAL_INDEX_TO_COCO = {2: 2, 3: 3, 5: 5, 7: 7} + + +class CoralAPIDetector(Detector): + """Detection using a coralapi Edge TPU inference server. + + https://github.com/nathan-v/coralapi — POST /v1/vision/detect with a + multipart image; detections come back with normalized [ymin, xmin, + ymax, xmax] boxes (TFLite SSD convention) scaled here against the + original frame, so optional client-side downscaling needs no box + rescaling bookkeeping. + """ + + def __init__( + self, + url: str = "http://localhost:8000", + model: str = "ssd_mobilenet_v2_coco_quant_postprocess_edgetpu", + confidence: float = 0.3, + timeout: int = 30, + resize_max_px: Optional[int] = None, + ) -> None: + self._url = url.rstrip("/") + self._model = model + self._confidence = confidence + self._timeout = timeout + self._resize_max_px = resize_max_px + + def detect(self, frame: np.ndarray) -> sv.Detections: + orig_h, orig_w = frame.shape[0], frame.shape[1] + if self._resize_max_px and max(orig_h, orig_w) > self._resize_max_px: + scale = self._resize_max_px / max(orig_h, orig_w) + frame = cv2.resize( + frame, + (round(orig_w * scale), round(orig_h * scale)), + interpolation=cv2.INTER_AREA, + ) + + ok, buf = cv2.imencode(".jpg", frame) + if not ok: + return sv.Detections.empty() + + boundary = "----ClockdBoundary" + body = bytearray() + body.extend(f"--{boundary}\r\n".encode()) + body.extend(b'Content-Disposition: form-data; name="file"; filename="frame.jpg"\r\n') + body.extend(b"Content-Type: image/jpeg\r\n\r\n") + body.extend(buf.tobytes()) + body.extend(f"\r\n--{boundary}--\r\n".encode()) + + params = urllib.parse.urlencode({"model": self._model, "threshold": self._confidence}) + url = f"{self._url}/v1/vision/detect?{params}" + req = urllib.request.Request(url, data=bytes(body), method="POST") + req.add_header("Content-Type", f"multipart/form-data; boundary={boundary}") + + try: + with urllib.request.urlopen(req, timeout=self._timeout) as resp: + data = json_loads(resp.read()) + except (urllib.error.URLError, OSError) as exc: + raise DetectorUnavailableError(f"coralapi detection failed: {exc}") from exc + + boxes = [] + class_ids = [] + confidences = [] + + for det in data.get("results", []): + label = (det.get("label") or "").lower() + coco_id = LABEL_TO_COCO.get(label) + if coco_id is None and not label: + # No labels file on the server: fall back to the class index. + # Coral zoo detection models use 0-indexed COCO-90, whose + # vehicle ids happen to equal our COCO-80 ids. A present but + # unrecognized label (e.g. "dog") is still skipped above. + coco_id = CORAL_INDEX_TO_COCO.get(det.get("index")) + if coco_id is None: + continue + box = det.get("box") or [0, 0, 0, 0] + ymin, xmin, ymax, xmax = box + boxes.append([xmin * orig_w, ymin * orig_h, xmax * orig_w, ymax * orig_h]) + class_ids.append(coco_id) + confidences.append(det.get("score", 0.0)) + + if not boxes: + return sv.Detections.empty() + + return sv.Detections( + xyxy=np.array(boxes, dtype=np.float32), + class_id=np.array(class_ids, dtype=int), + confidence=np.array(confidences, dtype=np.float32), + ) + + +class FallbackDetector(Detector): + """Wraps a remote detector, optionally falling back to local CPU inference. + + Policy: if the remote backend has never succeeded this job, the first + failure switches immediately (a down host shouldn't cost N timeouts); + after it has succeeded, MAX_CONSECUTIVE_FAILURES consecutive failures + trigger the switch. Once switched, the fallback is used for the rest of + the job. Without a fallback factory, failures log a warning and yield + empty detections (the pre-fallback behavior). + + The fallback detector is built lazily — LocalDetector loads YOLO weights + in __init__, which healthy remote jobs must not pay for. + """ + + MAX_CONSECUTIVE_FAILURES = 3 + + def __init__( + self, + primary: Detector, + fallback_factory: Optional[Callable[[], Detector]] = None, + ) -> None: + self._primary = primary + self._fallback_factory = fallback_factory + self._fallback: Optional[Detector] = None + self._consecutive_failures = 0 + self._ever_succeeded = False + self.using_fallback = False + self.fallback_reason: Optional[str] = None + + def detect(self, frame: np.ndarray) -> sv.Detections: + if self.using_fallback: + assert self._fallback is not None + return self._fallback.detect(frame) + try: + detections = self._primary.detect(frame) + except DetectorUnavailableError as exc: + self._consecutive_failures += 1 + should_switch = self._fallback_factory is not None and ( + not self._ever_succeeded + or self._consecutive_failures >= self.MAX_CONSECUTIVE_FAILURES + ) + if should_switch: + logger.warning( + "Remote detection backend unavailable (%s); " + "falling back to local CPU inference for the rest of this job", + exc, + ) + assert self._fallback_factory is not None + self._fallback = self._fallback_factory() + self.using_fallback = True + self.fallback_reason = str(exc) + return self._fallback.detect(frame) + logger.warning("Detection failed (%s); returning empty detections", exc) + return sv.Detections.empty() + self._consecutive_failures = 0 + self._ever_succeeded = True + return detections + + def create_detector( backend: str, model_name: str, @@ -279,28 +446,51 @@ def create_detector( roboflow_url: Optional[str] = None, roboflow_model_id: Optional[str] = None, roboflow_timeout: int = 30, + roboflow_resize_max_px: Optional[int] = None, localai_url: Optional[str] = None, localai_model: Optional[str] = None, localai_timeout: int = 30, + coralapi_url: Optional[str] = None, + coralapi_model: Optional[str] = None, + coralapi_timeout: int = 30, + coralapi_resize_max_px: Optional[int] = None, + fallback: str = "none", ) -> Detector: + remote: Optional[Detector] = None if backend == "codeproject_ai": - return CodeProjectAIDetector( + remote = CodeProjectAIDetector( url=codeproject_url or "http://localhost:32168", confidence=confidence, timeout=codeproject_timeout, ) - if backend == "roboflow": - return RoboflowInferenceDetector( + elif backend == "roboflow": + remote = RoboflowInferenceDetector( url=roboflow_url or "http://localhost:9001", model_id=roboflow_model_id or "yolo11n-640", confidence=confidence, timeout=roboflow_timeout, + resize_max_px=roboflow_resize_max_px, ) - if backend == "localai": - return LocalAIDetector( + elif backend == "localai": + remote = LocalAIDetector( url=localai_url or "http://localhost:8080", model=localai_model or "rfdetr-base", confidence=confidence, timeout=localai_timeout, ) - return LocalDetector(model_name=model_name, confidence=confidence) + elif backend == "coralapi": + remote = CoralAPIDetector( + url=coralapi_url or "http://localhost:8000", + model=coralapi_model or "ssd_mobilenet_v2_coco_quant_postprocess_edgetpu", + confidence=confidence, + timeout=coralapi_timeout, + resize_max_px=coralapi_resize_max_px, + ) + if remote is None: + return LocalDetector(model_name=model_name, confidence=confidence) + factory = ( + (lambda: LocalDetector(model_name=model_name, confidence=confidence)) + if fallback == "local" + else None + ) + return FallbackDetector(remote, fallback_factory=factory) diff --git a/src/clockd/services/metrics.py b/src/clockd/services/metrics.py index 277057f..5bcc4ca 100644 --- a/src/clockd/services/metrics.py +++ b/src/clockd/services/metrics.py @@ -90,9 +90,19 @@ def _write_influxdb_service_point( def _build_line_protocol(self, result: ProcessingResult, measurement: str) -> str: lines = [] ts_ns = int(time.time() * 1e9) + # what actually ran, so dashboards can split by model and CPU-vs-GPU + # backend (and alert on fallback frequency) + detect_tags = ( + f"backend={_escape_tag(result.detection_backend or 'unknown')}" + f",model={_escape_tag(result.detection_model or 'unknown')}" + f",fallback={str(result.detection_fallback).lower()}" + ) for v in result.vehicles: cam_id = _escape_tag(result.camera_id) - tags = f"{measurement},camera_id={cam_id},track_id={v.track_id},unit={v.unit}" + tags = ( + f"{measurement},camera_id={cam_id},track_id={v.track_id}," + f"unit={v.unit},{detect_tags}" + ) fields = ( f"speed_avg={v.speed_avg}," f"speed_min={v.speed_min}," @@ -102,7 +112,7 @@ def _build_line_protocol(self, result: ProcessingResult, measurement: str) -> st ) lines.append(f"{tags} {fields} {ts_ns}") # Processing summary point - tags = f"processing_summary,camera_id={_escape_tag(result.camera_id)}" + tags = f"processing_summary,camera_id={_escape_tag(result.camera_id)},{detect_tags}" fields = ( f"vehicle_count={len(result.vehicles)}i," f"processing_time_s={result.processing_time_s}," diff --git a/src/clockd/services/pipeline.py b/src/clockd/services/pipeline.py index c9f13fe..31fc1c8 100644 --- a/src/clockd/services/pipeline.py +++ b/src/clockd/services/pipeline.py @@ -11,7 +11,7 @@ from clockd.config import CameraConfig, ServerConfig from clockd.models import ProcessingResult, VehicleResult -from clockd.services.detector import create_detector +from clockd.services.detector import FallbackDetector, create_detector from clockd.services.view_transformer import ViewTransformer from clockd.utils.units import convert_speed, mph_to_ms from clockd.utils.video import MAX_FRAMES, validate_video @@ -88,11 +88,29 @@ def process_video( roboflow_url=server_cfg.roboflow.url, roboflow_model_id=server_cfg.roboflow.model_id, roboflow_timeout=server_cfg.roboflow.timeout, + roboflow_resize_max_px=server_cfg.roboflow.resize_max_px, localai_url=server_cfg.localai.url, localai_model=server_cfg.localai.model, localai_timeout=server_cfg.localai.timeout, + coralapi_url=server_cfg.coralapi.url, + coralapi_model=server_cfg.coralapi.model, + coralapi_timeout=server_cfg.coralapi.timeout, + coralapi_resize_max_px=server_cfg.coralapi.resize_max_px, + fallback=server_cfg.detection_fallback, ) + detection_backend = server_cfg.detection_backend + if detection_backend == "roboflow": + detection_model = server_cfg.roboflow.model_id + elif detection_backend == "localai": + detection_model = server_cfg.localai.model + elif detection_backend == "coralapi": + detection_model = server_cfg.coralapi.model + elif detection_backend == "codeproject_ai": + detection_model = "server-default" + else: + detection_model = model_name + if verbose: logger.info( "Detector initialized: backend=%s model=%s confidence=%.2f", @@ -286,6 +304,12 @@ def process_video( v.last_seen_frame, ) + fallback_used = isinstance(detector, FallbackDetector) and detector.using_fallback + if fallback_used: + detection_backend = "local" + detection_model = model_name + warnings.append("Remote detection backend unavailable; fell back to local CPU inference") + return ProcessingResult( camera_id=camera.camera_id, video_filename=video_path.rsplit("/", 1)[-1], @@ -297,4 +321,7 @@ def process_video( vehicles_filtered=filtered_count, processing_time_s=processing_time, warnings=warnings, + detection_backend=detection_backend, + detection_model=detection_model, + detection_fallback=fallback_used, ) diff --git a/tests/test_detector.py b/tests/test_detector.py index e5d8689..5fe39f5 100644 --- a/tests/test_detector.py +++ b/tests/test_detector.py @@ -1,13 +1,19 @@ from __future__ import annotations +import base64 import json from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread +import cv2 import numpy as np +import pytest from clockd.services.detector import ( CodeProjectAIDetector, + CoralAPIDetector, + DetectorUnavailableError, + FallbackDetector, LocalAIDetector, LocalDetector, RoboflowInferenceDetector, @@ -27,7 +33,8 @@ def test_create_detector_codeproject(): confidence=0.4, codeproject_url="http://localhost:32168", ) - assert isinstance(det, CodeProjectAIDetector) + assert isinstance(det, FallbackDetector) + assert isinstance(det._primary, CodeProjectAIDetector) def test_create_detector_roboflow(): @@ -38,7 +45,8 @@ def test_create_detector_roboflow(): roboflow_url="http://localhost:9001", roboflow_model_id="yolo11n-640", ) - assert isinstance(det, RoboflowInferenceDetector) + assert isinstance(det, FallbackDetector) + assert isinstance(det._primary, RoboflowInferenceDetector) def test_create_detector_localai(): @@ -49,7 +57,8 @@ def test_create_detector_localai(): localai_url="http://localhost:8080", localai_model="rfdetr-base", ) - assert isinstance(det, LocalAIDetector) + assert isinstance(det, FallbackDetector) + assert isinstance(det._primary, LocalAIDetector) def test_codeproject_detector_success(): @@ -145,8 +154,8 @@ def log_message(self, *args): def test_codeproject_detector_unreachable(): det = CodeProjectAIDetector(url="http://127.0.0.1:1", confidence=0.3, timeout=1) - detections = det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) - assert len(detections) == 0 # should not raise + with pytest.raises(DetectorUnavailableError): + det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) def test_roboflow_detector_success(): @@ -159,8 +168,13 @@ def test_roboflow_detector_success(): ] } + received = {} + class Handler(BaseHTTPRequestHandler): def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + received["path"] = self.path + received["body"] = json.loads(self.rfile.read(length)) resp = json.dumps(response_data).encode() self.send_response(200) self.send_header("Content-Type", "application/json") @@ -176,12 +190,20 @@ def log_message(self, *args): t = Thread(target=server.handle_request, daemon=True) t.start() - det = RoboflowInferenceDetector(url=f"http://127.0.0.1:{port}", confidence=0.3) + det = RoboflowInferenceDetector( + url=f"http://127.0.0.1:{port}", model_id="yolo26n-640", confidence=0.3 + ) detections = det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) t.join(timeout=5) server.server_close() + # Inference server >=1.x requires model_id and image in the JSON body + assert received["path"] == "/infer/object_detection" + assert received["body"]["model_id"] == "yolo26n-640" + assert received["body"]["image"]["type"] == "base64" + assert received["body"]["image"]["value"] + # car + bus (dog is not in LABEL_TO_COCO) assert len(detections) == 2 assert 2 in detections.class_id # car @@ -191,10 +213,68 @@ def log_message(self, *args): np.testing.assert_allclose(detections.xyxy[car_idx], [150, 260, 250, 340], atol=1) +def test_roboflow_detector_resize_max_px(): + # Prediction in the downscaled image's pixel space (1280x720 -> 640x360) + response_data = { + "predictions": [ + {"class": "car", "confidence": 0.9, "x": 100, "y": 150, "width": 50, "height": 40}, + ] + } + received = {} + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + received["body"] = json.loads(self.rfile.read(length)) + resp = json.dumps(response_data).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(resp))) + self.end_headers() + self.wfile.write(resp) + + def log_message(self, *args): + pass + + server = HTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + t = Thread(target=server.handle_request, daemon=True) + t.start() + + det = RoboflowInferenceDetector( + url=f"http://127.0.0.1:{port}", confidence=0.3, resize_max_px=640 + ) + detections = det.detect(np.zeros((720, 1280, 3), dtype=np.uint8)) + + t.join(timeout=5) + server.server_close() + + # The uploaded frame must have been downscaled to 640x360 + img = cv2.imdecode( + np.frombuffer(base64.b64decode(received["body"]["image"]["value"]), np.uint8), + cv2.IMREAD_COLOR, + ) + assert img.shape[:2] == (360, 640) + + # Boxes come back rescaled to the original 1280x720 space (scale=0.5) + assert len(detections) == 1 + np.testing.assert_allclose(detections.xyxy[0], [150, 260, 250, 340], atol=1) + + +def test_roboflow_detector_resize_noop_when_frame_smaller(): + # Frame smaller than resize_max_px is sent as-is (resize path must not + # crash on small frames; unreachable server then raises) + det = RoboflowInferenceDetector( + url="http://127.0.0.1:1", confidence=0.3, resize_max_px=640, timeout=1 + ) + with pytest.raises(DetectorUnavailableError): + det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) + + def test_roboflow_detector_unreachable(): det = RoboflowInferenceDetector(url="http://127.0.0.1:1", confidence=0.3, timeout=1) - detections = det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) - assert len(detections) == 0 + with pytest.raises(DetectorUnavailableError): + det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) def test_localai_detector_success(): @@ -244,5 +324,177 @@ def log_message(self, *args): def test_localai_detector_unreachable(): det = LocalAIDetector(url="http://127.0.0.1:1", confidence=0.3, timeout=1) + with pytest.raises(DetectorUnavailableError): + det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) + + +def test_create_detector_coralapi(): + det = create_detector( + backend="coralapi", + model_name="yolo11n.pt", + confidence=0.3, + coralapi_url="http://localhost:8000", + coralapi_model="ssd_mobilenet_v2_coco_quant_postprocess_edgetpu", + ) + assert isinstance(det, FallbackDetector) + assert isinstance(det._primary, CoralAPIDetector) + + +def test_coralapi_detector_success(): + # coralapi returns normalized [ymin, xmin, ymax, xmax] boxes + response_data = { + "model": "ssd_mobilenet_v2_coco_quant_postprocess_edgetpu", + "results": [ + {"box": [0.25, 0.125, 0.75, 0.375], "index": 2, "label": "car", "score": 0.9}, + {"box": [0.1, 0.5, 0.4, 0.9], "index": 5, "label": "bus", "score": 0.8}, + {"box": [0.0, 0.0, 0.2, 0.2], "index": 17, "label": "dog", "score": 0.95}, + {"box": [0.3, 0.3, 0.6, 0.6], "index": 99, "label": None, "score": 0.7}, + ], + } + received = {} + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + received["path"] = self.path + received["body"] = self.rfile.read(length) + received["content_type"] = self.headers.get("Content-Type", "") + resp = json.dumps(response_data).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(resp))) + self.end_headers() + self.wfile.write(resp) + + def log_message(self, *args): + pass + + server = HTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + t = Thread(target=server.handle_request, daemon=True) + t.start() + + det = CoralAPIDetector( + url=f"http://127.0.0.1:{port}", + model="ssd_mobilenet_v2_coco_quant_postprocess_edgetpu", + confidence=0.4, + ) detections = det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) - assert len(detections) == 0 + + t.join(timeout=5) + server.server_close() + + # model + threshold go in the query string; image is a multipart "file" field + assert received["path"].startswith("/v1/vision/detect?") + assert "model=ssd_mobilenet_v2_coco_quant_postprocess_edgetpu" in received["path"] + assert "threshold=0.4" in received["path"] + assert "multipart/form-data" in received["content_type"] + assert b'name="file"' in received["body"] + + # car + bus; dog is not a vehicle, null label skipped + assert len(detections) == 2 + assert 2 in detections.class_id + assert 5 in detections.class_id + # normalized box -> pixel xyxy against the 640x480 frame: + # car [ymin .25, xmin .125, ymax .75, xmax .375] -> [80, 120, 240, 360] + car_idx = list(detections.class_id).index(2) + np.testing.assert_allclose(detections.xyxy[car_idx], [80, 120, 240, 360], atol=1) + + +def test_coralapi_detector_resize_boxes_stay_in_original_space(): + response_data = { + "model": "m", + "results": [{"box": [0.5, 0.5, 1.0, 1.0], "index": 2, "label": "car", "score": 0.9}], + } + received = {} + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + length = int(self.headers.get("Content-Length", 0)) + received["body"] = self.rfile.read(length) + resp = json.dumps(response_data).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(resp))) + self.end_headers() + self.wfile.write(resp) + + def log_message(self, *args): + pass + + server = HTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + t = Thread(target=server.handle_request, daemon=True) + t.start() + + det = CoralAPIDetector(url=f"http://127.0.0.1:{port}", confidence=0.3, resize_max_px=640) + detections = det.detect(np.zeros((720, 1280, 3), dtype=np.uint8)) + + t.join(timeout=5) + server.server_close() + + # uploaded image was downscaled (multipart jpeg smaller than raw frame) + start = received["body"].index(b"\r\n\r\n", received["body"].index(b'name="file"')) + 4 + jpeg = received["body"][start : received["body"].rindex(b"\r\n----")] + img = cv2.imdecode(np.frombuffer(jpeg, np.uint8), cv2.IMREAD_COLOR) + assert img.shape[:2] == (360, 640) + + # normalized boxes scale against the ORIGINAL 1280x720 frame + np.testing.assert_allclose(detections.xyxy[0], [640, 360, 1280, 720], atol=1) + + +def test_coralapi_detector_unreachable(): + det = CoralAPIDetector(url="http://127.0.0.1:1", confidence=0.3, timeout=1) + with pytest.raises(DetectorUnavailableError): + det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) + + +def test_coralapi_detector_index_fallback_when_no_labels(): + # Zoo models without a labels file return label=null; vehicles must be + # recovered via the COCO-90 index. Known-but-non-vehicle labels stay skipped. + response_data = { + "model": "m", + "results": [ + {"box": [0.1, 0.1, 0.5, 0.5], "index": 2, "label": None, "score": 0.6}, # car by index + { + "box": [0.2, 0.2, 0.6, 0.6], + "index": 7, + "label": None, + "score": 0.5, + }, # truck by index + {"box": [0.0, 0.0, 0.9, 0.9], "index": 6, "label": None, "score": 0.5}, # train -> skip + { + "box": [0.3, 0.3, 0.7, 0.7], + "index": 2, + "label": "dog", + "score": 0.9, + }, # labeled non-vehicle -> skip + ], + } + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + resp = json.dumps(response_data).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(resp))) + self.end_headers() + self.wfile.write(resp) + + def log_message(self, *args): + pass + + server = HTTPServer(("127.0.0.1", 0), Handler) + port = server.server_address[1] + t = Thread(target=server.handle_request, daemon=True) + t.start() + + det = CoralAPIDetector(url=f"http://127.0.0.1:{port}", confidence=0.3) + detections = det.detect(np.zeros((480, 640, 3), dtype=np.uint8)) + + t.join(timeout=5) + server.server_close() + + assert len(detections) == 2 + assert 2 in detections.class_id # car via index fallback + assert 7 in detections.class_id # truck via index fallback diff --git a/tests/test_detector_fallback.py b/tests/test_detector_fallback.py new file mode 100644 index 0000000..16c90f4 --- /dev/null +++ b/tests/test_detector_fallback.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +import numpy as np +import pytest + +from clockd.services.detector import ( + Detector, + DetectorUnavailableError, + FallbackDetector, +) + +FRAME = np.zeros((480, 640, 3), dtype=np.uint8) + + +class _StubDetector(Detector): + """Scripted detector: each call pops the next behavior. + + "ok" returns a sentinel detections object; "fail" raises + DetectorUnavailableError. The last behavior repeats when exhausted. + """ + + def __init__(self, *script: str, sentinel=None): + import supervision as sv + + self._script = list(script) + self.calls = 0 + self.sentinel = sentinel if sentinel is not None else sv.Detections.empty() + + def detect(self, frame): + self.calls += 1 + behavior = self._script.pop(0) if len(self._script) > 1 else self._script[0] + if behavior == "fail": + raise DetectorUnavailableError("stub backend down") + return self.sentinel + + +def test_immediate_fallback_when_remote_never_succeeded(): + primary = _StubDetector("fail") + local = _StubDetector("ok") + det = FallbackDetector(primary, fallback_factory=lambda: local) + + det.detect(FRAME) + + assert det.using_fallback + assert primary.calls == 1 # single failure, no retry storm + assert local.calls == 1 + + +def test_fallback_sticky_for_rest_of_job(): + primary = _StubDetector("fail") + local = _StubDetector("ok") + det = FallbackDetector(primary, fallback_factory=lambda: local) + + for _ in range(5): + det.detect(FRAME) + + assert primary.calls == 1 # never consulted again after the switch + assert local.calls == 5 + + +def test_consecutive_failure_threshold_after_success(): + # ok, then permanent failure: 2 failures return empty, 3rd switches + primary = _StubDetector("ok", "fail") + local = _StubDetector("ok") + det = FallbackDetector(primary, fallback_factory=lambda: local) + + det.detect(FRAME) # ok + det.detect(FRAME) # fail 1 -> empty + det.detect(FRAME) # fail 2 -> empty + assert not det.using_fallback + assert local.calls == 0 + + det.detect(FRAME) # fail 3 -> switch + assert det.using_fallback + assert local.calls == 1 + assert "stub backend down" in (det.fallback_reason or "") + + +def test_transient_blip_does_not_switch(): + # a single failure between successes resets the counter + primary = _StubDetector("ok", "fail", "ok", "fail", "ok") + local = _StubDetector("ok") + det = FallbackDetector(primary, fallback_factory=lambda: local) + + for _ in range(5): + det.detect(FRAME) + + assert not det.using_fallback + assert local.calls == 0 + + +def test_no_fallback_configured_returns_empty(): + primary = _StubDetector("fail") + det = FallbackDetector(primary, fallback_factory=None) + + detections = det.detect(FRAME) + + assert len(detections) == 0 + assert not det.using_fallback + + +def test_fallback_factory_called_lazily(): + calls = [] + + def factory(): + calls.append(1) + return _StubDetector("ok") + + primary = _StubDetector("ok") + det = FallbackDetector(primary, fallback_factory=factory) + det.detect(FRAME) + + assert calls == [] # healthy remote never constructs the local detector + + +def test_detection_fallback_config_validation(): + from clockd.config import ServerConfig + + assert ServerConfig(detection_fallback="local").detection_fallback == "local" + with pytest.raises(ValueError): + ServerConfig(detection_fallback="gpu") diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 10d4a8b..6ca0d19 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -157,3 +157,26 @@ def test_influxdb_unreachable_does_not_raise(): svc = MetricsService(cfg) # Should log warning but not raise svc.record(_make_result()) + + +def test_line_protocol_detection_tags(): + svc = MetricsService(MetricsConfig()) + result = _make_result() + result.detection_backend = "roboflow" + result.detection_model = "yolo26l-640" + result.detection_fallback = False + lines = svc._build_line_protocol(result, "vehicle_speed").split("\n") + + for line in lines: + assert "backend=roboflow" in line + assert "model=yolo26l-640" in line + assert "fallback=false" in line + + # fallback run is tagged with what actually performed detection + result.detection_backend = "local" + result.detection_model = "yolo26n.pt" + result.detection_fallback = True + line = svc._build_line_protocol(result, "vehicle_speed").split("\n")[0] + assert "backend=local" in line + assert "model=yolo26n.pt" in line + assert "fallback=true" in line