4
\$\begingroup\$

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.

\$\endgroup\$

1 Answer 1

3
\$\begingroup\$

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.

\$\endgroup\$
2
  • \$\begingroup\$ Given the type hint breakdown: Dict[TimePeriod, List[Staff]], I don't think set(breakdown.values()) will work, because List[Staff] isn't hashable. \$\endgroup\$
    – RootTwo
    Sep 28, 2022 at 19:16
  • \$\begingroup\$ @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! \$\endgroup\$
    – Reinderien
    Sep 28, 2022 at 23:22

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.