Compare commits

...
13 Commits
5 changed files with 1037 additions and 81 deletions
+200 -55
View File
@@ -1,9 +1,10 @@
from pyomo.opt.results.container import ignore
from collections import defaultdict
import datetime
import os
import sys
import time
from rota_generator.shifts import MaxShiftsPerWeekConstraint, NoWorkers, PostShiftConstraint, PreShiftConstraint, RotaBuilder, SingleShift, WorkerRequirement, days
from rota_generator.shifts import MaxShiftsPerWeekConstraint, NoWorkers, PostShiftConstraint, PreShiftConstraint, RotaBuilder, SingleShift, WorkerRequirement, days, RotaConstraintOptions
import typer
import subprocess
import pandas as pd
@@ -38,11 +39,11 @@ def extract_leave_and_rota_from_calender(calender_df):
# Get rota data from the second row (index 1), starting from column 5 (index 4)
# Extract the rota data for each worker from the first data row (index 0), columns 5 onwards
rota_row = calender_df.iloc[0]
rota_data = {}
for i, worker in enumerate(worker_names):
assignment = str(rota_row.iloc[5 + i]).strip() if len(rota_row) > 5 + i else ""
rota_data[worker] = f"rota {assignment.lower()}"
#rota_row = calender_df.iloc[0]
#rota_data = {}
#for i, worker in enumerate(worker_names):
# assignment = str(rota_row.iloc[5 + i]).strip() if len(rota_row) > 5 + i else ""
# #rota_data[worker] = f"rota {assignment.lower()}"
leave_requests = []
work_requests = []
@@ -55,6 +56,9 @@ def extract_leave_and_rota_from_calender(calender_df):
date = datetime.datetime.strptime(date_str, "%Y-%m-%d").date()
except Exception:
continue # skip rows with invalid date
if date < datetime.date(2026, 2, 16):
continue # skip rows before rota start date
for i, worker in enumerate(worker_names):
if not worker or worker.lower() == "nan":
continue
@@ -77,7 +81,7 @@ def extract_leave_and_rota_from_calender(calender_df):
#print(leave_requests)
return leave_requests, work_requests, rota_data
return leave_requests, work_requests#, rota_data
def extract_shift_assignments_from_rota(rota_df):
"""
@@ -91,6 +95,7 @@ def extract_shift_assignments_from_rota(rota_df):
shifts = ["twilight", "oncall"]
assignments = []
skipped_rows = 0
for idx, row in rota_df.iloc[2:].iterrows():
week_start_str = str(row.iloc[0]).strip().split(" ")[0] # Get the week start date from the first column (index 0)
try:
@@ -99,6 +104,10 @@ def extract_shift_assignments_from_rota(rota_df):
print(f"Invalid date format in row {idx}: {week_start_str}")
continue # skip rows with invalid date
if week_start < datetime.date(2026, 2, 16):
skipped_rows += 1
continue # skip rows before rota start date
for i, day in enumerate(days):
for j, shift in enumerate(shifts):
if i < 5:
@@ -122,7 +131,7 @@ def extract_shift_assignments_from_rota(rota_df):
shift = "weekend"
assignments.append({
"week_start": week_start,
"week": idx-1, # 1-indexed week number
"week": idx-1-skipped_rows, # 1-indexed week number
"day": day,
"date": week_start + datetime.timedelta(days=i),
"shift": shift,
@@ -140,24 +149,24 @@ def extract_shift_assignments_from_rota(rota_df):
def load_workers():
# Path to the ODS file
ods_path = "cons/Oncall rota requests + New rota (first draft) - September 2025.ods"
ods_path = "cons/CONSULTANT TWILIGHT & ONCALL ROTA.ods"
# Read all sheets
xls = pd.read_excel(ods_path, engine="odf", sheet_name=None)
# Extract sheets
days_df = xls["Days"]
calender_df = xls["Calender"]
calender_df = xls["Calendar"]
rota_df = xls["Rota"]
rota_b_path = "cons/Draft Rota B Sept 25-Feb 26.xlsx"
rota_b_path = "cons/Rota B Feb -July 2026.xlsx"
rota_b_df = pd.read_excel(rota_b_path, engine="openpyxl", sheet_name="Sheet1", header=None)
# Extract rota_b assignments: date in column A, worker in column B
rota_b_assignments = defaultdict(list)
for idx, row in rota_b_df.iterrows():
date_val = row.iloc[0]
worker_val = row.iloc[1] if len(row) > 1 else None
worker_val = row.iloc[1].upper() if len(row) > 1 else None
if pd.isna(date_val) or pd.isna(worker_val):
continue
try:
@@ -171,11 +180,60 @@ 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}")
# --- Example: Print the first few rows of each sheet ---
#print("Days sheet:")
#print(days_df.head())
weekday_cols = ["Mon", "Tue", "Wed", "Thu", "Fri"]
workers_days = []
rota_data = {}
for idx, row in days_df.iloc[7:].iterrows():
# Extract initials from the name column (assumed to be in brackets at the end)
text = str(row.iloc[1]).strip()
@@ -188,38 +246,135 @@ def load_workers():
continue
availability = {}
requests = {}
rota_data[name] = f"rota {str(row.iloc[2]).strip().lower()}"
for i, day in enumerate(weekday_cols):
val = row.iloc[2 + i]
val = row.iloc[3 + i]
# You can adjust the logic below depending on your marking scheme (e.g. "Y", "Yes", "1", etc.)
available = not ("no" in str(val).strip().lower())
available = not ("no" in str(val).strip().lower() or str(val).strip().lower() == "yes*")
availability[day] = available
workers_days.append({"initial": initial, "name": name, "availability": availability, "requests": requests})
leave_requests, work_requests, rota_data = extract_leave_and_rota_from_calender(calender_df)
leave_requests, work_requests = extract_leave_and_rota_from_calender(calender_df)
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}")
workers = []
for worker_day in workers_days:
worker = worker_day["name"]
initial = worker_day["initial"]
start_date = "2025-09-01" # Default start date, can be adjusted later
start_date = "2026-02-16" # Default start date, can be adjusted later
if initial == "ND":
start_date = "2025-09-29" # Non-working day worker starts later
if initial == "MF":
start_date = "2026-01-15" # Medical leave worker starts earlier
if initial == "AA":
start_date = "2025-11-17" # Absent worker starts later
#if initial == "ND":
# start_date = "2025-09-29" # Non-working day worker starts later
#if initial == "MF":
# start_date = "2026-01-15" # Medical leave worker starts earlier
#if initial == "AA":
# start_date = "2025-11-17" # Absent worker starts later
end_date = None
if initial == "L1":
end_date = "2025-11-17" # L1 worker has a leave request
elif initial == "L2":
end_date = "2025-09-29" # L2 worker has a leave request
#if initial == "L1":
# end_date = "2025-11-17" # L1 worker has a leave request
#elif initial == "L2":
# end_date = "2025-09-29" # L2 worker has a leave request
non_working_days = []
for day, available in worker_day["availability"].items():
@@ -246,6 +401,7 @@ def load_workers():
) for req in work_requests if req["worker"] == worker
],
nwds=non_working_days,
previous_shifts=priors_map.get(initial, {}),
)
#if initial == "L1":
@@ -255,8 +411,6 @@ def load_workers():
# reason="even out shifts",
# )]
if initial == "ND":
print(w.not_available_to_work)
if initial in assignments_by_worker:
# Add shift assignments for this worker
@@ -291,10 +445,10 @@ def load_workers():
w.add_force_assign_with("weekend", "oncall")
if w.name in ("DS",):
w.add_hard_day_dependency({"Fri", "Sat", "Sun"}, start_date="2025-11-17")
w.add_hard_day_dependency({"Fri", "Sat", "Sun"}, ignore_dates=[datetime.date(2026, 5, 2)])
else:
w.add_hard_day_dependency({"Fri", "Sun"}, start_date="2025-11-17")
w.add_hard_day_exclusion({"Sat", "Sun"}, start_date="2025-11-17")
w.add_hard_day_dependency({"Fri", "Sun"})
w.add_hard_day_exclusion({"Sat", "Sun"})
#if w.name == "TB":
@@ -316,7 +470,7 @@ def main(
solve: bool = True,
time_to_run: int = 60 * 60,
ratio: float = 0.001,
start_date: datetime.datetime = "2025-09-01",
start_date: datetime.datetime = "2026-02-16",
weeks: int = 24,
bom: int = 1,
):
@@ -327,16 +481,21 @@ def main(
rota_start_date,
weeks_to_rota=weeks,
balance_offset_modifier=bom,
name="cons rota test",
use_previous_shifts=True,
name="cons rota feb 2026",
allow_force_assignment_with_leave_conflict=True,
constraint_options=RotaConstraintOptions(
balance_weekends=True,
max_shifts_per_week=6,
max_days_per_week_block=[(4, 2), (4, 4)],
#balance_weekend_days=True,
balance_weekend_days_quadratic=True,
weekend_day_hard_max_deviation=2,
balance_weekend_days_use_previous_shifts=True,
balance_weekend_days_weight=500000
),
)
# Rota = RotaBuilder(start_date, weeks_to_rota=20, balance_offset_modifier=1)
Rota.constraint_options["balance_weekends"] = True
#Rota.constraint_options["max_weekend_frequency"] = 2
Rota.constraint_options["max_shifts_per_week"] = 6
Rota.constraint_options["max_days_worked_per_week"] = 3
# Rota.constraint_options["avoid_st2_first_month"] = True
Rota.constraint_options["max_days_per_week_block"] = [(4, 2), (6, 3)] # (max_days, week_block)
#Rota.enable_unavailabilities = False
@@ -349,12 +508,8 @@ def main(
balance_offset=2,
assign_as_block=False,
constraints=[
PreShiftConstraint(days=2, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun")),
PreShiftConstraint(days=2, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun"), exclude_dates=["2026-04-06", "2026-05-04", "2026-05-25"]),
PostShiftConstraint(days=1, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun"), allow_self=True),
#{
# "name": "max_shifts_per_week",
# "options": 3,
#}
],
),
SingleShift(
@@ -365,9 +520,6 @@ def main(
days=days[5:],
balance_offset=1,
constraints=[
#{"name": "pre", "options": 2},
#{"name": "post", "options": 1},
PostShiftConstraint(days=1, start_date="2025-11-17", ignore_shifts=["oncall"]),
],
display_char="a",
@@ -380,10 +532,6 @@ def main(
length=12.5,
days=days[5:],
balance_offset=2,
#constraint=[
# {"name": "pre", "options": 1},
# {"name": "post", "options": 1},
#],
display_char="b",
),
SingleShift(
@@ -398,11 +546,8 @@ def main(
days=days[:5],
balance_offset=1,
constraints=[
#PreShiftConstraint(days=1, start_date="2025-11-17", ignore_shifts=["oncall"]),
PreShiftConstraint(days=1, start_date="2025-11-17", ignore_shifts=["oncall"], exclude_days=("Sat", "Sun")),
#PostShiftConstraint(days=1, start_date="2025-11-17", ignore_shifts=["oncall", "weekend a", "weekend b"], include_weekends=False),
MaxShiftsPerWeekConstraint(max_shifts=1),
#{"name": "post", "options": 1},
],
),
)
+201 -11
View File
@@ -173,14 +173,17 @@ function generateExtra() {
worker_td.get(0).dataset.shiftDiffSummed = summed_shift_diff.toPrecision(2);
worker_td.after(`<div class='worker-summary auto-generated'>
<span>Total shifts: ${total_shifts}, </span>
<span>Weekends: ${weekends_worked}, </span>
<span>Nights: ${nights_worked}, </span>
<span>Bank holidays: ${bank_holidays_worked}, </span>
<span class='shift-diff-span'>Summed shift diff: ${summed_shift_diff.toPrecision(2)}, </span>
</div>`)
// compute separate Sat / Sun counts for this worker
let sat_count = 0;
let sun_count = 0;
for (let i = 1; i <= weeks; i++) {
let sat = $(tr).find(`[data-week='${i}'][data-day='Sat']`).attr("data-shift");
let sun = $(tr).find(`[data-week='${i}'][data-day='Sun']`).attr("data-shift");
if (sat && sat !== "") sat_count += 1;
if (sun && sun !== "") sun_count += 1;
}
// worker-summary will be inserted after we compute prior-adjust badges (sat/sun)
if (summed_shift_diff >= 1) {
worker_td.addClass("large-shift-diff-pos")
@@ -235,8 +238,17 @@ function generateExtra() {
// }
summary_button = $("<button class='auto-generated'>Toggle Summary</button>").click(() => {
$(table).find(".rota-day").toggleClass("hidden");
$(table).find(".worker-summary").toggle();
let $rotaDays = $(table).find(".rota-day");
let $workerSummaries = $(table).find(".worker-summary");
// determine current visibility of rota days
let rotaVisible = $rotaDays.length ? !$rotaDays.first().hasClass('hidden') : true;
if (rotaVisible) {
$rotaDays.addClass('hidden');
$workerSummaries.show();
} else {
$rotaDays.removeClass('hidden');
$workerSummaries.hide();
}
});
@@ -261,10 +273,13 @@ if (shifts.size < 1) {
oshifts = Array.from(shifts).sort()
$("body").append("<div><button id='show-colour-diff'>Show colour diff</button><table class='tsummary'></table></div>")
$("body").append("<div><button id='show-colour-diff'>Show colour diff</button><button id='toggle-prior-adjust' style='margin-left:10px;'>Toggle Prior Adjust</button><table class='tsummary'></table></div>")
$("#show-colour-diff").on("click", (evt) => {
$("table.tsummary").find(`td.target-assigned`).toggleClass("no-colour");
});
$("#toggle-prior-adjust").on("click", (evt) => {
$("table.tsummary").find(`td.prior-adjust, th.prior-adjust-col`).toggle();
});
$("table.tsummary").append("<tr class='header'></tr>");
h = $("table.tsummary tr.header");
@@ -279,6 +294,9 @@ oshifts.forEach((s) => {
});
// Column for prior-adjustment (hidden by default)
h.append(`<th class='prior-adjust-col' style='display:none'>Prior Adjust</th>`);
$(".table-div .worker-row .worker").each((n, tr) => {
@@ -320,6 +338,153 @@ $(".table-div .worker-row .worker").each((n, tr) => {
}
})
// Build per-shift prior-adjust spans from data-previous-shifts
// Try both the raw attribute and jQuery-parsed data for robustness
let prev_attr = jtr.attr("data-previous-shifts");
if (!prev_attr) prev_attr = jtr.data("previous-shifts");
let partsHtml = "";
let sat_adj = null;
let sun_adj = null;
if (prev_attr) {
try {
let prev_map = (typeof prev_attr === 'object') ? prev_attr : JSON.parse(prev_attr);
let parts = [];
// Build parts for non-weekend shifts first; we'll always render weekend per-day below.
for (let k in prev_map) {
if (!prev_map.hasOwnProperty(k)) continue;
if (k === 'weekend') continue;
let entry = prev_map[k];
let adj = null;
if (entry && typeof entry === 'object' && !Array.isArray(entry)) {
if ('adjustment' in entry) adj = parseFloat(entry.adjustment);
else if ('allocated' in entry && 'worked' in entry) adj = parseFloat(entry.allocated) - parseFloat(entry.worked);
} else if (Array.isArray(entry) && entry.length >= 2) {
adj = parseFloat(entry[1]) - parseFloat(entry[0]);
}
if (adj && Math.abs(adj) > 0.0001) {
let s = (adj > 0 ? '+' : '') + Number(adj).toFixed(2);
// span includes data attributes for per-shift colouring later
partsHtml += `<span class='prior-adjust-part' data-shift='${k}' data-adj='${adj}' style='display:inline-block;padding:0 6px;margin-right:6px;border-radius:3px'>${k}:${s}</span>`;
}
}
// compute per-day weekend adjustments if present under 'weekend'
if ('weekend' in prev_map) {
let w = prev_map['weekend'];
// Prefer explicit per-day entries when present
if (w && typeof w === 'object' && !Array.isArray(w)) {
if ('Sat' in w) {
let e = w['Sat'];
if (e && typeof e === 'object' && 'allocated' in e && 'worked' in e) sat_adj = parseFloat(e.allocated) - parseFloat(e.worked);
else if (Array.isArray(e) && e.length >= 2) sat_adj = parseFloat(e[1]) - parseFloat(e[0]);
}
if ('Sun' in w) {
let e = w['Sun'];
if (e && typeof e === 'object' && 'allocated' in e && 'worked' in e) sun_adj = parseFloat(e.allocated) - parseFloat(e.worked);
else if (Array.isArray(e) && e.length >= 2) sun_adj = parseFloat(e[1]) - parseFloat(e[0]);
}
// If no per-day entries but there is an overall weekend value, split it evenly
if (sat_adj === null && sun_adj === null && 'allocated' in w && 'worked' in w) {
try {
let total = parseFloat(w.allocated) - parseFloat(w.worked);
if (isFinite(total)) {
sat_adj = sun_adj = total / 2.0;
}
} catch (err) {
// leave as null on parse error
}
}
// Handle a '__all__' bucket (tuple or object) as fallback
if (sat_adj === null && sun_adj === null && '__all__' in w) {
let e = w['__all__'];
if (Array.isArray(e) && e.length >= 2) {
let total = parseFloat(e[1]) - parseFloat(e[0]);
sat_adj = sun_adj = total / 2.0;
} else if (e && typeof e === 'object' && 'allocated' in e && 'worked' in e) {
let total = parseFloat(e.allocated) - parseFloat(e.worked);
sat_adj = sun_adj = total / 2.0;
}
}
} else if (Array.isArray(w) && w.length >= 2) {
// tuple-style: split evenly
let total = parseFloat(w[1]) - parseFloat(w[0]);
sat_adj = sun_adj = total / 2.0;
}
// Always append per-day prior-adjust parts for weekend so the section is per-day
if (sat_adj !== null && Math.abs(sat_adj) > 0.0001) {
let s = (sat_adj > 0 ? '+' : '') + Number(sat_adj).toFixed(2);
partsHtml += `<span class='prior-adjust-part' data-shift='weekend' data-day='Sat' data-adj='${sat_adj}' style='display:inline-block;padding:0 6px;margin-right:6px;border-radius:3px'>Sat:${s}</span>`;
}
if (sun_adj !== null && Math.abs(sun_adj) > 0.0001) {
let s = (sun_adj > 0 ? '+' : '') + Number(sun_adj).toFixed(2);
partsHtml += `<span class='prior-adjust-part' data-shift='weekend' data-day='Sun' data-adj='${sun_adj}' style='display:inline-block;padding:0 6px;margin-right:6px;border-radius:3px'>Sun:${s}</span>`;
}
}
} catch (e) {
// ignore parse errors
}
}
// helper to render small coloured adjustment badge
function _adjBadge(adj) {
if (adj === null || typeof adj === 'undefined' || isNaN(adj)) return '';
let adjf = parseFloat(adj);
if (!isFinite(adjf)) return '';
let sign = (adjf > 0) ? '+' : '';
let text = `${sign}${adjf.toFixed(2)}`;
// scale alpha by magnitude but ensure a minimum so small adjustments remain visible
let mag = Math.min(1.0, Math.abs(adjf) / 3.0);
if (mag < 0.12) mag = 0.12;
let bg, color;
if (adjf > 0) { bg = `rgba(0,128,0,${mag.toFixed(2)})`; color = '#fff'; }
else if (adjf < 0) { bg = `rgba(196,0,0,${mag.toFixed(2)})`; color = '#fff'; }
else { bg = 'rgba(128,128,128,0.15)'; color = '#000'; }
return ` <span class='weekend-day-adjust' style='margin-left:6px; padding:1px 6px; border-radius:4px; background:${bg}; color:${color}; font-weight:600; font-size:0.85em;'>(${text})</span>`;
}
let sat_adj_html = _adjBadge(sat_adj);
let sun_adj_html = _adjBadge(sun_adj);
// compute Sat / Sun counts for this worker row (ensure variables are defined)
let sat_count = 0;
let sun_count = 0;
try {
let $row = jtr.closest('tr');
$row.find("td[data-day='Sat']").each((i, td) => {
let s = $(td).attr('data-shift');
if (s && s !== '') sat_count += 1;
});
$row.find("td[data-day='Sun']").each((i, td) => {
let s = $(td).attr('data-shift');
if (s && s !== '') sun_count += 1;
});
} catch (e) {
// fallback: leave counts at 0
}
console.log("Computed weekend counts and adjustments for worker:", jtr.data("worker"));
console.log(`sat_count: ${sat_count}, sun_count: ${sun_count}, sat_adj: ${sat_adj}, sun_adj: ${sun_adj}`);
// Insert worker summary now that we have per-day adjustment badges
// Hidden by default so summaries only appear when rota days are hidden
let workerSummaryHtml = `<div class='worker-summary auto-generated' style='display: none;'>
<span>Total shifts: ${total_shifts}, </span>
<span>Weekends: ${weekends_worked}, </span>
<span>Sat: ${sat_count}${sat_adj_html}, </span>
<span>Sun: ${sun_count}${sun_adj_html}, </span>
<span>Nights: ${nights_worked}, </span>
<span>Bank holidays: ${bank_holidays_worked}, </span>
<span class='shift-diff-span'>Summed shift diff: ${summed_shift_diff.toPrecision(2)}, </span>
</div>`;
jtr.after(workerSummaryHtml);
row.append(`<td class='prior-adjust' style='display:none'>${partsHtml}</td>`)
$("table.tsummary").append(row)
})
@@ -368,6 +533,31 @@ oshifts.forEach((shift) => {
});
// Colour each per-shift prior-adjust span independently.
// For each shift, find the maximum absolute adjustment across rows and scale saturation accordingly.
let priorShiftMax = {};
$("table.tsummary").find("td.prior-adjust .prior-adjust-part").each((n, span) => {
let $span = $(span);
let shift = $span.attr('data-shift');
let adj = parseFloat($span.attr('data-adj')) || 0;
let absadj = Math.abs(adj);
if (!(shift in priorShiftMax)) priorShiftMax[shift] = 0;
priorShiftMax[shift] = Math.max(priorShiftMax[shift], absadj);
});
// Apply colouring to each span using hsv2rgb (green for positive, red for negative)
$("table.tsummary").find("td.prior-adjust .prior-adjust-part").each((n, span) => {
let $span = $(span);
let shift = $span.attr('data-shift');
let adj = parseFloat($span.attr('data-adj')) || 0;
if (Math.abs(adj) < 1e-9) return;
let max = priorShiftMax[shift] || Math.abs(adj) || 1;
let sat = (max === 0) ? 1 : Math.min(1, Math.abs(adj) / max);
let hue = (adj > 0) ? 120 : 0; // green or red
$span.css({ backgroundColor: hsv2rgb(hue, sat, 1), color: '#000' });
$span.addClass('no-colour');
});
let selectedShifts = new Set();
let shiftColors = {};
Regular → Executable
+4 -1
View File
@@ -18,4 +18,7 @@ typer
django>=6.0,<7
django-crispy-forms
crispy-bulma
django-debug-toolbar
django-debug-toolbar
pandas
odfpy
openpyxl
+629 -12
View File
@@ -93,12 +93,14 @@ class PreShiftConstraint(BaseShiftConstraint):
days: Optional[int] = None
ignore_shifts: List[str] = []
exclude_days: List[str] = []
exclude_dates: List[datetime.date] = []
allow_self: bool = True
class PostShiftConstraint(BaseShiftConstraint):
days: Optional[int] = None
ignore_shifts: List[str] = []
exclude_days: List[str] = []
exclude_dates: List[datetime.date] = []
allow_self: bool = True
class BalanceAcrossGroupsConstraint(BaseShiftConstraint):
@@ -343,6 +345,20 @@ class RotaConstraintOptions(BaseModel):
maximum_allowed_shift_diff: Optional[int] = Field(None, description="Maximum allowed per-worker shift difference (None = no limit)")
balance_weekends: bool = Field(True, description="Balance weekend allocations")
max_weekends: int = Field(100, description="Maximum number of weekends considered")
# If true, try to balance number of weekend days worked per worker separately
# for Saturday and Sunday (i.e. equivalent workers should have similar Sat counts
# and similar Sun counts).
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[float] = Field(
None,
description=(
"Hard maximum allowed absolute deviation (number) per-worker from the per-day "
"weekend target (Sat/Sun). If set, adds constraints enforcing |assigned - target| <= value."
),
)
max_weekend_frequency: int = Field(1, description="Don't assign multiple shifts every N weeks (requires balance_weekends)")
max_night_frequency: int = Field(1, description="Don't assign multiple night shifts every N weeks")
max_night_frequency_week_exclusions: List[int] = Field(default_factory=list, description="Weeks to exclude from night frequency checks")
@@ -383,6 +399,7 @@ class RotaBuilder(object):
SHIFT_BOUNDS=SHIFT_BOUNDS,
name: str = "",
bank_holidays: dict[datetime.date, str] = bank_holiday_map,
constraint_options: RotaConstraintOptions | dict | None = None,
):
self.name = name
@@ -418,7 +435,6 @@ class RotaBuilder(object):
# None (use defaults), a dict (raw values), or an existing model instance.
# Store the model instance in `self.constraint_options_model` and also
# expose it as `self.constraint_options` for minimal code changes.
constraint_options = None
if constraint_options is None:
opts_model = RotaConstraintOptions()
elif isinstance(constraint_options, RotaConstraintOptions):
@@ -430,6 +446,8 @@ class RotaBuilder(object):
self.constraint_options_model: RotaConstraintOptions = opts_model
logger.debug(f"Rota constraint options: {self.constraint_options_model.model_dump()}")
self.terminate_on_warning = [
"Worker/duplicate id",
"Worker/duplicate name",
@@ -1011,6 +1029,39 @@ class RotaBuilder(object):
initialize=0,
)
# Per-day weekend balancing (separate Sat / Sun counts)
# Also create the bookkeeping variables when a hard deviation constraint is requested
logger.debug(f"Weekend day balancing options: {self.constraint_options_model.balance_weekend_days=}, {self.constraint_options_model.balance_weekend_days_quadratic=}, {self.constraint_options_model.weekend_day_hard_max_deviation=}")
if (
self.constraint_options_model.balance_weekend_days
or self.constraint_options_model.balance_weekend_days_quadratic
or self.constraint_options_model.weekend_day_hard_max_deviation is not None
):
logger.debug("Creating per-day weekend balancing variables")
self.model.weekend_day_shift_count = Var(
((worker.id, d) for worker in self.workers for d in ("Sat", "Sun")),
domain=NonNegativeReals,
initialize=0,
)
self.model.weekend_day_shift_count_t1 = Var(
((worker.id, d) for worker in self.workers for d in ("Sat", "Sun")),
domain=NonNegativeReals,
initialize=0,
)
self.model.weekend_day_shift_count_t2 = Var(
((worker.id, d) for worker in self.workers for d in ("Sat", "Sun")),
domain=NonNegativeReals,
initialize=0,
)
self.model.weekend_day_shift_count_w = Var(
((worker.id, d) for worker in self.workers for d in ("Sat", "Sun")),
domain=NonNegativeReals,
initialize=0,
)
# self.model.weekend_count_t1 = Var(
# ((worker.id) for worker in self.workers),
# domain=NonNegativeReals,
@@ -1701,8 +1752,126 @@ class RotaBuilder(object):
)
for week in weeks_to_apply:
# compute which days in the day_group should be ignored
ignored_days = set()
try:
for ignore_date in getattr(dep, "ignore_dates", []) or []:
try:
ww, dd = self.date_week_day_map.get(ignore_date, (None, None))
except Exception:
ww = dd = None
if ww == week and dd in day_group:
ignored_days.add(dd)
except Exception:
ignored_days = set()
# If the worker has any forced assignment in the dependency group,
# emit a concise warning. For visibility include every day in the
# group showing the worker's forced shifts (if any) and other
# workers' forced shifts on those days so blocking days are visible.
worker_has_forced_in_group = False
for (fw, fd, fs) in getattr(worker, "forced_assignments", []):
if fw == week and fd in day_group:
worker_has_forced_in_group = True
break
if not worker_has_forced_in_group:
try:
for (d_date, d_shift) in getattr(worker, "forced_assignments_by_date", []):
ww, dd = self.date_week_day_map.get(d_date, (None, None))
if ww == week and dd in day_group:
worker_has_forced_in_group = True
break
except Exception:
pass
if worker_has_forced_in_group:
forced_day_details = []
try:
date_map_for_week = {self.get_date_by_week_day(week, d): d for d in day_group}
except Exception:
date_map_for_week = {}
for day in day_group:
if day in ignored_days:
# skip ignored dates for warning/visibility
continue
worker_forced_shifts = set()
# week-based forced assignments
for (fw, fd, fs) in getattr(worker, "forced_assignments", []):
if fw == week and fd == day:
worker_forced_shifts.add(fs)
# date-based forced assignments
try:
date = self.get_date_by_week_day(week, day)
for (d_date, d_shift) in getattr(worker, "forced_assignments_by_date", []):
try:
ww, dd = self.date_week_day_map.get(d_date, (None, None))
except Exception:
ww = dd = None
if ww == week and dd == day:
worker_forced_shifts.add(d_shift)
else:
try:
if str(d_date) == str(date):
worker_forced_shifts.add(d_shift)
except Exception:
pass
except Exception:
pass
# collect other workers' forced shifts on this day
others_forced = {}
try:
date_for_day = self.get_date_by_week_day(week, day)
except Exception:
date_for_day = None
for other in self.workers:
if other is worker:
continue
other_shifts = set()
# week-based forced assignments
for (ofw, ofd, ofs) in getattr(other, "forced_assignments", []):
if ofw == week and ofd == day:
other_shifts.add(ofs)
# date-based forced assignments: try mapping then fallback
for (odate, oshift) in getattr(other, "forced_assignments_by_date", []):
try:
oww, odd = self.date_week_day_map.get(odate, (None, None))
except Exception:
oww = odd = None
if oww == week and odd == day:
other_shifts.add(oshift)
continue
try:
if date_for_day is not None and str(odate) == str(date_for_day):
other_shifts.add(oshift)
except Exception:
pass
if other_shifts:
others_forced[other.name] = sorted(list(other_shifts))
forced_day_details.append((day, sorted(list(worker_forced_shifts)), others_forced))
parts = []
for d, wfs, others in forced_day_details:
olist = ", ".join([f"{n}: {', '.join(s)}" for n, s in others.items()]) or "none"
wfs_str = ", ".join(wfs) or "none"
parts.append(f"{d} - worker forced: {wfs_str}; others forced: {olist}")
details = "; ".join(parts)
logger.warning(f"HardDayDependency forced-assignment warning for worker={worker.name} week={week}: {details}")
self.add_warning(
"HardDayDependency forced assignment",
f"Worker '{worker.name}' has hard_day_dependency {set(day_group)} in week {week}; forced assignments on dependency days: {details}. If the rota is not infeasible this is unlikely to be an issue.",
)
# Build constraints only for non-ignored days
effective_days = [d for d in day_group if d not in ignored_days]
if not effective_days:
# nothing to constrain for this dependency in this week
continue
assigned_any_shift = {}
for day in day_group:
for day in effective_days:
shifts_today = self.get_shift_names_by_week_day(week, day)
var_name = f"hard_day_dep_{worker.id}_{week}_{day}"
if not hasattr(self.model, "hard_day_dep_vars"):
@@ -1714,7 +1883,7 @@ class RotaBuilder(object):
total_assigned = sum(self.model.works[worker.id, week, day, shift] for shift in shifts_today)
self.model.constraints.add(total_assigned >= assigned_any_shift[day])
self.model.constraints.add(total_assigned <= len(shifts_today) * assigned_any_shift[day])
# All-or-none: all days in the group must have the same assignment status
# All-or-none: all effective (non-ignored) days in the group must have the same assignment status
vals = list(assigned_any_shift.values())
for i in range(1, len(vals)):
self.model.constraints.add(vals[i] == vals[0])
@@ -2028,10 +2197,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:
@@ -2338,6 +2533,137 @@ class RotaBuilder(object):
- xU * xU
)
# Per-day weekend balancing (Sat and Sun separately)
logger.debug(f"Adding per-day weekend balancing constraints for worker {worker.name} ({worker.id})")
logger.debug(self.constraint_options_model.weekend_day_hard_max_deviation)
if (
self.constraint_options_model.balance_weekend_days
or self.constraint_options_model.balance_weekend_days_quadratic
or self.constraint_options_model.weekend_day_hard_max_deviation is not None
):
logger.debug(f"Worker {worker.name}: per-day weekend block active (hard_dev={self.constraint_options_model.weekend_day_hard_max_deviation})")
console.print(f"Worker {worker.name}: per-day weekend block active (hard_dev={self.constraint_options_model.weekend_day_hard_max_deviation})")
hard_weekend_constraints_added = 0
for day_name in ("Sat", "Sun"):
# count assignments on this specific weekend day
self.model.constraints.add(
self.model.weekend_day_shift_count[worker.id, day_name]
== sum(
self.model.works[worker.id, w, d, shift]
for w, d, shift in self.get_all_shiftname_combinations()
if d == day_name
)
)
# target number for this day (derived from shift targets)
weekend_day_shift_target_number = sum(
worker.shift_target_number[shift.name] / len(shift.days)
for shift in self.get_shifts()
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
# store adjusted per-day weekend target on the worker object for later use in objectives
if not hasattr(worker, 'weekend_day_target'):
worker.weekend_day_target = {}
worker.weekend_day_target[day_name] = weekend_day_shift_target_number
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]
== self.model.weekend_day_shift_count[worker.id, day_name]
- weekend_day_shift_target_number
)
xU_day = self.constraint_options_model.max_weekends
xL_day = 1
self.model.constraints.add(
inequality(
xL_day,
self.model.weekend_day_shift_count_t1[worker.id, day_name]
+ self.model.weekend_day_shift_count_t2[worker.id, day_name]
+ 1,
xU_day,
)
)
self.model.constraints.add(
self.model.weekend_day_shift_count_w[worker.id, day_name]
>= xL_day
* (
self.model.weekend_day_shift_count_t1[worker.id, day_name]
+ self.model.weekend_day_shift_count_t2[worker.id, day_name]
+ 1
)
* 2
- xL_day * xL_day
)
self.model.constraints.add(
self.model.weekend_day_shift_count_w[worker.id, day_name]
>= xU_day
* (
self.model.weekend_day_shift_count_t1[worker.id, day_name]
+ self.model.weekend_day_shift_count_t2[worker.id, day_name]
+ 1
)
* 2
- xU_day * xU_day
)
# Optional hard constraint: limit absolute deviation from per-day target
if self.constraint_options_model.weekend_day_hard_max_deviation is not None:
d = self.constraint_options_model.weekend_day_hard_max_deviation
console.print(f"Adding hard weekend-day deviation constraint: worker={worker.name} ({worker.id}), day={day_name}, target={weekend_day_shift_target_number:.3f}, d={d}")
# assigned - target <= d
self.model.constraints.add(
self.model.weekend_day_shift_count[worker.id, day_name]
- weekend_day_shift_target_number
<= d
)
# target - assigned <= d
self.model.constraints.add(
weekend_day_shift_target_number
- self.model.weekend_day_shift_count[worker.id, day_name]
<= d
)
hard_weekend_constraints_added += 2
if hard_weekend_constraints_added:
console.print(f"Worker {worker.name}: added {hard_weekend_constraints_added} hard weekend-day constraints")
# Ensure worker is not allocated shifts on non working days
if worker.non_working_day_list:
for week, day, shift in self.get_all_shiftclass_combinations():
@@ -2991,10 +3317,12 @@ class RotaBuilder(object):
ignore_shifts = constraint.ignore_shifts + [constraint_shift.name]
else:
ignore_shifts = constraint.ignore_shifts
if week not in constraint.weeks:
if constraint.weeks is not None and week not in constraint.weeks:
continue
date_main = self.get_date_by_week_day(week, day)
for n in range(0, constraint.days):
if day in constraint.exclude_days:
# skip if main day excluded by weekday or specific date
if day in getattr(constraint, "exclude_days", []) or date_main in getattr(constraint, "exclude_dates", []):
continue
if day in constraint_shift.days:
# Only apply if worker can work this shift
@@ -3006,6 +3334,14 @@ class RotaBuilder(object):
]
except KeyError:
continue
# compute the pre date and skip if it's in exclude_dates
try:
pre_date = self.get_date_by_week_day(pre_map[n][1], pre_map[n][2])
except Exception:
pre_date = None
if pre_date is not None and pre_date in getattr(constraint, "exclude_dates", []):
continue
self.model.constraints.add(
1
>= works
@@ -3031,10 +3367,12 @@ class RotaBuilder(object):
ignore_shifts = constraint.ignore_shifts + [constraint_shift.name]
else:
ignore_shifts = constraint.ignore_shifts
if week not in constraint.weeks:
if constraint.weeks is not None and week not in constraint.weeks:
continue
date_main = self.get_date_by_week_day(week, day)
for n in range(0, constraint.days):
if day in constraint.exclude_days:
# skip if main day excluded by weekday or specific date
if day in getattr(constraint, "exclude_days", []) or date_main in getattr(constraint, "exclude_dates", []):
continue
if day in constraint_shift.days:
# Only apply if worker can work this shift
@@ -3046,6 +3384,14 @@ class RotaBuilder(object):
]
except KeyError:
continue
# compute the post date and skip if it's in exclude_dates
try:
post_date = self.get_date_by_week_day(post_map[n][1], post_map[n][2])
except Exception:
post_date = None
if post_date is not None and post_date in getattr(constraint, "exclude_dates", []):
continue
self.model.constraints.add(
1
>= works
@@ -3239,6 +3585,31 @@ class RotaBuilder(object):
else:
weekend_shift_balancing = 0
# Balance each weekend day (Sat / Sun) separately if requested.
# Support either a linear term or a stronger quadratic penalty.
weekend_day_term = 0
if self.constraint_options_model.balance_weekend_days_quadratic:
wd_weight = self.constraint_options_model.balance_weekend_days_weight
# HiGHS/appsi does not accept arbitrary nonlinear objective expressions here.
# Use the existing t1/t2 linear representation (which approximates absolute
# deviation) in the objective instead of a true squared term.
weekend_day_term = sum(
wd_weight
* (
self.model.weekend_day_shift_count_t1[(worker.id, day_name)]
+ self.model.weekend_day_shift_count_t2[(worker.id, day_name)]
)
for worker in self.workers
for day_name in ("Sat", "Sun")
)
elif self.constraint_options_model.balance_weekend_days:
weekend_day_modifier = self.constraint_options_model.balance_weekend_days_weight
weekend_day_term = sum(
weekend_day_modifier * self.model.weekend_day_shift_count_w[(worker.id, day_name)]
for worker in self.workers
for day_name in ("Sat", "Sun")
)
preference_constant = 10
# Preferences (not to work)
preferences = sum(
@@ -3338,6 +3709,7 @@ class RotaBuilder(object):
+ true_quadratic_shift_balancing
- prefer_multi_shift_expr
+ prefer_block_expr
+ weekend_day_term
)
# add objective function to the model. rule (pass function) or expr (pass expression directly)
@@ -3425,6 +3797,47 @@ class RotaBuilder(object):
self.workers = sorted(self.workers)
# Pairwise hard constraints to limit per-day weekend skew among equivalent workers
# Group workers by (site, grade) and enforce for each pair and each day:
# assigned_i_day - assigned_j_day <= d
# assigned_j_day - assigned_i_day <= d
# Use `weekend_day_hard_max_deviation` as `d` when set.
if self.constraint_options_model.weekend_day_hard_max_deviation is not None:
d = self.constraint_options_model.weekend_day_hard_max_deviation
groups = defaultdict(list)
for w in self.workers:
groups[(w.site, w.grade)].append(w)
pairwise_added = 0
for (site, grade), workers_group in groups.items():
if len(workers_group) < 2:
continue
# iterate unique unordered pairs
for i in range(len(workers_group)):
for j in range(i + 1, len(workers_group)):
wi = workers_group[i]
wj = workers_group[j]
for day_name in ("Sat", "Sun"):
try:
# wi - wj <= d
self.model.constraints.add(
self.model.weekend_day_shift_count[wi.id, day_name]
- self.model.weekend_day_shift_count[wj.id, day_name]
<= d
)
# wj - wi <= d
self.model.constraints.add(
self.model.weekend_day_shift_count[wj.id, day_name]
- self.model.weekend_day_shift_count[wi.id, day_name]
<= d
)
pairwise_added += 2
except Exception:
# If variables aren't present for some worker/day, skip
continue
console.print(f"Added {pairwise_added} pairwise weekend-day hard constraints (d={d})")
self.full_time_equivalent = sum(
w.fte_adj for w in self.workers
) # TODO: rewrite as per shift
@@ -4297,6 +4710,21 @@ class RotaBuilder(object):
shift_count
+ f"#weekends_worked: {model.worker_weekend_count[worker.id].value}\\#"
)
# compute assigned Saturday / Sunday counts for this worker
sat_count = 0
sun_count = 0
for w, d in self.get_week_day_combinations():
if d == "Sat":
for s in self.get_shift_names_by_week_day(w, d, return_empty_if_week_day_not_found=True):
if model.works[worker.id, w, d, s].value > 0.5:
sat_count += 1
break
if d == "Sun":
for s in self.get_shift_names_by_week_day(w, d, return_empty_if_week_day_not_found=True):
if model.works[worker.id, w, d, s].value > 0.5:
sun_count += 1
break
timetable.append(f"{''.join(w)} {shift_count}")
if show_prefs:
@@ -4520,6 +4948,170 @@ class RotaBuilder(object):
diff = model.shift_count_diff[worker.id, shift.name].value
shift_diff_dict[shift.name] = diff
# compute assigned Saturday / Sunday counts for this worker (for HTML data attributes)
sat_count = 0
sun_count = 0
for w, d in self.get_week_day_combinations():
if d == "Sat":
for s in self.get_shift_names_by_week_day(w, d, return_empty_if_week_day_not_found=True):
if model.works[worker.id, w, d, s].value > 0.5:
sat_count += 1
break
if d == "Sun":
for s in self.get_shift_names_by_week_day(w, d, return_empty_if_week_day_not_found=True):
if model.works[worker.id, w, d, s].value > 0.5:
sun_count += 1
break
# Small HTML summary to be inserted inside the worker td (friendly display)
# Build a small adjustments summary from prior allocations (if any)
prior_map = {}
adjustments_list = []
for shift in self.get_shifts():
prev = getattr(worker, "previous_shifts", {}) or {}
if shift.name in prev:
entry = prev[shift.name]
# entry may be:
# - a tuple/list: (worked, allocated)
# - a dict with per-day entries (e.g., {'Sat': (w,a), 'Sun': (w,a)})
# - a dict with an '__all__' key or other buckets
if isinstance(entry, dict):
# If the dict already contains per-day keys (Sat/Sun) or an explicit
# '__all__' bucket, preserve the dict as-is so downstream code
# can render per-day adjustments. Otherwise, attempt to aggregate
# any tuple-like values found in the dict into totals.
if any(k in entry for k in ("Sat", "Sun")) or "__all__" in entry:
prior_map[shift.name] = entry
# also compute an overall adjustment for the adjustments list
try:
# compute totals by summing tuple/list values where present
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)))
adj = float(allocated) - float(worked)
except Exception:
adj = 0
else:
# fallback: sum any tuple values found in the dict
try:
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)))
adj = float(allocated) - float(worked)
except Exception:
worked = 0
allocated = 0
adj = 0
prior_map[shift.name] = dict(worked=worked, allocated=allocated, adjustment=adj)
else:
# tuple/list style entry
try:
worked, allocated = entry
adj = float(allocated) - float(worked)
except Exception:
worked = 0
allocated = 0
adj = 0
prior_map[shift.name] = dict(worked=worked, allocated=allocated, adjustment=adj)
if adj != 0:
adjustments_list.append(f"{shift.name}:{adj:+g}")
# If weekend prior is present as a top-level worked/allocated dict (not per-day),
# expand it into per-day entries so Sat/Sun adjustments are always available.
if 'weekend' in prior_map:
w = prior_map['weekend']
# detect the simple dict shape produced earlier: {'worked': x, 'allocated': y, 'adjustment': z}
if isinstance(w, dict) and 'worked' in w and 'allocated' in w and not ('Sat' in w or 'Sun' in w or '__all__' in w):
try:
worked = float(w.get('worked', 0))
allocated = float(w.get('allocated', 0))
# split evenly across Sat and Sun
sat_worked = worked / 2.0
sun_worked = worked - sat_worked
sat_alloc = allocated / 2.0
sun_alloc = allocated - sat_alloc
prior_map['weekend'] = {
'Sat': (sat_worked, sat_alloc),
'Sun': (sun_worked, sun_alloc),
}
except Exception:
# leave as-is if parsing fails
pass
adjustments_html = ""
if adjustments_list:
adjustments_html = (
"<div class='worker-adjustments auto-generated' style='margin-top:3px; font-size:0.85em; color:#666;'>"
+ "Adjust: "
+ ", ".join(adjustments_list)
+ "</div>"
)
# Compute per-day (Sat/Sun) adjustments for the 'weekend' shift if provided
sat_adj = None
sun_adj = None
prev_all = getattr(worker, "previous_shifts", {}) or {}
weekend_entry = prev_all.get("weekend")
if weekend_entry is not None:
try:
if isinstance(weekend_entry, dict):
if "Sat" in weekend_entry:
w_worked, w_alloc = weekend_entry["Sat"]
sat_adj = float(w_alloc) - float(w_worked)
if "Sun" in weekend_entry:
w_worked, w_alloc = weekend_entry["Sun"]
sun_adj = float(w_alloc) - float(w_worked)
if sat_adj is None and sun_adj is None and "__all__" in weekend_entry:
w_worked, w_alloc = weekend_entry["__all__"]
total = float(w_alloc) - float(w_worked)
sat_adj = sun_adj = total / 2.0
if sat_adj is None and sun_adj is None:
# fallback: sum any tuple values and split
worked_sum = sum(v[0] for v in weekend_entry.values() if isinstance(v, (list, tuple)))
alloc_sum = sum(v[1] for v in weekend_entry.values() if isinstance(v, (list, tuple)))
total = float(alloc_sum) - float(worked_sum)
sat_adj = sun_adj = total / 2.0
else:
# tuple-style entry, split evenly across Sat/Sun
w_worked, w_alloc = weekend_entry
total = float(w_alloc) - float(w_worked)
sat_adj = sun_adj = total / 2.0
except Exception:
sat_adj = sun_adj = None
def _adj_html(adj):
if adj is None:
return ""
try:
adjf = float(adj)
except Exception:
return ""
sign_text = f"{adjf:+.2f}"
# colour intensity scaled by magnitude (cap divisor at 3.0 to avoid overpowering)
mag = min(1.0, abs(adjf) / 3.0)
# ensure a minimum alpha so small adjustments remain visible
if mag < 0.12:
mag = 0.12
if adjf > 0:
bg = f"rgba(0,128,0,{mag:.2f})"
color = "#fff"
elif adjf < 0:
bg = f"rgba(196,0,0,{mag:.2f})"
color = "#fff"
else:
bg = "rgba(128,128,128,0.15)"
color = "#000"
return f" <span class='weekend-day-adjust' style='margin-left:6px; padding:1px 6px; border-radius:4px; background:{bg}; color:{color}; font-weight:600; font-size:0.85em;'>({sign_text})</span>"
sat_adj_html = _adj_html(sat_adj)
sun_adj_html = _adj_html(sun_adj)
worker_summary_html = (
f"<div class='worker-summary auto-generated' style='margin-top:4px; font-size:0.9em; color:#333;'>"
f"<span class='weekend-sat'>Sat: <strong>{sat_count}</strong>{sat_adj_html}</span>"
f" &nbsp; <span class='weekend-sun'>Sun: <strong>{sun_count}</strong>{sun_adj_html}</span>"
f"{adjustments_html}"
f"</div>"
)
worker_td = """<td title='Site: {site}' class='worker {site}'
data-worker-id='{worker_id}'
data-nwds='{nwds}' data-site='{site}' data-worker='{name}'
@@ -4535,10 +5127,13 @@ class RotaBuilder(object):
data-pair='{pair}'
data-shift-balance-extra='{shift_balance_extra}'
data-bank-holiday-extra='{bank_holiday_extra}'
data-weekend-sat='{weekend_sat}'
data-weekend-sun='{weekend_sun}'
data-previous-shifts='{previous_shifts}'
data-shift-diff='{shift_diff}'
data-shift-diff-summed='{shift_diff_summed}'
>
<span class='name' title='{name} ({worker_id})'>{name}</span> ({grade}) [{fte}]</td>""".format(
<span class='name' title='{name} ({worker_id})'>{name}</span> ({grade}) [{fte}]{worker_summary_html}</td>""".format(
site=worker.site,
nwds=nwds,
worker_id=worker.id,
@@ -4554,6 +5149,10 @@ class RotaBuilder(object):
weekend_target=worker.weekend_shift_target_number,
remote_site=worker.remote_site,
grade=worker.grade,
weekend_sat=sat_count,
weekend_sun=sun_count,
worker_summary_html=worker_summary_html,
previous_shifts=json.dumps(prior_map),
pair=worker.pair,
shift_balance_extra=json.dumps(worker.shift_balance_extra),
shift_diff=json.dumps(shift_diff_dict),
@@ -4647,6 +5246,24 @@ class RotaBuilder(object):
f"Unavailable: {''}",
f"Shift targets: {json.dumps(getattr(worker, 'shift_target_number', {}))}",
]
# compute Sat / Sun assigned counts for this worker
sat_count = 0
sun_count = 0
for w, d in self.get_week_day_combinations():
if d == "Sat":
for s in self.get_shift_names_by_week_day(w, d, return_empty_if_week_day_not_found=True):
if self.model.works[worker.id, w, d, s].value > 0.5:
sat_count += 1
break
if d == "Sun":
for s in self.get_shift_names_by_week_day(w, d, return_empty_if_week_day_not_found=True):
if self.model.works[worker.id, w, d, s].value > 0.5:
sun_count += 1
break
details.insert(
-1,
f"Sat assigned: {sat_count}, Sun assigned: {sun_count}",
)
# Build unavailability list for this worker
unavail_list = []
for entry in sorted(self.unavailable_to_work):
+3 -2
View File
@@ -36,6 +36,7 @@ class HardDayDependency(BaseModel):
weeks: Optional[List[int]] = None # If None, applies to all weeks
start_date: Optional[datetime.date] = None # Optional start date for the dependency
end_date: Optional[datetime.date] = None # Optional end date for the dependency
ignore_dates: list[datetime.date] = Field(default_factory=list, description="List of specific dates to ignore for this dependency")
@field_validator("weeks")
def weeks_none_if_dates_set(cls, weeks, values):
@@ -437,13 +438,13 @@ class Worker(BaseModel):
self.allowed_multi_shift_sets = []
self.allowed_multi_shift_sets.append(frozenset(shifts))
def add_hard_day_dependency(self, days: set[str], weeks: Optional[List[int]] = None, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None):
def add_hard_day_dependency(self, days: set[str], weeks: Optional[List[int]] = None, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, ignore_dates: Optional[List[datetime.date]] = None):
"""
Add a hard day dependency for this worker.
days: List of days that must be grouped together.
weeks: Optional list of weeks this dependency applies to. If None, applies to all weeks.
"""
self.hard_day_dependencies.append(HardDayDependency(days=days, weeks=weeks, start_date=start_date, end_date=end_date))
self.hard_day_dependencies.append(HardDayDependency(days=days, weeks=weeks, start_date=start_date, end_date=end_date, ignore_dates=ignore_dates or []))
def add_hard_day_exclusion(self, days: set[str], weeks: Optional[List[int]] = None, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None):
"""