ross/proc-rota
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

numberous fixes and improvements

Browse Source
This commit is contained in:
Ross
2025-06-22 09:38:03 +01:00
parent 06a551bb68
commit fc3102ec47
3 changed files with 333 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rota/shifts.py
+106 -24
View File
@@ -145,6 +145,7 @@ class SingleShift(BaseModel):
end_date: datetime.date | None = None
pair_proxy: bool = False
display_char: str | None = None # Add this line
force_assign_with: list[str] = [] # List of shift names to force assign with this shift on the same day
model_config = ConfigDict(
extra="allow",
@@ -1391,17 +1392,39 @@ class RotaBuilder(object):
)
logging.debug(f"Generate worker constraints: {worker.name}")
for week, day, shift in self.get_all_shiftclass_combinations():
# Only apply if this shift has force_assign_with set
if hasattr(shift, "force_assign_with") and shift.force_assign_with:
for other_shift_name in shift.force_assign_with:
# Only add constraint if the other shift is available on this day
if other_shift_name in self.get_shift_names_by_week_day(week, day):
for worker in self.get_workers_for_shift(shift):
self.model.constraints.add(
self.model.works[worker.id, week, day, shift.name]
<= self.model.works[worker.id, week, day, other_shift_name]
)
# Add hard shift dependencies for this worker
if hasattr(worker, "hard_day_dependencies"):
for from_day, to_day in worker.hard_day_dependencies.items():
for day_group in worker.hard_day_dependencies:
for week in self.weeks:
for shift in self.get_shift_names_by_week_day(week, from_day):
# Only apply if the shift also exists on the to_day
if shift in self.get_shift_names_by_week_day(week, to_day):
self.model.constraints.add(
self.model.works[worker.id, week, from_day, shift]
<= self.model.works[worker.id, week, to_day, shift]
)
assigned_any_shift = {}
for day in day_group:
shifts_today = self.get_shift_names_by_week_day(week, day)
var_name = f"hard_day_dep_{worker.id}_{week}_{day}"
if not hasattr(self.model, "hard_day_dep_vars"):
self.model.hard_day_dep_vars = {}
if (worker.id, week, day) not in self.model.hard_day_dep_vars:
self.model.hard_day_dep_vars[(worker.id, week, day)] = Var(within=Binary)
setattr(self.model, var_name, self.model.hard_day_dep_vars[(worker.id, week, day)])
assigned_any_shift[day] = self.model.hard_day_dep_vars[(worker.id, week, day)]
total_assigned = sum(self.model.works[worker.id, week, day, shift] for shift in shifts_today)
self.model.constraints.add(total_assigned >= assigned_any_shift[day])
self.model.constraints.add(total_assigned <= len(shifts_today) * assigned_any_shift[day])
# All-or-none: all days in the group must have the same assignment status
vals = list(assigned_any_shift.values())
for i in range(1, len(vals)):
self.model.constraints.add(vals[i] == vals[0])
# Add hard day exclusions for this worker
if hasattr(worker, "hard_day_exclusions"):
@@ -1448,6 +1471,17 @@ class RotaBuilder(object):
- len(allowed_set)
+ 1
)
shifts_today = self.get_shift_names_by_week_day(week, day)
for shift_name in shifts_today:
if hasattr(worker, "force_assign_with") and shift_name in worker.force_assign_with:
for other_shift in worker.force_assign_with[shift_name]:
if other_shift in shifts_today:
self.model.constraints.add(
self.model.works[worker.id, week, day, shift_name]
<= self.model.works[worker.id, week, day, other_shift]
)
if self.get_workers_who_require_locums():
for week, day, shift in self.get_all_shiftclass_combinations():
@@ -2215,17 +2249,20 @@ class RotaBuilder(object):
# for day in self.days[5:]
# for shift in self.get_shifts()) / 2)
# if self.constraint_options["balance_weekends"]:
self.model.constraints.add(
self.model.works_weekend[worker.id, week]
>= sum(
self.model.works[worker.id, week, day, shiftname]
for w, day, shiftname in self.get_all_shiftname_combinations(
week=week
)
if day in self.days[5:]
)
/ 2
)
# Try disabling tihs to allow more than 2 shifts ot be assigned on a weekend
#self.model.constraints.add(
# self.model.works_weekend[worker.id, week]
# >= sum(
# self.model.works[worker.id, week, day, shiftname]
# for w, day, shiftname in self.get_all_shiftname_combinations(
# week=week
# )
# if day in self.days[5:]
# )
# / 2
#)
self.model.constraints.add(
self.model.works_weekend[worker.id, week]
<= sum(
@@ -2492,11 +2529,24 @@ class RotaBuilder(object):
worker_allowed_sets = getattr(
worker, "allowed_multi_shift_sets", []
)
force_assign_sets = []
for shift_name in shifts_today:
if hasattr(worker, "force_assign_with") and shift_name in worker.force_assign_with:
force_set = set([shift_name] + list(worker.force_assign_with[shift_name]))
if force_set.issubset(shifts_today):
force_assign_sets.append(force_set)
shift_obj = self.get_shift_by_name(shift_name)
if getattr(shift_obj, "force_assign_with", []):
force_set = set([shift_name] + list(shift_obj.force_assign_with))
if force_set.issubset(shifts_today):
force_assign_sets.append(force_set)
# If no allowed sets, we can only assign one shift
allowed_sets_today = [
allowed_set
for allowed_set in worker_allowed_sets
if allowed_set.issubset(shifts_today)
]
] + force_assign_sets
if allowed_sets_today:
# Forbid any multi-shift assignment that includes shifts from different allowed sets
@@ -3182,6 +3232,24 @@ class RotaBuilder(object):
else:
s.end_date = self.rota_end_date
if hasattr(s, "force_assign_with") and s.force_assign_with:
missing = set(s.force_assign_with) - set(shift.name for shift in self.shifts)
if missing:
raise InvalidShift(
f"Shift '{s.name}' has force_assign_with referencing non-existent shift(s): {missing}"
)
all_shift_names = set(shift.name for shift in self.shifts)
for worker in self.workers:
if hasattr(worker, "force_assign_with"):
for shift_name, with_shifts in getattr(worker, "force_assign_with", {}).items():
missing = set([shift_name] + list(with_shifts)) - all_shift_names
if missing:
raise InvalidShift(
f"Worker '{worker.name}' has force_assign_with referencing non-existent shift(s): {missing}"
)
# Check worker requirements
if not isinstance(s.workers_required, int):
# Validate the worker requirements
@@ -4166,22 +4234,36 @@ class RotaBuilder(object):
return shifts
def get_worker_shift_list(self, worker: Worker, include_locums=False) -> List:
def get_worker_shift_list(self, worker: Worker, include_locums=False, search_multiple_assignments=False) -> List:
"""
This function returns a list of shifts assigned to a worker for each week and day.
It was originally designed to return a list of shift names (when only a single shift per day was supported), but it can also return a set of shift names if `search_multiple_assignments` is True.
"""
shifts = []
for week, day in self.get_week_day_combinations():
# d = self.start_date + datetime.timedelta(n)
# n = n + 1
temp = ""
if search_multiple_assignments:
temp = set()
else:
temp = ""
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0.90:
temp = shift
if search_multiple_assignments:
temp.add(shift)
else:
temp = shift
elif (
include_locums
and self.model.locum_works[worker.id, week, day, shift].value > 0.90
):
temp = shift
if search_multiple_assignments:
temp.add(shift)
else:
temp = shift
shifts.append(temp)
return shifts