Staff Scheduler: Design/Algorithm questions (Python)

Question

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:
 st.write("Please Pick A valid TIme")
 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.

Answer 1

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.

• RootTwo: Given the type hint breakdown: Dict[TimePeriod, List[Staff]], I don't think set(breakdown.values()) will work, because List[Staff] isn't hashable.
• Reinderien: @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!