Choosing the employee closest to the shift's location using optapy

I'm using the employee scheduling quick start on optapy's github page. Only difference is I modified the Employee class to have the employee's home address included.

@optapy.problem_fact class Employee: name: str skill_set: list[str] address: str
def __init__(self, name: str = None, skill_set: list[str] = None, address: str = None):
    self.name = name
    self.skill_set = skill_set
    self.address = address #new

def __str__(self):
    return f'Employee(name={self.name})'

def to_dict(self):
    return {
        'name': self.name,
        'skill_set': self.skill_set,
        'address': self.address
    }`

For example Instead of LOCATIONS = ["Ambulatory care", "Critical care", "Pediatric care"] as locations, each shift has a certain patient address as a location which caregivers can visit. From the original repository, the employees are matched to shifts based on their job expertise and availability as well as other constraints, but I would like to add the additional constraint of employees being matched to caregivers by the shortest distance to the patient as well.

I was thinking of using google maps API to retrieve the shortest possible route between two addresses:

import googlemaps
api_key = 'API_KEY'
def find_distance_and_duration_of_route(origin, destination, mode='driving'):
    gmaps = googlemaps.Client(api_key)
    route = gmaps.directions(origin, destination, mode=mode)
    return route[0]['legs'][0]['distance']['text']

I am looking for a way to incorporate this function into the existing constraints in optapy, so that I can match employees to shifts based on the shortest distance, or maybe a completely new solution by your suggestion.

I tried creating a constraint at constraints.py such as:

def closest_employee_to_shift(constraint_factory: ConstraintFactory):
    return constraint_factory.for_each(Shift).join(Employee,
                              Joiners.equal(lambda shift: shift.location,
                                            lambda employee: employee.address)
                            ) \
        .reward('Employee is too far from the shift', HardSoftScore.ONE_SOFT,
                  lambda shift, employee: find_distance_and_duration_of_route(shift, employee))

but unfortunately it didn't work.

Any suggestions on how to implement this would be greatly appreciated. Thank you.

Solution

I would precompute a distance matrix that contains distances from employee's addresses to shift location:

@optapy.problem_fact
class DistanceInfo:
    distance_matrix: dict
    
    def __init__(self, distance_matrix):
        self.distance_matrix = distance_matrix
    
    def get_distance(self, from_location, to_location):
        return self.distance_matrix[from_location][to_location]

distance_matrix = dict()
visited_locations = set()
for shift in shift_list:
    for employee in employee_list:
        if (shift.location, employee.address) not in visited_locations:
            if shift.location not in distance_matrix:
                distance_matrix[shift.location] = dict()
            if employee.address not in distance_matrix:
                distance_matrix[employee.address] = dict()
            distance_matrix[shift.location][employee.address] = find_distance_and_duration_of_route(shift.location, employee.address)
            distance_matrix[employee.address][shift.location] = find_distance_and_duration_of_route(employee.address, shift.location)
            visited_locations.add((shift.location, employee.address))
            visited_locations.add((employee.address, shift.location))

distance_info = DistanceInfo(distance_matrix)

and then you would use it in a constraint (after adding it to your @planning_solution like so:

def closest_employee_to_shift(constraint_factory: ConstraintFactory):
    return (
        constraint_factory.for_each(Shift)
        .join(DistanceInfo)
        .penalize('Employee distance to shift', HardSoftScore.ONE_SOFT,
                  lambda shift, distance_info: distance_info.get_distance(shift.employee.address, shift.location))
    )