Compare commits
8
Commits
33f50c7278
..
master
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d577b2cdf8 | ||
|
|
0874f05126 | ||
|
|
6ca3db86cb | ||
|
|
96e6a0d625 | ||
|
|
fc2413538c | ||
|
|
5ff78faa07 | ||
|
|
d1fbbe2dc9 | ||
|
|
b5525bd3fb |
@@ -16,10 +16,13 @@ try:
|
||||
WorkRequests,
|
||||
HardDayDependency,
|
||||
HardDayExclusion,
|
||||
ShiftStartDate,
|
||||
ShiftEndDate,
|
||||
)
|
||||
except Exception:
|
||||
NonWorkingDays = OutOfProgramme = NotAvailableToWork = PreferenceNotToWork = None
|
||||
WorkRequests = HardDayDependency = HardDayExclusion = None
|
||||
ShiftStartDate = ShiftEndDate = None
|
||||
|
||||
|
||||
class WorkerForm(forms.ModelForm):
|
||||
@@ -46,6 +49,9 @@ class WorkerForm(forms.ModelForm):
|
||||
assign_as_block_preferences = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->weight", label="Block preferences (JSON)")
|
||||
shift_fte_overrides = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->fte", label="Shift FTE overrides (JSON)")
|
||||
exact_shifts = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->exact_count", label="Exact shifts (JSON)")
|
||||
shift_start_dates = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->start_date (YYYY-MM-DD)", label="Shift start dates (JSON)")
|
||||
shift_end_dates = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON mapping shift_name->end_date (YYYY-MM-DD)", label="Shift end dates (JSON)")
|
||||
groups = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of group names (e.g. [\"group1\", \"group2\"])", label="Groups (JSON)")
|
||||
previous_shifts = forms.CharField(required=False, widget=forms.Textarea, help_text="Free JSON structure for previous shifts", label="Previous shifts (JSON)")
|
||||
shift_balance_extra = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON structure for extra shift-balance penalties", label="Shift balance extra (JSON)")
|
||||
allowed_multi_shift_sets = forms.CharField(required=False, widget=forms.Textarea, help_text="JSON list of sets (e.g. [[\"a\",\"b\"], ...])", label="Allowed multi-shift sets (JSON)")
|
||||
@@ -96,6 +102,9 @@ class WorkerForm(forms.ModelForm):
|
||||
"assign_as_block_preferences",
|
||||
"shift_fte_overrides",
|
||||
"exact_shifts",
|
||||
"shift_start_dates",
|
||||
"shift_end_dates",
|
||||
"groups",
|
||||
"previous_shifts",
|
||||
"shift_balance_extra",
|
||||
"allowed_multi_shift_sets",
|
||||
@@ -149,6 +158,9 @@ class WorkerForm(forms.ModelForm):
|
||||
"assign_as_block_preferences",
|
||||
"shift_fte_overrides",
|
||||
"exact_shifts",
|
||||
"shift_start_dates",
|
||||
"shift_end_dates",
|
||||
"groups",
|
||||
"previous_shifts",
|
||||
"shift_balance_extra",
|
||||
"allowed_multi_shift_sets",
|
||||
@@ -269,9 +281,10 @@ class WorkerForm(forms.ModelForm):
|
||||
cleaned["work_requests"] = _parse_and_validate("work_requests", WorkRequests)
|
||||
cleaned["locum_availability"] = _parse_and_validate("locum_availability", WorkRequests)
|
||||
|
||||
# Hard day dependency/exclusion structures
|
||||
cleaned["hard_day_dependencies"] = _parse_and_validate("hard_day_dependencies", HardDayDependency)
|
||||
cleaned["hard_day_exclusions"] = _parse_and_validate("hard_day_exclusions", HardDayExclusion)
|
||||
cleaned["shift_start_dates"] = _parse_and_validate("shift_start_dates", ShiftStartDate)
|
||||
cleaned["shift_end_dates"] = _parse_and_validate("shift_end_dates", ShiftEndDate)
|
||||
|
||||
# forced assignments: expect list of tuples [week, day, shift]
|
||||
fa = cleaned.get("forced_assignments")
|
||||
@@ -800,6 +813,16 @@ class ShiftForm(forms.Form):
|
||||
widget=forms.Textarea(attrs={"rows": 4, "class": "textarea"}),
|
||||
help_text="Optional JSON list of constraint objects for this shift (e.g. [{'name':'night','options':{}}])",
|
||||
)
|
||||
include_groups = forms.CharField(
|
||||
required=False,
|
||||
help_text="Comma separated list of groups that can work this shift (leave empty for all)",
|
||||
label="Include groups",
|
||||
)
|
||||
exclude_groups = forms.CharField(
|
||||
required=False,
|
||||
help_text="Comma separated list of groups that cannot work this shift",
|
||||
label="Exclude groups",
|
||||
)
|
||||
|
||||
def clean_sites(self):
|
||||
val = self.cleaned_data["sites"]
|
||||
@@ -841,3 +864,22 @@ class ShiftForm(forms.Form):
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise forms.ValidationError(f"Invalid JSON for constraints: {exc}")
|
||||
|
||||
def clean_include_groups(self):
|
||||
val = self.cleaned_data.get("include_groups") or ""
|
||||
return [g.strip() for g in val.split(",") if g.strip()]
|
||||
|
||||
def clean_exclude_groups(self):
|
||||
val = self.cleaned_data.get("exclude_groups") or ""
|
||||
return [g.strip() for g in val.split(",") if g.strip()]
|
||||
|
||||
def clean(self):
|
||||
cleaned_data = super().clean()
|
||||
include = cleaned_data.get("include_groups") or []
|
||||
exclude = cleaned_data.get("exclude_groups") or []
|
||||
overlap = set(include).intersection(set(exclude))
|
||||
if overlap:
|
||||
raise forms.ValidationError(
|
||||
f"Include groups and Exclude groups cannot overlap. Overlapping groups: {', '.join(overlap)}"
|
||||
)
|
||||
return cleaned_data
|
||||
|
||||
@@ -242,6 +242,8 @@ class Worker(models.Model):
|
||||
"hard_day_exclusions",
|
||||
"forced_assignments",
|
||||
"forced_assignments_by_date",
|
||||
"shift_start_dates",
|
||||
"shift_end_dates",
|
||||
]
|
||||
|
||||
for k in int_keys:
|
||||
@@ -274,6 +276,17 @@ class Worker(models.Model):
|
||||
else:
|
||||
pyd_kwargs[k] = []
|
||||
|
||||
set_keys = ["groups"]
|
||||
for k in set_keys:
|
||||
v = pyd_kwargs.get(k, None)
|
||||
try:
|
||||
if v is None:
|
||||
pyd_kwargs[k] = set()
|
||||
else:
|
||||
pyd_kwargs[k] = set(v)
|
||||
except Exception:
|
||||
pyd_kwargs[k] = set()
|
||||
|
||||
# allowed_multi_shift_sets is expected to be a list of sets; if the
|
||||
# DB contains lists-of-lists convert inner lists to sets.
|
||||
ams = pyd_kwargs.get("allowed_multi_shift_sets")
|
||||
|
||||
@@ -1690,6 +1690,8 @@ def shift_add(request, rota_id):
|
||||
"workers_required": int(data["workers_required"]),
|
||||
"assign_as_block": bool(data["assign_as_block"]),
|
||||
"balance_offset": data.get("balance_offset"),
|
||||
"include_groups": data.get("include_groups", []),
|
||||
"exclude_groups": data.get("exclude_groups", []),
|
||||
"constraints": data.get("constraints", []),
|
||||
}
|
||||
|
||||
@@ -1738,6 +1740,9 @@ def shift_edit(request, rota_id, idx):
|
||||
"days": data["days"],
|
||||
"workers_required": int(data["workers_required"]),
|
||||
"assign_as_block": bool(data["assign_as_block"]),
|
||||
"balance_offset": data.get("balance_offset"),
|
||||
"include_groups": data.get("include_groups", []),
|
||||
"exclude_groups": data.get("exclude_groups", []),
|
||||
"constraints": data.get("constraints", []),
|
||||
}
|
||||
current = rota.shifts or []
|
||||
@@ -1766,6 +1771,8 @@ def shift_edit(request, rota_id, idx):
|
||||
"workers_required": shift.get("workers_required"),
|
||||
"assign_as_block": shift.get("assign_as_block", False),
|
||||
"balance_offset": shift.get("balance_offset", None),
|
||||
"include_groups": ",".join(shift.get("include_groups") or []),
|
||||
"exclude_groups": ",".join(shift.get("exclude_groups") or []),
|
||||
"constraints": json.dumps(shift.get("constraints", [])) if shift.get("constraints") is not None else "",
|
||||
}
|
||||
form = ShiftForm(initial=initial)
|
||||
|
||||
+26
-3
@@ -507,11 +507,34 @@ def load_workers():
|
||||
|
||||
|
||||
|
||||
def parse_time_limit(t) -> int:
|
||||
"""Parses a time limit string (e.g. '10h', '30m', '45s') or an integer to seconds."""
|
||||
if isinstance(t, int):
|
||||
return t
|
||||
if isinstance(t, float):
|
||||
return int(t)
|
||||
t = str(t).strip().lower()
|
||||
if not t:
|
||||
return 0
|
||||
if t.isdigit():
|
||||
return int(t)
|
||||
if t.endswith("h"):
|
||||
return int(float(t[:-1]) * 3600)
|
||||
if t.endswith("m"):
|
||||
return int(float(t[:-1]) * 60)
|
||||
if t.endswith("s"):
|
||||
return int(float(t[:-1]))
|
||||
try:
|
||||
return int(float(t))
|
||||
except ValueError:
|
||||
raise ValueError(f"Invalid time format: {t}. Expected format like '10h', '30m', '45s' or a number of seconds.")
|
||||
|
||||
|
||||
@app.command()
|
||||
def main(
|
||||
suspend: bool = False,
|
||||
solve: bool = True,
|
||||
time_to_run: int = 60 * 60,
|
||||
time_to_run: str = "1h",
|
||||
ratio: float = 0.1,
|
||||
start_date: datetime.datetime = ROTA_START_DATE,
|
||||
weeks: int = 22,
|
||||
@@ -698,8 +721,8 @@ def main(
|
||||
# Rota.build_workers()
|
||||
# Rota.build_model()
|
||||
|
||||
solver_options = {"ratio": ratio, "seconds": time_to_run, "threads": 10}
|
||||
# solver_options = {"seconds": time_to_run, "threads": 10}
|
||||
seconds_to_run = parse_time_limit(time_to_run)
|
||||
solver_options = {"ratio": ratio, "seconds": seconds_to_run, "threads": 10}
|
||||
|
||||
# start_time = time.time()
|
||||
Rota.build_and_solve(solver_options, export=True, export_with_timestamp=False, solve=solve, solver="appsi_highs")
|
||||
|
||||
+49
-6
@@ -34,6 +34,7 @@ sites = (
|
||||
"truro ir",
|
||||
"exeter ir",
|
||||
"torbay ir",
|
||||
"ir nights",
|
||||
)
|
||||
|
||||
from rota_generator.workers import (
|
||||
@@ -45,12 +46,34 @@ from rota_generator.workers import (
|
||||
OutOfProgramme,
|
||||
)
|
||||
|
||||
def parse_time_limit(t) -> int:
|
||||
"""Parses a time limit string (e.g. '10h', '30m', '45s') or an integer to seconds."""
|
||||
if isinstance(t, int):
|
||||
return t
|
||||
if isinstance(t, float):
|
||||
return int(t)
|
||||
t = str(t).strip().lower()
|
||||
if not t:
|
||||
return 0
|
||||
if t.isdigit():
|
||||
return int(t)
|
||||
if t.endswith("h"):
|
||||
return int(float(t[:-1]) * 3600)
|
||||
if t.endswith("m"):
|
||||
return int(float(t[:-1]) * 60)
|
||||
if t.endswith("s"):
|
||||
return int(float(t[:-1]))
|
||||
try:
|
||||
return int(float(t))
|
||||
except ValueError:
|
||||
raise ValueError(f"Invalid time format: {t}. Expected format like '10h', '30m', '45s' or a number of seconds.")
|
||||
|
||||
|
||||
@app.command()
|
||||
def main(
|
||||
suspend: bool = False,
|
||||
solve: bool = True,
|
||||
time_to_run: int = 60 * 60 * 10,
|
||||
time_to_run: str = "10h",
|
||||
ratio: float = 0.1,
|
||||
start_date: datetime.datetime = "2026-09-07",
|
||||
weeks: int = 26,
|
||||
@@ -74,6 +97,7 @@ def main(
|
||||
Rota.constraint_options["max_weekend_frequency"] = 3
|
||||
|
||||
Rota.constraint_options["max_days_per_week_block"] = [(8,3)]
|
||||
Rota.constraint_options["maximum_allowed_shift_diff"] = 1
|
||||
|
||||
# wr = [
|
||||
# WorkerRequirement(
|
||||
@@ -174,11 +198,11 @@ def main(
|
||||
length=12.5,
|
||||
days=days[4:],
|
||||
# balance_offset=3,
|
||||
rota_on_nwds=True,
|
||||
force_as_block=True,
|
||||
#rota_on_nwds=True,
|
||||
#force_as_block=True,
|
||||
# assign_as_block=True,
|
||||
constraints=[PreShiftConstraint(days=2), PostShiftConstraint(days=2)],
|
||||
# force_as_block_unless_nwd=True
|
||||
force_as_block_unless_nwd=[["Fri"], ["Sat", "Sun"]],
|
||||
),
|
||||
SingleShift(
|
||||
sites=("torbay", "torbay twilights and weekends"),
|
||||
@@ -503,6 +527,24 @@ def main(
|
||||
"night_weekday": 0,
|
||||
}
|
||||
|
||||
override_shift_start_dates = []
|
||||
if worker in ["Anushka Kulkarni"]:
|
||||
override_shift_start_dates = [
|
||||
{
|
||||
"shift": "night_weekday",
|
||||
"start_date": datetime.datetime.strptime(
|
||||
"2026-11-01", "%Y-%m-%d"
|
||||
).date(),
|
||||
},
|
||||
{
|
||||
"shift": "night_weekend",
|
||||
"start_date": datetime.datetime.strptime(
|
||||
"2026-11-01", "%Y-%m-%d"
|
||||
).date(),
|
||||
},
|
||||
]
|
||||
|
||||
|
||||
# if worker_name == "Hadi Mohamed":
|
||||
# shift_fte_overrides = {
|
||||
# "weekend_exeter": 50,
|
||||
@@ -536,6 +578,7 @@ def main(
|
||||
bank_holiday_extra=w["bank_holiday_extra"],
|
||||
shift_fte_overrides=shift_fte_overrides,
|
||||
exact_shifts=exact_shifts,
|
||||
shift_start_dates=override_shift_start_dates,
|
||||
)
|
||||
|
||||
# print(w)
|
||||
@@ -546,8 +589,8 @@ def main(
|
||||
# Rota.build_workers()
|
||||
# Rota.build_model()
|
||||
|
||||
solver_options = {"ratio": ratio, "seconds": time_to_run, "threads": 10}
|
||||
# solver_options = {"seconds": time_to_run, "threads": 10}
|
||||
seconds_to_run = parse_time_limit(time_to_run)
|
||||
solver_options = {"ratio": ratio, "seconds": seconds_to_run, "threads": 10}
|
||||
|
||||
# start_time = time.time()
|
||||
Rota.build_and_solve(solver_options, export=True, solve=solve, solver="appsi_highs", export_with_timestamp=True)
|
||||
|
||||
@@ -33,7 +33,7 @@ def load_leave(Rota):
|
||||
|
||||
if live_rota:
|
||||
download = s.get(
|
||||
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSh64r9uUbPsCdhcV7zmZz10gVqIy6rhXFUvDgJxRD5W45IYEaZpKMxAqDsD1ph6dob4AiRzJXVVtOu/pub?gid=580824586&single=true&output=csv",
|
||||
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSh64r9uUbPsCdhcV7zmZz10gVqIy6rhXFUvDgJxRD5W45IYEaZpKMxAqDsD1ph6dob4AiRzJXVVtOu/pub?gid=396859373&single=true&output=csv",
|
||||
headers=headers,
|
||||
)
|
||||
# download = s.get("https://docs.google.com/spreadsheets/d/e/2PACX-1vSRx9VWXSlRubPyA0RhiI-Oqf5eHNYYEc6rFzlraDbR5_8qqr5g13-4uV-gn4u-TjZxiSMv1fBUaESq/pub?gid=814517272&single=true&output=csv", headers=headers)
|
||||
@@ -295,7 +295,7 @@ def load_academy(Rota):
|
||||
"""
|
||||
# Use the same published CSV as load_leave (main sheet)
|
||||
url = (
|
||||
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSh64r9uUbPsCdhcV7zmZz10gVqIy6rhXFUvDgJxRD5W45IYEaZpKMxAqDsD1ph6dob4AiRzJXVVtOu/pub?gid=580824586&single=true&output=csv"
|
||||
"https://docs.google.com/spreadsheets/d/e/2PACX-1vSh64r9uUbPsCdhcV7zmZz10gVqIy6rhXFUvDgJxRD5W45IYEaZpKMxAqDsD1ph6dob4AiRzJXVVtOu/pub?gid=396859373&single=true&output=csv"
|
||||
)
|
||||
|
||||
with Session() as s:
|
||||
|
||||
+199
-46
@@ -346,7 +346,7 @@ def _pydantic_to_dict(obj):
|
||||
# Recursively convert pydantic models, lists and dicts to JSON-serializable structures
|
||||
if obj is None:
|
||||
return None
|
||||
if isinstance(obj, list):
|
||||
if isinstance(obj, (list, set, tuple)):
|
||||
return [_pydantic_to_dict(x) for x in obj]
|
||||
if isinstance(obj, dict):
|
||||
return {k: _pydantic_to_dict(v) for k, v in obj.items()}
|
||||
@@ -417,7 +417,7 @@ class SingleShift(BaseModel):
|
||||
rota_on_nwds: bool = False
|
||||
assign_as_block: bool = False
|
||||
force_as_block: bool = False
|
||||
force_as_block_unless_nwd: bool = False
|
||||
force_as_block_unless_nwd: bool | List[List[str]] = False
|
||||
hard_constrain_shift: bool = True
|
||||
bank_holidays_only: bool = False
|
||||
#constraint: list[ShiftConstraint] = []
|
||||
@@ -433,6 +433,17 @@ class SingleShift(BaseModel):
|
||||
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
|
||||
include_groups: set[str] = set()
|
||||
exclude_groups: set[str] = set()
|
||||
|
||||
@model_validator(mode="after")
|
||||
def check_groups_non_overlapping(self) -> "SingleShift":
|
||||
intersection = self.include_groups.intersection(self.exclude_groups)
|
||||
if intersection:
|
||||
raise ValueError(
|
||||
f"include_groups and exclude_groups cannot overlap. Intersection: {intersection}"
|
||||
)
|
||||
return self
|
||||
|
||||
model_config = ConfigDict(
|
||||
extra="allow",
|
||||
@@ -855,6 +866,8 @@ class RotaBuilder(object):
|
||||
options["time_limit"] = options.pop("seconds")
|
||||
if "ratio" in options:
|
||||
options["mip_rel_gap"] = options.pop("ratio")
|
||||
if "threads" in options:
|
||||
options.pop("threads")
|
||||
else:
|
||||
self.opt = SolverFactory(solver)
|
||||
|
||||
@@ -1037,20 +1050,28 @@ class RotaBuilder(object):
|
||||
initialize=0,
|
||||
)
|
||||
|
||||
blocks_worker_keys = []
|
||||
for worker, week, shift_name in self.worker_week_shifts_to_assign_as_blocks():
|
||||
shift = self.get_shift_by_name(shift_name)
|
||||
sub_blocks = self.get_shift_sub_blocks(shift)
|
||||
for sb_idx in range(len(sub_blocks)):
|
||||
blocks_worker_keys.append((worker.id, week, shift_name, sb_idx))
|
||||
|
||||
self.model.blocks_worker_shift_assigned = Var(
|
||||
(
|
||||
(worker.id, week, shift)
|
||||
for worker, week, shift in self.worker_week_shifts_to_assign_as_blocks()
|
||||
),
|
||||
blocks_worker_keys,
|
||||
within=Binary,
|
||||
initialize=0,
|
||||
)
|
||||
|
||||
blocks_assigned_keys = []
|
||||
for week, shift_name in self.week_shifts_to_assign_as_blocks():
|
||||
shift = self.get_shift_by_name(shift_name)
|
||||
sub_blocks = self.get_shift_sub_blocks(shift)
|
||||
for sb_idx in range(len(sub_blocks)):
|
||||
blocks_assigned_keys.append((week, shift_name, sb_idx))
|
||||
|
||||
self.model.blocks_assigned = Var(
|
||||
(
|
||||
(week, shift)
|
||||
for week, shift in self.week_shifts_to_assign_as_blocks()
|
||||
),
|
||||
blocks_assigned_keys,
|
||||
within=NonNegativeIntegers,
|
||||
initialize=0,
|
||||
)
|
||||
@@ -1610,13 +1631,16 @@ class RotaBuilder(object):
|
||||
workers_required,
|
||||
site_required,
|
||||
) in self.get_required_workers_and_site_combinations():
|
||||
# print(week, day, shift, workers_required, site_required)
|
||||
shift_obj = self.get_shift_by_name(shift)
|
||||
eligible_workers = [
|
||||
worker for worker in self.workers
|
||||
if self.is_worker_eligible_for_shift(worker, shift_obj)
|
||||
]
|
||||
self.model.constraints.add(
|
||||
workers_required
|
||||
== sum(
|
||||
self.model.works[worker.id, week, day, shift]
|
||||
for worker in self.workers
|
||||
if worker.site in site_required
|
||||
for worker in eligible_workers
|
||||
)
|
||||
)
|
||||
|
||||
@@ -1625,7 +1649,7 @@ class RotaBuilder(object):
|
||||
self.model.locum_required[week, day, shift]
|
||||
== sum(
|
||||
self.model.works[worker.id, week, day, shift]
|
||||
for worker in self.workers
|
||||
for worker in eligible_workers
|
||||
if worker.locum
|
||||
)
|
||||
)
|
||||
@@ -1638,6 +1662,7 @@ class RotaBuilder(object):
|
||||
# if not worker.locum
|
||||
)
|
||||
)
|
||||
|
||||
self.model.constraints.add(
|
||||
0
|
||||
== sum(
|
||||
@@ -1647,14 +1672,17 @@ class RotaBuilder(object):
|
||||
)
|
||||
)
|
||||
|
||||
# And it is not assigned if the worker is from the wrong site
|
||||
if [worker for worker in self.workers if worker.site not in site_required]:
|
||||
# And it is not assigned if the worker is ineligible
|
||||
ineligible_workers = [
|
||||
worker for worker in self.workers
|
||||
if not self.is_worker_eligible_for_shift(worker, shift_obj)
|
||||
]
|
||||
if ineligible_workers:
|
||||
self.model.constraints.add(
|
||||
0
|
||||
== sum(
|
||||
self.model.works[worker.id, week, day, shift]
|
||||
for worker in self.workers
|
||||
if worker.site not in site_required
|
||||
for worker in ineligible_workers
|
||||
)
|
||||
)
|
||||
# # Ensure shifts are only assigned by workers from the allowed sites
|
||||
@@ -1880,30 +1908,33 @@ class RotaBuilder(object):
|
||||
|
||||
for week, shift_name in self.week_shifts_to_assign_as_blocks():
|
||||
shift = self.get_shift_by_name(shift_name)
|
||||
sub_blocks = self.get_shift_sub_blocks(shift)
|
||||
|
||||
for sb_idx, sb in enumerate(sub_blocks):
|
||||
for worker in self.get_workers_for_shift(shift):
|
||||
if (worker.id, week, shift.name) in self.model.blocks_worker_shift_assigned:
|
||||
if (worker.id, week, shift.name, sb_idx) in self.model.blocks_worker_shift_assigned:
|
||||
try:
|
||||
self.model.constraints.add(
|
||||
8
|
||||
* self.model.blocks_worker_shift_assigned[
|
||||
worker.id, week, shift_name
|
||||
worker.id, week, shift_name, sb_idx
|
||||
]
|
||||
>= sum(
|
||||
self.model.works[worker.id, week, day, shift_name]
|
||||
for day in shift.days
|
||||
for day in sb
|
||||
)
|
||||
)
|
||||
except KeyError:
|
||||
pass
|
||||
|
||||
self.model.constraints.add(
|
||||
self.model.blocks_assigned[week, shift_name]
|
||||
self.model.blocks_assigned[week, shift_name, sb_idx]
|
||||
== sum(
|
||||
self.model.blocks_worker_shift_assigned[
|
||||
worker.id, week, shift_name
|
||||
worker.id, week, shift_name, sb_idx
|
||||
]
|
||||
for worker in self.workers
|
||||
if (worker.id, week, shift.name) in self.model.blocks_worker_shift_assigned
|
||||
if (worker.id, week, shift.name, sb_idx) in self.model.blocks_worker_shift_assigned
|
||||
)
|
||||
)
|
||||
|
||||
@@ -1913,7 +1944,7 @@ class RotaBuilder(object):
|
||||
|
||||
if shift.force_as_block_unless_nwd:
|
||||
workers = self.get_workers_for_shift(shift)
|
||||
# Get workers who have a nwd on the shift
|
||||
# Get workers who have a nwd on the sub-block
|
||||
nwd_workers = []
|
||||
full_workers = []
|
||||
|
||||
@@ -1925,7 +1956,7 @@ class RotaBuilder(object):
|
||||
start_nwd_date,
|
||||
end_nwd_date,
|
||||
) in w.non_working_day_list:
|
||||
if nwd in shift.days:
|
||||
if nwd in sb:
|
||||
if start_nwd_date > self.get_week_start_date(week):
|
||||
continue
|
||||
if end_nwd_date < self.get_week_start_date(week):
|
||||
@@ -1936,7 +1967,6 @@ class RotaBuilder(object):
|
||||
nwd_workers.append(w)
|
||||
else:
|
||||
full_workers.append(w)
|
||||
|
||||
else:
|
||||
full_workers.append(w)
|
||||
|
||||
@@ -1944,20 +1974,21 @@ class RotaBuilder(object):
|
||||
self.model.constraints.add(
|
||||
sum(
|
||||
self.model.blocks_worker_shift_assigned[
|
||||
worker.id, week, shift.name
|
||||
worker.id, week, shift.name, sb_idx
|
||||
]
|
||||
for worker in self.workers
|
||||
if (worker.id, week, shift.name, sb_idx) in self.model.blocks_worker_shift_assigned
|
||||
)
|
||||
<= workers_required
|
||||
)
|
||||
|
||||
else:
|
||||
self.model.constraints.add(
|
||||
sum(
|
||||
self.model.blocks_worker_shift_assigned[
|
||||
worker.id, week, shift.name
|
||||
worker.id, week, shift.name, sb_idx
|
||||
]
|
||||
for worker in full_workers
|
||||
if (worker.id, week, shift.name, sb_idx) in self.model.blocks_worker_shift_assigned
|
||||
)
|
||||
<= workers_required + 1
|
||||
)
|
||||
@@ -1967,9 +1998,10 @@ class RotaBuilder(object):
|
||||
self.model.constraints.add(
|
||||
sum(
|
||||
self.model.blocks_worker_shift_assigned[
|
||||
worker.id, week, shift.name
|
||||
worker.id, week, shift.name, sb_idx
|
||||
]
|
||||
for worker in self.workers
|
||||
if (worker.id, week, shift.name, sb_idx) in self.model.blocks_worker_shift_assigned
|
||||
)
|
||||
<= workers_required
|
||||
)
|
||||
@@ -2647,14 +2679,14 @@ class RotaBuilder(object):
|
||||
self.get_shifts(), description="Generate shift balance constraints"
|
||||
):
|
||||
if (
|
||||
worker.site in shift.sites
|
||||
self.is_worker_eligible_for_shift(worker, shift)
|
||||
): # Each site specfies which sites self.workers can fullfill
|
||||
total_shifts = self.shift_worker_counts[shift.name]
|
||||
|
||||
# Find workers who have exact shifts for this shift type
|
||||
exact_workers = [
|
||||
w for w in self.workers
|
||||
if w.site in shift.sites and shift.name in getattr(w, "exact_shifts", {})
|
||||
if self.is_worker_eligible_for_shift(w, shift) and shift.name in getattr(w, "exact_shifts", {})
|
||||
]
|
||||
exact_shifts_sum = sum(w.exact_shifts[shift.name] for w in exact_workers)
|
||||
|
||||
@@ -2671,7 +2703,7 @@ class RotaBuilder(object):
|
||||
full_time_equivalent_joined = sum(
|
||||
w.get_fte(shift=shift.name)
|
||||
for w in self.workers
|
||||
if w.site in shift.sites and w not in exact_workers
|
||||
if self.is_worker_eligible_for_shift(w, shift) and w not in exact_workers
|
||||
)
|
||||
|
||||
if not full_time_equivalent_joined:
|
||||
@@ -2768,6 +2800,7 @@ class RotaBuilder(object):
|
||||
"target_shifts": target_shifts,
|
||||
}
|
||||
|
||||
if self.get_week_day_combinations_for_shift(shift):
|
||||
self.model.constraints.add(
|
||||
inequality(
|
||||
min_shifts,
|
||||
@@ -3705,6 +3738,22 @@ class RotaBuilder(object):
|
||||
)
|
||||
)
|
||||
|
||||
# Shift-specific active dates constraint
|
||||
for shift in self.get_shifts():
|
||||
if (worker.id, week, day, shift.name) in self.model.works:
|
||||
calc_start = getattr(worker, "calculated_shift_start_dates", {}).get(shift.name, worker.calculated_start_date)
|
||||
calc_end = getattr(worker, "calculated_shift_end_dates", {}).get(shift.name, worker.calculated_end_date)
|
||||
if calc_start is not None and calc_end is not None:
|
||||
date = self.week_day_date_map[(week, day)]
|
||||
if date < calc_start or date >= calc_end:
|
||||
self.model.constraints.add(
|
||||
self.model.works[worker.id, week, day, shift.name] == 0
|
||||
)
|
||||
if self.get_locum_workers() and (worker.id, week, day, shift.name) in self.model.locum_works:
|
||||
self.model.constraints.add(
|
||||
self.model.locum_works[worker.id, week, day, shift.name] == 0
|
||||
)
|
||||
|
||||
# single shift per day (unless multi-shift allowed)
|
||||
# This is signifantly slower so only enable if required
|
||||
shifts_today = self.get_shift_names_by_week_day(week, day)
|
||||
@@ -3825,7 +3874,7 @@ class RotaBuilder(object):
|
||||
continue
|
||||
if day in constraint_shift.days:
|
||||
# Only apply if worker can work this shift
|
||||
if worker.site not in constraint_shift.sites:
|
||||
if not self.is_worker_eligible_for_shift(worker, constraint_shift):
|
||||
continue
|
||||
try:
|
||||
works = self.model.works[
|
||||
@@ -3854,7 +3903,7 @@ class RotaBuilder(object):
|
||||
)
|
||||
if shiftname not in ignore_shifts
|
||||
for w in workers
|
||||
if w.site in constraint_shift.sites # Only workers who can work this shift
|
||||
if self.is_worker_eligible_for_shift(w, constraint_shift) # Only workers who can work this shift
|
||||
)
|
||||
)
|
||||
|
||||
@@ -3876,7 +3925,7 @@ class RotaBuilder(object):
|
||||
continue
|
||||
if day in constraint_shift.days:
|
||||
# Only apply if worker can work this shift
|
||||
if worker.site not in constraint_shift.sites:
|
||||
if not self.is_worker_eligible_for_shift(worker, constraint_shift):
|
||||
continue
|
||||
try:
|
||||
works = self.model.works[
|
||||
@@ -3905,7 +3954,7 @@ class RotaBuilder(object):
|
||||
)
|
||||
if shiftname not in ignore_shifts
|
||||
for w in workers
|
||||
if w.site in constraint_shift.sites # Only workers who can work this shift
|
||||
if self.is_worker_eligible_for_shift(w, constraint_shift) # Only workers who can work this shift
|
||||
)
|
||||
)
|
||||
|
||||
@@ -4164,24 +4213,27 @@ class RotaBuilder(object):
|
||||
if self.constraint_options_model.balance_blocks:
|
||||
blocks_balancing = sum(
|
||||
block_shift_balancing_constant
|
||||
* self.model.blocks_assigned[week, shift]
|
||||
* self.model.blocks_assigned[week, shift, sb_idx]
|
||||
for week in self.weeks
|
||||
for shift in self.shifts_to_assign_as_blocks()
|
||||
for sb_idx in range(len(self.get_shift_sub_blocks(self.get_shift_by_name(shift))))
|
||||
if (week, shift, sb_idx) in self.model.blocks_assigned
|
||||
)
|
||||
else:
|
||||
blocks_balancing = 0
|
||||
|
||||
prefer_block_expr = sum(
|
||||
worker.assign_as_block_preferences.get(shift.name, 0)
|
||||
* self.model.blocks_worker_shift_assigned[worker.id, week, shift.name]
|
||||
* self.model.blocks_worker_shift_assigned[worker.id, week, shift.name, sb_idx]
|
||||
for worker in self.workers
|
||||
for week in self.weeks
|
||||
for shift in self.shifts
|
||||
if hasattr(worker, "assign_as_block_preferences")
|
||||
and shift.name in worker.assign_as_block_preferences
|
||||
and (worker.id, week, shift.name)
|
||||
and self.is_worker_eligible_for_shift(worker, shift)
|
||||
for sb_idx in range(len(self.get_shift_sub_blocks(shift)))
|
||||
if (worker.id, week, shift.name, sb_idx)
|
||||
in self.model.blocks_worker_shift_assigned
|
||||
and worker.site in shift.sites
|
||||
)
|
||||
|
||||
# Quadratic :(
|
||||
@@ -4277,11 +4329,48 @@ class RotaBuilder(object):
|
||||
"Invalid exact shift site",
|
||||
f"Worker {worker.name} requested exact shifts for shift {s_name} but their site {worker.site} is not in the shift sites {s.sites}",
|
||||
)
|
||||
elif not self.is_worker_eligible_for_shift(worker, s):
|
||||
self.add_warning(
|
||||
"Invalid exact shift group",
|
||||
f"Worker {worker.name} requested exact shifts for shift {s_name} but is not eligible for it",
|
||||
)
|
||||
except KeyError:
|
||||
self.add_warning(
|
||||
"Invalid exact shift",
|
||||
f"Worker {worker.name} requested exact shifts for non-existent shift {s_name}",
|
||||
)
|
||||
|
||||
# Validate shift-dependent start/end dates
|
||||
start_dates = getattr(worker, "shift_start_dates", [])
|
||||
end_dates = getattr(worker, "shift_end_dates", [])
|
||||
|
||||
start_dict = {}
|
||||
if isinstance(start_dates, dict):
|
||||
start_dict = start_dates
|
||||
elif isinstance(start_dates, list):
|
||||
for item in start_dates:
|
||||
if hasattr(item, "shift"):
|
||||
start_dict[item.shift] = item.start_date
|
||||
|
||||
end_dict = {}
|
||||
if isinstance(end_dates, dict):
|
||||
end_dict = end_dates
|
||||
elif isinstance(end_dates, list):
|
||||
for item in end_dates:
|
||||
if hasattr(item, "shift"):
|
||||
end_dict[item.shift] = item.end_date
|
||||
|
||||
for s_name in set(list(start_dict.keys()) + list(end_dict.keys())):
|
||||
if s_name not in self.shifts_by_name:
|
||||
raise InvalidShift(
|
||||
f"Worker {worker.name} specified shift-dependent dates for non-existent shift {s_name}"
|
||||
)
|
||||
s = self.get_shift_by_name(s_name)
|
||||
if not self.is_worker_eligible_for_shift(worker, s):
|
||||
self.add_warning(
|
||||
"Worker/ineligible shift date constraint",
|
||||
f"Worker {worker.name} specified shift-dependent dates for shift {s_name} but is not eligible to work it"
|
||||
)
|
||||
wid = worker.id
|
||||
if wid in self.workers_id_map:
|
||||
message = f"Worker with id '{wid}' has been added twice"
|
||||
@@ -4294,7 +4383,13 @@ class RotaBuilder(object):
|
||||
self.add_warning("Worker/duplicate name", message)
|
||||
self.workers_name_map[worker.name] = worker
|
||||
|
||||
if worker.site not in self.sites:
|
||||
worker_has_valid_shift = False
|
||||
for shift in self.shifts:
|
||||
if self.is_worker_eligible_for_shift(worker, shift):
|
||||
worker_has_valid_shift = True
|
||||
break
|
||||
|
||||
if not worker_has_valid_shift:
|
||||
message = f"Worker with name '{worker.name}' ({worker.id}) has no valid shifts (site: {worker.site})"
|
||||
logger.warning(message)
|
||||
self.add_warning("Worker/no valid shifts", message)
|
||||
@@ -4873,8 +4968,7 @@ class RotaBuilder(object):
|
||||
|
||||
def get_shifts_for_worker(self, worker_id):
|
||||
worker = self.get_worker_by_id(worker_id)
|
||||
shifts = [shift for shift in self.shifts if worker.site in shift.sites]
|
||||
return shifts
|
||||
return [shift for shift in self.shifts if self.is_worker_eligible_for_shift(worker, shift)]
|
||||
|
||||
def get_shifts_with_constraint(self, constraint) -> List[SingleShift]:
|
||||
return [shift for shift in self.shifts if shift.has_constraint(constraint)]
|
||||
@@ -5015,6 +5109,27 @@ class RotaBuilder(object):
|
||||
)
|
||||
])
|
||||
|
||||
def get_shift_sub_blocks(self, shift: SingleShift) -> List[List[str]]:
|
||||
if getattr(shift, "force_as_block_unless_nwd", None):
|
||||
val = shift.force_as_block_unless_nwd
|
||||
if isinstance(val, (list, tuple)) and all(isinstance(x, (list, tuple, set)) for x in val):
|
||||
return [list(x) for x in val]
|
||||
return [list(shift.days)]
|
||||
|
||||
# Check if it is a block shift globally or per-worker
|
||||
is_block = (
|
||||
shift.assign_as_block
|
||||
or shift.force_as_block
|
||||
or any(
|
||||
hasattr(worker, "assign_as_block_preferences")
|
||||
and getattr(worker, "assign_as_block_preferences", {}).get(shift.name, 0) != 0
|
||||
for worker in self.workers
|
||||
)
|
||||
)
|
||||
if is_block:
|
||||
return [list(shift.days)]
|
||||
return []
|
||||
|
||||
def get_all_locum_availability(self):
|
||||
return self.locum_availability_map
|
||||
|
||||
@@ -5066,8 +5181,46 @@ class RotaBuilder(object):
|
||||
|
||||
return group_workers
|
||||
|
||||
def is_worker_eligible_for_shift(self, worker: Worker, shift: SingleShift) -> bool:
|
||||
"""Returns True if the worker is eligible for the shift based on site and group options."""
|
||||
# 1. Group checks
|
||||
worker_groups = getattr(worker, "groups", set())
|
||||
if not isinstance(worker_groups, (set, list, tuple)):
|
||||
worker_groups = {worker_groups} if worker_groups else set()
|
||||
worker_groups_set = set(worker_groups)
|
||||
|
||||
# exclude_groups check: worker MUST NOT belong to any group in exclude_groups
|
||||
if getattr(shift, "exclude_groups", None):
|
||||
exclude_set = set(shift.exclude_groups)
|
||||
if worker_groups_set.intersection(exclude_set):
|
||||
return False
|
||||
|
||||
# Additive Site or Include Group check
|
||||
has_site = worker.site in shift.sites
|
||||
has_include_group = False
|
||||
if getattr(shift, "include_groups", None):
|
||||
include_set = set(shift.include_groups)
|
||||
if worker_groups_set.intersection(include_set):
|
||||
has_include_group = True
|
||||
|
||||
if getattr(shift, "include_groups", None):
|
||||
if not (has_site or has_include_group):
|
||||
return False
|
||||
else:
|
||||
if not has_site:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def get_workers_in_group(self, group_name: str) -> List[Worker]:
|
||||
"""Returns a list of all workers belonging to the specified group."""
|
||||
return [
|
||||
w for w in self.workers
|
||||
if group_name in getattr(w, "groups", set())
|
||||
]
|
||||
|
||||
def get_workers_for_shift(self, shift: SingleShift) -> List[Worker]:
|
||||
return [worker for worker in self.workers if worker.site in shift.sites]
|
||||
return [worker for worker in self.workers if self.is_worker_eligible_for_shift(worker, shift)]
|
||||
|
||||
def get_workers_total_fte(self) -> float:
|
||||
"""Does not take into account shift adjusted ftes"""
|
||||
|
||||
+159
-3
@@ -169,6 +169,42 @@ class MaxUniqueShiftsPerWeekBlockConstraint(BaseModel):
|
||||
)
|
||||
|
||||
|
||||
class ShiftStartDate(BaseModel):
|
||||
shift: str
|
||||
start_date: datetime.date
|
||||
|
||||
@field_validator("start_date", mode="before")
|
||||
@classmethod
|
||||
def coerce_date(cls, v):
|
||||
if isinstance(v, datetime.date):
|
||||
return v
|
||||
if isinstance(v, str):
|
||||
for fmt in ("%d/%m/%Y", "%d/%m/%y", "%Y-%m-%d"):
|
||||
try:
|
||||
return datetime.datetime.strptime(v, fmt).date()
|
||||
except Exception:
|
||||
continue
|
||||
raise ValueError(f"Cannot parse date: {v}")
|
||||
|
||||
|
||||
class ShiftEndDate(BaseModel):
|
||||
shift: str
|
||||
end_date: datetime.date
|
||||
|
||||
@field_validator("end_date", mode="before")
|
||||
@classmethod
|
||||
def coerce_date(cls, v):
|
||||
if isinstance(v, datetime.date):
|
||||
return v
|
||||
if isinstance(v, str):
|
||||
for fmt in ("%d/%m/%Y", "%d/%m/%y", "%Y-%m-%d"):
|
||||
try:
|
||||
return datetime.datetime.strptime(v, fmt).date()
|
||||
except Exception:
|
||||
continue
|
||||
raise ValueError(f"Cannot parse date: {v}")
|
||||
|
||||
|
||||
class Worker(BaseModel):
|
||||
name: str
|
||||
site: str
|
||||
@@ -214,11 +250,56 @@ class Worker(BaseModel):
|
||||
shift_fte_overrides: dict[str, int] = {} # Need checks to ensure shifts exist
|
||||
|
||||
exact_shifts: dict[str, int] = {} # Map shift_name to exact number of shifts
|
||||
groups: set[str] = set() # Groups that the worker belongs to
|
||||
shift_start_dates: list[ShiftStartDate] = [] # List of ShiftStartDate models
|
||||
shift_end_dates: list[ShiftEndDate] = [] # List of ShiftEndDate models
|
||||
|
||||
weekend_shift_target_number: float = 0.0
|
||||
avoid_shifts_on_dates: list[AvoidShiftOnDates] = []
|
||||
|
||||
@field_validator("shift_start_dates", mode="before")
|
||||
@classmethod
|
||||
def coerce_shift_start_dates(cls, v):
|
||||
if v is None:
|
||||
return []
|
||||
if isinstance(v, dict):
|
||||
res = []
|
||||
for shift, d in v.items():
|
||||
res.append(ShiftStartDate(shift=shift, start_date=d))
|
||||
return res
|
||||
if isinstance(v, list):
|
||||
res = []
|
||||
for item in v:
|
||||
if isinstance(item, ShiftStartDate):
|
||||
res.append(item)
|
||||
elif isinstance(item, dict):
|
||||
res.append(ShiftStartDate(**item))
|
||||
else:
|
||||
raise ValueError(f"Invalid ShiftStartDate item: {item}")
|
||||
return res
|
||||
raise ValueError("Must be a list or dictionary")
|
||||
|
||||
@field_validator("shift_end_dates", mode="before")
|
||||
@classmethod
|
||||
def coerce_shift_end_dates(cls, v):
|
||||
if v is None:
|
||||
return []
|
||||
if isinstance(v, dict):
|
||||
res = []
|
||||
for shift, d in v.items():
|
||||
res.append(ShiftEndDate(shift=shift, end_date=d))
|
||||
return res
|
||||
if isinstance(v, list):
|
||||
res = []
|
||||
for item in v:
|
||||
if isinstance(item, ShiftEndDate):
|
||||
res.append(item)
|
||||
elif isinstance(item, dict):
|
||||
res.append(ShiftEndDate(**item))
|
||||
else:
|
||||
raise ValueError(f"Invalid ShiftEndDate item: {item}")
|
||||
return res
|
||||
raise ValueError("Must be a list or dictionary")
|
||||
|
||||
model_config = ConfigDict(
|
||||
extra="allow",
|
||||
@@ -415,13 +496,88 @@ class Worker(BaseModel):
|
||||
self.proportion_rota_to_work = days_to_work / Rota.rota_days_length
|
||||
self.days_to_work = days_to_work
|
||||
|
||||
# Shift-specific start/end dates and adjusted FTEs
|
||||
self.calculated_shift_start_dates = {}
|
||||
self.calculated_shift_end_dates = {}
|
||||
self.proportion_rota_to_work_shifts = {}
|
||||
|
||||
# Convert list of ShiftStartDate/ShiftEndDate models/dicts to dictionary mapping for quick lookup
|
||||
start_dates_dict = {item.shift: item.start_date for item in getattr(self, "shift_start_dates", []) if hasattr(item, "shift")}
|
||||
end_dates_dict = {item.shift: item.end_date for item in getattr(self, "shift_end_dates", []) if hasattr(item, "shift")}
|
||||
|
||||
for shift in Rota.get_shifts():
|
||||
s_date = start_dates_dict.get(shift.name, self.start_date)
|
||||
if s_date is None:
|
||||
calc_s = self.calculated_start_date
|
||||
else:
|
||||
if isinstance(s_date, str):
|
||||
for fmt in ("%d/%m/%Y", "%d/%m/%y", "%Y-%m-%d"):
|
||||
try:
|
||||
s_date = datetime.datetime.strptime(s_date, fmt).date()
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
calc_s = s_date
|
||||
if calc_s < Rota.start_date:
|
||||
calc_s = Rota.start_date
|
||||
elif calc_s > Rota.rota_end_date:
|
||||
calc_s = Rota.rota_end_date
|
||||
|
||||
e_date = end_dates_dict.get(shift.name, self.end_date)
|
||||
if e_date is None:
|
||||
calc_e = self.calculated_end_date
|
||||
else:
|
||||
if isinstance(e_date, str):
|
||||
for fmt in ("%d/%m/%Y", "%d/%m/%y", "%Y-%m-%d"):
|
||||
try:
|
||||
e_date = datetime.datetime.strptime(e_date, fmt).date()
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
calc_e = e_date
|
||||
if calc_e > Rota.rota_end_date:
|
||||
calc_e = Rota.rota_end_date
|
||||
elif calc_e < Rota.start_date:
|
||||
calc_e = Rota.start_date
|
||||
|
||||
if calc_s >= calc_e:
|
||||
calc_s = Rota.rota_end_date
|
||||
calc_e = Rota.rota_end_date
|
||||
|
||||
self.calculated_shift_start_dates[shift.name] = calc_s
|
||||
self.calculated_shift_end_dates[shift.name] = calc_e
|
||||
|
||||
days_active = (calc_e - calc_s).days
|
||||
|
||||
# Subtract overlapping OOP days from the active period of this shift
|
||||
for item in self.oop:
|
||||
start_oop = item.start_date
|
||||
end_oop = item.end_date
|
||||
if isinstance(start_oop, datetime.date):
|
||||
start_oop_date = start_oop
|
||||
else:
|
||||
start_oop_date = datetime.datetime.strptime(start_oop, "%d/%m/%y").date()
|
||||
|
||||
if isinstance(end_oop, datetime.date):
|
||||
end_oop_date = end_oop
|
||||
else:
|
||||
end_oop_date = datetime.datetime.strptime(end_oop, "%d/%m/%y").date()
|
||||
|
||||
overlap_start = max(calc_s, start_oop_date)
|
||||
overlap_end = min(calc_e, end_oop_date)
|
||||
if overlap_start < overlap_end:
|
||||
days_active -= (overlap_end - overlap_start).days
|
||||
|
||||
self.proportion_rota_to_work_shifts[shift.name] = max(0.0, days_active / Rota.rota_days_length)
|
||||
|
||||
# We have to adjust the full time equivalent for people who CCT / leave the rota early
|
||||
self.fte_adj = self.fte * self.proportion_rota_to_work
|
||||
|
||||
self.fte_adj_shifts = {}
|
||||
if self.shift_fte_overrides:
|
||||
for shift, fte in self.shift_fte_overrides.items():
|
||||
self.fte_adj_shifts[shift] = fte * self.proportion_rota_to_work
|
||||
for shift in Rota.get_shifts():
|
||||
prop = self.proportion_rota_to_work_shifts.get(shift.name, self.proportion_rota_to_work)
|
||||
base_fte = self.shift_fte_overrides.get(shift.name, self.fte)
|
||||
self.fte_adj_shifts[shift.name] = base_fte * prop
|
||||
|
||||
|
||||
if self.fte_adj > 100:
|
||||
|
||||
+58
-1
@@ -163,7 +163,7 @@ def test_nwd_force_as_block_force_split():
|
||||
for worker in Rota.workers:
|
||||
shifts = Rota.get_worker_shift_list(worker)
|
||||
shifts_string = "".join([i if i != "" else "-" for i in shifts])
|
||||
for week in weeks_from_list(shifts_string[7 * 5 :]):
|
||||
for week in weeks_from_list(shifts_string[7 * 6 :]):
|
||||
assert week in ("-----ww", "dddddww")
|
||||
if worker.name == "worker1":
|
||||
assert shifts_string[: 7 * 5].count("ww-") == 4
|
||||
@@ -214,3 +214,60 @@ def test_nwd_testing():
|
||||
assert week == "dddddww"
|
||||
else:
|
||||
assert week == "-----ww"
|
||||
|
||||
|
||||
def test_force_as_block_unless_nwd_with_subblocks():
|
||||
# Test that a Fri/Sat/Sun shift can be split into Fri and Sat/Sun blocks
|
||||
# if a worker does not work on Fridays.
|
||||
Rota = setup_rota(weeks_to_rota=2)
|
||||
start_date = Rota.start_date
|
||||
|
||||
# worker1 does not work on Fridays (Fri is NWD)
|
||||
worker1 = Worker(
|
||||
name="worker1", site="group1", grade=1,
|
||||
nwds=[{"day": "Fri", "start_date": start_date, "end_date": start_date + datetime.timedelta(weeks=2)}],
|
||||
)
|
||||
# worker2 works normally
|
||||
worker2 = Worker(name="worker2", site="group1", grade=1)
|
||||
|
||||
Rota.add_workers((worker1, worker2))
|
||||
|
||||
# Fri/Sat/Sun shift, requiring 1 worker, forced as block unless NWD
|
||||
# We specify sub-blocks: [["Fri"], ["Sat", "Sun"]]
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group1",), name="weekend", length=12.5,
|
||||
days=["Fri", "Sat", "Sun"],
|
||||
workers_required=1,
|
||||
force_as_block_unless_nwd=[["Fri"], ["Sat", "Sun"]],
|
||||
),
|
||||
)
|
||||
|
||||
# We remove no valid shifts warning
|
||||
Rota.terminate_on_warning.remove("Worker/no valid shifts")
|
||||
|
||||
Rota.build_and_solve()
|
||||
assert Rota.results.solver.status == "ok"
|
||||
|
||||
# Check assignments:
|
||||
# Since worker1 has NWD on Friday:
|
||||
# - worker1 cannot work Friday.
|
||||
# - But worker1 can work Sat and Sun as a block!
|
||||
# So on Saturday and Sunday, worker1 should be assigned.
|
||||
# On Friday, worker2 should be assigned.
|
||||
for week in Rota.weeks:
|
||||
w1_fri = Rota.model.works[worker1.id, week, "Fri", "weekend"].value
|
||||
w1_sat = Rota.model.works[worker1.id, week, "Sat", "weekend"].value
|
||||
w1_sun = Rota.model.works[worker1.id, week, "Sun", "weekend"].value
|
||||
|
||||
w2_fri = Rota.model.works[worker2.id, week, "Fri", "weekend"].value
|
||||
w2_sat = Rota.model.works[worker2.id, week, "Sat", "weekend"].value
|
||||
w2_sun = Rota.model.works[worker2.id, week, "Sun", "weekend"].value
|
||||
|
||||
assert w1_fri == 0
|
||||
|
||||
# Either worker1 works Sat/Sun and worker2 works Fri (split block)
|
||||
# OR worker2 works Fri/Sat/Sun (full block) and worker1 works nothing
|
||||
is_split = (w1_sat == 1 and w1_sun == 1 and w2_fri == 1 and w2_sat == 0 and w2_sun == 0)
|
||||
is_full_w2 = (w1_sat == 0 and w1_sun == 0 and w2_fri == 1 and w2_sat == 1 and w2_sun == 1)
|
||||
assert is_split or is_full_w2
|
||||
+263
-1
@@ -1,8 +1,9 @@
|
||||
import pytest
|
||||
from rota_generator.shifts import InvalidShift, MaxShiftsPerWeekConstraint, NoWorkers, PostShiftConstraint, RotaBuilder, SingleShift, WarningTermination, WorkerRequirement, days
|
||||
from pydantic import ValidationError
|
||||
|
||||
import datetime
|
||||
from rota_generator.workers import NonWorkingDays, NotAvailableToWork, Worker, generate_not_available_to_works
|
||||
from rota_generator.workers import NonWorkingDays, NotAvailableToWork, Worker, generate_not_available_to_works, ShiftStartDate, ShiftEndDate
|
||||
from rota_generator.workers import WorkRequests
|
||||
from rota_generator.shifts import PreShiftConstraint
|
||||
|
||||
@@ -1869,3 +1870,264 @@ def test_html_export_worker_details_and_unavailable_reason():
|
||||
|
||||
assert 'data-exact-shifts=\'{"a": 1}\'' in html
|
||||
assert 'title=\' (2022-03-07) / START DATE: 2022-03-14\'' in html
|
||||
|
||||
|
||||
def test_worker_groups_selector():
|
||||
weeks_to_rota = 2
|
||||
start_date = datetime.date(2022, 3, 7)
|
||||
|
||||
Rota = RotaBuilder(
|
||||
start_date,
|
||||
weeks_to_rota=weeks_to_rota,
|
||||
)
|
||||
Rota.add_workers([
|
||||
Worker(name="worker1", site="group1", grade=1, fte=100, groups={"groupA", "groupB"}),
|
||||
Worker(name="worker2", site="group1", grade=1, fte=100, groups={"groupB", "groupC"}),
|
||||
])
|
||||
|
||||
workers_A = Rota.get_workers_in_group("groupA")
|
||||
workers_B = Rota.get_workers_in_group("groupB")
|
||||
workers_C = Rota.get_workers_in_group("groupC")
|
||||
workers_D = Rota.get_workers_in_group("groupD")
|
||||
|
||||
assert [w.name for w in workers_A] == ["worker1"]
|
||||
assert set(w.name for w in workers_B) == {"worker1", "worker2"}
|
||||
assert [w.name for w in workers_C] == ["worker2"]
|
||||
assert workers_D == []
|
||||
|
||||
|
||||
def test_shift_groups_overlapping_validation():
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
SingleShift(
|
||||
sites=("group1",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
include_groups={"pool1", "pool2"},
|
||||
exclude_groups={"pool2", "pool3"},
|
||||
)
|
||||
assert "include_groups and exclude_groups cannot overlap" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_shift_group_availability_inclusion():
|
||||
weeks_to_rota = 2
|
||||
start_date = datetime.date(2022, 3, 7)
|
||||
|
||||
Rota = RotaBuilder(
|
||||
start_date,
|
||||
weeks_to_rota=weeks_to_rota,
|
||||
)
|
||||
Rota.add_workers([
|
||||
Worker(name="worker1", site="group1", grade=1, fte=100),
|
||||
Worker(name="worker2", site="group2", grade=1, fte=100, groups={"senior"}),
|
||||
Worker(name="worker3", site="group2", grade=1, fte=100, groups={"junior"}),
|
||||
])
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group1",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
include_groups={"senior"},
|
||||
),
|
||||
)
|
||||
Rota.terminate_on_warning.remove("Worker/no valid shifts")
|
||||
Rota.build_and_solve()
|
||||
Rota.export_rota_to_html("test_shift_group_availability_inclusion", folder="tests")
|
||||
assert Rota.results.solver.status == "ok"
|
||||
|
||||
worker_shift_counts = {w.name: 0 for w in Rota.get_workers()}
|
||||
for week, day, shiftname in Rota.get_all_shiftname_combinations():
|
||||
for worker in Rota.get_workers():
|
||||
val = Rota.model.works[worker.id, week, day, shiftname].value
|
||||
if val is not None and val > 0.5:
|
||||
worker_shift_counts[worker.name] += 1
|
||||
|
||||
assert worker_shift_counts["worker1"] in (1, 2)
|
||||
assert worker_shift_counts["worker2"] in (1, 2)
|
||||
assert worker_shift_counts["worker3"] == 0
|
||||
|
||||
|
||||
def test_shift_group_availability_exclusion():
|
||||
weeks_to_rota = 2
|
||||
start_date = datetime.date(2022, 3, 7)
|
||||
|
||||
Rota = RotaBuilder(
|
||||
start_date,
|
||||
weeks_to_rota=weeks_to_rota,
|
||||
)
|
||||
Rota.add_workers([
|
||||
Worker(name="worker1", site="group1", grade=1, fte=100, groups={"senior"}),
|
||||
Worker(name="worker2", site="group1", grade=1, fte=100, groups={"junior"}),
|
||||
])
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group1",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
exclude_groups={"junior"},
|
||||
),
|
||||
)
|
||||
Rota.terminate_on_warning.remove("Worker/no valid shifts")
|
||||
Rota.build_and_solve()
|
||||
assert Rota.results.solver.status == "ok"
|
||||
|
||||
worker_shift_counts = {w.name: 0 for w in Rota.get_workers()}
|
||||
for week, day, shiftname in Rota.get_all_shiftname_combinations():
|
||||
for worker in Rota.get_workers():
|
||||
val = Rota.model.works[worker.id, week, day, shiftname].value
|
||||
if val is not None and val > 0.5:
|
||||
worker_shift_counts[worker.name] += 1
|
||||
|
||||
assert worker_shift_counts["worker1"] == 2
|
||||
assert worker_shift_counts["worker2"] == 0
|
||||
|
||||
|
||||
def test_shift_dependent_active_dates_constraints():
|
||||
weeks_to_rota = 10
|
||||
start_date = datetime.date(2022, 3, 7) # Mon
|
||||
|
||||
Rota = RotaBuilder(
|
||||
start_date,
|
||||
weeks_to_rota=weeks_to_rota,
|
||||
)
|
||||
worker1 = Worker(name="worker1", site="group1", grade=1, fte=100)
|
||||
worker2 = Worker(
|
||||
name="worker2", site="group1", grade=1, fte=100,
|
||||
shift_start_dates={"a": datetime.date(2022, 3, 28)},
|
||||
shift_end_dates={"a": datetime.date(2022, 4, 25)},
|
||||
)
|
||||
Rota.add_workers([worker1, worker2])
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group1",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
),
|
||||
)
|
||||
Rota.build_and_solve(options={"ratio": 0.0})
|
||||
assert Rota.results.solver.status == "ok"
|
||||
|
||||
# Verify assignments for worker2
|
||||
for week in range(1, weeks_to_rota + 1):
|
||||
val = Rota.model.works[worker2.id, week, "Mon", "a"].value
|
||||
assigned = val is not None and val > 0.5
|
||||
if week < 4 or week >= 8:
|
||||
assert not assigned, f"Worker 2 should not be assigned 'a' in week {week}"
|
||||
|
||||
|
||||
def test_shift_dependent_fte_target_scaling():
|
||||
weeks_to_rota = 10
|
||||
start_date = datetime.date(2022, 3, 7) # Mon
|
||||
|
||||
Rota = RotaBuilder(
|
||||
start_date,
|
||||
weeks_to_rota=weeks_to_rota,
|
||||
)
|
||||
worker1 = Worker(name="worker1", site="group1", grade=1, fte=100)
|
||||
worker2 = Worker(
|
||||
name="worker2", site="group1", grade=1, fte=100,
|
||||
shift_start_dates={"a": datetime.date(2022, 3, 28)},
|
||||
shift_end_dates={"a": datetime.date(2022, 4, 25)},
|
||||
)
|
||||
Rota.add_workers([worker1, worker2])
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group1",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
),
|
||||
)
|
||||
Rota.build_and_solve(options={"ratio": 0.0})
|
||||
assert Rota.results.solver.status == "ok"
|
||||
|
||||
# Targets check
|
||||
assert abs(worker1.shift_target_number["a"] - 7.14) < 0.1
|
||||
assert abs(worker2.shift_target_number["a"] - 2.86) < 0.1
|
||||
|
||||
|
||||
def test_shift_dependent_dates_validation_invalid_shift():
|
||||
weeks_to_rota = 2
|
||||
start_date = datetime.date(2022, 3, 7)
|
||||
|
||||
Rota = RotaBuilder(start_date, weeks_to_rota=weeks_to_rota)
|
||||
worker = Worker(
|
||||
name="worker1", site="group1", grade=1, fte=100,
|
||||
shift_start_dates=[ShiftStartDate(shift="non_existent", start_date=datetime.date(2022, 3, 7))]
|
||||
)
|
||||
Rota.add_worker(worker)
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group1",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
)
|
||||
)
|
||||
with pytest.raises(InvalidShift):
|
||||
Rota.build_workers()
|
||||
|
||||
|
||||
def test_shift_dependent_dates_validation_ineligible_warning():
|
||||
weeks_to_rota = 2
|
||||
start_date = datetime.date(2022, 3, 7)
|
||||
|
||||
Rota = RotaBuilder(start_date, weeks_to_rota=weeks_to_rota)
|
||||
worker = Worker(
|
||||
name="worker1", site="group1", grade=1, fte=100,
|
||||
shift_start_dates=[ShiftStartDate(shift="a", start_date=datetime.date(2022, 3, 7))]
|
||||
)
|
||||
Rota.add_worker(worker)
|
||||
Rota.add_shifts(
|
||||
SingleShift(
|
||||
sites=("group2",),
|
||||
name="a",
|
||||
length=12.5,
|
||||
days=("Mon",),
|
||||
workers_required=1,
|
||||
)
|
||||
)
|
||||
Rota.terminate_on_warning.remove("Worker/no valid shifts")
|
||||
Rota.build_workers()
|
||||
|
||||
warnings = Rota.get_warnings("Worker/ineligible shift date constraint")
|
||||
assert len(warnings) == 1
|
||||
assert "is not eligible to work it" in warnings[0][1]
|
||||
|
||||
|
||||
def test_shift_dependent_dates_model_definition():
|
||||
worker = Worker(
|
||||
name="worker1", site="group1", grade=1, fte=100,
|
||||
shift_start_dates=[{"shift": "a", "start_date": "2022-03-07"}],
|
||||
shift_end_dates=[{"shift": "a", "end_date": "2022-04-07"}]
|
||||
)
|
||||
assert len(worker.shift_start_dates) == 1
|
||||
assert isinstance(worker.shift_start_dates[0], ShiftStartDate)
|
||||
assert worker.shift_start_dates[0].shift == "a"
|
||||
assert worker.shift_start_dates[0].start_date == datetime.date(2022, 3, 7)
|
||||
|
||||
assert len(worker.shift_end_dates) == 1
|
||||
assert isinstance(worker.shift_end_dates[0], ShiftEndDate)
|
||||
assert worker.shift_end_dates[0].shift == "a"
|
||||
assert worker.shift_end_dates[0].end_date == datetime.date(2022, 4, 7)
|
||||
|
||||
|
||||
def test_parse_time_limit():
|
||||
from gen_proc import parse_time_limit
|
||||
assert parse_time_limit(3600) == 3600
|
||||
assert parse_time_limit("3600") == 3600
|
||||
assert parse_time_limit("10h") == 36000
|
||||
assert parse_time_limit("30m") == 1800
|
||||
assert parse_time_limit("45s") == 45
|
||||
assert parse_time_limit(" 1.5h ") == 5400
|
||||
with pytest.raises(ValueError):
|
||||
parse_time_limit("invalid")
|
||||
Reference in New Issue
Block a user