1

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.

fuchcar
  • 15
  • 4
  • One general comment: you don't want long-running operations inside of your constraints. Constraints need to be super-fast, and the more time you spend waiting for external services, the less likely it is you'll get a good solution out of the solver. If you must do this, at the very least cache those distances; even better, pre-compute them before the solver starts. – Lukáš Petrovický Jan 31 '23 at 10:00
  • Thank you for your suggestion. I'm working on it – fuchcar Feb 01 '23 at 13:32

1 Answers1

2

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))
    )
Christopher Chianelli
  • 1,163
  • 1
  • 8
  • 8
  • Hey, thank you for your reply. I was getting a key error when running your first block of code. I've changed the code and I've fixed the error if your goal for "distance_matrix" was to look something like this: distance_matrix = {'location1': {'location2': 10, 'location3': 20}, 'location2': {'location1': 10}, 'location3': {'location1': 20}} I'm working on the rest of your suggestions now. Will update you asap – fuchcar Feb 01 '23 at 13:31
  • Thanks; updated my answer to prevent the KeyError (forgot to initialize the rows of the matrix to empty) – Christopher Chianelli Feb 01 '23 at 19:32
  • I've made all necessary changes in all three files I've changed the @planning_solution to look like [this](https://i.imgur.com/DbER8wt.png) However I'm getting this error when visiting http://localhost:5000/static/index.html: org.optaplanner.optapy.org.optaplanner.optapy.OptaPyException: org.optaplanner.optapy.OptaPyException: A problem occurred when wrapping the python problem (). Maybe an annotation was passed an incorrect type (for example, @problem_fact_collection_property(str) on a function that return a list of int). – fuchcar Feb 02 '23 at 09:55
  • The issue is you used `@optapy.problem_fact_collection_property` for something that not a collection (`get_distance_info`). The fix is to use `@optapy.problem_fact_property` on `get_distance_info` instead. – Christopher Chianelli Feb 02 '23 at 18:15