Minor improvements for shift-assignment

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

.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",

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", {})

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"]