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

a lot of changes

Browse Source
This commit is contained in:
Ross
2025-06-21 11:45:58 +01:00
parent e15cc6ee51
commit 06a551bb68
4 changed files with 659 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rota/shifts.py
+294 -152
View File
@@ -82,6 +82,7 @@ class WorkerRequirement(BaseModel):
"""
start date is inclusive, end date is not
"""
# NOTE: if number is changed mid week block allocations may not work?
number: int = 1
start_date: datetime.date | None = None
@@ -205,15 +206,15 @@ class SingleShift(BaseModel):
def get_worker_requirement_by_date(self, date: datetime.date):
"""Must be called after build_shifts
Returns the number of workers required for a given date
If none are required 0 is returned"""
if isinstance(self.workers_required, int):
return self.workers_required
for wr in self.workers_required:
#print(date, wr)
# print(date, wr)
if wr.start_date <= date < wr.end_date:
return wr.number
@@ -257,7 +258,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
@@ -306,7 +307,7 @@ class RotaBuilder(object):
"distribute_locum_shifts": True,
"balance_locum_shifts": False,
"maximum_allowed_locum_shifts_per_worker": None,
"minimum_allowed_locum_shifts_per_worker": None
"minimum_allowed_locum_shifts_per_worker": None,
}
self.terminate_on_warning = [
@@ -316,7 +317,6 @@ class RotaBuilder(object):
"Shift/invalid start date",
"Shift/invalid end date",
"Locum/no locum availability",
]
self.results = None
@@ -339,7 +339,7 @@ class RotaBuilder(object):
def set_rota_constraint(self, constraint: str, value: Any):
if constraint not in self.constraint_options:
raise ValueError(f"Constraint {constraint} not valid")
self.constraint_options[constraint] = value
def set_rota_dates(self, start_date: datetime.date, weeks_to_rota: int):
@@ -558,22 +558,10 @@ class RotaBuilder(object):
initialize=0,
)
self.model.worker_block_shift_split = Var(
(
(week, shift)
for week in self.weeks
for shift in self.shifts_to_assign_or_force_as_blocks()
),
within=Binary,
initialize=0,
)
self.model.blocks_worker_shift_assigned = Var(
(
(worker.id, week, shift)
for worker in self.workers
for week in self.weeks
for shift in self.shifts_to_assign_or_force_as_blocks()
for worker, week, shift in self.worker_week_shifts_to_assign_as_blocks()
),
within=Binary,
initialize=0,
@@ -582,8 +570,7 @@ class RotaBuilder(object):
self.model.blocks_assigned = Var(
(
(week, shift)
for week in self.weeks
for shift in self.shifts_to_assign_or_force_as_blocks()
for week, shift in self.week_shifts_to_assign_as_blocks()
),
within=NonNegativeIntegers,
initialize=0,
@@ -604,11 +591,15 @@ class RotaBuilder(object):
)
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))),
(
(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,
)
@@ -897,7 +888,6 @@ class RotaBuilder(object):
f"Invalid locum shift request [shift(s): ${invalid_shifts}]"
)
def locum_request_init(model, wid, week, day, shift):
if (wid, week, day, shift) in self.locum_availability:
self.locum_availability_map[(wid, week, day)] = shift
@@ -970,9 +960,9 @@ class RotaBuilder(object):
if self.get_workers_who_require_locums():
# We use this as a time check that their is some valid locum ability
if not self.get_all_locum_availability():
self.add_warning("Locum/no locum availability", "No locum availability set")
self.add_warning(
"Locum/no locum availability", "No locum availability set"
)
self.model.locum_required = Var(
(
@@ -994,28 +984,19 @@ class RotaBuilder(object):
)
self.model.locum_shifts_by_worker = Var(
(
worker.id
for worker in self.workers
),
(worker.id for worker in self.workers),
within=Integers,
initialize=0,
)
self.model.locum_shifts_t1 = Var(
(
worker.id
for worker in self.workers
),
(worker.id for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
self.model.locum_shifts_t2 = Var(
(
worker.id
for worker in self.workers
),
(worker.id for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
@@ -1058,7 +1039,7 @@ class RotaBuilder(object):
== sum(
self.model.locum_works[worker.id, week, day, shift]
for worker in self.workers
#if not worker.locum
# if not worker.locum
)
)
self.model.constraints.add(
@@ -1070,8 +1051,6 @@ class RotaBuilder(object):
)
)
# And it is not assigned if the worker is from the wrong site
if [worker for worker in self.workers if worker.site not in site_required]:
self.model.constraints.add(
@@ -1305,24 +1284,27 @@ class RotaBuilder(object):
weeks_days = self.get_week_day_combinations()
for week in track(self.weeks, description="Generating week constraints..."):
for shift_name in self.shifts_to_assign_or_force_as_blocks():
#for week in track(self.weeks, description="Generating week constraints..."):
# for shift_name in self.shifts_to_assign_or_force_as_blocks():
for week, shift_name in self.week_shifts_to_assign_as_blocks():
shift = self.get_shift_by_name(shift_name)
for worker in self.get_workers_for_shift(shift):
try:
self.model.constraints.add(
8
* self.model.blocks_worker_shift_assigned[
worker.id, week, shift_name
]
>= sum(
self.model.works[worker.id, week, day, shift_name]
for day in shift.days
if (worker.id, week, shift.name) in self.model.blocks_worker_shift_assigned:
try:
self.model.constraints.add(
8
* self.model.blocks_worker_shift_assigned[
worker.id, week, shift_name
]
>= sum(
self.model.works[worker.id, week, day, shift_name]
for day in shift.days
)
)
)
except KeyError:
pass
except KeyError:
pass
self.model.constraints.add(
self.model.blocks_assigned[week, shift_name]
== sum(
@@ -1330,6 +1312,7 @@ class RotaBuilder(object):
worker.id, week, shift_name
]
for worker in self.workers
if (worker.id, week, shift.name) in self.model.blocks_worker_shift_assigned
)
)
@@ -1408,10 +1391,40 @@ class RotaBuilder(object):
)
logging.debug(f"Generate worker constraints: {worker.name}")
# Add hard shift dependencies for this worker
if hasattr(worker, "hard_day_dependencies"):
for from_day, to_day in worker.hard_day_dependencies.items():
for week in self.weeks:
for shift in self.get_shift_names_by_week_day(week, from_day):
# Only apply if the shift also exists on the to_day
if shift in self.get_shift_names_by_week_day(week, to_day):
self.model.constraints.add(
self.model.works[worker.id, week, from_day, shift]
<= self.model.works[worker.id, week, to_day, shift]
)
# 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 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)
)
<= 1
)
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):
# 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(
@@ -1423,31 +1436,35 @@ class RotaBuilder(object):
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 <= 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
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():
for week, day, shift in self.get_all_shiftclass_combinations():
if not worker.locum:
self.model.constraints.add(
self.get_locum_availability(worker, week, day, shift)
>=
self.model.locum_works[worker.id, week, day, shift.name]
>= self.model.locum_works[worker.id, week, day, shift.name]
)
self.model.constraints.add(
self.model.locum_shifts_by_worker[worker.id] ==
sum(
self.model.locum_shifts_by_worker[worker.id]
== sum(
self.model.locum_works[worker.id, week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
)
)
self.model.constraints.add(
self.model.locum_shifts_t1[worker.id]
- self.model.locum_shifts_t2[worker.id]
@@ -1457,40 +1474,62 @@ class RotaBuilder(object):
if worker.locum_max_shifts:
self.model.constraints.add(
worker.locum_max_shifts
>=
sum(self.model.locum_works[worker.id, week, day, shiftname] for week, day, shiftname in self.get_all_shiftname_combinations())
>= sum(
self.model.locum_works[worker.id, week, day, shiftname]
for week, day, shiftname in self.get_all_shiftname_combinations()
)
)
if worker.locum_max_shifts_per_week:
for week_to_check in self.weeks:
self.model.constraints.add(
worker.locum_max_shifts_per_week
>=
sum(self.model.locum_works[worker.id, week, day, shiftname] for week, day, shiftname in self.get_all_shiftname_combinations(week=week_to_check))
>= sum(
self.model.locum_works[worker.id, week, day, shiftname]
for week, day, shiftname in self.get_all_shiftname_combinations(
week=week_to_check
)
)
)
if self.constraint_options["distribute_locum_shifts"]:
self.model.constraints.add(
sum(self.model.locum_required[week, day, shift] for week, day, shift in self.get_all_shiftname_combinations()) / (len(self.get_locum_workers())-1)
sum(
self.model.locum_required[week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
)
/ (len(self.get_locum_workers()) - 1)
>= self.model.locum_shifts_by_worker[worker.id]
)
if self.constraint_options["maximum_allowed_locum_shifts_per_worker"] is not None:
if (
self.constraint_options["maximum_allowed_locum_shifts_per_worker"]
is not None
):
self.model.constraints.add(
self.constraint_options["maximum_allowed_locum_shifts_per_worker"]
self.constraint_options[
"maximum_allowed_locum_shifts_per_worker"
]
>= self.model.locum_shifts_by_worker[worker.id]
)
if worker.locum_availability:
if self.constraint_options["minimum_allowed_locum_shifts_per_worker"] is not None:
if (
self.constraint_options[
"minimum_allowed_locum_shifts_per_worker"
]
is not None
):
self.model.constraints.add(
self.constraint_options["minimum_allowed_locum_shifts_per_worker"]
self.constraint_options[
"minimum_allowed_locum_shifts_per_worker"
]
<= self.model.locum_shifts_by_worker[worker.id]
)
if self.constraint_options["minimise_shift_diffs"] or self.constraint_options["maximum_allowed_shift_diff"] is not None:
if (
self.constraint_options["minimise_shift_diffs"]
or self.constraint_options["maximum_allowed_shift_diff"] is not None
):
self.model.constraints.add(
self.model.shift_count_diff_summed[worker.id]
== sum(
@@ -1606,19 +1645,20 @@ class RotaBuilder(object):
# calaculate the total number of shifts that need assigning over the rota
# total_shifts = (len(self.weeks) * len(shift.days) *
# TODO: check if shift_counts is still required?
#total_shifts = (
# total_shifts = (
# self.shift_counts[shift.name] * shift.workers_required
#)
total_shifts = (
self.shift_worker_counts[shift.name]
)
# )
total_shifts = self.shift_worker_counts[shift.name]
full_time_equivalent_joined = sum(
self.full_time_equivalent_sites[i][shift.name] for i in shift.sites
self.full_time_equivalent_sites[i][shift.name]
for i in shift.sites
)
target_shifts = (
total_shifts / full_time_equivalent_joined * worker.get_fte(shift=shift.name)
total_shifts
/ full_time_equivalent_joined
* worker.get_fte(shift=shift.name)
)
if self.use_previous_shifts:
@@ -2233,21 +2273,20 @@ class RotaBuilder(object):
# for shift in self.get_shifts_with_constraint("night"):
for shift in self.get_shifts(week=week):
if shift.force_as_block:
if shift.get_worker_requirement_by_date(
self.get_week_start_date(week)
):
# Force shifts to be assigned in blocks
self.model.constraints.add(
self.model.shift_week_worker_assigned[
shift.name, week, worker.id
]
* len(shift.days)
== sum(
self.model.works[worker.id, week, day, shift.name]
for day in shift.days
)
self.model.constraints.add(
self.model.shift_week_worker_assigned[
shift.name, week, worker.id
]
* len(shift.days)
== sum(
self.model.works[worker.id, week, day, shift.name]
for day in shift.days
)
)
#
@@ -2446,38 +2485,56 @@ class RotaBuilder(object):
# single shift per day (unless multi-shift allowed)
# This is signifantly slower so only enable if required
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", [])
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
allowed_set
for allowed_set in worker_allowed_sets
if allowed_set.issubset(shifts_today)
]
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)]
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)]
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(self.model.works[worker.id, week, day, shift] for shift in forbidden_set)
sum(
self.model.works[worker.id, week, day, shift]
for shift in forbidden_set
)
<= len(forbidden_set) - 1
)
# Still, at most all shifts per day
self.model.constraints.add(
sum(self.model.works[worker.id, week, day, shift] for shift in shifts_today)
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
)
self.model.constraints.add(sum(assigned_vars) <= 1)
# This applies to locums as well
if self.get_workers_who_require_locums():
@@ -2491,7 +2548,8 @@ class RotaBuilder(object):
self.model.constraints.add(
1
>= sum(
self.model.works[worker.id, week, day, shift] + self.model.locum_works[worker.id, week, day, shift]
self.model.works[worker.id, week, day, shift]
+ self.model.locum_works[worker.id, week, day, shift]
for shift in self.get_shift_names_by_week_day(week, day)
)
)
@@ -2628,13 +2686,13 @@ class RotaBuilder(object):
)
## Ensure each shift has the requisit number of workers assigned
#for (
# for (
# week,
# day,
# shift,
# workers_required,
# site_required,
#) in self.get_required_workers_and_site_combinations():
# ) in self.get_required_workers_and_site_combinations():
# # print(week, day, shift, workers_required, site_required)
# self.model.constraints.add(
# workers_required
@@ -2645,7 +2703,7 @@ class RotaBuilder(object):
# )
# )
#for worker in self.workers:
# for worker in self.workers:
# if worker.locum:
# for week, day, shift in self.get_all_shiftname_combinations():
# self.model.constraints.add(
@@ -2662,11 +2720,13 @@ class RotaBuilder(object):
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]
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", []))
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))
)
@@ -2679,9 +2739,11 @@ class RotaBuilder(object):
if self.get_workers_who_require_locums():
if self.constraint_options["balance_locum_shifts"]:
locum_shift_balancing = sum(
locum_shift_balance_modifier_constant *
(self.model.locum_shifts_t1[(worker.id)]
+ self.model.locum_shifts_t2[(worker.id)])
locum_shift_balance_modifier_constant
* (
self.model.locum_shifts_t1[(worker.id)]
+ self.model.locum_shifts_t2[(worker.id)]
)
for worker in self.workers
)
@@ -2731,7 +2793,6 @@ class RotaBuilder(object):
* self.model.shift_count_diff_summed[(worker.id)]
for worker in self.workers
)
if self.constraint_options["minimise_shift_diffs"]:
shift_diff_modifier_constant = 10
@@ -2832,13 +2893,27 @@ class RotaBuilder(object):
if self.constraint_options["balance_blocks"]:
blocks_balancing = sum(
block_shift_balancing_constant * self.model.blocks_assigned[week, shift]
block_shift_balancing_constant
* self.model.blocks_assigned[week, shift]
for week in self.weeks
for shift in self.shifts_to_assign_as_blocks()
)
else:
blocks_balancing = 0
prefer_block_expr = sum(
worker.assign_as_block_preferences.get(shift.name, 0)
* self.model.blocks_worker_shift_assigned[worker.id, week, shift.name]
for worker in self.workers
for week in self.weeks
for shift in self.shifts
if hasattr(worker, "assign_as_block_preferences")
and shift.name in worker.assign_as_block_preferences
and (worker.id, week, shift.name)
in self.model.blocks_worker_shift_assigned
and worker.site in shift.sites
)
# Quadratic :(
# worker_pairs_balancing = 0
# worker_pairs_constant = 100000
@@ -2863,12 +2938,13 @@ class RotaBuilder(object):
+ locum_shift_balancing
+ true_quadratic_shift_balancing
- prefer_multi_shift_expr
+ prefer_block_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):
# def allow_shifts_together(self, *shift_names: str):
# """
# Allow the given set of shift names to be assigned together on a single day.
# """
@@ -2937,7 +3013,9 @@ class RotaBuilder(object):
self.workers = sorted(self.workers)
self.full_time_equivalent = sum(w.fte_adj for w in self.workers) # TODO: rewrite as per shift
self.full_time_equivalent = sum(
w.fte_adj for w in self.workers
) # TODO: rewrite as per shift
self.full_time_equivalent_sites = {}
self.workers_at_sites = {}
@@ -3043,9 +3121,11 @@ class RotaBuilder(object):
raise ValueError(f"Must pair at least two shifts: {shifts}")
self.paired_shifts.append([*shifts])
def get_date_range(self, start_date: datetime.date=None, end_date: datetime.date = None):
def get_date_range(
self, start_date: datetime.date = None, end_date: datetime.date = None
):
"""Gets a range of dates
If either start_date or end_date are not provided defaults to the rota dates"""
if start_date is None:
start_date = self.start_date
@@ -3060,7 +3140,7 @@ class RotaBuilder(object):
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))
worker.allow_shifts_together(*shift_names)
def build_shifts(self):
"""
@@ -3111,24 +3191,23 @@ class RotaBuilder(object):
raise InvalidShift(
f"Invalid worker requirement: {worker_requirement}"
)
if worker_requirement.start_date is None:
worker_requirement.start_date = s.start_date
elif worker_requirement.start_date > s.end_date:
self.add_warning(
"Shift/worker requirement start date",
f"Worker requirement start date is after shift end date: {worker_requirement.start_date} > {s.end_date}"
f"Worker requirement start date is after shift end date: {worker_requirement.start_date} > {s.end_date}",
)
if worker_requirement.end_date is None:
worker_requirement.end_date = s.end_date
elif worker_requirement.end_date < s.start_date:
self.add_warning(
"Shift/worker requirement end date",
f"Worker requirement end date is before shift start date: {worker_requirement.end_date} < {s.start_date}"
f"Worker requirement end date is before shift start date: {worker_requirement.end_date} < {s.start_date}",
)
self.shifts_by_name[s.name] = s
self.shift_names.append(s.name)
@@ -3164,14 +3243,18 @@ class RotaBuilder(object):
self.week_day_shiftclass_product.append((week, day, s))
self.week_day_shifts_dict[(week, day)].add(s.name)
self.shift_counts[s.name] = self.shift_counts[s.name] + 1
self.shift_worker_counts[s.name] = self.shift_worker_counts[s.name] + workers_required
self.shift_worker_counts[s.name] = (
self.shift_worker_counts[s.name] + workers_required
)
else:
if day in s.days:
self.week_day_shift_product.append((week, day, s.name))
self.week_day_shiftclass_product.append((week, day, s))
self.week_day_shifts_dict[(week, day)].add(s.name)
self.shift_counts[s.name] = self.shift_counts[s.name] + 1
self.shift_worker_counts[s.name] = self.shift_worker_counts[s.name] + workers_required
self.shift_worker_counts[s.name] = (
self.shift_worker_counts[s.name] + workers_required
)
self.max_pre = 1
self.max_post = 1
@@ -3343,11 +3426,14 @@ class RotaBuilder(object):
return [self.get_shift_by_name(s) for s in shift_names]
def get_shift_names(self) -> List[ShiftName]:
"""Returns a list of all the registered shift names
(
"""Returns a list of all the registered shift names
Returns:
List[ShiftName]: List of names of all available shifts
""" """ """
"""
""" """
)
return self.shift_names
def get_week_block_iterator(self, block_length: int):
@@ -3420,12 +3506,49 @@ class RotaBuilder(object):
s = self.shifts_to_assign_as_blocks()
s.extend(self.shifts_to_force_as_blocks())
return s
def worker_week_shifts_to_assign_as_blocks(self) -> List[Tuple[Worker, int, str]]:
"""Returns a list of tuples containing worker id, week and shift name for shifts to assign as blocks"""
return [
(worker, week, shift.name)
for worker in self.workers
for week in self.weeks
for shift in self.get_shifts()
if (
shift.assign_as_block
or shift.force_as_block
or shift.force_as_block_unless_nwd
or (
hasattr(worker, "assign_as_block_preferences")
and shift.name in getattr(worker, "assign_as_block_preferences", {})
and getattr(worker, "assign_as_block_preferences")[shift.name] != 0
)
)
]
def week_shifts_to_assign_as_blocks(self) -> List[Tuple[int, str]]:
"""Returns a list of tuples containing week and shift name for shifts to assign as blocks"""
return set([
(week, shift.name)
for worker in self.workers
for week in self.weeks
for shift in self.get_shifts()
if (
shift.assign_as_block
or shift.force_as_block
or shift.force_as_block_unless_nwd
or (
hasattr(worker, "assign_as_block_preferences")
and shift.name in getattr(worker, "assign_as_block_preferences", {})
and getattr(worker, "assign_as_block_preferences")[shift.name] != 0
)
)
])
def get_all_locum_availability(self):
return self.locum_availability_map
def get_locum_availability(self, worker: Worker, week: int, day: DayStr, shift):
if (worker.id, week, day) in self.locum_availability_map:
if shift.name in self.locum_availability_map[(worker.id, week, day)]:
return 1
@@ -3502,7 +3625,11 @@ class RotaBuilder(object):
return [worker for worker in self.get_all_workers() if worker.locum]
def get_locum_workers(self):
return [worker for worker in self.get_all_workers() if not worker.locum and worker.locum_availability]
return [
worker
for worker in self.get_all_workers()
if not worker.locum and worker.locum_availability
]
def get_bank_holiday_week_days(self):
return [
@@ -3583,7 +3710,6 @@ class RotaBuilder(object):
week_table[week][day][shift].append(worker.get_details())
return week_table
def get_worker_timetable(self):
works = self.model.works
timetable = {
@@ -3600,7 +3726,6 @@ class RotaBuilder(object):
timetable[worker.name][week][day] = assigned_shifts
return timetable
def get_worker_timetable_brief(
self, show_prefs=False, marker_every=30, show_unavailable=False
):
@@ -3618,7 +3743,9 @@ class RotaBuilder(object):
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())
shift_names.append(
self.get_shift_by_name(shift).get_display_char()
)
if shift_names:
shift_display_char = "".join(shift_names)
shifts.extend(shift_names)
@@ -3674,7 +3801,7 @@ class RotaBuilder(object):
timetable.append(
f"<div>Start time: <span id='run_start_time'>{self.run_start_time:%Y-%m-%d %H:%M:%S%z}</span></div>"
f"<div>End time: <span id='run_end_time'>{self.run_end_time:%Y-%m-%d %H:%M:%S%z}</span></div>"
f"<div>Elapsed: <span id='run_elapsed_time'>{self.run_end_time-self.run_start_time}</span></div>"
f"<div>Elapsed: <span id='run_elapsed_time'>{self.run_end_time - self.run_start_time}</span></div>"
)
timetable.append(
@@ -3745,7 +3872,12 @@ class RotaBuilder(object):
assigned_shift_names.append(shift)
locum = True
if shift_names:
shift_display_char = "<br/>".join(["<span class='multi-shift-shift'>"+name+"</span>" for name in 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 ""
@@ -3754,7 +3886,7 @@ class RotaBuilder(object):
css_class = day
if multi_shift:
css_class += " multi-shift"
title = f"{shift_name} ({d})"
unavailable_reason = ""
if model.available[worker.id, week, day] > 0:
@@ -4002,9 +4134,9 @@ class RotaBuilder(object):
}
for worker in self.workers:
for shift in self.get_shift_names():
timetable[worker.name][shift] = round(self.model.shift_count[
worker.id, shift
].value)
timetable[worker.name][shift] = round(
self.model.shift_count[worker.id, shift].value
)
return timetable
@@ -4045,13 +4177,16 @@ class RotaBuilder(object):
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0.90:
temp = shift
elif include_locums and self.model.locum_works[worker.id, week, day, shift].value > 0.90:
elif (
include_locums
and self.model.locum_works[worker.id, week, day, shift].value > 0.90
):
temp = shift
shifts.append(temp)
return shifts
def get_worker_shift_count(self, worker: Worker, shift_name:str) -> int:
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():
@@ -4076,7 +4211,12 @@ class RotaBuilder(object):
shifts = self.get_worker_shift_list(worker)
# Convert shift to a string representation
return "".join([self.get_shift_by_name(i).get_display_char() if i != "" else "-" for i in shifts])
return "".join(
[
self.get_shift_by_name(i).get_display_char() if i != "" else "-"
for i in shifts
]
)
def get_shift_summary(self):
works = self.model.works
@@ -4108,7 +4248,9 @@ class RotaBuilder(object):
def get_shift_summary_html(self):
works = self.model.works
timetable = {
worker.get_details(): {shift.get_display_char(): "" for shift in self.get_shifts()}
worker.get_details(): {
shift.get_display_char(): "" for shift in self.get_shifts()
}
for worker in self.workers
}
# timetable = { worker.get_details() : { shift : "" for shift in ["truro_twilight"] } for worker in workers }
rota/workers.py
+15 -1
View File
@@ -114,6 +114,10 @@ class Worker(BaseModel):
allowed_multi_shift_sets: list[set] = [] # or list/set of frozensets
prefer_multi_shift_together: int = 0 # Default: no preference
# Set to a positive integer to prefer, negative to discourage, 0 for neutral
hard_day_dependencies: dict[str, str] = {} # e.g. {"Fri": "Sun"}
hard_day_exclusions: list[tuple[str, str]] = [] # e.g. [("Fri", "Sun")]
assign_as_block_preferences: dict[str, float] = {} # shift_name -> weight (positive = prefer block)
shift_fte_overrides: dict[str, int] = {} # Need checks to ensure shifts exist
@@ -396,4 +400,14 @@ class Worker(BaseModel):
def allow_shifts_together(self, *shifts):
if not hasattr(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, from_day: str, to_day: str):
if not hasattr(self, "hard_day_dependencies"):
self.hard_day_dependencies = {}
self.hard_day_dependencies[from_day] = to_day
def add_hard_day_exclusion(self, day1: str, day2: str):
if not hasattr(self, "hard_day_exclusions"):
self.hard_day_exclusions = []
self.hard_day_exclusions.append((day1, day2))
test/test_multiple_shifts_per_day.py
+47
View File
@@ -227,3 +227,50 @@ def test_cons_rota2(cons_rota):
# Check that days worked is now more than oncall count
days_worked = Rota.get_worker_days_worked(worker)
assert days_worked > oncall_count, f"{worker.name} worked {days_worked} days, more than oncall shifts {oncall_count}"
def test_cons_rota3(cons_rota):
Rota = cons_rota
Rota.terminate_on_warning.remove("Worker/no valid shifts")
Rota.add_shifts(
#SingleShift(sites=["a"], name="oncall weekday", length=8, days=days[:4], workers_required=1,
# constraint=[
# {"name": "pre", "options": 2},
# {"name": "post", "options": 2},
# ],
# ),
##SingleShift(sites=["a"], name="oncall weekend", length=8, days=days[4:], workers_required=1),
#SingleShift(sites=["a", "b"], name="evening", length=8, days=days[:4], workers_required=1,
# constraint=[
# {"name": "pre", "options": 2},
# {"name": "post", "options": 2},
# ],
# ),
SingleShift(sites=["a"], name="help", length=8, days=days[4:], workers_required=1,
#assign_as_block=True,
#force_as_block=True
#constraint=[
# {"name": "pre", "options": 2},
# {"name": "post", "options": 2},
#],
) ,
SingleShift(sites=["b"], name="weekend b", length=8, days=days[5:], workers_required=1,
#constraint=[
# {"name": "pre", "options": 2},
# {"name": "post", "options": 2},
#],
),
)
#print("Adding shifts")
#for worker in Rota.get_workers_by_site("a"):
# print(worker.name)
## worker.allow_shifts_together("oncall weekday", "evening")
## #worker.allow_shifts_together("oncall weekend", "weekend")
## worker.prefer_multi_shift_together = 1
# worker.assign_as_block_preferences = {"help":10}
#for worker in Rota.get_workers_by_site("b"):
# worker.assign_as_block_preferences = {"weekend b":-10}
Rota.build_and_solve()
Rota.export_rota_to_html("test_cons_rota", folder="tests")
assert Rota.results.solver.status == "ok"
test/test_workers.py
+303 -1
View File
@@ -867,4 +867,306 @@ def test_locums():
Rota.export_rota_to_html("test_locums", timestamp_filename=False)
assert Rota.results.solver.status == "ok"
assert Rota.results.solver.status == "ok"
def test_worker_assign_as_block_preference():
"""Test that a worker's assign_as_block preference is respected over the global setting."""
Rota = generate_basic_rota(workers=4, weeks_to_rota=12)
# Worker 1 prefers to have shift 'a' assigned as a block, worker 2 does not
workers = Rota.get_workers()
worker1 = workers[0]
worker2 = workers[1]
worker3 = workers[2]
worker4 = workers[3]
# Set a strong preference for worker1 to have 'a' as a block
worker1.assign_as_block_preferences = {"a":1}
worker2.assign_as_block_preferences = {"a":-1} # Worker 2 does not want 'a' as a block
# Add a shift 'a' that is NOT globally set as assign_as_block
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=8,
days=days[:4],
workers_required=1,
assign_as_block=False, # global setting is off
),
)
Rota.build_and_solve(options={"ratio": 0.0})
Rota.export_rota_to_html("test_worker_assign_as_block_preference", folder="tests")
# Worker 1 should have their 'a' shifts grouped together (block), worker 2 may not
shift_list_1 = Rota.get_worker_shift_list_string(worker1)
assert shift_list_1.count("aaaa") == 3, "Worker 1 should have all 'a' shifts grouped together (block)"
shift_list_2 = Rota.get_worker_shift_list_string(worker2)
assert shift_list_2.count("a") == 12, "Worker 2 should have at 12 'a' shift, but may not be grouped together"
shift_list_3 = Rota.get_worker_shift_list_string(worker3)
assert shift_list_3.count("a") == 12, "Worker 3 should have at least 12 'a' shifts, but may not be grouped together"
shift_list_4 = Rota.get_worker_shift_list_string(worker4)
assert shift_list_4.count("a") == 12, "Worker 4 should have at least 12 'a' shifts, but may not be grouped together"
# Modify worker1's preference to prevent block 'a' shifts
worker1.assign_as_block_preferences = {"a":-1}
Rota.build_and_solve(options={"ratio": 0.0})
shift_list_1 = Rota.get_worker_shift_list_string(worker1)
assert shift_list_1.count("aaaa") == 0, "Worker 1 should have all 'a' shifts grouped together (block)"
worker1.assign_as_block_preferences = {"a":0}
a_shift = Rota.get_shift_by_name("a")
a_shift.assign_as_block = True # Set the shift 'a' to be assigned as a block globally
Rota.build_and_solve(options={"ratio": 0.0})
for worker in Rota.get_workers():
shift_list = Rota.get_worker_shift_list_string(worker)
# All workers should have 'a' shifts grouped together (block)
assert shift_list.count("aaaa") == 3, f"Worker {worker.name} should have all 'a' shifts grouped together (block)"
def test_worker_assign_as_block_preference_multiple_shifts():
"""Test that a worker's assign_as_block preference is respected over the global setting."""
Rota = generate_basic_rota(workers=4, weeks_to_rota=12)
# Worker 1 prefers to have shift 'a' assigned as a block, worker 2 does not
workers = Rota.get_workers()
worker1 = workers[0]
worker2 = workers[1]
worker3 = workers[2]
worker4 = workers[3]
# Set a strong preference for worker1 to have 'a' as a block
worker1.assign_as_block_preferences = {"a":100}
worker2.assign_as_block_preferences = {"b":-10000} # Worker 2 does not want 'a' as a block
# Add a shift 'a' that is NOT globally set as assign_as_block
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=8,
days=days[:4],
workers_required=1,
assign_as_block=False, # global setting is off
),
SingleShift(
sites=("group1",),
name="b",
length=8,
days=days[4:],
workers_required=1,
assign_as_block=False, # global setting is off
),
)
shift_list_1 = Rota.get_worker_shift_list_string(worker1)
assert shift_list_1.count("aaaa") == 3, "Worker 1 should have all 'a' shifts grouped together (block)"
Rota.build_and_solve(options={"ratio": 0.0})
Rota.export_rota_to_html("test_worker_assign_as_block_preference", folder="tests")
def test_worker_assign_as_block_preference_multiple_sites():
"""Test that a worker's assign_as_block preference is respected over the global setting."""
Rota = generate_basic_rota(workers=6, weeks_to_rota=12)
# Worker 1 prefers to have shift 'a' assigned as a block, worker 2 does not
workers = Rota.get_workers()
worker1 = workers[0]
worker2 = workers[1]
worker3 = workers[2]
worker4 = workers[3]
worker5 = workers[4]
worker6 = workers[5]
# Set a strong preference for worker1 to have 'a' as a block
worker1.assign_as_block_preferences = {"a":1}
worker4.site = "group2"
worker5.site = "group2"
worker6.site = "group2"
worker4.assign_as_block_preferences = {"b":-10000} # Worker 2 does not want 'a' as a block
# Add a shift 'a' that is NOT globally set as assign_as_block
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=8,
days=days[:4],
workers_required=1,
assign_as_block=False, # global setting is off
),
SingleShift(
sites=("group2",),
name="b",
length=8,
days=days[4:],
workers_required=1,
assign_as_block=False, # global setting is off
),
)
Rota.build_and_solve(options={"ratio": 0.0})
Rota.export_rota_to_html("test_worker_assign_as_block_preference", folder="tests")
def test_worker_hard_day_dependency():
Rota = generate_basic_rota(workers=6, weeks_to_rota=16)
workers = Rota.get_workers()
worker1 = workers[0]
worker2 = workers[1]
worker3 = workers[2]
worker1.add_hard_day_dependency("Mon", "Tue")
worker2.add_hard_day_dependency("Mon", "Wed")
worker3.add_hard_day_dependency("Mon", "Thu")
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=8,
days=days,
workers_required=2,
assign_as_block=False, # global setting is off
),
)
Rota.build_and_solve(options={"ratio": 0.0})
Rota.export_rota_to_html("test_worker_hard_day_depend", folder="tests")
for worker in Rota.get_workers():
shift_list = Rota.get_worker_shift_list_string(worker)
if worker.name == "worker1":
# Check that every time 'a' is assigned on Monday, it is also assigned on Tuesday
for week in range(12):
mon_idx = week * 7 + days.index("Mon")
tue_idx = week * 7 + days.index("Tue")
if shift_list[mon_idx] == "a":
assert shift_list[tue_idx] == "a", f"Worker1 assigned 'a' on Mon (week {week}) but not on Tue"
if worker.name == "worker2":
# Check that every time 'a' is assigned on Monday, it is also assigned on Wednesday
for week in range(12):
mon_idx = week * 7 + days.index("Mon")
wed_idx = week * 7 + days.index("Wed")
if shift_list[mon_idx] == "a":
assert shift_list[wed_idx] == "a", f"Worker2 assigned 'a' on Mon (week {week}) but not on Wed"
if worker.name == "worker3":
# Check that every time 'a' is assigned on Monday, it is also assigned on Thursday
for week in range(12):
mon_idx = week * 7 + days.index("Mon")
thu_idx = week * 7 + days.index("Thu")
if shift_list[mon_idx] == "a":
assert shift_list[thu_idx] == "a", f"Worker3 assigned 'a' on Mon (week {week}) but not on Thu"
def test_worker_hard_day_dependency_weekend():
Rota = generate_basic_rota(workers=4, weeks_to_rota=16)
workers = Rota.get_workers()
worker1 = workers[0]
worker2 = workers[1]
worker3 = workers[2]
worker4 = workers[3]
worker1.add_hard_day_dependency("Fri", "Sun")
worker2.add_hard_day_dependency("Fri", "Sun")
worker3.add_hard_day_dependency("Fri", "Sun")
worker4.add_hard_day_dependency("Fri", "Sun")
worker1.add_hard_day_exclusion("Fri", "Sat")
worker2.add_hard_day_exclusion("Fri", "Sat")
worker3.add_hard_day_exclusion("Fri", "Sat")
worker4.add_hard_day_exclusion("Fri", "Sat")
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=8,
days=days[4:],
workers_required=1,
assign_as_block=False, # global setting is off
),
)
Rota.build_and_solve(options={"ratio": 0.0})
Rota.export_rota_to_html("test_worker_hard_day_depend", folder="tests")
for worker in Rota.get_workers():
shift_list = Rota.get_worker_shift_list_string(worker)
# Check that every time 'a' is assigned on Monday, it is also assigned on Tuesday
for week in range(16):
fri_idx = week * 7 + days.index("Fri")
sat_idx = week * 7 + days.index("Sat")
sun_idx = week * 7 + days.index("Sun")
if shift_list[fri_idx] == "a":
assert shift_list[sun_idx] == "a", f"Worker1 assigned 'a' on Fri (week {week}) but not on Sun"
assert shift_list[sat_idx] != "a", f"Worker1 assigned 'a' on Fri (week {week}) but also on Sat, which should not happen due to hard day exclusion"
def test_worker_hard_day_dependency_weekend2():
Rota = generate_basic_rota(workers=4, weeks_to_rota=16)
workers = Rota.get_workers()
worker1 = workers[0]
worker2 = workers[1]
worker3 = workers[2]
worker4 = workers[3]
worker1.add_hard_day_dependency("Fri", "Sun")
worker2.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", "Sat")
worker1.add_hard_day_exclusion("Fri", "Sat")
worker2.add_hard_day_exclusion("Fri", "Sat")
worker3.add_hard_day_exclusion("Fri", "Sat")
Rota.add_shifts(
SingleShift(
sites=("group1",),
name="a",
length=8,
days=days[4:],
workers_required=1,
assign_as_block=False, # global setting is off
),
)
Rota.build_and_solve(options={"ratio": 0.0})
Rota.export_rota_to_html("test_worker_hard_day_depend", folder="tests")
for worker in Rota.get_workers():
shift_list = Rota.get_worker_shift_list_string(worker)
# Check that every time 'a' is assigned on Monday, it is also assigned on Tuesday
for week in range(16):
fri_idx = week * 7 + days.index("Fri")
sat_idx = week * 7 + days.index("Sat")
sun_idx = week * 7 + days.index("Sun")
if worker != worker4 and shift_list[fri_idx] == "a":
assert shift_list[sun_idx] == "a", f"Worker1 assigned 'a' on Fri (week {week}) but not on Sun"
assert shift_list[sat_idx] != "a", f"Worker1 assigned 'a' on Fri (week {week}) but also on Sat, which should not happen due to hard day exclusion"
if worker == worker4 and shift_list[fri_idx] == "a":
assert shift_list[sun_idx] == "a", f"Worker4 assigned 'a' on Fri (week {week}) but not on Sun"
assert shift_list[sat_idx] == "a", f"Worker4 assigned 'a' on Fri (week {week}) but not on Sun"