diff --git a/.gitignore b/.gitignore index b59c1b0..cd57986 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ __pycache__/ *.py[cod] *$py.class cache.db* +.venv diff --git a/app.py b/app.py index 9ec5e4a..6541eb1 100644 --- a/app.py +++ b/app.py @@ -13,6 +13,8 @@ # limitations under the License. from __future__ import annotations +import multiprocess +from dataclasses import asdict from typing import TYPE_CHECKING, Any import diskcache @@ -21,9 +23,16 @@ import employee_scheduling import utils -from app_configs import (APP_TITLE, DEBUG, LARGE_SCENARIO, MEDIUM_SCENARIO, MIN_MAX_EMPLOYEES, - SMALL_SCENARIO) +from app_configs import ( + APP_TITLE, + DEBUG, + LARGE_SCENARIO, + MEDIUM_SCENARIO, + MIN_MAX_EMPLOYEES, + SMALL_SCENARIO, +) from app_html import errors_list, set_html +from src.demo_enums import SolverType if TYPE_CHECKING: from pandas import DataFrame @@ -33,9 +42,8 @@ # Fix for Dash background callbacks crashing on macOS 10.13+ (https://bugs.python.org/issue33725) # See https://github.com/dwave-examples/flow-shop-scheduling/pull/4 for more details. -import multiprocess if multiprocess.get_start_method(allow_none=True) is None: - multiprocess.set_start_method('spawn') + multiprocess.set_start_method("spawn") app = Dash( __name__, @@ -47,7 +55,9 @@ @app.callback( - Output({"type": "to-collapse-class", "index": MATCH}, "className", allow_duplicate=True), + Output( + {"type": "to-collapse-class", "index": MATCH}, "className", allow_duplicate=True + ), inputs=[ Input({"type": "collapse-trigger", "index": MATCH}, "n_clicks"), State({"type": "to-collapse-class", "index": MATCH}, "className"), @@ -114,7 +124,10 @@ def set_scenario( shifts_per_employees, employees_per_shift, random_seed, - False, False, False, False + False, + False, + False, + False, ) @@ -129,7 +142,9 @@ def set_scenario( State("employees-per-shift-select", "tooltip"), ], ) -def update_employees_per_shift(value: int, tooltip: dict[str, Any]) -> tuple[int, dict, dict]: +def update_employees_per_shift( + value: int, tooltip: dict[str, Any] +) -> tuple[int, dict, dict]: """Update the employees-per-shift slider max if num-employees is changed.""" marks = { MIN_MAX_EMPLOYEES["min"]: str(MIN_MAX_EMPLOYEES["min"]), @@ -222,10 +237,7 @@ def custom_random_seed(value: int, scenario: int) -> int: Output("schedule-tab", "disabled", allow_duplicate=True), Output("tabs", "value"), Output({"type": "to-collapse-class", "index": 1}, "style", allow_duplicate=True), - inputs=[ - Input("num-employees-select", "value"), - Input("seed-select", "value") - ], + inputs=[Input("num-employees-select", "value"), Input("seed-select", "value")], ) def disp_initial_sched( num_employees: int, rand_seed: int @@ -269,10 +281,7 @@ def update_error_sidebar(run_click: int, prev_classes) -> tuple[dict, str]: if "collapsed" in classes: return no_update, no_update - return ( - {"display": "none"}, - prev_classes + " collapsed" - ) + return ({"display": "none"}, prev_classes + " collapsed") @app.callback( @@ -288,11 +297,20 @@ def update_error_sidebar(run_click: int, prev_classes) -> tuple[dict, str]: State("checklist-input", "value"), State("consecutive-shifts-select", "value"), State("availability-content", "children"), + State("solver-select", "value"), ], running=[ # show cancel button and hide run button, and disable and animate results tab - (Output("cancel-button", "style"), {"display": "inline-block"}, {"display": "none"}), - (Output("run-button", "style"), {"display": "none"}, {"display": "inline-block"}), + ( + Output("cancel-button", "style"), + {"display": "inline-block"}, + {"display": "none"}, + ), + ( + Output("run-button", "style"), + {"display": "none"}, + {"display": "inline-block"}, + ), # switch to schedule tab while running (Output("schedule-tab", "disabled"), False, False), (Output("tabs", "value"), "schedule-tab", "schedule-tab"), @@ -308,6 +326,7 @@ def run_optimization( checklist: list[int], consecutive_shifts: int, sched_df: DataFrame, + solver: int, ) -> tuple[DataFrame, bool, dict, list]: """Run a job on the hybrid solver when the run button is clicked.""" if run_click == 0 or ctx.triggered_id != "run-button": @@ -322,20 +341,35 @@ def run_optimization( isolated_days_allowed = True if 0 in checklist else False manager_required = True if 1 in checklist else False - cqm = employee_scheduling.build_cqm( - availability, - shifts, - *shifts_per_employee, - *employees_per_shift, - manager_required, - isolated_days_allowed, - consecutive_shifts + 1, + params = utils.ModelParams( + availability=availability, + shifts=shifts, + min_shifts=min(shifts_per_employee), + max_shifts=max(shifts_per_employee), + shift_min=min(employees_per_shift), + shift_max=max(employees_per_shift), + requires_manager=manager_required, + allow_isolated_days_off=isolated_days_allowed, + max_consecutive_shifts=consecutive_shifts ) - feasible_sampleset, errors = employee_scheduling.run_cqm(cqm) - sample = feasible_sampleset.first.sample + solver_type = SolverType(solver) + + if solver_type is SolverType.CQM: + cqm = employee_scheduling.build_cqm(**asdict(params)) + + feasible_sampleset, errors = employee_scheduling.run_cqm(cqm) + sample = feasible_sampleset.first.sample + + sched = utils.build_schedule_from_sample(sample, employees) + + elif solver_type is SolverType.NL: + model, assignments = employee_scheduling.build_nl(**asdict(params)) + errors = employee_scheduling.run_nl(model, assignments, **asdict(params)) + sched = utils.build_schedule_from_state(assignments.state(), employees, shifts) - sched = utils.build_schedule_from_sample(sample, employees) + else: + raise ValueError(f"Solver value `{solver_type.label}` is unhandled.") return ( utils.display_schedule(sched, availability), diff --git a/app_html.py b/app_html.py index 768eb48..f5ae9fd 100644 --- a/app_html.py +++ b/app_html.py @@ -19,6 +19,7 @@ from app_configs import (DESCRIPTION, EXAMPLE_SCENARIO, MAIN_HEADER, MAX_CONSECUTIVE_SHIFTS, MIN_MAX_EMPLOYEES, MIN_MAX_SHIFTS, NUM_EMPLOYEES, REQUESTED_SHIFT_ICON, THUMBNAIL, UNAVAILABLE_ICON) +from src.demo_enums import SolverType def description_card(): @@ -73,6 +74,29 @@ def range_slider(name: str, id: str, config: dict) -> html.Div: ) +def dropdown(label: str, id: str, options: list) -> html.Div: + """Dropdown element for option selection. + + Args: + label: The title that goes above the dropdown. + id: A unique selector for this element. + options: A list of dictionaries of labels and values. + """ + return html.Div( + className="dropdown-wrapper", + children=[ + html.Label(label), + dcc.Dropdown( + id=id, + options=options, + value=options[0]["value"], + clearable=False, + searchable=False, + ), + ], + ) + + def generate_control_card() -> html.Div: """Generates the control card for the dashboard. @@ -81,21 +105,20 @@ def generate_control_card() -> html.Div: model, and solver. """ example_scenario = [{"label": size, "value": i} for i, size in enumerate(EXAMPLE_SCENARIO)] + solver_options = [{"label": s.label, "value": s.value} for s in SolverType] return html.Div( id="control-card", children=[ - html.Div( - children=[ - html.Label("Scenario preset (sets sliders below)"), - dcc.Dropdown( - id="example-scenario-select", - options=example_scenario, - value=example_scenario[0]["value"], - clearable=False, - searchable=False, - ), - ] + dropdown( + "Solver", + "solver-select", + solver_options, + ), + dropdown( + "Scenario preset (sets sliders below)", + "example-scenario-select", + example_scenario ), # add sliders for employees and shifts slider( diff --git a/employee_scheduling.py b/employee_scheduling.py index 47691ff..7fbeab6 100644 --- a/employee_scheduling.py +++ b/employee_scheduling.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. from collections import defaultdict +from typing import Optional from dimod import ( Binary, @@ -19,28 +20,54 @@ ConstrainedQuadraticModel, quicksum, ) -from dwave.system import LeapHybridCQMSampler +from dwave.optimization.model import Model +from dwave.optimization.mathematical import add +from dwave.optimization.symbols import BinaryVariable +from dwave.system import LeapHybridCQMSampler, LeapHybridNLSampler -from utils import DAYS, SHIFTS +from utils import DAYS, SHIFTS, validate_nl_schedule -def build_cqm( - availability, - shifts, - min_shifts, - max_shifts, - shift_min, - shift_max, - requires_manager, - allow_isolated_days_off, - max_consecutive_shifts, +MSGS = { + "unavailable": ("Employees scheduled when unavailable", "{employee} on {day}"), + "overtime": ("Employees with scheduled overtime", "{employee}"), + "insufficient": ("Employees with not enough scheduled time", "{employee}"), + "understaffed": ("Understaffed shifts", "{day} is understaffed"), + "overstaffed": ("Overstaffed shifts", "{day} is overstaffed"), + "isolated": ("Isolated shifts", "{day} is an isolated day off for {employee}"), + "manager_issue": ("Shifts with no manager", "No manager scheduled on {day}"), + "too_many_consecutive": ( + "Employees with too many consecutive shifts", + "{employee} starting with {day}", + ), + "trainee_issue": ( + "Shifts with trainee scheduling issues", + "Trainee scheduling issue on {day}", + ), +} + + +def build_cqm(#params: ModelParams + availability: dict[str, list[int]], + shifts: list[str], + min_shifts: int, + max_shifts: int, + shift_min: int, + shift_max: int, + requires_manager: bool, + allow_isolated_days_off: bool, + max_consecutive_shifts: int, ): """Builds the ConstrainedQuadraticModel for the given scenario.""" cqm = ConstrainedQuadraticModel() employees = list(availability.keys()) # Create variables: one per employee per shift - x = {(employee, shift): Binary(employee + "_" + shift) for shift in shifts for employee in employees} + x = { + (employee, shift): Binary(employee + "_" + shift) + for shift in shifts + for employee in employees + } # OBJECTIVES: # Objective: give employees preferred schedules (val = 2) @@ -53,18 +80,17 @@ def build_cqm( # Objective: for infeasible solutions, focus on right number of shifts for employees num_s = (min_shifts + max_shifts) / 2 for employee in employees: - obj += ( - quicksum(x[employee, shift] for shift in shifts) - num_s - ) ** 2 + obj += (quicksum(x[employee, shift] for shift in shifts) - num_s) ** 2 cqm.set_objective(obj) - # CONSTRAINTS: # Only schedule employees when they're available for employee, schedule in availability.items(): for i, shift in enumerate(shifts): if schedule[i] == 0: - cqm.add_constraint(x[employee, shift] == 0, label=f"unavailable,{employee},{shift}") + cqm.add_constraint( + x[employee, shift] == 0, label=f"unavailable,{employee},{shift}" + ) for employee in employees: # Schedule employees for at most max_shifts @@ -119,11 +145,15 @@ def build_cqm( # Don't exceed max_consecutive_shifts for employee in employees: - for s in range(len(shifts) - max_consecutive_shifts + 1): + for s in range(len(shifts) - max_consecutive_shifts): cqm.add_constraint( quicksum( - [x[employee, shifts[s + i]] for i in range(max_consecutive_shifts)] - ) <= max_consecutive_shifts - 1, + [ + x[employee, shifts[s + i]] + for i in range(max_consecutive_shifts + 1) + ] + ) + <= max_consecutive_shifts, label=f"too_many_consecutive,{employee},{shifts[s]}", ) @@ -131,8 +161,7 @@ def build_cqm( trainees = [employee for employee in employees if employee[-2:] == "Tr"] for shift in shifts: cqm.add_constraint( - x[trainees[0], shift] - - x[trainees[0], shift] * x[trainees[0][:-3], shift] + x[trainees[0], shift] - x[trainees[0], shift] * x[trainees[0][:-3], shift] == 0, label=f"trainee_issue,,{shift}", ) @@ -158,49 +187,11 @@ def run_cqm(cqm): if s_vals == {0.0}: sampleset.first.sample[list(cqm.variables)[0]] = 1.0 - msgs = { - "unavailable": ( - "Employees scheduled when unavailable", - "{employee} on {day}" - ), - "overtime": ( - "Employees with scheduled overtime", - "{employee}" - ), - "insufficient": ( - "Employees with not enough scheduled time", - "{employee}" - ), - "understaffed": ( - "Understaffed shifts", - "{day} is understaffed" - ), - "overstaffed": ( - "Overstaffed shifts", - "{day} is overstaffed" - ), - "isolated": ( - "Isolated shifts", - "{day} is an isolated day off for {employee}" - ), - "manager_issue": ( - "Shifts with no manager", - "No manager scheduled on {day}" - ), - "too_many_consecutive": ( - "Employees with too many consecutive shifts", - "{employee} starting with {day}" - ), - "trainee_issue": ( - "Shifts with trainee scheduling issues", - "Trainee scheduling issue on {day}" - ), - } for i, _ in enumerate(sat_array): if not sat_array[i]: key, *data = sampleset.info["constraint_labels"][i].split(",") try: - heading, error_msg = msgs[key] + heading, error_msg = MSGS[key] except KeyError: # ignore any unknown constraint labels continue @@ -218,3 +209,183 @@ def run_cqm(cqm): return sampleset, errors return feasible_sampleset, None + + +def build_nl( + availability: dict[str, list[int]], + shifts: list[str], + min_shifts: int, + max_shifts: int, + shift_min: int, + shift_max: int, + requires_manager: bool, + allow_isolated_days_off: bool, + max_consecutive_shifts: int, +) -> tuple[Model, BinaryVariable]: + """Builds an employee scheduling nonlinear model. + + Args: + availability (dict[str, list[int]]): Employee availability. + shifts (list[str]): Shift labels. + min_shifts (int): Minimum shifts per employee. + max_shifts (int): Maximum shifts per employee. + shift_min (int): Minimum employees per shift. + shift_max (int): Maximum employees per shift. + requires_manager (bool): Whether to require exactly one manager on every shift. + allow_isolated_days_off (bool): Whether to allow isolated days off. + max_consecutive_shifts (int): Maximum consecutive shifts per employee. + + Returns: + tuple[Model, BinaryVariable]: the NL model and assignments decision variable + """ + # Create list of employees + employees = list(availability) + model = Model() + + # Create a binary symbol representing the assignment of employees to shifts + # i.e. assignments[employee][shift] = 1 if assigned, else 0 + num_employees = len(employees) + num_shifts = len(shifts) + assignments = model.binary((num_employees, num_shifts)) + + # Create availability constant + availability_list = [availability[employee] for employee in employees] + # Boost objective value of preferred shifts from 2 to 100 + for i, sublist in enumerate(availability_list): + availability_list[i] = [a if a != 2 else 100 for a in sublist] + availability_const = model.constant(availability_list) + + # Initialize model constants + min_shifts_constant = model.constant(min_shifts) + max_shifts_constant = model.constant(max_shifts) + shift_min_constant = model.constant(shift_min) + shift_max_constant = model.constant(shift_max) + max_consecutive_shifts_c = model.constant(max_consecutive_shifts) + one_c = model.constant(1) + + # OBJECTIVES: + # Objective: give employees preferred schedules (val = 2) + obj = (assignments * availability_const).sum() + + # Objective: for infeasible solutions, focus on right number of shifts for employees + target_shifts = model.constant((min_shifts + max_shifts) / 2) + shift_difference_list = [ + (assignments[e, :].sum() - target_shifts) ** 2 for e in range(num_employees) + ] + obj += add(*shift_difference_list) + + model.minimize(-obj) + + # CONSTRAINTS: + # Only schedule employees when they're available + model.add_constraint((availability_const >= assignments).all()) + + for e in range(len(employees)): + # Schedule employees for at most max_shifts + model.add_constraint(assignments[e, :].sum() <= max_shifts_constant) + + # Schedule employees for at least min_shifts + model.add_constraint(assignments[e, :].sum() >= min_shifts_constant) + + # Every shift needs shift_min and shift_max employees working + for s in range(num_shifts): + model.add_constraint(assignments[:, s].sum() <= shift_max_constant) + model.add_constraint(assignments[:, s].sum() >= shift_min_constant) + + managers_c = model.constant( + [employees.index(e) for e in employees if e[-3:] == "Mgr"] + ) + trainees_c = model.constant( + [employees.index(e) for e in employees if e[-2:] == "Tr"] + ) + + if not allow_isolated_days_off: + negthree_c = model.constant(-3) + zero_c = model.constant(0) + # Adding many small constraints greatly improves feasibility + for e in range(len(employees)): + for s1 in range(len(shifts) - 2): + s2, s3 = s1 + 1, s1 + 2 + model.add_constraint( + negthree_c * assignments[e, s2] + + assignments[e][s1] * assignments[e][s2] + + assignments[e][s1] * assignments[e][s3] + + assignments[e][s2] * assignments[e][s3] + <= zero_c + ) + + if requires_manager: + for shift in range(len(shifts)): + model.add_constraint(assignments[managers_c][:, shift].sum() == one_c) + + # Don't exceed max_consecutive_shifts + for e in range(num_employees): + for s in range(num_shifts - max_consecutive_shifts + 1): + s_window = s + max_consecutive_shifts + 1 + model.add_constraint( + assignments[e][s : s_window + 1].sum() <= max_consecutive_shifts_c + ) + + # Trainee must work on shifts with trainer + trainers = [] + for i in trainees_c.state(): + trainer_name = employees[int(i)][:-3] + trainers.append(employees.index(trainer_name)) + trainers_c = model.constant(trainers) + + model.add_constraint((assignments[trainees_c] <= assignments[trainers_c]).all()) + + return model, assignments + + +def run_nl( + model: Model, + assignments: BinaryVariable, + availability: dict[str, list[int]], + shifts: list[str], + min_shifts: int, + max_shifts: int, + shift_min: int, + shift_max: int, + requires_manager: bool, + allow_isolated_days_off: bool, + max_consecutive_shifts: int, + time_limit: int | None = None, + msgs: dict[str, tuple[str, str]] = MSGS, +) -> Optional[defaultdict[str, list[str]]]: + """Solves the NL scheduling model and detects any errors. + + Args: + model (Model): NL model to solve + assignments (BinaryVariable): decision variable for employee shifts + time_limit (int | None, optional): time limit for sampling. Defaults to None. + """ + if not model.is_locked(): + model.lock() + + sampler = LeapHybridNLSampler() + sampler.sample(model, time_limit=time_limit) + errors = validate_nl_schedule( + assignments, + msgs, + availability, + shifts, + min_shifts, + max_shifts, + shift_min, + shift_max, + requires_manager, + allow_isolated_days_off, + max_consecutive_shifts, + ) + + # Return errors if any error message list is populated + for error_list in errors.values(): + if len(error_list) > 0: + return errors + + return None + + +if __name__ == "__main__": + ... diff --git a/requirements.txt b/requirements.txt index 4144089..7c7987d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ dash[diskcache]==2.16.1 dash-bootstrap-components==1.6.0 -dwave-ocean-sdk>=6.7 +dwave-ocean-sdk>=7.0.0 Faker==21.0.0 pandas>=2.0 +dwave-optimization>=0.3.0 \ No newline at end of file diff --git a/src/demo_enums.py b/src/demo_enums.py new file mode 100644 index 0000000..f3f3c86 --- /dev/null +++ b/src/demo_enums.py @@ -0,0 +1,29 @@ +# Copyright 2024 D-Wave +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from enum import Enum + + +class SolverType(Enum): + """Enum class representing the solver types used in the demo.""" + + CQM = 0 # Default value for application dropdown + NL = 1 + + @property + def label(self): + return { + SolverType.CQM: "CQM", + SolverType.NL: "Nonlinear", + }[self] diff --git a/tests/test_inputs.py b/tests/test_inputs.py index 2ade57e..36ce825 100644 --- a/tests/test_inputs.py +++ b/tests/test_inputs.py @@ -12,8 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. import unittest +from dataclasses import asdict from dash import dash_table +from numpy import asarray import employee_scheduling import utils @@ -21,30 +23,44 @@ class TestDemo(unittest.TestCase): + # Initialize shared data for tests + def setUp(self): + self.num_employees = 12 + self.sched_df = disp_initial_sched(self.num_employees, None)[0].data + self.shifts =list(self.sched_df[0].keys()) + self.shifts.remove("Employee") + self.availability = utils.availability_to_dict(self.sched_df) + self.test_params = utils.ModelParams( + availability=self.availability, + shifts=self.shifts, + min_shifts=1, + max_shifts=6, + shift_min=5, + shift_max=16, + requires_manager=True, + allow_isolated_days_off=False, + max_consecutive_shifts=6 + ) + # Check that initial schedule created is the right size def test_initial_sched(self): - num_employees = 12 - - sched_df = disp_initial_sched(num_employees, None)[0].data - - self.assertEqual(len(sched_df), num_employees) + self.assertEqual(len(self.sched_df), self.num_employees) # Check that CQM created has the right number of variables def test_cqm(self): - num_employees = 12 + cqm = employee_scheduling.build_cqm(**asdict(self.test_params)) - sched_df = disp_initial_sched(num_employees, None)[0].data - shifts = list(sched_df[0].keys()) - shifts.remove("Employee") - availability = utils.availability_to_dict(sched_df) + self.assertEqual(len(cqm.variables), + self.num_employees * len(self.shifts)) - cqm = employee_scheduling.build_cqm( - availability, shifts, 1, 6, 5, 16, True, False, 6 - ) + # Check that NL assignments variable is the correct shape + def test_nl(self): + _, assignments = employee_scheduling.build_nl(**asdict(self.test_params)) - self.assertEqual(len(cqm.variables), num_employees * len(shifts)) + self.assertEqual(assignments.shape(), + (self.num_employees, len(self.shifts))) - def test_samples(self): + def test_samples_cqm(self): shifts = [str(i + 1) for i in range(5)] # Make every employee available for every shift @@ -57,10 +73,20 @@ def test_samples(self): "E-Tr": [1] * 5, } - cqm = employee_scheduling.build_cqm( - availability, shifts, 1, 6, 5, 16, True, False, 6 + test_params = utils.ModelParams( + availability=availability, + shifts=shifts, + min_shifts=1, + max_shifts=6, + shift_min=5, + shift_max=16, + requires_manager=True, + allow_isolated_days_off=False, + max_consecutive_shifts=6 ) + cqm = employee_scheduling.build_cqm(**asdict(test_params)) + feasible_sample = { "A-Mgr_1": 0.0, "A-Mgr_2": 0.0, @@ -130,6 +156,69 @@ def test_samples(self): self.assertTrue(cqm.check_feasible(feasible_sample)) self.assertFalse(cqm.check_feasible(infeasible_sample)) + def test_states_nl(self): + shifts = [str(i + 1) for i in range(5)] + + # Make every employee available for every shift + availability = { + "A-Mgr": [1] * 5, + "B-Mgr": [1] * 5, + "C": [1] * 5, + "D": [1] * 5, + "E": [1] * 5, + "E-Tr": [1] * 5, + } + + test_params = utils.ModelParams( + availability=availability, + shifts=shifts, + min_shifts=1, + max_shifts=6, + shift_min=5, + shift_max=16, + requires_manager=True, + allow_isolated_days_off=False, + max_consecutive_shifts=6 + ) + + model, assignments = employee_scheduling.build_nl(**asdict(test_params)) + + if not model.is_locked(): + model.lock() + + # Resize model states + model.states.resize(2) + + feasible_state = [ + [0, 0, 1, 1, 1], # A-Mgr + [1, 1, 0, 0, 0], # B-Mgr + [1, 1, 1, 1, 1], # C + [1, 1, 1, 1, 1], # D + [1, 1, 1, 1, 1], # E + [1, 1, 1, 1, 1] # E-Tr + ] + + # Infeasible: multiple managers scheduled for the same shift + infeasible_state = [ + [1, 1, 1, 1, 1], # A-Mgr + [1, 1, 1, 1, 1], # B-Mgr + [1, 1, 1, 1, 1], # C + [1, 1, 1, 1, 1], # D + [1, 1, 1, 1, 1], # E + [1, 1, 1, 1, 1] # E-Tr + ] + + # Assign feasible state to index 0 + assignments.set_state(0, feasible_state) + # Assign infeasible state to index 1 + assignments.set_state(1, infeasible_state) + + constraints_0 = [int(c.state(0)) for c in model.iter_constraints()] + constraints_1 = [int(c.state(1)) for c in model.iter_constraints()] + + self.assertEqual(len(constraints_0), sum(constraints_0)) + self.assertGreater(len(constraints_1), sum(constraints_1)) + def test_build_from_sample(self): employees = ["A-Mgr", "B-Mgr", "C", "D", "E", "E-Tr"] @@ -183,3 +272,34 @@ def test_build_from_sample(self): # This should verify we don't have any issues in the object created for display from a sample self.assertEqual(type(disp_datatable), dash_table.DataTable) + + def test_build_from_state(self): + employees = ["A-Mgr", "B-Mgr", "C", "D", "E", "E-Tr"] + shifts = [str(i+1) for i in range(14)] + + # Make every employee available for every shift + availability = { + "A-Mgr": [1] * 14, + "B-Mgr": [1] * 14, + "C": [1] * 14, + "D": [1] * 14, + "E": [1] * 14, + "E-Tr": [1] * 14, + } + + state = asarray([ + # Give managers alternating shifts + [i % 2 for i in range(14)], + [(i+1) % 2 for i in range(14)], + [1 for _ in range(14)], + [1 for _ in range(14)], + [1 for _ in range(14)], + [1 for _ in range(14)], + ]) + + disp_datatable = utils.display_schedule( + utils.build_schedule_from_state(state, employees, shifts), + availability + ) + + self.assertIsInstance(disp_datatable, dash_table.DataTable) \ No newline at end of file diff --git a/utils.py b/utils.py index 09b8da7..b72d823 100644 --- a/utils.py +++ b/utils.py @@ -15,12 +15,15 @@ import datetime import random import string +from collections import defaultdict +from dataclasses import dataclass from app_configs import REQUESTED_SHIFT_ICON, UNAVAILABLE_ICON import numpy as np import pandas as pd from dash import dash_table from faker import Faker +from dwave.optimization.symbols import BinaryVariable NOW = datetime.datetime.now() SCHEDULE_LENGTH = 14 @@ -35,6 +38,44 @@ DAYS = ["Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat"] WEEKEND_IDS = ["1", "7", "8", "14"] +@dataclass +class ModelParams: + """Convenience class for defining and passing model parameters. + + Attributes: + availability (dict[str, list[int]]): Employee availability for each shift, + structured as follows: + ``` + availability = { + 'Employee Name': [ + 0, # 0 if unavailable for shift at index i + 1, # 1 if available for shift at index i + 2, # 2 if shift at index i is preferred + ... + ] + } + ``` + shifts (list[str]): List of shift labels. + min_shifts (int): Min shifts per employee. + max_shifts (int): Max shifts per employee. + shift_min (int): Min employees per shift. + shift_max (int): Max employees per shift. + requires_manager (bool): Whether a manager is required on every shift. + allow_isolated_days_off (bool): Whether isolated shifts off are allowed + (pattern of on-off-on). + max_consecutive_shifts (int): Max consecutive shifts for each employee. + """ + + availability: dict[str, list[int]] + shifts: list[str] + min_shifts: int + max_shifts: int + shift_min: int + shift_max: int + requires_manager: bool + allow_isolated_days_off: bool + max_consecutive_shifts: int + def get_random_string(length): """Generate a random string of a given length.""" @@ -86,8 +127,8 @@ def build_random_sched(num_employees, rand_seed=None): data.insert(0, "Employee", employees) - data[COL_IDS[0]].replace(UNAVAILABLE_ICON, " ", inplace=True) - data[COL_IDS[-1]].replace(UNAVAILABLE_ICON, " ", inplace=True) + data.replace({COL_IDS[0]: {UNAVAILABLE_ICON, " "}}) + data.replace({COL_IDS[-1]: {UNAVAILABLE_ICON, " "}}) data.loc[data.Employee == employees[-1], data.columns[1:]] = " " @@ -108,6 +149,22 @@ def build_schedule_from_sample(sample, employees): return data + +def build_schedule_from_state(state: np.ndarray, employees: list[str], shifts: list[str]): + """Builds a schedule from the state of a BinaryVariable.""" + data = pd.DataFrame(columns=COL_IDS) + data.insert(0, "Employee", employees) + + for e, employee in enumerate(employees): + for s, shift in enumerate(shifts): + if state[e, s] == 1.0: + data.loc[data["Employee"] == employee, shift] = " " + else: + data.loc[data["Employee"] == employee, shift] = UNAVAILABLE_ICON + + return data + + def get_cols(): """Gets information for column headers, including months and days.""" start_month = START_DATE.strftime("%B %Y") # Get month and year @@ -121,6 +178,7 @@ def get_cols(): ]} for i, c in enumerate(SHIFTS)] ) + def get_cell_styling(cols): """Sets conditional cell styling.""" return [ @@ -253,3 +311,238 @@ def availability_to_dict(availability_list): ] return availability_dict + + +def validate_nl_schedule( + assignments: BinaryVariable, + msgs: dict[str, tuple[str, str]], + availability: dict[str, list[int]], + shifts: list[str], + min_shifts: int, + max_shifts: int, + shift_min: int, + shift_max: int, + requires_manager: bool, + allow_isolated_days_off: bool, + max_consecutive_shifts: int, +) -> defaultdict[str, list[str]]: + """Detect any errors in a solved NL scheduling model. + + Since NL models do not currently support constraint labels, this function + is required to detect any constraint violations and format them for display + in the user interface. + + Args: + assignments (BinaryVariable): Assignments generated from sampling the + NL model. + params (ModelParams): Model parameters. + msgs (dict[str, tuple[str, str]]): Error message template dictionary. + Must be formatted like: + ``` + msgs = { + 'error_type': ('Error message', 'message template with {replacement fields}'), + ... + } + ``` + + Raises: + ValueError: Raised if the `msgs` dictionary doesn't contain the required keys. + + Returns: + errors (defaultdict[str, list[str]]): Error descriptions and messages. + """ + # Required keys to match existing error messages in employee_scheduling.py + required_msg_keys = [ + "unavailable", + "overtime", + "insufficient", + "understaffed", + "overstaffed", + "isolated", + "manager_issue", + "too_many_consecutive", + "trainee_issue", + ] + for key in required_msg_keys: + if key not in msgs: + raise ValueError(f"`msgs` dictionary missing required key `{key}`") + + # Pull solution state as ndarray, employees as list + result = assignments.state() + employees = list(availability.keys()) + + # Convert shifts to day/date labels + shift_labels = [f"{DAYS[i%7]} {SHIFTS[i]}" for i in range(len(shifts))] + + errors = defaultdict(list) + + _validate_availability(result, availability, employees, shift_labels, errors, msgs) + _validate_shifts_per_employee(result, employees, min_shifts, max_shifts, errors, msgs) + _validate_employees_per_shift(result, shift_min, shift_max, shift_labels, errors, msgs) + _validate_max_consecutive_shifts(result, max_consecutive_shifts, employees, shift_labels, errors, msgs) + _validate_trainee_shifts(result, employees, shift_labels, errors, msgs) + if requires_manager: + _validate_requires_manager(result, employees, shift_labels, errors, msgs) + if not allow_isolated_days_off: + _validate_isolated_days_off(result, employees, shift_labels, errors, msgs) + + return errors + + +def _validate_availability( + results: np.ndarray, + availability: dict[str, list[int]], + employees: list[str], + shift_labels: list[int], + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates employee availability for the solution and updates the `errors` + dictionary with any errors found. Requires the `msgs` dict to have the + key `'unavailable'`.""" + msg_key, msg_template = msgs["unavailable"] + for e, employee in enumerate(employees): + for s, day in enumerate(shift_labels): + if results[e, s] > availability[employee][s]: + errors[msg_key].append( + msg_template.format(employee=employee, day=day) + ) + return errors + + +def _validate_shifts_per_employee( + results: np.ndarray, + employees: list[str], + min_shifts: int, + max_shifts: int, + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates the number of shifts per employee for the solution and updates + the `errors` dictionary with any errors found. Requires the `msgs` dict + to have the keys `'insufficient'` and `'overtime'`.""" + insufficient_key, insufficient_template = msgs["insufficient"] + overtime_key, overtime_template = msgs["overtime"] + for e, employee in enumerate(employees): + num_shifts = results[e, :].sum() + if num_shifts < min_shifts: + errors[insufficient_key].append( + insufficient_template.format(employee=employee) + ) + elif num_shifts > max_shifts: + errors[overtime_key].append(overtime_template.format(employee=employee)) + return errors + + +def _validate_employees_per_shift( + results: np.ndarray, + shift_min: int, + shift_max: int, + shift_labels: list[int], + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates the number of employees per shift for the solution and updates + the `errors` dictionary with any errors found. Requires the `msgs` dict + to have the keys `'understaffed'` and `'overstaffed'`.""" + understaffed_key, understaffed_template = msgs["understaffed"] + overstaffed_key, overstaffed_template = msgs["overstaffed"] + + for s, day in enumerate(shift_labels): + num_employees = results[:, s].sum() + if num_employees < shift_min: + errors[understaffed_key].append(understaffed_template.format(day=day)) + elif num_employees > shift_max: + errors[overstaffed_key].append(overstaffed_template.format(day=day)) + return errors + + +def _validate_requires_manager( + results: np.ndarray, + employees: list[str], + shift_labels: list[str], + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates the number of managers per shift for the solution and updates + the `errors` dictionary with any errors found. Requires the `msgs` dict + to have the key `'manager_issue'`.""" + key, template = msgs["manager_issue"] + employee_arr = np.asarray( + [employees.index(e) for e in employees if e[-3:] == "Mgr"] + ) + managers_per_shift = results[employee_arr].sum(axis=0) + for shift, num_managers in enumerate(managers_per_shift): + if num_managers == 0: + errors[key].append(template.format(day=shift_labels[shift])) + return errors + + +def _validate_isolated_days_off( + results: np.ndarray, + employees: list[str], + shift_labels: list[str], + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates the number of managers per shift for the solution and updates + the `errors` dictionary with any errors found. Requires the `msgs` dict + to have the key `'isolated'`.""" + key, template = msgs["isolated"] + isolated_pattern = np.array([1, 0, 1]) + for e, employee in enumerate(employees): + shift_triples = [results[e, i : i + 3] for i in range(results.shape[1] - 2)] + for s, shift_set in enumerate(shift_triples): + if np.equal(shift_set, isolated_pattern).all(): + day = shift_labels[s + 1] + errors[key].append(template.format(employee=employee, day=day)) + return errors + + +def _validate_max_consecutive_shifts( + results: np.ndarray, + max_consecutive_shifts: int, + employees: list[str], + shift_labels: list[str], + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates the max number of consecutive shifts for the solution and updates + the `errors` dictionary with any errors found. Requires the `msgs` dict + to have the key `'too_many_consecutive'`.""" + key, template = msgs["too_many_consecutive"] + for e, employee in enumerate(employees): + consecutive_shift_arrays = ( + [results[e, i : i + max_consecutive_shifts] + for i in range(results.shape[1] - max_consecutive_shifts)] + ) + for shift, shift_arr in enumerate(consecutive_shift_arrays): + if shift_arr.sum() > max_consecutive_shifts: + errors[key].append( + template.format(employee=employee, day=shift_labels[shift]) + ) + break + return errors + + +def _validate_trainee_shifts( + results: np.ndarray, + employees: list[str], + shift_labels: list[str], + errors: defaultdict[str, list[str]], + msgs: dict[str, tuple[str, str]], +) -> defaultdict[str, list[str]]: + """Validates that trainees are on-shift with their manager for the solution + and updates the `errors` dictionary with any errors found. Requires the + `msgs` dict to have the key `'trainee_issue'`.""" + key, template = msgs["trainee_issue"] + trainees = {employees.index(e): e for e in employees if e[-2:] == "Tr"} + trainers = { + employees.index(e): e for e in employees if e + "-Tr" in trainees.values() + } + for (trainee_i), (trainer_i) in zip(trainees.keys(), trainers.keys()): + same_shifts = np.less_equal(results[trainee_i], results[trainer_i]) + for i, s in enumerate(same_shifts): + if not s: + errors[key].append(template.format(day=shift_labels[i])) + return errors