diff --git a/.gitignore b/.gitignore
index 99c1a1b..7325ecf 100644
--- a/.gitignore
+++ b/.gitignore
@@ -13,3 +13,5 @@ logs/
*.db
web/rota
+
+cons/
\ No newline at end of file
diff --git a/gen_cons.py b/gen_cons.py
index fd5d6ed..dd21537 100644
--- a/gen_cons.py
+++ b/gen_cons.py
@@ -1,3 +1,4 @@
+from collections import defaultdict
import datetime
import os
import sys
@@ -5,6 +6,7 @@ import time
from rota.shifts import NoWorkers, RotaBuilder, SingleShift, WorkerRequirement, days
import typer
import subprocess
+import pandas as pd
from rota.workers import (
Worker,
@@ -21,6 +23,275 @@ app = typer.Typer()
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()
def main(
@@ -29,7 +300,7 @@ def main(
time_to_run: int = 60 * 60,
ratio: float = 0.001,
start_date: datetime.datetime = "2025-09-01",
- weeks: int = 22,
+ weeks: int = 24,
bom: int = 1,
):
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.constraint_options["balance_weekends"] = True
- Rota.constraint_options["max_weekend_frequency"] = 4
- Rota.constraint_options["max_shifts_per_week"] = 20
+ #Rota.constraint_options["max_weekend_frequency"] = 2
+ Rota.constraint_options["max_shifts_per_week"] = 6
# Rota.constraint_options["avoid_st2_first_month"] = True
Rota.add_shifts(
@@ -68,7 +339,7 @@ def main(
workers_required=1,
length=12.5,
days=days[5:],
- balance_offset=2,
+ balance_offset=1,
constraint=[
#{"name": "pre", "options": 2},
#{"name": "post", "options": 1},
@@ -77,28 +348,29 @@ def main(
#force_assign_with=["oncall"]
),
SingleShift(
- sites=("rota b",),
+ sites=("rota b", "rota c"),
name="weekend b",
workers_required=1,
length=12.5,
days=days[5:],
- balance_offset=2,
- constraint=[
- {"name": "pre", "options": 1},
- {"name": "post", "options": 1},
- ],
+ balance_offset=1,
+ #constraint=[
+ # {"name": "pre", "options": 1},
+ # {"name": "post", "options": 1},
+ #],
display_char="b",
),
SingleShift(
sites=(
"rota a",
"rota b",
+ "rota d"
),
name="twilight",
workers_required=1,
length=12.5,
days=days[:5],
- balance_offset=5,
+ balance_offset=1,
#constraint=[
# {"name": "pre", "options": 1},
# {"name": "post", "options": 1},
@@ -120,74 +392,81 @@ def main(
Rota.build_shifts()
# 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")
- w.add_force_assign_with("weekend", "oncall")
- w.set_max_shifts_per_week("twilight", 1)
+ workers = load_workers()
+ Rota.add_workers(workers)
- 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.prefer_multi_shift_together = 10
- #w.allow_shifts_together("oncall", "twilight")
- #w.allow_shifts_together("oncall", "weekend")
+ # w.add_force_assign_with("twilight", "oncall")
+ # w.add_force_assign_with("weekend", "oncall")
+ # w.set_max_shifts_per_week("twilight", 1)
- #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")
- w13.add_force_assign_with("weekend", "oncall")
- #w.set_max_shifts_per_week("twilight", 1)
+ # print(w)
- Rota.add_worker(w13)
+ # Rota.add_worker(w)
- w13 = Worker(name=f"A14", site="rota a", grade=1, start_date="2025-12-01")
- #w.prefer_multi_shift_together = 10
- #w.allow_shifts_together("oncall", "twilight")
- #w.allow_shifts_together("oncall", "weekend")
+ #w13 = Worker(name=f"A13", site="rota a", grade=1, start_date="2025-10-01")
+ ##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")
+ ##w.add_hard_day_dependency("Fri", "Sat", "Sun")
- w13.add_force_assign_with("twilight", "oncall")
- w13.add_force_assign_with("weekend", "oncall")
- #w.set_max_shifts_per_week("twilight", 1)
+ #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)
+ #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)
+ #w13 = Worker(name=f"A14", site="rota a", grade=1, start_date="2025-12-01")
+ ##w.prefer_multi_shift_together = 10
+ ##w.allow_shifts_together("oncall", "twilight")
+ ##w.allow_shifts_together("oncall", "weekend")
- 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_model()
diff --git a/output/timetable.css b/output/timetable.css
index 0adbba2..e4e8e0a 100644
--- a/output/timetable.css
+++ b/output/timetable.css
@@ -325,4 +325,9 @@ table.transposed th.bank-holiday {
.multi-shift {
line-height: 1;
+}
+
+.force-assigned {
+ background-color: #ffe066 !important;
+ border: 2px solid #ff8800 !important;
}
\ No newline at end of file
diff --git a/rota/shifts.py b/rota/shifts.py
index 1e498c3..4531d25 100644
--- a/rota/shifts.py
+++ b/rota/shifts.py
@@ -332,6 +332,7 @@ class RotaBuilder(object):
"Shift/invalid start date",
"Shift/invalid end date",
"Locum/no locum availability",
+ "Force assignment/leave conflict", # Currently will always be infeasible
]
self.results = None
@@ -381,11 +382,13 @@ class RotaBuilder(object):
# Generate a map for week, day combinations to a given date
self.week_day_date_map = {}
+ self.date_week_day_map = {}
n: int = 0
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
self.week_day_date_map[(week, day)] = d
+ self.date_week_day_map[d] = (week, day)
n = n + 1
self.unavailable_to_work = set()
@@ -1430,13 +1433,24 @@ class RotaBuilder(object):
for week, day, shift_name in getattr(worker, "forced_assignments", []):
# 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):
+ # 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.works[worker.id, week, day, shift_name] == 1
)
else:
+ date = self.get_date_by_week_day(week, day)
self.add_warning(
"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"):
@@ -1468,8 +1482,23 @@ class RotaBuilder(object):
# Add hard shift dependencies for this worker
if hasattr(worker, "hard_day_dependencies"):
- for day_group in worker.hard_day_dependencies:
- for week in self.weeks:
+ for dep in worker.hard_day_dependencies:
+ 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 = {}
for day in day_group:
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
if hasattr(worker, "hard_day_exclusions"):
- for day1, day2 in worker.hard_day_exclusions:
- for week in self.weeks:
+ for exclusion in worker.hard_day_exclusions:
+ 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
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)
+ self.model.works[worker.id, week, day, shift]
+ for day in days_to_exclude
+ for shift in self.get_shift_names_by_week_day(week, day)
)
<= 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():
@@ -2075,13 +2115,20 @@ class RotaBuilder(object):
for n, start, end in worker.non_working_day_list:
if not shift.rota_on_nwds and day == n:
# 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:
- self.model.constraints.add(
- 0
- == self.model.works[
- worker.id, week, day, shift.name
- ]
- )
+ if (week, day, shift.name) in getattr(worker, "forced_assignments", []):
+ self.add_warning(
+ "Force assignment/NWD 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 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 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})"
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.full_time_equivalent = sum(
@@ -3142,6 +3201,7 @@ class RotaBuilder(object):
for p in pairs:
self.worker_pairs.append(tuple(pairs[p]))
+
def add_worker_name_constraint_by_week(
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:
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):
# NOTE: currently designed to pair two shifts
# this could be extended...
@@ -3503,6 +3582,21 @@ class RotaBuilder(object):
"""
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:
return [
(week, day)
@@ -3799,12 +3893,14 @@ class RotaBuilder(object):
"""
Force a specific worker to be assigned to a shift on a specific date.
"""
- # Find (week, day) for this date
- for (week, day), dt in self.week_day_date_map.items():
- 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.")
+ worker = self.get_worker_by_name(worker_name)
+ worker.forced_assignments_by_date.append((date, shift_name))
+
+
+
+
+
+
# RESULTS
@@ -4041,17 +4137,25 @@ class RotaBuilder(object):
assigned_shift_names = []
locum = False
shift_name = ""
+ force_assigned = False
+
for shift in self.get_shift_names_by_week_day(week, day):
shift_obj = self.get_shift_by_name(shift)
if model.works[worker.id, week, day, shift].value > 0.8:
shift_names.append(shift_obj.get_display_char())
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():
if model.locum_works[worker.id, week, day, shift].value > 0.8:
locum_shifts.append(shift)
shift_names.append(shift_obj.get_display_char())
assigned_shift_names.append(shift)
locum = True
+ # Check if this shift was force assigned
+ if (week, day, shift) in getattr(worker, "forced_assignments", []):
+ force_assigned = True
if shift_names:
shift_display_char = "
".join(
[
@@ -4086,6 +4190,14 @@ class RotaBuilder(object):
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 = ""
if d in self.bank_holidays:
title = f"{title} [Bank Holiday]"
@@ -4243,11 +4355,62 @@ class RotaBuilder(object):
**limits,
})
+ # Add warnings export
+ warnings_html = "Warnings
\n"
+ for warning_type, message in self.warnings:
+ warnings_html += f"{warning_type}: {message}\n"
+ warnings_html += "
+ {worker_details_human_str}
+
+ {json.dumps(force_assigned_info, indent=4)}
+