julian
6933406ae3
Skeleton of the host-side vision service per the packaging decision: a Python/FastAPI app at apps/vision/, uv-managed, wired into the Turbo graph via a thin package.json shim (dev/lint/test/build → uv/uvicorn/ruff/pytest). A per-package turbo.json sets build outputs [] so the no-op build is warning-free. Endpoints: GET /health (readiness + model version) and POST /analyze (raw octet-stream body, so Node POSTs Snapshot.bytes directly; empty→400, oversize→413, recognizer-not-ready→503). The recognizer is a Protocol with a StubRecognizer (no models, boots/tests offline — the dev/CI default) and a FastAlprRecognizer (the real MIT YOLOv9+CCT/ONNX stack, lazily imported; missing models ⇒ ready=False, not a crash) — the device-adapter pattern applied to the model. fast-alpr + onnxruntime are an optional `alpr` extra, so `uv sync` needs no model download. Verified: turbo run lint|test|build includes @parking/vision and stays green; uv run mypy strict-clean; uvicorn boots and serves /health + /analyze live; pnpm workspace 6→7. Not built yet: the Node VisionClient adapter, a Dockerfile + model fetch, and Job 2 (vehicle verification). Updates the packaging decision (As-scaffolded) + log. Claude-Session: https://claude.ai/code/session_01Xcm6ikLgGoCxxHrxtjkk5V
89 lines
3.3 KiB
Python
89 lines
3.3 KiB
Python
"""FastAPI app: POST /analyze (snapshot → plate) + GET /health.
|
|
|
|
Called by the Node backend over localhost HTTP (the camera driver already holds the
|
|
JPEG bytes — Snapshot.bytes). This service is a SEPARATE PROCESS with its own failure
|
|
domain: if it's down or unsure, the host falls back to the ticket path — recognition is
|
|
advisory, never the sole authority. See wiki/entities/opencv-anpr-service.md.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
|
|
from .recognizer import Recognizer, build_recognizer
|
|
from .schemas import AnalyzeResponse, HealthResponse
|
|
from .settings import Settings, get_settings
|
|
|
|
# Cap an upload so a malformed/huge POST can't exhaust memory (a camera JPEG is well
|
|
# under this). 413 beyond it.
|
|
MAX_IMAGE_BYTES = 12 * 1024 * 1024
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|
settings = get_settings()
|
|
app.state.settings = settings
|
|
# Build the recognizer once at startup (models load here, not per-request).
|
|
app.state.recognizer = build_recognizer(settings)
|
|
yield
|
|
|
|
|
|
app = FastAPI(title="parking-vision", version="0.0.0", lifespan=lifespan)
|
|
|
|
|
|
# Typed accessors over the untyped `app.state` (so mypy --strict sees the real types).
|
|
def _recognizer(request: Request) -> Recognizer:
|
|
rec: Recognizer = request.app.state.recognizer
|
|
return rec
|
|
|
|
|
|
def _settings(request: Request) -> Settings:
|
|
settings: Settings = request.app.state.settings
|
|
return settings
|
|
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
async def health(request: Request) -> HealthResponse:
|
|
rec = _recognizer(request)
|
|
settings = _settings(request)
|
|
ready = bool(rec.ready)
|
|
return HealthResponse(
|
|
status="ok" if ready else "degraded",
|
|
recognizer=settings.recognizer,
|
|
ready=ready,
|
|
model_version=rec.model_version,
|
|
detail=getattr(rec, "error", None),
|
|
)
|
|
|
|
|
|
@app.post("/analyze", response_model=AnalyzeResponse)
|
|
async def analyze(request: Request) -> AnalyzeResponse:
|
|
"""Analyze raw image bytes (the camera JPEG). Body is the octet-stream itself, so
|
|
the Node side POSTs Snapshot.bytes directly with Content-Type:
|
|
application/octet-stream — no multipart wrapping. We read the raw body ourselves
|
|
(rather than a required Body param) so an empty/oversize body returns our own clean
|
|
400/413 instead of FastAPI's generic 422."""
|
|
image = await request.body()
|
|
if not image:
|
|
raise HTTPException(status_code=400, detail="empty image body")
|
|
if len(image) > MAX_IMAGE_BYTES:
|
|
raise HTTPException(status_code=413, detail="image too large")
|
|
|
|
rec = _recognizer(request)
|
|
if not rec.ready:
|
|
# The real recognizer failed to load — be explicit so Node falls back rather
|
|
# than treating a silent empty result as "no plate present".
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail=f"recognizer not ready: {getattr(rec, 'error', 'unavailable')}",
|
|
)
|
|
try:
|
|
return rec.analyze(image)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
|
except Exception as exc: # noqa: BLE001 - never leak a stack to the caller
|
|
raise HTTPException(status_code=500, detail=f"analysis failed: {exc}") from exc
|