# Staff Scheduler: Design/Algorithm questions (Python)

## Background

I worked at a community center where everyday one of my coworkers would spend 20ish minutes trying to organize the day's schedule. The facility had three rooms: The Great Hall, The Club House, and The Sub Cellar. Each of these rooms were open at different times in the day, and when he was organizing the schedule, he didn't want too many staff in there at once. For some rooms, he allowed more staff than others. Every day, we had different people working. Each person had a different shift. Some people came at 9am, while other people came at 1pm to cover the evening shifts. How long each person's shift was for that day determined the length of the break they were given that day in order to grab lunch or dinner.

## Repo

Scheduler

scheduler
|_________ src
|          |________ scheduler
|          |        |________ __init__.py
|          |        |________ __main__.py
|          |
|          |_______ _scheduler
|                   |________ __init__.py
|                   |________ work.py
|                   |________ timemodule.py
|                   |________ manager.py
|_________testing
|
|_________setup.py


manager.py

""" This module contains manager classes that are responsible for
assigning staff to rooms based on the required hard conditions"""
from collections import defaultdict
from copy import deepcopy
from typing import List, Dict

from _scheduler.work import Room, Staff, RType, EType, Shift
from _scheduler.timemodule import TimePeriod

class Manager():
def __init__(self, staff: List):
self.staff = staff

def manage(self):
raise NotImplementedError()

class RoomManager(Manager):
def __init__(self, room: Room, staff: List):
super().__init__(staff)
self.room = room

def manage(self) -> (List[TimePeriod], List[List[Staff]]):
available_staff = []
staff = self._get_available_staff(self.staff)
while(True):
if self._is_enough_coverage(staff):
breakdown = self._get_breakdown(staff)
result = self._verify_breakdown(breakdown, len(staff))
if result:
return self.get_possible_shifts(breakdown)
else:
staff = self._remove_extra_staff(breakdown)
else:
return {}

def _get_available_staff(self, staff: List):
""" Given a list of staff, this checks to see which
ones are available """
avail_staff = []
for s in staff:
if s._is_coincides(self.room):
avail_staff.append(s)
return avail_staff

def _get_breakdown(self, staff: List) -> Dict[TimePeriod, List[Staff]]:
room_schedule = defaultdict(list)
avail_staff = self._get_available_staff(staff)
num_of_staff = len(avail_staff)
split_times = self.room.time_open._split(num_of_staff)
for time in split_times:
for staff in avail_staff:
if staff._is_available(time):
room_schedule[time].append(staff)
return room_schedule

def _verify_breakdown(self,
breakdown: Dict[TimePeriod, List[Staff]],
expected: int) -> bool:
valid_staff = set()
for s in breakdown.values():
valid_staff = valid_staff.union(set(s))
return len(valid_staff) == expected

def _remove_extra_staff(self, breakdown) -> List[Staff]:
valid_staff = set()
for s in breakdown.values():
valid_staff = valid_staff.union(set(s))
return list(valid_staff)

def _is_enough_coverage(self, staff: List) -> bool:
""" Given a list of staff, this checks that their combined
times cover the room's time"""
room_time = set(self.room.time_open.comp)
total_coverage = set()
for s in staff:
total_coverage  = total_coverage.union(s.shift.comp)
return room_time.issubset(total_coverage)

def _find_valid_path(self, time_list: List,
curr_list: List, i: int,
valid_path: List) -> None:
if i >= len(time_list):
valid_path.append(curr_list)
return
staff_list = list(time_list.values())
staff_list = staff_list[i]
for staff in staff_list:
if staff not in curr_list:
new_list = deepcopy(curr_list)
new_list.append(staff)
self._find_valid_path(time_list, new_list, i + 1, valid_path)
else:
continue
return

def get_possible_shifts(self, time_list: List
)-> (List[TimePeriod], List[List[Staff]]):
possible_schedules = []
self._find_valid_path(time_list, [], 0, possible_schedules)
times = list(time_list.keys())
return times, possible_schedules

class BreakManager(Manager):
def __init__(self, staff: List):
super().__init__(staff)

def manage(self):
pass



work.py

from enum import Enum, auto
from datetime import datetime
from typing import Dict, Any

from _scheduler.timemodule import TimePeriod

