Refactor shift constraints in tests to use new constraint classes

- Updated test cases in `test_workers.py` to replace dictionary-based constraints with instances of `PreShiftConstraint`, `PostShiftConstraint`, and `MaxShiftsPerWeekConstraint`.
- Added a new test `test_worker_assign_as_block_preference2` to verify worker preferences for block assignments.
- Marked tests as slow where appropriate to improve test suite performance.
This commit is contained in:
Ross
2025-08-11 13:03:08 +01:00
parent d7bcfce78c
commit 76f3b48c1d
12 changed files with 790 additions and 987 deletions
+215 -126
View File
@@ -4,7 +4,7 @@ import itertools
from typing import Dict, Iterable, List, Sequence, Tuple, Set, Any, Type, Literal, Optional
import time
from pydantic import BaseModel, ConfigDict, Extra, constr, model_validator, StrictInt, Field
from pydantic import BaseModel, ConfigDict, Extra, constr, field_validator, model_validator, StrictInt, Field
from pathlib import Path
@@ -59,18 +59,74 @@ SHIFT_BOUNDS = {
"weekend_count": (0, 60),
}
VALID_SHIFT_CONSTRAINTS = Literal[
"night",
"pre",
"post",
"balance_across_groups",
"limit_grade_number",
"minimum_grade_number",
"max_shifts_per_week",
"max_shifts_per_week_block",
"require_remote_site_presence",
"require_remote_site_presence_week",
]
#VALID_SHIFT_CONSTRAINTS = Literal[
# "night",
# "pre",
# "post",
# "balance_across_groups",
# "limit_grade_number",
# "minimum_grade_number",
# "max_shifts_per_week",
# "max_shifts_per_week_block",
# "require_remote_site_presence",
# "require_remote_site_presence_week",
#]
class BaseShiftConstraint(BaseModel):
weeks: Optional[List[int]] = None
start_date: Optional[datetime.date] = None
end_date: Optional[datetime.date] = None
model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
@field_validator("weeks")
def weeks_none_if_dates_set(cls, weeks, values):
data = values.data
if (data.get("start_date") is not None or data.get("end_date") is not None) and weeks is not None:
raise ValueError("If start_date or end_date is set, weeks must not be set")
return weeks
#class ShiftConstraint(BaseShiftConstraint):
# name: str
# options: Any = None
class NightConstraint(BaseShiftConstraint):
options: Any = None
class PreShiftConstraint(BaseShiftConstraint):
days: Any = None
class PostShiftConstraint(BaseShiftConstraint):
days: Any = None
class BalanceAcrossGroupsConstraint(BaseShiftConstraint):
""""""
class MinSummedGradeByShiftsPerDayConstraint(BaseShiftConstraint):
shifts: List[str] = Field(..., description="List of shift names to sum grades over")
min_grade_sum: int = Field(..., description="Minimum total grade required")
days: Optional[List[str]] = None
class LimitGradeNumberConstraint(BaseShiftConstraint):
grade: int
max_number: int
class MinimumGradeNumberConstraint(BaseShiftConstraint):
grade: int
min_number: int
class RequireRemoteSitePresenceConstraint(BaseShiftConstraint):
site: str
required_number: int
class MaxShiftsPerWeekConstraint(BaseShiftConstraint):
max_shifts: int | None = None
class MaxShiftsPerWeekBlockConstraint(BaseShiftConstraint):
week_block : int
max_shifts: int | None = None
def _pydantic_to_dict(obj):
if isinstance(obj, list):
@@ -80,17 +136,6 @@ def _pydantic_to_dict(obj):
return obj
class ShiftConstraint(BaseModel):
name: VALID_SHIFT_CONSTRAINTS
options: bool | int | Dict | tuple = False
class MinSummedGradeByShiftsPerDayConstraint(BaseModel):
shifts: List[str] = Field(..., description="List of shift names to sum grades over")
min_grade_sum: int = Field(..., description="Minimum total grade required")
weeks: Optional[List[int]] = None # If None, applies to all weeks
days: Optional[List[str]] = None # If None, applies to all days
class WorkerRequirement(BaseModel):
"""
start date is inclusive, end date is not
@@ -153,7 +198,9 @@ class SingleShift(BaseModel):
force_as_block_unless_nwd: bool = False
hard_constrain_shift: bool = True
bank_holidays_only: bool = False
constraint: list[ShiftConstraint] = []
#constraint: list[ShiftConstraint] = []
constraints: List[BaseModel] = Field(default_factory=list)
start_date: datetime.date | None = None
end_date: datetime.date | None = None
pair_proxy: bool = False
@@ -166,6 +213,11 @@ class SingleShift(BaseModel):
)
def __init__(self, **data: Any):
if "constraint" in data:
raise ValueError(
"The 'constraint' field on SingleShift has been deprecated and removed. "
"Please use the 'constraints' field with Pydantic constraint models instead."
)
super().__init__(**data)
# self.site = sites
# self.name = name
@@ -187,13 +239,13 @@ class SingleShift(BaseModel):
# self.assign_as_block = assign_as_block
self.constraint_options = {}
self.constraints = []
for c in self.constraint:
self.constraints.append(c.name)
#self.constraint_options = {}
#self.constraints = []
#for c in self.constraint:
# self.constraints.append(c.name)
if c.options:
self.constraint_options[c.name] = c.options
# if c.options:
# self.constraint_options[c.name] = c.options
# constraint = c.constraint
# match c:
@@ -206,7 +258,34 @@ class SingleShift(BaseModel):
self.days = [self.days]
def __str__(self):
return f"name: {self.name}, site: {self.sites}, length: {self.length}, balance weighting: {self.balance_weighting}, balance offset: {self.balance_offset}, hard_constrain: {self.hard_constrain_shift}, workers required : {self.workers_required}, rota on nwds {self.rota_on_nwds}, assign as block: {self.assign_as_block}, force as block: {self.force_as_block}, constraints: {self.constraints}, constraint_options: {self.constraint_options}" # , total shifts: {self.total_shifts}"
return (
f"name: {self.name}, "
f"sites: {self.sites}, "
f"length: {self.length}, "
f"balance_weighting: {self.balance_weighting}, "
f"balance_offset: {self.balance_offset}, "
f"hard_constrain_shift: {self.hard_constrain_shift}, "
f"workers_required: {self.workers_required}, "
f"rota_on_nwds: {self.rota_on_nwds}, "
f"assign_as_block: {self.assign_as_block}, "
f"force_as_block: {self.force_as_block}, "
f"constraints: {self.constraints}"
)
def has_constraint(self, constraint: Type) -> bool:
for c in self.constraints:
if isinstance(c, constraint):
return True
return False
def get_constraint(self, constraint: Type) -> BaseShiftConstraint | None:
for c in self.constraints:
if isinstance(c, constraint):
return c
return None
def get_constraints(self, constraint: Type) -> list[BaseShiftConstraint]:
return [c for c in self.constraints if isinstance(c, constraint)]
def get_shift_summary(self):
"""
@@ -641,7 +720,7 @@ class RotaBuilder(object):
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
for shift in self.get_shifts_with_constraint(NightConstraint)
),
# within=Binary,
within=NonNegativeIntegers,
@@ -653,7 +732,7 @@ class RotaBuilder(object):
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
for shift in self.get_shifts_with_constraint(NightConstraint)
),
domain=NonNegativeIntegers,
initialize=0,
@@ -663,7 +742,7 @@ class RotaBuilder(object):
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
for shift in self.get_shifts_with_constraint(NightConstraint)
),
domain=NonNegativeIntegers,
initialize=0,
@@ -675,7 +754,7 @@ class RotaBuilder(object):
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("balance_across_groups")
for shift in self.get_shifts_with_constraint(BalanceAcrossGroupsConstraint)
),
# within=Binary,
within=NonNegativeIntegers,
@@ -687,7 +766,7 @@ class RotaBuilder(object):
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("balance_across_groups")
for shift in self.get_shifts_with_constraint(BalanceAcrossGroupsConstraint)
),
domain=NonNegativeIntegers,
initialize=0,
@@ -697,7 +776,7 @@ class RotaBuilder(object):
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("balance_across_groups")
for shift in self.get_shifts_with_constraint(BalanceAcrossGroupsConstraint)
),
domain=NonNegativeIntegers,
initialize=0,
@@ -1161,12 +1240,10 @@ class RotaBuilder(object):
<= limit
)
for shift in self.get_shifts_with_constraint("limit_grade_number"):
if not shift.constraint_options["limit_grade_number"]:
raise ValueError(
"Constraint option must be defined for 'limit_grade_number'"
)
for grade in shift.constraint_options["limit_grade_number"]:
for shift in self.get_shifts_with_constraint(LimitGradeNumberConstraint):
for constraint in shift.get_constraints(LimitGradeNumberConstraint):
grade = constraint.grade
limit = constraint.max_number
# self.model.limit_grades_constraint = Constraint(
setattr(
self.model,
@@ -1176,7 +1253,7 @@ class RotaBuilder(object):
# [shift for shift in self.get_shifts_with_constraint("limit_grade_number")],
[shift.name],
[grade],
[shift.constraint_options["limit_grade_number"][grade]],
[limit],
rule=limitGradesRule,
# name=f"limit_grades_constraint_{shift.name}_{grade}"
),
@@ -1194,12 +1271,10 @@ class RotaBuilder(object):
>= limit
)
for shift in self.get_shifts_with_constraint("minimum_grade_number"):
if not shift.constraint_options["minimum_grade_number"]:
raise ValueError(
"Constraint option must be defined for 'minimum_grade_number'"
)
grade, min_required = shift.constraint_options["minimum_grade_number"]
for shift in self.get_shifts_with_constraint(MinimumGradeNumberConstraint):
constraint = shift.get_constraints(MinimumGradeNumberConstraint)[0]
grade, min_required = constraint.grade, constraint.min_number
# self.model.limit_grades_constraint = Constraint(
setattr(
self.model,
@@ -1231,12 +1306,11 @@ class RotaBuilder(object):
)
for shift in self.get_shifts_with_constraint(
"require_remote_site_presence_week"
RequireRemoteSitePresenceConstraint
):
site, required_number = shift.constraint_options[
"require_remote_site_presence_week"
]
# self.model.require_presence_at_site_overnight_rule = Constraint(
constraint = shift.get_constraints(RequireRemoteSitePresenceConstraint)[0]
site, required_number = constraint.site, constraint.required_number
setattr(
self.model,
f"require_remote_site_presence_week_{shift.name}",
@@ -1249,39 +1323,39 @@ class RotaBuilder(object):
),
)
def presenceAtRemoteSite(
model, day, week, shift, required_site, required_number
):
required_site_workers = [
w for w in self.workers if required_site == w.remote_site
]
#def presenceAtRemoteSite(
# model, day, week, shift, required_site, required_number
#):
# required_site_workers = [
# w for w in self.workers if required_site == w.remote_site
# ]
for w in required_site_workers:
if (w.id, week, day, shift) not in model.works:
return Constraint.Skip
# for w in required_site_workers:
# if (w.id, week, day, shift) not in model.works:
# return Constraint.Skip
return (
sum(model.works[w.id, week, day, shift] for w in required_site_workers)
>= required_number
)
# return (
# sum(model.works[w.id, week, day, shift] for w in required_site_workers)
# >= required_number
# )
for shift in self.get_shifts_with_constraint("require_remote_site_presence"):
site, required_number = shift.constraint_options[
"require_remote_site_presence"
]
# self.model.require_presence_at_site_overnight_rule = Constraint(
setattr(
self.model,
f"require_remote_site_presence_{shift.name}",
Constraint(
[day for day in self.days],
[week for week in self.weeks],
[shift.name],
[site],
[required_number],
rule=presenceAtRemoteSite,
),
)
#for shift in self.get_shifts_with_constraint("require_remote_site_presence"):
# site, required_number = shift.constraint_options[
# "require_remote_site_presence"
# ]
# # self.model.require_presence_at_site_overnight_rule = Constraint(
# setattr(
# self.model,
# f"require_remote_site_presence_{shift.name}",
# Constraint(
# [day for day in self.days],
# [week for week in self.weeks],
# [shift.name],
# [site],
# [required_number],
# rule=presenceAtRemoteSite,
# ),
# )
# # Count the number of workers from each site on each night shift
# # As 1 or 0 is optimum we can simply subtract 1 from the number
@@ -1309,7 +1383,7 @@ class RotaBuilder(object):
for site in track(
self.sites, description="Generating site balance constraints..."
):
for shift in self.get_shifts_with_constraint("balance_across_groups"):
for shift in self.get_shifts_with_constraint(BalanceAcrossGroupsConstraint):
if site in shift.sites:
block = shift.name
for week in self.weeks:
@@ -2056,7 +2130,7 @@ class RotaBuilder(object):
== sum(
self.model.works[worker.id, week, day, shift.name]
for week, day, shift in self.get_week_day_shift_combinations_for_constraint(
"night"
NightConstraint
)
# for week, day in self.get_week_day_combinations_for_shift(shift)
# for shift in self.get_shifts_with_constraint("night")
@@ -2065,7 +2139,7 @@ class RotaBuilder(object):
night_shift_target_number = sum(
worker.shift_target_number[shift.name]
for shift in self.get_shifts_with_constraint("night")
for shift in self.get_shifts_with_constraint(NightConstraint)
)
self.model.constraints.add(
@@ -2217,7 +2291,7 @@ class RotaBuilder(object):
):
continue
if self.get_shifts_with_constraint("night"):
if self.get_shifts_with_constraint(NightConstraint):
# Prevent nights more than once every n weeks
self.model.constraints.add(
1
@@ -2227,7 +2301,7 @@ class RotaBuilder(object):
]
for week in week_blocks
for shift in self.get_shifts_with_constraint(
"night"
NightConstraint
)
)
)
@@ -2247,19 +2321,18 @@ class RotaBuilder(object):
)
for constraint_shift in self.get_shifts_with_constraints(
"max_shifts_per_week_block"
MaxShiftsPerWeekBlockConstraint
):
constraint_options = constraint_shift.constraint_options[
"max_shifts_per_week_block"
]
constraint = constraint_shift.get_constraint(MaxShiftsPerWeekBlockConstraint)
# If not supplied, default to the number of days per shift
shift_number = constraint_shift.get_shift_number()
if "shift_number" in constraint_options:
shift_number = constraint_options["shift_number"]
if constraint.max_shifts is not None:
shift_number = constraint.max_shifts
else:
shift_number = constraint_shift.get_shift_number()
for week_blocks in self.get_week_block_iterator(
constraint_options["weeks"]
constraint.week_block
):
# Prevent shifts more than once every n weeks
self.model.constraints.add(
@@ -2289,11 +2362,19 @@ class RotaBuilder(object):
for constraint_shift in self.get_shifts_with_constraints(
"max_shifts_per_week"
MaxShiftsPerWeekConstraint
):
# If not supplied, default to the number of days per shift
constraint = constraint_shift.get_constraint(MaxShiftsPerWeekConstraint)
if constraint.max_shifts is not None:
shift_number = constraint.max_shifts
else:
shift_number = constraint_shift.get_shift_number()
try:
self.model.constraints.add(
constraint_shift.constraint_options["max_shifts_per_week"]
shift_number
>= sum(
self.model.works[
worker.id, w, day, constraint_shift.name
@@ -2814,9 +2895,12 @@ class RotaBuilder(object):
# shift spans 7 days you may get >7 allocations in a row
# as it only checks for a different shift allocation
for constraint_shift in self.get_shifts_with_constraints(
"pre" # , week=week
PreShiftConstraint
):
for n in range(0, constraint_shift.constraint_options["pre"]):
constraint = constraint_shift.get_constraint(PreShiftConstraint)
if week not in constraint.weeks:
continue
for n in range(0, constraint.days):
if day in constraint_shift.days:
# Only apply if worker can work this shift
if worker.site not in constraint_shift.sites:
@@ -2845,9 +2929,12 @@ class RotaBuilder(object):
)
for constraint_shift in self.get_shifts_with_constraints(
"post" # , week=week
PostShiftConstraint
):
for n in range(0, constraint_shift.constraint_options["post"]):
constraint = constraint_shift.get_constraint(PostShiftConstraint)
if week not in constraint.weeks:
continue
for n in range(0, constraint.days):
if day in constraint_shift.days:
# Only apply if worker can work this shift
if worker.site not in constraint_shift.sites:
@@ -2877,7 +2964,7 @@ class RotaBuilder(object):
# Night constraint means we won't assign a shift the day before
# an unavailability
for constraint_shift in self.get_shifts_with_constraint("night"):
for constraint_shift in self.get_shifts_with_constraint(NightConstraint):
if (
worker.id,
# pweek,
@@ -3081,13 +3168,13 @@ class RotaBuilder(object):
* 20
# self.model.night_per_site2[week, block, site]
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
for shift in self.get_shifts_with_constraint(NightConstraint)
for site in self.sites
)
else:
nights_site_balancing = 0
if self.get_shifts_with_constraint("balance_across_groups"):
if self.get_shifts_with_constraint(BalanceAcrossGroupsConstraint):
block_site_balancing = sum(
(
self.model.shift_per_site_t1[week, shift.name, site]
@@ -3096,9 +3183,7 @@ class RotaBuilder(object):
* 2000
# self.model.night_per_site2[week, block, site]
for week in self.weeks
for shift in self.get_shifts_with_constraint(
"balance_across_groups"
)
for shift in self.get_shifts_with_constraint(BalanceAcrossGroupsConstraint)
for site in self.sites
)
else:
@@ -3326,7 +3411,6 @@ class RotaBuilder(object):
if return_empty_if_week_day_not_found:
return set()
raise WeekDayNotFound(f"Week {week}, Day {day} not found in week_day_shifts_dict")
#print(self.week_day_shifts_dict)
return self.week_day_shifts_dict[(week, day)]
def get_week_day_shift_names_by_week_day(self, week, day: DayStr) -> set:
@@ -3363,7 +3447,7 @@ class RotaBuilder(object):
def get_date_by_week_day(self, week: WeekInt, day: DayStr) -> datetime.date:
return self.week_day_date_map[(week, day)]
def get_weeks_by_date_range(self, start_date: datetime.date, end_date: datetime.date) -> list[WeekInt]:
def get_weeks_by_date_range(self, start_date: datetime.date | None, end_date: datetime.date | None) -> list[WeekInt]:
"""Get a list of weeks that fall within a date range.
Args:
@@ -3375,6 +3459,11 @@ class RotaBuilder(object):
"""
# TODO: think about error if not monday start date?
weeks = []
if start_date is None:
start_date = self.start_date
if end_date is None:
end_date = self.rota_end_date
for week in self.weeks:
week_start = self.get_date_by_week_day(week, "Mon")
week_end = self.get_date_by_week_day(week, "Sun")
@@ -3471,7 +3560,6 @@ class RotaBuilder(object):
# Check worker requirements
if not isinstance(s.workers_required, int):
# Validate the worker requirements
print(s.workers_required)
for worker_requirement in s.workers_required:
if not isinstance(worker_requirement, WorkerRequirement):
raise InvalidShift(
@@ -3503,6 +3591,10 @@ class RotaBuilder(object):
for site in list(s.sites):
self.sites.add(site)
for constraint in s.constraints:
if constraint.weeks is None:
constraint.weeks = self.get_weeks_by_date_range(constraint.start_date, constraint.end_date)
self.shift_counts = defaultdict(int)
self.shift_worker_counts = defaultdict(int)
for week, day in self.weeks_days_product:
@@ -3544,11 +3636,11 @@ class RotaBuilder(object):
self.max_pre = 1
self.max_post = 1
for shift in self.get_shifts_with_constraint("pre"):
self.max_pre = max(shift.constraint_options["pre"], self.max_pre)
for shift in self.get_shifts_with_constraint(PreShiftConstraint):
self.max_pre = max(shift.get_constraint(PreShiftConstraint).days, self.max_pre)
for shift in self.get_shifts_with_constraint("post"):
self.max_post = max(shift.constraint_options["post"], self.max_post)
for shift in self.get_shifts_with_constraint(PostShiftConstraint):
self.max_post = max(shift.get_constraint(PostShiftConstraint).days, self.max_post)
# --- Check allowed_multi_shift_sets validity ---
all_shift_names = set(s.name for s in self.shifts)
@@ -3708,7 +3800,7 @@ class RotaBuilder(object):
return shifts
def get_shifts_with_constraint(self, constraint) -> List[SingleShift]:
return [shift for shift in self.shifts if constraint in shift.constraints]
return [shift for shift in self.shifts if shift.has_constraint(constraint)]
def get_shifts_with_constraints(
self, *constraints, week: int | None = None
@@ -3720,7 +3812,7 @@ class RotaBuilder(object):
[
shift.name
for shift in self.get_shifts(week)
if constraint in shift.constraints
if shift.has_constraint(constraint)
]
)
@@ -4283,9 +4375,9 @@ class RotaBuilder(object):
remote_site = ""
if shift_name:
shift = self.get_shift_by_name(shift_name)
if "require_remote_site_presence_week" in shift.constraints:
remote_site = f" data-shift-remote-site='{shift.constraints['require_remote_site_presence_week'].options[0]}'"
if "night" in shift.constraints:
if shift.has_constraint(RequireRemoteSitePresenceConstraint):
remote_site = f" data-shift-remote-site='{shift.get_constraint(RequireRemoteSitePresenceConstraint).site}'"
if shift.has_constraint(NightConstraint):
css_class = " ".join((css_class, "night-shift"))
shift_tds.append(
@@ -4586,7 +4678,6 @@ class RotaBuilder(object):
shifts = {}
for worker in self.workers:
print("*", worker.name)
shifts[worker.name] = len(
[i for i in self.get_worker_shift_list(worker) if i != ""]
)
@@ -4703,7 +4794,6 @@ class RotaBuilder(object):
if c > 0:
l.append(f"{shift} ({c})")
total_shifts = total_shifts + c
# print(worker.id, shift)
timetable[worker.get_details()][shift] = c
t.append(f"{worker.get_full_details()} [{total_shifts}]: {', '.join(l)}")
return "\n".join(t)
@@ -4732,7 +4822,6 @@ class RotaBuilder(object):
if c > 0:
l.append(f"{shift} ({c})")
total_shifts = total_shifts + c
# print(worker.id, shift)
shift_obj = self.get_shift_by_name(shift)
timetable[worker.get_details()][shift_obj.get_display_char()] = c
t.append(f"{worker.get_full_details()} [{total_shifts}]: {', '.join(l)}")