ross/proc-rota
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add max days per week block constraint to Worker class

Browse Source 
- Introduced a new method `add_max_days_per_week_block_constraint` to the Worker class.
- This method allows for the addition of constraints that limit the number of days a worker can be assigned within a specified sliding block of weeks.
- The method accepts parameters for maximum days, the week block size, and optional dates to ignore.
This commit is contained in:
Ross
2026-06-02 13:50:31 +01:00
parent 4cad1158c6
commit e288585cd3
3 changed files with 114 additions and 530 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gen_cons.py
+110 -143
View File
@@ -2,6 +2,7 @@ from pyomo.opt.results.container import ignore
from collections import defaultdict
import datetime
import os
import re
import sys
import time
from rota_generator.shifts import MaxShiftsPerWeekConstraint, NoWorkers, PostShiftConstraint, PreShiftConstraint, RotaBuilder, SingleShift, WorkerRequirement, days, RotaConstraintOptions, MaxShiftsPerWeekBlockConstraint
@@ -38,6 +39,100 @@ def _parse_sheet_date(value):
return None
return parsed.date()
def _parse_worked_target_cell(value):
"""Parse values like '5 (6.88)' into (worked, target)."""
if pd.isna(value):
return None
text = str(value).strip()
if not text or text.lower() in {"nan", "-"}:
return None
match = re.match(r"^\s*(-?\d+(?:\.\d+)?)\s*\(([^)]+)\)\s*$", text)
if match:
try:
worked = float(match.group(1).strip())
target = float(match.group(2).strip())
return worked, target
except Exception:
return None
# Fallback for cells that contain only one number.
try:
target = float(text)
return 0.0, target
except Exception:
return None
def _load_priors_from_running_totals(running_totals_df):
"""Extract previous shifts map from Running totals tab.
Reads only oncall nights, twilights and weekend A values.
"""
priors_map = {}
if running_totals_df is None or running_totals_df.empty:
return priors_map
# Find the header row that starts with 'Worker'.
header_row_index = None
for i in range(len(running_totals_df)):
first_cell = str(running_totals_df.iloc[i, 0]).strip().lower()
if first_cell == "worker":
header_row_index = i
break
if header_row_index is None:
return priors_map
header_values = [
str(c).strip().lower() if not pd.isna(c) else ""
for c in running_totals_df.iloc[header_row_index].tolist()
]
def _find_col(header_name):
for idx, col in enumerate(header_values):
if col == header_name:
return idx
return None
worker_col = _find_col("worker")
oncall_col = _find_col("oncall night")
twilight_col = _find_col("twilight")
weekend_a_col = _find_col("weekend a")
if worker_col is None:
return priors_map
for i in range(header_row_index + 1, len(running_totals_df)):
row = running_totals_df.iloc[i]
worker_raw = row.iloc[worker_col] if worker_col < len(row) else None
worker = str(worker_raw).strip() if not pd.isna(worker_raw) else ""
if not worker or worker.lower() in {"nan", "total"}:
continue
prev = {}
if oncall_col is not None and oncall_col < len(row):
parsed = _parse_worked_target_cell(row.iloc[oncall_col])
if parsed is not None:
prev["oncall"] = parsed
if twilight_col is not None and twilight_col < len(row):
parsed = _parse_worked_target_cell(row.iloc[twilight_col])
if parsed is not None:
prev["twilight"] = parsed
if weekend_a_col is not None and weekend_a_col < len(row):
parsed = _parse_worked_target_cell(row.iloc[weekend_a_col])
if parsed is not None:
prev["weekend"] = parsed
if prev:
priors_map[worker] = prev
return priors_map
def extract_leave_and_rota_from_calender(calender_df):
"""
Extract leave and rota assignments from calender_df.
@@ -168,6 +263,7 @@ def load_workers():
days_df = xls["Days"]
calender_df = xls["Calendar"]
rota_df = xls["Rota"]
running_totals_df = xls.get("Running totals")
rota_b_path = ROTA_B_PATH
@@ -196,53 +292,9 @@ def load_workers():
print(rota_b_assignments)
# You can now use rota_b_assignments as needed
# --- Load prior allocations (targets) from CSV ---
priors_path = "cons/sep2025priors.csv"
priors_map = {}
try:
priors_df = pd.read_csv(priors_path)
for _, prow in priors_df.iterrows():
initial = str(prow.get("Worker", "")).strip()
if not initial or initial.lower() == "nan":
continue
prev = {}
# expected columns: oncall, twilight, weekend
for col in ("oncall", "twilight", "weekend"):
raw = prow.get(col, "")
if pd.isna(raw) or raw == "" or str(raw).strip() in ("-", "nan"):
continue
s = str(raw).strip()
# Expect formats like: '4 (2.50)' or '3 (2.86)'
worked = 0.0
allocated = 0.0
if "(" in s:
try:
worked_part = s.split("(")[0].strip()
worked = float(worked_part)
except Exception:
worked = 0.0
try:
inner = s.split("(", 1)[1].split(")", 1)[0]
allocated = float(inner.strip())
except Exception:
allocated = worked
else:
try:
allocated = float(s)
except Exception:
continue
worked = 0.0
# store as tuple (worked, allocated) as expected by RotaBuilder
prev[col] = (worked, allocated)
if prev:
priors_map[initial] = prev
logger.debug("priors_map: {}", priors_map)
except FileNotFoundError:
print(f"Prior allocations file not found: {priors_path}")
except Exception as e:
print(f"Error reading prior allocations: {e}")
# --- Load prior allocations from requests.ods -> Running totals tab ---
priors_map = _load_priors_from_running_totals(running_totals_df)
logger.debug("priors_map from Running totals: {}", priors_map)
# --- Example: Print the first few rows of each sheet ---
#print("Days sheet:")
@@ -272,101 +324,7 @@ 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
logger.debug("priors_map after per-day merge: {}", priors_map)
except FileNotFoundError:
print(f"Per-day prior file not found: {priordays_path}")
except Exception as e:
print(f"Error reading per-day priors: {e}")
logger.debug("Using Running totals priors for oncall/twilight/weekend A only")
workers = []
for worker_day in workers_days:
@@ -452,6 +410,7 @@ def load_workers():
shift_name="weekend b",
)
w.add_hard_day_exclusion({"Sun", "Mon"})
if w.site == "rota a":
w.prefer_multi_shift_together = 10
@@ -459,10 +418,12 @@ def load_workers():
#w.add_force_assign_with("twilight", "oncall")
w.add_force_assign_with("weekend", "oncall")
if w.name in ("DS","RG"):
if w.name in ("DS","RG","MOB","TB"):
w.add_hard_day_dependency({"Fri", "Sat", "Sun"}, ignore_dates=[datetime.date(2026, 5, 2)])
w.max_days_worked_per_week = 3
else:
w.max_days_worked_per_week = 2
w.add_max_days_per_week_block_constraint(max_days=2, week_block=2)
w.add_hard_day_dependency({"Fri", "Sun"})
w.add_hard_day_exclusion({"Sat", "Sun"})
else:
@@ -535,6 +496,8 @@ def main(
balance_offset=1,
constraints=[
PostShiftConstraint(days=1, start_date="2025-11-17", ignore_shifts=["oncall"]),
#PreShiftConstraint(days=1),
#PostShiftConstraint(days=3),
],
display_char="a",
#force_assign_with=["oncall"]
@@ -548,6 +511,10 @@ def main(
days=days[5:],
balance_offset=2,
display_char="b",
constraints=[
PreShiftConstraint(days=2),
PostShiftConstraint(days=2),
],
),
SingleShift(
sites=(