| from __future__ import annotations |
|
|
| import math |
| from typing import Any |
|
|
| |
| DEMO_ROUTING_DATA = True |
|
|
| ROOM_COORDINATES = { |
| "LOBBY": (12.9718, 77.5947), |
| "RESTAURANT B": (12.9721, 77.5954), |
| "FLOOR 3": (12.9725, 77.5950), |
| "POOL DECK": (12.9715, 77.5958), |
| "BALLROOM": (12.9719, 77.5961), |
| "SERVICE CORRIDOR": (12.9723, 77.5942), |
| } |
|
|
| RESPONDER_DIRECTORY = [ |
| { |
| "id": "sec-team-alpha", |
| "name": "Security Alpha", |
| "role": "Security", |
| "channel": "radio://security-alpha", |
| "rooms": ["LOBBY", "SERVICE CORRIDOR"], |
| "lat": 12.9718, |
| "lng": 77.5944, |
| }, |
| { |
| "id": "medic-team", |
| "name": "Medic Unit", |
| "role": "Medical", |
| "channel": "app://medic", |
| "rooms": ["BALLROOM", "POOL DECK", "RESTAURANT B"], |
| "lat": 12.9719, |
| "lng": 77.5959, |
| }, |
| { |
| "id": "ops-control", |
| "name": "Ops Control", |
| "role": "Control", |
| "channel": "dashboard://command", |
| "rooms": ["LOBBY", "RESTAURANT B", "FLOOR 3", "POOL DECK", "BALLROOM", "SERVICE CORRIDOR"], |
| "lat": 12.9720, |
| "lng": 77.5951, |
| }, |
| ] |
|
|
|
|
| def _haversine_km(lat1: float, lng1: float, lat2: float, lng2: float) -> float: |
| radius = 6371.0 |
| phi1 = math.radians(lat1) |
| phi2 = math.radians(lat2) |
| dphi = math.radians(lat2 - lat1) |
| dlambda = math.radians(lng2 - lng1) |
| a = math.sin(dphi / 2) ** 2 + math.cos(phi1) * math.cos(phi2) * math.sin(dlambda / 2) ** 2 |
| return radius * (2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))) |
|
|
|
|
| def infer_incident_coordinates(location: str | None, lat: float | None, lng: float | None) -> tuple[float | None, float | None]: |
| if lat is not None and lng is not None: |
| return float(lat), float(lng) |
| if not location: |
| return None, None |
| return ROOM_COORDINATES.get(location.upper(), (None, None)) |
|
|
|
|
| def route_alert(alert_payload: dict[str, Any], radius_km: float = 0.45) -> list[dict[str, Any]]: |
| location = alert_payload.get("location") |
| lat = alert_payload.get("lat") |
| lng = alert_payload.get("lng") |
| incident_lat, incident_lng = infer_incident_coordinates(location, lat, lng) |
| resolved_location = (location or "").upper() |
| recipients: list[dict[str, Any]] = [] |
|
|
| for responder in RESPONDER_DIRECTORY: |
| direct_room_match = bool(resolved_location and resolved_location in [room.upper() for room in responder["rooms"]]) |
| nearby = False |
| distance_km = None |
| if incident_lat is not None and incident_lng is not None: |
| distance_km = _haversine_km(incident_lat, incident_lng, responder["lat"], responder["lng"]) |
| nearby = distance_km <= radius_km |
|
|
| if direct_room_match or nearby or responder["role"] == "Control": |
| recipients.append( |
| { |
| "recipient_id": responder["id"], |
| "recipient_name": responder["name"], |
| "role": responder["role"], |
| "channel": responder["channel"], |
| "distance_km": round(distance_km, 3) if distance_km is not None else None, |
| "reason": "room-match" if direct_room_match else ("nearby-radius" if nearby else "control-fallback"), |
| "demo_routing": DEMO_ROUTING_DATA, |
| } |
| ) |
|
|
| return recipients |
|
|
