utils/anomaly_detector.py

"""
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 ""