diff --git a/.github/workflows/python-lint.yml b/.github/workflows/python-lint.yml index cf78a2fa..c9581b24 100644 --- a/.github/workflows/python-lint.yml +++ b/.github/workflows/python-lint.yml @@ -24,7 +24,7 @@ jobs: - name: install dependencies run: | python -m pip install --upgrade pip - pip install ruff==0.1.6 + pip install ruff==0.6.4 - name: lint with ruff run: ruff check --output-format=github . diff --git a/.nextmv/golden/python-ortools-shiftassignment/inputs/input.json.golden b/.nextmv/golden/python-ortools-shiftassignment/inputs/input.json.golden index b9e031c6..cb884c10 100644 --- a/.nextmv/golden/python-ortools-shiftassignment/inputs/input.json.golden +++ b/.nextmv/golden/python-ortools-shiftassignment/inputs/input.json.golden @@ -19,13 +19,13 @@ }, { "end_time": "2023-11-20T14:00:00+02:00", - "shift_id": "normal_monday-early", + "shift_id": "welder_monday-early", "start_time": "2023-11-20T06:00:00+02:00", "worker_id": "Jacob Cunningham" }, { "end_time": "2023-11-20T14:00:00+02:00", - "shift_id": "welder_monday-early", + "shift_id": "normal_monday-early", "start_time": "2023-11-20T06:00:00+02:00", "worker_id": "Ray Heard" }, @@ -47,7 +47,7 @@ "result": { "custom": { "active_workers": 5, - "constraints": 89, + "constraints": 59, "mean_hours_worked": 8, "preferences_matched": 4, "status": "optimal", diff --git a/python-ortools-shiftassignment/main.py b/python-ortools-shiftassignment/main.py index dcfe2eb4..eba08de2 100644 --- a/python-ortools-shiftassignment/main.py +++ b/python-ortools-shiftassignment/main.py @@ -95,37 +95,41 @@ def solve(self, input: nextmv.Input) -> nextmv.Output: f"Shift_{s['id']}", ) - # Each worker must be assigned to at least their minimum number of shifts for e in workers: rules = rules_per_worker[e["id"]] - solver.Add( - solver.Sum([x_assign[(e["id"], s["id"])] for s in shifts]) >= rules["min_shifts"], - f"worker_{e['id']}", - ) + + # Each worker must be assigned to at least their minimum number of shifts + if "min_shifts" in rules: + solver.Add( + solver.Sum([x_assign[(e["id"], s["id"])] for s in shifts]) >= rules["min_shifts"], + f"worker_{e['id']}", + ) # Each worker must be assigned to at most their maximum number of shifts - solver.Add( - solver.Sum([x_assign[(e["id"], s["id"])] for s in shifts]) <= rules["max_shifts"], - f"worker_{e['id']}", - ) + if "max_shifts" in rules: + solver.Add( + solver.Sum([x_assign[(e["id"], s["id"])] for s in shifts]) <= rules["max_shifts"], + f"worker_{e['id']}", + ) # Ensure that the minimum rest time between shifts is respected - rest_time = datetime.timedelta(hours=rules_per_worker[e["id"]]["min_rest_hours_between_shifts"]) - for s1, shift1 in enumerate(shifts): - for s2, shift2 in enumerate(shifts): - if s1 >= s2: - continue - if ( - shift1["end_time"] + rest_time < shift2["start_time"] - or shift2["end_time"] + rest_time < shift1["start_time"] - ): - continue - # The two shifts are closer to each other than the minimum rest time, so we need to ensure that - # the worker is not assigned to both. - solver.Add( - x_assign[(e["id"], shift1["id"])] + x_assign[(e["id"], shift2["id"])] <= 1, - f"Rest_{e['id']}_{shift1['id']}_{shift2['id']}", - ) + if "min_rest_hours_between_shifts" in rules: + rest_time = datetime.timedelta(hours=rules["min_rest_hours_between_shifts"]) + for s1, shift1 in enumerate(shifts): + for s2, shift2 in enumerate(shifts): + if s1 >= s2: + continue + if ( + shift1["end_time"] + rest_time < shift2["start_time"] + or shift2["end_time"] + rest_time < shift1["start_time"] + ): + continue + # The two shifts are closer to each other than the minimum rest time, so we need to ensure that + # the worker is not assigned to both. + solver.Add( + x_assign[(e["id"], shift1["id"])] + x_assign[(e["id"], shift2["id"])] <= 1, + f"Rest_{e['id']}_{shift1['id']}_{shift2['id']}", + ) # Ensure that availabilities are respected for s in shifts: @@ -146,34 +150,34 @@ def solve(self, input: nextmv.Input) -> nextmv.Output: # The worker does not have the required qualification (worker cannot be assigned) x_assign[(e["id"], s["id"])].SetBounds(0, 0) - # Ensure that the minimum and maximum work hours per day are respected for day in range((latest_shift_end_time - earliest_shift_start_time).days + 1): + # Ensure that the minimum and maximum work hours per day are respected day_start = earliest_shift_start_time + datetime.timedelta(days=day) day_end = day_start + datetime.timedelta(days=1) - solver.Add( - solver.Sum( - [ - x_assign[(e["id"], s["id"])] - * ((min(s["end_time"], day_end) - max(s["start_time"], day_start)).total_seconds() / 3600) - for s in shifts - if s["start_time"] < day_end and s["end_time"] >= day_start - ] + if "max_work_hours_per_day" in rules: + solver.Add( + solver.Sum( + [ + x_assign[(e["id"], s["id"])] * overlap(s, day_start, day_end) + for s in shifts + if overlap(s, day_start, day_end) > 0 + ] + ) + <= rules["max_work_hours_per_day"], + f"MaxWorkHours_{e['id']}_{day}", ) - <= rules_per_worker[e["id"]]["max_work_hours_per_day"], - f"MaxWorkHours_{e['id']}_{day}", - ) - solver.Add( - solver.Sum( - [ - x_assign[(e["id"], s["id"])] - * ((min(s["end_time"], day_end) - max(s["start_time"], day_start)).total_seconds() / 3600) - for s in shifts - if s["start_time"] < day_end and s["end_time"] >= day_start - ] + if "min_work_hours_per_day" in rules: + solver.Add( + solver.Sum( + [ + x_assign[(e["id"], s["id"])] * overlap(s, day_start, day_end) + for s in shifts + if overlap(s, day_start, day_end) > 0 + ] + ) + >= rules["min_work_hours_per_day"], + f"MinWorkHours_{e['id']}_{day}", ) - >= rules_per_worker[e["id"]]["min_work_hours_per_day"], - f"MinWorkHours_{e['id']}_{day}", - ) # Ensure total hours worked by each worker are correctly calculated for e in workers: solver.Add( @@ -188,28 +192,34 @@ def solve(self, input: nextmv.Input) -> nextmv.Output: ) # Ensure that the maximum work hours per week are respected - for week in range((latest_shift_end_time - earliest_shift_start_time).days // 7 + 1): - week_start = earliest_shift_start_time + datetime.timedelta(weeks=week) - week_end = week_start + datetime.timedelta(days=7) - solver.Add( - solver.Sum( - [ - x_assign[(e["id"], s["id"])] - * ((min(s["end_time"], week_end) - max(s["start_time"], week_start)).total_seconds() / 3600) - for s in shifts - if s["start_time"] < week_end and s["end_time"] > week_start - ] + if "max_work_hours_per_week" in rules: + for week in range((latest_shift_end_time - earliest_shift_start_time).days // 7 + 1): + week_start = earliest_shift_start_time + datetime.timedelta(weeks=week) + week_end = week_start + datetime.timedelta(days=7) + solver.Add( + solver.Sum( + [ + x_assign[(e["id"], s["id"])] * overlap(s, week_start, week_end) + for s in shifts + if overlap(s, week_start, week_end) > 0 + ] + ) + <= rules["max_work_hours_per_week"], + f"MaxWorkHours_{e['id']}_Week{week}", ) - <= rules_per_worker[e["id"]]["max_work_hours_per_week"], - f"MaxWorkHours_{e['id']}_Week{week}", - ) + + # Calculate deviation from mean hours worked + balance_hours_weight = input.options.factor_balance_total_hours + if balance_hours_weight > 0: + avg_hours = solver.Sum([total_hours[e["id"]] for e in workers]) / len(workers) + for e in workers: + deviation = total_hours[e["id"]] - avg_hours + solver.Add(deviations[e["id"]] == deviation) # >>> Objective objective = solver.Objective() preference_weight = input.options.factor_maximize_preferences - balance_hours_weight = input.options.factor_balance_total_hours weekly_hours_weight = input.options.factor_maximize_weekly_hours_per_worker - avg_hours = solver.Sum([total_hours[e["id"]] for e in workers]) / len(workers) for e in workers: # Maximize preferences @@ -219,12 +229,12 @@ def solve(self, input: nextmv.Input) -> nextmv.Output: objective.SetCoefficient(x_assign[(e["id"], s["id"])], pref * preference_weight) # Minimize variance in total hours worked - deviation = total_hours[e["id"]] - avg_hours - solver.Add(deviations[e["id"]] == deviation) - objective.SetCoefficient(deviations[e["id"]], -balance_hours_weight) + if balance_hours_weight > 0: + objective.SetCoefficient(deviations[e["id"]], -balance_hours_weight) # Maximize total hours worked up to the maximum allowed - objective.SetCoefficient(total_hours[e["id"]], weekly_hours_weight) + if weekly_hours_weight > 0: + objective.SetCoefficient(total_hours[e["id"]], weekly_hours_weight) objective.SetMaximization() @@ -287,6 +297,11 @@ def solve(self, input: nextmv.Input) -> nextmv.Output: ) +def overlap(shift: dict, start: datetime.datetime, end: datetime.datetime) -> float: + """Calculates the overlap between a shift and a time interval. Returns the overlap in hours.""" + return max(0, (min(shift["end_time"], end) - max(shift["start_time"], start)).total_seconds() / 3600) + + def convert_input(input_data: dict[str, Any]) -> tuple[list, list, dict]: """Converts the input data to the format expected by the model.""" workers = input_data["workers"] @@ -301,14 +316,6 @@ def convert_input(input_data: dict[str, Any]) -> tuple[list, list, dict]: a["start_time"] = datetime.datetime.fromisoformat(a["start_time"]) a["end_time"] = datetime.datetime.fromisoformat(a["end_time"]) - # Add default values for rules - for r in input_data["rules"]: - r["min_shifts"] = r.get("min_shifts", 0) - r["max_shifts"] = r.get("max_shifts", 1000) - r["min_work_hours_per_day"] = r.get("min_work_hours_per_day", 0) - r["max_work_hours_per_day"] = r.get("max_work_hours_per_day", 24) - r["max_work_hours_per_week"] = r.get("max_work_hours_per_week", 24 * 7) - # Add default values for workers for e in workers: e["preferences"] = e.get("preferences", {}) diff --git a/ruff.toml b/ruff.toml index 25983e04..46fb29fd 100644 --- a/ruff.toml +++ b/ruff.toml @@ -1,7 +1,7 @@ target-version="py312" # Rule selection. -select = [ +lint.select = [ "E", # pycodestyle errors "W", # pycodestyle warnings "F", # pyflakes @@ -15,4 +15,11 @@ select = [ line-length = 120 # Allow generous cyclomatic complexity for examples. -mccabe.max-complexity = 25 +lint.mccabe.max-complexity = 25 + +[lint.per-file-ignores] +# Set a different max complexity for a specific file +"python-ortools-shiftassignment/main.py" = ["C901"] +"python-gurobi-knapsack/main.ipynb" = ["I001"] +"python-highs-knapsack/main.ipynb" = ["I001"] +"python-nextroute/main.ipynb" = ["I001"]