wrongmove/crawler/rec/routing.py

import enum
import os
from typing import Any
import requests
from rec.utils import nextMonday
from rec.exceptions import RoutingApiError

ROUTES_API_URL = "https://routes.googleapis.com/directions/v2:computeRoutes"
API_KEY_ENVIRONMENT_VARIABLE = "ROUTING_API_KEY"
ROUTES_FIELD_MASK = (
    "routes.distanceMeters,"
    "routes.duration,"
    "routes.staticDuration,"
    "routes.legs.steps.distanceMeters,"
    "routes.legs.steps.staticDuration,"
    "routes.legs.steps.travelMode"
)


class TravelMode(enum.StrEnum):
    TRANSIT = "TRANSIT"
    BICYCLE = "BICYCLE"
    WALK = "WALK"
    DRIVE = "DRIVE"


def transit_route(
    origin_lat: float,
    origin_lon: float,
    dest_address: str,
    travel_mode: TravelMode,
    compute_alternative_routes: bool = True,
) -> dict[str, Any]:
    monday9am = nextMonday()

    # must be set
    api_key = os.environ[API_KEY_ENVIRONMENT_VARIABLE]

    header = {
        "X-Goog-Api-Key": api_key,
        "Content-Type": "application/json",
        "X-Goog-FieldMask": ROUTES_FIELD_MASK,
    }

    body = {
        "origin": {
            "location": {"latLng": {"latitude": origin_lat, "longitude": origin_lon}}
        },
        "destination": {
            "address": dest_address
        },
        "travelMode": travel_mode.value,
        "departureTime": monday9am.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
        "computeAlternativeRoutes": compute_alternative_routes,
        "languageCode": "en-US",
        "units": "METRIC",
    }

    r = requests.post(ROUTES_API_URL, json=body, headers=header)
    if r.status_code == 200:
        return r.json()

    raise RoutingApiError(r.status_code, r.json())