import streamlit as st import cv2 import numpy as np from ultralytics import YOLO import requests import math import time import os from tempfile import NamedTemporaryFile import folium from streamlit_folium import st_folium # ============================================================ # CONFIGURATION # ============================================================ MODEL_PATH = "model/best.pt" CLASS_NAMES = {0: "Accident", 1: "Non-accident", 2: "Fire"} ALERT_CLASSES = {"Accident", "Fire"} DEFAULT_CONFIDENCE = 0.3 HOSPITAL_RADIUS_M = 5000 # ============================================================ # PAGE CONFIG # ============================================================ st.set_page_config( page_title="AccidentAI β€” Real-Time Detection & Alert", page_icon="🚨", layout="wide", initial_sidebar_state="expanded", ) # ============================================================ # CUSTOM CSS # ============================================================ st.markdown(""" """, unsafe_allow_html=True) # ============================================================ # MODEL LOADING # ============================================================ @st.cache_resource def load_model(): if not os.path.exists(MODEL_PATH): st.error(f"Model file not found at `{MODEL_PATH}`.") st.stop() return YOLO(MODEL_PATH) model = load_model() # ============================================================ # UTILITY FUNCTIONS # ============================================================ def haversine_km(lat1, lon1, lat2, lon2): """Return distance in km between two lat/lon points.""" R = 6371 dlat = math.radians(lat2 - lat1) dlon = math.radians(lon2 - lon1) a = ( math.sin(dlat / 2) ** 2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2 ) return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) # ============================================================ # DETECTION FUNCTIONS # ============================================================ def detect_in_image(image_array, conf): """Run YOLO on a single image. Returns (detections, annotated_rgb).""" results = model.predict(image_array, conf=conf, verbose=False) detections = [] for result in results: for box in result.boxes: cls_id = int(box.cls[0]) detections.append( { "label": CLASS_NAMES.get(cls_id, "Unknown"), "confidence": float(box.conf[0]), } ) annotated_bgr = results[0].plot() annotated_rgb = cv2.cvtColor(annotated_bgr, cv2.COLOR_BGR2RGB) return detections, annotated_rgb def detect_in_video(video_path, conf, frame_skip=4, progress_cb=None): """Run YOLO on video frames. Returns (detections, best_annotated_rgb).""" cap = cv2.VideoCapture(video_path) total = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) or 1 detections = [] best_frame = None best_conf = 0 idx = 0 while cap.isOpened(): ret, frame = cap.read() if not ret: break if idx % frame_skip == 0: results = model.predict(frame, conf=conf, verbose=False) for result in results: for box in result.boxes: cls_id = int(box.cls[0]) c = float(box.conf[0]) label = CLASS_NAMES.get(cls_id, "Unknown") if label in ALERT_CLASSES: detections.append( {"label": label, "confidence": c, "frame": idx} ) if c > best_conf: best_conf = c best_frame = results[0].plot() idx += 1 if progress_cb: progress_cb(min(idx / total, 1.0)) cap.release() if best_frame is not None: best_frame = cv2.cvtColor(best_frame, cv2.COLOR_BGR2RGB) return detections, best_frame # ============================================================ # LOCATION & HOSPITAL FUNCTIONS # ============================================================ def get_ip_location(): """Free IP-based geolocation (no API key required).""" try: r = requests.get("http://ip-api.com/json/", timeout=5) d = r.json() if d.get("status") == "success": return { "lat": d["lat"], "lon": d["lon"], "city": d.get("city", ""), "region": d.get("regionName", ""), "country": d.get("country", ""), } except Exception: pass return None def fetch_hospitals(lat, lon, radius_m=5000): """Query OpenStreetMap Overpass API for hospitals within radius.""" query = f""" [out:json][timeout:10]; ( node["amenity"="hospital"](around:{radius_m},{lat},{lon}); way["amenity"="hospital"](around:{radius_m},{lat},{lon}); relation["amenity"="hospital"](around:{radius_m},{lat},{lon}); ); out center body; """ try: r = requests.post( "https://overpass-api.de/api/interpreter", data={"data": query}, timeout=15, ) elements = r.json().get("elements", []) except Exception: return [] hospitals = [] for el in elements: tags = el.get("tags", {}) if el["type"] == "node": h_lat, h_lon = el["lat"], el["lon"] else: c = el.get("center", {}) h_lat = c.get("lat", lat) h_lon = c.get("lon", lon) dist = haversine_km(lat, lon, h_lat, h_lon) hospitals.append( { "name": tags.get("name", "Unnamed Hospital"), "lat": h_lat, "lon": h_lon, "distance_km": round(dist, 2), "phone": tags.get("phone", tags.get("contact:phone", "β€”")), "beds": tags.get("beds", "β€”"), "emergency": tags.get("emergency", "unknown"), } ) hospitals.sort(key=lambda h: h["distance_km"]) return hospitals def build_map(lat, lon, hospitals, radius_m=5000): """Create a folium map with accident marker, radius circle, and hospitals.""" m = folium.Map(location=[lat, lon], zoom_start=14, tiles="CartoDB dark_matter") # Accident location folium.Marker( [lat, lon], popup="🚨 Accident", icon=folium.Icon(color="red", icon="exclamation-triangle", prefix="fa"), ).add_to(m) # Radius folium.Circle( [lat, lon], radius=radius_m, color="#ff3b30", fill=True, fill_opacity=0.06, weight=2, dash_array="6 4", ).add_to(m) # Hospitals for h in hospitals: folium.Marker( [h["lat"], h["lon"]], popup=f"πŸ₯ {h['name']} β€” {h['distance_km']} km", icon=folium.Icon(color="green", icon="plus-square", prefix="fa"), ).add_to(m) return m # ============================================================ # SIDEBAR # ============================================================ with st.sidebar: st.markdown("## βš™οΈ Detection Settings") confidence = st.slider( "Confidence threshold", 0.10, 0.90, DEFAULT_CONFIDENCE, 0.05 ) st.markdown("---") st.markdown("## πŸ“ Location") st.caption( "Location is auto-detected via IP. On cloud deployments this returns " "the server location β€” use manual override for accurate testing." ) use_manual = st.checkbox("Use manual coordinates") if use_manual: manual_lat = st.number_input("Latitude", value=28.6139, format="%.4f") manual_lon = st.number_input("Longitude", value=77.2090, format="%.4f") else: manual_lat, manual_lon = None, None st.markdown("---") st.markdown("## ℹ️ About") st.markdown( "**AccidentAI** uses a custom-trained **YOLOv8** model to detect " "accidents and fires in CCTV footage, then automatically locates " "nearby hospitals and sends simulated emergency alerts." ) st.markdown( "Built with Ultralytics, Streamlit, OpenStreetMap Overpass API, and Folium." ) # ============================================================ # HERO # ============================================================ st.markdown( """

