utils/incident_logger.py

"""
Logs vehicle detection events to a CSV file.
Each row is also ingested into ChromaDB for RAG retrieval.
"""

import csv
import uuid
from datetime import datetime
from pathlib import Path
import pandas as pd
import pytz

IST = pytz.timezone("Asia/Kolkata")

LOG_PATH = Path("data/incidents.csv")
FIELDNAMES = ["id", "timestamp", "plate", "vehicle_class", "zone", "status", "notes"]


def _ensure_file():
    LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    if not LOG_PATH.exists():
        with open(LOG_PATH, "w", newline="", encoding="utf-8") as f:
            csv.DictWriter(f, fieldnames=FIELDNAMES).writeheader()


def log_incident(plate, vehicle_class, zone="Entry", status="normal", notes=""):
    """
    Append one incident to the CSV.
    status: 'normal' | 'flagged' | 'unauthorized' | 'anomaly'
    Returns the logged row dict.
    """
    _ensure_file()
    row = {
        "id": str(uuid.uuid4())[:8],
        "timestamp": datetime.now(IST).strftime("%Y-%m-%d %H:%M:%S"),
        "plate": plate or "UNKNOWN",
        "vehicle_class": vehicle_class,
        "zone": zone,
        "status": status,
        "notes": notes,
    }
    with open(LOG_PATH, "a", newline="", encoding="utf-8") as f:
        csv.DictWriter(f, fieldnames=FIELDNAMES).writerow(row)
    return row


def load_incidents():
    """Return all incidents as a sorted DataFrame."""
    _ensure_file()
    df = pd.read_csv(LOG_PATH, encoding='utf-8', encoding_errors='replace')
    if df.empty:
        return pd.DataFrame(columns=FIELDNAMES)
    df["timestamp"] = pd.to_datetime(df["timestamp"])
    return df.sort_values("timestamp", ascending=False)


def get_flagged_plates():
    df = load_incidents()
    if df.empty:
        return []
    return list(set(df[df["status"].isin(["flagged", "unauthorized"])]["plate"].tolist()))


def is_plate_flagged(plate):
    return plate.upper() in [p.upper() for p in get_flagged_plates()]