Ross
1056 lines
41 KiB
Python
1056 lines
41 KiB
Python
import datetime
|
|
import itertools
|
|
from typing import List, Tuple
|
|
|
|
from pyomo.environ import *
|
|
from pyomo.opt import SolverFactory
|
|
|
|
from workers import Worker
|
|
|
|
import csv
|
|
|
|
from collections import defaultdict
|
|
|
|
ShiftName = str
|
|
DayStr = str
|
|
WeekInt = int
|
|
|
|
days = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]
|
|
|
|
# weeks = [i for i in range(1,weeks_to_rota+1)]
|
|
|
|
# weeks_days_product = list(itertools.product(weeks, days))
|
|
|
|
|
|
class SingleShift(object):
|
|
"""Class to hold all details for a shift"""
|
|
def __init__(
|
|
self,
|
|
sites,
|
|
name,
|
|
length,
|
|
shift_days,
|
|
# balance_by_site=True,
|
|
balance_offset=2, # this could be generated dynamically
|
|
balance_weighting=1,
|
|
workers_required=1,
|
|
rota_on_nwds=False,
|
|
assign_as_block=False,
|
|
force_as_block=False,
|
|
constraints=[],
|
|
):
|
|
self.site = sites
|
|
self.name = name
|
|
self.length = length
|
|
self.shift_days = shift_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)
|
|
self.balance_offset = balance_offset
|
|
|
|
# weight the shift for balancing, default is 1
|
|
self.balance_weighting = balance_weighting
|
|
|
|
self.workers_required = workers_required
|
|
self.rota_on_nwds = rota_on_nwds
|
|
self.assign_as_block = assign_as_block
|
|
self.force_as_block = force_as_block
|
|
self.constraints = constraints
|
|
|
|
self.total_shifts = length * len(days) * len(shift_days)
|
|
|
|
def get_shift_summary(self):
|
|
"""
|
|
Returns a text summary of the shift
|
|
"""
|
|
|
|
return "{}: {} workers for sites ({}) on days ({})".format(
|
|
self.name, self.workers_required, ", ".join(self.site),
|
|
", ".join(self.shift_days))
|
|
|
|
|
|
class RotaBuilder(object):
|
|
"""Class to hold and manipulate shifts"""
|
|
def __init__(self,
|
|
start_date,
|
|
weeks_to_rota=26,
|
|
balance_offset_modifier=1):
|
|
self.shifts = [] # type List[SingleShift]
|
|
|
|
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))
|
|
|
|
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)
|
|
|
|
self.unavailable_to_work = set()
|
|
self.pref_not_to_work = {}
|
|
|
|
self.workers = []
|
|
|
|
self.night_blocks = ["weekday", "weekend", "none"]
|
|
|
|
self.sites = None
|
|
|
|
self.balance_offset_modifier = balance_offset_modifier
|
|
|
|
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, shift) for worker in self.workers
|
|
for week, day in self.get_week_day_combinations()
|
|
for shift in self.get_shift_names()),
|
|
within=Binary,
|
|
initialize=0,
|
|
)
|
|
|
|
# The nights self.model is used to ensure nights are assigned as a block (and limit the number of workers required)
|
|
self.model.nights = Var(
|
|
((worker.id, week, block) for worker in self.workers
|
|
for week in self.weeks for block in self.night_blocks),
|
|
within=Binary,
|
|
initialize=0,
|
|
)
|
|
|
|
# The nights self.model is used to ensure nights are assigned as a block (and limit the number of workers required)
|
|
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_counts = Var(
|
|
((worker.id, week, shift) for week in self.weeks
|
|
for shift in self.shifts_to_assign_or_force_as_blocks()
|
|
for worker in self.workers),
|
|
within=NonNegativeIntegers,
|
|
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,
|
|
)
|
|
|
|
# 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
|
|
self.model.night_per_site = Var(
|
|
((week, block, site) for site in self.sites for week in self.weeks
|
|
for block in ["weekday", "weekend"]),
|
|
#within=Binary,
|
|
within=NonNegativeIntegers,
|
|
initialize=0,
|
|
)
|
|
|
|
self.model.night_per_site_t1 = Var(
|
|
((week, block, site) for site in self.sites for week in self.weeks
|
|
for block in ["weekday", "weekend"]),
|
|
domain=NonNegativeIntegers,
|
|
initialize=0,
|
|
)
|
|
self.model.night_per_site_t2 = Var(
|
|
((week, block, site) for site in self.sites for week in self.weeks
|
|
for block in ["weekday", "weekend"]),
|
|
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,
|
|
)
|
|
|
|
self.model.shift_count_t1 = Var(
|
|
((worker.id) for worker in self.workers),
|
|
domain=NonNegativeReals,
|
|
initialize=0,
|
|
)
|
|
|
|
self.model.shift_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,
|
|
)
|
|
|
|
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, self.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))
|
|
|
|
# # 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))
|
|
|
|
# Ensure no workers are assigned to shifts that are not required
|
|
for week, day, shift in self.get_not_required_shifts():
|
|
self.model.constraints.add(0 == sum(self.model.works[worker.id,
|
|
week, day,
|
|
shift.name]
|
|
for worker in self.workers))
|
|
|
|
# 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
|
|
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 to 1 ST2 (or below) on a night shift
|
|
# def nightShiftMaxSTRule(model, week, block):
|
|
# """Limits to 1 ST2 (or below) per night"""
|
|
# single_workers = [w for w in self.workers if w.grade < 3]
|
|
# if not single_workers:
|
|
# return Constraint.Skip
|
|
# return sum(model.nights[w.id, week, block]
|
|
# for w in single_workers) <= 1
|
|
|
|
# # self.model.night_shifts_max_st_constraint = Constraint(
|
|
# # [week for week in self.weeks],
|
|
# # [block for block in ["weekday", "weekend"]],
|
|
# # rule=nightShiftMaxSTRule,
|
|
# # )
|
|
|
|
# Limit to 1 ST2 (or below) on a night shift
|
|
def nightShiftMaxSTRule_new(model, week, day, shift):
|
|
"""Limits to 1 ST2 (or below) per night"""
|
|
single_workers = [w for w in self.workers if w.grade < 3]
|
|
if not single_workers:
|
|
return Constraint.Skip
|
|
return sum(model.works[w.id, week, day, shift]
|
|
for w in single_workers) <= 1
|
|
|
|
self.model.night_shifts_max_st_constraint = Constraint(
|
|
[week for week in self.weeks],
|
|
[day for day in self.days],
|
|
[shift.name for shift in self.get_shifts_with_constraint("night")],
|
|
rule=nightShiftMaxSTRule_new,
|
|
)
|
|
|
|
# Enusre at least 1 ST4+ on a night shift
|
|
def nightShiftMinST4Rule(model, week, block):
|
|
single_workers = [w for w in self.workers if w.grade > 3]
|
|
if not single_workers:
|
|
return Constraint.Skip
|
|
return sum(model.nights[w.id, week, block]
|
|
for w in single_workers) >= 1
|
|
|
|
self.model.night_shifts_min_st4_constraint = Constraint(
|
|
[week for week in self.weeks],
|
|
[block for block in ["weekday", "weekend"]],
|
|
rule=nightShiftMinST4Rule,
|
|
)
|
|
|
|
# #Constraint (def of self.model.needed)
|
|
# for worker in self.workers:
|
|
# self.model.constraints.add(
|
|
# 10000 * self.model.needed[worker.id] >= sum(
|
|
# self.model.works[worker.id, week, day, shift] for week,
|
|
# day, shift in self.get_all_shiftname_combinations())
|
|
# ) # if any self.model.works[worker, ·, ·] non-zero, self.model.needed[worker] must be one; else is zero to reduce the obj function
|
|
# #10000 is to remark, but 5 was enough since max of 40 hours yields max of 5 shifts, the maximum possible sum
|
|
|
|
# #Constraint (def of self.model.no_pref)
|
|
# for worker in self.workers:
|
|
# for week in weeks:
|
|
# self.model.constraints.add(
|
|
# self.model.no_pref[worker.id] >=
|
|
# sum(self.model.works[worker.id, week, 'Sat', shift]
|
|
# for shift in self.get_shift_names_by_day("Sat")) -
|
|
# sum(self.model.works[worker.id, week, 'Sun', shift]
|
|
# for shift in self.get_shift_names_by_day("Sun"))
|
|
# ) # if not working on sunday but working saturday self.model.needed must be 1; else will be zero to reduce the obj function
|
|
|
|
# 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
|
|
for site in self.sites:
|
|
for block in ("weekend", "weekday"):
|
|
for week in self.weeks:
|
|
self.model.constraints.add(
|
|
self.model.night_per_site[week, block, site] == sum(
|
|
self.model.nights[worker.id, week, block]
|
|
for worker in self.workers if worker.site == site)
|
|
#1 >= sum(self.model.nights[worker.id, week, block] 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)
|
|
|
|
weeks_days = self.get_week_day_combinations()
|
|
|
|
for week in self.weeks:
|
|
for shift_name in self.shifts_to_assign_or_force_as_blocks():
|
|
shift = self.get_shift_by_name(shift_name)
|
|
for worker in self.workers:
|
|
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.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:
|
|
# Force nights to be assigned in blocks
|
|
self.model.constraints.add(
|
|
self.model.worker_block_shift_counts[worker.id, week,
|
|
shift.name] ==
|
|
sum(self.model.works[worker.id, week, day, shift.name]
|
|
for day in self.days))
|
|
|
|
self.model.constraints.add(
|
|
sum(self.model.blocks_worker_shift_assigned[worker.id,
|
|
week,
|
|
shift.name]
|
|
for worker in self.workers) <= 1)
|
|
|
|
# Most of our constraints apply per worker
|
|
for worker in self.workers:
|
|
|
|
# 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 self.get_shifts():
|
|
if (
|
|
worker.site in shift.site
|
|
): # 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.shift_days) *
|
|
shift.workers_required)
|
|
|
|
full_time_equivalent_joined = sum([
|
|
self.full_time_equivalent_sites[i] for i in shift.site
|
|
])
|
|
|
|
target_shifts = (total_shifts /
|
|
full_time_equivalent_joined *
|
|
worker.fte_adj)
|
|
|
|
worker.shift_target_number[shift.name] = target_shifts
|
|
|
|
max_shifts = target_shifts + shift.balance_offset * self.balance_offset_modifier
|
|
min_shifts = target_shifts - shift.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()),
|
|
max_shifts,
|
|
))
|
|
else:
|
|
# Make sure shifts aren't assigned to those from other sites
|
|
self.model.constraints.add(0 == sum(
|
|
self.model.works[worker.id, week, day, shift.name]
|
|
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()))
|
|
|
|
# Define shift_count_t1 and 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)
|
|
self.model.constraints.add(
|
|
self.model.shift_count_t1[worker.id] -
|
|
self.model.shift_count_t2[worker.id] == sum(
|
|
(self.model.shift_count[worker.id, shift.name] -
|
|
worker.shift_target_number[shift.name]) *
|
|
shift.balance_weighting for shift in self.get_shifts()))
|
|
|
|
# Ensure worker is not allocated shifts on non working days
|
|
if worker.nwd:
|
|
for week, day, shift in self.get_all_shiftclass_combinations():
|
|
if not shift.rota_on_nwds and day in worker.nwd:
|
|
self.model.constraints.add(
|
|
0 == self.model.works[worker.id, week, day,
|
|
shift.name])
|
|
|
|
if "night" in self.get_shift_names():
|
|
for week_blocks in self.get_week_block_iterator(2):
|
|
# Prevent nights more than once every n weeks
|
|
self.model.constraints.add(
|
|
1 >= sum(self.model.nights[worker.id, week, block]
|
|
for week in week_blocks
|
|
for block in ["weekday", "weekend"]))
|
|
|
|
for week in self.weeks:
|
|
|
|
# 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")))
|
|
|
|
if "night" in self.get_shift_names():
|
|
# Force nights to be assigned in blocks
|
|
self.model.constraints.add(1 == sum(
|
|
self.model.nights[worker.id, week, block]
|
|
for block in self.night_blocks))
|
|
|
|
# if night block is weekday make sure Mon - Thurs is assigned as nights
|
|
self.model.constraints.add(
|
|
self.model.nights[worker.id, week, "weekday"] *
|
|
4 == sum(self.model.works[worker.id, week, day,
|
|
"night"]
|
|
for day in days[:4]))
|
|
self.model.constraints.add(
|
|
self.model.nights[worker.id, week, "weekend"] *
|
|
3 == sum(self.model.works[worker.id, week, day,
|
|
"night"]
|
|
for day in days[4:]))
|
|
|
|
#
|
|
|
|
for n in range(len(weeks_days)):
|
|
|
|
week, day = weeks_days[n]
|
|
|
|
p1 = 1
|
|
p2 = 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]
|
|
n2 = 0
|
|
|
|
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
|
|
|
|
# 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_day(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_day(day)))
|
|
|
|
# if working a night ensure preceeding (1) or subsequent (2) shifts can only be nights
|
|
|
|
if "night" in self.get_shift_names():
|
|
# 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, "night"])
|
|
|
|
self.model.constraints.add(
|
|
1 >= self.model.works[worker.id, week, day, "night"] +
|
|
sum(n1 *
|
|
self.model.works[worker.id, nweek, nday, shift]
|
|
for shift in self.get_shift_names_by_day(nday)
|
|
if shift != "night"))
|
|
|
|
self.model.constraints.add(
|
|
1 >= self.model.works[worker.id, week, day, "night"] +
|
|
sum(n2 *
|
|
self.model.works[worker.id, n2week, n2day, shift]
|
|
for shift in self.get_shift_names_by_day(n2day)
|
|
if shift != "night"))
|
|
self.model.constraints.add(
|
|
1 >= self.model.works[worker.id, week, day, "night"] +
|
|
sum(p1 *
|
|
self.model.works[worker.id, pweek, pday, shift]
|
|
for shift in self.get_shift_names_by_day(pday)
|
|
if shift != "night"))
|
|
self.model.constraints.add(
|
|
1 >= self.model.works[worker.id, week, day, "night"] +
|
|
sum(p2 *
|
|
self.model.works[worker.id, p2week, p2day, shift]
|
|
for shift in self.get_shift_names_by_day(p2day)
|
|
if shift != "night"))
|
|
|
|
self.define_objectives()
|
|
|
|
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 = 1000
|
|
|
|
shift_balancing = sum(balance_modifier_constant *
|
|
(self.model.shift_count_t1[(worker.id)] +
|
|
self.model.shift_count_t2[(worker.id)])
|
|
for worker in self.workers)
|
|
|
|
preference_constant = 10
|
|
# Preferences
|
|
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())
|
|
|
|
# # Spread nights
|
|
|
|
nights_balancing = sum(
|
|
self.model.night_per_site_t1[week, block, site] * 20
|
|
#self.model.night_per_site2[week, block, site]
|
|
for week in self.weeks for block in ("weekday", "weekend")
|
|
for site in self.sites)
|
|
#nights_balancing = 0
|
|
|
|
blocks_balancing = sum(
|
|
100 * self.model.blocks_assigned[week, shift]
|
|
for week in self.weeks
|
|
for shift in self.shifts_to_assign_as_blocks())
|
|
|
|
return shift_balancing + preferences + nights_balancing + blocks_balancing
|
|
|
|
# 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):
|
|
"""Add a worker to the rota
|
|
|
|
Args:
|
|
worker (Worker):
|
|
"""
|
|
self.workers.append(worker)
|
|
|
|
def add_workers(self, workers: List):
|
|
"""Add multiple worker to the rota
|
|
|
|
Args:
|
|
workers (List(Worker)):
|
|
"""
|
|
self.workers.extend(workers)
|
|
|
|
def build_shifts_and_workers(self):
|
|
"""Process loaded shifts and workers
|
|
|
|
Must be called prior to attempting to solve
|
|
"""
|
|
self.build_shifts()
|
|
|
|
self.workers = sorted(self.workers)
|
|
|
|
self.full_time_equivalent = sum([w.fte_adj for w in self.workers])
|
|
self.full_time_equivalent_sites = {}
|
|
|
|
for site in self.sites:
|
|
self.full_time_equivalent_sites[site] = sum(
|
|
[w.fte_adj for w in self.workers if w.site == site])
|
|
|
|
def add_shift(self, shift):
|
|
"""Add a shift to the collection
|
|
|
|
:param SingleShift shift: Shift object
|
|
|
|
"""
|
|
self.shifts.append(shift)
|
|
|
|
def add_shifts(self, *shifts: SingleShift):
|
|
"""Add multiple shifts
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
self.shifts.extend(shifts)
|
|
|
|
def get_shift_names_by_day(self, day: DayStr):
|
|
"""Returns the shifts required for a specific day
|
|
|
|
Returns:
|
|
set: Set of ShiftName
|
|
|
|
"""
|
|
return self.day_shifts_dict[day]
|
|
|
|
def get_shift_by_name(self, name):
|
|
"""
|
|
|
|
: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):
|
|
""" """
|
|
self.shifts_by_name = {}
|
|
self.shift_names = [] # type: List[ShiftName]
|
|
|
|
self.day_shifts_dict = {day: set() for day in days}
|
|
|
|
self.sites = set()
|
|
|
|
for s in self.shifts:
|
|
self.shifts_by_name[s.name] = s
|
|
self.shift_names.append(s.name)
|
|
|
|
for day in s.shift_days:
|
|
self.day_shifts_dict[day].add(s.name)
|
|
|
|
for site in s.site:
|
|
self.sites.add(site)
|
|
|
|
self.week_day_shift_product = []
|
|
self.week_day_shiftclass_product = []
|
|
for week, day in self.weeks_days_product:
|
|
for s in self.shifts:
|
|
if day in s.shift_days:
|
|
self.week_day_shift_product.append((week, day, s.name))
|
|
self.week_day_shiftclass_product.append((week, day, s))
|
|
|
|
self.day_shift_product = []
|
|
self.day_shiftclass_product = []
|
|
for s in self.shifts:
|
|
for day in days:
|
|
if day in s.shift_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) -> 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
|
|
"""
|
|
return self.week_day_shift_product
|
|
|
|
def get_all_shiftclass_combinations(
|
|
self, ) -> 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
|
|
"""
|
|
return self.week_day_shiftclass_product
|
|
|
|
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_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_with_constraint(self, constraint) -> List[SingleShift]:
|
|
return [
|
|
shift for shift in self.shifts if constraint in shift.constraints
|
|
]
|
|
|
|
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):
|
|
"""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)):
|
|
blocks.append(self.weeks[i:i + block_length])
|
|
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 site)
|
|
"""
|
|
l = []
|
|
for week, day, shift in self.week_day_shiftclass_product:
|
|
if day in shift.shift_days:
|
|
l.append((week, day, shift.name, shift.workers_required,
|
|
shift.site))
|
|
|
|
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):
|
|
return [
|
|
shift.name for shift in self.get_shifts() if shift.assign_as_block
|
|
]
|
|
|
|
def shifts_to_force_as_blocks(self):
|
|
return [
|
|
shift.name for shift in self.get_shifts() if shift.force_as_block
|
|
]
|
|
|
|
def shifts_to_assign_or_force_as_blocks(self):
|
|
s = self.shifts_to_assign_as_blocks()
|
|
s.extend(self.shifts_to_force_as_blocks())
|
|
return s
|
|
|
|
|
|
class RotaResults(object):
|
|
def __init__(self, rota, results):
|
|
self.rota = rota
|
|
|
|
def export_rota_to_csv(self, filename="rota"):
|
|
works = self.rota.model.works
|
|
with open("{}.csv".format(filename), 'w', newline='') as f:
|
|
wr = csv.writer(f, quoting=csv.QUOTE_ALL)
|
|
l = ["Name"]
|
|
l.extend([worker.name for worker in self.rota.workers])
|
|
wr.writerow(l)
|
|
|
|
l2 = ["Site"]
|
|
l2.extend([worker.site for worker in self.rota.workers])
|
|
wr.writerow(l2)
|
|
|
|
l3 = ["Grade"]
|
|
l3.extend([worker.grade for worker in self.rota.workers])
|
|
wr.writerow(l3)
|
|
|
|
l4 = ["FTE"]
|
|
l4.extend([worker.fte for worker in self.rota.workers])
|
|
wr.writerow(l4)
|
|
|
|
for week, day in self.rota.get_week_day_combinations():
|
|
d = ["Week {} Day {}".format(week, day)]
|
|
|
|
for worker in self.rota.workers:
|
|
i = ""
|
|
for shift in self.rota.get_shift_names_by_day(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.rota.model.works
|
|
week_table = {
|
|
week: {
|
|
day: {shift: []
|
|
for shift in self.rota.get_shift_names()}
|
|
for day in days
|
|
}
|
|
for week in self.rota.weeks
|
|
}
|
|
for week in self.rota.weeks:
|
|
for worker in self.rota.workers:
|
|
for day, shift in self.rota.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.rota.model.works
|
|
timetable = {
|
|
worker.name:
|
|
{week: {day: ""
|
|
for day in days}
|
|
for week in self.rota.weeks}
|
|
for worker in self.rota.workers
|
|
}
|
|
for worker in self.rota.workers:
|
|
for week in self.rota.weeks:
|
|
for day, shift in self.rota.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=True, marker_every=19):
|
|
model = self.rota.model
|
|
week_string = "{:20}".format("-Week-") + "".join(
|
|
[7 * str("{}".format(w))[-1:] for w in self.rota.weeks])
|
|
days_string = "{:20}".format("-Day-") + "".join(
|
|
"MTWTFSS" * len(self.rota.weeks))
|
|
timetable = []
|
|
for worker in self.rota.workers:
|
|
shifts = []
|
|
w = ["{:20}".format(worker.name)]
|
|
for week, day in self.rota.get_week_day_combinations():
|
|
a = "-"
|
|
for shift in self.rota.get_shift_names_by_day(day):
|
|
if model.works[worker.id, week, day, shift].value == 1:
|
|
shifts.append(shift)
|
|
a = shift[0]
|
|
w.append(a)
|
|
|
|
shift_count = ""
|
|
for s in set(shifts):
|
|
shift_count = shift_count + "{}: {},".format(
|
|
s, shifts.count(s))
|
|
timetable.append("{} {}".format("".join(w), shift_count))
|
|
|
|
if show_prefs:
|
|
# prefs
|
|
w = ["{:20}".format("Preferences")]
|
|
for week, day in self.rota.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))
|
|
|
|
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_shift_summary(self):
|
|
works = self.rota.model.works
|
|
timetable = {
|
|
worker.get_details():
|
|
{shift[0]: ""
|
|
for shift in self.rota.get_shift_names()}
|
|
for worker in self.rota.workers
|
|
}
|
|
#timetable = { worker.get_details() : { shift : "" for shift in ["truro_twilight"] } for worker in workers }
|
|
t = []
|
|
for worker in self.rota.workers:
|
|
l = []
|
|
|
|
total_shifts = 0
|
|
for shift in self.rota.get_shift_names():
|
|
#for shift in ["truro_twilight"]:
|
|
c = [
|
|
works[worker.id, week, day, shift]
|
|
for week in self.rota.weeks for day in days
|
|
].count(1)
|
|
if c > 0:
|
|
l.append("{} ({})".format(shift, c))
|
|
total_shifts = total_shifts + c
|
|
#print(worker.id, shift)
|
|
timetable[worker.get_details()][shift[0]] = c
|
|
t.append("{} [{}]: {}".format(worker.get_full_details(),
|
|
total_shifts, ", ".join(l)))
|
|
return "\n".join(t)
|
|
|
|
#def get_no_preference(no_pref):
|
|
# """Extract to a list the workers not satisfied with their weekend preference."""
|
|
# return [worker.id for worker in workers if no_pref[worker.id].value == 1]
|
|
|
|
def get_night_blocks(self):
|
|
nights = self.rota.model.nights
|
|
timetable = {
|
|
worker.name: {
|
|
week: {block: ""
|
|
for block in self.rota.night_blocks}
|
|
for week in self.rota.weeks
|
|
}
|
|
for worker in self.rota.workers
|
|
}
|
|
for worker in self.rota.workers:
|
|
for week in self.rota.weeks:
|
|
for block in self.rota.night_blocks:
|
|
if nights[worker.id, week, block].value == 1:
|
|
timetable[worker.name][week][block] = "true"
|
|
return timetable
|
|
|
|
def get_multinight(self):
|
|
for site in self.rota.sites:
|
|
for block in ("weekend", "weekday"):
|
|
for week in self.rota.weeks:
|
|
print(
|
|
site, week, block,
|
|
self.rota.model.night_per_site[week, block,
|
|
site].value,
|
|
self.rota.model.night_per_site_t1[week, block,
|
|
site].value,
|
|
self.rota.model.night_per_site_t2[week, block,
|
|
site].value)
|