File size: 1,923 Bytes
06e54e9 c3845b8 06e54e9 c3845b8 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 54 55 56 57 58 59 60 61 62 63 64 65 66 | """
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()]
|