From a347ad7777c4f2bfbe3d5c16a9188d10c8a284a9 Mon Sep 17 00:00:00 2001 From: Ross Date: Wed, 17 Dec 2025 11:27:16 +0000 Subject: [PATCH] . --- gen_cons.py | 96 ++++++++++++++++++++++++++++++++++++++++ rota_generator/shifts.py | 89 ++++++++++++++++++++++++++++++++++--- 2 files changed, 178 insertions(+), 7 deletions(-) diff --git a/gen_cons.py b/gen_cons.py index eafdc19..eddd348 100644 --- a/gen_cons.py +++ b/gen_cons.py @@ -259,6 +259,100 @@ def load_workers(): assignments, assignments_by_worker = extract_shift_assignments_from_rota(rota_df) print(assignments) + # --- Load per-day prior worked data (Sat/Sun) from sep2025priordays.csv for Rota A workers --- + # Build set of initials who are on Rota A + rota_a_initials = set() + for wd in workers_days: + name = wd["name"] + initial = wd["initial"] + if rota_data.get(name, "") == "rota a": + rota_a_initials.add(initial) + + priordays_path = "cons/sep2025priordays.csv" + try: + import csv + + perday_counts = defaultdict(lambda: defaultdict(int)) + with open(priordays_path, newline="", encoding="utf-8") as f: + reader = csv.reader(f) + for row in reader: + if len(row) < 4: + continue + day = row[2].strip() + if day not in ("Sat", "Sun"): + continue + # fields from index 3 onwards are assignments; some may be empty + for field in row[3:]: + if not field: + continue + val = field.strip() + if not val: + continue + # entries may be like: RK (oncall, weekend) or CK (weekend b) + import re + + m = re.match(r"\s*([A-Za-z0-9]+)\s*\(([^)]*)\)", val) + if not m: + # sometimes multiple assignments are concatenated in a single cell separated by commas + parts = [p.strip() for p in re.split(r",\s*", val) if p.strip()] + for p in parts: + m2 = re.match(r"\s*([A-Za-z0-9]+)\s*\(([^)]*)\)", p) + if m2: + initial = m2.group(1).strip() + paren = m2.group(2).lower() + if initial in rota_a_initials and "weekend" in paren: + perday_counts[initial][day] += 1 + continue + + initial = m.group(1).strip() + paren = m.group(2).lower() + if initial in rota_a_initials and "weekend" in paren: + perday_counts[initial][day] += 1 + + # Merge per-day counts into priors_map under 'weekend' per-day keys + for initial, daymap in perday_counts.items(): + prev = priors_map.get(initial, {}) + existing = prev.get("weekend") + # If existing is a tuple, convert to dict preserving overall tuple under '__all__' + if existing and not isinstance(existing, dict): + prev["weekend"] = {"__all__": existing} + existing = prev["weekend"] + + for day, worked_count in daymap.items(): + # Determine allocated value for this per-day entry + allocated_day = worked_count + if existing: + # prefer an explicit per-day allocated if present + if isinstance(existing, dict) and day in existing: + try: + allocated_day = float(existing[day][1]) + except Exception: + allocated_day = worked_count + elif isinstance(existing, dict) and "__all__" in existing: + try: + overall_alloc = float(existing["__all__"][1]) + allocated_day = overall_alloc / 2.0 + except Exception: + allocated_day = worked_count + elif isinstance(existing, tuple): + try: + allocated_day = float(existing[1]) / 2.0 + except Exception: + allocated_day = worked_count + + # ensure structure + if "weekend" not in prev or not isinstance(prev["weekend"], dict): + prev.setdefault("weekend", {}) + + prev["weekend"][day] = (float(worked_count), float(allocated_day)) + + if prev: + priors_map[initial] = prev + except FileNotFoundError: + print(f"Per-day prior file not found: {priordays_path}") + except Exception as e: + print(f"Error reading per-day priors: {e}") + workers = [] for worker_day in workers_days: worker = worker_day["name"] @@ -393,6 +487,8 @@ def main( max_days_per_week_block=[(4, 2), (4, 4)], balance_weekend_days=True, #weekend_day_hard_max_deviation=2, + balance_weekend_days_use_previous_shifts=True, + ), ) diff --git a/rota_generator/shifts.py b/rota_generator/shifts.py index 04899cd..b3b135b 100644 --- a/rota_generator/shifts.py +++ b/rota_generator/shifts.py @@ -351,6 +351,7 @@ class RotaConstraintOptions(BaseModel): balance_weekend_days: bool = Field(False, description="Balance weekend days (Sat/Sun) separately") balance_weekend_days_weight: int = Field(5, description="Objective weight for balancing weekend days per day") balance_weekend_days_quadratic: bool = Field(False, description="Use quadratic penalty to balance weekend days per day") + balance_weekend_days_use_previous_shifts: bool = Field(False, description="When true, adjust per-day weekend (Sat/Sun) targets using worker.previous_shifts (allocated-worked) distributed per shift day") weekend_day_hard_max_deviation: Optional[int] = Field( None, description=( @@ -2194,10 +2195,36 @@ class RotaBuilder(object): if self.use_previous_shifts: if shift.name in worker.previous_shifts: - worked, allocated = worker.previous_shifts[shift.name] - target_shifts = ( - target_shifts + float(allocated) - float(worked) - ) + entry = worker.previous_shifts[shift.name] + # entry may be a tuple (worked, allocated) or a dict of per-day tuples + if isinstance(entry, dict): + # prefer overall tuple if present + if "__all__" in entry: + try: + worked = float(entry["__all__"][0]) + allocated = float(entry["__all__"][1]) + target_shifts = target_shifts + allocated - worked + except Exception: + pass + else: + # sum any per-day entries + try: + total_worked = 0.0 + total_alloc = 0.0 + for v in entry.values(): + total_worked += float(v[0]) + total_alloc += float(v[1]) + target_shifts = target_shifts + total_alloc - total_worked + except Exception: + pass + else: + try: + worked, allocated = entry + target_shifts = ( + target_shifts + float(allocated) - float(worked) + ) + except Exception: + pass if self.use_shift_balance_extra: if shift.name in worker.shift_balance_extra: @@ -2533,6 +2560,41 @@ class RotaBuilder(object): if day_name in shift.days ) + # Optionally adjust per-day weekend targets using previous_shifts + if self.use_previous_shifts and self.constraint_options_model.balance_weekend_days_use_previous_shifts: + # For each shift that includes this day, if previous_shifts contains an entry + # add (allocated - worked) / len(shift.days) to the per-day target + prev_adj = 0.0 + for shift in self.get_shifts(): + if day_name in shift.days and shift.name in getattr(worker, "previous_shifts", {}): + entry = worker.previous_shifts.get(shift.name) + # support per-day nested dicts: { 'Sat': (worked, allocated), 'Sun': (...) } + if isinstance(entry, dict): + # prefer explicit day entry if present + if day_name in entry: + try: + worked = float(entry[day_name][0]) + allocated = float(entry[day_name][1]) + prev_adj += (allocated - worked) + except Exception: + continue + # fallback: if an overall tuple was preserved under '__all__', use distributed value + elif "__all__" in entry: + try: + worked = float(entry["__all__"][0]) + allocated = float(entry["__all__"][1]) + prev_adj += (allocated - worked) / max(1, len(shift.days)) + except Exception: + continue + else: + try: + worked = float(entry[0]) + allocated = float(entry[1]) + prev_adj += (allocated - worked) / max(1, len(shift.days)) + except Exception: + continue + weekend_day_shift_target_number = weekend_day_shift_target_number + prev_adj + self.model.constraints.add( self.model.weekend_day_shift_count_t1[worker.id, day_name] - self.model.weekend_day_shift_count_t2[worker.id, day_name] @@ -4904,12 +4966,25 @@ class RotaBuilder(object): for shift in self.get_shifts(): prev = getattr(worker, "previous_shifts", {}) or {} if shift.name in prev: - worked, allocated = prev[shift.name] + entry = prev[shift.name] + # entry may be a tuple (worked, allocated) or a dict with per-day entries + if isinstance(entry, dict): + # prefer a day-specific entry if available (variable `d` holds the day name in scope) + if d in entry: + worked, allocated = entry[d] + elif "__all__" in entry: + worked, allocated = entry["__all__"] + else: + # fallback: sum any tuple values found in the dict + worked = sum(v[0] for v in entry.values() if isinstance(v, (list, tuple))) + allocated = sum(v[1] for v in entry.values() if isinstance(v, (list, tuple))) + else: + worked, allocated = entry try: adj = float(allocated) - float(worked) except Exception: - adj = 0.0 - prior_map[shift.name] = {"worked": worked, "allocated": allocated, "adjustment": adj} + adj = 0 + prior_map[shift.name] = dict(worked=worked, allocated=allocated, adjustment=adj) if adj != 0: adjustments_list.append(f"{shift.name}:{adj:+g}")