google_distance.py

"""Real driving distances/times via the Google Maps Routes API.

Uses the computeRouteMatrix endpoint (the successor to the legacy Distance
Matrix API). Produces the same (miles, minutes) matrices as
distance.build_matrices, so solver.py is unchanged.

Setup:
  1. In Google Cloud Console, create a project, enable the "Routes API",
     and create an API key (billing must be enabled on the project).
  2. export GOOGLE_MAPS_API_KEY="your-key"
  3. python main.py --distance-source google

Cost awareness:
  Google bills per matrix *element* (origin x destination pair). A problem
  with A adjusters and C claims needs (A+C)^2 elements — e.g. 30 locations
  = 900 elements per uncached build. Results are cached on disk keyed by
  the exact set of coordinates, so re-solving the same day's data is free.
  For heavy experimentation, a self-hosted OSRM server gives similar
  road-network accuracy at zero marginal cost.
"""

from __future__ import annotations

import json
import os
import urllib.error
import urllib.request

import config
from data_gen import Adjuster, Claim
from distance import haversine_miles

ENDPOINT = "https://routes.googleapis.com/distanceMatrix/v2:computeRouteMatrix"
# computeRouteMatrix allows at most 625 elements per request with
# TRAFFIC_UNAWARE routing, so tile the full matrix into 25x25 blocks.
BLOCK = 25
METERS_PER_MILE = 1609.344

CACHE_PATH = os.path.join(config.DATA_DIR, "google_matrix_cache.json")


class GoogleMatrixError(RuntimeError):
    pass


def _coords(adjusters: list[Adjuster], claims: list[Claim]) -> list[list[float]]:
    pts = [(a.home_lat, a.home_lon) for a in adjusters]
    pts += [(c.lat, c.lon) for c in claims]
    # Round so cache keys are stable across float formatting.
    return [[round(lat, 6), round(lon, 6)] for lat, lon in pts]


def _load_cache(coords: list[list[float]]):
    if not os.path.exists(CACHE_PATH):
        return None
    try:
        with open(CACHE_PATH) as f:
            cache = json.load(f)
    except (json.JSONDecodeError, OSError):
        return None
    if cache.get("coords") == coords:
        return cache["miles"], cache["minutes"]
    return None


def _save_cache(coords, miles, minutes) -> None:
    os.makedirs(config.DATA_DIR, exist_ok=True)
    with open(CACHE_PATH, "w") as f:
        json.dump({"coords": coords, "miles": miles, "minutes": minutes}, f)


def _waypoint(lat: float, lon: float) -> dict:
    return {"waypoint": {"location": {"latLng": {"latitude": lat,
                                                 "longitude": lon}}}}


def _request_block(origins, destinations, api_key: str) -> list[dict]:
    """POST one origins x destinations block; returns the element list."""
    body = {
        "origins": [_waypoint(lat, lon) for lat, lon in origins],
        "destinations": [_waypoint(lat, lon) for lat, lon in destinations],
        "travelMode": "DRIVE",
        "routingPreference": config.GOOGLE_ROUTING_PREFERENCE,
    }
    req = urllib.request.Request(
        ENDPOINT,
        data=json.dumps(body).encode(),
        headers={
            "Content-Type": "application/json",
            "X-Goog-Api-Key": api_key,
            "X-Goog-FieldMask":
                "originIndex,destinationIndex,distanceMeters,duration,condition",
        },
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.loads(resp.read())
    except urllib.error.HTTPError as e:
        detail = e.read().decode(errors="replace")[:500]
        raise GoogleMatrixError(
            f"Routes API returned HTTP {e.code}: {detail}") from e
    except urllib.error.URLError as e:
        raise GoogleMatrixError(f"Could not reach Routes API: {e.reason}") from e


def build_matrices(adjusters: list[Adjuster], claims: list[Claim],
                   api_key: str | None = None, use_cache: bool = True,
                   ) -> tuple[list[list[float]], list[list[int]]]:
    """Return (miles, minutes) matrices from real Google driving routes.

    Note: unlike the haversine matrices these are asymmetric (one-way
    streets, divided highways), which the solver handles natively.
    """
    api_key = api_key or os.environ.get("GOOGLE_MAPS_API_KEY")
    if not api_key:
        raise GoogleMatrixError(
            "No API key. Set the GOOGLE_MAPS_API_KEY environment variable "
            "(Google Cloud Console -> enable 'Routes API' -> create key).")

    coords = _coords(adjusters, claims)
    if use_cache:
        cached = _load_cache(coords)
        if cached:
            print(f"Using cached Google matrix ({CACHE_PATH})")
            return cached

    n = len(coords)
    n_elements = n * n
    n_requests = ((n + BLOCK - 1) // BLOCK) ** 2
    print(f"Fetching Google route matrix: {n} locations, "
          f"{n_elements} billable elements, {n_requests} request(s) ...")

    miles = [[0.0] * n for _ in range(n)]
    minutes = [[0] * n for _ in range(n)]
    fallbacks = 0
    for oi in range(0, n, BLOCK):
        for di in range(0, n, BLOCK):
            o_block = coords[oi:oi + BLOCK]
            d_block = coords[di:di + BLOCK]
            for el in _request_block(o_block, d_block, api_key):
                i = oi + el.get("originIndex", 0)
                j = di + el.get("destinationIndex", 0)
                if i == j:
                    continue
                if el.get("condition") == "ROUTE_EXISTS":
                    meters = el.get("distanceMeters", 0)
                    secs = int(el.get("duration", "0s").rstrip("s"))
                    miles[i][j] = meters / METERS_PER_MILE
                    minutes[i][j] = int(round(secs / 60))
                else:
                    # No drivable route reported; estimate so the solver
                    # still has a finite (but discouraging) value.
                    d = haversine_miles(coords[i][0], coords[i][1],
                                        coords[j][0], coords[j][1])
                    d *= config.CIRCUITY_FACTOR * 2
                    miles[i][j] = d
                    minutes[i][j] = int(round(d / config.AVG_SPEED_MPH * 60))
                    fallbacks += 1
    if fallbacks:
        print(f"Warning: {fallbacks} pair(s) had no route; "
              f"used pessimistic haversine estimates.")

    if use_cache:
        _save_cache(coords, miles, minutes)
    return miles, minutes