# Author: Sarvamangala Kokatanur
# Smart Traffic & EV Analytics System
# YOLOS + OpenCV + EasyOCR + SQLite + Gradio
# Dynamic AI Traffic Analytics Dashboard
import gradio as gr
import torch
import numpy as np
import cv2
import sqlite3
import os
import tempfile
import uuid
import pandas as pd
import matplotlib.pyplot as plt
import easyocr
from datetime import datetime, timedelta
from PIL import Image, ImageDraw
from transformers import (
YolosImageProcessor,
YolosForObjectDetection
)
# ============================================================
# CONFIGURATION
# ============================================================
MODEL_NAME = (
"nickmuchi/"
"yolos-small-finetuned-license-plate-detection"
)
EV_DISCOUNT = 50
# ============================================================
# LOAD YOLOS LICENSE PLATE DETECTION MODEL
# ============================================================
print("Loading YOLOS license plate detection model...")
processor = YolosImageProcessor.from_pretrained(
MODEL_NAME
)
model = YolosForObjectDetection.from_pretrained(
MODEL_NAME
)
model.eval()
print("YOLOS model loaded successfully.")
# ============================================================
# LOAD EASY OCR
# ============================================================
print("Loading EasyOCR...")
try:
reader = easyocr.Reader(
["en"],
gpu=torch.cuda.is_available()
)
print("EasyOCR loaded successfully.")
except Exception as e:
print("EasyOCR loading error:", e)
reader = None
# ============================================================
# DATABASE
# ============================================================
conn = sqlite3.connect(
"vehicles.db",
check_same_thread=False
)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS vehicles(
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
license_plate TEXT,
vehicle_status TEXT,
discount INTEGER
)
""")
conn.commit()
# ============================================================
# GET VEHICLE DATA
# ============================================================
def get_vehicle_dataframe():
try:
df = pd.read_sql_query(
"SELECT * FROM vehicles ORDER BY id DESC",
conn
)
return df
except Exception as e:
print("Database error:", e)
return pd.DataFrame(
columns=[
"id",
"timestamp",
"license_plate",
"vehicle_status",
"discount"
]
)
# ============================================================
# PLATE COLOR CLASSIFIER
# ============================================================
def classify_plate_color(plate_img):
try:
img = np.array(plate_img)
if img.size == 0:
return "Non-EV"
# PIL RGB -> OpenCV RGB
hsv = cv2.cvtColor(
img,
cv2.COLOR_RGB2HSV
)
h, w = hsv.shape[:2]
if w == 0:
return "Non-EV"
# Inspect left 20% of the plate
left_width = max(
1,
int(w * 0.20)
)
left = hsv[:, :left_width]
# Green HSV range
green = cv2.inRange(
left,
(35, 40, 40),
(90, 255, 255)
)
green_ratio = (
np.count_nonzero(green)
/ green.size
)
if green_ratio > 0.15:
return "EV"
return "Non-EV"
except Exception as e:
print(
"Plate color classification error:",
e
)
return "Non-EV"
# ============================================================
# OCR PREPROCESSING
# ============================================================
def preprocess_plate_for_ocr(plate_img):
try:
img = np.array(plate_img)
if img.size == 0:
return None
# Convert RGB -> grayscale
gray = cv2.cvtColor(
img,
cv2.COLOR_RGB2GRAY
)
# Resize for OCR
scale = 3
gray = cv2.resize(
gray,
None,
fx=scale,
fy=scale,
interpolation=cv2.INTER_CUBIC
)
# Noise reduction
gray = cv2.bilateralFilter(
gray,
9,
75,
75
)
# Contrast enhancement
gray = cv2.equalizeHist(
gray
)
return gray
except Exception as e:
print(
"OCR preprocessing error:",
e
)
return None
# ============================================================
# OCR FUNCTION
# ============================================================
def read_license_plate(plate_img):
if reader is None:
return "OCR_UNAVAILABLE"
try:
processed = preprocess_plate_for_ocr(
plate_img
)
if processed is None:
return "UNKNOWN"
results = reader.readtext(
processed,
detail=1,
paragraph=False
)
if not results:
return "UNKNOWN"
candidates = []
for result in results:
if len(result) >= 3:
text = str(
result[1]
).strip()
confidence = float(
result[2]
)
if confidence >= 0.20:
candidates.append(
(text, confidence)
)
if not candidates:
return "UNKNOWN"
# Select highest-confidence result
candidates.sort(
key=lambda x: x[1],
reverse=True
)
text = candidates[0][0]
# Remove unwanted characters
cleaned = "".join(
char
for char in text
if char.isalnum()
)
if cleaned == "":
return "UNKNOWN"
return cleaned.upper()
except Exception as e:
print(
"OCR error:",
e
)
return "UNKNOWN"
# ============================================================
# DUPLICATE VEHICLE CHECK
# ============================================================
def is_duplicate_vehicle(plate_number):
if (
plate_number is None
or plate_number == ""
or plate_number == "UNKNOWN"
):
return False
try:
cursor.execute("""
SELECT timestamp
FROM vehicles
WHERE license_plate=?
ORDER BY id DESC
LIMIT 1
""", (plate_number,))
row = cursor.fetchone()
if row is None:
return False
last_time = datetime.strptime(
row[0],
"%Y-%m-%d %H:%M:%S"
)
elapsed = (
datetime.now()
- last_time
)
if elapsed < timedelta(
minutes=5
):
return True
return False
except Exception as e:
print(
"Duplicate check error:",
e
)
return False
# ============================================================
# SAVE VEHICLE
# ============================================================
def save_vehicle(
plate,
status
):
if status == "EV":
discount = EV_DISCOUNT
else:
discount = 0
# Do not use duplicate filtering for UNKNOWN OCR
if plate != "UNKNOWN":
if is_duplicate_vehicle(
plate
):
return "Duplicate"
current_time = datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
try:
cursor.execute("""
INSERT INTO vehicles(
timestamp,
license_plate,
vehicle_status,
discount
)
VALUES (?, ?, ?, ?)
""", (
current_time,
plate,
status,
discount
))
conn.commit()
return "Saved"
except Exception as e:
print(
"Database save error:",
e
)
return "Error"
# ============================================================
# STATIC DASHBOARD
# ============================================================
def get_dashboard():
df = get_vehicle_dataframe()
fig, axs = plt.subplots(
2,
2,
figsize=(10, 7)
)
# --------------------------------------------------------
# NO DATA
# --------------------------------------------------------
if df.empty:
for ax in axs.flatten():
ax.text(
0.5,
0.5,
"No Data Available",
ha="center",
va="center",
fontsize=12
)
ax.axis("off")
plt.tight_layout()
return fig
# --------------------------------------------------------
# VEHICLE STATUS
# --------------------------------------------------------
status_counts = (
df["vehicle_status"]
.value_counts()
)
axs[0, 0].bar(
status_counts.index,
status_counts.values
)
axs[0, 0].set_title(
"EV vs Non-EV Vehicles"
)
axs[0, 0].set_ylabel(
"Number of Vehicles"
)
# --------------------------------------------------------
# PIE CHART
# --------------------------------------------------------
axs[0, 1].pie(
status_counts.values,
labels=status_counts.index,
autopct="%1.1f%%"
)
axs[0, 1].set_title(
"Vehicle Distribution"
)
# --------------------------------------------------------
# DISCOUNT
# --------------------------------------------------------
total_discount = int(
df["discount"].sum()
)
axs[1, 0].bar(
["Discount"],
[total_discount]
)
axs[1, 0].set_title(
"Total EV Discount"
)
axs[1, 0].set_ylabel(
"Amount (₹)"
)
# --------------------------------------------------------
# REPORT
# --------------------------------------------------------
axs[1, 1].axis("off")
ev_count = len(
df[
df["vehicle_status"] == "EV"
]
)
non_ev_count = len(
df[
df["vehicle_status"] == "Non-EV"
]
)
ev_percentage = (
ev_count / len(df) * 100
if len(df) > 0
else 0
)
report = (
"TRAFFIC ANALYTICS REPORT\n"
f"Total Vehicles : {len(df)}\n\n"
f"EV Vehicles : {ev_count}\n\n"
f"Non-EV Vehicles : {non_ev_count}\n\n"
f"EV Percentage : "
f"{ev_percentage:.2f}%\n\n"
f"Discount Given : "
f"₹{total_discount}"
)
axs[1, 1].text(
0.02,
0.98,
report,
fontsize=10,
va="top"
)
plt.tight_layout()
plt.close(fig)
return fig
# ============================================================
# AI ANALYTICS FUNCTIONS
# ============================================================
def get_total_vehicles():
df = get_vehicle_dataframe()
return len(df)
def get_ev_count():
df = get_vehicle_dataframe()
if df.empty:
return 0
return len(
df[
df["vehicle_status"] == "EV"
]
)
def get_non_ev_count():
df = get_vehicle_dataframe()
if df.empty:
return 0
return len(
df[
df["vehicle_status"] == "Non-EV"
]
)
def get_total_discount():
df = get_vehicle_dataframe()
if df.empty:
return 0
return int(
df["discount"].sum()
)
def get_ev_percentage():
total = get_total_vehicles()
ev = get_ev_count()
if total == 0:
return 0
return round(
ev / total * 100,
2
)
# ============================================================
# DAILY TRAFFIC
# ============================================================
def get_daily_traffic():
df = get_vehicle_dataframe()
if df.empty:
return pd.DataFrame()
try:
df["timestamp"] = pd.to_datetime(
df["timestamp"]
)
df["date"] = (
df["timestamp"]
.dt.date
)
daily = (
df.groupby("date")
.size()
.reset_index(
name="vehicles"
)
)
return daily
except Exception as e:
print(
"Daily traffic error:",
e
)
return pd.DataFrame()
# ============================================================
# GRAPH - EV VS NON-EV
# ============================================================
def generate_ev_graph():
df = get_vehicle_dataframe()
fig, ax = plt.subplots(
figsize=(8, 5)
)
if df.empty:
ax.text(
1,
1,
"No Vehicle Data Available",
ha="center",
va="center"
)
ax.axis("off")
return fig
counts = (
df["vehicle_status"]
.value_counts()
)
ax.bar(
counts.index,
counts.values
)
ax.set_title(
"EV vs Non-EV Analysis"
)
ax.set_xlabel(
"Vehicle Status"
)
ax.set_ylabel(
"Number of Vehicles"
)
plt.tight_layout()
plt.close(fig)
return fig
# ============================================================
# GRAPH - DISCOUNT
# ============================================================
def generate_discount_graph():
df = get_vehicle_dataframe()
fig, ax = plt.subplots(
figsize=(8, 5)
)
if df.empty:
ax.text(
0.5,
0.5,
"No Discount Data Available",
ha="center",
va="center"
)
ax.axis("off")
return fig
discount_data = (
df.groupby(
"vehicle_status"
)["discount"]
.sum()
)
ax.bar(
discount_data.index,
discount_data.values
)
ax.set_title(
"Discount Analysis"
)
ax.set_xlabel(
"Vehicle Status"
)
ax.set_ylabel(
"Discount Amount (₹)"
)
plt.tight_layout()
plt.close(fig)
return fig
# ============================================================
# GRAPH - DAILY TRAFFIC
# ============================================================
def generate_daily_traffic_graph():
daily = get_daily_traffic()
fig, ax = plt.subplots(
figsize=(9, 5)
)
if daily.empty:
ax.text(
0.5,
0.5,
"No Traffic Data Available",
ha="center",
va="center"
)
ax.axis("off")
return fig
ax.plot(
daily["date"].astype(str),
daily["vehicles"],
marker="o"
)
ax.set_title(
"Daily Traffic Volume"
)
ax.set_xlabel(
"Date"
)
ax.set_ylabel(
"Number of Vehicles"
)
plt.xticks(
rotation=45
)
plt.tight_layout()
plt.close(fig)
return fig
# ============================================================
# GRAPH - EV DAILY TRAFFIC
# ============================================================
def generate_ev_daily_graph():
df = get_vehicle_dataframe()
fig, ax = plt.subplots(
figsize=(9, 5)
)
if df.empty:
ax.text(
0.5,
0.5,
"No EV Data Available",
ha="center",
va="center"
)
ax.axis("off")
return fig
try:
df["timestamp"] = pd.to_datetime(
df["timestamp"]
)
df["date"] = (
df["timestamp"]
.dt.date
)
ev_df = df[
df["vehicle_status"] == "EV"
]
if ev_df.empty:
ax.text(
0.5,
0.5,
"No EV Vehicles Available",
ha="center",
va="center"
)
ax.axis("off")
return fig
daily_ev = (
ev_df.groupby("date")
.size()
.reset_index(
name="EV Vehicles"
)
)
ax.plot(
daily_ev["date"].astype(str),
daily_ev["EV Vehicles"],
marker="o"
)
ax.set_title(
"Daily EV Traffic"
)
ax.set_xlabel(
"Date"
)
ax.set_ylabel(
"EV Vehicles"
)
plt.xticks(
rotation=45
)
plt.tight_layout()
plt.close(fig)
return fig
except Exception as e:
print(
"EV graph error:",
e
)
ax.text(
0.5,
0.5,
"Unable to generate graph",
ha="center",
va="center"
)
ax.axis("off")
return fig
# ============================================================
# AGENTIC AI QUERY ENGINE
# ============================================================
def traffic_ai_agent(user_query):
if user_query is None:
return (
"Please enter a question.",
None
)
query = user_query.strip().lower()
if query == "":
return (
"Please enter a question.",
None
)
# ========================================================
# GET LIVE DATABASE VALUES
# ========================================================
total = get_total_vehicles()
ev = get_ev_count()
non_ev = get_non_ev_count()
discount = get_total_discount()
ev_percentage = get_ev_percentage()
# ========================================================
# EV COUNT
# ========================================================
if (
"how many ev" in query
or "number of ev" in query
or "ev count" in query
or "ev vehicles" in query
):
response = (
"AI TRAFFIC ANALYSIS\n"
f"EV vehicles detected are: {ev}"
f"out of {total} Total vehicles: \n"
f"EV vehicle percentage is : "
f"{ev_percentage:.2f}%"
)
return (
response,
generate_ev_graph()
)
# ========================================================
# NON-EV
# ========================================================
if (
"non ev" in query
or "non-ev" in query
or "non electric" in query
or "Nonelectric" in query
):
response = (
"AI TRAFFIC ANALYSIS\n\n"
f"Non-EV vehicles detected are: "
f"{non_ev}\n\n"
f"EV vehicles detected are: {ev}\n\n"
f"Total vehicles: {total}"
)
return (
response,
generate_ev_graph()
)
# ========================================================
# EV PERCENTAGE
# ========================================================
if (
"percentage" in query
or "percent" in query
or "ratio" in query
or "proportion" in query
):
response = (
" AI EV ANALYSIS\n\n"
f"Total vehicles: {total}\n\n"
f"EV vehicles: {ev}\n\n"
f"Non-EV vehicles: {non_ev}\n\n"
f"EV percentage: "
f"{ev_percentage:.2f}%"
)
return (
response,
generate_ev_graph()
)
# ========================================================
# DISCOUNT
# ========================================================
if (
"discount" in query
or "incentive" in query
or "saving" in query
or "savings" in query
):
response = (
" AI INCENTIVE ANALYSIS\n\n"
f"EV vehicles are: {ev}\n\n"
f"Discount per EV: "
f"₹{EV_DISCOUNT}\n\n"
f"Total discount provided is: ₹{discount} for {ev} vehicles"
)
return (
response,
generate_discount_graph()
)
# ========================================================
# DAILY TRAFFIC
# ========================================================
if (
"daily traffic" in query
or "traffic by day" in query
or "traffic trend" in query
or "trend" in query
):
daily = get_daily_traffic()
if daily.empty:
response = (
" No daily traffic data "
"is currently available."
)
else:
highest = daily.loc[
daily["vehicles"].idxmax()
]
response = (
" DAILY TRAFFIC ANALYSIS\n\n"
f"Number of recorded days: "
f"{len(daily)}\n\n"
f"Highest traffic date: "
f"{highest['date']}\n\n"
f" On highest traffic date:{highest['date']} number of Vehicles are "
f"{highest['vehicles']}"
)
return (
response,
generate_daily_traffic_graph()
)
# ========================================================
# EV DAILY TRAFFIC
# ========================================================
if (
"ev trend" in query
or "daily ev" in query
or "ev traffic" in query
):
response = (
" EV TRAFFIC TREND\n"
f"Total EV vehicles: {ev}\n"
f"Overall EV percentage: "
f"{ev_percentage:.2f}%"
)
return (
response,
generate_ev_daily_graph()
)
# ========================================================
# TOTAL VEHICLES
# ========================================================
if (
"total vehicle" in query
or "total traffic" in query
or "how many vehicle" in query
or "number of vehicle" in query
):
response = (
" TRAFFIC ANALYSIS\n\n"
f"Total vehicles detected are: {total}\n"
f"No. of EV vehicles are: {ev}\n"
f"No. of Non-EV vehicles: {non_ev}"
)
return (
response,
generate_ev_graph()
)
# ========================================================
# CO2
# ========================================================
if (
"co2" in query
or "CO2" in query
or "carbon" in query
or "environment" in query
or "emission" in query
or "CO2 savings" in query
or "saving" in query
):
# Project-level illustrative estimate.
# Replace with validated emission methodology
# for final research evaluation.
estimated_co2 = ev * 2.3
response = (
" AI ENVIRONMENTAL ANALYSIS\n\n"
f"EV vehicles detected: {ev}\n\n"
f"Illustrative estimated CO₂ saving: "
f"{estimated_co2:.2f} kg\n\n"
"This is a project-level estimation. "
"For a research publication, the emission "
"factor and travel-distance assumptions "
"should be validated using an appropriate "
"transportation emissions methodology."
)
return (
response,
generate_ev_graph()
)
# ========================================================
# SUMMARY / REPORT
# ========================================================
if (
"summary" in query
or "overall" in query
or "report" in query
or "dashboard" in query
):
response = (
"SMART TRAFFIC AI SUMMARY\n"
f"Total Vehicles: {total}\n"
f" EV Vehicles: {ev}\n"
f"Non-EV Vehicles: {non_ev}\n"
f" EV Percentage: "
f"{ev_percentage:.2f}%\n"
f"Total Discount provided: ₹{discount}"
)
return (
response,
generate_ev_graph()
)
# ========================================================
# HELP
# ========================================================
response = (
"Smart Traffic AI Assistant\n\n"
"I can analyse the vehicle database. "
"Try asking:\n"
"• How many EV vehicles were detected?\n"
"• How many Non-EV vehicles are there?\n"
"• What is the total traffic?\n"
"• What percentage of vehicles are EVs?\n"
"• What is the total discount?\n"
"• Show daily traffic trend.\n"
"• Show EV traffic trend.\n"
"• Give me an overall summary.\n"
"• What are the CO2 savings?"
)
return (
response,
None
)
# ============================================================
# VEHICLE IMAGE PROCESSING
# ============================================================
def process_image(img):
if img is None:
return (
None,
"Please upload an image.",
"0",
"0",
"₹0",
get_dashboard()
)
try:
# ----------------------------------------------------
# Convert input to PIL
# ----------------------------------------------------
if isinstance(img, np.ndarray):
image = Image.fromarray(
img.astype(np.uint8)
)
elif isinstance(img, Image.Image):
image = img.copy()
else:
return (
None,
"Unsupported image format.",
"0",
"0",
"₹0",
get_dashboard()
)
# ----------------------------------------------------
# YOLOS PROCESSING
# ----------------------------------------------------
inputs = processor(
images=image,
return_tensors="pt"
)
with torch.no_grad():
outputs = model(
**inputs
)
target_sizes = torch.tensor(
[
[
image.size[1],
image.size[0]
]
]
)
results = (
processor
.post_process_object_detection(
outputs,
threshold=0.30,
target_sizes=target_sizes
)[0]
)
draw = ImageDraw.Draw(
image
)
ev_count = 0
non_ev_count = 0
discount_total = 0
output_text = ""
# ----------------------------------------------------
# NO PLATE DETECTION
# ----------------------------------------------------
if len(results["boxes"]) == 0:
return (
image,
"No license plate detected.",
"0",
"0",
"₹0",
get_dashboard()
)
# ----------------------------------------------------
# PROCESS EACH DETECTED PLATE
# ----------------------------------------------------
processed_count = 0
for i, box in enumerate(
results["boxes"]
):
x1, y1, x2, y2 = map(
int,
box.tolist()
)
# ------------------------------------------------
# Boundary validation
# ------------------------------------------------
x1 = max(
0,
min(x1, image.width - 1)
)
y1 = max(
0,
min(y1, image.height - 1)
)
x2 = max(
0,
min(x2, image.width)
)
y2 = max(
0,
min(y2, image.height)
)
if x2 <= x1 or y2 <= y1:
continue
# ------------------------------------------------
# Crop license plate
# ------------------------------------------------
plate = image.crop(
(
x1,
y1,
x2,
y2
)
)
# ------------------------------------------------
# EV / NON-EV
# ------------------------------------------------
status = classify_plate_color(
plate
)
# ------------------------------------------------
# OCR
# ------------------------------------------------
plate_number = read_license_plate(
plate
)
# ------------------------------------------------
# Save database record
# ------------------------------------------------
saved = save_vehicle(
plate_number,
status
)
if saved == "Duplicate":
output_text += (
f"Vehicle {i + 1}\n"
f"Plate: {plate_number}\n"
f"Status: {status}\n"
"Result: Duplicate - "
"not added again\n"
)
continue
if saved == "Error":
output_text += (
f"Vehicle {i + 1}\n"
f"Plate: {plate_number}\n"
f"Status: {status}\n"
"Database: Error\n"
)
# ------------------------------------------------
# EV
# ------------------------------------------------
if status == "EV":
ev_count += 1
discount = EV_DISCOUNT
discount_total += discount
label = (
f"{plate_number} | "
f"EV | ₹{discount}"
)
outline_color = "green"
# ------------------------------------------------
# NON-EV
# ------------------------------------------------
else:
non_ev_count += 1
discount = 0
label = (
f"{plate_number} | Non-EV"
)
outline_color = "red"
processed_count += 1
# ------------------------------------------------
# Draw bounding box
# ------------------------------------------------
draw.rectangle(
[
x1,
y1,
x2,
y2
],
outline=outline_color,
width=4
)
# ------------------------------------------------
# Draw label background
# ------------------------------------------------
text_y = max(
0,
y1 - 22
)
draw.rectangle(
[
x1,
text_y,
min(
image.width,
x1 + 300
),
y1
],
fill="black"
)
draw.text(
(
x1 + 3,
text_y + 3
),
label,
fill=outline_color
)
# ------------------------------------------------
# Output summary
# ------------------------------------------------
output_text += (
f"Vehicle {i + 1}\n"
f"Plate: {plate_number}\n"
f"Status: {status}\n"
f"Discount: ₹{discount}\n\n"
)
# ----------------------------------------------------
# If every detection was invalid
# ----------------------------------------------------
if processed_count == 0:
return (
image,
"No valid license plate was processed.",
"0",
"0",
"₹0",
get_dashboard()
)
# ----------------------------------------------------
# Refresh dashboard
# ----------------------------------------------------
dashboard = get_dashboard()
return (
image,
output_text,
str(ev_count),
str(non_ev_count),
f"₹{discount_total}",
dashboard
)
except Exception as e:
print(
"Image processing error:",
e
)
return (
img,
f"Processing error: {str(e)}",
"0",
"0",
"₹0",
get_dashboard()
)
# ============================================================
# VIDEO PROCESSING
# ============================================================
def process_video(video_path, frame_skip=5, max_frames=0):
"""Process an uploaded traffic video and create an annotated output video."""
if video_path is None:
return None, "Please upload a video.", "0", "0", "₹0", get_dashboard()
cap = None
writer = None
try:
cap = cv2.VideoCapture(video_path)
if not cap.isOpened():
return None, "Unable to open the uploaded video.", "0", "0", "₹0", get_dashboard()
fps = cap.get(cv2.CAP_PROP_FPS)
if not fps or fps <= 0:
fps = 25.0
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
if width <= 0 or height <= 0:
return None, "Invalid video dimensions.", "0", "0", "₹0", get_dashboard()
try:
frame_skip = max(1, int(frame_skip))
except Exception:
frame_skip = 5
try:
max_frames = max(0, int(max_frames))
except Exception:
max_frames = 0
output_path = os.path.join(
tempfile.gettempdir(),
f"traffic_result_{uuid.uuid4().hex}.mp4"
)
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
writer = cv2.VideoWriter(
output_path,
fourcc,
fps,
(width, height)
)
if not writer.isOpened():
return None, "Unable to create the processed output video.", "0", "0", "₹0", get_dashboard()
ev_count = 0
non_ev_count = 0
discount_total = 0
detected_plates = 0
processed_frames = 0
frame_number = 0
# Prevent repeated database records for the same known plate
# across multiple frames of this uploaded video.
video_seen_plates = set()
output_lines = [
"VIDEO TRAFFIC ANALYSIS",
"",
f"Input frames: {total_frames if total_frames > 0 else 'Unknown'}",
f"Processing every {frame_skip} frame(s)",
""
]
while True:
ret, frame = cap.read()
if not ret:
break
if max_frames > 0 and frame_number >= max_frames:
break
# Write unprocessed frames unchanged.
if frame_number % frame_skip != 0:
writer.write(frame)
frame_number += 1
continue
processed_frames += 1
rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
image = Image.fromarray(rgb_frame)
inputs = processor(
images=image,
return_tensors="pt"
)
with torch.no_grad():
outputs = model(**inputs)
target_sizes = torch.tensor(
[[image.size[1], image.size[0]]]
)
results = processor.post_process_object_detection(
outputs,
threshold=0.30,
target_sizes=target_sizes
)[0]
draw = ImageDraw.Draw(image)
for i, box in enumerate(results["boxes"]):
x1, y1, x2, y2 = map(int, box.tolist())
x1 = max(0, min(x1, image.width - 1))
y1 = max(0, min(y1, image.height - 1))
x2 = max(0, min(x2, image.width))
y2 = max(0, min(y2, image.height))
if x2 <= x1 or y2 <= y1:
continue
plate = image.crop((x1, y1, x2, y2))
status = classify_plate_color(plate)
plate_number = read_license_plate(plate)
detected_plates += 1
normalized_plate = str(plate_number).strip().upper()
should_count = True
if normalized_plate not in ("", "UNKNOWN", "OCR_UNAVAILABLE"):
if normalized_plate in video_seen_plates:
should_count = False
else:
video_seen_plates.add(normalized_plate)
if should_count:
saved = save_vehicle(plate_number, status)
if saved not in ("Duplicate", "Error"):
if status == "EV":
ev_count += 1
discount_total += EV_DISCOUNT
else:
non_ev_count += 1
output_lines.extend([
f"Vehicle {ev_count + non_ev_count}",
f"Plate: {plate_number}",
f"Status: {status}",
f"Discount: ₹{EV_DISCOUNT if status == 'EV' else 0}",
""
])
if status == "EV":
outline_color = "green"
label = f"{plate_number} | EV | ₹{EV_DISCOUNT}"
else:
outline_color = "red"
label = f"{plate_number} | Non-EV"
draw.rectangle(
[x1, y1, x2, y2],
outline=outline_color,
width=4
)
text_y = max(0, y1 - 22)
draw.rectangle(
[
x1,
text_y,
min(image.width, x1 + 330),
y1
],
fill="black"
)
draw.text(
(x1 + 3, text_y + 3),
label,
fill=outline_color
)
output_frame = cv2.cvtColor(
np.array(image),
cv2.COLOR_RGB2BGR
)
writer.write(output_frame)
frame_number += 1
cap.release()
cap = None
writer.release()
writer = None
if processed_frames == 0:
return None, "No frames were processed.", "0", "0", "₹0", get_dashboard()
total_unique = ev_count + non_ev_count
ev_percentage = (
ev_count / total_unique * 100
if total_unique > 0 else 0
)
output_lines.extend([
"VIDEO SUMMARY",
f"Processed frames: {processed_frames}",
f"Detected plate instances: {detected_plates}",
f"Unique vehicles recorded: {total_unique}",
f"EV vehicles: {ev_count}",
f"Non-EV vehicles: {non_ev_count}",
f"EV percentage: {ev_percentage:.2f}%",
f"Total EV discount: ₹{discount_total}"
])
return (
output_path,
"\n".join(output_lines),
str(ev_count),
str(non_ev_count),
f"₹{discount_total}",
get_dashboard()
)
except Exception as e:
print("Video processing error:", e)
if cap is not None:
cap.release()
if writer is not None:
writer.release()
return None, f"Video processing error: {str(e)}", "0", "0", "₹0", get_dashboard()
# ============================================================
# REFRESH DASHBOARD
# ============================================================
def refresh_dashboard():
return get_dashboard()
# ============================================================
# CLEAR AI QUERY
# ============================================================
def clear_ai():
return (
"",
"",
None
)
# ============================================================
# CSS
# ============================================================
css = """
.gradio-container {
max-width: 98% !important;
margin: auto;
}
textarea {
font-size: 15px !important;
}
h1 {
text-align: center;
}
.block {
border-radius: 12px;
}
.gr-image {
min-height: 350px;
}
"""
# ============================================================
# GRADIO APPLICATION
# ============================================================
with gr.Blocks(
css=css
) as demo:
# ========================================================
# TITLE
# ========================================================
gr.Markdown(
"# Smart Traffic & EV Analytics System"
)
gr.Markdown(
"YOLOS License Plate Detection • "
"OpenCV EV Classification • "
"EasyOCR • EV Incentives • "
"Dynamic AI Analytics"
)
# ========================================================
# TAB 1 - VEHICLE DETECTION
# ========================================================
with gr.Tab(
"Vehicle Detection"
):
with gr.Row(
equal_height=True
):
# ------------------------------------------------
# INPUT
# ------------------------------------------------
with gr.Column(
scale=1
):
input_img = gr.Image(
type="numpy",
label="Input Vehicle Image",
sources=[
"upload",
"webcam"
],
height=350
)
btn = gr.Button(
"Scan Vehicle",
variant="primary",
size="lg"
)
# ------------------------------------------------
# OUTPUT
# ------------------------------------------------
with gr.Column(
scale=1
):
output_img = gr.Image(
label="Detection Result",
height=350
)
# ------------------------------------------------
# SUMMARY
# ------------------------------------------------
with gr.Column(
scale=1
):
summary_box = gr.Textbox(
label="Detection Summary",
lines=16
)
# ====================================================
# STATISTICS
# ====================================================
with gr.Row():
ev_box = gr.Textbox(
label="EV Vehicles"
)
non_ev_box = gr.Textbox(
label="Non-EV Vehicles"
)
discount_box = gr.Textbox(
label="Total Discount"
)
# ====================================================
# STATIC DASHBOARD
# ====================================================
gr.Markdown(
"## Traffic Analytics Dashboard"
)
dashboard = gr.Plot(
label="Analytics"
)
btn.click(
fn=process_image,
inputs=input_img,
outputs=[
output_img,
summary_box,
ev_box,
non_ev_box,
discount_box,
dashboard
]
)
# ========================================================
# TAB 2 - VIDEO PROCESSING
# ========================================================
with gr.Tab("Video Processing"):
gr.Markdown("## Video Traffic & EV Analysis")
gr.Markdown(
"Upload a traffic video to detect license plates, "
"classify EV/Non-EV vehicles, read plate numbers using "
"EasyOCR, calculate EV incentives, and generate an "
"annotated output video."
)
with gr.Row(equal_height=True):
with gr.Column(scale=1):
input_video = gr.Video(
label="Upload Traffic Video",
sources=["upload"],
format="mp4"
)
frame_skip = gr.Slider(
minimum=1,
maximum=15,
value=5,
step=1,
label="Process Every Nth Frame",
info="1 = every frame; higher values reduce processing time."
)
max_frames = gr.Number(
value=0,
precision=0,
label="Maximum Frames",
info="0 = process the complete video."
)
video_btn = gr.Button(
"Process Video",
variant="primary",
size="lg"
)
with gr.Column(scale=1):
output_video = gr.Video(
label="Processed Video",
autoplay=False
)
video_summary = gr.Textbox(
label="Video Analysis Summary",
lines=18
)
with gr.Row():
video_ev_box = gr.Textbox(label="EV Vehicles")
video_non_ev_box = gr.Textbox(label="Non-EV Vehicles")
video_discount_box = gr.Textbox(label="Total Discount")
gr.Markdown("## Video Traffic Analytics Dashboard")
video_dashboard = gr.Plot(label="Video Analytics")
video_btn.click(
fn=process_video,
inputs=[
input_video,
frame_skip,
max_frames
],
outputs=[
output_video,
video_summary,
video_ev_box,
video_non_ev_box,
video_discount_box,
video_dashboard
]
)
# ========================================================
# TAB 3 - AGENTIC AI DASHBOARD
# ========================================================
with gr.Tab(
"Agentic AI Dashboard"
):
gr.Markdown(
"## Agentic AI Traffic Analytics Assistant"
)
gr.Markdown(
"""
Ask questions about your vehicle database
using natural language.
The AI analytics layer dynamically reads the
SQLite database and selects the appropriate
analytics operation.
"""
)
# ----------------------------------------------------
# Query input
# ----------------------------------------------------
user_query = gr.Textbox(
label="Ask your traffic analytics question",
placeholder=(
"Example: How many EV vehicles "
"were detected?"
),
lines=3
)
with gr.Row():
ask_button = gr.Button(
"Ask AI",
variant="primary",
size="lg"
)
clear_button = gr.Button(
"Clear",
size="lg"
)
# ----------------------------------------------------
# AI RESPONSE
# ----------------------------------------------------
ai_response = gr.Textbox(
label=" AI Analysis",
lines=12
)
# ----------------------------------------------------
# DYNAMIC GRAPH
# ----------------------------------------------------
ai_graph = gr.Plot(
label="📊 Dynamic AI Visualization"
)
# ----------------------------------------------------
# BUTTON ACTION
# ----------------------------------------------------
ask_button.click(
fn=traffic_ai_agent,
inputs=user_query,
outputs=[
ai_response,
ai_graph
]
)
clear_button.click(
fn=clear_ai,
inputs=[],
outputs=[
user_query,
ai_response,
ai_graph
]
)
# ====================================================
# SUGGESTED QUESTIONS
# ====================================================
gr.Markdown(
"""
### Suggested Questions
**Traffic**
- How many vehicles were detected?
- What is the total traffic?
- Show daily traffic trend.
**EV Analysis**
- How many EV vehicles were detected?
- How many Non-EV vehicles are there?
- What percentage of vehicles are EVs?
- Show EV traffic trend.
**Incentives**
- What is the total discount?
- How much incentive was provided?
**Environmental**
- What are the CO2 savings?
**Overall**
- Give me an overall traffic summary.
"""
)
# ========================================================
# TAB 4 - ADMIN / DATABASE
# ========================================================
with gr.Tab(
" Admin Dashboard"
):
gr.Markdown(
"## Vehicle Database"
)
database_table = gr.Dataframe(
label="Stored Vehicle Records",
interactive=False
)
with gr.Row():
refresh_button = gr.Button(
"Refresh Database"
)
refresh_graph_button = gr.Button(
"Refresh Dashboard"
)
admin_dashboard = gr.Plot(
label="Current Database Analytics"
)
def load_database():
return get_vehicle_dataframe()
refresh_button.click(
fn=load_database,
inputs=[],
outputs=database_table
)
refresh_graph_button.click(
fn=refresh_dashboard,
inputs=[],
outputs=admin_dashboard
)
# ============================================================
# APPLICATION START
# ============================================================
if __name__ == "__main__":
print(
"Starting Smart Traffic & EV Analytics System..."
)
demo.queue()
demo.launch(
ssr_mode=False
)
# Author: Sarvamangala Kokatanur
# Smart Traffic & EV Analytics System
# YOLOS + OpenCV + EasyOCR + SQLite + Gradio
# Dynamic AI Traffic Analytics Dashboard
import gradio as gr
import torch
import numpy as np
import cv2
import sqlite3
import os
import tempfile
import uuid
import pandas as pd
import matplotlib.pyplot as plt
import easyocr
from datetime import datetime, timedelta
from PIL import Image, ImageDraw
from transformers import (
YolosImageProcessor,
YolosForObjectDetection
)
# ============================================================
# CONFIGURATION
# ============================================================
MODEL_NAME = (
"nickmuchi/"
"yolos-small-finetuned-license-plate-detection"
)
EV_DISCOUNT = 50
# ============================================================
# LOAD YOLOS LICENSE PLATE DETECTION MODEL
# ============================================================
print("Loading YOLOS license plate detection model...")
processor = YolosImageProcessor.from_pretrained(
MODEL_NAME
)
model = YolosForObjectDetection.from_pretrained(
MODEL_NAME
)
model.eval()
print("YOLOS model loaded successfully.")
# ============================================================
# LOAD EASY OCR
# ============================================================
print("Loading EasyOCR...")
try:
reader = easyocr.Reader(
["en"],
gpu=torch.cuda.is_available()
)
print("EasyOCR loaded successfully.")
except Exception as e:
print("EasyOCR loading error:", e)
reader = None
# ============================================================
# DATABASE
# ============================================================
conn = sqlite3.connect(
"vehicles.db",
check_same_thread=False
)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS vehicles(
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT,
license_plate TEXT,
vehicle_status TEXT,