ross/proc-rota
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
3f78ecc67152e49b6821021f411c12324dea7096
proc-rota/rota/shifts.py
T
Ross 3f78ecc671 update a few things
2023-05-23 14:48:12 +01:00

3361 lines
129 KiB
Python
Raw Blame History

from calendar import week
import datetime
from distutils.log import debug
import itertools
from typing import Dict, Iterable, List, Sequence, Tuple, Set, Any, Type
import time
from pydantic import BaseModel, Extra, constr
from pathlib import Path
import datetime
from pyomo.environ import *
from pyomo.opt import SolverFactory
from rota.workers import Worker
from rota.console import console
import json
import csv
from collections import defaultdict
from io import StringIO
import sys
from rich.pretty import pprint
from rich.progress import track
from rich.panel import Panel
import logging
logging.basicConfig(
filename="rota.log",
filemode="w",
format="%(name)s - %(levelname)s - %(message)s",
level=logging.DEBUG,
)
ShiftName = str
DayStr = str
WeekInt = int
days = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
full_weekend_count = 2
# Load uk based bank holidays
from govuk_bank_holidays.bank_holidays import BankHolidays
bank_holidays = BankHolidays()
bank_holiday_map = {}
logging.debug("Creating bank holiday map")
for bank_holiday in bank_holidays.get_holidays(division="england-and-wales"):
bank_holiday_map[bank_holiday["date"]] = bank_holiday["title"]
logging.debug(bank_holiday_map)
SHIFT_BOUNDS = {
"bank_holiday": (0, 9),
"shift_count": (0, 400),
"night_shift_count": (0, 100),
"weekend_count": (0, 60),
}
class ShiftConstraint(BaseModel):
name: str
options: bool | int | Dict | tuple = False
class SingleShift(BaseModel):
"""Class to hold all details for a shift
Valid constraints
balance_across_groups (requires block assignment)
postclear(2) / preclear(2)
require_remote_site_presence_week:
options: (site, required_number)
night
limit_grade_number:
options: {grade: max_number, grade2: max_number2}
minimum_grade_number:
options: (grade, min_number)
"""
sites: List[str]
name: str
length: float
days: str | List[str]
balance_offset: float | None = None
balance_weighting: float = 1
workers_required: float = 1
rota_on_nwds: bool = False
assign_as_block: bool = False
force_as_block: bool = False
force_as_block_unless_nwd: bool = False
hard_constrain_shift: bool = True
bank_holidays_only: bool = False
constraint: List[ShiftConstraint] = []
class Config:
extra = Extra.allow
orm_mode = True
def __init__(self, **data: Any):
super().__init__(**data)
# self.site = sites
# self.name = name
# self.length = length
# self.days = days
# self.balance_by_site = balance_by_site
# balance_offset defines the max difference in allocated shifts
# versus target shifts (hard constraint)
# if 0 exactly equal shifts must be assigend (unlikely to be possible)
if self.balance_offset is None:
self.balance_offset = len(self.days)
if self.force_as_block_unless_nwd:
self.assign_as_block = True
# self.assign_as_block = assign_as_block
self.constraint_options = {}
self.constraints = []
for c in self.constraint:
self.constraints.append(c.name)
if c.options:
self.constraint_options[c.name] = c.options
# constraint = c.constraint
# match c:
# case (constraint, options):
# self.constraint_options[constraint] = options
# case constraint:
# self.constraints.append(constraint)
if isinstance(self.days, str):
self.days = [self.days]
def __str__(self):
return f"name: {self.name}, site: {self.sites}, length: {self.length}, balance weighting: {self.balance_weighting}, balance offset: {self.balance_offset}, hard_constrain: {self.hard_constrain_shift}, workers required : {self.workers_required}, rota on nwds {self.rota_on_nwds}, assign as block: {self.assign_as_block}, force as block: {self.force_as_block}, constraints: {self.constraints}, constraint_options: {self.constraint_options}" # , total shifts: {self.total_shifts}"
def get_shift_summary(self):
"""
Returns a text summary of the shift
"""
return f"{self.name}: {self.workers_required} workers for sites ({', '.join(self.site)}) on days ({', '.join(self.days)})"
def get_shift_number(self):
return len(self.days)
class RotaBuilder(object):
"""Class to hold and manipulate shifts"""
def __init__(
self,
start_date: datetime.date = datetime.date.today(),
weeks_to_rota: int = 26,
balance_offset_modifier: int = 1,
ltft_balance_offset: int = 1,
use_previous_shifts: bool = False,
use_shift_balance_extra: bool = False,
use_bank_holiday_extra: bool = False,
SHIFT_BOUNDS=SHIFT_BOUNDS,
):
console.print(Panel(f"""[white]
{locals()}
""", style="green", title="Generating Rota"), justify="left")
self.SHIFT_BOUNDS = SHIFT_BOUNDS
self.shifts: List[SingleShift] = [] # type List[SingleShift]
self.use_previous_shifts = use_previous_shifts
self.use_shift_balance_extra = use_shift_balance_extra
self.use_bank_holiday_extra = use_bank_holiday_extra
self.workers: list[Worker] = []
# self.night_blocks = ["weekday", "weekend", "none"]
self.sites = None
self.balance_offset_modifier = balance_offset_modifier
self.ltft_balance_offset = ltft_balance_offset
self.constraint_options = {
"balance_nights": True,
"constrain_time_off_after_nights": False,
"balance_nights_across_sites": True,
"balance_bank_holidays": True,
"balance_blocks": True,
"balance_shifts": True, # Does not use a quadratic function
"balance_shifts_quadratic": False, # Will prevent spreading of spreading across different shifts
"balance_shifts_over_workers": True,
"minimise_shift_diffs": False, # less sophisticated version of balance_shifts_over_workers
"balance_weekends": True,
"max_weekends": 100,
# Don't assign multiple shifts every (n) weeks
"max_weekend_frequency": 1, # Requires balance weekends
# Don't assign multiple shifts every (n) weeks
"max_night_frequency": 1,
"max_night_frequency_week_exclusions": [],
"max_shifts_per_week": 7,
"max_shifts_per_month": 40,
# The following will cause an unsolvable problem if a shift
# is forced as a block and spans the time peroid
"prevent_monday_after_full_weekends": [],
"prevent_monday_and_tuesday_after_full_weekends": [],
"prevent_fridays_before_full_weekends": [],
"prevent_thursdays_before_full_weekends": [],
"avoid_st2_first_month": False,
"hard_constrain_pair_separation": False,
"avoid_shifts_by_grades": [],
}
self.results = None
self.ignore_valid_shifts = False
self.set_rota_dates(start_date, weeks_to_rota)
def set_rota_dates(self, start_date: datetime.datetime, weeks_to_rota: int):
self.weeks_to_rota = weeks_to_rota
self.days = days
self.weeks = [i for i in range(1, weeks_to_rota + 1)]
self.weeks_days_product = list(itertools.product(self.weeks, days))
if start_date.weekday() != 0:
raise ValueError(
f"Start date {start_date.isoformat()} must be a Mon (not a {days[start_date.weekday()]})"
)
self.start_date = start_date
self.rota_days_length = len(self.weeks) * 7
self.rota_end_date = self.start_date + datetime.timedelta(self.rota_days_length)
console.print(
f"Weeks to rota: {weeks_to_rota} ({self.start_date} - {self.rota_end_date})"
)
# Generate a map for week, day combinations to a given date
self.week_day_date_map = {}
n = 0
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
self.week_day_date_map[(week, day)] = d
n = n + 1
self.unavailable_to_work = set()
self.unavailable_to_work_reason = {}
self.pref_not_to_work = {}
self.pref_not_to_work_reason = {}
self.work_requests: set[Tuple[str, WeekInt, DayStr, ShiftName]] = set()
self.work_requests_map = {}
def solve_model(
self,
solver: str = "cbc",
use_neos: bool = False,
options: dict = {},
debug_if_fail: bool = False,
):
console.print("Setting up solver")
#solver = "scip"
if solver == "scip":
self.opt = SolverFactory(solver, executable="scip")
else:
self.opt = SolverFactory(solver)
try:
console.print("Solving")
if use_neos:
solver_manager = SolverManagerFactory("neos") # Solve using neos server
# results = solver_manager.solve(Rota.model, opt=opt, logfile="test.log")
results = solver_manager.solve(
self.model, keepfiles=True, tee=True, opt=self.opt, logfile="test.log"
)
else:
results = self.opt.solve(
self.model,
tee=True,
options=options,
# options={
# "threads": 10,
# },
logfile="test.log",
keepfiles=True,
)
except KeyboardInterrupt:
return
pass
self.results = results
console.print(f"Complete - outcome: {results.solver.status}")
if results.solver.status != "ok" and debug_if_fail:
console.print(f"Attempting each shift individually")
self.solve_shifts_individually(options)
#if not results.solver.status:
# sys.exit(0)
def solve_shifts_by_block(self, options, block_length: int, folder="block_shifts"):
shifts = self.shifts
outcomes = {}
no_blocks = self.weeks_to_rota - block_length
original_start_date = self.start_date
for block in range(0, no_blocks):
start_time = time.time()
start_date = original_start_date + datetime.timedelta(weeks=block)
self.set_rota_dates(start_date = start_date,weeks_to_rota=block_length)
console.print(f"Testing block: start_date - {self.start_date} , length {self.weeks_to_rota} weeks")
self.clear_shifts()
self.add_shifts(*shifts)
self.build_and_solve(options)
self.export_rota_to_html(filename=f"{start_date}_{self.weeks_to_rota}", folder=folder)
end_time = time.time()
time_taken = end_time - start_time
outcomes[f"{start_date}_{self.weeks}"] = (
self.results.solver.status,
self.results.solver.termination_condition,
time_taken
)
console.print(outcomes)
sys.exit(0)
def solve_shifts_individually(self, options, folder="individual_shifts"):
shifts = self.shifts
outcomes = {}
for shift in shifts:
console.print(f"Testing shifts: {shift.name}")
self.ignore_valid_shifts = True
self.clear_shifts()
self.add_shift(shift)
self.build_and_solve(options)
self.export_rota_to_html(filename=shift.name, folder=folder)
outcomes[shift.name] = (
self.results.solver.status,
self.results.solver.termination_condition,
)
console.print(outcomes)
def build_and_solve(
self,
options={"ratio": 0.1, "seconds": 1000, "threads": 10},
solve=True,
debug_if_fail: bool = False,
):
self.build_shifts()
self.build_workers()
self.build_model()
if solve:
self.solve_model(options=options, debug_if_fail=debug_if_fail)
def build_model(self):
# Initialize model
self.model = ConcreteModel()
# binary variables representing if a worker is scheduled somewhere
# NOTE: this will assign to all possible shift combinations (which we probably don't want)
self.model.works = Var(
(
(worker.id, week, day, shiftname)
for worker in self.workers
for week, day, shiftname in self.get_all_shiftname_combinations()
),
within=Binary,
initialize=0,
)
self.model.shift_week_worker_assigned = Var(
(
(shift, week, worker.id)
for worker in self.workers
for week in self.weeks
for shift in self.get_shift_names()
),
within=Binary,
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()
),
within=Binary,
initialize=0,
)
self.model.blocks_assigned = Var(
(
(week, shift)
for week in self.weeks
for shift in self.shifts_to_assign_or_force_as_blocks()
),
within=NonNegativeIntegers,
initialize=0,
)
if self.constraint_options["balance_bank_holidays"]:
# Bank holidays
self.model.bank_holiday_count = Var(
((worker.id) for worker in self.workers),
within=NonNegativeIntegers,
initialize=0,
bounds=self.SHIFT_BOUNDS["bank_holiday"],
)
self.model.bank_holiday_count_w = Var(
((worker.id) for worker in self.workers),
within=NonNegativeIntegers,
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
if self.constraint_options["balance_nights_across_sites"]:
self.model.night_per_site = Var(
(
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
),
# within=Binary,
within=NonNegativeIntegers,
initialize=0,
)
self.model.night_per_site_t1 = Var(
(
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
),
domain=NonNegativeIntegers,
initialize=0,
)
self.model.night_per_site_t2 = Var(
(
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
),
domain=NonNegativeIntegers,
initialize=0,
)
# Replacement for balance_nights_across_sites
self.model.shift_per_site = Var(
(
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("balance_across_groups")
),
# within=Binary,
within=NonNegativeIntegers,
initialize=0,
)
self.model.shift_per_site_t1 = Var(
(
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("balance_across_groups")
),
domain=NonNegativeIntegers,
initialize=0,
)
self.model.shift_per_site_t2 = Var(
(
(week, shift.name, site)
for site in self.sites
for week in self.weeks
for shift in self.get_shifts_with_constraint("balance_across_groups")
),
domain=NonNegativeIntegers,
initialize=0,
)
self.model.shift_count = Var(
(
(worker.id, shift)
for worker in self.workers
for shift in self.get_shift_names()
),
domain=NonNegativeReals,
initialize=0,
bounds=self.SHIFT_BOUNDS["shift_count"],
)
self.model.shift_count_diff = Var(
(
(worker.id, shift)
for worker in self.workers
for shift in self.get_shift_names()
),
within=Reals,
initialize=0,
)
self.model.shift_count_diff_summed = Var(
(worker.id for worker in self.workers),
within=Reals,
initialize=0,
)
if self.constraint_options["balance_shifts_over_workers"]:
self.model.worker_shift_count_t1 = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
self.model.worker_shift_count_t2 = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
if (
self.constraint_options["balance_shifts"]
or self.constraint_options["balance_shifts_quadratic"]
):
self.model.shift_count_t1 = Var(
(
(worker.id, shift.name)
for worker in self.workers
for shift in self.get_shifts()
),
domain=NonNegativeReals,
initialize=0,
)
self.model.shift_count_t2 = Var(
(
(worker.id, shift.name)
for worker in self.workers
for shift in self.get_shifts()
),
domain=NonNegativeReals,
initialize=0,
)
if self.constraint_options["balance_shifts_quadratic"]:
self.model.shift_count_w = Var(
(
(worker.id, shift.name)
for worker in self.workers
for shift in self.get_shifts()
),
domain=NonNegativeReals,
initialize=0,
)
if self.constraint_options["balance_nights"]:
# We also try to even out the night shifts seperately
self.model.night_shift_count = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
bounds=self.SHIFT_BOUNDS["night_shift_count"],
)
self.model.night_shift_count_t1 = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
self.model.night_shift_count_t2 = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
self.model.night_shift_count_w = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
# self.model.worker_weekend_assigned = Var(
# ((worker.id, week) for worker in self.workers
# for week in self.weeks),
# domain=NonNegativeReals,
# initialize=0,
# )
# self.model.works_weekend_count = Var(
# ((worker.id, week) for worker in self.workers
# for week in self.weeks),
# domain=NonNegativeIntegers,
# initialize=0,
# )
# self.model.works_saturday = Var(
# ((worker.id, week) for worker in self.workers
# for week in self.weeks),
# domain=Binary,
# initialize=0,
# )
# self.model.works_sunday = Var(
# ((worker.id, week) for worker in self.workers
# for week in self.weeks),
# domain=Binary,
# initialize=0,
# )
# self.model.works_whole_weekend = Var(
# ((worker.id, week) for worker in self.workers
# for week in self.weeks),
# domain=Binary,
# initialize=0,
# )
self.model.works_weekend = Var(
((worker.id, week) for worker in self.workers for week in self.weeks),
domain=Binary,
initialize=0,
)
self.model.worker_weekend_count = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeIntegers,
initialize=0,
bounds=self.SHIFT_BOUNDS["weekend_count"],
)
if self.constraint_options["balance_weekends"]:
self.model.weekend_shift_count_t1 = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
self.model.weekend_shift_count_t2 = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
self.model.weekend_shift_count_w = Var(
((worker.id) for worker in self.workers),
domain=NonNegativeReals,
initialize=0,
)
# self.model.weekend_count_t1 = Var(
# ((worker.id) for worker in self.workers),
# domain=NonNegativeReals,
# initialize=0,
# )
# self.model.weekend_count_t2 = Var(
# ((worker.id) for worker in self.workers),
# domain=NonNegativeReals,
# initialize=0,
# )
# self.model.unavailable = Var(((worker, week, day) for worker in self.workers for week in weeks for day in days),
# within=Binary, initialize=0)
def availability_init(model, wid, week, day):
if (wid, week, day) in self.unavailable_to_work:
# print((wid, week, day))
return 0
return 1
self.model.available = Param(
(
(worker.id, week, day)
for worker in self.workers
for week, day in self.get_week_day_combinations()
),
initialize=availability_init,
)
work_request_sets = set()
for work_request in self.work_requests:
if work_request[3] == "*":
pass
else:
work_request_sets.add(work_request[3])
# Validate shifts are valid
# work_request_sets = set([i[3] for i in self.work_requests])
invalid_shifts = work_request_sets - set(self.get_shift_names())
if invalid_shifts and not self.ignore_valid_shifts:
raise InvalidShift(
f"Invalid worker shift request [shift(s): ${invalid_shifts}]"
)
def work_request_init(model, wid, week, day, shift):
if (wid, week, day, shift) in self.work_requests:
self.work_requests_map[(wid, week, day)] = shift
return 1
return 0
self.model.work_requests = Param(
(
(worker.id, week, day, shift)
for worker in self.workers
for week, day, shift in self.get_all_shiftname_combinations()
),
initialize=work_request_init,
)
def pref_init(model, wid, week, day):
if (wid, week, day) in self.pref_not_to_work:
return self.pref_not_to_work[(wid, week, day)]
return 0
# It would also be possible to assign sequential days as blocks
# which would allow more weight to be given if the whole block
# is not allocated
self.model.pref_not_to_work = Param(
(
(worker.id, week, day)
for worker in self.workers
for week, day in self.get_week_day_combinations()
),
initialize=pref_init,
)
# binary variables representing if a worker is necessary
# self.model.needed = Var([worker.id for worker in self.workers], within=Binary, initialize=0)
# binary variables representing if a worker worked on sunday but not on saturday (avoid if possible)
# self.model.no_pref = Var([worker.id for worker in self.workers], within=Binary, initialize=0)
self.build_model_constraints()
def build_model_constraints(self):
self.model.constraints = ConstraintList() # Create a set of constraints
# Ensure each shift has the requisit number of workers assigned
for (
week,
day,
shift,
workers_required,
site_required,
) in self.get_required_workers_and_site_combinations():
# print(week, day, shift, workers_required, site_required)
self.model.constraints.add(
workers_required
== sum(
self.model.works[worker.id, week, day, shift]
for worker in self.workers
if worker.site in site_required
)
)
# 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(
0
== sum(
self.model.works[worker.id, week, day, shift]
for worker in self.workers
if worker.site not in site_required
)
)
# # Ensure shifts are only assigned by workers from the allowed sites
# self.model.constraints.add(0 == sum(
# self.model.works[worker.id, week, day, shift]
# for worker in self.workers if worker.site not in site_required))
# Constraint: total hours worked hours worked
# for worker in self.workers:
# self.model.constraints.add(
# 1200 >= sum(shift_lengths[shift] * self.model.works[worker, week, day, shift] for week in weeks for day in days for shift in days_shifts[day])
# )
# for week in weeks:
# for worker in self.workers:
# self.model.constraints.add(
# 48 >= sum(shift_lengths[shift] * self.model.works[worker, week, day, shift] for day in days for shift in days_shifts[day])
# )
# Define a maximum number hours that can be worked per week
# NOTE: this works on a calander week not a 7 day period
# TODO: consider fixing
# def maxHoursPerWeekRule(model, wid, week):
# return 72 >= sum(
# shift.length * model.works[wid, week, day, shift.name]
# for day, shift in self.get_day_shiftclass_combinations())
# self.model.max_hours_per_week_constraint = Constraint(
# [worker.id for worker in self.workers],
# self.weeks,
# rule=maxHoursPerWeekRule)
# Define an upper limit for hours worked over the rota period
# def maxHoursRule(model, wid):
# return 8000 >= sum(
# shift.length * model.works[wid, week, day, shift.name]
# for week, day, shift in self.get_all_shiftclass_combinations())
# self.model.max_hours_constraint = Constraint(
# [worker.id for worker in self.workers], rule=maxHoursRule)
# Limit grades
def limitGradesRule(model, week, shift, grade, limit):
workers = [w for w in self.workers if w.grade == grade]
if not workers:
return Constraint.Skip
return (
sum(
model.shift_week_worker_assigned[shift, week, w.id] for w in workers
)
<= limit
)
for shift in self.get_shifts_with_constraint("limit_grade_number"):
if not shift.constraint_options["limit_grade_number"]:
raise ValueError(
"Constraint option must be defined for 'limit_grade_number'"
)
for grade in shift.constraint_options["limit_grade_number"]:
# self.model.limit_grades_constraint = Constraint(
setattr(
self.model,
f"limit_grades_constraint_{shift.name}_{grade}",
Constraint(
[week for week in self.weeks],
# [shift for shift in self.get_shifts_with_constraint("limit_grade_number")],
[shift.name],
[grade],
[shift.constraint_options["limit_grade_number"][grade]],
rule=limitGradesRule,
# name=f"limit_grades_constraint_{shift.name}_{grade}"
),
)
def minimumGradesRule(model, week, shift, grade, limit):
workers = [w for w in self.workers if w.grade >= grade]
if len(workers) < limit:
# return Constraint.Skip
raise ValueError("minimum_grade_number: not enough valid workers")
return (
sum(
model.shift_week_worker_assigned[shift, week, w.id] for w in workers
)
>= limit
)
for shift in self.get_shifts_with_constraint("minimum_grade_number"):
if not shift.constraint_options["minimum_grade_number"]:
raise ValueError(
"Constraint option must be defined for 'minimum_grade_number'"
)
grade, min_required = shift.constraint_options["minimum_grade_number"]
# self.model.limit_grades_constraint = Constraint(
setattr(
self.model,
f"minimum_grade_number_{shift.name}",
Constraint(
[week for week in self.weeks],
# [shift for shift in self.get_shifts_with_constraint("limit_grade_number")],
[shift.name],
[grade],
[min_required],
rule=minimumGradesRule,
# name=f"limit_grades_constraint_{shift.name}_{grade}"
),
)
def presenceAtRemoteSiteWeek(
model, week, shift, required_site, required_number
):
required_site_workers = [
w for w in self.workers if required_site == w.remote_site
]
return (
sum(
model.shift_week_worker_assigned[shift, week, w.id]
for w in required_site_workers
)
>= required_number
)
for shift in self.get_shifts_with_constraint(
"require_remote_site_presence_week"
):
site, required_number = shift.constraint_options[
"require_remote_site_presence_week"
]
# self.model.require_presence_at_site_overnight_rule = Constraint(
setattr(
self.model,
f"require_remote_site_presence_week_{shift.name}",
Constraint(
[week for week in self.weeks],
[shift.name],
[site],
[required_number],
rule=presenceAtRemoteSiteWeek,
),
)
def presenceAtRemoteSite(
model, day, week, shift, required_site, required_number
):
required_site_workers = [
w for w in self.workers if required_site == w.remote_site
]
for w in required_site_workers:
if (w.id, week, day, shift) not in model.works:
return Constraint.Skip
return (
sum(model.works[w.id, week, day, shift] for w in required_site_workers)
>= required_number
)
for shift in self.get_shifts_with_constraint("require_remote_site_presence"):
site, required_number = shift.constraint_options[
"require_remote_site_presence"
]
# self.model.require_presence_at_site_overnight_rule = Constraint(
setattr(
self.model,
f"require_remote_site_presence_{shift.name}",
Constraint(
[day for day in self.days],
[week for week in self.weeks],
[shift.name],
[site],
[required_number],
rule=presenceAtRemoteSite,
),
)
# # Count the number of workers from each site on each night shift
# # As 1 or 0 is optimum we can simply subtract 1 from the number
# if self.constraint_options["balance_nights_across_sites"]:
# for site in self.sites:
# for shift in self.get_shifts_with_constraint("night"):
# block = shift.name
# for week in self.weeks:
# self.model.constraints.add(
# self.model.night_per_site[week, block, site]
# == sum(
# self.model.shift_week_worker_assigned[
# block, week, worker.id
# ]
# for worker in self.workers
# if worker.site == site
# )
# # 1 >= sum(self.model.shift_week_worker_assigned[shift.name, week, worker.id] for worker in self.workers if worker.site == site)
# )
# self.model.constraints.add(
# self.model.night_per_site_t1[week, block, site]
# - self.model.night_per_site_t2[week, block, site]
# == self.model.night_per_site[week, block, site] - 1
# )
for site in track(self.sites, description="Generating site balance constraints..."):
for shift in self.get_shifts_with_constraint("balance_across_groups"):
if site in shift.sites:
block = shift.name
for week in self.weeks:
self.model.constraints.add(
self.model.shift_per_site[week, block, site]
== sum(
self.model.shift_week_worker_assigned[
block, week, worker.id
]
for worker in self.workers
if worker.site == site
)
# 1 >= sum(self.model.shift_week_worker_assigned[shift.name, week, worker.id] for worker in self.workers if worker.site == site)
)
self.model.constraints.add(
self.model.shift_per_site_t1[week, block, site]
- self.model.shift_per_site_t2[week, block, site]
== self.model.shift_per_site[week, block, site] - 1
)
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():
shift = self.get_shift_by_name(shift_name)
for worker in self.get_workers_for_shift(shift):
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
)
)
self.model.constraints.add(
self.model.blocks_assigned[week, shift_name]
== sum(
self.model.blocks_worker_shift_assigned[
worker.id, week, shift_name
]
for worker in self.workers
)
)
if shift.force_as_block_unless_nwd:
workers = self.get_workers_for_shift(shift)
# Get workers who have a nwd on the shift
nwd_workers = []
full_workers = []
for w in workers:
if w.non_working_day_list:
l = []
for (
nwd,
start_nwd_date,
end_nwd_date,
) in w.non_working_day_list:
if nwd in shift.days:
if start_nwd_date > self.get_week_start_date(week):
continue
if end_nwd_date < self.get_week_start_date(week):
continue
l.append(w)
if l:
nwd_workers.append(w)
else:
full_workers.append(w)
else:
full_workers.append(w)
if not nwd_workers: # Just do the usual
self.model.constraints.add(
sum(
self.model.blocks_worker_shift_assigned[
worker.id, week, shift.name
]
for worker in self.workers
)
<= shift.workers_required
)
else:
self.model.constraints.add(
sum(
self.model.blocks_worker_shift_assigned[
worker.id, week, shift.name
]
for worker in full_workers
)
<= shift.workers_required + 1
)
elif shift.force_as_block:
self.model.constraints.add(
sum(
self.model.blocks_worker_shift_assigned[
worker.id, week, shift.name
]
for worker in self.workers
)
<= shift.workers_required
)
# Most of our constraints apply per worker
for worker in self.workers:
console.rule(f"Generate worker constraints: [bold blue]{worker.name}[/bold blue]")
logging.debug(f"Generate worker constraints: {worker.name}")
if self.constraint_options["minimise_shift_diffs"]:
self.model.constraints.add(
self.model.shift_count_diff_summed[worker.id]
== sum(
self.model.shift_count_diff[worker.id, shift.name]
for shift in self.get_shifts()
)
)
for week_blocks in self.get_week_block_iterator(4):
# Prevent more than n number shifts per 4 weeks
try:
self.model.constraints.add(
self.constraint_options["max_shifts_per_month"]
>= sum(
self.model.works[worker.id, week, day, shiftname]
for week, day, shiftname in self.get_all_shiftname_combinations()
if week in week_blocks
)
)
except ValueError:
# Occurs if there are no shifts within a defined block
# TODO: test if this breaks (and we should check rathar than except)
logging.debug(f"Failed to constrain max_shifts_per_month for worker: {worker.name}")
pass
for constraint_dict in self.constraint_options["avoid_shifts_by_grades"]:
if invalid_shifts := set(constraint_dict["shifts"]).difference(
set(self.get_shift_names())
):
if self.ignore_valid_shifts:
continue # Skip if we don't want to raise an error
else:
raise InvalidShift(
f"avoid shifts by grade constraint contains a non existent shift: {invalid_shifts}"
)
if invalid_grades := set(constraint_dict["grades"]).difference(
set(self.get_worker_grades())
):
raise ValueError(
f"avoid shifts by grade constraint contains a non existent worker grade: {invalid_grades}"
)
if worker.grade in constraint_dict["grades"]:
self.model.constraints.add(
0
== sum(
self.model.works[worker.id, week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
if week in constraint_dict["weeks"]
and shift in constraint_dict["shifts"]
)
)
# sys.exit(0)
if self.constraint_options["avoid_st2_first_month"] and worker.grade == 2:
# Avoid ST1s on the first month
try:
self.model.constraints.add(
0
== sum(
self.model.works[worker.id, week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
if week in (1, 2, 3, 4)
and shift in ("night_weekday", "night_weekend")
)
)
except ValueError as e:
print("Failure setting constraint", "avoid_st2_first_month")
print(e)
# Count number of weekends an worker works
# if self.constraint_options["balance_weekends"]:
self.model.constraints.add(
self.model.worker_weekend_count[worker.id]
== sum(self.model.works_weekend[worker.id, week] for week in self.weeks)
)
if self.constraint_options["max_weekends"] > -1:
self.model.constraints.add(
self.constraint_options["max_weekends"]
>= self.model.worker_weekend_count[worker.id]
)
# Balance shifts within required limits (set by balance_offset)
# If balance_offset is too restrictive (for the rota length)
# no solution will be possible
for shift in track(self.get_shifts(), description="Generate shift balance constraints"):
if (
worker.site in shift.sites
): # Each site specfies which sites self.workers can fullfill it
# calaculate the total number of shifts that need assigning over the rota
# total_shifts = (len(self.weeks) * len(shift.days) *
total_shifts = (
self.shift_counts[shift.name] * shift.workers_required
)
full_time_equivalent_joined = sum(
self.full_time_equivalent_sites[i] for i in shift.sites
)
target_shifts = (
total_shifts / full_time_equivalent_joined * worker.fte_adj
)
if self.use_previous_shifts:
if shift.name in worker.previous_shifts:
worked, allocated = worker.previous_shifts[shift.name]
target_shifts = (
target_shifts + float(allocated) - float(worked)
)
if self.use_shift_balance_extra:
if shift.name in worker.shift_balance_extra:
extra = worker.shift_balance_extra[shift.name]
target_shifts = target_shifts + extra
# print(worker.name, shift.name, target_shifts)
worker.shift_target_number[shift.name] = target_shifts
if shift.hard_constrain_shift:
adjusted_balance_offset = shift.balance_offset
if worker.fte_adj < 100:
adjusted_balance_offset = (
adjusted_balance_offset * self.ltft_balance_offset
)
max_shifts = (
target_shifts
+ adjusted_balance_offset * self.balance_offset_modifier
)
min_shifts = (
target_shifts
- adjusted_balance_offset * self.balance_offset_modifier
)
self.model.constraints.add(
inequality(
min_shifts,
sum(
self.model.works[worker.id, week, day, shift.name]
for week, day in self.get_week_day_combinations_for_shift(
shift
)
# for week, day in self.get_week_day_combinations()
# if shift.name in self.get_shift_names_by_week_day(week, day)
),
max_shifts,
)
)
else:
# Make sure shifts aren't assigned to those from other sites
# Not needed if there are no shifts
if self.get_week_day_combinations_for_shift(shift):
self.model.constraints.add(
0
== sum(
self.model.works[worker.id, week, day, shift.name]
for week, day in self.get_week_day_combinations_for_shift(
shift
)
# for week, day in self.get_week_day_combinations()
)
)
# model.shift_count stores the number of shifts that a worker is assigned to worker by shift
# this is used (by the objective) to balance the total number of shifts worked
self.model.constraints.add(
self.model.shift_count[worker.id, shift.name]
== sum(
self.model.works[worker.id, week, day, shift.name]
for week, day in self.get_week_day_combinations_for_shift(shift)
)
)
self.model.constraints.add(
self.model.shift_count_diff[worker.id, shift.name]
== self.model.shift_count[worker.id, shift.name]
- worker.shift_target_number[shift.name]
)
if self.constraint_options["balance_shifts"]:
self.model.constraints.add(
self.model.shift_count_t1[worker.id, shift.name]
- self.model.shift_count_t2[worker.id, shift.name]
== self.model.shift_count_diff[worker.id, shift.name]
)
if self.constraint_options["balance_shifts_quadratic"]:
# This may need to be updated
xU = 25
xL = 1
self.model.constraints.add(
inequality(
xL,
self.model.shift_count_t1[worker.id, shift.name]
+ self.model.shift_count_t2[worker.id, shift.name]
+ 1,
xU,
)
)
self.model.constraints.add(
self.model.shift_count_w[worker.id, shift.name]
>= xL
* (
self.model.shift_count_t1[worker.id, shift.name]
+ self.model.shift_count_t2[worker.id, shift.name]
+ 1
)
* 2
- xL * xL
)
self.model.constraints.add(
self.model.shift_count_w[worker.id, shift.name]
>= xU
* (
self.model.shift_count_t1[worker.id, shift.name]
+ self.model.shift_count_t2[worker.id, shift.name]
+ 1
)
* 2
- xU * xU
)
# Define worker_shift_count_t1 and worker_shift_count_t2 constraints for the object
# Thus bypassing the need for a quadratic solver
# t1-t2 is the target
# As the objective is to minimise t1+t2 and t1 and t2 are positive reals
# t1+t2 approximates the absolute target (which otherwise requires a quadratic solver)
# TODO: quadratic implementation so perfect solutions will be chosen
if self.constraint_options["balance_shifts_over_workers"]:
self.model.constraints.add(
self.model.worker_shift_count_t1[worker.id]
- self.model.worker_shift_count_t2[worker.id]
== sum(
(self.model.shift_count_diff[worker.id, shift.name])
* shift.balance_weighting
for shift in self.get_shifts()
)
)
if self.constraint_options["balance_bank_holidays"]:
extra_bank_holiday = 0
if self.use_bank_holiday_extra:
extra_bank_holiday = worker.bank_holiday_extra
self.model.constraints.add(
self.model.bank_holiday_count[worker.id]
== sum(
self.model.works[worker.id, week, day, shift]
for week, day, shift in self.get_all_shiftname_combinations()
if self.week_day_date_map[(week, day)] in bank_holiday_map
)
+ extra_bank_holiday
)
xU = len(self.get_bank_holiday_week_days()) + 1
xL = 1
self.model.constraints.add(
inequality(
xL,
self.model.bank_holiday_count[worker.id] + 1,
xU,
)
)
self.model.constraints.add(
self.model.bank_holiday_count_w[worker.id]
>= xL * (self.model.bank_holiday_count[worker.id] + 1) * 2 - xL * xL
)
self.model.constraints.add(
self.model.bank_holiday_count_w[worker.id]
>= xU * (self.model.bank_holiday_count[worker.id] + 1) * 2 - xU * xU
)
if self.constraint_options["balance_nights"]:
self.model.constraints.add(
self.model.night_shift_count[worker.id]
== sum(
self.model.works[worker.id, week, day, shift.name]
for week, day, shift in self.get_week_day_shift_combinations_for_constraint(
"night"
)
# for week, day in self.get_week_day_combinations_for_shift(shift)
# for shift in self.get_shifts_with_constraint("night")
)
)
night_shift_target_number = sum(
worker.shift_target_number[shift.name]
for shift in self.get_shifts_with_constraint("night")
)
self.model.constraints.add(
self.model.night_shift_count_t1[worker.id]
- self.model.night_shift_count_t2[worker.id]
== self.model.night_shift_count[worker.id]
- night_shift_target_number
)
# This may need to be updated
xU = 6
xL = 1
self.model.constraints.add(
inequality(
xL,
self.model.night_shift_count_t1[worker.id]
+ self.model.night_shift_count_t2[worker.id]
+ 1,
xU,
)
)
self.model.constraints.add(
self.model.night_shift_count_w[worker.id]
>= xL
* (
self.model.night_shift_count_t1[worker.id]
+ self.model.night_shift_count_t2[worker.id]
+ 1
)
* 2
- xL * xL
)
self.model.constraints.add(
self.model.night_shift_count_w[worker.id]
>= xU
* (
self.model.night_shift_count_t1[worker.id]
+ self.model.night_shift_count_t2[worker.id]
+ 1
)
* 2
- xU * xU
)
# self.model.constraints.add(
# self.model.night_shift_count_w[worker.id] >= 0)
# We use a similar method to balance the number of weekends worked
# This works as long as weekend shifts are assigned as blocks!
weekend_shift_target_number = sum(
worker.shift_target_number[shift.name] / len(shift.days)
for shift in self.get_shifts()
if "Sat" in shift.days or "Sun" in shift.days
)
worker.weekend_shift_target_number = weekend_shift_target_number
if self.constraint_options["balance_weekends"]:
if weekend_shift_target_number > 0:
self.model.constraints.add(
self.model.weekend_shift_count_t1[worker.id]
- self.model.weekend_shift_count_t2[worker.id]
== self.model.worker_weekend_count[worker.id]
- weekend_shift_target_number
)
xU = self.constraint_options["max_weekends"]
xL = 1
self.model.constraints.add(
inequality(
xL,
self.model.weekend_shift_count_t1[worker.id]
+ self.model.weekend_shift_count_t2[worker.id]
+ 1,
xU,
)
)
self.model.constraints.add(
self.model.weekend_shift_count_w[worker.id]
>= xL
* (
self.model.weekend_shift_count_t1[worker.id]
+ self.model.weekend_shift_count_t2[worker.id]
+ 1
)
* 2
- xL * xL
)
self.model.constraints.add(
self.model.weekend_shift_count_w[worker.id]
>= xU
* (
self.model.weekend_shift_count_t1[worker.id]
+ self.model.weekend_shift_count_t2[worker.id]
+ 1
)
* 2
- xU * xU
)
# Ensure worker is not allocated shifts on non working days
if worker.non_working_day_list:
for week, day, shift in self.get_all_shiftclass_combinations():
for n, start, end in worker.non_working_day_list:
if not shift.rota_on_nwds and day == n:
# print(start, self.week_day_date_map[(week, day)], end)
if start <= self.week_day_date_map[(week, day)] < end:
self.model.constraints.add(
0
== self.model.works[
worker.id, week, day, shift.name
]
)
if self.constraint_options["balance_blocks"]:
if self.constraint_options["max_night_frequency"]:
for week_blocks in self.get_week_block_iterator(
self.constraint_options["max_night_frequency"]
):
# Ignore excluded weeks
if set(week_blocks).intersection(set(self.constraint_options["max_night_frequency_week_exclusions"])):
continue
if self.get_shifts_with_constraint("night"):
# Prevent nights more than once every n weeks
self.model.constraints.add(
1
>= sum(
self.model.shift_week_worker_assigned[
shift.name, week, worker.id
]
for week in week_blocks
for shift in self.get_shifts_with_constraint(
"night"
)
)
)
# if self.constraint_options["balance_weekends"]:
if self.constraint_options["max_weekend_frequency"]:
for week_blocks in self.get_week_block_iterator(
self.constraint_options["max_weekend_frequency"]
):
# Prevent weekend shifts more than once every n weeks
self.model.constraints.add(
1
>= sum(
self.model.works_weekend[worker.id, week]
for week in week_blocks
)
)
for constraint_shift in self.get_shifts_with_constraints(
"max_shifts_per_week_block"
):
constraint_options = constraint_shift.constraint_options[
"max_shifts_per_week_block"
]
# If not supplied, default to the number of days per shift
shift_number = constraint_shift.get_shift_number()
if "shift_number" in constraint_options:
shift_number = constraint_options["shift_number"]
for week_blocks in self.get_week_block_iterator(
constraint_options["weeks"]
):
# Prevent shifts more than once every n weeks
self.model.constraints.add(
shift_number
>= sum(
self.model.works[
worker.id, week, day, constraint_shift.name
]
for week in week_blocks
for day in constraint_shift.days
)
)
for week in self.weeks:
for constraint_shift in self.get_shifts_with_constraints(
"max_shifts_per_week"
):
self.model.constraints.add(
constraint_shift.constraint_options["max_shifts_per_week"]
>= sum(
self.model.works[worker.id, w, day, constraint_shift.name]
# for shiftname in self.get_shift_names_by_week_day(week, day)
# for day in self.days
for w, day in self.get_week_day_combinations_for_shift(
constraint_shift
)
if w == week
)
)
try:
self.model.constraints.add(
self.constraint_options["max_shifts_per_week"]
>= sum(
self.model.works[worker.id, w, day, shiftname]
# for shiftname in self.get_shift_names_by_week_day(week, day)
# for day in self.days
for w, day, shiftname in self.get_all_shiftname_combinations(
week=week
)
)
)
except ValueError:
pass
# TODO: consider excluding shifts that span relevant period
if (
worker.site
in self.constraint_options["prevent_monday_after_full_weekends"]
):
# Ignore last week
if week + 1 in self.weeks:
self.model.constraints.add(
full_weekend_count
>= sum(
self.model.works[worker.id, week, "Sat", shift]
for shift in self.get_shift_names_by_week_day(
week, "Sat"
)
)
+ sum(
self.model.works[worker.id, week, "Sun", shift]
for shift in self.get_shift_names_by_week_day(
week, "Sun"
)
)
+ sum(
self.model.works[worker.id, week + 1, "Mon", shift]
for shift in self.get_shift_names_by_week_day(
week + 1, "Mon"
)
)
)
if (
worker.site
in self.constraint_options[
"prevent_monday_and_tuesday_after_full_weekends"
]
):
# Ignore last week
if week + 1 in self.weeks:
self.model.constraints.add(
full_weekend_count
>= sum(
self.model.works[worker.id, week, "Sat", shift]
for shift in self.get_shift_names_by_week_day(
week, "Sat"
)
)
+ sum(
self.model.works[worker.id, week, "Sun", shift]
for shift in self.get_shift_names_by_week_day(
week, "Sun"
)
)
+ sum(
self.model.works[worker.id, week + 1, "Mon", shift]
for shift in self.get_shift_names_by_week_day(
week + 1, "Mon"
)
)
+ sum(
self.model.works[worker.id, week + 1, "Tue", shift]
for shift in self.get_shift_names_by_week_day(
week + 1, "Tue"
)
)
)
if (
worker.site
in self.constraint_options["prevent_fridays_before_full_weekends"]
):
self.model.constraints.add(
full_weekend_count
>= sum(
self.model.works[worker.id, week, "Sat", shift.name]
for shift in self.get_shift_names_by_week_day(week, "Sat")
)
+ sum(
self.model.works[worker.id, week, "Sun", shift.name]
for shift in self.get_shift_names_by_week_day(week, "Sat")
)
+ sum(
self.model.works[worker.id, week, "Fri", shift.name]
for shift in self.get_shift_names_by_week_day(week, "Sat")
)
)
if (
worker.site
in self.constraint_options["prevent_thursdays_before_full_weekends"]
):
self.model.constraints.add(
full_weekend_count
>= sum(
self.model.works[worker.id, week, "Sat", shift.name]
for shift in self.get_shift_names_by_week_day(week, "Sat")
)
+ sum(
self.model.works[worker.id, week, "Sun", shift.name]
for shift in self.get_shift_names_by_week_day(week, "Sun")
)
+ sum(
self.model.works[worker.id, week, "Thu", shift.name]
for shift in self.get_shift_names_by_week_day(week, "Thu")
)
)
# # model.weekend_count stores the number of weekend shifts that a worker is assigned to wor
# # this is used (by the objective) to balance the total number of weekend shifts worked
# self.model.constraints.add(
# # We could use a helper to get required shifts to check
# self.model.works_whole_weekend[worker.id, week] >= sum(
# self.model.works[worker.id, week, day, shift.name]
# for day in self.days[5:]
# for shift in self.get_shifts()) - 1)
# self.model.constraints.add(
# self.model.works_whole_weekend[worker.id, week] <= sum(
# self.model.works[worker.id, week, day, shift.name]
# for day in self.days[5:]
# for shift in self.get_shifts()) / 2)
# if self.constraint_options["balance_weekends"]:
self.model.constraints.add(
self.model.works_weekend[worker.id, week]
>= sum(
self.model.works[worker.id, week, day, shiftname]
for w, day, shiftname in self.get_all_shiftname_combinations(
week=week
)
if day in self.days[5:]
)
/ 2
)
self.model.constraints.add(
self.model.works_weekend[worker.id, week]
<= sum(
self.model.works[worker.id, week, day, shiftname]
for w, day, shiftname in self.get_all_shiftname_combinations(
week=week
)
if day in self.days[5:]
)
)
# self.model.constraints.add(
# self.model.works_saturday[worker.id, week] == sum(
# self.model.works[worker.id, week, "Sat", shift.name]
# for shift in self.get_shifts()))
# self.model.constraints.add(
# self.model.works_sunday[worker.id, week] == sum(
# self.model.works[worker.id, week, "Sun", shift.name]
# for shift in self.get_shifts()))
# self.model.constraints.add(1 == sum(
# self.model.shift_week_worker_assigned[s.name, week, worker.id]
# for s in self.get_shifts_with_constraint("night")))
# for shift in self.get_shifts_with_constraint("single_block_per_week"):
# if shift.force_as_block:
# # Force nights to be assigned in blocks
# # self.model.constraints.add(8* self.model.shift_week_worker_assigned[shift.name, week, worker.id] >= sum(self.model.works[worker.id, week, day, shift.name] for day in self.days)
# # )
# # Force night 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, w, d, s]
# for w, d, s in self.get_all_shiftname_combinations(week=week)
# #for day in self.days
# )
# )
# else:
# raise ValueError(f"Requires {shift.name} to have force_as_block")
# for shift in self.get_shifts_with_constraint("night"):
for shift in self.get_shifts():
if shift.force_as_block:
# Force nights to be assigned in blocks
# self.model.constraints.add(8* self.model.shift_week_worker_assigned[shift.name, week, worker.id] >= sum(self.model.works[worker.id, week, day, shift.name] for day in self.days)
# )
# Force night 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
)
)
#
# def get_pre_may(week_days, max_pre)
for n in track(range(len(weeks_days)), description="Generate week/day constraints"):
week, day = weeks_days[n]
pre_map = []
for pre_n in range(1, self.max_pre + 1):
try:
pweek, pday = weeks_days[n - pre_n]
p = 1
except:
pweek, pday = weeks_days[n]
p = 0
pre_map.append((p, pweek, pday))
post_map = []
for post_n in range(1, self.max_post + 1):
try:
pweek, pday = weeks_days[n + post_n]
p = 1
except:
pweek, pday = weeks_days[n]
p = 0
post_map.append((p, pweek, pday))
# p1 = 1
# p2 = 1
# p3 = 1
# p4 = 1
# p5 = 1
# p6 = 1
# p7 = 1
# if n > 0:
# pweek, pday = weeks_days[n - 1]
# else:
# p1 = 0
# pweek, pday = weeks_days[n]
# try:
# p2week, p2day = weeks_days[n - 2]
# except IndexError:
# p2week, p2day = weeks_days[n]
# p2 = 0
# try:
# p3week, p3day = weeks_days[n - 3]
# except IndexError:
# p3week, p3day = weeks_days[n]
# p3 = 0
# try:
# p4week, p4day = weeks_days[n - 4]
# except IndexError:
# p4week, p4day = weeks_days[n]
# p4 = 0
# try:
# p5week, p5day = weeks_days[n - 5]
# except IndexError:
# p5week, p5day = weeks_days[n]
# p5 = 0
# try:
# p6week, p6day = weeks_days[n - 6]
# except IndexError:
# p6week, p6day = weeks_days[n]
# p6 = 0
# try:
# p7week, p7day = weeks_days[n - 7]
# except IndexError:
# p7week, p7day = weeks_days[n]
# p7 = 0
# pre_map = [
# (p1, pweek, pday),
# (p2, p2week, p2day),
# (p3, p3week, p3day),
# (p4, p4week, p4day),
# (p5, p5week, p5day),
# (p6, p6week, p6day),
# (p7, p7week, p7day),
# ]
# n1 = 1
# n2 = 1
# n3 = 1
# n4 = 1
# n5 = 1
# n6 = 1
# n7 = 1
# if n > 0:
# nweek, nday = weeks_days[n - 1]
# else:
# n1 = 0
# nweek, nday = weeks_days[n]
# try:
# n2week, n2day = weeks_days[n - 2]
# except IndexError:
# n2week, n2day = weeks_days[n]
# n2 = 0
# try:
# n3week, n3day = weeks_days[n - 3]
# except IndexError:
# n3week, n3day = weeks_days[n]
# n3 = 0
# try:
# n4week, n4day = weeks_days[n - 4]
# except IndexError:
# n4week, n4day = weeks_days[n]
# n4 = 0
# try:
# n5week, n5day = weeks_days[n - 5]
# except IndexError:
# n5week, n5day = weeks_days[n]
# n5 = 0
# try:
# n6week, n6day = weeks_days[n - 6]
# except IndexError:
# n6week, n6day = weeks_days[n]
# n6 = 0
# try:
# n7week, n7day = weeks_days[n - 7]
# except IndexError:
# n7week, n7day = weeks_days[n]
# n7 = 0
# post_map = [
# (n1, nweek, nday),
# (n2, n2week, n2day),
# (n3, n3week, n3day),
# (n4, n4week, n4day),
# (n5, n5week, n5day),
# (n6, n6week, n6day),
# (n7, n7week, n7day),
# ]
# n1 = 1
# n2 = 1
# try:
# nweek, nday = weeks_days[n + 1]
# except IndexError:
# nweek, nday = weeks_days[n]
# n1 = 0
# try:
# n2week, n2day = weeks_days[n + 2]
# except IndexError:
# n2week, n2day = weeks_days[n]
# n2 = 0
# try:
# n3week, n3day = weeks_days[n + 3]
# except IndexError:
# n3week, n3day = weeks_days[n]
# n3 = 0
# IF paired we check the following against both workers
workers = [worker]
if self.constraint_options["hard_constrain_pair_separation"]:
for worker_pairs in self.worker_pairs:
if worker_pairs[0] == worker:
workers = worker_pairs
if self.get_shift_names_by_week_day(week, day):
# Unable to work (hard constraint not preference)
self.model.constraints.add(
self.model.available[worker.id, week, day]
>= sum(
self.model.works[worker.id, week, day, shift]
for shift in self.get_shift_names_by_week_day(week, day)
)
)
# single shift 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)
)
)
if self.constraint_options["hard_constrain_pair_separation"]:
for worker_pairs in self.worker_pairs:
if worker_pairs[0] == worker:
self.model.constraints.add(
1
>= sum(
self.model.works[w.id, week, day, shift]
for shift in self.get_shift_names_by_week_day(
week, day
)
for w in worker_pairs
)
)
for constraint_shift in self.get_shifts_with_constraints(
"pre",
):
for n in range(0, constraint_shift.constraint_options["pre"]):
if day in constraint_shift.days:
self.model.constraints.add(
1
>= self.model.works[
worker.id, week, day, constraint_shift.name
]
+ sum(
pre_map[n][0]
* self.model.works[
w.id, pre_map[n][1], pre_map[n][2], shiftname
]
for shiftname in self.get_shift_names_by_week_day(
pre_map[n][1], pre_map[n][2]
)
if shiftname != constraint_shift.name
for w in workers
)
)
# for constraint_shift in self.get_shifts_with_constraint( "pre"):
# to_sum = []
# for n in range(
# 0, constraint_shift.constraint_options["pre"]
# ):
# for w in workers:
# for wk, dy, shiftname in self.get_week_day_shift_names_by_week_day(
# pre_map[n][1], pre_map[n][2]
# ):
# if shiftname != constraint_shift.name:
# to_sum.append(pre_map[n][0]
# * self.model.works[
# w.id, wk, dy, shiftname
# ])
# if day in constraint_shift.days:
# self.model.constraints.add(
# 1
# >= self.model.works[
# worker.id, week, day, constraint_shift.name
# ]
# + sum(
# to_sum
# )
# )
for constraint_shift in self.get_shifts_with_constraints(
"post",
):
for n in range(0, constraint_shift.constraint_options["post"]):
if day in constraint_shift.days:
self.model.constraints.add(
1
>= self.model.works[
worker.id, week, day, constraint_shift.name
]
+ sum(
post_map[n][0]
* self.model.works[
w.id, post_map[n][1], post_map[n][2], shiftname
]
for shiftname in self.get_shift_names_by_week_day(
post_map[n][1], post_map[n][2]
)
if shiftname != constraint_shift.name
for w in workers
)
)
# for constraint_shift in self.get_shifts_with_constraints(
# "preclear", "preclear2"
# ):
# if day in constraint_shift.days:
# self.model.constraints.add(
# 1
# >= self.model.works[
# worker.id, week, day, constraint_shift.name
# ]
# + sum(
# p1 * self.model.works[w.id, pweek, pday, shiftname]
# for shiftname in self.get_shift_names_by_week_day(
# pweek, pday
# )
# if shiftname != constraint_shift.name
# for w in workers
# )
# )
#
# for constraint_shift in self.get_shifts_with_constraint("preclear2"):
# if day in constraint_shift.days:
# self.model.constraints.add(
# 1
# >= self.model.works[
# worker.id, week, day, constraint_shift.name
# ]
# + sum(
# p2 * self.model.works[w.id, p2week, p2day, shiftname]
# for shiftname in self.get_shift_names_by_week_day(
# p2week, p2day
# )
# if shiftname != constraint_shift.name
# for w in workers
# )
# )
# for constraint_shift in self.get_shifts_with_constraints(
# "postclear", "postclear2"
# ):
# if day in constraint_shift.days:
# self.model.constraints.add(
# 1
# >= self.model.works[
# worker.id, week, day, constraint_shift.name
# ]
# + sum(
# n1 * self.model.works[w.id, nweek, nday, shiftname]
# for shiftname in self.get_shift_names_by_week_day(
# nweek, nday
# )
# if shiftname != constraint_shift.name
# for w in workers
# )
# )
# for constraint_shift in self.get_shifts_with_constraints("postclear2"):
# if day in constraint_shift.days:
# self.model.constraints.add(
# 1
# >= self.model.works[
# worker.id, week, day, constraint_shift.name
# ]
# + sum(
# n2 * self.model.works[w.id, n2week, n2day, shiftname]
# for shiftname in self.get_shift_names_by_week_day(
# n2week, n2day
# )
# if shiftname != constraint_shift.name
# for w in workers
# )
# )
# Night constraint means we won't assign a shift the day before
# an unavailability
for constraint_shift in self.get_shifts_with_constraint("night"):
if (
worker.id,
# pweek,
pre_map[0][1],
# pday,
pre_map[0][2],
constraint_shift.name,
) in self.model.works:
# Ensure night prior to unavalibity is not assigned
self.model.constraints.add(
self.model.available[worker.id, week, day]
>= self.model.works[
# worker.id, pweek, pday, constraint_shift.name
worker.id,
pre_map[0][1],
pre_map[0][2],
constraint_shift.name,
]
)
self.define_objectives()
print("Building model completed")
def define_objectives(self):
# Define an objective function with model as input, to pass later
def obj_rule(m):
# c = len(workers)
balance_modifier_constant = 1
if self.constraint_options["balance_shifts_over_workers"]:
worker_shift_balancing = sum(
balance_modifier_constant
* (
self.model.worker_shift_count_t1[(worker.id)]
+ self.model.worker_shift_count_t2[(worker.id)]
)
for worker in self.workers
)
else:
worker_shift_balancing = 0
if self.constraint_options["balance_shifts"]:
shift_balancing = sum(
balance_modifier_constant
* (
self.model.shift_count_t1[worker.id, shift.name]
+ self.model.shift_count_t2[worker.id, shift.name]
)
for worker in self.workers
for shift in self.get_shifts()
)
else:
shift_balancing = 0
if self.constraint_options["balance_shifts_quadratic"]:
shift_balancing = sum(
balance_modifier_constant
* (
self.model.shift_count_t1[worker.id, shift.name]
+ self.model.shift_count_t2[worker.id, shift.name]
)
for worker in self.workers
for shift in self.get_shifts()
)
else:
shift_balancing = 0
if self.constraint_options["minimise_shift_diffs"]:
shift_diff_modifier_constant = 10
shift_diff_balancing = sum(
shift_diff_modifier_constant
* self.model.shift_count_diff_summed[(worker.id)]
for worker in self.workers
)
else:
shift_diff_balancing = 0
if self.constraint_options["balance_nights"]:
night_balance_modifier_constant = 10
night_shift_balancing = sum(
night_balance_modifier_constant
* self.model.night_shift_count_w[(worker.id)]
# (self.model.night_shift_count_t1[(worker.id)] +
# self.model.night_shift_count_t2[(worker.id)])
for worker in self.workers
)
else:
night_shift_balancing = 0
if self.constraint_options["balance_bank_holidays"]:
bank_holiday_balance_modifier_constant = 1000
bank_holiday_balancing = sum(
bank_holiday_balance_modifier_constant
* (self.model.bank_holiday_count_w[(worker.id)] - 1)
# (self.model.night_shift_count_t1[(worker.id)] +
# self.model.night_shift_count_t2[(worker.id)])
for worker in self.workers
)
else:
bank_holiday_balancing = 0
if self.constraint_options["balance_weekends"]:
weekend_balance_modifier_constant = 5
weekend_shift_balancing = sum(
weekend_balance_modifier_constant
* self.model.weekend_shift_count_w[(worker.id)]
for worker in self.workers
)
else:
weekend_shift_balancing = 0
preference_constant = 10
# Preferences (not to work)
preferences = sum(
self.model.pref_not_to_work[worker.id, week, day]
* self.model.works[worker.id, week, day, shift]
* preference_constant
for worker in self.workers
for week, day, shift in self.get_all_shiftname_combinations()
)
# Work requsets
work_request_constant = 10000 # Needs to be very big to override loss from bank holiday requests
work_requests = sum(
self.model.work_requests[worker.id, week, day, shift]
* self.model.works[worker.id, week, day, shift]
* work_request_constant
for worker in self.workers
for week, day, shift in self.get_all_shiftname_combinations()
)
# # Spread nights
if self.constraint_options["balance_nights_across_sites"]:
nights_site_balancing = sum(
(
self.model.night_per_site_t1[week, shift.name, site]
+ self.model.night_per_site_t2[week, shift.name, site]
)
* 20
# self.model.night_per_site2[week, block, site]
for week in self.weeks
for shift in self.get_shifts_with_constraint("night")
for site in self.sites
)
else:
nights_site_balancing = 0
if self.get_shifts_with_constraint("balance_across_groups"):
block_site_balancing = sum(
(
self.model.shift_per_site_t1[week, shift.name, site]
+ self.model.shift_per_site_t2[week, shift.name, site]
)
* 2000
# self.model.night_per_site2[week, block, site]
for week in self.weeks
for shift in self.get_shifts_with_constraint(
"balance_across_groups"
)
for site in self.sites
)
else:
block_site_balancing = 0
if self.constraint_options["balance_blocks"]:
blocks_balancing = sum(
1 * self.model.blocks_assigned[week, shift]
for week in self.weeks
for shift in self.shifts_to_assign_as_blocks()
)
else:
blocks_balancing = 0
# Quadratic :(
# worker_pairs_balancing = 0
# worker_pairs_constant = 100000
# if self.worker_pairs:
# for worker_a, worker_b in self.worker_pairs:
# print(worker_a, worker_b)
# for week, day in self.get_week_day_combinations():
# worker_pairs_balancing = worker_pairs_balancing + sum(self.model.works[worker_a.id, week, day, shift] for shift in self.get_shift_names_by_week_day(week, day)) * sum(self.model.works[worker_b.id, week, day, shift] for shift in self.get_shift_names_by_week_day(week, day)) * worker_pairs_constant
return (
weekend_shift_balancing
+ shift_balancing
+ worker_shift_balancing
+ night_shift_balancing
+ shift_diff_balancing
+ preferences
+ nights_site_balancing
+ block_site_balancing
+ bank_holiday_balancing
+ blocks_balancing
- work_requests
)
# add objective function to the model. rule (pass function) or expr (pass expression directly)
self.model.obj = Objective(rule=obj_rule, sense=minimize)
def add_worker(self, worker: Worker) -> None:
"""Add a worker to the rota
Args:
worker (Worker):
"""
self.workers.append(worker)
def add_workers(self, workers: List) -> None:
"""Add multiple workers to the rota
Args:
workers (List(Worker)):
"""
for worker in workers:
self.add_worker(worker)
# self.workers.extend(workers)
def build_workers(self) -> None:
"""Process loaded shifts and workers
Must be called prior to attempting to solve
"""
if not self.workers:
raise NoWorkers("Workers must be added prior to calling build_workers")
self.workers_id_map = {}
self.workers_name_map = {}
for worker in track(self.workers, description="Building workers"):
worker.load_rota(self)
wid = worker.id
if wid in self.workers_id_map:
raise ValueError(f"Worker with id '{wid}' has been added twice")
self.workers_id_map[wid] = worker
if worker.name in self.workers_name_map:
raise ValueError(
f"Worker with name '{worker.name}' has been added twice"
)
self.workers_name_map[worker.name] = worker
self.workers = sorted(self.workers)
self.full_time_equivalent = sum(w.fte_adj for w in self.workers)
self.full_time_equivalent_sites = {}
self.workers_at_sites = {}
for site in track(self.sites, description="Generating site workers"):
self.workers_at_sites[site] = [w for w in self.workers if w.site == site]
self.full_time_equivalent_sites[site] = sum(
w.fte_adj for w in self.workers if w.site == site
)
self.worker_pairs = []
pairs = defaultdict(list)
for w in self.workers:
if w.pair is not None:
pairs[w.pair].append(w)
for p in pairs:
self.worker_pairs.append(tuple(pairs[p]))
def add_grade_constraint_by_week(
self, grades: Iterable[int], weeks: Iterable[int], shifts
):
self.constraint_options["avoid_shifts_by_grades"].append(
{"grades": grades, "weeks": weeks, "shifts": shifts}
)
def add_shift(self, shift: SingleShift) -> None:
"""Add a shift to the collection
:param SingleShift shift: Shift object
"""
self.shifts.append(shift)
def add_shifts(self, *shifts: SingleShift ) -> None:
"""Add multiple shifts
Returns:
None
"""
self.shifts.extend(shifts)
def clear_shifts(self) -> None:
self.shifts = []
def get_shift_names_by_week_day(self, week, day: DayStr) -> set:
"""Returns the shifts required for a specific day
Returns:
set: Set of ShiftName
"""
return self.week_day_shifts_dict[(week, day)]
def get_week_day_shift_names_by_week_day(self, week, day: DayStr) -> set:
"""Returns the shifts required for a specific day
Returns:
set: Set of ShiftName
"""
return set(
(week, day, shift_name)
for shift_name in self.week_day_shifts_dict[(week, day)]
)
def get_shift_by_name(self, name: str) -> SingleShift:
"""
:param name:
"""
return self.shifts_by_name[name]
def get_shift_length_by_name(self, name):
"""Get the length of a shift
Args:
name (str): Name of a shift
Returns:
int: Length of the shift
"""
return self.shifts_by_name[name].length
def build_shifts(self):
"""
Process the added shifts
"""
self.shifts_by_name = {}
self.shift_names = [] # type: List[ShiftName]
# self.week_day_shifts_dict = {(week, day): set() for week in self.weeks for day in days}
self.week_day_shifts_dict = {}
self.sites = set()
self.week_day_shift_product = []
self.week_day_shiftclass_product = []
for s in self.shifts:
pass
for s in self.shifts:
if s.name in self.shift_names:
raise InvalidShift(f"Duplicate shift: {s.name}")
self.shifts_by_name[s.name] = s
self.shift_names.append(s.name)
# for day in s.days:
# self.week_day_shifts_dict[(week, day)].add(s.name)
for site in list(s.sites):
self.sites.add(site)
self.shift_counts = defaultdict(int)
for week, day in self.weeks_days_product:
self.week_day_shifts_dict[(week, day)] = set()
for s in self.shifts:
if s.bank_holidays_only:
if self.week_day_date_map[(week, day)] in bank_holiday_map:
print(s.name, self.week_day_date_map[(week, day)])
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
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
# print(self.week_day_shift_product)
self.max_pre = 1
self.max_post = 1
for shift in self.get_shifts_with_constraint("pre"):
self.max_pre = max(shift.constraint_options["pre"], self.max_pre)
for shift in self.get_shifts_with_constraint("post"):
self.max_post = max(shift.constraint_options["post"], self.max_post)
# # Todo replace with week_day.....
# self.day_shift_product = []
# self.day_shiftclass_product = []
# for s in self.shifts:
# for day in days:
# if day in s.days:
# self.day_shift_product.append((day, s.name))
# self.day_shiftclass_product.append((day, s))
#
# def get_day_shiftname_combinations(self):
# """Returns a list of all required day / shift combinations
#
# Returns:
# list: list of in the following format (str->day, str->shift)
# """
# return self.day_shift_product
#
# def get_day_shiftclass_combinations(self) -> List[Tuple[DayStr, SingleShift]]:
# """Returns a list of all required day / shift combinations
#
# Returns:
# list: list of in the following format (str->day, shiftObject->shift)
# """
# return self.day_shiftclass_product
def get_all_shiftname_combinations(
self, week: int = None, day: int = None
) -> List[Tuple[WeekInt, DayStr, ShiftName]]:
"""Returns a list of all possible week / day / shift combinations
Returns:
list: contains tuple of all possible week / day / shift combinations
"""
week_day_shifts = self.week_day_shift_product
if week is not None:
week_day_shifts = [t for t in week_day_shifts if t[0] == week]
if day is not None:
week_day_shifts = [t for t in week_day_shifts if t[1] == day]
return week_day_shifts
def get_all_shiftclass_combinations(
self, week: int = None, day: int = None
) -> List[Tuple[WeekInt, DayStr, SingleShift]]:
"""Returns a list of all possible week / day / shift combinations
Returns:
list: contains tuple of all possible week / day / shift combinations
"""
week_day_shifts = self.week_day_shiftclass_product
if week is not None:
week_day_shifts = [t for t in week_day_shifts if t[0] == week]
if day is not None:
week_day_shifts = [t for t in week_day_shifts if t[1] == day]
return week_day_shifts
def get_week_day_combinations(self) -> list:
"""Returns a list of all week / day tuple combinations
Returns:
list: list of possible week day combinations
[(1, "Mon"), (1, "Tue"), ... (n, "Fri")]
"""
return self.weeks_days_product
def get_week_day_combinations_for_shift(self, shift) -> list:
return [
(week, day)
for week, day in self.get_week_day_combinations()
if shift.name in self.get_shift_names_by_week_day(week, day)
]
def get_week_day_shift_combinations_for_constraint(self, constraint) -> list:
constraint_shifts = self.get_shifts_with_constraint(constraint)
week_day_shifts = []
for tup in self.get_all_shiftclass_combinations():
if tup[2] in constraint_shifts:
week_day_shifts.append(tup)
return week_day_shifts
def get_shifts(self) -> List[SingleShift]:
"""Returns a list of all the registered shifts
Returns:
List[SingleShift]: list of registered shifts (as SingleShift)
"""
return self.shifts
def get_shifts_for_worker_site(self, worker_site):
shifts = [shift for shift in self.shifts if worker_site in shift.sites]
return shifts
def get_shifts_for_worker(self, worker_id):
worker = self.get_worker_by_id(worker_id)
shifts = [shift for shift in self.shifts if worker.site in shift.sites]
return shifts
def get_shifts_with_constraint(self, constraint) -> List[SingleShift]:
return [shift for shift in self.shifts if constraint in shift.constraints]
def get_shifts_with_constraints(self, *constraints) -> List[SingleShift]:
shift_names = set()
for constraint in constraints:
shift_names.update(
[shift.name for shift in self.shifts if constraint in shift.constraints]
)
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:
List[ShiftName]: List of names of all available shifts
""" """ """
return self.shift_names
def get_week_block_iterator(self, block_length: int):
"""Gets a two dimensional list of week blocks in specified length
e.g. block_length = 4
[
[ 1, 2, 3, 4],
[ 2, 3, 4, 5],
...
]
Args:
block_length (int): length of blocks to create
Returns:
list: two dimensional list containing weeks in blocks
"""
blocks = []
for i in range(len(self.weeks) - block_length + 1):
block = self.weeks[i : i + block_length]
blocks.append(block)
return blocks
def get_required_workers_and_site_combinations(self):
"""Returns a list of all possible shifts, the workers required and site
Returns:
list: list (week, day, shift name, workers required, shift.sites)
"""
l = []
for week, day, shift in self.week_day_shiftclass_product:
# if day in shift.days:
l.append((week, day, shift.name, shift.workers_required, shift.sites))
return l
def get_not_required_shifts(self):
"""Returns a set of all possible shifts combinations that are not required
Includes those on days
Returns:
list: (week, day, shift)
"""
l = set()
for week in self.weeks:
for day in days:
for shift in self.shifts:
l.add((week, day, shift))
return l - set(self.get_all_shiftclass_combinations())
def shifts_to_assign_as_blocks(self) -> List[str]:
return [shift.name for shift in self.get_shifts() if shift.assign_as_block]
def shifts_to_force_as_blocks(self) -> List[str]:
return [
shift.name
for shift in self.get_shifts()
if (shift.force_as_block or shift.force_as_block_unless_nwd)
]
def shifts_to_assign_or_force_as_blocks(self) -> List[str]:
s = self.shifts_to_assign_as_blocks()
s.extend(self.shifts_to_force_as_blocks())
return s
def get_worker_by_id(self, worker_id: str) -> Worker:
return self.workers_id_map[worker_id]
def get_worker_by_name(self, name: str) -> Worker:
return self.workers_name_map[name]
def get_workers_by_group(self) -> Dict[str, Worker]:
group_workers = defaultdict(list)
for worker in self.workers:
group_workers[worker.site].append(worker)
return group_workers
def get_workers_by_remote_group(self) -> Dict[str, Worker]:
group_workers = defaultdict(list)
for worker in self.workers:
group_workers[worker.remote_site].append(worker)
return group_workers
def get_workers_by_grade(self) -> Dict[str, Worker]:
group_workers = defaultdict(list)
for worker in self.workers:
group_workers[worker.grade].append(worker)
return group_workers
def get_workers_for_shift(self, shift: SingleShift) -> List[Worker]:
return [worker for worker in self.workers if worker.site in shift.sites]
def get_workers_total_fte(self) -> float:
return self.full_time_equivalent
def get_worker_details(self):
w = defaultdict(list)
for worker in self.workers:
# print(worker)
w[worker.site].append(worker)
l = []
for site in w:
l.append(f"{site}: {sum(worker.fte_adj for worker in w[site]) / 100}")
t = "\n".join(l)
return f"Full time equivalent trainees by site:\n{t}"
def get_workers(self):
return self.workers
def get_worker_grades(self) -> set[int]:
return set([worker.grade for worker in self.workers])
def get_bank_holiday_week_days(self):
return [
(week, day)
for week, day in self.get_week_day_combinations()
if self.week_day_date_map[(week, day)] in bank_holiday_map
]
def get_week_start_date(self, week: int):
return self.start_date + datetime.timedelta(weeks=week)
# RESULTS
def export_rota_to_html(self, filename: str = "rota", folder=None):
if folder is not None:
output_file = Path("output", folder, f"{filename}.html")
else:
output_file = Path("output", f"{filename}.html")
# Make sure path exits
output_file.parent.mkdir(parents=True, exist_ok=True)
with open(output_file, "w") as f:
f.write(self.get_worker_timetable_html(True))
def export_rota_to_csv(self, filename: str = "rota"):
output_file = Path("output", f"{filename}.csv")
works = self.model.works
with open(output_file, "w", newline="") as f:
wr = csv.writer(f, quoting=csv.QUOTE_ALL)
l = ["Name"]
l.extend([worker.name for worker in self.workers])
wr.writerow(l)
l2 = ["Site"]
l2.extend([worker.site for worker in self.workers])
wr.writerow(l2)
l3 = ["Grade"]
l3.extend([worker.grade for worker in self.workers])
wr.writerow(l3)
l4 = ["FTE"]
l4.extend([worker.fte for worker in self.workers])
wr.writerow(l4)
for week, day in self.get_week_day_combinations():
d = [f"Week {week} Day {day}"]
for worker in self.workers:
i = ""
for shift in self.get_shift_names_by_week_day(week, day):
if works[worker.id, week, day, shift].value == 1:
i = shift
d.append(i)
wr.writerow(d)
def get_work_table(self):
"""Build a timetable of the week as a dictionary from the model's optimal solution."""
works = self.model.works
week_table = {
week: {day: {shift: [] for shift in self.get_shift_names()} for day in days}
for week in self.weeks
}
for week in self.weeks:
for worker in self.workers:
for day, shift in self.get_day_shiftname_combinations():
if works[worker.id, week, day, shift].value == 1:
week_table[week][day][shift].append(worker.get_details())
return week_table
def get_worker_timetable(self):
works = self.model.works
timetable = {
worker.name: {week: {day: "" for day in days} for week in self.weeks}
for worker in self.workers
}
for worker in self.workers:
for week in self.weeks:
for day, shift in self.get_day_shiftname_combinations():
if works[worker.id, week, day, shift].value == 1:
timetable[worker.name][week][day] = shift
return timetable
def get_worker_timetable_brief(
self, show_prefs=False, marker_every=30, show_unavailable=False
):
model = self.model
week_string = f"{'-Week-':20}" + "".join(
[7 * str(f"{w}")[-1:] for w in self.weeks]
)
days_string = f"{'-Day-':20}" + "".join("MTWTFSS" * len(self.weeks))
timetable = []
for worker in self.workers:
shifts = []
w = [f"{worker.name:20}"]
for week, day in self.get_week_day_combinations():
a = "-"
for shift in self.get_shift_names_by_week_day(week, day):
if model.works[worker.id, week, day, shift].value > 0:
shifts.append(shift)
a = shift[0]
w.append(a)
shift_count = ""
for s in set(shifts):
shift_count = shift_count + f"{s}: {shifts.count(s)}, "
shift_count = (
shift_count
+ f"#weekends_worked: {model.worker_weekend_count[worker.id].value}\\#"
)
timetable.append(f"{''.join(w)} {shift_count}")
if show_prefs:
# prefs
w = [f"{'Preferences':20}"]
for week, day in self.get_week_day_combinations():
if model.pref_not_to_work[worker.id, week, day] > 0:
w.append("Y")
else:
w.append("N")
timetable.append("".join(w))
if show_unavailable:
# prefs
w = [f"{'Unavailable':20}"]
for week, day in self.get_week_day_combinations():
if model.available[worker.id, week, day] > 0:
w.append("A")
else:
w.append("U")
timetable.append("".join(w))
i = marker_every
while i < len(timetable):
timetable.insert(i, week_string)
timetable.insert(i + 1, days_string)
i = i + marker_every + 2
timetable.append(week_string)
timetable.insert(0, week_string)
timetable.insert(0, days_string)
return "\n".join(timetable)
def get_worker_timetable_html(
self, include_html_tag=False, table_name="rota-table"
):
model = self.model
timetable = []
timetable.append(
f"<h2>Rota start date: {self.start_date.isoformat()} ({self.weeks[-1]} weeks)</h2>"
)
date_row = ["<th class='worker'></th>"]
n = 0
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
th_class = "date"
if d in bank_holiday_map:
th_class = th_class + " bank-holiday"
date_row.append(
f"<th title='{d}' class='{th_class}' data-date='{d}'>Week {week}: {day}</th>"
)
n = n + 1
timetable.append(f"<tr class='data-row'>{''.join(date_row)}</tr>")
current_site = ""
for worker in self.workers:
if worker.site != current_site:
try:
timetable.append(
f"<tr><th class='site-title'>{worker.site} n={len(self.workers_at_sites[worker.site])} fte={self.full_time_equivalent_sites[worker.site]}</th><tr>"
)
except KeyError as e:
print(e)
current_site = worker.site
shifts = []
nwds = json.dumps(worker.non_working_day_list, default=str)
# if worker.nwd:
# # TODO: limit to dates
# nwds = ", ".join([i[0] for i in worker.nwd])
# else:
# nwds = None
worker_targets = json.dumps(worker.shift_target_number)
tds = []
n = 0
shift_tds = []
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
n = n + 1
a = "-"
shift_name = ""
# Loop through all the days possible shifts and see
# if the worker has been assigned
for shift in self.get_shift_names_by_week_day(week, day):
if model.works[worker.id, week, day, shift].value > 0:
shifts.append(shift)
a = shift[0]
shift_name = shift
break
title = f"{shift_name} ({d})"
css_class = day
unavailable_reason = ""
if model.available[worker.id, week, day] > 0:
available = True
else:
available = False
css_class = "unavailable"
try:
unavailable_reason = self.unavailable_to_work_reason[
(worker.id, week, day)
]
except KeyError:
print("Error getting reason")
print(worker.id, worker.name)
print(f"Week {week}, Day {day}")
unavailable_reason = "????"
title = f"{title} / {unavailable_reason}"
bank_holiday = ""
if d in bank_holiday_map:
title = f"{title} [Bank Holiday]"
css_class = " ".join((css_class, "bank-holiday"))
bank_holiday = f" data-bank-holiday='{bank_holiday_map[d]}'"
requests = ""
if (worker.id, week, day) in self.work_requests_map:
css_class = " ".join((css_class, "shift-requested"))
title = " ".join((title, "[REQUESTED]"))
requests = f" data-shift-request='{self.work_requests_map[worker.id, week, day]}'"
remote_site = ""
if shift_name:
shift = self.get_shift_by_name(shift_name)
if "require_remote_site_presence_week" in shift.constraints:
remote_site = f" data-shift-remote-site='{shift.constraint_options['require_remote_site_presence_week'][0]}'"
shift_tds.append(
f"<td title='{title}' class='rota-day {css_class}' data-shift='{shift_name}' data-available='{available}' data-unavailable_reason='{unavailable_reason}' data-date='{d}' data-week='{week}' data-day='{day}'{remote_site}{requests}{bank_holiday}>{a}</td>"
)
shift_count = ""
shift_count_dict = {}
for s in set(shifts):
c = shifts.count(s)
shift_count_dict[s] = c
shift_count = shift_count + f"{s}: {c}, "
shift_diff_dict = {}
for shift in self.get_shifts():
diff = model.shift_count_diff[worker.id, shift.name].value
shift_diff_dict[shift.name] = diff
worker_td = """<td title='Site: {site}' class='worker {site}'
data-nwds='{nwds}' data-site='{site}' data-worker='{name}'
data-fte='{fte}' data-fte_adj='{fte_adj}'
data-start_date='{start_date}'
data-end_date='{end_date}'
data-worker-targets='{targets}' data-shift-counts='{worker_shift_counts}'
data-weekend-target='{weekend_target}'
data-remote-site='{remote_site}'
data-pair='{pair}'
data-shift-balance-extra='{shift_balance_extra}'
data-bank-holiday-extra='{bank_holiday_extra}'
data-shift-diff='{shift_diff}'
data-shift-diff-summed='{shift_diff_summed}'
>
<span class='name' title='{name}'>{name}</span> ({grade}) [{fte}]</td>""".format(
site=worker.site,
nwds=nwds,
name=worker.name,
fte=worker.fte,
fte_adj=worker.fte_adj,
start_date=worker.calculated_start_date,
end_date=worker.calculated_end_date,
targets=worker_targets,
worker_shift_counts=json.dumps(shift_count_dict),
weekend_target=worker.weekend_shift_target_number,
remote_site=worker.remote_site,
grade=worker.grade,
pair=worker.pair,
shift_balance_extra=json.dumps(worker.shift_balance_extra),
shift_diff=json.dumps(shift_diff_dict),
shift_diff_summed=model.shift_count_diff_summed[worker.id].value,
bank_holiday_extra=worker.bank_holiday_extra,
)
shift_count = (
shift_count
+ f"#weekends_worked: {model.worker_weekend_count[worker.id].value}\\#"
)
if self.constraint_options["balance_bank_holidays"]:
bank_holiday_count = model.bank_holiday_count[worker.id].value
bank_holiday_count_w = model.bank_holiday_count_w[worker.id].value - 1
else:
bank_holiday_count = -1
bank_holiday_count_w = -1
# print(worker.name, bank_holiday_count, bank_holiday_count_w)
timetable.append(
f"<tr class='worker-row'>{worker_td}{''.join(shift_tds)}</tr>"
)
# timetable.append("<tr>{}</tr>".format("".join(w), shift_count))
# if show_prefs:
# # prefs
# w = ["{:20}".format("Preferences")]
# for week, day in self.get_week_day_combinations():
# if model.pref_not_to_work[worker.id, week, day] > 0:
# w.append("Y")
# else:
# w.append("N")
# timetable.append("".join(w))
# if show_unavailable:
# # prefs
# w = ["{:20}".format("Unavailable")]
# for week, day in self.get_week_day_combinations():
# if model.available[worker.id, week, day] > 0:
# w.append("A")
# else:
# w.append("U")
# timetable.append("".join(w))
result_stream = StringIO()
if self.results is not None:
self.results.write(ostream=result_stream)
self.results.write_json(ostream=result_stream)
result_string = result_stream.getvalue()
else:
result_string = "Rota not run"
joined_timetable = "\n".join(timetable)
html = f"""
<body>
<div class="table-div" id="{table_name}">
<table id="main-table">{joined_timetable}</table>
</div>
<details>
<summary><h2>Rota settings</h2></summary>
<div>
<pre>
{json.dumps(self.constraint_options, indent=4)}
</pre>
</details>
</div>
<details>
<summary><h2>Shifts settings</h2></summary>
<div id="shifts-container" data-shifts='{json.dumps([i.name for i in self.shifts])}'>
{"<br/>".join(str(i) for i in self.shifts)}
</div>
</details>
<details>
<summary><h2>Output</h2></summary>
<pre>
{result_string}
</pre>
</details>
<div>
<details>
<summary><h2>Export table</h2></summary>
<div id="export-div">
<div>
</details>
<details>
<summary><h2>Shift timetables</h2></summary>
<div id="shift-timetable-options">
</div>
<div id="shift-timetable-div">
<div>
</details>
</div
</body>
"""
if include_html_tag:
html = f"""<html>
<head>
<link rel="stylesheet" type="text/css" href="timetable.css">
<link rel="stylesheet" type="text/css" href="https://code.jquery.com/ui/1.13.0/themes/base/jquery-ui.css">
<script src="https://ajax.googleapis.com/ajax/libs/jquery/3.5.1/jquery.min.js"></script>
<script src="https://code.jquery.com/ui/1.13.0/jquery-ui.min.js"></script>
<script src="timetable.js" defer></script>
</head>
{html}</html>"""
return html
def get_shift_summary_dict(self):
""""""
timetable = {
worker.name: {shift: "" for shift in self.get_shift_names()}
for worker in self.workers
}
for worker in self.workers:
for shift in self.get_shift_names():
timetable[worker.name][shift] = self.model.shift_count[
worker.id, shift
].value
return timetable
def get_worker_shifts_by_date(self, worker: Worker) -> Dict:
shifts = {}
n = 0
for week, day in self.get_week_day_combinations():
d = self.start_date + datetime.timedelta(n)
n = n + 1
temp = ""
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0:
shifts[d] = shift
return shifts
def get_workers_total_shifts(self) -> Dict[str, int]:
shifts = {}
for worker in self.workers:
shifts[worker.name] = len(
[i for i in self.get_worker_shift_list(worker) if i != ""]
)
return shifts
def get_worker_shift_list(self, worker: Worker) -> List:
shifts = []
for week, day in self.get_week_day_combinations():
# d = self.start_date + datetime.timedelta(n)
# n = n + 1
temp = ""
for shift in self.get_shift_names_by_week_day(week, day):
if self.model.works[worker.id, week, day, shift].value > 0:
temp = shift
shifts.append(temp)
return shifts
def get_worker_shift_list_string(self, worker: Worker) -> str:
shifts = self.get_worker_shift_list(worker)
# Convert shift to a string representation
return "".join([i if i != "" else "-" for i in shifts])
def get_shift_summary(self):
works = self.model.works
timetable = {
worker.get_details(): {shift: "" for shift in self.get_shift_names()}
for worker in self.workers
}
# timetable = { worker.get_details() : { shift : "" for shift in ["truro_twilight"] } for worker in workers }
t = []
for worker in self.workers:
l = []
total_shifts = 0
for shift in self.get_shift_names():
# for shift in ["truro_twilight"]:
c = [
works[worker.id, week, day, shift].value
for week in self.weeks
for day in days
].count(1)
if c > 0:
l.append(f"{shift} ({c})")
total_shifts = total_shifts + c
# print(worker.id, shift)
timetable[worker.get_details()][shift] = c
t.append(f"{worker.get_full_details()} [{total_shifts}]: {', '.join(l)}")
return "\n".join(t)
def get_shift_summary_html(self):
works = self.model.works
timetable = {
worker.get_details(): {shift[0]: "" for shift in self.get_shift_names()}
for worker in self.workers
}
# timetable = { worker.get_details() : { shift : "" for shift in ["truro_twilight"] } for worker in workers }
t = []
for worker in self.workers:
l = []
total_shifts = 0
for shift in self.get_shift_names():
# for shift in ["truro_twilight"]:
c = [
works[worker.id, week, day, shift]
for week in self.weeks
for day in days
].count(1)
if c > 0:
l.append(f"{shift} ({c})")
total_shifts = total_shifts + c
# print(worker.id, shift)
timetable[worker.get_details()][shift[0]] = c
t.append(f"{worker.get_full_details()} [{total_shifts}]: {', '.join(l)}")
return "\n".join(t)
class NoActiveSites(Exception):
"""Raised when there are no active sites"""
pass
class NoWorkers(Exception):
"""Raised when there are no active sites"""
pass
class InvalidShift(Exception):
"""Raised when there are no active sites"""
pass
Reference in New Issue View Git Blame Copy Permalink