percoandClaude Sonnet 4.6
65b74ce46d
- Remplace la source Frigate par l'API GetAiState de la caméra Reolink - Capture RTSP en temps réel (~20s) quand véhicule détecté, garde la meilleure frame - LPR avec YOLOv9 (détection plaque) + PaddleOCR v4 (lecture texte) via ONNX - Modèles partagés avec Frigate (volume local ./models/) - Cooldown 60s entre événements pour éviter les doublons Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
136 lines
4.7 KiB
Python
136 lines
4.7 KiB
Python
import os
|
|
import cv2
|
|
import numpy as np
|
|
import logging
|
|
|
|
log = logging.getLogger("lpr")
|
|
|
|
MODEL_CACHE = os.environ.get("MODEL_CACHE", "/models")
|
|
|
|
# Output format: [N, 7] where each row = [batch_idx, x1, y1, x2, y2, class_id, confidence]
|
|
_YOLO_CONF_THRESHOLD = 0.03
|
|
|
|
|
|
class PlateAnalyzer:
|
|
def __init__(self):
|
|
self._yolo = None
|
|
self._rec = None
|
|
self._keys: list[str] = []
|
|
self._load()
|
|
|
|
def _load(self):
|
|
try:
|
|
import onnxruntime as ort
|
|
providers = ["CPUExecutionProvider"]
|
|
|
|
yolo_path = os.path.join(MODEL_CACHE, "yolov9_license_plate", "yolov9-256-license-plates.onnx")
|
|
rec_path = os.path.join(MODEL_CACHE, "paddleocr-onnx", "recognition_v4.onnx")
|
|
keys_path = os.path.join(MODEL_CACHE, "paddleocr-onnx", "ppocr_keys_v1.txt")
|
|
|
|
if os.path.exists(yolo_path):
|
|
self._yolo = ort.InferenceSession(yolo_path, providers=providers)
|
|
log.info("YOLOv9 plate detector loaded")
|
|
else:
|
|
log.warning(f"YOLOv9 model not found at {yolo_path}")
|
|
|
|
if os.path.exists(rec_path):
|
|
self._rec = ort.InferenceSession(rec_path, providers=providers)
|
|
log.info("PaddleOCR recognition model loaded")
|
|
|
|
if os.path.exists(keys_path):
|
|
with open(keys_path) as f:
|
|
self._keys = f.read().splitlines()
|
|
log.info(f"Loaded {len(self._keys)} OCR characters")
|
|
|
|
except ImportError:
|
|
log.warning("onnxruntime not installed — LPR disabled")
|
|
except Exception as e:
|
|
log.error(f"LPR model load error: {e}")
|
|
|
|
def detect_plates(self, frame: np.ndarray) -> list[tuple]:
|
|
"""Returns [(x1, y1, x2, y2, conf), ...] in original frame coordinates."""
|
|
if self._yolo is None:
|
|
return []
|
|
|
|
h, w = frame.shape[:2]
|
|
inp = cv2.resize(frame, (256, 256))
|
|
inp = cv2.cvtColor(inp, cv2.COLOR_BGR2RGB).astype(np.float32) / 255.0
|
|
inp = inp.transpose(2, 0, 1)[np.newaxis]
|
|
|
|
out = self._yolo.run(None, {"images": inp})[0] # [N, 7]
|
|
|
|
boxes = []
|
|
for row in out:
|
|
# row: [batch_idx, x1, y1, x2, y2, class_id, confidence]
|
|
if len(row) >= 7:
|
|
_, x1, y1, x2, y2, _, conf = row
|
|
if conf < _YOLO_CONF_THRESHOLD:
|
|
continue
|
|
# Scale from 256x256 to original
|
|
x1 = max(0.0, float(x1) / 256.0 * w)
|
|
y1 = max(0.0, float(y1) / 256.0 * h)
|
|
x2 = min(float(w), float(x2) / 256.0 * w)
|
|
y2 = min(float(h), float(y2) / 256.0 * h)
|
|
if x2 > x1 + 4 and y2 > y1 + 4:
|
|
boxes.append((x1, y1, x2, y2, float(conf)))
|
|
|
|
return sorted(boxes, key=lambda b: -b[4])
|
|
|
|
def _ocr_crop(self, crop: np.ndarray) -> tuple[str, float]:
|
|
"""Run PaddleOCR recognition on a crop. Returns (text, confidence)."""
|
|
if self._rec is None or not self._keys or crop.size == 0:
|
|
return "", 0.0
|
|
|
|
h, w = crop.shape[:2]
|
|
if h == 0 or w == 0:
|
|
return "", 0.0
|
|
|
|
inp_h = 48
|
|
inp_w = max(10, int(inp_h * w / h))
|
|
|
|
resized = cv2.resize(crop, (inp_w, inp_h))
|
|
rgb = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB).astype(np.float32)
|
|
normalized = (rgb / 127.5) - 1.0
|
|
inp = normalized.transpose(2, 0, 1)[np.newaxis]
|
|
|
|
out = self._rec.run(None, {"x": inp})[0][0] # [seq, 6625]
|
|
|
|
chars: list[str] = []
|
|
confs: list[float] = []
|
|
prev_idx = 0
|
|
|
|
for step in out:
|
|
idx = int(np.argmax(step))
|
|
conf = float(step[idx])
|
|
if idx != prev_idx and idx != 0 and idx <= len(self._keys):
|
|
chars.append(self._keys[idx - 1])
|
|
confs.append(conf)
|
|
prev_idx = idx
|
|
|
|
text = "".join(chars)
|
|
avg_conf = float(np.mean(confs)) if confs else 0.0
|
|
return text, avg_conf
|
|
|
|
def read_plate(self, frame: np.ndarray) -> tuple[str, float]:
|
|
"""Detect plate in frame, read text. Returns (plate_text, confidence)."""
|
|
plate_boxes = self.detect_plates(frame)
|
|
if not plate_boxes:
|
|
return "", 0.0
|
|
|
|
x1, y1, x2, y2, plate_conf = plate_boxes[0]
|
|
pad = 8
|
|
cx1 = max(0, int(x1) - pad)
|
|
cy1 = max(0, int(y1) - pad)
|
|
cx2 = min(frame.shape[1], int(x2) + pad)
|
|
cy2 = min(frame.shape[0], int(y2) + pad)
|
|
|
|
crop = frame[cy1:cy2, cx1:cx2]
|
|
text, ocr_conf = self._ocr_crop(crop)
|
|
|
|
# Filter noise: plate must have at least 4 alphanumeric chars
|
|
alnum = "".join(c for c in text if c.isalnum())
|
|
if len(alnum) < 4:
|
|
return "", 0.0
|
|
|
|
return text, (plate_conf + ocr_conf) / 2.0
|