class EType(Enum):
COUNSELOR = auto()
FRONT_DESK = auto()

class RType(Enum):
GH = auto()
SC = auto()
CH = auto()

class Shift(TimePeriod):
def __init__(self, st: int, et: int):
super().__init__(st, et)
hours = self.dur.seconds // 3600
if hours > 5:
self.break_length = 1
else:
self.break_length = .5

class Staff:
def __init__(self, name: str, emp_type: EType, st: int = None,
et: int = None, shift: Shift = None,):
if shift:
self.shift = shift
else:
self.shift = Shift(st, et)
self.name = name
self.emp_type = emp_type

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

def __repr__(self):
return f'Staff("{self.name}", {self.emp_type}, Shift={self.shift})'

def __eq__(self, other):
if isinstance(other, self.__class__):
return self.name == other.name
return False

def __hash__(self):
return hash(self.name)

def _get_possible_break_periods(self):
emp_shift = self.shift
break_length = emp_shift.break_length
shifts = []
i = emp_shift.st + break_length
while i <= emp_shift.et:
shifts.append(Shift(i-break_length, i))
i += .5
return shifts

def _is_coincides(self, shift: Any) -> bool:
""" This function determins whether the staff object's
shift happens within the same time as another TimePeriod
returns true if it does, and false if it doesn't."""
if type(shift) == Staff:
shift = shift.shift
elif type(shift) == Room:
shift = shift.time_open
coincides = self.shift._coincides(shift)
return coincides

def _is_available(self, shift: Any) -> bool:
""" This function determins whether the staff object's
shift contains the entire period. If it does, then the staff
is available"""
if type(shift) == Staff:
shift = shift.shift
elif type(shift) == Room:
shift = shift.time_open
is_available = self.shift._contains(shift)
return is_available

class Room:
def __init__(self, name: RType):
room_info = self._room_assignment(name)
self.max_cap = room_info["max_cap"]
self.name = name
self.time_open = room_info["time_open"]

def _room_assignment(self, name: RType) -> Dict[str, Any]:
room_info = {}
times = [datetime(1, 1, 1, 9, 0),
datetime(1, 1, 1, 21, 0),
datetime(1, 1, 1, 14, 30, 0)]

if name == RType.CH:
room_info["max_cap"] = 2
room_info["time_open"] = TimePeriod(times[0], times[2])
elif name == RType.GH:
room_info["max_cap"] = 3
room_info["time_open"] = TimePeriod(times[0], times[1])
elif name == RType.SC:
room_info["max_cap"] = 1
room_info["time_open"] = TimePeriod(times[0], times[2])
return room_info



timemodule.py

from typing import List
from datetime import datetime, timedelta

import scheduler

class TimePeriod:
"""
This class represents a time period between two points in time.
The smallest unit of time in this representation is 30mins, and
each time period is composed of 30 minute intervals.
---------------------------------------------------------------
++++++++++++++++++++++++ ARGS +++++++++++++++++++++++++++++++++
---------------------------------------------------------------
(int) st: Start Time
(int) et: End Time
"""
num = 0

def __init__(self, st: datetime, et: datetime):
if et <= st:
raise scheduler.TimeError(
"End time needs to be later than start time.")
self.st = st  # datetime
self.et = et  # datetime
self.dur = et - st  # timedelta in seconds
self.comp = self._get_composition(self.dur)
self._id = self.update(1)

def __eq__(self, other):
"""
Allows one to check equality with instances
>>> start = datetime(1,1,1,1,30)
>>> end = datetime(1,1,1,4,30)
>>> TimePeriod(start, end) == TimePeriod(start, end)
True
"""
if isinstance(other, self.__class__):
return str(self) == str(other)
return False

def __str__(self):
return f'{self.st.strftime("%I:%M %p")} - {self.et.strftime("%I:%M %p")}'

def __repr__(self):
return f'{self.__class__}({self.st}, {self.et})'

def __hash__(self):
return hash(self._id)

def _split(self, part: int) -> List:
""" Split uses the partition argument to split the TimePeriod into
equal parts by blocks of .5 """
if part > len(self.comp):
raise BaseException("Cannot divide time segment into that many parts")

split_time = []
part_size = len(self.comp) // part

