""" Google Maps Distance Matrix API — Monsoon Speed Validation Script ================================================================= Collects real-world travel time data for the Bhopal corridor used in APOO validation, comparing clear vs pessimistic (monsoon-proxy) conditions. API Key Required: Google Maps Distance Matrix API (or DistanceMatrix.ai) Corridor: Bhopal Hoshangabad Road / Arera Colony arterial Total Length: ~2.5 km across 5 segments Usage: export GOOGLE_MAPS_API_KEY="your_key_here" python google_maps_api_collection.py Author: APOO Validation Team Date: 2026-05-06 """ import os import sys import time import json import urllib.request from datetime import datetime, timedelta # ============================================================ # CONFIGURATION # ============================================================ # API Configuration GOOGLE_API_KEY = os.environ.get("GOOGLE_MAPS_API_KEY", "") DISTANCEMATRIX_AI_KEY = os.environ.get("DISTANCEMATRIX_AI_KEY", "") USE_DISTANCEMATRIX_AI = not DISTANCEMATRIX_AI_KEY == "" # Bhopal Corridor Coordinates (from APOO validation/apoo_validation_results.csv) WAYPOINTS = [ # (origin_lat, origin_lon, dest_lat, dest_lon, segment_name, expected_length_m) (23.2599, 77.4126, 23.2555, 77.4110, "Segment_1_MP_Nagar_to_Arera_Hills", 500), (23.2555, 77.4110, 23.2512, 77.4095, "Segment_2_Arera_Hills_to_DB_City", 520), (23.2512, 77.4095, 23.2470, 77.4080, "Segment_3_DB_City_to_New_Market", 480), (23.2470, 77.4080, 23.2428, 77.4065, "Segment_4_New_Market_to_TT_Nagar", 510), (23.2428, 77.4065, 23.2385, 77.4050, "Segment_5_TT_Nagar_to_Roshanpura", 490), ] # Departure times for testing (IST = UTC+5:30) # Next Monday 09:00 AM IST = 03:30 UTC def next_monday_9am_utc(): """Get timestamp for next Monday 09:00 AM IST.""" now = datetime.utcnow() days_until_monday = (7 - now.weekday()) % 7 if days_until_monday == 0: days_until_monday = 7 # Next Monday, not today next_monday = now + timedelta(days=days_until_monday) next_monday = next_monday.replace(hour=3, minute=30, second=0, microsecond=0) return int(next_monday.timestamp()) # Test conditions CONDITIONS = [ { "name": "Clear_Weekday_Peak", "traffic_model": "best_guess", "departure_time": "now", # Or use next_monday_9am_utc() "description": "Clear weather baseline — weekday morning peak", }, { "name": "Pessimistic_Monsoon_Proxy", "traffic_model": "pessimistic", "departure_time": "now", "description": "Monsoon proxy — pessimistic traffic model simulates worst-case congestion + incidents", }, { "name": "Optimistic_Clear_Comparison", "traffic_model": "optimistic", "departure_time": "now", "description": "Optimistic model — upper bound for clear conditions", }, ] # ============================================================ # API CALL FUNCTIONS # ============================================================ def call_google_maps_distance_matrix(origins, destinations, departure_time=None, traffic_model="best_guess", api_key=""): """ Call Google Maps Distance Matrix API. Args: origins: str like "lat1,lon1|lat2,lon2" or single "lat,lon" destinations: str like "lat1,lon1|lat2,lon2" or single "lat,lon" departure_time: "now" or Unix timestamp traffic_model: "best_guess", "pessimistic", or "optimistic" api_key: Google Maps API key Returns: dict with parsed results or error """ if not api_key: return {"error": "No API key provided"} url = (f"https://maps.googleapis.com/maps/api/distancematrix/json" f"?origins={origins}" f"&destinations={destinations}" f"&mode=driving" f"&traffic_model={traffic_model}" f"&key={api_key}") if departure_time: url += f"&departure_time={departure_time}" try: req = urllib.request.Request(url, headers={'User-Agent': 'APOO-Validation/1.0'}) resp = urllib.request.urlopen(req, timeout=30) data = json.loads(resp.read()) if data.get("status") != "OK": return {"error": f"API status: {data.get('status')}", "raw": data} return {"status": "OK", "raw": data} except Exception as e: return {"error": str(e)} def call_distancematrix_ai(origins, destinations, api_key=""): """ Call DistanceMatrix.ai API (alternative provider). Note: This provider does not support traffic_model parameter. """ if not api_key: return {"error": "No API key provided"} url = (f"https://api.distancematrix.ai/maps/api/distancematrix/json" f"?origins={origins}" f"&destinations={destinations}" f"&key={api_key}") try: req = urllib.request.Request(url, headers={'User-Agent': 'APOO-Validation/1.0'}) resp = urllib.request.urlopen(req, timeout=30) data = json.loads(resp.read()) return {"status": "OK", "raw": data} except Exception as e: return {"error": str(e)} def parse_google_response(data, waypoint_idx): """Extract duration, distance, and compute speed from Google response.""" try: row = data["raw"]["rows"][0] element = row["elements"][0] if element["status"] != "OK": return {"status": element["status"]} duration_s = element["duration_in_traffic"]["value"] distance_m = element["distance"]["value"] speed_kmh = (distance_m / 1000) / (duration_s / 3600) return { "status": "OK", "duration_s": duration_s, "distance_m": distance_m, "speed_kmh": round(speed_kmh, 2), "duration_text": element["duration_in_traffic"]["text"], "distance_text": element["distance"]["text"], } except KeyError as e: return {"status": "PARSE_ERROR", "error": str(e), "raw_element": element} def parse_distancematrix_ai_response(data, waypoint_idx): """Extract duration, distance from DistanceMatrix.ai response.""" try: row = data["raw"]["rows"][0] element = row["elements"][0] duration_s = element["duration"]["value"] distance_m = element["distance"]["value"] speed_kmh = (distance_m / 1000) / (duration_s / 3600) return { "status": "OK", "duration_s": duration_s, "distance_m": distance_m, "speed_kmh": round(speed_kmh, 2), "duration_text": element["duration"]["text"], "distance_text": element["distance"]["text"], } except KeyError as e: return {"status": "PARSE_ERROR", "error": str(e)} # ============================================================ # MAIN COLLECTION PIPELINE # ============================================================ def collect_segment_data(orig_lat, orig_lon, dest_lat, dest_lon, segment_name, expected_length_m, condition, api_key): """Collect data for a single segment under a single condition.""" origins = f"{orig_lat},{orig_lon}" destinations = f"{dest_lat},{dest_lon}" if USE_DISTANCEMATRIX_AI: raw = call_distancematrix_ai(origins, destinations, api_key) parsed = parse_distancematrix_ai_response(raw, 0) else: raw = call_google_maps_distance_matrix( origins, destinations, departure_time=condition.get("departure_time"), traffic_model=condition.get("traffic_model", "best_guess"), api_key=api_key ) parsed = parse_google_response(raw, 0) return { "segment_name": segment_name, "expected_length_m": expected_length_m, "condition": condition["name"], "traffic_model": condition.get("traffic_model", "N/A"), "description": condition["description"], "timestamp_utc": datetime.utcnow().isoformat(), **parsed, } def run_validation(): """Run the full validation pipeline.""" api_key = DISTANCEMATRIX_AI_KEY if USE_DISTANCEMATRIX_AI else GOOGLE_API_KEY if not api_key: print("ERROR: No API key configured.") print("Set either GOOGLE_MAPS_API_KEY or DISTANCEMATRIX_AI_KEY environment variable.") sys.exit(1) print("=" * 70) print("APOO Monsoon Speed Validation — Google Maps API Collection") print("=" * 70) print(f"Provider: {'DistanceMatrix.ai' if USE_DISTANCEMATRIX_AI else 'Google Maps'}") print(f"Corridor: Bhopal Hoshangabad Road / Arera Colony") print(f"Total Segments: {len(WAYPOINTS)}") print(f"Total Distance: ~{sum(w[5] for w in WAYPOINTS)} m") print("=" * 70) all_results = [] for condition in CONDITIONS: print(f"\n--- Condition: {condition['name']} ---") print(f" Description: {condition['description']}") print(f" Traffic Model: {condition.get('traffic_model', 'N/A')}") condition_results = [] for i, (orig_lat, orig_lon, dest_lat, dest_lon, seg_name, exp_len) in enumerate(WAYPOINTS): print(f" Segment {i+1}: {seg_name} ({exp_len}m)...", end=" ") result = collect_segment_data( orig_lat, orig_lon, dest_lat, dest_lon, seg_name, exp_len, condition, api_key ) condition_results.append(result) if result.get("status") == "OK": print(f"OK — {result['duration_text']} ({result['distance_text']}) " f"→ {result['speed_kmh']:.1f} km/h") else: print(f"ERROR — {result.get('status', 'Unknown')}: {result.get('error', '')}") time.sleep(1.0) # Rate limit compliance # Compute corridor totals for this condition total_distance = sum(r["distance_m"] for r in condition_results if r.get("status") == "OK") total_duration = sum(r["duration_s"] for r in condition_results if r.get("status") == "OK") avg_speed = (total_distance / 1000) / (total_duration / 3600) if total_duration > 0 else 0 summary = { "condition": condition["name"], "total_distance_m": total_distance, "total_duration_s": total_duration, "avg_speed_kmh": round(avg_speed, 2), "segments": condition_results, } all_results.append(summary) print(f" Corridor Summary: {total_distance}m in {total_duration}s → {avg_speed:.1f} km/h") # Save results output_file = f"/app/bhopal_monsoon_validation_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json" with open(output_file, "w") as f: json.dump({ "metadata": { "corridor": "Bhopal Hoshangabad Road / Arera Colony", "total_segments": len(WAYPOINTS), "total_expected_distance_m": sum(w[5] for w in WAYPOINTS), "api_provider": "DistanceMatrix.ai" if USE_DISTANCEMATRIX_AI else "Google Maps", "collection_timestamp_utc": datetime.utcnow().isoformat(), "conditions_tested": [c["name"] for c in CONDITIONS], }, "results": all_results, }, f, indent=2) print(f"\n{'=' * 70}") print(f"Results saved to: {output_file}") # Speed reduction analysis if len(all_results) >= 2: clear_speed = all_results[0]["avg_speed_kmh"] pessimistic_speed = all_results[1]["avg_speed_kmh"] reduction_pct = ((clear_speed - pessimistic_speed) / clear_speed * 100) if clear_speed > 0 else 0 print(f"\n--- SPEED REDUCTION ANALYSIS ---") print(f"Clear Weather Avg Speed: {clear_speed:.1f} km/h") print(f"Pessimistic/Monsoon Avg Speed: {pessimistic_speed:.1f} km/h") print(f"Speed Reduction: {reduction_pct:.1f}%") print(f"APOO Simulation Assumption: 35.0%") print(f"Difference from APOO: {reduction_pct - 35.0:.1f} percentage points") if 20 <= reduction_pct <= 50: print(f"\nVERDICT: ✅ Real-world speed reduction ({reduction_pct:.1f}%) falls within") print(f" the 20-50% range documented in Indian monsoon literature.") print(f" APOO's 35% assumption is VALIDATED.") elif reduction_pct < 20: print(f"\nVERDICT: ⚠️ Real-world reduction ({reduction_pct:.1f}%) is LOWER than") print(f" literature. Monsoon proxy may not capture full impact.") else: print(f"\nVERDICT: ✅ Real-world reduction ({reduction_pct:.1f}%) is HIGHER than") print(f" APOO's 35%. The simulation is CONSERVATIVE.") print("=" * 70) return all_results if __name__ == "__main__": run_validation()