import os import base64 import time import json import asyncio import functools import cv2 import numpy as np import gradio as gr import structlog import modal # Import Sentinel components from gatekeeper import ( FrameChangeDetector, ObjectDetector, AudioMonitor, PoseAnalyzer, GatekeeperDecision, SensorSnapshot, AudioClass ) from kokoro_tts import AlertSpeaker from cost_tracker import CostTracker from fallback import FallbackRouter from cohere_rag import SafetyKnowledgeBase from flux_images import AlertImageGenerator from sensor_bridge import SENSOR_BRIDGE_HTML # Setup logger logger = structlog.get_logger() # --- SYSTEM PROMPT FOR NEMOTRON --- SYSTEM_PROMPT = """You are Sentinel, an autonomous AI guardian for visually impaired and elderly users. You receive visual descriptions and sensor data. Your job is to: 1. Identify potential dangers (tripping hazards, approaching vehicles, strangers, fire) 2. Provide navigation guidance (door ahead, stairs, obstacles) 3. Alert ONLY when genuinely dangerous — avoid false alarms 4. Respond in 1-2 sentences maximum (user hears this via TTS) Format: [LEVEL] message Where LEVEL is: CRITICAL, WARNING, or OK Examples: [CRITICAL] Stairs ahead, stop immediately. [WARNING] Person approaching from your left, about 2 meters. [OK] Clear path ahead, hallway is empty.""" # --- CUSTOM CSS (Dark Glassmorphism Theme) --- CUSTOM_CSS = """ body { background: #0a0a1a; } .glass-panel { background: rgba(255,255,255,0.04); border: 1px solid rgba(255,255,255,0.08); border-radius: 16px; padding: 20px; backdrop-filter: blur(12px); } #activate-btn { font-size: 1.3em; letter-spacing: 2px; font-weight: 700; } #status-row { display: flex; justify-content: space-around; padding: 10px 16px; background: rgba(99,102,241,0.1); border-radius: 10px; font-size: 0.9em; margin-top: 10px; } #sensor-dashboard { padding: 12px; background: rgba(0,0,0,0.2); border-radius: 10px; margin-top: 10px; } .sensor-val { padding: 4px 8px; font-family: monospace; font-size: 0.85em; color: #a5b4fc; } .alert-banner-critical { background: #dc2626; color: white; padding: 16px; border-radius: 10px; font-size: 1.2em; font-weight: bold; animation: pulse 1s infinite; } .alert-banner-warning { background: #f59e0b; color: #1a1a2e; padding: 16px; border-radius: 10px; font-size: 1.1em; font-weight: 600; } .alert-banner-info { background: rgba(99,102,241,0.15); color: #c7d2fe; padding: 12px; border-radius: 10px; } @keyframes pulse { 0%, 100% { opacity: 1; } 50% { opacity: 0.7; } } .sponsor-card { display: inline-block; background: rgba(255,255,255,0.05); border: 1px solid rgba(255,255,255,0.1); border-radius: 12px; padding: 14px 18px; margin: 6px; min-width: 140px; text-align: center; } .sponsor-card h4 { margin: 0 0 4px 0; color: #a5b4fc; } .sponsor-card p { margin: 0; font-size: 0.8em; color: #94a3b8; } #sim-panel { margin-top: 16px; padding: 16px; background: rgba(245, 158, 11, 0.08); border: 1px solid rgba(245, 158, 11, 0.2); border-radius: 12px; } #sim-panel h3 { margin: 0 0 12px 0; color: #fbbf24; font-size: 1em; } .sim-btn { margin: 4px !important; } """ # --- ARCHITECTURE MARKDOWN --- ARCHITECTURE_MARKDOWN = """ ## Two-Tier Gatekeeper Architecture Sentinel uses a **cost-efficient two-tier architecture** that minimizes GPU usage: **Tier 1 — CPU Gatekeeper (Free, runs 24/7)** - Frame differencing (OpenCV) — detects significant scene changes (>30% pixel delta) - YOLO11n zero-shot detection — identifies persons, vehicles, fire, animals - MediaPipe Pose — detects falls via head-hip landmark inversion - Browser Audio Energy & Doppler Detection — detects loud events & ultrasonic motion **Tier 2 — GPU Analyst (Modal A10G, on-demand only)** - MiniCPM-V 4.6 — vision-language scene understanding - Nemotron-3-Nano-4B — safety reasoning and alert generation - Cohere RAG (Command-R) — emergency protocol lookup, runs **inside Modal container** (co-located with VLM) - Kokoro TTS — natural speech alert synthesis (82M params, ONNX) **Fallback Chain (when Modal is offline)** - Reasoning: OpenRouter Nemotron-30B-A3B → OpenAI gpt-4o-mini - Vision: OpenBMB MiniCPM-V API → OpenAI gpt-4o-mini vision - RAG: local SafetyKnowledgeBase (keyword-based, zero latency) **Result:** ~240x cost reduction vs naive always-on VLM streaming. """ # --- SPONSOR INTEGRATIONS HTML --- SPONSOR_HTML = """
""" # --- INITIALIZE GLOBAL AI ENGINES & ROUTERS --- logger.info("Initializing global AI controllers...") # Connect to Modal backend try: SentinelEngine = modal.Cls.from_name("sentinel-backend", "SentinelEngine") modal_engine = SentinelEngine() logger.info("Successfully bound connection to Modal GPU backend.") except Exception as e: logger.error("Could not bind Modal backend. Running with fallback client.", error=str(e)) modal_engine = None # Initialize fallback router (OpenRouter, OpenBMB, OpenAI) fallback_router = FallbackRouter( openrouter_key=os.environ.get("OPENROUTER_API_KEY", ""), openbmb_key=os.environ.get("OPENBMB_API_KEY", ""), openai_key=os.environ.get("OPENAI_API_KEY", "") ) # Local SafetyKnowledgeBase: offline RAG fallback when Modal is unavailable # Primary RAG now runs inside the Modal container via modal_engine.rag_query() cohere_rag = SafetyKnowledgeBase(cohere_api_key=os.environ.get("COHERE_API_KEY", "")) flux_gen = AlertImageGenerator(hf_token=os.environ.get("HF_TOKEN", "")) # Load read-only models globally to save memory try: object_detector = ObjectDetector() except Exception as e: logger.error("ObjectDetector (YOLO) failed to initialize. Visual detection disabled.", error=str(e)) object_detector = None try: audio_monitor = AudioMonitor() except Exception as e: logger.error("AudioMonitor (YAMNet) failed to initialize.", error=str(e)) audio_monitor = None speaker = AlertSpeaker() # --- HELPER FUNCTIONS --- def get_compass_direction(deg: float) -> str: """ Converts degrees to a compass heading string. """ dirs = ["N", "NE", "E", "SE", "S", "SW", "W", "NW"] idx = int(((deg + 22.5) % 360) / 45) return dirs[idx] # --- SIMULATION MODE --- SIMULATION_SCENARIOS = { "stairs_ahead": { "label": "Stairs Ahead", "prompt": "Analyze this scene for a visually impaired user. Context: Dark environment, stairs detected directly ahead, user walking forward. Light level: 3.2 lux. Battery: 78%.", "question": "A visually impaired user is walking toward stairs in a dimly lit hallway. Describe the danger and give guidance.", "color": (40, 30, 80), "shapes": "stairs", }, "person_approaching": { "label": "Person Approaching", "prompt": "Analyze this scene for a visually impaired user. Context: Person detected approaching from left side, about 2 meters away. User heading: 45° NE. Light level: 320 lux. Battery: 65%.", "question": "A person is approaching a visually impaired user from the left side in a public space. Describe the situation.", "color": (80, 60, 30), "shapes": "person", }, "fall_detected": { "label": "Fall Detected", "prompt": "URGENT: Fall detected via pose analysis. Head position below hip level (head_y: 0.82, hip_y: 0.55). Accelerometer spike: 22.4 m/s². User may have fallen.", "question": "An elderly user's pose data indicates they have fallen. Provide an emergency assessment and safety guidance.", "color": (30, 30, 100), "shapes": "fall", }, "fire_alert": { "label": "Fire Detected", "prompt": "CRITICAL: Fire detected in scene. Loud alarm audio confirmed (85% confidence). Smoke visible. User heading: 180° S. Light level fluctuating rapidly.", "question": "Fire has been detected in the environment of a visually impaired user. Provide urgent evacuation guidance.", "color": (20, 40, 100), "shapes": "fire", }, "clear_path": { "label": "Clear Path", "prompt": "Analyze this scene for a visually impaired user. Context: Clear hallway, no obstacles detected, good lighting. User heading: 90° E. Light level: 450 lux. Battery: 92%.", "question": "The path ahead appears clear for a visually impaired user. Confirm the safe conditions and provide brief navigation guidance.", "color": (50, 60, 40), "shapes": "clear", }, } def generate_sim_image(scenario_key: str) -> tuple: """ Generates a synthetic 640x480 test image for simulation mode. Returns (frame_bgr_numpy, base64_jpeg_string). """ cfg = SIMULATION_SCENARIOS[scenario_key] img = np.zeros((480, 640, 3), dtype=np.uint8) img[:] = cfg["color"] shapes = cfg["shapes"] if shapes == "stairs": for i in range(8): y = 380 - i * 40 cv2.rectangle(img, (120, y), (520, y + 35), (70, 70, 140), -1) cv2.rectangle(img, (120, y), (520, y + 35), (90, 90, 170), 2) cv2.putText(img, "STAIRS AHEAD", (160, 60), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (255, 255, 255), 3) elif shapes == "person": cv2.circle(img, (220, 160), 35, (180, 180, 200), -1) cv2.rectangle(img, (185, 200), (255, 340), (180, 180, 200), -1) cv2.line(img, (185, 240), (140, 310), (180, 180, 200), 8) cv2.line(img, (255, 240), (300, 310), (180, 180, 200), 8) cv2.line(img, (205, 340), (175, 420), (180, 180, 200), 8) cv2.line(img, (235, 340), (265, 420), (180, 180, 200), 8) cv2.arrowedLine(img, (300, 250), (350, 250), (100, 200, 255), 3, tipLength=0.3) cv2.putText(img, "PERSON APPROACHING", (120, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.9, (255, 255, 255), 2) elif shapes == "fall": cv2.circle(img, (400, 350), 30, (100, 100, 220), -1) cv2.rectangle(img, (260, 340), (390, 380), (100, 100, 220), -1) cv2.line(img, (260, 360), (220, 400), (100, 100, 220), 8) cv2.line(img, (350, 380), (380, 430), (100, 100, 220), 8) cv2.line(img, (300, 380), (280, 430), (100, 100, 220), 8) cv2.putText(img, "FALL DETECTED", (160, 80), cv2.FONT_HERSHEY_SIMPLEX, 1.4, (80, 80, 255), 3) elif shapes == "fire": pts = np.array([[320, 100], [240, 300], [280, 280], [320, 350], [360, 280], [400, 300]], np.int32) cv2.fillPoly(img, [pts], (0, 140, 255)) pts2 = np.array([[320, 160], [270, 300], [300, 280], [320, 320], [340, 280], [370, 300]], np.int32) cv2.fillPoly(img, [pts2], (0, 200, 255)) cv2.putText(img, "FIRE ALERT", (180, 60), cv2.FONT_HERSHEY_SIMPLEX, 1.4, (80, 80, 255), 3) else: cv2.line(img, (200, 480), (280, 120), (100, 180, 100), 3) cv2.line(img, (440, 480), (360, 120), (100, 180, 100), 3) cv2.putText(img, "CLEAR PATH", (190, 80), cv2.FONT_HERSHEY_SIMPLEX, 1.2, (150, 255, 150), 2) cv2.putText(img, "Safe to proceed", (180, 420), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (150, 200, 150), 1) cv2.rectangle(img, (0, 0), (639, 479), (255, 255, 255), 2) cv2.putText(img, f"SIM: {cfg['label']}", (10, 470), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (200, 200, 200), 1) _, buffer = cv2.imencode(".jpg", img) b64 = base64.b64encode(buffer).decode("utf-8") return img, b64 async def simulate_scenario( scenario_name: str, monitoring_active: bool, alert_history_state: list, cost_tracker_state: CostTracker, frame_count_state: int, frame_detector_state: FrameChangeDetector, pose_analyzer_state: PoseAnalyzer, decision_engine_state: GatekeeperDecision, last_vlm_time_state: float, ): """ Injects a synthetic scenario into the full VLM pipeline, bypassing Tier 1 gatekeeper. Used for demo on iOS, recording videos, and generating training data. """ if scenario_name not in SIMULATION_SCENARIOS: return ( gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, 0.0, decision_engine_state, last_vlm_time_state ) cfg = SIMULATION_SCENARIOS[scenario_name] monitoring_active = True frame_count_state += 1 cost_tracker_state.log_frame() frame, frame_b64 = generate_sim_image(scenario_name) current_time = time.time() last_vlm_time_state = current_time logger.info("Simulation scenario triggered", scenario=scenario_name, prompt=cfg["prompt"]) vlm_response = "" tokens_used = 0 duration_ms = 0 model_name = "modal-backend" try: if modal_engine: vlm_start = time.perf_counter() vlm_res = await modal_engine.see.remote.aio(frame_b64, cfg["question"]) vlm_text = vlm_res.get("text", "") reason_res = await modal_engine.reason.remote.aio(vlm_text, SYSTEM_PROMPT) vlm_response = reason_res.get("text", "") tokens_used = vlm_res.get("tokens", 0) + reason_res.get("tokens", 0) duration_ms = int((time.perf_counter() - vlm_start) * 1000) else: raise ConnectionError("Modal engine not initialized.") except Exception as me: logger.error("Simulation: Modal failed, using fallback router.", error=str(me)) fallback_res = await fallback_router.fallback_see(frame_b64, cfg["question"]) vlm_text = fallback_res.get("text", "") reason_res = await fallback_router.fallback_reason(vlm_text, SYSTEM_PROMPT) vlm_response = reason_res.get("text", "") tokens_used = fallback_res.get("tokens", 0) + reason_res.get("tokens", 0) duration_ms = 1500 model_name = reason_res.get("model", "fallback-router") cost_tracker_state.log(model_name, tokens_used, duration_ms) alert_level = "none" alert_text = vlm_response if "[CRITICAL]" in vlm_response: alert_level = "critical" alert_text = vlm_response.replace("[CRITICAL]", "").strip() elif "[WARNING]" in vlm_response: alert_level = "warning" alert_text = vlm_response.replace("[WARNING]", "").strip() elif "[OK]" in vlm_response: alert_level = "info" alert_text = vlm_response.replace("[OK]", "").strip() if not alert_text: alert_level = "info" alert_text = f"[SIM] {cfg['label']} scenario executed. VLM pipeline operational." # Run RAG and TTS concurrently rag_task = None tts_task = None if alert_level in ["critical", "warning"]: async def run_sim_rag(lvl, txt): try: if modal_engine: res = await modal_engine.rag_query.remote.aio(lvl, txt) return res.get("advice", "") else: raise RuntimeError("Modal engine not available") except Exception as e: logger.warn("Simulation: Modal RAG failed, using local fallback.", error=str(e)) res = cohere_rag.query(lvl, txt) return res.get("advice", "") rag_task = asyncio.create_task(run_sim_rag(alert_level, alert_text)) tts_task = asyncio.create_task(speaker.speak(alert_text, level=alert_level)) rag_advice = "" audio_data_uri = None if rag_task or tts_task: tasks = [] if rag_task: tasks.append(rag_task) if tts_task: tasks.append(tts_task) results = await asyncio.gather(*tasks, return_exceptions=True) idx = 0 if rag_task: res_val = results[idx] if not isinstance(res_val, Exception): rag_advice = res_val idx += 1 if tts_task: res_val = results[idx] if not isinstance(res_val, Exception) and res_val.get("audio_base64"): audio_data_uri = f"data:audio/wav;base64,{res_val['audio_base64']}" if rag_advice: alert_text = f"{alert_text} {rag_advice}" alert_image_html = "" if alert_level in ["critical", "warning"]: try: flux_result = flux_gen.generate(alert_text, alert_level) if flux_result.get("image_base64"): alert_image_html = f"Alert illustration" except Exception as flux_err: logger.error("Simulation: FLUX image generation failed", error=str(flux_err)) if alert_level != "none": autoplay_tag = f"" if audio_data_uri else "" alert_html = f"
⚠️ [SIM] {alert_text}{autoplay_tag}{alert_image_html}
" else: alert_html = f"
[SIM] Status: Normal. Path clear.
" new_alert = [ time.strftime("%H:%M:%S", time.localtime(current_time)), alert_level.upper(), f"[SIM] {alert_text}", 0.95 ] alert_history_state.insert(0, new_alert) stats = cost_tracker_state.get_stats() status_row_html = ( f"
" f"Status: 🟡 SIMULATION" f"Calls: {stats['total_calls']}" f"Cost: ${stats['total_cost_usd']:.4f}" f"Uptime: {stats['uptime_hours']:.3f}h" f"
" ) sensor_dashboard_html = ( f"
" f"
Mode: SIMULATION — {cfg['label']}
" f"
Prompt: {cfg['prompt']}
" f"
Model: {model_name} | Tokens: {tokens_used} | Latency: {duration_ms}ms
" f"
VLM Response: {vlm_response}
" f"
" ) stats_data = [ ["Total Frames Evaluated", str(stats["frames_processed"] + stats["gatekeeper_filtered"]), "N/A", "N/A"], ["Total VLM GPU Inferences", str(stats["total_calls"]), str(stats["frames_processed"] + stats["gatekeeper_filtered"]), f"{stats['savings_pct']}% Filtered"], ["Total Tokens Consumed", f"{stats['total_tokens']:,}", "N/A", "N/A"], ["Estimated Cost (USD)", f"${stats['total_cost_usd']:.6f}", f"${stats['naive_cost_usd']:.6f}", f"{stats['savings_pct']}% Saved"], ["Average Call Latency", f"{stats['avg_latency_ms']} ms", "N/A", "N/A"], ] call_logs = cost_tracker_state.get_gradio_rows() return ( alert_html, status_row_html, alert_history_state, stats_data, call_logs, sensor_dashboard_html, frame, monitoring_active, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, 0.0, decision_engine_state, last_vlm_time_state ) # --- GRADIO CALLBACKS --- def activate_sentinel(): """ Callback fired when user clicks 'ACTIVATE SENTINEL'. Turns on UI elements and enables monitoring state. """ logger.info("Sentinel Activated") return ( gr.update(visible=True), # Show camera feed gr.update(value="System Active. Monitoring..."), # Alert banner text True # Set monitoring_active = True ) def clear_alert_history(): """ Resets the alert dataframe logs. """ return [[]] async def process_frame( frame_base64: str, accel_x: float, accel_y: float, accel_z: float, gyro_beta: float, gyro_gamma: float, gps_lat: float, gps_lon: float, light_level: float, battery_pct: float, heading: float, loud_audio_flag: str, doppler_motion_flag: str, audio_level: float, voice_query: str, monitoring_active: bool, alert_history_state: list, cost_tracker_state: CostTracker, frame_count_state: int, frame_detector_state: FrameChangeDetector, pose_analyzer_state: PoseAnalyzer, decision_engine_state: GatekeeperDecision, last_vlm_time_state: float ): """ Core Loop called every 500ms when the camera frame changes. Executes Tier 1 logic on CPU and conditionally triggers Tier 2 on GPU. """ if not monitoring_active or not frame_base64: yield ( gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state, gr.update(), gr.update(), gr.update(), gr.update(), "idle" ) return # 1. Parse frame and update basic telemetry states frame_count_state += 1 cost_tracker_state.log_frame() try: img_bytes = base64.b64decode(frame_base64) nparr = np.frombuffer(img_bytes, np.uint8) frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR) except Exception as e: logger.error("Failed to decode frame base64", error=str(e)) yield ( gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state, gr.update(), gr.update(), gr.update(), gr.update(), "idle" ) return if frame is None: yield ( gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), gr.update(), alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state, gr.update(), gr.update(), gr.update(), gr.update(), "idle" ) return # 2. RUN TIER 1 GATES (CPU only, free) # 2.1 Frame Differencing change_result = frame_detector_state.detect(frame) # 2.2 YOLO Detections detections = [] yolo_summary = "No critical visual targets detected." if object_detector is not None: detections = object_detector.detect(frame) yolo_summary = object_detector.get_trigger_summary(detections) serialized_detections = json.dumps([ { "class_name": d.class_name, "confidence": float(d.confidence), "bbox": [int(x) for x in d.bbox] } for d in detections ]) # 2.3 Pose Analysis (MediaPipe) pose_data = pose_analyzer_state.analyze(frame) # 2.4 Audio Check # Real YAMNet classification disabled to reduce cold start. Using browser-side RMS energy detection. audio_classes = [] if loud_audio_flag == "true": audio_classes.append(AudioClass(class_name="Scream", confidence=0.85, alert_level="critical")) if doppler_motion_flag == "true": audio_classes.append(AudioClass(class_name="Alarm", confidence=0.75, alert_level="warning")) # 2.5 Collect Sensor telemetries sensor_data = SensorSnapshot( accelerometer=(accel_x, accel_y, accel_z), gyroscope=(gyro_beta, gyro_gamma, 0.0), gps=(gps_lat, gps_lon) if (gps_lat != 0.0 and gps_lon != 0.0) else None, light_level=light_level, battery_pct=battery_pct, heading=heading ) # 2.6 Gatekeeper Decision (uses persistent state for cooldown tracking) decision = decision_engine_state.decide( frame_change=change_result, detections=detections, audio_classes=audio_classes, pose_data=pose_data, user_query=voice_query, sensor_data=sensor_data ) alert_html = gr.update() status_row_html = gr.update() alert_history_row = gr.update() sensor_dashboard_html = gr.update() # 3. RUN TIER 2 VLM GATES (GPU, triggered) if decision.should_trigger: current_time = time.time() time_since_vlm = current_time - last_vlm_time_state # Enforce rate throttling (skip VLM if last call was < 4.0s ago, unless CRITICAL urgency) if time_since_vlm < 4.0 and decision.urgency != "critical": logger.info("VLM call suppressed by throttle window.") else: last_vlm_time_state = current_time # Yield VLM state update to frontend immediately so stage lights up stats = cost_tracker_state.get_stats() stats_data = [ ["Total Frames Evaluated", str(stats["frames_processed"] + stats["gatekeeper_filtered"]), "N/A", "N/A"], ["Total VLM GPU Inferences", str(stats["total_calls"]), str(stats["frames_processed"] + stats["gatekeeper_filtered"]), f"{stats['savings_pct']}% Filtered"], ["Total Tokens Consumed", f"{stats['total_tokens']:,}", "N/A", "N/A"], ["Estimated Cost (USD)", f"${stats['total_cost_usd']:.6f}", f"${stats['naive_cost_usd']:.6f}", f"{stats['savings_pct']}% Saved"], ["Average Call Latency", f"{stats['avg_latency_ms']} ms", "N/A", "N/A"], ] call_logs = cost_tracker_state.get_gradio_rows() sensor_dashboard_html = ( f"
" f"
Accelerometer: [X: {accel_x:.2f}, Y: {accel_y:.2f}, Z: {accel_z:.2f}] m/s²
" f"
GPS Telemetry: [Lat: {gps_lat:.6f}, Lng: {gps_lon:.6f}]
" f"
Ambient Light Level: {light_level:.1f} lux
" f"
Battery Telemetry: {battery_pct:.0f}%
" f"
Compass Heading: {heading:.1f}° {get_compass_direction(heading)}
" f"
" ) yield ( gr.update(), gr.update(), gr.update(), stats_data, call_logs, sensor_dashboard_html, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state, serialized_detections, frame_count_state, stats['total_cost_usd'], "", "vlm" ) # Format contextual RAG/VLM question direction = get_compass_direction(heading) sensor_context = ( f"Camera shows: {yolo_summary}. " f"User heading: {heading:.0f}° {direction}. " f"Light level: {light_level:.1f} lux. " f"Battery: {battery_pct:.0f}%." ) question = f"Analyze this scene and provide a warning prompt. Context: {sensor_context}" logger.info("VLM Triggered, calling Modal backend...", context=sensor_context) # Resize image to 640x480 for fast VLM transfer resized_vlm = cv2.resize(frame, (640, 480)) _, buffer = cv2.imencode(".jpg", resized_vlm) frame_vlm_base64 = base64.b64encode(buffer).decode("utf-8") vlm_response = "" tokens_used = 0 duration_ms = 0 model_name = "modal-backend" # Execute async calls to Modal with fallback router protection try: if modal_engine: vlm_start = time.perf_counter() # 1. Call VLM vlm_res = await modal_engine.see.remote.aio(frame_vlm_base64, question) # 2. Call Nemotron text logic vlm_text = vlm_res.get("text", "") reason_res = await modal_engine.reason.remote.aio(vlm_text, SYSTEM_PROMPT) vlm_response = reason_res.get("text", "") tokens_used = vlm_res.get("tokens", 0) + reason_res.get("tokens", 0) duration_ms = int((time.perf_counter() - vlm_start) * 1000) else: raise ConnectionError("Modal engine not initialized.") except Exception as me: logger.error("Modal connection failed. Invoking Fallback Router.", error=str(me)) # Call backup client fallback_res = await fallback_router.fallback_see(frame_vlm_base64, question) vlm_text = fallback_res.get("text", "") reason_res = await fallback_router.fallback_reason(vlm_text, SYSTEM_PROMPT) vlm_response = reason_res.get("text", "") tokens_used = fallback_res.get("tokens", 0) + reason_res.get("tokens", 0) duration_ms = 1500 # estimated fallback latency model_name = reason_res.get("model", "fallback-router") # Update Cost Tracker metrics cost_tracker_state.log(model_name, tokens_used, duration_ms) logger.info("VLM Response received", response=vlm_response) # Parse Alert levels alert_level = "none" alert_text = vlm_response if "[CRITICAL]" in vlm_response: alert_level = "critical" alert_text = vlm_response.replace("[CRITICAL]", "").strip() elif "[WARNING]" in vlm_response: alert_level = "warning" alert_text = vlm_response.replace("[WARNING]", "").strip() elif "[OK]" in vlm_response: alert_level = "info" alert_text = vlm_response.replace("[OK]", "").strip() # Generate spoken overlay (Kokoro TTS) and RAG concurrently rag_task = None tts_task = None if alert_level in ["critical", "warning"]: async def run_rag_async(lvl, txt): try: if modal_engine: res = await modal_engine.rag_query.remote.aio(lvl, txt) return res.get("advice", "") else: raise RuntimeError("Modal engine not available") except Exception as e: logger.warn("Modal RAG failed, using local fallback.", error=str(e)) res = cohere_rag.query(lvl, txt) return res.get("advice", "") rag_task = asyncio.create_task(run_rag_async(alert_level, alert_text)) tts_task = asyncio.create_task(speaker.speak(alert_text, level=alert_level)) rag_advice = "" audio_data_uri = None if rag_task or tts_task: tasks = [] if rag_task: tasks.append(rag_task) if tts_task: tasks.append(tts_task) results = await asyncio.gather(*tasks, return_exceptions=True) idx = 0 if rag_task: res_val = results[idx] if not isinstance(res_val, Exception): rag_advice = res_val idx += 1 if tts_task: res_val = results[idx] if not isinstance(res_val, Exception) and res_val.get("audio_base64"): audio_data_uri = f"data:audio/wav;base64,{res_val['audio_base64']}" if rag_advice: alert_text = f"{alert_text} {rag_advice}" # Generate alert illustration (FLUX.1 or PIL fallback) alert_image_html = "" if alert_level in ["critical", "warning"]: try: flux_result = flux_gen.generate(alert_text, alert_level) if flux_result.get("image_base64"): alert_image_html = f"Alert illustration" except Exception as flux_err: logger.error("FLUX image generation failed", error=str(flux_err)) # Render HTML alert banner if alert_level != "none": autoplay_tag = f"" if audio_data_uri else "" alert_html = f"
⚠️ {alert_text}{autoplay_tag}{alert_image_html}
" else: alert_html = f"
Status: Normal. Path clear.
" # Log to alert history state new_alert = [ time.strftime("%H:%M:%S", time.localtime(current_time)), alert_level.upper(), alert_text, decision.confidence ] alert_history_state.insert(0, new_alert) alert_history_row = alert_history_state # 4. Render Telemetry & cost updates stats = cost_tracker_state.get_stats() # Render Status row status_row_html = ( f"
" f"Status: 🟢 ACTIVE" f"Calls: {stats['total_calls']}" f"Cost: ${stats['total_cost_usd']:.4f}" f"Uptime: {stats['uptime_hours']:.3f}h" f"
" ) # Render Sensor dashboard sensor_dashboard_html = ( f"
" f"
Accelerometer: [X: {accel_x:.2f}, Y: {accel_y:.2f}, Z: {accel_z:.2f}] m/s²
" f"
GPS Telemetry: [Lat: {gps_lat:.6f}, Lng: {gps_lon:.6f}]
" f"
Ambient Light Level: {light_level:.1f} lux
" f"
Battery Telemetry: {battery_pct:.0f}%
" f"
Compass Heading: {heading:.1f}° {get_compass_direction(heading)}
" f"
" ) # Refresh Cost dataframe structures stats_data = [ ["Total Frames Evaluated", str(stats["frames_processed"] + stats["gatekeeper_filtered"]), "N/A", "N/A"], ["Total VLM GPU Inferences", str(stats["total_calls"]), str(stats["frames_processed"] + stats["gatekeeper_filtered"]), f"{stats['savings_pct']}% Filtered"], ["Total Tokens Consumed", f"{stats['total_tokens']:,}", "N/A", "N/A"], ["Estimated Cost (USD)", f"${stats['total_cost_usd']:.6f}", f"${stats['naive_cost_usd']:.6f}", f"{stats['savings_pct']}% Saved"], ["Average Call Latency", f"{stats['avg_latency_ms']} ms", "N/A", "N/A"], ] call_logs = cost_tracker_state.get_gradio_rows() yield ( alert_html, status_row_html, alert_history_row, stats_data, call_logs, sensor_dashboard_html, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state, serialized_detections, frame_count_state, stats['total_cost_usd'], "", "idle" ) def refresh_costs(cost_tracker_state: CostTracker): """ Auto-refresh method for the cost tab dataframe values. """ stats = cost_tracker_state.get_stats() stats_data = [ ["Total Frames Evaluated", str(stats["frames_processed"] + stats["gatekeeper_filtered"]), "N/A", "N/A"], ["Total VLM GPU Inferences", str(stats["total_calls"]), str(stats["frames_processed"] + stats["gatekeeper_filtered"]), f"{stats['savings_pct']}% Filtered"], ["Total Tokens Consumed", f"{stats['total_tokens']:,}", "N/A", "N/A"], ["Estimated Cost (USD)", f"${stats['total_cost_usd']:.6f}", f"${stats['naive_cost_usd']:.6f}", f"{stats['savings_pct']}% Saved"], ["Average Call Latency", f"{stats['avg_latency_ms']} ms", "N/A", "N/A"], ] call_logs = cost_tracker_state.get_gradio_rows() return stats_data, call_logs # --- GRADIO BLOCKS LAYOUT DEFINITION --- with gr.Blocks(css=CUSTOM_CSS, theme=gr.themes.Soft(primary_hue="indigo")) as demo: # Session-isolated states monitoring_active = gr.State(value=False) alert_history_state = gr.State(value=[]) cost_tracker_state = gr.State(value=CostTracker()) frame_count_state = gr.State(value=0) frame_detector_state = gr.State(value=FrameChangeDetector()) pose_analyzer_state = gr.State(value=PoseAnalyzer()) decision_engine_state = gr.State(value=GatekeeperDecision(cooldown_seconds=15.0)) last_vlm_time_state = gr.State(value=0.0) # Hidden text boxes for JS sensor bridge feeds image_data = gr.Textbox(visible=False, elem_id="image-data") loud_audio_flag = gr.Textbox(value="false", visible=False, elem_id="loud-audio-flag") doppler_motion_flag = gr.Textbox(value="false", visible=False, elem_id="doppler-motion-flag") sentinel_active_state = gr.Textbox(value="false", visible=False, elem_id="sentinel-active-state") detection_data = gr.Textbox(value="[]", visible=False, elem_id="detection-data") frame_count_display = gr.Number(value=0, visible=False, elem_id="frame-count") cost_total_display = gr.Number(value=0.0, visible=False, elem_id="cost-total") voice_query = gr.Textbox(value="", visible=False, elem_id="voice-query") pipeline_stage = gr.Textbox(value="idle", visible=False, elem_id="pipeline-stage") gr.Markdown("# 🛡️ Sentinel\n### Autonomous AI Guardian") # TAB 1: Live Monitor with gr.Tab("📡 Monitor"): with gr.Column(elem_classes="glass-panel"): with gr.Row(): activate_btn = gr.Button("🛡️ ACTIVATE SENTINEL", variant="primary", size="lg", elem_id="activate-btn") voice_btn = gr.Button("🎤 VOICE QUERY", variant="secondary", size="lg", elem_id="voice-btn") sos_btn = gr.Button("🆘 SOS", variant="stop", size="lg", elem_id="sos-btn") camera_feed = gr.Image(label="Live Camera", interactive=False, visible=False, elem_id="camera-feed") pipeline_indicator = gr.HTML("
", elem_id="pipeline-indicator-container") alert_banner = gr.HTML("
System idle. Ready to activate.
", elem_id="alert-banner-container") status_row = gr.HTML( "
" "Status: 🟥 INACTIVE" "Calls: 0" "Cost: $0.0000" "Uptime: 0.0h" "
" ) cost_ticker = gr.HTML("
", elem_id="cost-ticker-container") # Hidden Number fields populated by JS accel_x = gr.Number(value=0.0, visible=False, elem_id="accel-x") accel_y = gr.Number(value=0.0, visible=False, elem_id="accel-y") accel_z = gr.Number(value=0.0, visible=False, elem_id="accel-z") gyro_beta = gr.Number(value=0.0, visible=False, elem_id="gyro-beta") gyro_gamma = gr.Number(value=0.0, visible=False, elem_id="gyro-gamma") gps_lat = gr.Number(value=0.0, visible=False, elem_id="gps-lat") gps_lon = gr.Number(value=0.0, visible=False, elem_id="gps-lon") light_level = gr.Number(value=0.0, visible=False, elem_id="light-level") battery_pct = gr.Number(value=100.0, visible=False, elem_id="battery-pct") heading = gr.Number(value=0.0, visible=False, elem_id="heading-val") audio_level = gr.Number(value=0.0, visible=False, elem_id="audio-level") # HTML sensor bridge script injection gr.HTML(SENSOR_BRIDGE_HTML) # Simulation Mode Panel with gr.Column(elem_id="sim-panel"): gr.Markdown("### Simulation Mode") gr.Markdown("*Test the full AI pipeline with preset scenarios — works on any device*") sim_buttons = {} for key, cfg in SIMULATION_SCENARIOS.items(): sim_buttons[key] = gr.Button( cfg["label"], size="sm", elem_classes="sim-btn" ) # TAB 2: Alert History with gr.Tab("🔔 Alerts"): with gr.Column(elem_classes="glass-panel"): alert_history = gr.Dataframe( headers=["Time", "Level", "Message", "Confidence"], datatype=["str", "str", "str", "number"], wrap=True ) clear_alerts_btn = gr.Button("Clear History", variant="secondary") # TAB 3: Cost Dashboard with gr.Tab("💰 Costs"): with gr.Column(elem_classes="glass-panel"): cost_stats = gr.Dataframe( headers=["Metric", "Sentinel", "Naive (GPT-4o)", "Savings"], datatype=["str", "str", "str", "str"], wrap=True ) gr.Markdown("### Recent Inference Calls") call_log = gr.Dataframe( headers=["Time", "Model", "Tokens", "Latency(ms)", "Cost($)"], datatype=["str", "str", "number", "number", "number"] ) # TAB 4: Architecture with gr.Tab("🏗️ Architecture"): with gr.Column(elem_classes="glass-panel"): gr.Markdown(ARCHITECTURE_MARKDOWN) gr.Markdown("### Sponsor Integrations") gr.HTML(SPONSOR_HTML) # TAB 5: Live Sensors Telemetry with gr.Tab("📱 Sensors"): with gr.Column(elem_classes="glass-panel"): sensor_display = gr.HTML( "
" "
Telemetry Offline. Click 'Activate Sentinel' to view live sensors.
" "
" ) gr.Markdown("*Sensor data collected from your device via browser APIs*") # --- EVENT BINDINGS --- # Activation Trigger activate_btn.click( fn=activate_sentinel, inputs=None, outputs=[camera_feed, alert_banner, monitoring_active] ) # Frame Loop Trigger (Change in image_data triggers processing) image_data.change( fn=process_frame, inputs=[ image_data, accel_x, accel_y, accel_z, gyro_beta, gyro_gamma, gps_lat, gps_lon, light_level, battery_pct, heading, loud_audio_flag, doppler_motion_flag, audio_level, voice_query, monitoring_active, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, decision_engine_state, last_vlm_time_state ], outputs=[ alert_banner, status_row, alert_history, cost_stats, call_log, sensor_display, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state, detection_data, frame_count_display, cost_total_display, voice_query, pipeline_stage ] ) # Clear History Trigger clear_alerts_btn.click( fn=clear_alert_history, inputs=None, outputs=alert_history ) # Simulation Scenario Buttons _sim_inputs = [ monitoring_active, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, decision_engine_state, last_vlm_time_state ] _sim_outputs = [ alert_banner, status_row, alert_history, cost_stats, call_log, sensor_display, camera_feed, monitoring_active, alert_history_state, cost_tracker_state, frame_count_state, frame_detector_state, pose_analyzer_state, audio_level, decision_engine_state, last_vlm_time_state ] for _sim_key, _sim_btn in sim_buttons.items(): _sim_btn.click( fn=functools.partial(simulate_scenario, _sim_key), inputs=_sim_inputs, outputs=_sim_outputs ) # Auto-refresh Cost telemetries every 5 seconds timer = gr.Timer(5) timer.tick( fn=refresh_costs, inputs=cost_tracker_state, outputs=[cost_stats, call_log] ) if __name__ == "__main__": demo.launch()