"""ParkSight — Parking-Induced Congestion Intelligence for Bengaluru.
Serving layer: reads only the small precomputed artifacts in data/processed/.
Run locally: streamlit run app.py
"""
import io
import re
import json
import hashlib
from datetime import datetime
from pathlib import Path
import pandas as pd
import pydeck as pdk
import plotly.graph_objects as go
import streamlit as st
import streamlit.components.v1 as components
try:
from streamlit_option_menu import option_menu
HAS_MENU = True
except Exception:
HAS_MENU = False
try:
from streamlit_mic_recorder import speech_to_text
HAS_MIC = True
except Exception:
HAS_MIC = False
try:
from streamlit_autorefresh import st_autorefresh
HAS_AUTOREFRESH = True
except Exception:
HAS_AUTOREFRESH = False
# Make this folder importable no matter where the app is launched from
# (fixes "ModuleNotFoundError: No module named 'src'" on some setups).
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from src.i18n import LANGS, SPEECH_LANG, t, build_area_vocab, resolve_area
from src import ops
import time as _time
ROOT = Path(__file__).resolve().parent
PROC = ROOT / "data" / "processed"
VIOLET = "#8B5CF6"
GOLD = "#D4AF37"
DONUT_COLORS = ["#8B5CF6", "#22D3EE", "#EC4899", "#6366F1", "#0EA5E9",
"#F472B6", "#A855F7", "#34D399", "#FB7185", "#818CF8"]
# ---- demo-grade login (self-contained, no fragile dependency) ----
# NOTE: demonstration auth — credentials are shown on the login screen so judges
# can always get in. Production would use a real identity provider. To DISABLE
# the login entirely, comment out the `require_login()` call below.
_USERS = {"admin": "Traffic Admin (BTP)", "officer": "Patrol Officer"}
_PW = {"admin": hashlib.sha256(b"admin123").hexdigest(),
"officer": hashlib.sha256(b"officer123").hexdigest()}
def require_login():
if st.session_state.get("authed"):
return
# keep the session logged in across theme-switch page reloads (demo only)
if st.query_params.get("auth") == "1":
st.session_state.authed = True
st.session_state.user = "Traffic Admin (BTP)"
return
st.markdown(
"
"
"
दृष्टि — DRISHTI
"
"
Digital Real-time Intelligence for "
"Smart Hotspot & Traffic Insights · हर सड़क पर नज़र, हर सफ़र आसान
",
unsafe_allow_html=True)
st.subheader("🔐 Secure sign-in")
with st.form("login_form"):
u = st.text_input("Username")
p = st.text_input("Password", type="password")
ok = st.form_submit_button("Sign in")
if ok:
if u in _PW and _PW[u] == hashlib.sha256(p.encode()).hexdigest():
st.session_state.authed = True
st.session_state.user = _USERS[u]
st.query_params["auth"] = "1"
st.rerun()
st.error("Invalid username or password.")
st.caption("Demo credentials — sign in with either account:")
st.table(pd.DataFrame(
{"USERNAME": ["admin", "officer"], "PASSWORD": ["admin123", "officer123"]}
).set_index("USERNAME"))
st.stop()
# sidebar starts open so the nav is always visible
st.set_page_config(page_title="DRISHTI · Bengaluru", page_icon="🚦",
layout="wide", initial_sidebar_state="expanded")
# ---------------- data ----------------
@st.cache_data
def load():
hot = pd.read_parquet(PROC / "hotspots.parquet")
fc = pd.read_parquet(PROC / "forecast.parquet")
off = pd.read_parquet(PROC / "offenders.parquet")
meta = json.loads((PROC / "meta.json").read_text(encoding="utf-8"))
fc = fc.merge(hot[["h3", "location", "junction_name", "cii"]], on="h3", how="left")
return hot, fc, off, meta
@st.cache_data
def load_trends():
return pd.read_parquet(PROC / "trends.parquet")
@st.cache_data
def load_byday():
return pd.read_parquet(PROC / "trends_byday.parquet")
def cii_color(cii):
x = max(0.0, min(1.0, cii / 100.0))
g, y, r = (22, 163, 74), (245, 158, 11), (220, 38, 38)
if x < 0.5:
f, a, b = x / 0.5, g, y
else:
f, a, b = (x - 0.5) / 0.5, y, r
return [int(a[i] + (b[i] - a[i]) * f) for i in range(3)] + [185]
def cii_to_hex(c):
r, g, b, _ = cii_color(c)
return f"rgb({r},{g},{b})"
# ---------------- theme CSS (theme-AGNOSTIC: adapts to light & dark) ----------------
def inject_css():
# No hardcoded background/text colours -> the native Light/Dark theme (⋮ menu)
# stays fully consistent, including tables. Only shape + accent + font here.
st.markdown("""
""", unsafe_allow_html=True)
# ---------------- voice ----------------
def play_tts(text, lang_code):
"""Reliable, multilingual TTS via gTTS (plays an MP3). Browser fallback if offline."""
try:
from gtts import gTTS
buf = io.BytesIO()
gTTS(text=text, lang=lang_code).write_to_fp(buf)
st.audio(buf.getvalue(), format="audio/mp3", autoplay=True)
return True
except Exception:
loc = {"en": "en-IN", "hi": "hi-IN", "kn": "kn-IN"}.get(lang_code, "en-IN")
safe = json.dumps(text)
components.html(f"""""",
height=0)
return False
def parse_command(text):
low = (text or "").lower().strip()
out = {"speak": any(w in low for w in ["read", "speak", "say", "aloud", "tell",
"ಮಾತ", "ಓದ", "बोल", "पढ"])}
# which language to SPEAK the answer in (overrides the UI language)
if any(w in low for w in ["hindi", "हिंदी", "हिन्दी", "हिंदी में"]):
out["say_lang"] = "hi"
elif any(w in low for w in ["kannada", "ಕನ್ನಡ", "kannad"]):
out["say_lang"] = "kn"
elif "english" in low:
out["say_lang"] = "en"
m = re.search(r"(?:top|ಮೇಲಿನ|शीर्ष)\s*(\d+)", low)
if m:
out["topn"] = max(5, min(50, int(m.group(1))))
if any(w in low for w in ["worst", "high impact", "critical", "severe",
"ಕೆಟ್ಟ", "खराब", "गंभीर"]):
out["min_cii"] = 80
return out
def speak_summary(rows, say_lang, n):
"""Build a spoken summary of the top-n zones in the requested language."""
rows = rows.head(n)
if say_lang == "hi":
parts = [f"शीर्ष {len(rows)} क्षेत्र।"]
for i, (_, r) in enumerate(rows.iterrows(), 1):
parts.append(f"{i}. {r['location'].split(',')[0]}, "
f"सी आई आई {r['cii']:.0f}, {int(r['n_violations'])} उल्लंघन।")
return " ".join(parts)
if say_lang == "kn":
parts = [f"ಮೇಲಿನ {len(rows)} ಪ್ರದೇಶಗಳು."]
for i, (_, r) in enumerate(rows.iterrows(), 1):
parts.append(f"{i}. {r['location'].split(',')[0]}, "
f"ಸಿ ಐ ಐ {r['cii']:.0f}, {int(r['n_violations'])} ಉಲ್ಲಂಘನೆ.")
return " ".join(parts)
nums = ["one", "two", "three", "four", "five", "six", "seven", "eight"]
parts = [f"Top {len(rows)} zones."]
for i, (_, r) in enumerate(rows.iterrows()):
label = nums[i] if i < len(nums) else str(i + 1)
parts.append(f"{label}: {r['location'].split(',')[0]}, "
f"C I I {r['cii']:.0f}, {int(r['n_violations'])} violations.")
return " ".join(parts)
# ---------------- plotly helpers (let theme="streamlit" adapt to light/dark) ----------------
def _layout(fig, height=240, title=None):
fig.update_layout(height=height, margin=dict(l=10, r=10, t=36 if title else 8, b=8),
title=dict(text=title, font=dict(size=14)) if title else None,
paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)",
showlegend=False)
return fig
def line_chart(daily, title, mark_last=False):
fig = go.Figure(go.Scatter(x=daily["label"], y=daily["value"], mode="lines",
line=dict(color=VIOLET, width=2.5), fill="tozeroy",
fillcolor="rgba(139,92,246,0.22)"))
if mark_last and len(daily):
row = daily.iloc[-1]
fig.add_trace(go.Scatter(x=[row["label"]], y=[row["value"]], mode="markers",
marker=dict(color="#F472B6", size=13)))
fig.update_xaxes(showgrid=False)
return _layout(fig, title=title)
def bar_chart(d, title, ramp=False):
colors = [cii_to_hex(v / (max(d["value"]) or 1) * 100) for v in d["value"]] if ramp else VIOLET
fig = go.Figure(go.Bar(x=d["label"], y=d["value"], marker_color=colors))
fig.update_xaxes(showgrid=False)
return _layout(fig, title=title)
def rainbow_gauge(value, title):
stops = [(0.0, (34, 211, 238)), (0.4, (52, 211, 153)),
(0.7, (250, 204, 21)), (1.0, (244, 114, 182))]
def lerp(tt):
for i in range(len(stops) - 1):
t0, c0 = stops[i]
t1, c1 = stops[i + 1]
if tt <= t1:
f = (tt - t0) / (t1 - t0 + 1e-9)
return tuple(int(c0[j] + (c1[j] - c0[j]) * f) for j in range(3))
return stops[-1][1]
seg = 28
steps = [{"range": [i / seg * 100, (i + 1) / seg * 100],
"color": f"rgb{lerp((i + 0.5) / seg)}"} for i in range(seg)]
fig = go.Figure(go.Indicator(
mode="gauge+number", value=value, number={"suffix": "%", "font": {"size": 38}},
gauge={"axis": {"range": [0, 100], "tickwidth": 0},
"bar": {"color": "rgba(255,255,255,0)"}, "borderwidth": 0, "steps": steps}))
return _layout(fig, height=250, title=title)
# ---------------- gradient KPI card + donut ----------------
def sparkline_svg(series, color, w=120, h=36):
if not series or len(series) < 2:
return ""
mn, mx = min(series), max(series)
rng = (mx - mn) or 1
pts = " ".join(
f"{i/(len(series)-1)*w:.1f},{h - (v-mn)/rng*(h-7) - 4:.1f}"
for i, v in enumerate(series))
last = pts.split()[-1]
return (f"")
def bars_svg(vals, colors, w=120, h=36):
mx = max(vals) or 1
bw = w / (len(vals) * 1.7)
gap = bw * 0.7
rects = "".join(
f""
for i, v in enumerate(vals))
return f""
def _delta(series, lower_is_better=True):
if not series or len(series) < 14:
return ""
k = min(30, len(series) // 2)
recent = sum(series[-k:]) / k
prior = sum(series[-2 * k:-k]) / k
if prior == 0:
return ""
pct = (recent - prior) / prior * 100
up = pct >= 0
good = (not up) if lower_is_better else up
color = "#34D399" if good else "#F87171"
return (f"{'▲' if up else '▼'} "
f"{abs(pct):.1f}%"
f"vs prev {k}d")
def kpi_card(icon, label, value, accent=VIOLET, series=None, viz="spark",
sub=None, lower_is_better=True):
if viz == "bars" and series:
chart = bars_svg(series, ["rgba(170,170,190,0.30)", accent])
elif series:
chart = sparkline_svg(series, accent)
else:
chart = ""
delta = sub if sub else _delta(series, lower_is_better)
delta_html = (f"
{delta}
"
if delta else "")
st.markdown(
f"
"
f"
"
f"
{icon}
"
f"
{chart}
"
f"
{label}
"
f"
{value}
"
f"{delta_html}
", unsafe_allow_html=True)
def donut(labels, values, title):
fig = go.Figure(go.Pie(labels=list(labels), values=list(values), hole=0.62,
marker=dict(colors=DONUT_COLORS), textinfo="percent",
textfont=dict(color="#fff", size=12), sort=True))
fig.update_layout(height=300, margin=dict(l=10, r=10, t=42, b=10),
title=dict(text=title, font=dict(size=14)),
paper_bgcolor="rgba(0,0,0,0)", plot_bgcolor="rgba(0,0,0,0)",
legend=dict(orientation="v", x=1, y=0.5, font=dict(size=11)))
return fig
def active_theme():
"""Active theme ('light'/'dark'); defaults to dark on first run / older Streamlit."""
try:
tp = st.context.theme.type
return tp if tp in ("light", "dark") else "dark"
except Exception:
return "dark"
def theme_css(mode):
"""Light / system overrides layered on top of the dark default.
No !important on text colours, so inline-coloured bits (logo, KPI accents,
status cards) keep their colours; only the *defaults* get recoloured."""
light = """
.stApp { background-color:#F6F3FF; }
.stApp p, .stApp li, .stApp label,
.stApp h1, .stApp h2, .stApp h3, .stApp h4 { color:#241748; }
[data-testid="stCaptionContainer"], [data-testid="stCaptionContainer"] p { color:#5f548b; }
section[data-testid="stSidebar"] > div:first-child { background:#ECE6FB; }
section[data-testid="stSidebar"], section[data-testid="stSidebar"] p,
section[data-testid="stSidebar"] label, section[data-testid="stSidebar"] div { color:#2b1d55; }
[data-testid="stSegmentedControl"] button p { color:#2b1d55; }
.drishti-sub { color:#2563EB !important; }
"""
if mode == "light":
return f""
if mode == "system":
return f""
return ""
# ==================== APP ====================
inject_css()
require_login() # comment this line out to disable the login gate
hot, fc, off, meta = load()
THEME = active_theme() # follows the ⋮ menu (top-right) -> Settings -> Theme
ACCENT = "#FACC15" if THEME == "dark" else "#2563EB" # gold text -> yellow (dark) / blue (light)
MAP_STYLE = "dark" if THEME == "dark" else "light"
NAV_COLOR = "#9B8FC2" if THEME == "dark" else "#4C3A82"
area_vocab = build_area_vocab(hot)
hot["fill"] = hot["cii"].apply(cii_color)
mm = meta["model_metrics"]
NAV_KEYS = ["tab_map", "tab_ops", "tab_trends", "tab_rank", "tab_off", "tab_fc",
"tab_patrol", "tab_whatif", "tab_event"]
NAV_ICONS = ["geo-alt-fill", "broadcast", "graph-up", "list-check",
"exclamation-triangle-fill", "magic",
"signpost-split-fill", "sliders", "calendar-event-fill"]
with st.sidebar:
st.markdown(
"