Compare commits

...
8 Commits
10 changed files with 900 additions and 144 deletions
+43 -1
View File
@@ -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
+13
View File
@@ -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")
+7
View File
@@ -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
View File
@@ -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
View File
@@ -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)
+2 -2
View File
@@ -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
View File
@@ -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
View File
@@ -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
View File
@@ -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
View File
@@ -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")