Compare commits

..
7 Commits
5 changed files with 631 additions and 56 deletions
+59 -47
View File
@@ -15,23 +15,36 @@ from rota_generator.workers import (
NonWorkingDays,
WorkRequests,
PreferenceNotToWork,
OutOfProgramme,
OutOfProgramme, MaxUniqueShiftsPerWeekBlockConstraint,
)
from loguru import logger
app = typer.Typer()
ROTA_REQUESTS = "cons/requests.ods"
ROTA_B_PATH = "cons/rota_b.xlsx"
ROTA_START_DATE = "2026-08-03"
sites = ("rota a", "rota b", "rota c")
def _parse_sheet_date(value):
if pd.isna(value):
return None
parsed = pd.to_datetime(value, errors="coerce")
if pd.isna(parsed):
return None
return parsed.date()
def extract_leave_and_rota_from_calender(calender_df):
"""
Extract leave and rota assignments from calender_df.
- The first row contains worker names, starting from the 5th column (index 4).
- The second column (index 1) contains dates in dd/mm/yyyy format.
- The second column (index 1) also contains the rota/leave code for that date.
Returns a list of dicts: {"date": ..., "worker": ..., "rota": ...}
Returns a list of dicts: {"date": ..., "/worker": ..., "rota": ...}
"""
# Get worker names from the first row, starting from column 5 (index 4)
worker_names = [str(x).strip() for x in calender_df.columns[5:]]
@@ -49,15 +62,11 @@ def extract_leave_and_rota_from_calender(calender_df):
work_requests = []
# Process leave data from row 3 onwards (index 2 and up)
for idx, row in calender_df.iloc[3:].iterrows():
date_str = str(row.iloc[1]).strip().split(" ")[0] # Get the date string from the second column (index 1)
if not date_str or date_str.lower() == "nan":
date = _parse_sheet_date(row.iloc[1])
if date is None:
continue
try:
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):
if date < datetime.date.fromisoformat(ROTA_START_DATE):
continue # skip rows before rota start date
for i, worker in enumerate(worker_names):
if not worker or worker.lower() == "nan":
@@ -96,29 +105,30 @@ def extract_shift_assignments_from_rota(rota_df):
assignments = []
skipped_rows = 0
rota_start = datetime.date.fromisoformat(ROTA_START_DATE)
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:
week_start = datetime.datetime.strptime(week_start_str, "%Y-%m-%d").date()
except Exception:
print(f"Invalid date format in row {idx}: {week_start_str}")
week_start = _parse_sheet_date(row.iloc[0])
if week_start is None:
print(f"Invalid date format in row {idx}: {row.iloc[0]}")
continue # skip rows with invalid date
if week_start < datetime.date(2026, 2, 16):
if week_start < rota_start:
skipped_rows += 1
continue # skip rows before rota start date
week = ((week_start - rota_start).days // 7) + 1
for i, day in enumerate(days):
for j, shift in enumerate(shifts):
if i < 5:
# Mon-Fri: two columns per day (twilight, oncall)
col_idx = 1 + i * 2 + j # 1 for Mon twilight, 2 for Mon oncall, etc.
col_idx = 2 + i * 2 + j # 2 for Mon twilight, 3 for Mon oncall, etc.
weekend = False
else:
# Sat/Sun: only one column for "weekend" shift (twilight), skip "night"
weekend = True
if j == 0:
col_idx = 11 + (i - 5) # 11 for Sat, 12 for Sun
col_idx = 12 + (i - 5) # 12 for Sat, 13 for Sun
else:
continue # No night shift column for Sat/Sun
if col_idx >= len(row):
@@ -131,7 +141,7 @@ def extract_shift_assignments_from_rota(rota_df):
shift = "weekend"
assignments.append({
"week_start": week_start,
"week": idx-1-skipped_rows, # 1-indexed week number
"week": week,
"day": day,
"date": week_start + datetime.timedelta(days=i),
"shift": shift,
@@ -149,7 +159,7 @@ def extract_shift_assignments_from_rota(rota_df):
def load_workers():
# Path to the ODS file
ods_path = "cons/CONSULTANT TWILIGHT & ONCALL ROTA.ods"
ods_path = ROTA_REQUESTS
# Read all sheets
xls = pd.read_excel(ods_path, engine="odf", sheet_name=None)
@@ -160,13 +170,19 @@ def load_workers():
rota_df = xls["Rota"]
rota_b_path = "cons/Rota B Feb -July 2026.xlsx"
rota_b_path = ROTA_B_PATH
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():
logger.debug(f"{idx}: {row}")
date_val = row.iloc[0]
worker_val = row.iloc[1].upper() if len(row) > 1 else None
try:
worker_val = row.iloc[1].upper() if len(row) > 1 else None
except AttributeError:
# end of file
break
if pd.isna(date_val) or pd.isna(worker_val):
continue
try:
@@ -236,19 +252,15 @@ def load_workers():
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()
if "(" in text and ")" in text:
initial = text.split("(")[-1].split(")")[0].strip()
else:
raise ValueError(f"Invalid format in row {idx}: {text}")
name = text.split("(")[0].strip()
name = str(row.iloc[1]).strip()
initial = str(row.iloc[2]).strip()
if not name or name.lower() == "nan":
continue
availability = {}
requests = {}
rota_data[name] = f"rota {str(row.iloc[2]).strip().lower()}"
rota_data[name] = f"rota {str(row.iloc[3]).strip().lower()}"
for i, day in enumerate(weekday_cols):
val = row.iloc[3 + i]
val = row.iloc[4 + 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() or str(val).strip().lower() == "yes*")
availability[day] = available
@@ -361,7 +373,8 @@ def load_workers():
worker = worker_day["name"]
initial = worker_day["initial"]
start_date = "2026-02-16" # Default start date, can be adjusted later
start_date = ROTA_START_DATE
#if initial == "ND":
# start_date = "2025-09-29" # Non-working day worker starts later
@@ -402,10 +415,9 @@ def load_workers():
],
nwds=non_working_days,
previous_shifts=priors_map.get(initial, {}),
max_days_worked_per_week=2
)
if initial == "LB":
w.start_date = datetime.date(2026, 6, 15)
#if initial == "L1":
# w.oop = [OutOfProgramme(
@@ -447,19 +459,16 @@ def load_workers():
#w.add_force_assign_with("twilight", "oncall")
w.add_force_assign_with("weekend", "oncall")
if w.name in ("DS",):
if w.name in ("DS","RG"):
w.add_hard_day_dependency({"Fri", "Sat", "Sun"}, ignore_dates=[datetime.date(2026, 5, 2)])
w.max_days_worked_per_week = 3
else:
w.add_hard_day_dependency({"Fri", "Sun"})
w.add_hard_day_exclusion({"Sat", "Sun"})
else:
w.max_unique_shifts_per_week_block = [MaxUniqueShiftsPerWeekBlockConstraint(week_block=1, max_unique_shifts=1)]
#if w.name == "TB":
# w.add_hard_day_exclusion({"Sat", "Sun"}, start_date="2025-11-17")
#w.set_max_shifts
#w.set_max_shifts_per_week("weekend", 1)
workers.append(w)
@@ -472,9 +481,9 @@ def main(
suspend: bool = False,
solve: bool = True,
time_to_run: int = 60 * 60,
ratio: float = 0.001,
start_date: datetime.datetime = "2026-02-16",
weeks: int = 24,
ratio: float = 0.1,
start_date: datetime.datetime = ROTA_START_DATE,
weeks: int = 22,
bom: int = 1,
):
rota_start_date = start_date.date()
@@ -485,13 +494,13 @@ def main(
weeks_to_rota=weeks,
balance_offset_modifier=bom,
use_previous_shifts=True,
name="cons rota feb 2026 run 5",
name="cons rota 2026 july",
allow_force_assignment_with_leave_conflict=True,
constraint_options=RotaConstraintOptions(
balance_weekends=True,
max_shifts_per_week=6,
max_weekend_frequency=3,
max_days_per_week_block=[(4, 2), (4, 4)],
max_days_per_week_block=[(3, 2), (4, 4)],
#balance_weekend_days=True,
balance_weekend_days_quadratic=True,
weekend_day_hard_max_deviation=2,
@@ -512,8 +521,8 @@ def main(
balance_offset=2,
assign_as_block=False,
constraints=[
PreShiftConstraint(days=2, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun"), exclude_dates=["2026-04-06", "2026-05-04", "2026-05-25"]),
PreShiftConstraint(days=1, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun"), allow_self=False),
#PreShiftConstraint(days=2, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun"), exclude_dates=["2026-04-06", "2026-05-04", "2026-05-25"]),
#PreShiftConstraint(days=1, start_date="2025-11-17", exclude_days=("Fri", "Sat", "Sun"), allow_self=False),
MaxShiftsPerWeekConstraint(max_shifts=1, days=days[:5]),
],
),
@@ -533,7 +542,7 @@ def main(
SingleShift(
sites=("rota b", "rota c"),
name="weekend b",
end_date="2026-07-20",
#end_date="2026-07-20",
workers_required=1,
length=12.5,
days=days[5:],
@@ -685,6 +694,9 @@ def main(
subprocess.run(
["scp", Rota.exported_rota_file, "ross@46.101.13.46:proc/proc-rota/output/"]
)
subprocess.run(
["scp", "output/timetable.js", "ross@46.101.13.46:proc/proc-rota/output/"]
)
if suspend_on_finish:
os.system("systemctl suspend")
+264 -8
View File
@@ -410,17 +410,43 @@ function renderWorkerList() {
const filter = ($("#worker-filter").val() || '').toLowerCase();
const shiftFilter = ($("#worker-shift-select").val() || '');
// Parse the original pre block into per-worker text blocks so rich details are preserved.
const detailsByName = {};
const preText = (($wd.find("pre").first().text() || '') + '').trim();
const escapeHtml = (text) => String(text)
.replace(/&/g, '&amp;')
.replace(/</g, '&lt;')
.replace(/>/g, '&gt;')
.replace(/"/g, '&quot;')
.replace(/'/g, '&#39;');
if (preText.length > 0) {
preText.split(/\n\s*\n(?=Name:\s*)/g).forEach(block => {
const trimmed = (block || '').trim();
if (!trimmed) return;
const nameMatch = trimmed.match(/^Name:\s*(.+)$/m);
if (!nameMatch) return;
detailsByName[nameMatch[1].trim()] = trimmed;
});
}
let $list = $("#worker-list");
if ($list.length === 0) {
$list = $("<div id='worker-list' style='margin-top:8px;'></div>");
$("#worker_details").append($list);
}
// Hide the static pre block since we render interactive cards
$("#worker_details pre").hide();
$list.show();
$list.empty();
if (!workers) return;
workers.forEach(w => {
const name = (w.name || '') + '';
const site = (w.site || '') + '';
if (filter && name.toLowerCase().indexOf(filter) === -1 && site.toLowerCase().indexOf(filter) === -1) return;
if (filter.length > 0) {
const matchesName = name.toLowerCase().indexOf(filter) !== -1;
const matchesSite = site.toLowerCase().indexOf(filter) !== -1;
if (!matchesName && !matchesSite) return;
}
// if shiftFilter, ensure worker has target or assigned for that shift
// attempt to read per-worker targets and counts from DOM if not present in JSON
let targets = w.shift_target_number || {};
@@ -443,21 +469,242 @@ function renderWorkerList() {
if (!hasTarget && !hasAssigned) return;
}
const fte = w.fte || '';
const card = $(`<div class='worker-card' style='border:1px solid #ccc;padding:8px;margin-bottom:6px;'><strong>${name}</strong> <span style='color:#666'>${site}</span><div>FTE: ${fte}</div></div>`);
const grade = w.grade || '';
const id = w.id ? String(w.id).substring(0, 8) : '';
const card = $(`<div class='worker-card'><div class='worker-header'><strong>${name}</strong> <span class='worker-id'>${id}</span></div><div class='worker-meta'><span class='worker-site'>${site}</span> • <span class='worker-grade'>Grade: ${grade}</span> • <span class='worker-fte'>FTE: ${fte}%</span></div></div>`);
// show shift targets summary
if (w.shift_target_number) {
const tbl = $("<table style='margin-top:6px; width:100%;'></table>");
Object.keys(w.shift_target_number).forEach(s => {
const t = w.shift_target_number[s];
const c = (w.shift_counts && w.shift_counts[s]) ? w.shift_counts[s] : 0;
tbl.append(`<tr><td style='width:60%'>${s}</td><td style='text-align:right'>${c} assigned</td><td style='text-align:right'>target ${parseFloat(t).toFixed(2)}</td></tr>`);
if (targets && Object.keys(targets).length > 0) {
const tbl = $("<table class='worker-shifts-table'></table>");
const header = $("<thead><tr><th>Shift</th><th class='text-right'>Assigned</th><th class='text-right'>Target</th></tr></thead>");
tbl.append(header);
const tbody = $("<tbody></tbody>");
Object.keys(targets).forEach(s => {
const t = targets[s];
const c = (counts && counts[s]) ? counts[s] : 0;
tbody.append(`<tr><td>${s}</td><td class='text-right'>${c}</td><td class='text-right'>${parseFloat(t).toFixed(2)}</td></tr>`);
});
tbl.append(tbody);
card.append(tbl);
}
// Include the original full worker detail text block so no information is lost.
const fullDetails = detailsByName[name] || '';
if (fullDetails) {
card.append(`<details class='worker-full-details'><summary>Full details</summary><pre>${escapeHtml(fullDetails)}</pre></details>`);
}
$list.append(card);
});
}
// Add CSS for worker details styling and comprehensive dark mode support
if ($('#worker-details-style').length === 0) {
$('head').append(`<style id='worker-details-style'>
/* ===== Light mode defaults ===== */
body {
background-color: #fff;
color: #000;
}
#worker_details {
background: #fff;
color: #000;
}
#worker-filter, #worker-shift-select {
background: #fff;
color: #000;
border: 1px solid #ccc;
}
#worker-filter:focus, #worker-shift-select:focus {
outline: none;
border-color: #0066cc;
box-shadow: 0 0 4px rgba(0,102,204,0.3);
}
.worker-card {
border: 1px solid #ccc;
border-radius: 4px;
padding: 8px;
margin-bottom: 8px;
background: #fff;
color: #000;
transition: background-color 0.2s, border-color 0.2s;
}
.worker-card:hover {
background: #f5f5f5;
border-color: #999;
}
.worker-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 6px;
}
.worker-header strong {
flex: 1;
margin: 0;
color: #000;
}
.worker-id {
font-size: 0.8em;
color: #999;
font-family: monospace;
}
.worker-meta {
font-size: 0.9em;
color: #666;
margin-bottom: 6px;
}
.worker-site, .worker-grade, .worker-fte {
display: inline;
}
.worker-shifts-table {
margin-top: 6px;
width: 100%;
border-collapse: collapse;
font-size: 0.9em;
background: #fff;
color: #000;
}
.worker-shifts-table thead {
background: #f0f0f0;
font-weight: bold;
color: #000;
}
.worker-shifts-table th, .worker-shifts-table td {
padding: 4px 6px;
border: 1px solid #ddd;
color: #000;
}
.worker-shifts-table tr:nth-child(even) {
background: #fafafa;
}
.worker-full-details {
margin-top: 8px;
}
.worker-full-details summary {
cursor: pointer;
font-weight: 600;
}
.worker-full-details pre {
margin-top: 6px;
white-space: pre-wrap;
}
.text-right {
text-align: right;
}
pre {
background: #f5f5f5;
color: #000;
border: 1px solid #ddd;
}
input, select, textarea {
background: #fff;
color: #000;
border: 1px solid #ccc;
}
table {
background: #fff;
color: #000;
}
table th {
background: #f9f9f9;
color: #000;
}
table td {
border-color: #ddd;
}
/* ===== Dark mode overrides (toggled via html.dark-mode class) ===== */
html.dark-mode body {
background-color: #1e1e1e;
color: #e0e0e0;
}
html.dark-mode #worker_details {
background: #2d2d2d;
color: #e0e0e0;
}
html.dark-mode #worker-filter, html.dark-mode #worker-shift-select {
background: #3d3d3d;
color: #e0e0e0;
border: 1px solid #555;
}
html.dark-mode #worker-filter:focus, html.dark-mode #worker-shift-select:focus {
border-color: #0088ff;
box-shadow: 0 0 4px rgba(0,136,255,0.5);
}
html.dark-mode .worker-card {
background: #2d2d2d;
border-color: #444;
color: #e0e0e0;
}
html.dark-mode .worker-card:hover {
background: #333;
border-color: #666;
}
html.dark-mode .worker-header strong {
color: #e0e0e0;
}
html.dark-mode .worker-meta, html.dark-mode .worker-id {
color: #aaa;
}
html.dark-mode .worker-shifts-table {
background: #2d2d2d;
color: #e0e0e0;
}
html.dark-mode .worker-shifts-table thead {
background: #1e1e1e;
color: #e0e0e0;
}
html.dark-mode .worker-shifts-table th, html.dark-mode .worker-shifts-table td {
border-color: #444;
color: #e0e0e0;
}
html.dark-mode .worker-shifts-table tr:nth-child(even) {
background: #252525;
}
html.dark-mode .worker-full-details summary {
color: #cfd8e3;
}
html.dark-mode pre {
background: #1e1e1e;
color: #e0e0e0;
border: 1px solid #444;
}
html.dark-mode input, html.dark-mode select, html.dark-mode textarea {
background: #3d3d3d;
color: #e0e0e0;
border: 1px solid #555;
}
html.dark-mode table {
background: #2d2d2d;
color: #e0e0e0;
}
html.dark-mode table th {
background: #1e1e1e;
color: #e0e0e0;
}
html.dark-mode table td {
border-color: #444;
color: #e0e0e0;
}
html.dark-mode h1, html.dark-mode h2, html.dark-mode h3, html.dark-mode h4, html.dark-mode h5, html.dark-mode h6 {
color: #e0e0e0;
}
html.dark-mode a {
color: #4db8ff;
}
html.dark-mode a:visited {
color: #8866ff;
}
html.dark-mode button {
background: #3d3d3d;
color: #e0e0e0;
border: 1px solid #555;
}
html.dark-mode button:hover {
background: #4d4d4d;
}
</style>`);
}
// initialize panels
renderShiftSettings();
renderWorkerFilterOptions();
@@ -492,6 +739,15 @@ if ($('#tsummary-style').length === 0) {
/* Slightly thinner avoid highlight for compact layout */
td.rota-day.avoid-shift.avoid-highlight { border-width: 1.5px !important; box-shadow: 0 0 4px rgba(255,140,0,0.18); }
/* Higher-contrast dark mode for timetable readability */
html.dark-mode table.tsummary { background: #111922; color: #e8eef5; }
html.dark-mode table.tsummary th,
html.dark-mode table.tsummary td { border-color: #516274; }
html.dark-mode table.tsummary th { background: #253241; color: #f5f8fc; }
html.dark-mode table.tsummary td { background: #16212d; color: #e8eef5; }
html.dark-mode table.tsummary tr:nth-child(even) td { background: #1a2734; }
html.dark-mode table.tsummary td.target-assigned { color: #f2f6fb; }
</style>
`);
}
+104
View File
@@ -1055,6 +1055,35 @@ class RotaBuilder(object):
initialize=0,
)
# Helper binary variables for per-worker unique-shift block limits.
# Index is (worker_id, block_length, block_start_week, shift_name).
unique_shift_block_indices = []
for worker in self.workers:
for constraint in worker.max_unique_shifts_per_week_block:
for week_blocks in self.get_week_block_iterator(constraint.week_block):
if not week_blocks:
continue
block_start_week = week_blocks[0]
for shift_name in self.get_shift_names():
if any(
shift_name in self.get_shift_names_by_week_day(
week,
day,
return_empty_if_week_day_not_found=True,
)
for week in week_blocks
for day in days
):
unique_shift_block_indices.append(
(worker.id, constraint.week_block, block_start_week, shift_name)
)
self.model.worker_shift_used_in_week_block = Var(
unique_shift_block_indices,
within=Binary,
initialize=0,
)
if self.constraint_options_model.balance_bank_holidays:
# Bank holidays
self.model.bank_holiday_count = Var(
@@ -2374,6 +2403,81 @@ class RotaBuilder(object):
<= max_days
)
for constraint in worker.max_days_per_week_block:
for week_blocks in self.get_week_block_iterator(constraint.week_block):
# Determine which (week, day) combinations to ignore
ignored_week_days = set()
for ignore_date in (constraint.ignore_dates or []):
try:
w, d = self.date_week_day_map.get(ignore_date, (None, None))
if w is not None and d is not None:
ignored_week_days.add((w, d))
except Exception:
pass
included_days = [
(week, day)
for week, day in self.get_week_day_combinations()
if week in week_blocks and (week, day) not in ignored_week_days
]
if included_days:
self.model.constraints.add(
sum(
self.model.works_day[worker.id, week, day]
for week, day in included_days
)
<= constraint.max_days
)
for constraint in worker.max_unique_shifts_per_week_block:
for week_blocks in self.get_week_block_iterator(constraint.week_block):
if not week_blocks:
continue
# Determine which (week, day) combinations to ignore
ignored_week_days = set()
for ignore_date in (constraint.ignore_dates or []):
try:
w, d = self.date_week_day_map.get(ignore_date, (None, None))
if w is not None and d is not None:
ignored_week_days.add((w, d))
except Exception:
pass
block_start_week = week_blocks[0]
used_shift_vars = []
for shift_name in self.get_shift_names():
shift_assignments = []
for week, day in self.get_week_day_combinations():
if week in week_blocks and (week, day) not in ignored_week_days:
if shift_name in self.get_shift_names_by_week_day(
week,
day,
return_empty_if_week_day_not_found=True,
):
shift_assignments.append(
self.model.works[worker.id, week, day, shift_name]
)
if not shift_assignments:
continue
used_shift_var = self.model.worker_shift_used_in_week_block[
worker.id,
constraint.week_block,
block_start_week,
shift_name,
]
used_shift_vars.append(used_shift_var)
self.model.constraints.add(
sum(shift_assignments)
<= len(shift_assignments) * used_shift_var
)
if used_shift_vars:
self.model.constraints.add(
sum(used_shift_vars) <= constraint.max_unique_shifts
)
for week_blocks in self.get_week_block_iterator(4):
# Prevent more than n number shifts per 4 weeks
try:
+36
View File
@@ -135,6 +135,40 @@ class AvoidShiftOnDates(BaseModel):
raise ValueError(f"Cannot parse date: {v}")
class MaxDaysPerWeekBlockConstraint(BaseModel):
max_days: int = Field(
...,
ge=1,
description="Maximum number of worked days allowed inside each sliding week block.",
)
week_block: int = Field(
...,
ge=1,
description="Length of the sliding week block (for example, 2 means every consecutive 2-week window).",
)
ignore_dates: list[datetime.date] = Field(
default_factory=list,
description="List of specific dates to exclude from this constraint.",
)
class MaxUniqueShiftsPerWeekBlockConstraint(BaseModel):
max_unique_shifts: int = Field(
...,
ge=1,
description="Maximum number of distinct shift names allowed inside each sliding week block.",
)
week_block: int = Field(
...,
ge=1,
description="Length of the sliding week block (for example, 2 means every consecutive 2-week window).",
)
ignore_dates: list[datetime.date] = Field(
default_factory=list,
description="List of specific dates to exclude from this constraint.",
)
class Worker(BaseModel):
name: str
site: str
@@ -171,6 +205,8 @@ class Worker(BaseModel):
forced_assignments: List[tuple[int, str, str]] = [] # (week, day, shift_name)
forced_assignments_by_date: List[tuple[datetime.date, str]] = [] # (date, shift_name)
max_days_worked_per_week: int | None= None # Default: no limit, can be set to a specific number
max_days_per_week_block: list[MaxDaysPerWeekBlockConstraint] = []
max_unique_shifts_per_week_block: list[MaxUniqueShiftsPerWeekBlockConstraint] = []
assign_as_block_preferences: dict[str, float] = {} # shift_name -> weight (positive = prefer block)
+168 -1
View File
@@ -1,11 +1,19 @@
import datetime
import pytest
from pydantic import ValidationError
from rota_generator.shifts import (
InvalidShift, MaxShiftsPerWeekBlockConstraint, MaxShiftsPerWeekConstraint,
NoWorkers, PostShiftConstraint, PreShiftConstraint, RotaBuilder, SingleShift,
WarningTermination, days, WorkerRequirement
)
from rota_generator.workers import NotAvailableToWork, WorkRequests, Worker, generate_not_available_to_works
from rota_generator.workers import (
MaxDaysPerWeekBlockConstraint,
MaxUniqueShiftsPerWeekBlockConstraint,
NotAvailableToWork,
WorkRequests,
Worker,
generate_not_available_to_works,
)
def generate_basic_rota(weeks_to_rota=10, workers=2, start_date=datetime.date(2022, 3, 7)):
Rota = RotaBuilder(
@@ -89,6 +97,165 @@ def test_max_shifts_per_week_block_shift_number_fail():
Rota.export_rota_to_html("max_shifts_per_week_block")
assert Rota.results.solver.status in ("warning", "error")
def test_worker_max_days_per_week_block_fail():
Rota = RotaBuilder(datetime.date(2022, 3, 7), weeks_to_rota=2)
Rota.add_worker(
Worker(
name="worker1",
site="group1",
grade=1,
max_days_per_week_block=[
MaxDaysPerWeekBlockConstraint(max_days=1, week_block=1)
],
)
)
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=12.5,
days=days[:3],
workers_required=1,
)
)
Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.termination_condition == "infeasible"
def test_worker_max_unique_shifts_per_week_block_fail():
Rota = RotaBuilder(datetime.date(2022, 3, 7), weeks_to_rota=2)
Rota.add_worker(
Worker(
name="worker1",
site="group1",
grade=1,
max_unique_shifts_per_week_block=[
MaxUniqueShiftsPerWeekBlockConstraint(max_unique_shifts=1, week_block=1)
],
)
)
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=12.5,
days=["Mon"],
workers_required=1,
),
SingleShift(
sites=("group1",),
name="b",
length=12.5,
days=["Tue"],
workers_required=1,
),
)
Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.termination_condition == "infeasible"
def test_worker_block_constraint_validation():
with pytest.raises(ValidationError):
MaxDaysPerWeekBlockConstraint(max_days=0, week_block=1)
with pytest.raises(ValidationError):
MaxUniqueShiftsPerWeekBlockConstraint(max_unique_shifts=0, week_block=1)
def test_worker_max_days_per_week_block_pass():
"""Positive test: constraint allowing reasonable days."""
Rota = generate_basic_rota()
# Add constraint allowing up to 4 days per week
Rota.workers[0].max_days_per_week_block = [
MaxDaysPerWeekBlockConstraint(max_days=4, week_block=1)
]
# Add shifts to the rota
Rota.add_shifts(
SingleShift(
sites=("group1", "group2"),
name="a",
length=12.5,
days=days,
)
)
Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"
def test_worker_max_unique_shifts_per_week_block_pass():
"""Positive test: allowing multiple shifts within week block."""
Rota = generate_basic_rota()
# Add constraint allowing up to 2 unique shifts per week
Rota.workers[0].max_unique_shifts_per_week_block = [
MaxUniqueShiftsPerWeekBlockConstraint(max_unique_shifts=2, week_block=1)
]
# Add shifts to the rota
Rota.add_shifts(
SingleShift(
sites=("group1", "group2"),
name="a",
length=12.5,
days=days,
)
)
Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.termination_condition == "optimal"
def test_worker_max_days_per_week_block_with_ignore_dates():
"""Test max_days_per_week_block with ignored dates works without error."""
Rota = generate_basic_rota()
rota_start = Rota.start_date
ignored_date = rota_start # Ignore first Monday
# Add constraint with ignore_dates
Rota.workers[0].max_days_per_week_block = [
MaxDaysPerWeekBlockConstraint(
max_days=3,
week_block=1,
ignore_dates=[ignored_date],
)
]
# Add shifts to the rota
Rota.add_shifts(
SingleShift(
sites=("group1", "group2"),
name="a",
length=12.5,
days=days,
)
)
Rota.build_and_solve(options={"ratio": 0.000})
assert Rota.results.solver.status == "ok"
def test_worker_max_unique_shifts_per_week_block_with_ignore_dates():
"""Test max_unique_shifts_per_week_block with ignored dates works without error."""
Rota = generate_basic_rota()
rota_start = Rota.start_date
ignored_date = rota_start + datetime.timedelta(days=1) # Ignore Tuesday
# Add constraint with ignore_dates
Rota.workers[0].max_unique_shifts_per_week_block = [
MaxUniqueShiftsPerWeekBlockConstraint(
max_unique_shifts=1,
week_block=1,
ignore_dates=[ignored_date],
)
]
# Add shifts to the rota
Rota.add_shifts(
SingleShift(
sites=("group1", "group2"),
name="a",
length=12.5,
days=days,
)
)
Rota.build_and_solve(options={"ratio": 0.000})
# This should still be feasible with only one shift type in main rota
assert Rota.results.solver.status == "ok"
def test_pre_shift_constraint():
Rota = generate_basic_rota()
for i, d in enumerate(days[:6]):