From 6ca3db86cbdffda3b1ff193d6312a9189ab3a11b Mon Sep 17 00:00:00 2001 From: Ross Date: Tue, 14 Jul 2026 21:38:59 +0100 Subject: [PATCH] add prtial shifts --- djangorota/rota/forms.py | 12 ++- djangorota/rota/models.py | 2 + gen_proc.py | 19 +++++ rota_generator/shifts.py | 48 ++++++++++++ rota_generator/workers.py | 161 +++++++++++++++++++++++++++++++++++++- test/test_workers.py | 138 +++++++++++++++++++++++++++++++- 6 files changed, 374 insertions(+), 6 deletions(-) diff --git a/djangorota/rota/forms.py b/djangorota/rota/forms.py index 635f97c..f653a7a 100644 --- a/djangorota/rota/forms.py +++ b/djangorota/rota/forms.py @@ -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,8 @@ 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)") @@ -97,6 +102,8 @@ 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", @@ -151,6 +158,8 @@ 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", @@ -272,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") diff --git a/djangorota/rota/models.py b/djangorota/rota/models.py index 7102d79..3dbb47a 100644 --- a/djangorota/rota/models.py +++ b/djangorota/rota/models.py @@ -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: diff --git a/gen_proc.py b/gen_proc.py index 54d03b9..c6ca1d5 100644 --- a/gen_proc.py +++ b/gen_proc.py @@ -504,6 +504,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, @@ -537,6 +555,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) diff --git a/rota_generator/shifts.py b/rota_generator/shifts.py index 1de76d5..c4f1b76 100644 --- a/rota_generator/shifts.py +++ b/rota_generator/shifts.py @@ -3738,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) @@ -4323,6 +4339,38 @@ class RotaBuilder(object): "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" diff --git a/rota_generator/workers.py b/rota_generator/workers.py index 4ab29cb..b8906b2 100644 --- a/rota_generator/workers.py +++ b/rota_generator/workers.py @@ -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 @@ -215,11 +251,55 @@ class Worker(BaseModel): 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", @@ -416,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: diff --git a/test/test_workers.py b/test/test_workers.py index 732185d..932e099 100644 --- a/test/test_workers.py +++ b/test/test_workers.py @@ -3,7 +3,7 @@ from rota_generator.shifts import InvalidShift, MaxShiftsPerWeekConstraint, NoWo 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 @@ -1984,4 +1984,138 @@ def test_shift_group_availability_exclusion(): worker_shift_counts[worker.name] += 1 assert worker_shift_counts["worker1"] == 2 - assert worker_shift_counts["worker2"] == 0 \ No newline at end of file + 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) \ No newline at end of file