Enhance Rota Management System

- Updated .gitignore to exclude 'cons/' directory.
- Refactored gen_cons.py to extract leave and rota assignments from calendar data, and added functionality to handle shift assignments from rota data.
- Modified timetable.css to include styles for force-assigned shifts.
- Enhanced shifts.py to manage forced assignments and handle leave conflicts, including new methods for date-based assignments.
- Updated workers.py to introduce structured hard day dependencies and exclusions, allowing for more flexible scheduling.
- Adjusted test_workers.py to reflect changes in hard day dependency and exclusion methods, ensuring compatibility with new data structures.
This commit is contained in:
Ross
2025-08-09 22:11:39 +01:00
parent 6bd47ba5b4
commit 48710f50e8
6 changed files with 612 additions and 116 deletions
+2
View File
@@ -13,3 +13,5 @@ logs/
*.db *.db
web/rota web/rota
cons/
+342 -63
View File
@@ -1,3 +1,4 @@
from collections import defaultdict
import datetime import datetime
import os import os
import sys import sys
@@ -5,6 +6,7 @@ import time
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, WorkerRequirement, days from rota.shifts import NoWorkers, RotaBuilder, SingleShift, WorkerRequirement, days
import typer import typer
import subprocess import subprocess
import pandas as pd
from rota.workers import ( from rota.workers import (
Worker, Worker,
@@ -21,6 +23,275 @@ app = typer.Typer()
sites = ("rota a", "rota b", "rota c") sites = ("rota a", "rota b", "rota c")
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": ...}
"""
# 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:]]
# Iterate over each row (skip the header row)
# 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()}"
leave_requests = []
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":
continue
try:
date = datetime.datetime.strptime(date_str, "%Y-%m-%d").date()
except Exception:
continue # skip rows with invalid date
for i, worker in enumerate(worker_names):
if not worker or worker.lower() == "nan":
continue
leave_val = str(row.iloc[5 + i]).strip().lower() if len(row) > 5 + i else ""
if "no" in leave_val:
leave_requests.append({
"date": date,
"worker": worker,
})
elif "yes" in leave_val:
work_requests.append({
"date": date,
"worker": worker,
})
#print(leave_requests)
return leave_requests, work_requests, rota_data
def extract_shift_assignments_from_rota(rota_df):
"""
Extract shift assignments from rota_df.
- Data starts from the 3rd row (index 2).
- The first column is the week start date.
- For each day (Mon-Sun), two columns: twilight initials, night initials.
Returns a list of dicts: {"week_start": ..., "day": ..., "shift": ..., "initial": ...}
"""
days = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
shifts = ["twilight", "oncall"]
assignments = []
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}")
continue # skip rows with invalid date
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.
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
else:
continue # No night shift column for Sat/Sun
if col_idx >= len(row):
continue
initial = str(row.iloc[col_idx]).strip()
if initial and initial.lower() != "nan" and initial.lower() != "nat":
if weekend:
shift = "weekend"
assignments.append({
"week_start": week_start,
"week": idx-1, # 1-indexed week number
"day": day,
"date": week_start + datetime.timedelta(days=i),
"shift": shift,
"initial": initial,
})
# Build a mapping of assignments by worker initial
assignments_by_worker = {}
for assignment in assignments:
initial = assignment["initial"]
if initial not in assignments_by_worker:
assignments_by_worker[initial] = []
assignments_by_worker[initial].append(assignment)
return assignments, assignments_by_worker
def load_workers():
# Path to the ODS file
ods_path = "cons/Oncall rota requests + New rota (first draft) - September 2025.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"]
rota_df = xls["Rota"]
rota_b_path = "cons/Draft Rota B Sept 25-Feb 26.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
if pd.isna(date_val) or pd.isna(worker_val):
continue
try:
date = pd.to_datetime(date_val).date()
except Exception:
continue
worker = str(worker_val).strip()
if not worker or worker.lower() == "nan":
continue
rota_b_assignments[worker].append(date)
print(rota_b_assignments)
# You can now use rota_b_assignments as needed
# --- 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 = []
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()
if not name or name.lower() == "nan":
continue
availability = {}
requests = {}
for i, day in enumerate(weekday_cols):
val = row.iloc[2 + 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())
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)
assignments, assignments_by_worker = extract_shift_assignments_from_rota(rota_df)
print(assignments)
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
if initial == "ND":
start_date = "2025-09-29" # Non-working day worker starts later
if initial == "MF":
start_date = "2025-10-06" # 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-06" # 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():
if not available:
non_working_days.append(NonWorkingDays(day=day, start_date=datetime.datetime(2025, 11, 17)))
w = Worker(
name=worker_day["initial"],
site=rota_data[worker],
grade=1,
start_date=start_date,
end_date=end_date,
# Add leave requests for this worker
not_available_to_work=[
NotAvailableToWork(
date=req["date"],
reason="leave",
) for req in leave_requests if req["worker"] == worker
],
work_requests=[
WorkRequests(
date=req["date"],
shift="*"
) for req in work_requests if req["worker"] == worker
],
nwds=non_working_days,
)
if initial == "L1":
w.oop = [OutOfProgramme(
start_date="2025-09-15",
end_date="2025-10-14",
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
#if initial in ("RK", "HB", "DS", "L1", "L2", "L3", "GFM", "TS", "AB", "IR", "MF", "ND", "RG", "AA", "BR", "MS", "HA", "COD", "JL", "TB", "NHO"):
for assignment in assignments_by_worker[initial]:
#if initial == "NHO":
# pass
# #if assignment["shift"] == "oncall":
# # continue
# #if assignment["shift"] == "twilight":
# # continue
# #1if assignment["shift"] == "weekend":
# #1 continue
#print(assignment)
w.force_assign_shift(
week=assignment["week"],
day=assignment["day"],
shift_name=assignment["shift"],
)
if initial in rota_b_assignments:
for date in rota_b_assignments.get(initial, []):
w.force_assign_shift_by_date(
date=date,
shift_name="weekend b",
)
if w.site == "rota a":
w.prefer_multi_shift_together = 10
w.allow_shifts_together("oncall", "twilight")
#w.add_force_assign_with("twilight", "oncall")
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")
else:
w.add_hard_day_dependency({"Fri", "Sun"}, start_date="2025-11-17")
#w.set_max_shifts_per_week("weekend", 1)
workers.append(w)
return workers
@app.command() @app.command()
def main( def main(
@@ -29,7 +300,7 @@ def main(
time_to_run: int = 60 * 60, time_to_run: int = 60 * 60,
ratio: float = 0.001, ratio: float = 0.001,
start_date: datetime.datetime = "2025-09-01", start_date: datetime.datetime = "2025-09-01",
weeks: int = 22, weeks: int = 24,
bom: int = 1, bom: int = 1,
): ):
rota_start_date = start_date.date() rota_start_date = start_date.date()
@@ -43,8 +314,8 @@ def main(
) )
# Rota = RotaBuilder(start_date, weeks_to_rota=20, balance_offset_modifier=1) # Rota = RotaBuilder(start_date, weeks_to_rota=20, balance_offset_modifier=1)
Rota.constraint_options["balance_weekends"] = True Rota.constraint_options["balance_weekends"] = True
Rota.constraint_options["max_weekend_frequency"] = 4 #Rota.constraint_options["max_weekend_frequency"] = 2
Rota.constraint_options["max_shifts_per_week"] = 20 Rota.constraint_options["max_shifts_per_week"] = 6
# Rota.constraint_options["avoid_st2_first_month"] = True # Rota.constraint_options["avoid_st2_first_month"] = True
Rota.add_shifts( Rota.add_shifts(
@@ -68,7 +339,7 @@ def main(
workers_required=1, workers_required=1,
length=12.5, length=12.5,
days=days[5:], days=days[5:],
balance_offset=2, balance_offset=1,
constraint=[ constraint=[
#{"name": "pre", "options": 2}, #{"name": "pre", "options": 2},
#{"name": "post", "options": 1}, #{"name": "post", "options": 1},
@@ -77,28 +348,29 @@ def main(
#force_assign_with=["oncall"] #force_assign_with=["oncall"]
), ),
SingleShift( SingleShift(
sites=("rota b",), sites=("rota b", "rota c"),
name="weekend b", name="weekend b",
workers_required=1, workers_required=1,
length=12.5, length=12.5,
days=days[5:], days=days[5:],
balance_offset=2, balance_offset=1,
constraint=[ #constraint=[
{"name": "pre", "options": 1}, # {"name": "pre", "options": 1},
{"name": "post", "options": 1}, # {"name": "post", "options": 1},
], #],
display_char="b", display_char="b",
), ),
SingleShift( SingleShift(
sites=( sites=(
"rota a", "rota a",
"rota b", "rota b",
"rota d"
), ),
name="twilight", name="twilight",
workers_required=1, workers_required=1,
length=12.5, length=12.5,
days=days[:5], days=days[:5],
balance_offset=5, balance_offset=1,
#constraint=[ #constraint=[
# {"name": "pre", "options": 1}, # {"name": "pre", "options": 1},
# {"name": "post", "options": 1}, # {"name": "post", "options": 1},
@@ -120,74 +392,81 @@ def main(
Rota.build_shifts() Rota.build_shifts()
# Rota.add_grade_constraint_by_week([2], [1], Rota.get_shift_names()) # Rota.add_grade_constraint_by_week([2], [1], Rota.get_shift_names())
for n in range(1, 12):
w = Worker(name=f"A{n:02d}", site="rota a", grade=1)
#w.prefer_multi_shift_together = 10
#w.allow_shifts_together("oncall", "twilight")
#w.allow_shifts_together("oncall", "weekend")
#w.add_hard_day_dependency("Fri", "Sat", "Sun") # Build workers
w.add_force_assign_with("twilight", "oncall") workers = load_workers()
w.add_force_assign_with("weekend", "oncall") Rota.add_workers(workers)
w.set_max_shifts_per_week("twilight", 1)
if n <= 2:
pass
w.add_hard_day_dependency("Fri", "Sat", "Sun")
else:
w.add_hard_day_dependency("Fri", "Sun")
#w.set_max_shifts_per_week("weekend", 1)
# #w.add_hard_day_dependency("Sat", "Sun")
print(w) #for n in range(1, 12):
# w = Worker(name=f"A{n:02d}", site="rota a", grade=1)
# #w.prefer_multi_shift_together = 10
# #w.allow_shifts_together("oncall", "twilight")
# #w.allow_shifts_together("oncall", "weekend")
Rota.add_worker(w) # #w.add_hard_day_dependency("Fri", "Sat", "Sun")
w13 = Worker(name=f"A13", site="rota a", grade=1, start_date="2025-10-01") # w.add_force_assign_with("twilight", "oncall")
#w.prefer_multi_shift_together = 10 # w.add_force_assign_with("weekend", "oncall")
#w.allow_shifts_together("oncall", "twilight") # w.set_max_shifts_per_week("twilight", 1)
#w.allow_shifts_together("oncall", "weekend")
#w.add_hard_day_dependency("Fri", "Sat", "Sun") # if n <= 2:
# pass
# w.add_hard_day_dependency("Fri", "Sat", "Sun")
# else:
# w.add_hard_day_dependency("Fri", "Sun")
# #w.set_max_shifts_per_week("weekend", 1)
# # #w.add_hard_day_dependency("Sat", "Sun")
w13.add_force_assign_with("twilight", "oncall") # print(w)
w13.add_force_assign_with("weekend", "oncall")
#w.set_max_shifts_per_week("twilight", 1)
Rota.add_worker(w13) # Rota.add_worker(w)
w13 = Worker(name=f"A14", site="rota a", grade=1, start_date="2025-12-01") #w13 = Worker(name=f"A13", site="rota a", grade=1, start_date="2025-10-01")
#w.prefer_multi_shift_together = 10 ##w.prefer_multi_shift_together = 10
#w.allow_shifts_together("oncall", "twilight") ##w.allow_shifts_together("oncall", "twilight")
#w.allow_shifts_together("oncall", "weekend") ##w.allow_shifts_together("oncall", "weekend")
#w.add_hard_day_dependency("Fri", "Sat", "Sun") ##w.add_hard_day_dependency("Fri", "Sat", "Sun")
w13.add_force_assign_with("twilight", "oncall") #w13.add_force_assign_with("twilight", "oncall")
w13.add_force_assign_with("weekend", "oncall") #w13.add_force_assign_with("weekend", "oncall")
#w.set_max_shifts_per_week("twilight", 1) ##w.set_max_shifts_per_week("twilight", 1)
Rota.add_worker(w13) #Rota.add_worker(w13)
for w in range(14, 24): #w13 = Worker(name=f"A14", site="rota a", grade=1, start_date="2025-12-01")
w = Worker( ##w.prefer_multi_shift_together = 10
name=f"B{w}", ##w.allow_shifts_together("oncall", "twilight")
site="rota b", ##w.allow_shifts_together("oncall", "weekend")
grade=1,
nwds=[
NonWorkingDays(
day="Fri",
)
],
)
#w.add_hard_day_exclusion("Sat", "Sun")
w.set_max_shifts_per_week("twilight", 2)
w.set_max_shifts_per_week("weekend b", 1)
print(w) ##w.add_hard_day_dependency("Fri", "Sat", "Sun")
Rota.add_worker(w) #w13.add_force_assign_with("twilight", "oncall")
#w13.add_force_assign_with("weekend", "oncall")
##w.set_max_shifts_per_week("twilight", 1)
#Rota.add_worker(w13)
#for w in range(14, 24):
# w = Worker(
# name=f"B{w}",
# site="rota b",
# grade=1,
# nwds=[
# NonWorkingDays(
# day="Fri",
# )
# ],
# )
# #w.add_hard_day_exclusion("Sat", "Sun")
# w.set_max_shifts_per_week("twilight", 2)
# w.set_max_shifts_per_week("weekend b", 1)
# print(w)
# Rota.add_worker(w)
# Rota.build_workers() # Rota.build_workers()
# Rota.build_model() # Rota.build_model()
+5
View File
@@ -326,3 +326,8 @@ table.transposed th.bank-holiday {
.multi-shift { .multi-shift {
line-height: 1; line-height: 1;
} }
.force-assigned {
background-color: #ffe066 !important;
border: 2px solid #ff8800 !important;
}
+193 -23
View File
@@ -332,6 +332,7 @@ class RotaBuilder(object):
"Shift/invalid start date", "Shift/invalid start date",
"Shift/invalid end date", "Shift/invalid end date",
"Locum/no locum availability", "Locum/no locum availability",
"Force assignment/leave conflict", # Currently will always be infeasible
] ]
self.results = None self.results = None
@@ -381,11 +382,13 @@ class RotaBuilder(object):
# Generate a map for week, day combinations to a given date # Generate a map for week, day combinations to a given date
self.week_day_date_map = {} self.week_day_date_map = {}
self.date_week_day_map = {}
n: int = 0 n: int = 0
for week, day in self.get_week_day_combinations(): for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n) d = self.start_date + datetime.timedelta(n)
self.week_day_date_map[(week, day)] = d self.week_day_date_map[(week, day)] = d
self.date_week_day_map[d] = (week, day)
n = n + 1 n = n + 1
self.unavailable_to_work = set() self.unavailable_to_work = set()
@@ -1430,13 +1433,24 @@ class RotaBuilder(object):
for week, day, shift_name in getattr(worker, "forced_assignments", []): for week, day, shift_name in getattr(worker, "forced_assignments", []):
# Only add if this shift is valid for this week/day # Only add if this shift is valid for this week/day
if shift_name in self.get_shift_names_by_week_day(week, day, return_empty_if_week_day_not_found=True): if shift_name in self.get_shift_names_by_week_day(week, day, return_empty_if_week_day_not_found=True):
# Check for leave conflict
leave_conflict = False
if hasattr(self, "unavailable_to_work"):
if (worker.id, week, day) in self.unavailable_to_work:
leave_conflict = True
if leave_conflict:
self.add_warning(
"Force assignment/leave conflict",
f"Worker '{worker.name}' has a forced assignment for shift '{shift_name}' on week {week}, day {day} (date: {self.get_date_by_week_day(week, day)}), which conflicts with a leave request.",
)
self.model.constraints.add( self.model.constraints.add(
self.model.works[worker.id, week, day, shift_name] == 1 self.model.works[worker.id, week, day, shift_name] == 1
) )
else: else:
date = self.get_date_by_week_day(week, day)
self.add_warning( self.add_warning(
"Invalid forced assignment", "Invalid forced assignment",
f"Worker '{worker.name}' has forced assignment for non-existent shift: {shift_name} on {week}, {day}", f"Worker '{worker.name}' has forced assignment for non-existent shift: {shift_name} on week {week}, day {day} (date: {date})",
) )
if hasattr(worker, "max_shifts_per_week_by_shift_name"): if hasattr(worker, "max_shifts_per_week_by_shift_name"):
@@ -1468,8 +1482,23 @@ class RotaBuilder(object):
# Add hard shift dependencies for this worker # Add hard shift dependencies for this worker
if hasattr(worker, "hard_day_dependencies"): if hasattr(worker, "hard_day_dependencies"):
for day_group in worker.hard_day_dependencies: for dep in worker.hard_day_dependencies:
for week in self.weeks: day_group = dep.days
weeks_to_apply = self.weeks
if dep.weeks is not None:
weeks_to_apply = dep.weeks
if dep.start_date is not None or dep.end_date is not None:
if dep.start_date is None:
dep.start_date = self.start_date
if dep.end_date is None:
dep.end_date = self.rota_end_date
weeks_to_apply = self.get_weeks_by_date_range(
dep.start_date, dep.end_date
)
for week in weeks_to_apply:
assigned_any_shift = {} assigned_any_shift = {}
for day in day_group: for day in day_group:
shifts_today = self.get_shift_names_by_week_day(week, day) shifts_today = self.get_shift_names_by_week_day(week, day)
@@ -1490,20 +1519,31 @@ class RotaBuilder(object):
# Add hard day exclusions for this worker # Add hard day exclusions for this worker
if hasattr(worker, "hard_day_exclusions"): if hasattr(worker, "hard_day_exclusions"):
for day1, day2 in worker.hard_day_exclusions: for exclusion in worker.hard_day_exclusions:
for week in self.weeks: days_to_exclude = exclusion.days
weeks_to_apply = exclusion.weeks if exclusion.weeks is not None else self.weeks
for week in weeks_to_apply:
# For each week, prevent any shift being assigned on both days # For each week, prevent any shift being assigned on both days
self.model.constraints.add( self.model.constraints.add(
sum( sum(
self.model.works[worker.id, week, day1, shift] self.model.works[worker.id, week, day, shift]
for shift in self.get_shift_names_by_week_day(week, day1) for day in days_to_exclude
) for shift in self.get_shift_names_by_week_day(week, day)
+ sum(
self.model.works[worker.id, week, day2, shift]
for shift in self.get_shift_names_by_week_day(week, day2)
) )
<= 1 <= 1
) )
#self.model.constraints.add(
# sum(
# self.model.works[worker.id, week, day1, shift]
# for shift in self.get_shift_names_by_week_day(week, day1)
# )
# + sum(
# self.model.works[worker.id, week, day2, shift]
# for shift in self.get_shift_names_by_week_day(week, day2)
# )
# <= 1
#)
for week, day in self.get_week_day_combinations(): for week, day in self.get_week_day_combinations():
@@ -2075,13 +2115,20 @@ class RotaBuilder(object):
for n, start, end in worker.non_working_day_list: for n, start, end in worker.non_working_day_list:
if not shift.rota_on_nwds and day == n: if not shift.rota_on_nwds and day == n:
# print(start, self.week_day_date_map[(week, day)], end) # print(start, self.week_day_date_map[(week, day)], end)
# If this shift is force assigned, allow it but add a warning
if start <= self.week_day_date_map[(week, day)] < end: if start <= self.week_day_date_map[(week, day)] < end:
self.model.constraints.add( if (week, day, shift.name) in getattr(worker, "forced_assignments", []):
0 self.add_warning(
== self.model.works[ "Force assignment/NWD conflict",
worker.id, week, day, shift.name f"Worker '{worker.name}' has a forced assignment for shift '{shift.name}' on week {week}, day {day} (date: {self.get_date_by_week_day(week, day)}), which is a non-working day."
] )
) else:
self.model.constraints.add(
0
== self.model.works[
worker.id, week, day, shift.name
]
)
if self.get_workers_who_require_locums(): if self.get_workers_who_require_locums():
if not worker.locum_on_nwds and day == n: if not worker.locum_on_nwds and day == n:
@@ -3115,6 +3162,18 @@ class RotaBuilder(object):
message = f"Worker with name '{worker.name}' ({worker.id}) has no valid shifts (site: {worker.site})" message = f"Worker with name '{worker.name}' ({worker.id}) has no valid shifts (site: {worker.site})"
self.add_warning("Worker/no valid shifts", message) self.add_warning("Worker/no valid shifts", message)
for date, shift_name in worker.forced_assignments_by_date:
try:
week, day = self.get_week_day_by_date(date)
worker.force_assign_shift(week, day, shift_name)
except WeekDayNotFound:
self.add_warning(
"Worker/forced assignment date not found",
f"Worker {worker.name} ({worker.id}) has a forced assignment for {date} ({shift_name}) but this date is not in the rota",
)
continue
self.workers = sorted(self.workers) self.workers = sorted(self.workers)
self.full_time_equivalent = sum( self.full_time_equivalent = sum(
@@ -3142,6 +3201,7 @@ class RotaBuilder(object):
for p in pairs: for p in pairs:
self.worker_pairs.append(tuple(pairs[p])) self.worker_pairs.append(tuple(pairs[p]))
def add_worker_name_constraint_by_week( def add_worker_name_constraint_by_week(
self, names: Iterable[str], weeks: Iterable[int], shifts self, names: Iterable[str], weeks: Iterable[int], shifts
): ):
@@ -3241,6 +3301,25 @@ class RotaBuilder(object):
def get_date_by_week_day(self, week: WeekInt, day: DayStr) -> datetime.date: def get_date_by_week_day(self, week: WeekInt, day: DayStr) -> datetime.date:
return self.week_day_date_map[(week, day)] return self.week_day_date_map[(week, day)]
def get_weeks_by_date_range(self, start_date: datetime.date, end_date: datetime.date) -> list[WeekInt]:
"""Get a list of weeks that fall within a date range.
Args:
start_date (datetime.date): The start date of the range.
end_date (datetime.date): The end date of the range.
Returns:
list[WeekInt]: A list of week numbers that fall within the date range.
"""
# TODO: think about error if not monday start date?
weeks = []
for week in self.weeks:
week_start = self.get_date_by_week_day(week, "Mon")
week_end = self.get_date_by_week_day(week, "Sun")
if week_start >= start_date and week_end <= end_date:
weeks.append(week)
return weeks
def pair_shifts(self, *shifts: SingleShift): def pair_shifts(self, *shifts: SingleShift):
# NOTE: currently designed to pair two shifts # NOTE: currently designed to pair two shifts
# this could be extended... # this could be extended...
@@ -3503,6 +3582,21 @@ class RotaBuilder(object):
""" """
return self.weeks_days_product return self.weeks_days_product
def get_week_day_by_date(self, date: datetime.date) -> tuple[int, DayStr]:
"""Returns the week and day for a given date
Args:
date (datetime.date): Date to get week and day for
Returns:
tuple[int, DayStr]: Tuple containing week number and day string
"""
try:
week, day = self.date_week_day_map[date]
return week, day
except KeyError:
raise WeekDayNotFound(f"No week/day found for date {date}")
def get_week_day_combinations_for_shift(self, shift: SingleShift) -> list: def get_week_day_combinations_for_shift(self, shift: SingleShift) -> list:
return [ return [
(week, day) (week, day)
@@ -3799,12 +3893,14 @@ class RotaBuilder(object):
""" """
Force a specific worker to be assigned to a shift on a specific date. Force a specific worker to be assigned to a shift on a specific date.
""" """
# Find (week, day) for this date worker = self.get_worker_by_name(worker_name)
for (week, day), dt in self.week_day_date_map.items(): worker.forced_assignments_by_date.append((date, shift_name))
if dt == date:
self.force_assign_shift(worker_name, week, day, shift_name)
return
raise ValueError(f"Date {date} is not in the rota period.")
# RESULTS # RESULTS
@@ -4041,17 +4137,25 @@ class RotaBuilder(object):
assigned_shift_names = [] assigned_shift_names = []
locum = False locum = False
shift_name = "" shift_name = ""
force_assigned = False
for shift in self.get_shift_names_by_week_day(week, day): for shift in self.get_shift_names_by_week_day(week, day):
shift_obj = self.get_shift_by_name(shift) shift_obj = self.get_shift_by_name(shift)
if model.works[worker.id, week, day, shift].value > 0.8: if model.works[worker.id, week, day, shift].value > 0.8:
shift_names.append(shift_obj.get_display_char()) shift_names.append(shift_obj.get_display_char())
assigned_shift_names.append(shift) assigned_shift_names.append(shift)
# Check if this shift was force assigned
if (week, day, shift) in getattr(worker, "forced_assignments", []):
force_assigned = True
elif self.get_workers_who_require_locums(): elif self.get_workers_who_require_locums():
if model.locum_works[worker.id, week, day, shift].value > 0.8: if model.locum_works[worker.id, week, day, shift].value > 0.8:
locum_shifts.append(shift) locum_shifts.append(shift)
shift_names.append(shift_obj.get_display_char()) shift_names.append(shift_obj.get_display_char())
assigned_shift_names.append(shift) assigned_shift_names.append(shift)
locum = True locum = True
# Check if this shift was force assigned
if (week, day, shift) in getattr(worker, "forced_assignments", []):
force_assigned = True
if shift_names: if shift_names:
shift_display_char = "<br/>".join( shift_display_char = "<br/>".join(
[ [
@@ -4086,6 +4190,14 @@ class RotaBuilder(object):
unavailable_reason = "????" unavailable_reason = "????"
title = f"{title} / {unavailable_reason}" title = f"{title} / {unavailable_reason}"
# --- Highlight force assigned shifts ---
if force_assigned:
css_class += " force-assigned"
title = f"{shift_name} ({d}) [FORCE ASSIGNED]"
else:
title = f"{shift_name} ({d})"
bank_holiday = "" bank_holiday = ""
if d in self.bank_holidays: if d in self.bank_holidays:
title = f"{title} [Bank Holiday]" title = f"{title} [Bank Holiday]"
@@ -4243,11 +4355,62 @@ class RotaBuilder(object):
**limits, **limits,
}) })
# Add warnings export
warnings_html = "<details><summary><h2>Warnings</h2></summary><pre>\n"
for warning_type, message in self.warnings:
warnings_html += f"{warning_type}: {message}\n"
warnings_html += "</pre></details>"
# Human-readable worker details export
worker_details_human = []
for worker in self.get_all_workers():
details = [
f"Name: {worker.name}",
f"ID: {worker.id}",
f"Site: {worker.site}",
f"Grade: {worker.grade}",
f"FTE: {worker.fte}",
f"Remote site: {getattr(worker, 'remote_site', '')}",
f"Pair: {getattr(worker, 'pair', '')}",
f"Start date: {getattr(worker, 'calculated_start_date', '')}",
f"End date: {getattr(worker, 'calculated_end_date', '')}",
f"Bank holiday extra: {getattr(worker, 'bank_holiday_extra', '')}",
f"Shift balance extra: {getattr(worker, 'shift_balance_extra', '')}",
f"Non-working days: {getattr(worker, 'non_working_day_list', '')}",
f"OOP: {getattr(worker, 'oop', '')}",
f"Shift targets: {json.dumps(getattr(worker, 'shift_target_number', {}))}",
]
worker_details_human.append("\n".join(details))
worker_details_human_str = "\n\n".join(worker_details_human)
# Collect force assigned shifts for export
force_assigned_info = []
for worker in self.get_all_workers():
for assignment in getattr(worker, "forced_assignments", []):
# Support both tuple and pydantic model
if hasattr(assignment, "week") and hasattr(assignment, "day") and hasattr(assignment, "shift_name"):
week = assignment.week
day = assignment.day
shift_name = assignment.shift_name
else:
week, day, shift_name = assignment
date = self.get_date_by_week_day(week, day)
force_assigned_info.append({
"worker": worker.name,
"week": week,
"day": day,
"date": str(date),
"shift": shift_name,
})
html = f""" html = f"""
<body> <body>
<div class="table-div" id="{table_name}"> <div class="table-div" id="{table_name}">
<table id="main-table">{joined_timetable}</table> <table id="main-table">{joined_timetable}</table>
</div> </div>
{warnings_html}
<details> <details>
<summary><h2>Rota settings</h2></summary> <summary><h2>Rota settings</h2></summary>
<div> <div>
@@ -4267,6 +4430,9 @@ class RotaBuilder(object):
<div id="worker_details" data-workers='{json.dumps([json.loads(worker.model_dump_json()) for worker in self.get_all_workers()])}'> <div id="worker_details" data-workers='{json.dumps([json.loads(worker.model_dump_json()) for worker in self.get_all_workers()])}'>
<span id="workers-json-target"></span> <span id="workers-json-target"></span>
<pre>
{worker_details_human_str}
</pre>
</div> </div>
</details> </details>
<details> <details>
@@ -4297,6 +4463,10 @@ class RotaBuilder(object):
{"TEST"} {"TEST"}
{self.locum_availability_map} {self.locum_availability_map}
<br/> <br/>
<h3>Force assigned shifts</h3>
<pre>
{json.dumps(force_assigned_info, indent=4)}
</pre>
<div> <div>
</details> </details>
<details><summary><h2>Hard Constrain Shift Min/Max</h2></summary> <details><summary><h2>Hard Constrain Shift Min/Max</h2></summary>
+49 -9
View File
@@ -1,7 +1,7 @@
import datetime import datetime
from collections import defaultdict from collections import defaultdict
from typing import Iterable, List, Literal, Optional from typing import Iterable, List, Literal, Optional, Set
from pydantic import BaseModel, ConfigDict, field_validator from pydantic import BaseModel, ConfigDict, field_validator, Field
from rich.pretty import pprint from rich.pretty import pprint
from rota.console import console from rota.console import console
@@ -31,6 +31,31 @@ class NotAvailableToWork(BaseModel):
date: datetime.date date: datetime.date
reason: str = "unknown" reason: str = "unknown"
class HardDayDependency(BaseModel):
days: Set[str] = Field(..., description="List of days for the dependency group")
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
@field_validator("weeks")
def weeks_none_if_dates_set(cls, weeks, values):
data = values.data
if (data.get("start_date") is not None or data.get("end_date") is not None) and weeks is not None:
raise ValueError("If start_date or end_date is set, weeks must not be set")
return weeks
class HardDayExclusion(BaseModel):
days: set[str] = Field(..., description="List of days to exclude together")
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
@field_validator("weeks")
def weeks_none_if_dates_set(cls, weeks, values):
data = values.data
if (data.get("start_date") is not None or data.get("end_date") is not None) and weeks is not None:
raise ValueError("If start_date or end_date is set, weeks must not be set")
return weeks
def generate_not_available_to_works( def generate_not_available_to_works(
start_date: datetime.date, start_date: datetime.date,
@@ -114,11 +139,12 @@ class Worker(BaseModel):
allowed_multi_shift_sets: list[set] = [] # or list/set of frozensets allowed_multi_shift_sets: list[set] = [] # or list/set of frozensets
prefer_multi_shift_together: int = 0 # Default: no preference prefer_multi_shift_together: int = 0 # Default: no preference
# Set to a positive integer to prefer, negative to discourage, 0 for neutral # Set to a positive integer to prefer, negative to discourage, 0 for neutral
hard_day_dependencies: set[frozenset[str]] = set() # e.g. {frozenset({"Mon", "Tue"}), frozenset({"Fri", "Sun"})} hard_day_dependencies: list[HardDayDependency] = []
hard_day_exclusions: list[tuple[str, str]] = [] # e.g. [("Fri", "Sun")] hard_day_exclusions: list[HardDayExclusion] = []
force_assign_with: dict[str, list[str]] = {} # e.g. {"oncall": ["weekend"]} force_assign_with: dict[str, list[str]] = {} # e.g. {"oncall": ["weekend"]}
max_shifts_per_week_by_shift_name: dict[str, int] = {} # e.g. {"night": 2, "a": 3} max_shifts_per_week_by_shift_name: dict[str, int] = {} # e.g. {"night": 2, "a": 3}
forced_assignments: List[tuple[int, str, str]] = [] # (week, day, shift_name) forced_assignments: List[tuple[int, str, str]] = [] # (week, day, shift_name)
forced_assignments_by_date: List[tuple[datetime.date, str]] = [] # (date, shift_name)
assign_as_block_preferences: dict[str, float] = {} # shift_name -> weight (positive = prefer block) assign_as_block_preferences: dict[str, float] = {} # shift_name -> weight (positive = prefer block)
@@ -410,15 +436,25 @@ class Worker(BaseModel):
self.allowed_multi_shift_sets = [] self.allowed_multi_shift_sets = []
self.allowed_multi_shift_sets.append(frozenset(shifts)) self.allowed_multi_shift_sets.append(frozenset(shifts))
def add_hard_day_dependency(self, *days: str): 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):
"""
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.
"""
if not hasattr(self, "hard_day_dependencies"): if not hasattr(self, "hard_day_dependencies"):
self.hard_day_dependencies = set() self.hard_day_dependencies = []
self.hard_day_dependencies.add(frozenset(days)) self.hard_day_dependencies.append(HardDayDependency(days=days, weeks=weeks, start_date=start_date, end_date=end_date))
def add_hard_day_exclusion(self, day1: str, day2: str): 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):
"""
Add a hard day exclusion for this worker.
days: List of days that must be excluded together.
weeks: Optional list of weeks this exclusion applies to. If None, applies to all weeks.
"""
if not hasattr(self, "hard_day_exclusions"): if not hasattr(self, "hard_day_exclusions"):
self.hard_day_exclusions = [] self.hard_day_exclusions = []
self.hard_day_exclusions.append((day1, day2)) self.hard_day_exclusions.append(HardDayExclusion(days=days, weeks=weeks, start_date=start_date, end_date=end_date))
def add_force_assign_with(self, shift: str, with_shifts: list[str] | str): def add_force_assign_with(self, shift: str, with_shifts: list[str] | str):
if not hasattr(self, "force_assign_with"): if not hasattr(self, "force_assign_with"):
@@ -436,3 +472,7 @@ class Worker(BaseModel):
def force_assign_shift(self, week: int, day: str, shift_name: str): def force_assign_shift(self, week: int, day: str, shift_name: str):
"""Force this worker to be assigned to a shift on a specific week/day.""" """Force this worker to be assigned to a shift on a specific week/day."""
self.forced_assignments.append((week, day, shift_name)) self.forced_assignments.append((week, day, shift_name))
def force_assign_shift_by_date(self, date: datetime.date, shift_name: str):
"""Force this worker to be assigned to a shift on a specific date."""
self.forced_assignments_by_date.append((date, shift_name))
+21 -21
View File
@@ -1024,9 +1024,9 @@ def test_worker_hard_day_dependency():
worker2 = workers[1] worker2 = workers[1]
worker3 = workers[2] worker3 = workers[2]
worker1.add_hard_day_dependency("Mon", "Tue") worker1.add_hard_day_dependency({"Mon", "Tue"})
worker2.add_hard_day_dependency("Mon", "Wed") worker2.add_hard_day_dependency({"Mon", "Wed"})
worker3.add_hard_day_dependency("Mon", "Thu") worker3.add_hard_day_dependency({"Mon", "Thu"})
Rota.add_shifts( Rota.add_shifts(
SingleShift( SingleShift(
@@ -1080,15 +1080,15 @@ def test_worker_hard_day_dependency_weekend():
worker3 = workers[2] worker3 = workers[2]
worker4 = workers[3] worker4 = workers[3]
worker1.add_hard_day_dependency("Fri", "Sun") worker1.add_hard_day_dependency({"Fri", "Sun"})
worker2.add_hard_day_dependency("Fri", "Sun") worker2.add_hard_day_dependency({"Fri", "Sun"})
worker3.add_hard_day_dependency("Fri", "Sun") worker3.add_hard_day_dependency({"Fri", "Sun"})
worker4.add_hard_day_dependency("Fri", "Sun") worker4.add_hard_day_dependency({"Fri", "Sun"})
worker1.add_hard_day_exclusion("Fri", "Sat") worker1.add_hard_day_exclusion({"Fri", "Sat"})
worker2.add_hard_day_exclusion("Fri", "Sat") worker2.add_hard_day_exclusion({"Fri", "Sat"})
worker3.add_hard_day_exclusion("Fri", "Sat") worker3.add_hard_day_exclusion({"Fri", "Sat"})
worker4.add_hard_day_exclusion("Fri", "Sat") worker4.add_hard_day_exclusion({"Fri", "Sat"})
Rota.add_shifts( Rota.add_shifts(
SingleShift( SingleShift(
@@ -1128,15 +1128,15 @@ def test_worker_hard_day_dependency_weekend2():
worker3 = workers[2] worker3 = workers[2]
worker4 = workers[3] worker4 = workers[3]
worker1.add_hard_day_dependency("Fri", "Sun") worker1.add_hard_day_dependency({"Fri", "Sun"})
worker2.add_hard_day_dependency("Fri", "Sun") worker2.add_hard_day_dependency({"Fri", "Sun"})
worker3.add_hard_day_dependency("Fri", "Sun") worker3.add_hard_day_dependency({"Fri", "Sun"})
worker4.add_hard_day_dependency("Fri", "Sun") worker4.add_hard_day_dependency({"Fri", "Sun"})
worker4.add_hard_day_dependency("Fri", "Sat") worker4.add_hard_day_dependency({"Fri", "Sat"})
worker1.add_hard_day_exclusion("Fri", "Sat") worker1.add_hard_day_exclusion({"Fri", "Sat"})
worker2.add_hard_day_exclusion("Fri", "Sat") worker2.add_hard_day_exclusion({"Fri", "Sat"})
worker3.add_hard_day_exclusion("Fri", "Sat") worker3.add_hard_day_exclusion({"Fri", "Sat"})
Rota.add_shifts( Rota.add_shifts(
SingleShift( SingleShift(
@@ -1260,7 +1260,7 @@ def test_worker_force_assign_with_and_hard_days():
worker.add_force_assign_with("a", "b") worker.add_force_assign_with("a", "b")
#worker.add_hard_day_dependency("Sat", "Sun") #worker.add_hard_day_dependency("Sat", "Sun")
worker.add_hard_day_dependency("Mon", "Tue") worker.add_hard_day_dependency({"Mon", "Tue"})
Rota.add_shifts( Rota.add_shifts(
SingleShift( SingleShift(
@@ -1294,7 +1294,7 @@ def test_worker_force_assign_with_and_hard_days_weekend():
worker.add_force_assign_with("a", "b") worker.add_force_assign_with("a", "b")
#worker.add_hard_day_dependency("Sat", "Sun") #worker.add_hard_day_dependency("Sat", "Sun")
worker.add_hard_day_dependency("Sat", "Sun") worker.add_hard_day_dependency({"Sat", "Sun"})
Rota.add_shifts( Rota.add_shifts(
SingleShift( SingleShift(