from __future__ import annotations import gradio as gr import pathlib import numpy as np import pandas as pd from hardware.setup import start_gpio from models.registry import MODEL_YOLO from ui.context import AppContext from ui.history import HistoryStore from ui.inference_handlers import InferenceHandlers from ui.input_section import build_input_section from ui.model_section import build_model_section from ui.model_section import model_to_tab_id, tab_value_to_model from ui.output_section import build_output_section CUSTOM_UI_STYLING = """ @import url('https://fonts.googleapis.com/css2?family=Inter:wght=400;600&family=JetBrains+Mono&family=Space+Grotesk:wght=500;700&display=swap'); body, .gradio-container { background-color: #F8FAFC !important; font-family: 'Inter', sans-serif !important; color: #334155 !important; } h1, h2, h3 { font-family: 'Space Grotesk', sans-serif !important; font-weight: 700 !important; color: #0F172A !important; } .tabs { border-bottom: 2px solid #E2E8F0 !important; } .tab-nav button { font-family: 'Space Grotesk', sans-serif !important; font-weight: 500 !important; color: #64748B !important; } .tab-nav button.selected { color: #16A34A !important; border-bottom: 2px solid #16A34A !important; } .metric-box { background: #FFFFFF; border: 1px solid #E2E8F0; border-radius: 8px; padding: 14px; text-align: center; box-shadow: 0 1px 3px rgba(0,0,0,0.05); } .metric-num { font-family: 'JetBrains Mono', monospace; font-size: 22px; font-weight: bold; color: #2563EB; } .badge-yolo { background-color: #D1FAE5; color: #065F46; padding: 4px 10px; border-radius: 6px; font-size: 12px; font-weight: 600; display: inline-block; } .badge-rpi { background-color: #FEF3C7; color: #92400E; padding: 4px 10px; border-radius: 6px; font-size: 12px; font-weight: 600; display: inline-block; margin-left: 6px; } .badge-gradio { background-color: #DBEAFE; color: #1E40AF; padding: 4px 10px; border-radius: 6px; font-size: 12px; font-weight: 600; display: inline-block; margin-left: 6px; } .badge-gpio { background-color: #FEE2E2; color: #991B1B; padding: 4px 10px; border-radius: 6px; font-size: 12px; font-weight: 600; display: inline-block; margin-left: 6px; } #traffic-light-housing { background: #E2E8F0; border-radius: 16px; padding: 12px; display: flex; flex-direction: column; align-items: center; justify-content: center; width: 70px; margin: auto; border: 2px solid #CBD5E1; } .light-bulb { width: 26px; height: 26px; border-radius: 50%; background: #94A3B8; margin: 4px 0; } .light-red-active { background: #EF4444 !important; box-shadow: 0 0 14px #EF4444; } """ def _wire_callbacks(inp, model, out, local_counters, input_handlers, inference, history, loop=None) -> None: inp.shutdown_button.click( fn=inference.shutdown_now, inputs=None, outputs=[out.status_output], show_progress="hidden", ) inp.target_fps.change( fn=input_handlers.timer_interval_update, inputs=[inp.target_fps], outputs=[inp.timer], show_progress="hidden", ) inp.timer.tick( fn=input_handlers.sync_video_preview, inputs=[ inp.resolution, inp.input_source, inp.latest_capture_frame, inp.perf_state, inp.target_fps, ], outputs=[ inp.input_preview, inp.camera_status_output, inp.latest_capture_frame, inp.perf_state, ], show_progress="hidden", ) inp.input_source.change( fn=input_handlers.source_visibility, inputs=[inp.input_source], outputs=[ inp.rpi_note, inp.input_preview, inp.camera_status_output, inp.resolution, inp.target_fps, ], show_progress="hidden", ) def on_model_tab(evt: gr.SelectData) -> str: return inference.sync_selected_model(tab_value_to_model(evt.value)) def on_gpio_model_radio(model_name: str) -> tuple[str, dict]: model_id = inference.sync_selected_model(model_name) return model_id, gr.Tabs(selected=model_to_tab_id(model_id)) model.model_tabs.select(on_model_tab, None, model.selected_model, show_progress="hidden") model.selected_model.change( fn=on_gpio_model_radio, inputs=[model.selected_model], outputs=[model.selected_model, model.model_tabs], show_progress="hidden", ) model.yolo_run.click( lambda: inference.sync_selected_model(MODEL_YOLO), None, model.selected_model, show_progress="hidden", ) def safe_yolo_run(src, frame, conf, iou, ai_hist, err_hist): raw_outputs = inference.run_yolo(src, frame, None, None, conf, iou, ai_hist, err_hist) if not isinstance(raw_outputs, (list, tuple)): return raw_outputs return tuple(raw_outputs) def safe_gpio_inference(src, frame, conf, iou, ai_hist, err_hist): raw_outputs = inference.dispatch_gpio_inference(src, frame, None, None, conf, iou, ai_hist, err_hist) if not isinstance(raw_outputs, (list, tuple)): return raw_outputs return tuple(raw_outputs) # 8 Base Fields + ERROR_DELETE_BUTTON_SLOTS + 4 Counters + 1 State + 1 BarPlot = 24 Outputs Total yolo_click_outputs = [ model.yolo_output, model.yolo_count_output, out.status_output, out.server_status_output, out.ai_history_state, out.error_history_state, out.ai_accordion_output, out.error_accordion_output, ] + list(out.error_delete_buttons) + local_counters + [ out.plot_line, out.plot_bar, ] model.yolo_run.click( fn=safe_yolo_run, inputs=[ inp.input_source, inp.latest_capture_frame, model.yolo_conf, model.yolo_iou, out.ai_history_state, out.error_history_state, ], outputs=yolo_click_outputs, show_progress="hidden", ) # 8 Base Fields + ERROR_DELETE_BUTTON_SLOTS + 4 Counters + 1 State + 1 BarPlot + 1 Tab = 25 Outputs Total gpio_tick_outputs = [ model.yolo_output, model.yolo_count_output, out.status_output, out.server_status_output, out.ai_history_state, out.error_history_state, out.ai_accordion_output, out.error_accordion_output, ] + list(out.error_delete_buttons) + local_counters + [ out.plot_line, out.plot_bar, ] + [model.model_tabs] gpio_timer = gr.Timer(value=0.15, active=True) gpio_timer.tick( fn=safe_gpio_inference, inputs=[ inp.input_source, inp.latest_capture_frame, model.yolo_conf, model.yolo_iou, out.ai_history_state, out.error_history_state, ], outputs=gpio_tick_outputs, show_progress="hidden", ) if loop is not None: hw_timer = gr.Timer(value=0.25, active=True) def _hw_status(): data = loop.get_latest() running = loop.is_running() frame = data.get("annotated") if frame is None: frame = np.zeros((480, 640, 3), dtype=np.uint8) distances = data.get("distances", {}) sensor_str = " | ".join( f"{k.upper()}: {v}cm" for k, v in distances.items() ) if distances else "Awaiting ultrasonic array inputs..." status_str = ( f"đŸŸĸ Pipeline Active\n{data.get('status', 'Processing data stream')}\n\nSensors: {sensor_str}" if running else f"âšĒ Engine Standby — click housing button to execute cycles\n\nSensors: {sensor_str}" ) return frame, status_str hw_timer.tick( fn=_hw_status, inputs=None, outputs=[out.hw_preview, out.hw_status], show_progress="hidden", ) for idx, delete_button in enumerate(out.error_delete_buttons): delete_button.click( fn=lambda errors, idx=idx: history.delete_error_at_index(errors, idx), inputs=[out.error_history_state], outputs=[out.error_history_state, out.error_accordion_output, *out.error_delete_buttons], show_progress="hidden", ) def build_dashboard(ctx: AppContext, loop=None) -> tuple[gr.Blocks, InferenceHandlers]: from ui.input_handlers import InputHandlers history = HistoryStore() input_handlers = InputHandlers(ctx) inference = InferenceHandlers(ctx, history) with gr.Blocks(title="ASTDC Traffic Dashboard") as demo: with gr.Row(): with gr.Column(scale=3): gr.Markdown("# đŸšĻ ASTDC — Control Center") gr.HTML("""
YOLO Object Detection Raspberry Pi Gradio Interface GPIO Hardware
""") with gr.Column(scale=1): gr.HTML("""
Signal Status
""") with gr.Tabs() as operational_tabs: with gr.Tab("ℹī¸ About / Onboarding"): gr.Markdown("## 💡 Project Identity & Strategic Concept") with gr.Row(): with gr.Column(scale=2): gr.Markdown( "### Working Title: AI Smart Traffic Density Controller (ASTDC)\n" "**Student Name:** Summiya Yousaf \n" "**Sparring Partner:** Brandon-Ray NJINKENG \n\n" "**Objective:** Build an adaptive, computer vision-based infrastructure prototype " "that resolves inefficient urban traffic signal timing. By implementing real-time vehicle " "tracking loops on the edge, the system optimizes queue delays, maximizes intersection " "throughput, and mitigates idle emissions." ) with gr.Column(scale=1): gr.Markdown("### 🔩 Edge Infrastructure Core") gr.Markdown( "- **Vision Engine:** YOLO Network Pipeline\n" "- **Host Hardware:** Raspberry Pi Embedded Stack\n" "- **Sensor Web:** HC-SR04 Proximity Modules\n" "- **Telematic Output:** I2C 1602 Character LCD" ) gr.Markdown("---") gr.Markdown("### 🔍 Monitored Class Matrix Models") with gr.Row(): gr.HTML("""

