README.md

Schedule a variable number of people into a variable number slots for multiple time steps.

Maximize for diversity of slots per person.

Not optimal

scheduling_people.py

#!/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)