105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
import pandas as pd
|
|
import requests
|
|
import os
|
|
from tatami.classes import Participant, Group
|
|
|
|
|
|
GOOGLE_MAPS_API_URL = (
|
|
"https://routes.googleapis.com/distanceMatrix/v2:computeRouteMatrix"
|
|
)
|
|
GOOGLE_MAPS_API_KEY = os.getenv("GOOGLE_MAPS_API_KEY")
|
|
if not GOOGLE_MAPS_API_KEY:
|
|
raise ValueError("GOOGLE_MAPS_API_KEY environment variable is not set.")
|
|
|
|
|
|
def get_distance_matrix(
|
|
addresses: list[str], mode: str = "BICYCLE", value: str = "duration"
|
|
) -> pd.DataFrame:
|
|
if not addresses:
|
|
raise ValueError("The list of addresses cannot be empty.")
|
|
|
|
locations = [{"waypoint": {"address": address}} for address in addresses]
|
|
|
|
params = {
|
|
"origins": locations,
|
|
"destinations": locations,
|
|
"travelMode": mode,
|
|
}
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Goog-Api-Key": GOOGLE_MAPS_API_KEY,
|
|
"X-Goog-FieldMask": "originIndex,destinationIndex,duration,distanceMeters",
|
|
}
|
|
|
|
response = requests.post(GOOGLE_MAPS_API_URL, json=params, headers=headers)
|
|
|
|
if response.status_code != 200:
|
|
raise Exception(f"Error fetching data from Google Maps API: {response.text}")
|
|
|
|
data = response.json()
|
|
df = pd.DataFrame(
|
|
data
|
|
) # Keys: originIndex, destinationIndex, duration, distanceMeters
|
|
distance_matrix = pd.DataFrame(
|
|
index=df["originIndex"].unique(), columns=df["destinationIndex"].unique()
|
|
)
|
|
for index, row in df.iterrows():
|
|
origin = row["originIndex"]
|
|
destination = row["destinationIndex"]
|
|
if value == "duration":
|
|
distance_matrix.at[origin, destination] = pd.to_timedelta(row["duration"])
|
|
elif value == "distanceMeters":
|
|
distance_matrix.at[origin, destination] = float(row["distanceMeters"])
|
|
else:
|
|
raise ValueError(
|
|
"Invalid value specified. Use 'duration' or 'distanceMeters'."
|
|
)
|
|
|
|
# Sort the DataFrame by index and columns
|
|
distance_matrix = distance_matrix.sort_index().sort_index(axis=1)
|
|
return distance_matrix
|
|
|
|
|
|
def get_participant_distance_matrix(
|
|
participants: list[Participant], mode: str = "BICYCLE"
|
|
) -> pd.DataFrame:
|
|
addresses = [p.address for p in participants]
|
|
|
|
distance_matrix = get_distance_matrix(addresses, mode)
|
|
|
|
distance_matrix.index = pd.Index([p.uuid for p in participants])
|
|
distance_matrix.columns = pd.Index([p.uuid for p in participants])
|
|
|
|
return distance_matrix
|
|
|
|
|
|
def reduce_distance_matrix(
|
|
distance_matrix: pd.DataFrame, groups: list[Group]
|
|
) -> pd.DataFrame:
|
|
uuids = [g.main_member.uuid for g in groups]
|
|
reduced_distance_matrix = distance_matrix.loc[uuids, uuids]
|
|
reduced_distance_matrix.columns = pd.Index(uuids)
|
|
reduced_distance_matrix.index = pd.Index(uuids)
|
|
for group in groups:
|
|
reduced_distance_matrix.loc[group.main_member.uuid] += (
|
|
group.main_member.get_penalty()
|
|
)
|
|
|
|
return reduced_distance_matrix
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# Example usage
|
|
participants = [
|
|
Participant("Alice", "Römerstr. 12 76189 Karlsruhe", "555-1234", 8.0, "None"),
|
|
Participant(
|
|
"Bob", "Gottesauerstr. 30 76131 Karlsruhe", "555-5678", 7.5, "Peanuts"
|
|
),
|
|
Participant(
|
|
"Charlie", "Hermann-Hesse-Str 50 76189 Karlsruhe", "555-8765", 9.0, "None"
|
|
),
|
|
]
|
|
|
|
distance_matrix = get_participant_distance_matrix(participants)
|
|
print(distance_matrix)
|