🚨 AccidentAI

Real-Time Accident Detection & Emergency Alert System

YOLOv8 CCTV Analysis Hospital Alerts Live Map
""", unsafe_allow_html=True, ) # ============================================================ # FILE UPLOAD # ============================================================ uploaded = st.file_uploader( "Upload CCTV footage or image", type=["jpg", "jpeg", "png", "mp4", "avi", "mov"], help="Supported formats: JPG, PNG images and MP4, AVI, MOV videos", ) if not uploaded: st.info("πŸ‘† Upload an image or video to start detection.") st.stop() # ============================================================ # RUN DETECTION # ============================================================ is_video = uploaded.name.lower().endswith((".mp4", ".avi", ".mov")) if is_video: with NamedTemporaryFile(delete=False, suffix=".mp4") as tmp: tmp.write(uploaded.read()) tmp_path = tmp.name st.markdown("### 🎬 Analyzing video frames…") pbar = st.progress(0) detections, annotated = detect_in_video( tmp_path, confidence, progress_cb=pbar.progress ) pbar.empty() try: os.unlink(tmp_path) except OSError: pass else: raw = np.asarray(bytearray(uploaded.read()), dtype=np.uint8) image = cv2.imdecode(raw, cv2.IMREAD_COLOR) detections, annotated = detect_in_image(image, confidence) alert_dets = [d for d in detections if d["label"] in ALERT_CLASSES] # ============================================================ # RESULTS # ============================================================ st.markdown("---") col_img, col_stats = st.columns([2, 1], gap="large") with col_img: st.markdown("### πŸ” Detection Output") if annotated is not None: st.image(annotated, use_container_width=True) elif is_video: st.info("No hazard frames captured β€” video appears safe.") with col_stats: st.markdown("### πŸ“Š Analysis") n_acc = sum(1 for d in alert_dets if d["label"] == "Accident") n_fire = sum(1 for d in alert_dets if d["label"] == "Fire") total = len(detections) st.markdown( f"""
{n_acc}
Accidents
{n_fire}
Fires
{total}
Total Detections
""", unsafe_allow_html=True, ) if alert_dets: peak = max(d["confidence"] for d in alert_dets) avg = sum(d["confidence"] for d in alert_dets) / len(alert_dets) st.metric("Peak Confidence", f"{peak:.1%}") st.metric("Avg Confidence", f"{avg:.1%}") if is_video: frames_hit = len(set(d.get("frame", 0) for d in alert_dets)) st.metric("Frames with Hazards", frames_hit) else: st.success("βœ… No hazards detected in the uploaded media.") # ============================================================ # EMERGENCY ALERT SYSTEM # ============================================================ if not alert_dets: st.stop() st.markdown("---") # Alert banner top_label = "Accident" if n_acc else "Fire" st.markdown( f"""

