Integrate PlateRecognizer API as primary ANPR engine

Adds call_platerecognizer() to lpr.py which sends frames to the
PlateRecognizer cloud API (regions=fr) with a 1920px cap to stay
within API limits. In watcher.py, switches to a two-pass frame
scoring strategy: Laplacian sharpness on all frames first, then
YOLO only on the top-10 sharpest frames, then PlateRecognizer on
the top-3 by score. Falls back to local YOLO+Tesseract/PaddleOCR
if the API returns no result.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

app/lpr.py

@@ -38,6 +38,38 @@ def _nms(boxes: list, iou_threshold: float = 0.3) -> list:
     return result
 
 
+def call_platerecognizer(frame: np.ndarray, api_key: str, region: str = "fr") -> tuple[str, float]:
+    """Send a frame to PlateRecognizer API. Returns (plate_text, confidence)."""
+    import requests as req
+
+    h, w = frame.shape[:2]
+    if w > 1920:
+        scale = 1920 / w
+        frame = cv2.resize(frame, (1920, int(h * scale)), interpolation=cv2.INTER_AREA)
+
+    _, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
+    try:
+        resp = req.post(
+            "https://api.platerecognizer.com/v1/plate-reader/",
+            headers={"Authorization": f"Token {api_key}"},
+            files={"upload": ("frame.jpg", buf.tobytes(), "image/jpeg")},
+            data={"regions": region},
+            timeout=15,
+        )
+        data = resp.json()
+        results = data.get("results", [])
+        if results:
+            best = max(results, key=lambda r: r.get("score", 0))
+            plate = best.get("plate", "").upper().strip()
+            conf = float(best.get("score", 0))
+            if len([c for c in plate if c.isalnum()]) >= 4:
+                log.info(f"PlateRecognizer: {plate!r} conf={conf:.2f}")
+                return plate, conf
+    except Exception as e:
+        log.warning(f"PlateRecognizer API error: {e}")
+    return "", 0.0
+
+
 def _order_points(pts: np.ndarray) -> np.ndarray:
     """Order 4 points: top-left, top-right, bottom-right, bottom-left."""
     rect = np.zeros((4, 2), dtype=np.float32)

app/watcher.py

@@ -27,6 +27,7 @@ POLL_INTERVAL = float(os.environ.get("POLL_INTERVAL", "2"))
 CAPTURE_DURATION = int(os.environ.get("CAPTURE_DURATION", "15"))
 CAPTURE_FPS = int(os.environ.get("CAPTURE_FPS", "5"))
 COOLDOWN = int(os.environ.get("COOLDOWN", "60"))
+PLATERECOGNIZER_KEY = os.environ.get("PLATERECOGNIZER_API_KEY", "")
 
 _token: str | None = None
 _token_time = 0.0
@@ -144,32 +145,67 @@ def _process_clip(event_id: str, event_dir: str, clip_path: str, camera_name: st
         shutil.rmtree(event_dir, ignore_errors=True)
         return None
 
-    # Score ALL frames, rename sorted (best = frame_0001)
-    scored: list[tuple[float, np.ndarray, str]] = []
+    # Pass 1: rank all frames by sharpness (fast, no LPR)
+    sharpness: list[tuple[float, np.ndarray, str]] = []
     for path in frame_files:
         frame = cv2.imread(path)
         if frame is None:
             continue
+        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+        lap = cv2.Laplacian(gray, cv2.CV_64F).var()
+        sharpness.append((lap, frame, path))
+    sharpness.sort(key=lambda x: -x[0])
+
+    # Pass 2: run YOLO only on top 10 sharpest frames → LPR-based score
+    top10 = sharpness[:10]
+    scored: list[tuple[float, np.ndarray, str]] = []
+    for _, frame, path in top10:
         plates = _analyzer.detect_plates(frame) if _analyzer else []
         score = _frame_score(frame, plates)
         scored.append((score, frame, path))
-
     scored.sort(key=lambda x: -x[0])
 
-    for i, (_, _, old_path) in enumerate(scored):
+    # Merge: LPR-ranked top10 first, then remaining by sharpness
+    lpr_paths = {path for _, _, path in scored}
+    rest = [(lap * 0.001, frame, path) for lap, frame, path in sharpness if path not in lpr_paths]
+    full_sorted = scored + rest
+
+    for i, (_, _, old_path) in enumerate(full_sorted):
         os.rename(old_path, old_path + ".tmp")
-    for i, (_, _, old_path) in enumerate(scored):
+    for i, (_, _, old_path) in enumerate(full_sorted):
         os.rename(old_path + ".tmp", os.path.join(event_dir, f"frame_{i+1:04d}.jpg"))
 
-    best_frame = scored[0][1] if scored else None
+    best_frame = full_sorted[0][1] if full_sorted else None
     if best_frame is None:
         shutil.rmtree(event_dir, ignore_errors=True)
         return None
 
-    # LPR on best frame
+    # LPR: PlateRecognizer API (top 3 frames) → fallback to local
     plate, conf, plate_bbox, plate_corrected = ("", 0.0, None, None)
-    if _analyzer:
+    if PLATERECOGNIZER_KEY:
+        from lpr import call_platerecognizer
+        for _, frame, _ in scored[:3]:
+            p, c = call_platerecognizer(frame, PLATERECOGNIZER_KEY)
+            if p and c > conf:
+                plate, conf = p, c
+            if conf >= 0.7:
+                break
+        if plate:
+            # Get local perspective-corrected crop for display
+            if _analyzer:
+                plates = _analyzer.detect_plates(best_frame)
+                if plates:
+                    x1, y1, x2, y2, _ = plates[0]
+                    quad = _analyzer._find_plate_quad(best_frame, int(x1), int(y1), int(x2), int(y2))
+                    plate_bbox = (int(x1), int(y1), int(x2), int(y2))
+                    plate_corrected = _analyzer._perspective_correct(best_frame, quad) if quad else None
+        else:
+            log.info("PlateRecognizer returned no result, falling back to local LPR")
+            if _analyzer:
+                plate, conf, plate_bbox, plate_corrected = _analyzer.read_plate(best_frame)
+    elif _analyzer:
         plate, conf, plate_bbox, plate_corrected = _analyzer.read_plate(best_frame)
+
     log.info(f"LPR: plate={plate!r} conf={conf:.2f} bbox={plate_bbox}")
 
     # Save raw plate crop (4× upscale) and the perspective-corrected OCR crop

docker-compose.yml

@@ -19,6 +19,7 @@ services:
       - CAPTURE_DURATION=15
       - CAPTURE_FPS=5
       - COOLDOWN=60
+      - PLATERECOGNIZER_API_KEY=c1127284b5ef06b49ade33f1dfc0f50169b88c1d
     labels:
       - traefik.enable=true
       - traefik.http.routers.camwatch.rule=Host(`camwatch.nas.percolouco.com`)