.
This commit is contained in:
Ross
+96
@@ -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,
|
||||
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@@ -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}")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user