Initial Commit. I'm afraid of testing it.
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
def main() -> None:
|
||||
print("Hello from tatami-core!")
|
||||
@@ -0,0 +1,114 @@
|
||||
import datetime as dt
|
||||
import pandas as pd
|
||||
from uuid import uuid4
|
||||
|
||||
|
||||
class Participant:
|
||||
def __init__(
|
||||
self, name: str, address: str, phone: str, kitchen_size: float, allergies: str
|
||||
):
|
||||
self.uuid = str(uuid4())
|
||||
self.name = name
|
||||
self.address = address
|
||||
self.phone = phone
|
||||
self.kitchen_size = kitchen_size # bigger is better; range 0-10
|
||||
self.allergies = allergies
|
||||
|
||||
def get_penalty(self) -> dt.timedelta:
|
||||
return dt.timedelta(minutes=3 * (10 - self.kitchen_size))
|
||||
|
||||
def get_after_party_time(
|
||||
self, distance_matrix: pd.DataFrame, after_party_group: "Group"
|
||||
) -> dt.timedelta:
|
||||
return (
|
||||
self.get_penalty()
|
||||
+ pd.to_timedelta(
|
||||
distance_matrix.loc[self.uuid, after_party_group.main_member.uuid]
|
||||
).to_pytimedelta()
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"Participant(name={self.name}, uuid={self.uuid})[address={self.address}, phone={self.phone}, kitchen_size={self.kitchen_size}, allergies={self.allergies}]"
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.__repr__()
|
||||
|
||||
def dict(self) -> dict:
|
||||
return {
|
||||
"uuid": self.uuid,
|
||||
"name": self.name,
|
||||
"address": self.address,
|
||||
"phone": self.phone,
|
||||
"kitchen_size": self.kitchen_size,
|
||||
"allergies": self.allergies,
|
||||
}
|
||||
|
||||
|
||||
class Group:
|
||||
def __init__(self, members: list[Participant], main_member: int = 0):
|
||||
self.uuid = "Group_" + str(uuid4())
|
||||
self.members = members
|
||||
self.main_member = members[main_member]
|
||||
self.course: str | None = None # For ordering the groups allowed values: "starter", "main", "dessert"
|
||||
self.hosts: list[Group] | None = None
|
||||
|
||||
def set_course(self, course: str):
|
||||
self.course = course
|
||||
|
||||
def set_hosts(self, hosts: list["Group"]):
|
||||
self.hosts = hosts
|
||||
self.sort_hosts()
|
||||
|
||||
def sort_hosts(self):
|
||||
if self.hosts is None:
|
||||
return
|
||||
|
||||
courses = {"starter": 0, "main": 1, "dessert": 2}
|
||||
self.hosts.sort(key=lambda x: courses[x.course] if x.course in courses else 3)
|
||||
|
||||
def add_host(self, host: "Group"):
|
||||
if self.hosts is None:
|
||||
self.hosts = []
|
||||
self.hosts.append(host)
|
||||
self.sort_hosts()
|
||||
|
||||
def add_member(self, member: Participant, main_member: bool = False):
|
||||
if main_member:
|
||||
self.main_member = member
|
||||
self.members.append(member)
|
||||
|
||||
def get_total_time(
|
||||
self, distance_matrix: pd.DataFrame, after_party_group: "Group"
|
||||
) -> dt.timedelta:
|
||||
if self.hosts is None:
|
||||
raise ValueError("Hosts must be set.")
|
||||
if after_party_group.uuid in [h.uuid for h in self.hosts]:
|
||||
groups = self.hosts
|
||||
else:
|
||||
groups = self.hosts + [after_party_group]
|
||||
total_time = dt.timedelta()
|
||||
for group, next_group in zip(groups[:-1], groups[1:]):
|
||||
total_time += pd.to_timedelta(
|
||||
distance_matrix.loc[group.main_member.uuid, next_group.main_member.uuid]
|
||||
).to_pytimedelta()
|
||||
total_time += group.main_member.get_penalty()
|
||||
|
||||
return total_time
|
||||
|
||||
def dict(self) -> dict:
|
||||
return {
|
||||
"uuid": self.uuid,
|
||||
"members": [member.uuid for member in self.members],
|
||||
"main_member": self.main_member.uuid,
|
||||
"course": self.course,
|
||||
"hosts": [host.uuid for host in self.hosts] if self.hosts else None,
|
||||
}
|
||||
|
||||
def get_guests(self, groups: list["Group"]) -> list[Participant]:
|
||||
guests: list[Participant] = []
|
||||
if self.hosts is None:
|
||||
return guests
|
||||
for group in groups:
|
||||
if group.hosts is not None and self.uuid in [g.uuid for g in group.hosts]:
|
||||
guests.extend(group.members)
|
||||
return guests
|
||||
@@ -0,0 +1,210 @@
|
||||
from tatami.classes import Participant, Group
|
||||
from tatami.traveltimes import reduce_distance_matrix, get_participant_distance_matrix
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
import random
|
||||
from tqdm import tqdm
|
||||
from itertools import permutations
|
||||
|
||||
|
||||
def get_after_party_group(address: str) -> Group:
|
||||
"""
|
||||
Create a group for the after party with a single participant.
|
||||
"""
|
||||
participant = Participant(
|
||||
name="After Party", address=address, phone="", kitchen_size=10, allergies=""
|
||||
)
|
||||
return Group(members=[participant], main_member=0)
|
||||
|
||||
|
||||
def get_masterplan(
|
||||
participants: list[Participant],
|
||||
distance_matrix: pd.DataFrame,
|
||||
after_party_group: Group,
|
||||
) -> tuple[list[dict], list[dict]]:
|
||||
groups_per_course = np.floor(len(participants) / 6).astype(int)
|
||||
participants.sort(
|
||||
key=lambda x: x.get_after_party_time(distance_matrix, after_party_group)
|
||||
)
|
||||
hosts = participants[: 3 * groups_per_course]
|
||||
semi_hosts = participants[3 * groups_per_course :]
|
||||
courses = ["starter", "main", "dessert"] * groups_per_course
|
||||
|
||||
groups = []
|
||||
for host in hosts:
|
||||
group = Group(members=[host])
|
||||
groups.append(group)
|
||||
|
||||
random.shuffle(semi_hosts)
|
||||
for i, member in enumerate(semi_hosts):
|
||||
groups[i % len(groups)].add_member(member)
|
||||
|
||||
distance_matrix = reduce_distance_matrix(
|
||||
distance_matrix, [*groups, after_party_group]
|
||||
)
|
||||
best_order = run_simulated_annealing(
|
||||
groups,
|
||||
distance_matrix,
|
||||
initial_temperature=1000,
|
||||
cooling_rate=0.99,
|
||||
max_iterations=10000,
|
||||
multiprocessing=1,
|
||||
)
|
||||
assign_courses(best_order, courses)
|
||||
group_dicts = [group.dict() for group in best_order]
|
||||
participant_dicts = [participant.dict() for participant in participants]
|
||||
return group_dicts, participant_dicts
|
||||
|
||||
|
||||
def assign_courses(groups: list[Group], courses: list[str]) -> None:
|
||||
"""
|
||||
Assign courses to groups.
|
||||
"""
|
||||
for i, (group, course) in enumerate(zip(groups, courses)):
|
||||
group.set_course(course)
|
||||
hosts = [groups[i] for i in get_courses(i, len(groups))]
|
||||
group.set_hosts(hosts)
|
||||
|
||||
|
||||
def run_simulated_annealing(
|
||||
groups: list[Group],
|
||||
reduced_distance_matrix: pd.DataFrame,
|
||||
initial_temperature: float,
|
||||
cooling_rate: float,
|
||||
max_iterations: int,
|
||||
multiprocessing: int = 1,
|
||||
) -> list[Group]:
|
||||
"""
|
||||
Run simulated annealing to find the optimal order of groups.
|
||||
"""
|
||||
group_indices = list(range(len(groups)))
|
||||
distance_matrix = reduced_distance_matrix.to_numpy()
|
||||
|
||||
# Convert group indices to a list of integers
|
||||
group_indices = [int(i) for i in group_indices]
|
||||
|
||||
if multiprocessing > 1:
|
||||
raise NotImplementedError("Multiprocessing is not implemented yet.")
|
||||
else:
|
||||
# Run simulated annealing without multiprocessing
|
||||
best_order = simulated_annealing(
|
||||
distance_matrix,
|
||||
group_indices,
|
||||
initial_temperature,
|
||||
cooling_rate,
|
||||
max_iterations,
|
||||
)
|
||||
best_ordererd_groups = [groups[i] for i in best_order]
|
||||
return best_ordererd_groups
|
||||
|
||||
|
||||
def next_permutation(
|
||||
group_indices: list[int],
|
||||
distance_matrix: np.ndarray,
|
||||
T: float,
|
||||
) -> list[int]:
|
||||
"""
|
||||
Generate the next permutation of group indices that minimizes the total travel time using boltzmann annealing.
|
||||
"""
|
||||
all_permutations = list(permutations(group_indices))
|
||||
times = {
|
||||
i: fast_total_time(distance_matrix, perm)
|
||||
for i, perm in enumerate(all_permutations)
|
||||
}
|
||||
min_time = min(times.values())
|
||||
probabilities = {i: np.exp(-(time - min_time) / T) for i, time in times.items()}
|
||||
key = np.random.choice(list(probabilities.keys()), p=list(probabilities.values()))
|
||||
return all_permutations[key]
|
||||
|
||||
|
||||
def simulated_annealing(
|
||||
distance_matrix: np.ndarray,
|
||||
group_indices: list[int],
|
||||
initial_temperature: float,
|
||||
cooling_rate: float,
|
||||
max_iterations: int,
|
||||
) -> list[int]:
|
||||
"""
|
||||
Simulated annealing algorithm to find the optimal order of groups.
|
||||
"""
|
||||
solution = group_indices.copy()
|
||||
np.random.shuffle(solution)
|
||||
current_temperature = initial_temperature
|
||||
|
||||
for iteration in tqdm(range(max_iterations)):
|
||||
try:
|
||||
solution = next_permutation(solution, distance_matrix, current_temperature)
|
||||
time = fast_total_time(distance_matrix, solution)
|
||||
if iteration % 100 == 0:
|
||||
print(f"Iteration {iteration}: Time = {time}")
|
||||
current_temperature *= cooling_rate
|
||||
except KeyboardInterrupt:
|
||||
print("Simulation interrupted. Returning current solution.")
|
||||
break
|
||||
return solution
|
||||
|
||||
|
||||
def get_courses(
|
||||
group_index: int,
|
||||
total_meetings: int,
|
||||
):
|
||||
a = group_index
|
||||
b = (group_index + 1) % total_meetings
|
||||
c = (group_index - 4) % total_meetings
|
||||
order = sorted((a, b, c), key=lambda x: x % 3)
|
||||
return order
|
||||
|
||||
|
||||
def fast_total_time(distance_matrix: np.ndarray, group_indices: list[int]):
|
||||
"""Calculates the total travel time for all groups.
|
||||
|
||||
Args:
|
||||
distance_matrix (np.ndarray): 2D array representing the distance matrix, where
|
||||
distance_matrix[i][j] is the travel time from group i to group j in seconds. Should be of shape (n+1, n+1). The last column/row should be the after party group.
|
||||
group_indices (list[int]): List of group indices for which to calculate the total travel time. Should be of length n.
|
||||
"""
|
||||
n = len(group_indices)
|
||||
after_party_idx = n # Since distance_matrix is (n+1)x(n+1)
|
||||
|
||||
total_time = 0
|
||||
total_meetings = n
|
||||
|
||||
for group_index in group_indices:
|
||||
a, b, c = get_courses(group_index, total_meetings)
|
||||
total_time += distance_matrix[a][b]
|
||||
total_time += distance_matrix[b][c]
|
||||
total_time += distance_matrix[c][after_party_idx]
|
||||
|
||||
return total_time
|
||||
|
||||
|
||||
def load_csv_to_participants(file_path: str) -> list[Participant]:
|
||||
"""
|
||||
Load participants from a CSV file.
|
||||
"""
|
||||
df = pd.read_csv(file_path, sep="\t")
|
||||
participants = []
|
||||
for _, row in df.iterrows():
|
||||
participant = Participant(
|
||||
name=row["name"],
|
||||
address=row["address"],
|
||||
phone=row["phone"],
|
||||
kitchen_size=row["kitchen_size"],
|
||||
allergies=row["allergies"],
|
||||
)
|
||||
participants.append(participant)
|
||||
return participants
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Example usage
|
||||
participants = load_csv_to_participants("test-config.csv")
|
||||
after_party_group = get_after_party_group(
|
||||
"Sebastian-Kneipp-Straße 6, 76131 Karlsruhe"
|
||||
)
|
||||
distance_matrix = get_participant_distance_matrix(
|
||||
[*participants, after_party_group.main_member], mode="BICYCLE"
|
||||
)
|
||||
masterplan = get_masterplan(participants, distance_matrix, after_party_group)
|
||||
print(masterplan)
|
||||
print("Masterplan generated successfully.")
|
||||
@@ -0,0 +1,104 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user