for i in range(part):
if i == (part - 1):
split_time.append(TimePeriod(self.comp[i * part_size],
self.comp[-1]))
else:
split_time.append(TimePeriod(self.comp[i * part_size],
self.comp[(i+1) * part_size]))
return split_time

def _contains(self, other_tp):
if self.st <= other_tp.st and self.et >= other_tp.et:
return True
return False

def _coincides(self, t2):
composition1 = set(self.comp)
composition2 = set(t2.comp)
in_common = composition1 & composition2

return bool(in_common)

def _get_composition(self, duration: timedelta) -> int:
""" It splits the duration into 30 minute segments and creates/returns a list
of the 30 minute segments the TimePeriod is composed from"""
hours = duration.seconds // 3600
mins = duration.seconds - (hours * 3600)
quant = hours * 2
quant = quant + 1 if int(mins) > 0 else quant
comp = [self.st + i * timedelta(minutes=30) for i in range(quant + 1)]
return comp

@classmethod
def update(cls, value):
cls.num += value
return cls.num


driver.py

import streamlit as st
from _scheduler.work import Room, Staff, EType, RType
from _scheduler.manager import RoomManager
import graphviz as graphviz
import datetime as dt
from datetime import datetime, date, timedelta

import scheduler

def get_num_of_staff():
num_of_staff = st.text_input("How many staff do you want?", "0")
num_of_staff = int(num_of_staff)

return num_of_staff

def setup_times():
base_date = date(1, 1, 1)
start_time = dt.time(9, 0)
start_time = datetime.combine(base_date, start_time)
avail_times = [start_time + (i * timedelta(minutes=30)) for i in range(25)]

return avail_times

def create_staff_list(num_of_staff, avail_times):
staff_list = []
for i in range(num_of_staff):
name = st.text_input("* Enter the Staff's name",
str(i*num_of_staff))

start_time = st.selectbox(
f"Please Choose a Starting Time for {name}",
avail_times,
index=i * num_of_staff + 1,
format_func=lambda x: str(x.strftime("%I:%M %p")))

end_time = st.selectbox(
f"Please Choose an Ending Time for {name}",
avail_times,
index=i * num_of_staff + 2,
format_func=lambda x: str(x.strftime("%I:%M %p")))
try:
staff_list.append(Staff(name,
EType.COUNSELOR,
st=start_time,
et=end_time))
except scheduler.TimeError:
return None
return staff_list

def setup_room_and_manager(staff_list):
club_house = Room(RType.CH)  # room
chmanager = RoomManager(club_house, staff_list)
return chmanager

def draw_graph(times, order):
graph = graphviz.Digraph()
colorx = .000
for current in order:
final_color = f'{colorx} .999 .400'
for i, v in enumerate(current):
if i == len(current) - 1:
continue
time = str(times[i]).replace(":", " ")
time2 = str(times[i+1]).replace(":", " ")
node1 = v.name + " " + time
node2 = current[i+1].name + " " + time2
graph.edge(node1, node2, color=final_color)
colorx += .070
st.graphviz_chart(graph)

def get_schedule():
times, order = [], []
try:
times, order = manager.manage()
except Exception:
st.write("Not A valid Schedule")
return times, order

if __name__ == '__main__':
st.title("Break Scheduler")
number_of_staff = get_num_of_staff()
if number_of_staff > 0:
time_choices = setup_times()
staff_list = create_staff_list(number_of_staff, time_choices)

manager = setup_room_and_manager(staff_list)
times, order = get_schedule()
if len(times) > 0:
draw_graph(times, order)
else:
st.write("""
Please get more coverage. Can't make schedule from current shifts
""")
else:
st.write("Please begin filling out the information above")


## Design

I'd love it if I could get advice and feedback on my current design. I've broken down the problem and created classes for Staff, Rooms, Shifts. I have a TimePeriod class that has a start and end time and some other attributes that allow for splitting a time period into multiple components that together add up to the original TimePeriod. It's made a little easy because for this program, the smallest unit of time is 30 mins, and the front end portion only provides users to select hours in the half hour interval. 9am, 9:30am, ..., 8:30pm, 9pm.

I have a Manager class that's in charge of creating the schedules, and my current manager class takes a Room and a list of Staff and provides possible combinations in which those staff can cover that room only for staff that are available to work at the time the room is open.

