diff --git a/rota/shifts.py b/rota/shifts.py
index b50ff7a..8f1783e 100644
--- a/rota/shifts.py
+++ b/rota/shifts.py
@@ -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"Start time: {self.run_start_time:%Y-%m-%d %H:%M:%S%z}
"
f"End time: {self.run_end_time:%Y-%m-%d %H:%M:%S%z}
"
- f"Elapsed: {self.run_end_time-self.run_start_time}
"
+ f"Elapsed: {self.run_end_time - self.run_start_time}
"
)
timetable.append(
@@ -3745,7 +3872,12 @@ class RotaBuilder(object):
assigned_shift_names.append(shift)
locum = True
if shift_names:
- shift_display_char = "
".join([""+name+"" for name in shift_names])
+ shift_display_char = "
".join(
+ [
+ "" + name + ""
+ 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 }
diff --git a/rota/workers.py b/rota/workers.py
index 682e6db..93f70c6 100644
--- a/rota/workers.py
+++ b/rota/workers.py
@@ -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))
\ No newline at end of file
+ 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))
\ No newline at end of file
diff --git a/test/test_multiple_shifts_per_day.py b/test/test_multiple_shifts_per_day.py
index b4a36f6..e7fb903 100644
--- a/test/test_multiple_shifts_per_day.py
+++ b/test/test_multiple_shifts_per_day.py
@@ -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"
diff --git a/test/test_workers.py b/test/test_workers.py
index 6dd98b2..c16cc67 100644
--- a/test/test_workers.py
+++ b/test/test_workers.py
@@ -867,4 +867,306 @@ def test_locums():
Rota.export_rota_to_html("test_locums", timestamp_filename=False)
- assert Rota.results.solver.status == "ok"
\ No newline at end of file
+ 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"
+
+
+