Schedule a variable number of people into a variable number slots for multiple time steps.
Maximize for diversity of slots per person.
Not optimal
#!/usr/bin/env python3 | |
import copy # https://docs.python.org/3/library/copy.html | |
import random | |
""" | |
This script focuses on three variables: | |
* (discrete) time | |
* slots | |
* people | |
The purpose of this script is to rotate people across slots | |
and maximize the diversity of each person's experience over time. | |
A person should not be in the same slot on consecutive times. | |
Two equivilant ways of viewing the relation are | |
provided in table 1 and table 2. | |
+--------+------------+-----------+---------+ | |
| | time_1 | time_2 | time_3 | | |
+--------+------------+-----------+---------+ | |
| slot_A | person_1 | person_3 | | | |
| | person_2 | person_4 | | | |
+--------+------------+-----------+---------+ | |
| slot_B | person_3 | person_1 | | | |
+--------+------------+-----------+---------+ | |
| slot_C | person_4 | person_2 | | | |
| | person_5 | | | | |
|--------+------------+-----------+---------+ | |
| slot_D | N/A | person_5 | | | |
+--------+------------+-----------+---------+ | |
| slot_E | N/A | N/A | | | |
+--------+------------+-----------+---------+ | |
Table 1: slot versus time -- schedule of which slot is address by whom | |
source: https://asciiflow.com | |
Multiple people can be assigned one slot. | |
The number of slots and people is not one-to-one, | |
and the number of slots varies over time, | |
and the number of people varies over time. | |
In table 1, a person should not be in the same row for | |
two consecutive columns. | |
Also, every slot should be occupied by one or more people for | |
a given time (unless that slot is explicitly "N/A"). | |
The people per slot should be uniformly distributed as much as feasible. | |
+---------+--------+--------+--------+--------+--------+ | |
| | time_1 | time_2 | time_3 | time_4 | time_5 | | |
+---------+--------+--------+--------+--------+--------+ | |
| person_1| slot_A | slot_B | | | | | |
+---------+--------+--------+--------+--------+--------+ | |
| person_2| slot_A | slot_C | | | | | |
+---------+--------+--------+--------+--------+--------+ | |
| person_3| slot_B | slot_A | | | | | |
+---------+--------+--------+--------+--------+--------+ | |
| person_4| slot_C | slot_A | | | | | |
+---------+--------+--------+--------+--------+--------+ | |
| person_5| slot_C | slot_D | | | | | |
+---------+--------+--------+--------+--------+--------+ | |
| person_6| N/A | N/A | | | | | |
+---------+--------+--------+--------+--------+--------+ | |
Table 2: person versus time -- schedule of who is covering which slot | |
source: https://asciiflow.com | |
Some people may not be present for each time. | |
constraint: Each person who is present for a given time | |
gets assigned one task. | |
""" | |
def print_slots_to_be_filled(slots_available_per_time: dict) -> None: | |
""" | |
Produces an ASCII table that looks like | |
slots to be filled | |
| t1 | t2 | t3 | | |
A | ___ | ___ | ___ | | |
B | ___ | ___ | ___ | | |
C | ___ | ___ | ___ | | |
D | N/A | ___ | ___ | | |
E | N/A | N/A | ___ | | |
""" | |
complete_list_of_slots = [] | |
for this_time, list_of_slots in slots_available_per_time.items(): | |
complete_list_of_slots+= list_of_slots | |
complete_list_of_slots = list(set(complete_list_of_slots)) | |
complete_list_of_slots.sort() | |
print(" slots to be filled") | |
str_to_print = " |" | |
for this_time in slots_available_per_time.keys(): | |
str_to_print+=" " | |
str_to_print+=this_time | |
str_to_print+=" |" | |
print(str_to_print) | |
for this_slot in complete_list_of_slots: | |
str_to_print=" " + this_slot + " |" | |
for this_time, list_of_slots in slots_available_per_time.items(): | |
if this_slot in list_of_slots: | |
str_to_print+=" ___ |" | |
else: | |
str_to_print+=" N/A |" | |
print(str_to_print) | |
return | |
# def print_people_to_be_slotted(people_available_per_time: dict) -> None: | |
# """ | |
# """ | |
# return | |
def print_schedule(person_allocations:dict) -> None: | |
""" | |
Produces an ASCII table that looks like | |
schedule per person | |
| t1 | t2 | | |
p1 | A | B | | |
p2 | A | A | | |
p3 | B | B | | |
p4 | C | C | | |
p5 | C | C | | |
""" | |
#print("[Trace] def print_schedule") | |
complete_list_of_people = [] | |
for this_time, people_slot_dict in person_allocations.items(): | |
complete_list_of_people += list(people_slot_dict.keys()) | |
complete_list_of_people = list(set(complete_list_of_people)) | |
complete_list_of_people.sort() | |
print("\n schedule per person") | |
str_to_print = " |" | |
for this_time in person_allocations.keys(): | |
str_to_print+=" " | |
str_to_print+=this_time | |
str_to_print+=" |" | |
print(str_to_print) | |
for person in complete_list_of_people: | |
str_to_print = " " | |
str_to_print += person | |
str_to_print += " |" | |
for this_time, people_slot_dict in person_allocations.items(): | |
if person in people_slot_dict.keys(): | |
str_to_print += " " | |
str_to_print += people_slot_dict[person] | |
str_to_print += " |" | |
else: | |
str_to_print += " N/A |" | |
print(str_to_print) | |
return | |
def create_candidate_schedule(slots_available_per_time:dict, | |
people_available_per_time:dict, | |
new_schedule:dict, | |
this_time:str) -> dict: | |
""" | |
for an unscheduled time (this_time), assign slots to people. | |
""" | |
#print("[Trace] def create_candidate_schedule") | |
new_schedule[this_time] = {} | |
# Create a random list of slots that is uniformly distributed | |
# and matches the number of people who need slots at this time | |
slots_to_assign = [] | |
#print("len(people_available_per_time[this_time])=",len(people_available_per_time[this_time])) | |
while len(slots_to_assign)<len(people_available_per_time[this_time]): | |
slots_to_pick_from = random.sample( slots_available_per_time[this_time], | |
len(slots_available_per_time[this_time])) | |
#print('slots_to_pick_from=',slots_to_pick_from) | |
while ((len(slots_to_assign) < len(people_available_per_time[this_time])) | |
and (len(slots_to_pick_from)>0)): | |
#print('slots_to_assign=',slots_to_assign,'; slots_to_pick_from=',slots_to_pick_from) | |
slots_to_assign.append(slots_to_pick_from.pop()) | |
#print("slots_to_assign=",slots_to_assign) | |
# assign each person the slot in the new time | |
for index, this_person in enumerate(people_available_per_time[this_time]): | |
new_schedule[this_time][this_person] = slots_to_assign[index] | |
#print_schedule(new_schedule) | |
# check whether any person has two consecutive slots for this_time and previous_time | |
index_of_this_time = list(slots_available_per_time.keys()).index(this_time) | |
previous_time = list(slots_available_per_time.keys())[index_of_this_time-1] | |
#print("previous_time =",previous_time,"currently",this_time) | |
for person, slot in new_schedule[this_time].items(): | |
try: | |
#print("old:",person,new_schedule[previous_time][person],"new:",new_schedule[this_time][person]) | |
if new_schedule[previous_time][person]==new_schedule[this_time][person]: | |
raise Exception("repetative slot for person") | |
except KeyError: | |
#print(person,"was not present") | |
pass | |
return new_schedule | |
def score_schedule(new_schedule: dict,this_time: str) -> int: | |
""" | |
""" | |
#print("[Trace] def score_schedule") | |
index_of_this_time = list(new_schedule.keys()).index(this_time) | |
previous_time = list(new_schedule.keys())[index_of_this_time-1] | |
#print("previous_time =",previous_time,"currently",this_time) | |
complete_list_of_people = [] | |
for a_time, people_and_slot in new_schedule.items(): | |
complete_list_of_people += list(people_and_slot.keys()) | |
complete_list_of_people = list(set(complete_list_of_people)) | |
#print("complete_list_of_people=",complete_list_of_people) | |
score = 0 | |
for person in complete_list_of_people: | |
list_of_slots = [] | |
for a_time, people_and_slot in new_schedule.items(): | |
try: | |
list_of_slots.append(new_schedule[a_time][person]) | |
except KeyError: | |
list_of_slots.append(" ") | |
#print(person,this_time,list_of_slots) | |
# maximize diversity of rotations | |
# TODO: the following includes " " in the scoring, which may skew results | |
for index in range(2,len(list_of_slots)+1): | |
#print('comparing',list_of_slots[-1],list_of_slots[-1*index]) | |
if list_of_slots[-1]!=list_of_slots[-1*index]: | |
score+=1 | |
#print("added 1") | |
return score | |
def find_schedule(slots_available_per_time:dict, | |
people_available_per_time:dict, | |
old_schedule:dict, | |
this_time:str, | |
number_of_candidate_schedules_per_time: int, | |
maximum_number_of_trials_per_time: int) -> dict: | |
""" | |
for an unscheduled time (this_time), assign slots to people. | |
""" | |
#print("[Trace] def find_schedule") | |
number_of_tries = 0 | |
max_score = 0 | |
list_of_scores = [] | |
list_of_schedules = [] | |
while ((len(list_of_schedules)<number_of_candidate_schedules_per_time) | |
and (number_of_tries<maximum_number_of_trials_per_time)): | |
number_of_tries+=1 | |
try: | |
new_schedule = create_candidate_schedule(slots_available_per_time, | |
people_available_per_time, | |
old_schedule, | |
this_time) | |
score = score_schedule(new_schedule,this_time) | |
list_of_scores.append(score) | |
if score > max_score: | |
max_score = score | |
#print('score=',score) | |
#print_schedule(new_schedule) | |
list_of_schedules.append(new_schedule) | |
except Exception: | |
pass | |
#print("after creating multiple schedules, the scores are",list_of_scores) | |
highest_scoring_schedule = list_of_schedules[list_of_scores.index(max_score)] | |
#print_schedule(highest_scoring_schedule) | |
return highest_scoring_schedule | |
if __name__ == "__main__": | |
number_of_candidate_schedules_per_time = 10 | |
maximum_number_of_trials_per_time = 1000 | |
# To represent the slots available to be filled per time, | |
slots_available_per_time = {"t1": ["A", "B", "C"], | |
"t2": ["A", "B", "C", "D"], | |
"t3": ["A", "B", "C", "D", "E"], | |
"t4": ["A", "B", "C", "D"]} | |
# the ASCII table is just the transpose of the dict | |
print_slots_to_be_filled(slots_available_per_time) | |
# specify which people are available per time, | |
people_available_per_time = {"t1": ["p1", "p2", "p3", "p4", "p5"], | |
"t2": ["p1", "p2", "p3", "p4", "p5"], | |
"t3": ["p1", "p2", "p3", "p4", "p5", "p6"], | |
"t4": ["p1", "p3", "p4", "p6"]} | |
# creating an ASCII table of this isn't that interesting because it's just the transpose of the dict | |
#print_people_to_be_slotted(people_available_per_time) | |
# verify that the times are the same | |
time_diff = set(slots_available_per_time.keys()) - set(people_available_per_time.keys()) | |
#print(time_diff) | |
assert(time_diff==set()) | |
# what happened in the past? | |
historical_allocations = {"t1": {"p1":"A", "p2": "A", "p3":"B", "p4":"C", "p5":"C"}, | |
"t2": {"p1":"B", "p2": "C", "p3":"A", "p4":"B", "p5":"A"}} | |
print_schedule(historical_allocations) | |
times_to_be_scheduled = set(people_available_per_time.keys()) - set(historical_allocations.keys()) | |
times_to_be_scheduled = list(times_to_be_scheduled) | |
times_to_be_scheduled.sort() | |
print('times to schedule:', times_to_be_scheduled) | |
new_schedule = copy.deepcopy(historical_allocations) # https://docs.python.org/3/library/copy.html | |
for this_time in times_to_be_scheduled: | |
new_schedule = find_schedule(slots_available_per_time, | |
people_available_per_time, | |
new_schedule, | |
this_time, | |
number_of_candidate_schedules_per_time, | |
maximum_number_of_trials_per_time) | |
print("new schedule:") | |
print_schedule(new_schedule) |