File size: 1,858 Bytes
06e54e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | """
Rule-based anomaly detection layer applied on top of YOLO detections.
Flags unusual behaviours: wrong zone entry, repeated appearances, overspeeding (future).
"""
from collections import defaultdict
from datetime import datetime, timedelta
# Track how often each plate has been seen in the last N minutes
_plate_history: dict[str, list[datetime]] = defaultdict(list)
REPEAT_WINDOW_MINUTES = 10
REPEAT_COUNT_THRESHOLD = 5 # flag if same plate seen >5 times in window
# Zones where heavy vehicles are not allowed
RESTRICTED_HEAVY_ZONES = {"VIP_LOT", "COMPACT_ONLY"}
def analyse(plate: str, vehicle_class: str, zone: str) -> tuple[str, str]:
"""
Evaluate one detection event and return (status, notes).
status: "normal" | "flagged" | "unauthorized" | "anomaly"
"""
now = datetime.now()
# 1. Prune old history
cutoff = now - timedelta(minutes=REPEAT_WINDOW_MINUTES)
_plate_history[plate] = [t for t in _plate_history[plate] if t > cutoff]
# 2. Record this sighting
_plate_history[plate].append(now)
notes_parts = []
status = "normal"
# Rule A: Heavy vehicle in restricted zone
if vehicle_class in ("bus", "truck") and zone.upper() in RESTRICTED_HEAVY_ZONES:
status = "unauthorized"
notes_parts.append(f"Heavy vehicle ({vehicle_class}) in restricted zone {zone}")
# Rule B: Same plate seen too many times in short window
count = len(_plate_history[plate])
if count > REPEAT_COUNT_THRESHOLD:
status = "anomaly"
notes_parts.append(f"Plate {plate} seen {count}x in last {REPEAT_WINDOW_MINUTES} min")
# Rule C: Unknown / unreadable plate
if not plate or plate == "UNKNOWN":
status = "flagged"
notes_parts.append("Unreadable plate — manual review required")
return status, "; ".join(notes_parts) if notes_parts else ""
|