🚨 EMERGENCY β€” {top_label} Detected

Initiating automated alert protocol β€’ Searching hospitals within 5 km radius

""", unsafe_allow_html=True, ) # -- Get location -- if use_manual and manual_lat is not None: loc = { "lat": manual_lat, "lon": manual_lon, "city": "Manual", "region": "", "country": "", } else: with st.spinner("πŸ“ Detecting location…"): loc = get_ip_location() if loc is None: st.warning( "Could not detect location automatically. " "Enable **manual coordinates** in the sidebar." ) st.stop() loc_str = ", ".join(filter(None, [loc["city"], loc["region"], loc["country"]])) st.markdown( f"**πŸ“ Incident Location:** {loc_str}  |  " f"`{loc['lat']:.4f}, {loc['lon']:.4f}`" ) # -- Fetch hospitals -- with st.spinner("πŸ₯ Querying OpenStreetMap for nearby hospitals…"): hospitals = fetch_hospitals(loc["lat"], loc["lon"], HOSPITAL_RADIUS_M) if not hospitals: st.warning( "No hospitals found within 5 km. Try different coordinates or increase radius." ) st.stop() st.markdown(f"**Found {len(hospitals)} hospital(s) within 5 km**") # -- Map & hospital list side by side -- col_map, col_list = st.columns([1, 1], gap="large") with col_map: st.markdown("### πŸ—ΊοΈ Incident Map") m = build_map(loc["lat"], loc["lon"], hospitals) st_folium(m, height=420, use_container_width=True) with col_list: st.markdown("### πŸ₯ Nearby Hospitals") for i, h in enumerate(hospitals): emoji = "πŸ₯‡" if i == 0 else "πŸ₯ˆ" if i == 1 else "πŸ₯‰" if i == 2 else "πŸ₯" st.markdown( f"""
{emoji} {h['name']}
πŸ“ {h['distance_km']} km  |  πŸ“ž {h['phone']}  |  πŸ›οΈ Beds: {h['beds']}
""", unsafe_allow_html=True, ) if len(hospitals) > 8: st.caption(f"Showing all {len(hospitals)} results") # ============================================================ # SIMULATED NOTIFICATIONS # ============================================================ st.markdown("---") st.markdown("### πŸ“¨ Alert Dispatch Log") st.caption("Sending automated emergency alerts to the nearest hospitals…") notify_count = min(len(hospitals), 3) log_container = st.container() for i in range(notify_count): h = hospitals[i] time.sleep(0.7) log_container.markdown( f"""
βœ…   Alert dispatched to {h['name']} β€” {h['distance_km']} km away
""", unsafe_allow_html=True, ) st.success( f"βœ… Emergency alerts sent to **{notify_count}** hospital(s). " f"Nearest: **{hospitals[0]['name']}** ({hospitals[0]['distance_km']} km)" )