🚗 Cars

Standard volume index

🚌 Buses

Public priority modifier

đŸšļ Pedestrians

Crosswalk demand safety indicator

🏍ī¸ Motorbikes

Low queue profile

""") with gr.Tab("📊 Data Analysis"): gr.Markdown("## 📈 Live Analytical Performance Telemetry") with gr.Row(): counter_car = gr.Number(label="🚗 Cars Detected", value=0, precision=0) counter_bus = gr.Number(label="🚌 Buses Detected", value=0, precision=0) counter_pedestrian = gr.Number(label="đŸšļ Pedestrians Detected", value=0, precision=0) counter_moto = gr.Number(label="🏍ī¸ Motorbikes Detected", value=0, precision=0) # Binds counters in the exact alignment sequence: car, bus, pedestrian, moto local_counters = [counter_car, counter_bus, counter_pedestrian, counter_moto] # Invisible state slot — replaces LinePlot, keeps output count intact ui_plot_line = gr.State() with gr.Row(): gr.Markdown("### đŸšļ Fleet & Pedestrian Classification Breakdown") with gr.Row(): ui_plot_bar = gr.BarPlot( x="Class", y="Count", title="Detected Intersection Class Aggregates", height=260 ) with gr.Tab("🎮 Operations & Diagnostics"): gr.Markdown("## đŸŽĨ USB Camera Stream & Core Infrastructure") with gr.Row(): with gr.Column(scale=2): hw_preview_box = gr.Image( label="📷 Live USB Camera Feed (/dev/video0) — Real-Time Inference Overlay", interactive=False, elem_id="vision-stream-preview", type="numpy" ) with gr.Column(scale=1): hw_status_box = gr.Textbox( label="Hardware Processing Engine Status", value="âšĒ Awaiting ignition step...", lines=8, interactive=False, elem_id="hardware-status-terminal" ) gr.Markdown("---") gr.Markdown("### 🛠ī¸ Manual System Intervention Environment") with gr.Row(): with gr.Column(): inp = build_input_section(ctx.cfg) with gr.Column(): model = build_model_section() out = build_output_section() out.plot_line = ui_plot_line out.plot_bar = ui_plot_bar out.hw_preview = hw_preview_box out.hw_status = hw_status_box _wire_callbacks(inp, model, out, local_counters, input_handlers, inference, history, loop=loop) def _trigger_live_startup(): return MODEL_YOLO, "usb_camera" demo.load( fn=_trigger_live_startup, inputs=None, outputs=[model.selected_model, inp.input_source], show_progress="hidden", ) ctx.gpio = None return demo, inference def launch_dashboard(ctx: AppContext, demo: gr.Blocks) -> None: css_path = pathlib.Path(__file__).parent / "theme.css" active_styling = CUSTOM_UI_STYLING if css_path.exists(): active_styling = active_styling + "\n" + css_path.read_text() demo.queue(default_concurrency_limit=10).launch( server_name=ctx.cfg.gradio_host, server_port=ctx.cfg.gradio_port, css=active_styling, show_error=True, prevent_thread_lock=True, share=True )