113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
import enum
|
|
import os
|
|
from typing import Any
|
|
import requests
|
|
from rec.utils import nextMonday
|
|
from collections import defaultdict
|
|
|
|
url = "https://routes.googleapis.com/directions/v2:computeRoutes"
|
|
API_KEY_ENVIRONMENT_VARIABLE = "ROUTING_API_KEY"
|
|
|
|
|
|
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=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.*",
|
|
"routes.distanceMeters,routes.duration,routes.staticDuration,routes.legs.steps.distanceMeters,routes.legs.steps.staticDuration,routes.legs.steps.travelMode",
|
|
}
|
|
|
|
body = {
|
|
"origin": {
|
|
# "address": origin_address
|
|
"location": {
|
|
"latLng": {
|
|
"latitude": origin_lat,
|
|
"longitude": origin_lon
|
|
}
|
|
}
|
|
},
|
|
"destination": {
|
|
"address": dest_address
|
|
# "location": {
|
|
# "latLng": {
|
|
# "latitude": dest_lat,
|
|
# "longitude": dest_lon
|
|
# }
|
|
# }
|
|
},
|
|
"travelMode": travel_mode.value,
|
|
# "2023-10-15T15:01:23.045123456Z"
|
|
"departureTime": monday9am.strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
|
|
"computeAlternativeRoutes": compute_alternative_routes,
|
|
# "routeModifiers": {
|
|
# "avoidTolls": false,
|
|
# "avoidHighways": false,
|
|
# "avoidFerries": false
|
|
# },
|
|
"languageCode": "en-US",
|
|
"units": "METRIC",
|
|
}
|
|
|
|
r = requests.post(url, json=body, headers=header)
|
|
if r.status_code == 200:
|
|
return r.json()
|
|
|
|
raise Exception(r.json())
|
|
|
|
|
|
def extract_time(d, limit: int = 2):
|
|
res = []
|
|
for route in d["routes"]:
|
|
distance = route["distanceMeters"]
|
|
duration = int(route["duration"].strip("s"))
|
|
duration_static = int(route["staticDuration"].strip("s"))
|
|
|
|
steps = route["legs"][0]["steps"]
|
|
initial_walk_duration = 0
|
|
used_transit = False
|
|
duration_per_transit = defaultdict(lambda: 0)
|
|
distance_per_transit = defaultdict(lambda: 0)
|
|
number_of_transit_stops = 0
|
|
|
|
for step in steps:
|
|
if used_transit == False and step["travelMode"] == "WALK":
|
|
initial_walk_duration += int(step["staticDuration"].strip("s"))
|
|
else:
|
|
used_transit = True
|
|
duration_per_transit[step["travelMode"]] += int(
|
|
step["staticDuration"].strip("s"))
|
|
distance_per_transit[step["travelMode"]] += step.get(
|
|
"distanceMeters", 0)
|
|
if step["travelMode"] == "TRANSIT":
|
|
number_of_transit_stops += 1
|
|
|
|
res.append({
|
|
"duration": duration,
|
|
"distance": distance,
|
|
"duration_static": duration_static,
|
|
"initial_walk_duration": initial_walk_duration,
|
|
"duration_per_transit": dict(duration_per_transit),
|
|
"distance_per_transit": dict(distance_per_transit),
|
|
"number_of_transit_stops": number_of_transit_stops,
|
|
})
|
|
|
|
return res[:limit]
|