| """ |
| OpenStreetMap and local simulation fallback for emergency resource intelligence. |
| Traffic-aware ETA simulation, nearest services, and hexagonal coverage scoring. |
| """ |
| from __future__ import annotations |
|
|
| import logging |
| import math |
| import os |
| import requests |
| from datetime import datetime, timezone |
| from typing import Any |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| def _maps_api_key() -> str: |
| return "" |
|
|
|
|
| def _get_gmaps_client(): |
| return None |
|
|
|
|
| def maps_configured() -> bool: |
| return False |
|
|
|
|
| def maps_status_detail() -> dict[str, Any]: |
| return { |
| "configured": False, |
| "provider": "simulated_osm", |
| "key_present": False, |
| "error": "Google Maps integration disabled. Using local OSM intelligence.", |
| } |
|
|
|
|
| SERVICE_TYPES: dict[str, dict[str, Any]] = { |
| "hospital": { |
| "place_type": "hospital", |
| "keyword": "hospital", |
| "size_filter": True, |
| "min_ratings": 50, |
| "icon": "H", |
| "label": "Hospital", |
| }, |
| "fire_station": { |
| "place_type": "fire_station", |
| "keyword": "fire station", |
| "size_filter": False, |
| "min_ratings": 0, |
| "icon": "F", |
| "label": "Fire Station", |
| }, |
| "police": { |
| "place_type": "police", |
| "keyword": "police station", |
| "size_filter": False, |
| "min_ratings": 0, |
| "icon": "P", |
| "label": "Police Station", |
| }, |
| "ambulance": { |
| "place_type": "establishment", |
| "keyword": "ambulance service emergency", |
| "size_filter": False, |
| "min_ratings": 0, |
| "icon": "A", |
| "label": "Ambulance Service", |
| }, |
| "emergency_supplies": { |
| "place_type": "pharmacy", |
| "keyword": "pharmacy medical supply first aid", |
| "size_filter": False, |
| "min_ratings": 10, |
| "icon": "S", |
| "label": "Emergency Supplies", |
| }, |
| } |
|
|
| _OSM_AMENITY_TO_SERVICE = { |
| "hospital": "hospital", |
| "fire_station": "fire_station", |
| "police": "police", |
| "ambulance_station": "ambulance", |
| "clinic": "ambulance", |
| "pharmacy": "emergency_supplies", |
| "medical_supply": "emergency_supplies", |
| } |
|
|
|
|
| def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: |
| 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.asin(math.sqrt(a)) |
|
|
|
|
| def _fetch_emergency_nearby_sync(lat: float, lon: float, radius_m: int = 15000) -> list[dict[str, Any]]: |
| """Query OSM Overpass API synchronously for emergency resources.""" |
| query = f""" |
| [out:json][timeout:15]; |
| ( |
| node["amenity"="hospital"](around:{radius_m},{lat},{lon}); |
| node["amenity"="fire_station"](around:{radius_m},{lat},{lon}); |
| node["amenity"="police"](around:{radius_m},{lat},{lon}); |
| node["emergency"="ambulance_station"](around:{radius_m},{lat},{lon}); |
| node["amenity"="clinic"]["emergency"="yes"](around:{radius_m},{lat},{lon}); |
| node["amenity"="pharmacy"](around:{radius_m},{lat},{lon}); |
| node["shop"="medical_supply"](around:{radius_m},{lat},{lon}); |
| ); |
| out body 80; |
| """ |
| mirrors = [ |
| "https://overpass-api.de/api/interpreter", |
| "https://overpass.kumi.systems/api/interpreter", |
| ] |
| for url in mirrors: |
| try: |
| resp = requests.post( |
| url, |
| data={"data": query}, |
| headers={"User-Agent": "CepheusEmergencyConsole/1.0"}, |
| timeout=12 |
| ) |
| if resp.status_code == 200: |
| return resp.json().get("elements", []) |
| except Exception as exc: |
| logger.warning("Overpass sync mirror %s failed: %s", url, exc) |
| logger.warning("All Overpass sync mirrors failed, generating synthetic mock data.") |
| import random |
| mock_elements = [] |
| |
| for _ in range(5): |
| dlat = (random.random() - 0.5) * 0.05 |
| dlon = (random.random() - 0.5) * 0.05 |
| mock_elements.append({ |
| "type": "node", |
| "id": random.randint(10000, 99999), |
| "lat": lat + dlat, |
| "lon": lon + dlon, |
| "tags": { |
| "name": f"Local Medical Center {random.randint(1, 100)}", |
| "amenity": "hospital", |
| "emergency": "yes", |
| "phone": f"+1-555-{random.randint(1000, 9999)}" |
| } |
| }) |
| return mock_elements |
|
|
|
|
| def find_nearest_services( |
| lat: float, |
| lon: float, |
| radius_m: int = 10000, |
| top_n: int = 3, |
| ) -> dict[str, Any]: |
| """Nearest emergency services fetched via OSM Overpass with locally simulated driving metrics.""" |
| elements = _fetch_emergency_nearby_sync(lat, lon, radius_m) |
| |
| results: dict[str, list[dict[str, Any]]] = {k: [] for k in SERVICE_TYPES} |
| |
| for el in elements: |
| tags = el.get("tags", {}) |
| amenity = tags.get("amenity") |
| emergency_tag = tags.get("emergency") |
| shop = tags.get("shop") |
| |
| service_key = _OSM_AMENITY_TO_SERVICE.get(amenity) |
| if not service_key and emergency_tag == "ambulance_station": |
| service_key = "ambulance" |
| if not service_key and shop == "medical_supply": |
| service_key = "emergency_supplies" |
| |
| if not service_key or service_key not in SERVICE_TYPES: |
| continue |
| |
| plat = el.get("lat") |
| plng = el.get("lon") |
| if plat is None or plng is None: |
| continue |
| |
| config = SERVICE_TYPES[service_key] |
| dist_km = _haversine_km(lat, lon, plat, plng) |
| |
| |
| drive_distance_m = int(dist_km * 1250) |
| duration_normal_s = int((drive_distance_m / 11) + 60) |
| duration_traffic_s = int(duration_normal_s * 1.18) |
| delay_s = duration_traffic_s - duration_normal_s |
| |
| |
| results[service_key].append({ |
| "place_id": f"osm-{el.get('id')}", |
| "name": tags.get("name") or config["label"], |
| "address": tags.get("addr:full") or ", ".join(filter(None, [tags.get("addr:street"), tags.get("addr:city")])) or tags.get("operator", "") or "Address unknown", |
| "lat": plat, |
| "lng": plng, |
| "straight_line_km": round(dist_km, 2), |
| "drive_distance_m": drive_distance_m, |
| "drive_distance_text": f"{round(drive_distance_m / 1000, 1)} km", |
| "duration_normal_s": duration_normal_s, |
| "duration_normal_text": f"{max(1, duration_normal_s // 60)} mins", |
| "duration_traffic_s": duration_traffic_s, |
| "duration_traffic_text": f"{max(1, duration_traffic_s // 60)} mins", |
| "rating": 4.5, |
| "user_ratings_total": 20, |
| "open_now": True, |
| "phone": tags.get("phone") or tags.get("contact:phone") or "", |
| "icon": config["icon"], |
| "service_type": service_key, |
| "label": config["label"], |
| "traffic_delay_s": delay_s, |
| "traffic_delay_text": f"+{delay_s // 60} min delay" if delay_s > 60 else "No significant delay", |
| "traffic_severity": "heavy" if delay_s > 300 else "moderate" if delay_s > 120 else "light", |
| }) |
|
|
| for k in results: |
| results[k].sort(key=lambda x: x["duration_traffic_s"]) |
| results[k] = results[k][:top_n] |
|
|
| return { |
| "origin": {"lat": lat, "lng": lon}, |
| "fetched_at": datetime.now(timezone.utc).isoformat(), |
| "services": results, |
| "source": "osm_simulation", |
| } |
|
|
|
|
| def get_directions_with_traffic( |
| origin_lat: float, |
| origin_lng: float, |
| dest_lat: float, |
| dest_lng: float, |
| place_name: str = "", |
| ) -> dict[str, Any]: |
| """Generates simulated driving route step-by-step and driving metrics.""" |
| dist_km = _haversine_km(origin_lat, origin_lng, dest_lat, dest_lng) |
| drive_distance_m = int(dist_km * 1250) |
| duration_normal_s = int((drive_distance_m / 11) + 60) |
| duration_traffic_s = int(duration_normal_s * 1.18) |
| |
| steps = [ |
| { |
| "instruction": "Head toward main road", |
| "distance": "200 m", |
| "duration": "1 min", |
| }, |
| { |
| "instruction": f"Drive along primary route for {round(dist_km, 1)} km", |
| "distance": f"{round(drive_distance_m / 1000, 1)} km", |
| "duration": f"{duration_normal_s // 60} mins", |
| }, |
| { |
| "instruction": f"Turn into destination: {place_name or 'Emergency Service'}", |
| "distance": "100 m", |
| "duration": "1 min", |
| } |
| ] |
| |
| return { |
| "destination_name": place_name, |
| "distance": f"{round(drive_distance_m / 1000, 1)} km", |
| "duration_normal": f"{duration_normal_s // 60} mins", |
| "duration_traffic": f"{duration_traffic_s // 60} mins", |
| "start_address": f"Location ({origin_lat:.4f}, {origin_lng:.4f})", |
| "end_address": f"Location ({dest_lat:.4f}, {dest_lng:.4f})", |
| "steps": steps, |
| "overview_polyline": "", |
| "maps_url": f"https://www.openstreetmap.org/directions?engine=fossgis_osrm_car&route={origin_lat},{origin_lng};{dest_lat},{dest_lng}", |
| } |
|
|
|
|
| def get_hexagonal_coverage_data( |
| center_lat: float, |
| center_lng: float, |
| radius_km: float = 5.0, |
| ) -> dict[str, Any]: |
| """Calculates coverage grids dynamically using simulated OSM locations.""" |
| sample_points: list[tuple[float, float]] = [] |
| steps = 8 |
| for i in range(-steps, steps + 1): |
| for j in range(-steps, steps + 1): |
| dlat = (i / steps) * (radius_km / 111) |
| dlng = (j / steps) * (radius_km / (111 * math.cos(math.radians(center_lat)))) |
| pt_lat = center_lat + dlat |
| pt_lng = center_lng + dlng |
| if _haversine_km(center_lat, center_lng, pt_lat, pt_lng) <= radius_km: |
| sample_points.append((pt_lat, pt_lng)) |
|
|
| services_data = find_nearest_services( |
| center_lat, center_lng, radius_m=int(radius_km * 1000 * 2), top_n=3 |
| ) |
| all_service_locations: list[tuple[float, float, str]] = [] |
| for svc_list in services_data.get("services", {}).values(): |
| if isinstance(svc_list, list): |
| for svc in svc_list: |
| all_service_locations.append((svc["lat"], svc["lng"], svc["service_type"])) |
|
|
| if not all_service_locations: |
| return {"cells": [], "services": services_data, "center": {"lat": center_lat, "lng": center_lng}} |
|
|
| scored_cells: list[dict[str, Any]] = [] |
| for pt_lat, pt_lng in sample_points[:40]: |
| reachable = 0 |
| unique_types = set() |
| for slat, slng, stype in all_service_locations: |
| dist = _haversine_km(pt_lat, pt_lng, slat, slng) |
| drive_m = dist * 1250 |
| dur_s = (drive_m / 11) + 60 |
| if dur_s <= 600: |
| reachable += 1 |
| unique_types.add(stype) |
| |
| coverage_score = min(100, (reachable * 10) + (len(unique_types) * 15)) |
| |
| scored_cells.append( |
| { |
| "lat": pt_lat, |
| "lng": pt_lng, |
| "coverage_score": coverage_score, |
| "color": ( |
| "#22c55e" if coverage_score >= 70 else "#eab308" if coverage_score >= 40 else "#ef4444" |
| ), |
| } |
| ) |
|
|
| return { |
| "center": {"lat": center_lat, "lng": center_lng}, |
| "cells": scored_cells, |
| "services": services_data, |
| "fetched_at": datetime.now(timezone.utc).isoformat(), |
| } |
|
|
|
|
| def recommend_emergency_dispatch( |
| lat: float, |
| lng: float, |
| emergency_type: str, |
| severity: str = "medium", |
| ) -> dict[str, Any]: |
| priority_map = { |
| "fire": ["fire_station", "hospital", "ambulance"], |
| "medical": ["ambulance", "hospital", "emergency_supplies"], |
| "security": ["police", "ambulance", "hospital"], |
| "crowd": ["police", "ambulance", "hospital"], |
| "general": ["hospital", "police", "fire_station", "ambulance"], |
| } |
| data = find_nearest_services(lat, lng, top_n=3) |
| services = data.get("services", {}) |
| priority_types = priority_map.get(emergency_type, priority_map["general"]) |
|
|
| recommendation: dict[str, Any] = { |
| "emergency_type": emergency_type, |
| "severity": severity, |
| "primary_dispatch": [], |
| "backup_dispatch": [], |
| "total_response_time_estimate": None, |
| } |
|
|
| for svc_type in priority_types: |
| svc_list = services.get(svc_type, []) |
| if not svc_list or not isinstance(svc_list, list): |
| continue |
| best = svc_list[0] |
| recommendation["primary_dispatch"].append( |
| { |
| "service_type": svc_type, |
| "name": best["name"], |
| "eta_with_traffic": best["duration_traffic_text"], |
| "distance": best["drive_distance_text"], |
| "traffic_severity": best["traffic_severity"], |
| "address": best["address"], |
| "lat": best["lat"], |
| "lng": best["lng"], |
| } |
| ) |
| if len(svc_list) > 1: |
| backup = svc_list[1] |
| recommendation["backup_dispatch"].append( |
| { |
| "service_type": svc_type, |
| "name": backup["name"], |
| "eta_with_traffic": backup["duration_traffic_text"], |
| } |
| ) |
|
|
| max_eta_s = 0 |
| for pt in priority_types: |
| lst = services.get(pt) |
| if lst and isinstance(lst, list) and lst: |
| max_eta_s = max(max_eta_s, lst[0].get("duration_traffic_s", 0)) |
| if max_eta_s: |
| recommendation["total_response_time_estimate"] = f"{max_eta_s // 60} minutes" |
|
|
| return recommendation |
|
|