My front end runs on streamline and asks how many staff and for each staff collects their shift. It then, if possible with the given staff shifts, returns a graph with the possible combinations that they can cover the club house which is open from 9AM-2:30PM.

## Goals

I eventually want to be able to provide a more generalized front end. I'd like to give the user the ability to create any room and as many rooms as they need. I'd like to have an algorithm that can place the staff in each of those rooms.

I'd also like for it to figure out when the best time for staff to take breaks. I have a function in the Staff class that creates a list of possible break times within that staff's shift. I have an algorithm in mind that ranks each of the times in that list with the times in the being ranked higher, so that when it looks at all staff, it gives them breaks that don't overlap but are as close to the middle of their shift as possible. All while making sure that all rooms are covered by a staff member.

For my specific purpose I only need three rooms, two of which have the same time, but I'd like this to be very generalized so that anyone could use it for their workplace.

## Questions

My questions have kind of been scattered throughout the text above and so I'll consolidate them here so that it's easier to refer to.

• Is the current design good for the goals that I have in mind, if not what can I change to make my goals easier to accomplish

• What algorithms should I use when it comes to scheduling the rooms

• Am I overlooking anything?

• Where else can I look for help on this? I'm only one person and can't imagine this is something I get done alone in a short amount of time.

## Type hints

You're using them. Great! As for this one:

def __init__(self, staff: List):


What is staff a list of? If you know, specify it as List[Thing]. If you do not know, leave it as list.

## Parens

We're not in Java/C++/etc., so this

    while(True):


does not need parentheses.

## Order of conditions

I find this:

        if self._is_enough_coverage(staff):
breakdown = self._get_breakdown(staff)
result = self._verify_breakdown(breakdown, len(staff))
if result:
return self.get_possible_shifts(breakdown)
else:
staff = self._remove_extra_staff(breakdown)
else:
return {}


would be more legible as

if not self._is_enough_coverage(staff):
return {}

breakdown = self._get_breakdown(staff)
result = self._verify_breakdown(breakdown, len(staff))
if result:
return self.get_possible_shifts(breakdown)

staff = self._remove_extra_staff(breakdown)


## Generators

Some of your functions can be simplified with yield:

    avail_staff = []
for s in staff:
if s._is_coincides(self.room):
avail_staff.append(s)
return avail_staff


can be

for s in staff:
if s._is_coincides(self.room):
yield s


though in this case, you can condense this further:

return (s for s in staff if s._is_coincides(self.room))


A grammar nitpick: "is coincides" does not make sense; use either "coincides" or "does coincide".

## Set simplification

    valid_staff = set()
for s in breakdown.values():
valid_staff = valid_staff.union(set(s))


can be

valid_staff = set(itertools.chain.from_iterable(breakdown.values()))


This pattern appears a few times.

## Redundant return

At the end of _find_valid_path.

## Do not do your own time math

Here.

    hours = self.dur.seconds // 3600


The way that the Python built-in recommends:

from datetime import timedelta
# ...

hours = self.dur / timedelta(hours=1)


self.dur is already a timedelta. break_length should also be.

## No-op format

f'{self.name}' should just be self.name.

## Combined predicates

    if isinstance(other, self.__class__):
return self.name == other.name
return False


should be

return isinstance(other, self.__class__) and self.name == other.name


## Stringy representation

Why does _room_assignment return a dict? You already have a strong class structure. You should make a class with members max_cap and time_open and return an instance of this.

## Overlap algorithm

    composition1 = set(self.comp)
composition2 = set(t2.comp)
in_common = composition1 & composition2


is a bad idea. Re-think this in terms of the start and end times of the two objects. I will leave this as an exercise to you.

## Stringy times

draw_graph, first of all, is missing type hints - but even without them I can tell that times is some sequence of strings. It should not be, nor should you be doing string manipulation on formatted times. Instead, pass them as actual time objects, and format them as appropriate.

• Given the type hint breakdown: Dict[TimePeriod, List[Staff]], I don't think set(breakdown.values()) will work, because List[Staff] isn't hashable. Sep 28, 2022 at 19:16
• @RootTwo You're right to point out that that expression has problems, but hashability isn't really the problem. If the inner values were tuples the result would still be wrong. The actual fix is to flatten the values collection before passing it to the set constructor. Thanks! Sep 28, 2022 at 23:22