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
|_________ 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.
