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

numerous improvemts

Browse Source
This commit is contained in:
Ross
2025-06-08 14:00:15 +01:00
parent ad75b6b8b5
commit d7e5f9e5f1
6 changed files with 331 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rota/shifts.py
+137 -59
View File
@@ -257,7 +257,7 @@ class RotaBuilder(object):
self.SHIFT_BOUNDS = SHIFT_BOUNDS
self.shifts: List[SingleShift] = [] # type List[SingleShift]
self.allowed_multi_shift_sets: list[set[str]] = []
#self.allowed_multi_shift_sets: list[set[str]] = []
self.use_previous_shifts = use_previous_shifts
self.use_shift_balance_extra = use_shift_balance_extra
@@ -603,6 +603,16 @@ class RotaBuilder(object):
initialize=0,
)
self.model.multi_shift_together = Var(
((worker.id, week, day, idx)
for worker in self.workers
for week, day in self.get_week_day_combinations()
for idx, allowed_set in enumerate(getattr(worker, "allowed_multi_shift_sets", []))
if allowed_set.issubset(self.get_shift_names_by_week_day(week, day))),
domain=Binary,
initialize=0,
)
# Used to limit number of workers on night shift per site
# if we force a binary it will in effect hard constrain to < 2
# from a single site
@@ -1398,6 +1408,25 @@ class RotaBuilder(object):
)
logging.debug(f"Generate worker constraints: {worker.name}")
for week, day in self.get_week_day_combinations():
shifts_today = self.get_shift_names_by_week_day(week, day)
worker_allowed_sets = getattr(worker, "allowed_multi_shift_sets", [])
#for idx, allowed_set in enumerate(worker_allowed_sets):
# if allowed_set.issubset(shifts_today):
# var = self.model.multi_shift_together[worker.id, week, day, idx]
# self.model.constraints.add(
# sum(self.model.works[worker.id, week, day, shift] for shift in allowed_set)
# == len(allowed_set) * var
# )
for idx, allowed_set in enumerate(worker_allowed_sets):
if allowed_set.issubset(shifts_today):
var = self.model.multi_shift_together[worker.id, week, day, idx]
# var is 1 iff all shifts in allowed_set are assigned
for shift in allowed_set:
self.model.constraints.add(var <= self.model.works[worker.id, week, day, shift])
self.model.constraints.add(
var >= sum(self.model.works[worker.id, week, day, shift] for shift in allowed_set) - len(allowed_set) + 1
)
if self.get_workers_who_require_locums():
@@ -2416,55 +2445,40 @@ class RotaBuilder(object):
# single shift per day (unless multi-shift allowed)
# This is signifantly slower so only enable if required
if self.allowed_multi_shift_sets:
shifts_today = self.get_shift_names_by_week_day(week, day)
if shifts_today:
# Build all allowed sets that are subsets of today's shifts
allowed_sets_today = [
allowed_set for allowed_set in self.allowed_multi_shift_sets
if allowed_set.issubset(shifts_today)
]
shifts_today = self.get_shift_names_by_week_day(week, day)
assigned_vars = [self.model.works[worker.id, week, day, shift] for shift in shifts_today]
worker_allowed_sets = getattr(worker, "allowed_multi_shift_sets", [])
allowed_sets_today = [
allowed_set for allowed_set in worker_allowed_sets
if allowed_set.issubset(shifts_today)
]
# For each possible assignment, only allow:
# - a single shift
# - OR exactly one of the allowed sets
assigned_vars = [self.model.works[worker.id, week, day, shift] for shift in shifts_today]
# The sum of assigned shifts must be 1 (single shift) or exactly len(allowed_set) for one allowed set
allowed_sums = [1] + [len(allowed_set) for allowed_set in allowed_sets_today]
# Enforce: sum in allowed_sums
if allowed_sets_today:
# Forbid any multi-shift assignment that includes shifts from different allowed sets
# Build all possible multi-shift assignments
all_multi = [set(c) for i in range(2, len(shifts_today)+1) for c in itertools.combinations(shifts_today, i)]
# Build all allowed multi-shift patterns (all non-empty subsets of each allowed set)
allowed_patterns = [{shift} for shift in shifts_today]
for allowed_set in allowed_sets_today:
allowed_patterns += [set(s) for i in range(2, len(allowed_set)+1) for s in itertools.combinations(allowed_set, i)]
# Forbid all other multi-shift combinations
forbidden = [s for s in all_multi if s not in allowed_patterns]
for forbidden_set in forbidden:
self.model.constraints.add(
sum(assigned_vars) <= max(allowed_sums)
sum(self.model.works[worker.id, week, day, shift] for shift in forbidden_set)
<= len(forbidden_set) - 1
)
self.model.constraints.add(
sum(assigned_vars) >= min(allowed_sums)
)
# Now, for each allowed set, prevent partial assignment of the set
for allowed_set in allowed_sets_today:
# If any shift in allowed_set is assigned, all must be assigned
for shift in allowed_set:
self.model.constraints.add(
sum(self.model.works[worker.id, week, day, s] for s in allowed_set)
>= len(allowed_set) * self.model.works[worker.id, week, day, shift]
)
# No other shifts can be assigned together with the allowed set
for shift in shifts_today:
if shift not in allowed_set:
self.model.constraints.add(
sum(self.model.works[worker.id, week, day, s] for s in allowed_set)
+ self.model.works[worker.id, week, day, shift]
<= len(allowed_set)
)
# Otherwise only one shift per day
else:
# Still, at most all shifts per day
self.model.constraints.add(
1
>= sum(
self.model.works[worker.id, week, day, shift]
for shift in self.get_shift_names_by_week_day(week, day)
)
sum(self.model.works[worker.id, week, day, shift] for shift in shifts_today)
<= len(shifts_today)
)
else:
# Only one shift per day if no allowed sets
self.model.constraints.add(
sum(assigned_vars) <= 1
)
# This applies to locums as well
if self.get_workers_who_require_locums():
self.model.constraints.add(
@@ -2647,6 +2661,14 @@ class RotaBuilder(object):
# Define an objective function with model as input, to pass later
def obj_rule(m):
# c = len(workers)
prefer_multi_shift_expr = sum(
getattr(worker, "prefer_multi_shift_together", 0) *
self.model.multi_shift_together[worker.id, week, day, idx]
for worker in self.workers
for week, day in self.get_week_day_combinations()
for idx, allowed_set in enumerate(getattr(worker, "allowed_multi_shift_sets", []))
if allowed_set.issubset(self.get_shift_names_by_week_day(week, day))
)
balance_modifier_constant = 10
balance_quadratic_shift_modifier_constant = 3
@@ -2840,16 +2862,17 @@ class RotaBuilder(object):
- work_requests
+ locum_shift_balancing
+ true_quadratic_shift_balancing
- prefer_multi_shift_expr
)
# add objective function to the model. rule (pass function) or expr (pass expression directly)
self.model.obj = Objective(rule=obj_rule, sense=minimize)
def allow_shifts_together(self, *shift_names: str):
"""
Allow the given set of shift names to be assigned together on a single day.
"""
self.allowed_multi_shift_sets.append(set(shift_names))
#def allow_shifts_together(self, *shift_names: str):
# """
# Allow the given set of shift names to be assigned together on a single day.
# """
# self.allowed_multi_shift_sets.append(set(shift_names))
def add_warning(self, warning_type: str, message: str):
print(f"[bold red]WARNING:[/bold red] {warning_type} - {message}")
@@ -3032,6 +3055,13 @@ class RotaBuilder(object):
for n in range(int((end_date - start_date).days) + 1):
yield start_date + datetime.timedelta(n)
def allow_shifts_together_for_all_workers(self, *shift_names: str):
"""
Allow the given set of shift names to be assigned together on a single day for all workers.
"""
for worker in self.workers:
worker.allowed_multi_shift_sets.append(frozenset(shift_names))
def build_shifts(self):
"""
Process the added shifts
@@ -3151,6 +3181,16 @@ class RotaBuilder(object):
for shift in self.get_shifts_with_constraint("post"):
self.max_post = max(shift.constraint_options["post"], self.max_post)
# --- Check allowed_multi_shift_sets validity ---
all_shift_names = set(s.name for s in self.shifts)
for worker in self.workers:
for allowed_set in getattr(worker, "allowed_multi_shift_sets", []):
missing = set(allowed_set) - all_shift_names
if missing:
raise InvalidShift(
f"Worker '{worker.name}' has allowed_multi_shift_set(s) with non-existent shift(s): {missing}"
)
## For paired shifts we create proxy shifts
# self.paired_shift_map = {}
# for shifts in self.paired_shifts:
@@ -3414,6 +3454,18 @@ class RotaBuilder(object):
return group_workers
def get_workers_by_site(self, site: str) -> list[Worker]:
"""
Returns a list of workers assigned to a specific site.
Args:
site (str): The site name.
Returns:
list[Worker]: List of workers at the given site.
"""
return [worker for worker in self.workers if worker.site == site]
def get_workers_by_grade(self) -> Dict[int, list[Worker]]:
group_workers: dict[int, list[Worker]] = defaultdict(list)
for worker in self.workers:
@@ -3466,7 +3518,7 @@ class RotaBuilder(object):
# RESULTS
def export_rota_to_html(
self, filename: str = "rota", folder=None, timestamp_filename=True
self, filename: str = "rota", folder=None, timestamp_filename=False
):
if timestamp_filename:
filename = f"{filename}_{datetime.datetime.now().strftime('%Y%m%d-%H%M%S')}"
@@ -3562,15 +3614,15 @@ class RotaBuilder(object):
shifts = []
w = [f"{worker.name:20}"]
for week, day in self.get_week_day_combinations():
a = "-"
shift_display_char = "-"
shift_names = []
for shift in self.get_shift_names_by_week_day(week, day):
if model.works[worker.id, week, day, shift].value > 0.5:
shift_names.append(self.get_shift_by_name(shift).get_display_char())
if shift_names:
a = "".join(shift_names)
shift_display_char = "".join(shift_names)
shifts.extend(shift_names)
w.append(a)
w.append(shift_display_char)
shift_count = ""
for s in set(shifts):
shift_count = shift_count + f"{s}: {shifts.count(s)}, "
@@ -3676,7 +3728,7 @@ class RotaBuilder(object):
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
n = n + 1
a = "-"
shift_display_char = "-"
shift_names = []
assigned_shift_names = []
locum = False
@@ -3693,7 +3745,7 @@ class RotaBuilder(object):
assigned_shift_names.append(shift)
locum = True
if shift_names:
a = "<br/>".join(shift_names)
shift_display_char = "<br/>".join(["<span class='multi-shift-shift'>"+name+"</span>" for name in shift_names])
shifts.extend(assigned_shift_names)
shift_name = assigned_shift_names[0] if assigned_shift_names else ""
@@ -3704,7 +3756,6 @@ class RotaBuilder(object):
css_class += " multi-shift"
title = f"{shift_name} ({d})"
css_class = day
unavailable_reason = ""
if model.available[worker.id, week, day] > 0:
available = True
@@ -3751,7 +3802,13 @@ class RotaBuilder(object):
css_class = " ".join((css_class, "night-shift"))
shift_tds.append(
f"<td title='{','.join(shift_names)} ({d})' class='rota-day {css_class}' data-shift='{','.join(shift_names)}' data-available='{available}' data-unavailable_reason='{unavailable_reason}' data-date='{d}' data-week='{week}' data-day='{day}'{remote_site}{requests}{bank_holiday}>{a}</td>"
f"<td title='{','.join(shift_names)} ({d})' class='rota-day {css_class}'"
f" data-shift='{','.join(assigned_shift_names)}'"
f" data-shift-display='{','.join(shift_names)}'"
f" data-available='{available}'"
f" data-unavailable_reason='{unavailable_reason}'"
f" data-date='{d}' data-week='{week}' data-day='{day}'"
f"{remote_site}{requests}{bank_holiday}>{shift_display_char}</td>"
)
shift_count = ""
@@ -3961,7 +4018,7 @@ class RotaBuilder(object):
temp = ""
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0:
if self.model.works[worker.id, week, day, shift].value > 0.5:
shifts[d] = shift
return shifts
@@ -3994,6 +4051,27 @@ class RotaBuilder(object):
return shifts
def get_worker_shift_count(self, worker: Worker, shift_name:str) -> int:
"""Returns the number of shifts of a specific type assigned to a worker."""
count = 0
for week, day in self.get_week_day_combinations():
if shift_name in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift_name].value > 0.5:
count += 1
return count
def get_worker_days_worked(self, worker: Worker) -> int:
"""
Returns the number of unique days on which the worker is assigned at least one shift.
"""
days_worked = 0
for week, day in self.get_week_day_combinations():
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0.5:
days_worked += 1
break # Only count each day once, even if multiple shifts
return days_worked
def get_worker_shift_list_string(self, worker: Worker) -> str:
shifts = self.get_worker_shift_list(worker)