# ============================================================
# Smart Traffic & EV Analytics System
# Agentic AI Dynamic Dashboard
# Author: Sarvamangala Kokatanur
# ============================================================
# ---------------- IMPORT LIBRARIES ----------------
# ============================================================
# IMPORT LIBRARIES
# ============================================================
import os
import json
import sqlite3
import gradio as gr
import torch
import numpy as np
import cv2
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from PIL import Image, ImageDraw
from transformers import (
YolosImageProcessor,
YolosForObjectDetection
)
from huggingface_hub import InferenceClient
# ============================================================
# CONFIGURATION
# ============================================================
MODEL_NAME = "nickmuchi/yolos-small-finetuned-license-plate-detection"
EV_DISCOUNT = 50
LLM_MODEL = "openai/gpt-oss-120b"
# ============================================================
# HUGGING FACE LLM INITIALIZATION
# ============================================================
HF_TOKEN = os.getenv("HF_TOKEN")
if HF_TOKEN:
os.environ["HF_HUB_TOKEN"] = HF_TOKEN.strip()
llm_client = None
if HF_TOKEN and HF_TOKEN.strip():
try:
llm_client = InferenceClient(
api_key=HF_TOKEN.strip(),
provider="auto"
)
print("========================================")
print("✓ Hugging Face LLM initialized")
print("✓ Model:", LLM_MODEL)
print("========================================")
except Exception as e:
print("LLM initialization error:", str(e))
llm_client = None
else:
print("⚠ HF_TOKEN not found")
# ============================================================
# LOAD YOLOS MODEL
# ============================================================
processor = YolosImageProcessor.from_pretrained(
MODEL_NAME
)
model = YolosForObjectDetection.from_pretrained(
MODEL_NAME
)
model.eval()
print("✓ YOLOS model loaded")
# ============================================================
# 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()
# ============================================================
# PLATE COLOUR CLASSIFICATION
# ============================================================
def classify_plate_color(plate_img):
try:
img = np.array(plate_img)
if img.size == 0:
return "Non-EV"
# RGB → HSV
hsv = cv2.cvtColor(
img,
cv2.COLOR_RGB2HSV
)
h, w = hsv.shape[:2]
if w == 0:
return "Non-EV"
# Inspect the left portion of the plate
left_width = max(
1,
int(w * 0.20)
)
left = hsv[:, :left_width]
# Green colour range
green = cv2.inRange(
left,
(35, 40, 40),
(90, 255, 255)
)
green_ratio = (
np.count_nonzero(green)
/ green.size
)
# EV if sufficient green pixels
if green_ratio > 0.15:
return "EV"
return "Non-EV"
except Exception as e:
print(
"Plate colour classification error:",
e
)
return "Non-EV"
# ============================================================
# DUPLICATE VEHICLE CHECK
# ============================================================
def is_duplicate_vehicle(plate_number):
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
try:
last_time = datetime.strptime(
row[0],
"%Y-%m-%d %H:%M:%S"
)
if (
datetime.now() - last_time
< timedelta(minutes=5)
):
return True
except Exception:
return False
return False
# ============================================================
# SAVE VEHICLE
# ============================================================
def save_vehicle(
plate,
status
):
if status == "EV":
discount = EV_DISCOUNT
else:
discount = 0
if is_duplicate_vehicle(plate):
return "Duplicate"
current_time = datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
cursor.execute("""
INSERT INTO vehicles(
timestamp,
license_plate,
vehicle_status,
discount
)
VALUES (?, ?, ?, ?)
""", (
current_time,
plate,
status,
discount
))
conn.commit()
return "Saved"
# ============================================================
# AGENT ANALYTICS TOOL 1
# OVERALL SUMMARY
# ============================================================
def get_database_summary():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return {
"total_vehicles": 0,
"ev_vehicles": 0,
"non_ev_vehicles": 0,
"ev_percentage": 0,
"non_ev_percentage": 0,
"total_discount": 0
}
total = len(df)
ev = len(
df[
df["vehicle_status"] == "EV"
]
)
non_ev = len(
df[
df["vehicle_status"] == "Non-EV"
]
)
total_discount = int(
df["discount"].sum()
)
return {
"total_vehicles": total,
"ev_vehicles": ev,
"non_ev_vehicles": non_ev,
"ev_percentage": round(
(ev / total) * 100,
2
),
"non_ev_percentage": round(
(non_ev / total) * 100,
2
),
"total_discount": total_discount
}
# ============================================================
# AGENT ANALYTICS TOOL 2
# TODAY'S ANALYSIS
# ============================================================
def get_today_analysis():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return {
"today_total": 0,
"today_ev": 0,
"today_non_ev": 0,
"today_ev_percentage": 0,
"today_discount": 0
}
df["timestamp"] = pd.to_datetime(
df["timestamp"],
errors="coerce"
)
today = datetime.now().date()
today_df = df[
df["timestamp"].dt.date == today
]
total = len(today_df)
ev = len(
today_df[
today_df["vehicle_status"] == "EV"
]
)
non_ev = len(
today_df[
today_df["vehicle_status"] == "Non-EV"
]
)
discount = int(
today_df["discount"].sum()
)
ev_percentage = (
round((ev / total) * 100, 2)
if total > 0
else 0
)
return {
"today_total": total,
"today_ev": ev,
"today_non_ev": non_ev,
"today_ev_percentage": ev_percentage,
"today_discount": discount
}
# ============================================================
# AGENT ANALYTICS TOOL 3
# EV ANALYSIS
# ============================================================
def get_ev_analysis():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return {
"total_vehicles": 0,
"ev_vehicles": 0,
"non_ev_vehicles": 0,
"ev_percentage": 0,
"non_ev_percentage": 0,
"ev_discount": 0
}
total = len(df)
ev_df = df[
df["vehicle_status"] == "EV"
]
non_ev_df = df[
df["vehicle_status"] == "Non-EV"
]
ev_count = len(ev_df)
non_ev_count = len(non_ev_df)
ev_percentage = round(
ev_count / total * 100,
2
)
non_ev_percentage = round(
non_ev_count / total * 100,
2
)
ev_discount = int(
ev_df["discount"].sum()
)
return {
"total_vehicles": total,
"ev_vehicles": ev_count,
"non_ev_vehicles": non_ev_count,
"ev_percentage": ev_percentage,
"non_ev_percentage": non_ev_percentage,
"ev_discount": ev_discount
}
# ============================================================
# AGENT ANALYTICS TOOL 4
# DISCOUNT ANALYSIS
# ============================================================
def get_discount_analysis():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return {
"total_discount": 0,
"ev_count": 0,
"discount_per_ev": EV_DISCOUNT
}
ev_count = len(
df[
df["vehicle_status"] == "EV"
]
)
total_discount = int(
df["discount"].sum()
)
return {
"total_discount": total_discount,
"ev_count": ev_count,
"discount_per_ev": EV_DISCOUNT
}
# ============================================================
# AGENT ANALYTICS TOOL 5
# TRAFFIC TREND
# ============================================================
def get_traffic_trend():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return []
df["timestamp"] = pd.to_datetime(
df["timestamp"],
errors="coerce"
)
df = df.dropna(
subset=["timestamp"]
)
if df.empty:
return []
df["date"] = (
df["timestamp"]
.dt.strftime("%Y-%m-%d")
)
trend = (
df.groupby("date")
.size()
.reset_index(
name="vehicles"
)
)
return trend.to_dict(
orient="records"
)
# ============================================================
# AGENT ANALYTICS TOOL 6
# EV TREND
# ============================================================
def get_ev_trend():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return []
df["timestamp"] = pd.to_datetime(
df["timestamp"],
errors="coerce"
)
df = df.dropna(
subset=["timestamp"]
)
ev_df = df[
df["vehicle_status"] == "EV"
]
if ev_df.empty:
return []
ev_df["date"] = (
ev_df["timestamp"]
.dt.strftime("%Y-%m-%d")
)
trend = (
ev_df.groupby("date")
.size()
.reset_index(
name="ev_vehicles"
)
)
return trend.to_dict(
orient="records"
)
# ============================================================
# AGENT ANALYTICS TOOL 7
# VEHICLE RECORDS
# ============================================================
def get_vehicle_records():
df = pd.read_sql(
"SELECT * FROM vehicles",
conn
)
if df.empty:
return []
return df.to_dict(
orient="records"
)
# ============================================================
# AGENT TOOL DEFINITIONS
# ============================================================
AGENT_TOOLS = [
"SUMMARY",
"TODAY",
"EV_ANALYSIS",
"DISCOUNT",
"TRAFFIC_TREND",
"EV_TREND",
"RECORDS"
]
# ============================================================
# LLM AGENT PLANNER
# ============================================================
def choose_agent_tool(user_query):
# Fallback when LLM is unavailable
if llm_client is None:
query = user_query.lower()
if (
"trend" in query
and "ev" in query
):
return "EV_TREND"
if "trend" in query:
return "TRAFFIC_TREND"
if (
"discount" in query
or "saving" in query
or "incentive" in query
):
return "DISCOUNT"
if (
"today" in query
or "current day" in query
):
return "TODAY"
if (
"ev" in query
or "electric" in query
or "adoption" in query
):
return "EV_ANALYSIS"
if (
"record" in query
or "plate" in query
or "vehicle details" in query
):
return "RECORDS"
return "SUMMARY"
system_prompt = """
You are the planning agent for a Smart Traffic
and EV Analytics System.
Your job is to select exactly ONE analytics tool.
Available tools:
SUMMARY
- Overall traffic statistics.
TODAY
- Statistics for today's detected vehicles.
EV_ANALYSIS
- EV count, Non-EV count, EV percentage,
and EV adoption.
DISCOUNT
- EV incentive, discount and savings.
TRAFFIC_TREND
- Number of vehicles detected over time.
EV_TREND
- Number of EV vehicles detected over time.
RECORDS
- Vehicle records and license plate records.
Rules:
1. Select exactly ONE tool.
2. Return ONLY the tool name.
3. Do not explain your decision.
4. Do not return JSON.
5. Do not return anything else.
"""
try:
response = llm_client.chat.completions.create(
model=LLM_MODEL,
messages=[
{
"role": "system",
"content": system_prompt
},
{
"role": "user",
"content": user_query
}
],
temperature=0,
max_tokens=30
)
result = (
response
.choices[0]
.message
.content
.strip()
.upper()
)
for tool in AGENT_TOOLS:
if tool in result:
return tool
return "SUMMARY"
except Exception as e:
print(
"Agent planner error:",
e
)
return "SUMMARY"
# ============================================================
# EXECUTE SELECTED ANALYTICS TOOL
# ============================================================
def execute_agent_tool(tool):
if tool == "SUMMARY":
return get_database_summary()
elif tool == "TODAY":
return get_today_analysis()
elif tool == "EV_ANALYSIS":
return get_ev_analysis()
elif tool == "DISCOUNT":
return get_discount_analysis()
elif tool == "TRAFFIC_TREND":
return get_traffic_trend()
elif tool == "EV_TREND":
return get_ev_trend()
elif tool == "RECORDS":
return get_vehicle_records()
return get_database_summary()
# ============================================================
# LLM ANALYST
# ============================================================
def generate_agent_response(
user_query,
selected_tool,
tool_result
):
if llm_client is None:
return (
"### ⚠️ Agentic AI Not Configured\n\n"
"The analytics system is working, but the "
"LLM is not configured.\n\n"
"**Please add `HF_TOKEN` to the Hugging Face "
"Space Secrets and restart the Space.**\n\n"
f"Selected analytics tool: **{selected_tool}**"
)
system_prompt = f"""
You are the AI analyst of a Smart Traffic
and EV Analytics System.
The system contains:
1. YOLOS license plate detection
2. OpenCV license plate colour analysis
3. EV / Non-EV classification
4. License plate recognition
5. EV discount calculation
6. SQLite traffic analytics
7. Agentic AI analytics
The prototype gives an EV incentive of ₹{EV_DISCOUNT}.
The user asked a question.
An analytics agent selected one tool:
{selected_tool}
You MUST use only the supplied analytics result.
Do not invent data.
Do not estimate missing values.
Do not create statistics that are not present
in the analytics result.
Answer the user's question directly.
Use this structure when appropriate:
**Answer**
Give the direct result.
**Key Statistics**
Give the relevant numbers.
**Interpretation**
Explain what the result means.
**Recommendation**
Give a short recommendation only when useful.
Keep the response concise and professional.
"""
prompt = f"""
User question:
{user_query}
Selected analytics tool:
{selected_tool}
Analytics result:
{json.dumps(
tool_result,
indent=2,
default=str
)}
Generate the final analytical response.
"""
try:
response = llm_client.chat.completions.create(
model=LLM_MODEL,
messages=[
{
"role": "system",
"content": system_prompt
},
{
"role": "user",
"content": prompt
}
],
temperature=0.2,
max_tokens=500
)
return (
response
.choices[0]
.message
.content
.strip()
)
except Exception as e:
print(
"LLM response error:",
e
)
return (
"### ⚠️ LLM Response Error\n\n"
f"{str(e)}"
)
# ============================================================
# DYNAMIC GRAPH GENERATOR
# ============================================================
def generate_agent_graph(
selected_tool
):
fig, ax = plt.subplots(
figsize=(9, 5)
)
# --------------------------------------------------------
# SUMMARY / EV ANALYSIS
# --------------------------------------------------------
if selected_tool in [
"SUMMARY",
"EV_ANALYSIS"
]:
result = get_database_summary()
labels = [
"EV",
"Non-EV"
]
values = [
result["ev_vehicles"],
result["non_ev_vehicles"]
]
ax.bar(
labels,
values
)
ax.set_title(
"EV vs Non-EV Vehicles"
)
ax.set_ylabel(
"Number of Vehicles"
)
for i, value in enumerate(values):
ax.text(
i,
value,
str(value),
ha="center",
va="bottom"
)
# --------------------------------------------------------
# TODAY
# --------------------------------------------------------
elif selected_tool == "TODAY":
result = get_today_analysis()
labels = [
"EV",
"Non-EV"
]
values = [
result["today_ev"],
result["today_non_ev"]
]
ax.bar(
labels,
values
)
ax.set_title(
"Today's Traffic"
)
ax.set_ylabel(
"Vehicles"
)
for i, value in enumerate(values):
ax.text(
i,
value,
str(value),
ha="center",
va="bottom"
)
# --------------------------------------------------------
# DISCOUNT
# --------------------------------------------------------
elif selected_tool == "DISCOUNT":
result = get_discount_analysis()
discount = result[
"total_discount"
]
ax.bar(
["Total EV Discount"],
[discount]
)
ax.set_title(
"EV Incentive / Discount"
)
ax.set_ylabel(
"Discount (₹)"
)
ax.text(
0,
discount,
f"₹{discount}",
ha="center",
va="bottom"
)
# --------------------------------------------------------
# TRAFFIC TREND
# --------------------------------------------------------
elif selected_tool == "TRAFFIC_TREND":
records = get_traffic_trend()
if not records:
ax.text(
0.5,
0.5,
"No traffic trend data available",
ha="center",
va="center"
)
ax.axis("off")
else:
dates = [
item["date"]
for item in records
]
vehicles = [
item["vehicles"]
for item in records
]
ax.plot(
dates,
vehicles,
marker="o"
)
ax.set_title(
"Traffic Trend"
)
ax.set_xlabel(
"Date"
)
ax.set_ylabel(
"Vehicles"
)
ax.tick_params(
axis="x",
rotation=45
)
# --------------------------------------------------------
# EV TREND
# --------------------------------------------------------
elif selected_tool == "EV_TREND":
records = get_ev_trend()
if not records:
ax.text(
0.5,
0.5,
"No EV trend data available",
ha="center",
va="center"
)
ax.axis("off")
else:
dates = [
item["date"]
for item in records
]
ev_values = [
item["ev_vehicles"]
for item in records
]
ax.plot(
dates,
ev_values,
marker="o"
)
ax.set_title(
"EV Adoption Trend"
)
ax.set_xlabel(
"Date"
)
ax.set_ylabel(
"EV Vehicles"
)
ax.tick_params(
axis="x",
rotation=45
)
# --------------------------------------------------------
# RECORDS
# --------------------------------------------------------
elif selected_tool == "RECORDS":
result = get_database_summary()
labels = [
"EV",
"Non-EV"
]
values = [
result["ev_vehicles"],
result["non_ev_vehicles"]
]
ax.bar(
labels,
values
)
ax.set_title(
"Vehicle Records"
)
ax.set_ylabel(
"Number of Records"
)
else:
ax.text(
0.5,
0.5,
"No visualization available",
ha="center",
va="center"
)
ax.axis("off")
plt.tight_layout()
return fig
# ============================================================
# RUN AGENTIC DASHBOARD
# ============================================================
def run_agentic_dashboard(
user_query
):
if not user_query:
return (
"### Please enter a question\n\n"
"Ask the Traffic AI Agent a question.",
None
)
user_query = user_query.strip()
# --------------------------------------------------------
# STEP 1: LLM PLANNER
# --------------------------------------------------------
selected_tool = choose_agent_tool(
user_query
)
print(
"Agent selected:",
selected_tool
)
# --------------------------------------------------------
# STEP 2: EXECUTE ANALYTICS TOOL
# --------------------------------------------------------
tool_result = execute_agent_tool(
selected_tool
)
# --------------------------------------------------------
# STEP 3: LLM ANALYST
# --------------------------------------------------------
answer = generate_agent_response(
user_query,
selected_tool,
tool_result
)
# --------------------------------------------------------
# STEP 4: GENERATE DYNAMIC GRAPH
# --------------------------------------------------------
graph = generate_agent_graph(
selected_tool
)
return answer, graph
# ============================================================
# MAIN VEHICLE PROCESSING PIPELINE
# ============================================================
def process_image(img):
if img is None:
return (
None,
"Please upload an image.",
"0",
"0",
"₹0"
)
try:
image = Image.fromarray(
img
).convert("RGB")
# ----------------------------------------------------
# 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.3,
target_sizes=target_sizes
)[0]
)
draw = ImageDraw.Draw(
image
)
ev_count = 0
non_ev_count = 0
discount_total = 0
output_text = ""
# ----------------------------------------------------
# NO DETECTION
# ----------------------------------------------------
if len(results["boxes"]) == 0:
return (
image,
"No license plate detected.",
"0",
"0",
"₹0"
)
# ----------------------------------------------------
# PROCESS DETECTED PLATES
# ----------------------------------------------------
for i, box in enumerate(
results["boxes"]
):
x1, y1, x2, y2 = map(
int,
box.tolist()
)
# Prevent invalid coordinates
x1 = max(
0,
x1
)
y1 = max(
0,
y1
)
x2 = min(
image.width,
x2
)
y2 = min(
image.height,
y2
)
if x2 <= x1 or y2 <= y1:
continue
# Crop license plate
plate = image.crop(
(
x1,
y1,
x2,
y2
)
)
# ------------------------------------------------
# EV / NON-EV CLASSIFICATION
# ------------------------------------------------
status = classify_plate_color(
plate
)
# ------------------------------------------------
# TEMPORARY PLATE NUMBER
#
# Replace this section with OCR later.
# ------------------------------------------------
plate_number = (
f"Vehicle_"
f"{datetime.now().strftime('%H%M%S%f')}_"
f"{i}"
)
# ------------------------------------------------
# SAVE DATABASE RECORD
# ------------------------------------------------
saved = save_vehicle(
plate_number,
status
)
if saved == "Duplicate":
continue
# ------------------------------------------------
# EV
# ------------------------------------------------
if status == "EV":
ev_count += 1
discount = EV_DISCOUNT
discount_total += (
discount
)
box_color = "green"
label = (
f"{plate_number}\n"
f"EV | ₹{discount}"
)
# ------------------------------------------------
# NON-EV
# ------------------------------------------------
else:
non_ev_count += 1
discount = 0
box_color = "red"
label = (
f"{plate_number}\n"
f"Non-EV"
)
# ------------------------------------------------
# DRAW BOUNDING BOX
# ------------------------------------------------
draw.rectangle(
[
x1,
y1,
x2,
y2
],
outline=box_color,
width=3
)
# ------------------------------------------------
# DRAW LABEL
# ------------------------------------------------
draw.text(
(
x1,
max(
0,
y1 - 35
)
),
label,
fill=box_color
)
# ------------------------------------------------
# TEXT SUMMARY
# ------------------------------------------------
output_text += (
f"Vehicle {i + 1}\n"
f"Plate : "
f"{plate_number}\n"
f"Status : "
f"{status}\n"
f"Discount : "
f"₹{discount}\n\n"
)
# ----------------------------------------------------
# DATABASE COMMIT
# ----------------------------------------------------
conn.commit()
# ----------------------------------------------------
# IF NO NEW VEHICLES
# ----------------------------------------------------
if (
ev_count == 0
and non_ev_count == 0
):
output_text = (
"All detected vehicles "
"were duplicate records "
"within the 5-minute window."
)
return (
image,
output_text,
str(ev_count),
str(non_ev_count),
f"₹{discount_total}"
)
except Exception as e:
print(
"Image processing error:",
e
)
return (
img,
f"Processing error:\n{str(e)}",
"0",
"0",
"₹0"
)
# ============================================================
# CUSTOM 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 USER INTERFACE
# ============================================================
with gr.Blocks(
title="Smart Traffic & EV Analytics"
) as demo:
# ========================================================
# TITLE
# ========================================================
gr.Markdown(
"""
# 🚗 Smart Traffic & EV Analytics System
### YOLOS License Plate Detection • OpenCV EV Classification •
SQLite Analytics • EV Incentives • Agentic AI Dashboard
"""
)
# ========================================================
# IMAGE PROCESSING SECTION
# ========================================================
with gr.Row(
equal_height=True
):
# ----------------------------------------------------
# INPUT
# ----------------------------------------------------
with gr.Column(
scale=1
):
input_img = gr.Image(
type="numpy",
label="Input Image",
sources=[
"upload",
"webcam"
],
height=350
)
btn = gr.Button(
" Scan Vehicle",
variant="primary",
size="lg"
)
# ----------------------------------------------------
# OUTPUT IMAGE
# ----------------------------------------------------
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
)
# ========================================================
# VEHICLE STATISTICS
# ========================================================
gr.Markdown(
"## Detection Statistics"
)
with gr.Row():
ev_box = gr.Textbox(
label="EV Vehicles",
value="0"
)
non_ev_box = gr.Textbox(
label="Non-EV Vehicles",
value="0"
)
discount_box = gr.Textbox(
label="EV Discount",
value="₹0"
)
# ========================================================
# AGENTIC AI DASHBOARD
# ========================================================
gr.Markdown(
"""
## Agentic AI Traffic Dashboard
Ask questions in natural language. The AI agent
will select the appropriate analytics tool, retrieve
data from the SQLite database, generate an analytical
response, and create a relevant visualization.
"""
)
# --------------------------------------------------------
# SAMPLE QUESTIONS
# --------------------------------------------------------
gr.Markdown(
"""
**Example questions:**
- How many EV vehicles were detected?
- What percentage of vehicles are EVs?
- What is the total discount?
- How many vehicles were detected today?
- Show today's traffic.
- Show the traffic trend.
- Show the EV adoption trend.
- Give me an overall traffic report.
- Show the vehicle records.
"""
)
# --------------------------------------------------------
# QUERY BOX
# --------------------------------------------------------
agent_query = gr.Textbox(
label="Ask the Traffic AI Agent",
placeholder=(
"Example: "
"What percentage of vehicles are EVs?"
),
lines=3
)
# --------------------------------------------------------
# ASK BUTTON
# --------------------------------------------------------
agent_button = gr.Button(
"Ask AI Agent",
variant="primary",
size="lg"
)
# ========================================================
# AI RESPONSE
# ========================================================
gr.Markdown(
"### AI Generated Analysis"
)
agent_response = gr.Markdown(
"Ask a question to generate "
"an AI-powered traffic report."
)
# ========================================================
# DYNAMIC VISUALIZATION
# ========================================================
gr.Markdown(
"### Dynamic Visualization"
)
agent_graph = gr.Plot(
label="AI Dynamic Analytics"
)
# ========================================================
# SCAN BUTTON EVENT
# ========================================================
btn.click(
fn=process_image,
inputs=input_img,
outputs=[
output_img,
summary_box,
ev_box,
non_ev_box,
discount_box
]
)
# ========================================================
# AGENT BUTTON EVENT
# ========================================================
agent_button.click(
fn=run_agentic_dashboard,
inputs=agent_query,
outputs=[
agent_response,
agent_graph
]
)
# ============================================================
# LAUNCH
# ============================================================
if __name__ == "__main__":
demo.queue()
demo.launch(
ssr_mode=False,
css=css
)
# ============================================================
# Smart Traffic & EV Analytics System
# Agentic AI Dynamic Dashboard
# Author: Sarvamangala Kokatanur
# ============================================================
# ---------------- IMPORT LIBRARIES ----------------
# ============================================================
# IMPORT LIBRARIES
# ============================================================
import os
import json
import sqlite3
import gradio as gr
import torch
import numpy as np
import cv2
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
from PIL import Image, ImageDraw
from transformers import (
YolosImageProcessor,
YolosForObjectDetection
)
from huggingface_hub import InferenceClient
# ============================================================
# CONFIGURATION
# ============================================================
MODEL_NAME = "nickmuchi/yolos-small-finetuned-license-plate-detection"
EV_DISCOUNT = 50
LLM_MODEL = "openai/gpt-oss-120b"
# ============================================================
# HUGGING FACE LLM INITIALIZATION
# ============================================================
HF_TOKEN = os.getenv("HF_TOKEN")
if HF_TOKEN:
os.environ["HF_HUB_TOKEN"] = HF_TOKEN.strip()
llm_client = None
if HF_TOKEN and HF_TOKEN.strip():
try:
llm_client = InferenceClient(
api_key=HF_TOKEN.strip(),
provider="auto"
)
print("========================================")
print("✓ Hugging Face LLM initialized")
print("✓ Model:", LLM_MODEL)
print("========================================")
except Exception as e:
print("LLM initialization error:", str(e))
llm_client = None
else:
print("⚠ HF_TOKEN not found")
# ============================================================
# LOAD YOLOS MODEL
# ============================================================
processor = YolosImageProcessor.from_pretrained(
MODEL_NAME
)
model = YolosForObjectDetection.from_pretrained(
MODEL_NAME
)
model.eval()
print("✓ YOLOS model loaded")
# ============================================================