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, hard_constrain_shift=True, 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.hard_constrain_shift = hard_constrain_shift 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, ltft_balance_offset=4, max_night_frequency=2, max_weekend_frequency=2): 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 self.ltft_balance_offset = ltft_balance_offset # Don't assign multiple shifts every (n) weeks self.max_night_frequency = max_night_frequency self.max_weekend_frequency = max_weekend_frequency self.constraint_options = { "limit_to_1_st1_on_nights": True, "ensure_1_st4_plus_on_nights": True, "balance_nights": True, "constrain_time_off_after_nights": True, "balance_nights_across_sites": True, "balance_blocks": True, "balance_shifts": True, "balance_weekends": True, "prevent_monday_after_full_weekends": False, "prevent_fridays_before_full_weekends": True, } 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 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, ) self.model.shift_count = Var( ((worker.id, shift) for worker in self.workers for shift in self.get_shift_names()), domain=NonNegativeReals, initialize=0, ) if self.constraint_options["balance_shifts"]: 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, ) 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, ) 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, # ) if self.constraint_options["balance_weekends"]: 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, ) 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 #print(self.unavailable_to_work) 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, 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.shift_week_worker_assigned[shift, week, w.id] for w in single_workers) <= 1 if self.constraint_options["limit_to_1_st1_on_nights"]: self.model.night_shifts_max_st_constraint = Constraint( [week for week in self.weeks], [ shift.name for shift in self.get_shifts_with_constraint("night") ], rule=nightShiftMaxSTRule, ) def nightShiftMinST4Rule(model, week, shift): single_workers = [w for w in self.workers if w.grade >= 4] if not single_workers: print(single_workers) return Constraint.Skip return sum(model.shift_week_worker_assigned[shift, week, w.id] for w in single_workers) >= 1 if self.constraint_options["ensure_1_st4_plus_on_nights"]: self.model.night_shifts_min_st4_constraint = Constraint( [week for week in self.weeks], [ shift.name for shift in self.get_shifts_with_constraint("night") ], rule=nightShiftMinST4Rule, ) # # 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) 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: 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: # 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)) # 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 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()), 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())) 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 in self.get_week_day_combinations() for shift in self.get_shifts_with_constraint("night"))) # 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) if self.constraint_options["balance_shifts"]: 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())) #if "night" not in shift.constraints)) if self.constraint_options["balance_nights"]: night_shift_target_number = sum( worker.shift_target_number[shift.name] for shift in self.get_shifts_with_constraint("night")) # min_shifts = night_shift_target_number - 20 # max_shifts = night_shift_target_number + 20 # self.model.constraints.add( # inequality( # min_shifts, # self.model.night_shift_count[worker.id], # max_shifts, # )) 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 # I'm not entirely sure how split shifts will affect this if self.constraint_options["balance_weekends"]: weekend_shift_target_number = sum( worker.shift_target_number[shift.name] for shift in self.get_shifts() if not set(("Sat", "Sun")).isdisjoint(shift.shift_days)) # min_shifts = weekend_shift_target_number - 20 # max_shifts = weekend_shift_target_number + 20 # self.model.constraints.add( # inequality( # min_shifts, # self.model.worker_weekend_count[worker.id], # max_shifts, # )) 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 = 10 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.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 self.constraint_options["balance_blocks"]: for week_blocks in self.get_week_block_iterator( self.max_night_frequency): 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"))) for week_blocks in self.get_week_block_iterator( self.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 week in self.weeks: if self.constraint_options[ "prevent_monday_after_full_weekends"]: # Ignore last week if week + 1 in self.weeks: self.model.constraints.add( 2 >= sum(self.model.works[worker.id, week, "Sat", shift.name] for shift in self.get_shifts()) + sum(self.model.works[worker.id, week, "Sun", shift.name] for shift in self.get_shifts()) + sum(self.model.works[worker.id, week + 1, "Mon", shift.name] for shift in self.get_shifts())) if self.constraint_options[ "prevent_fridays_before_full_weekends"]: self.model.constraints.add( 2 >= sum(self.model.works[worker.id, week, "Sat", shift.name] for shift in self.get_shifts() if shift.name not in ("night_weekend")) + sum(self.model.works[worker.id, week, "Sun", shift.name] for shift in self.get_shifts() if shift.name not in ("night_weekend")) + sum(self.model.works[worker.id, week, "Fri", shift.name] for shift in self.get_shifts() if shift.name not in ("night_weekend"))) # # 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, shift.name] for day in self.days[5:] for shift in self.get_shifts()) / 2) self.model.constraints.add( self.model.works_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())) # 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("night"): # 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.shift_days) == sum( self.model.works[worker.id, week, day, shift.name] for day in shift.shift_days)) # 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] p2 = 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 try: n3week, n3day = weeks_days[n + 3] except IndexError: n3week, n3day = weeks_days[n] n3 = 0 if self.get_shift_names_by_day(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_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 self.constraint_options["constrain_time_off_after_nights"]: for constraint_shift in self.get_shifts_with_constraint( "night"): # 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]) self.model.constraints.add( 1 >= self.model.works[worker.id, week, day, constraint_shift.name] + sum(n1 * self.model.works[worker.id, nweek, nday, shift] for shift in self.get_shift_names_by_day(nday) if shift != constraint_shift.name)) self.model.constraints.add( 1 >= self.model.works[worker.id, week, day, constraint_shift.name] + sum(n2 * self.model.works[worker.id, n2week, n2day, shift] for shift in self.get_shift_names_by_day(n2day) if shift != constraint_shift.name)) # self.model.constraints.add( # 1 >= self.model.works[worker.id, week, day, # constraint_shift.name] + # sum(n3 * self.model.works[worker.id, n3week, n3day, # shift] # for shift in self.get_shift_names_by_day(n3day) # if shift != constraint_shift.name)) self.model.constraints.add( 1 >= self.model.works[worker.id, week, day, constraint_shift.name] + sum(p1 * self.model.works[worker.id, pweek, pday, shift] for shift in self.get_shift_names_by_day(pday) if shift != constraint_shift.name)) self.model.constraints.add( 1 >= self.model.works[worker.id, week, day, constraint_shift.name] + sum(p2 * self.model.works[worker.id, p2week, p2day, shift] for shift in self.get_shift_names_by_day(p2day) if shift != constraint_shift.name)) 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 = 1 if self.constraint_options["balance_shifts"]: 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) else: shift_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_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 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 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.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 #return weekend_shift_balancing + blocks_balancing #return shift_balancing + preferences + blocks_balancing #return shift_balancing + preferences + nights_site_balancing + blocks_balancing return shift_balancing + night_shift_balancing + preferences + nights_site_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 def get_workers_total_fte(self): return self.full_time_equivalent def get_worker_details(self): w = defaultdict(list) for worker in self.workers: w[worker.site].append(worker) l = [] for site in w: l.append("{}: {}".format( site, sum([worker.fte_adj for worker in w[site]]) / 100)) t = "\n".join(l) return "Full time equivalent trainees by site:\n{}".format(t) 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=False, marker_every=30, show_unavailable=False): 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)) shift_count = shift_count + "#weekends_worked: {}\\#".format( model.worker_weekend_count[worker.id].value) 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)) if show_unavailable: # prefs w = ["{:20}".format("Unavailable")] for week, day in self.rota.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="Test"): model = self.rota.model timetable = [] date_row = [""] n = 0 for week, day in self.rota.get_week_day_combinations(): d = self.rota.start_date + datetime.timedelta(n) date_row.append("Week {}: {}".format( d, week, day)) n = n + 1 timetable.append("{}".format( "".join(date_row))) for worker in self.rota.workers: shifts = [] if worker.nwd is not None: nwds = ", ".join(worker.nwd) else: nwds = None w = [ "{} ({}) [{}]" .format(worker.site, worker.site, nwds, worker.site, worker.name, worker.fte, worker.fte_adj, worker.end_date, worker.name, worker.grade, worker.fte) ] n = 0 for week, day in self.rota.get_week_day_combinations(): d = self.rota.start_date + datetime.timedelta(n) n = n + 1 a = "-" shift_name = "" 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] shift_name = shift break css_class = "" if model.available[worker.id, week, day] > 0: available = True else: available = False css_class = "unavailable" if worker.name == "Delilah Trimmer": pass w.append( "{}" .format(shift_name, d, css_class, shift_name, available, d, week, day, a)) shift_count = "" for s in set(shifts): shift_count = shift_count + "{}: {}, ".format( s, shifts.count(s)) shift_count = shift_count + "#weekends_worked: {}\\#".format( model.worker_weekend_count[worker.id].value) timetable.append("{}".format( "".join(w))) #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)) # if show_unavailable: # # prefs # w = ["{:20}".format("Unavailable")] # for week, day in self.rota.get_week_day_combinations(): # if model.available[worker.id, week, day] > 0: # w.append("A") # else: # w.append("U") # timetable.append("".join(w)) html = """
{}
""".format(table_name, "\n".join(timetable)) if include_html_tag: html = """ {}""".format(html) return html 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_shift_summary_html(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 shift in self.get_shifts_with_constraint("night"): block = shift.name 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) def get_night_details(self): for worker in self.rota.workers: print( "{:20}".format(worker.name), "worked: {},".format( self.rota.model.night_shift_count[(worker.id)].value), "target: {},".format( worker.shift_target_number["night_weekday"] + worker.shift_target_number["night_weekend"]), "target_diff: {},".format( self.rota.model.night_shift_count_t1[(worker.id)].value + self.rota.model.night_shift_count_t2[(worker.id)].value + 1), "balance: {},".format( self.rota.model.night_shift_count_w[worker.id].value), ) def get_weekend_details(self): for worker in self.rota.workers: print( "{:20}".format(worker.name), "worked: {},".format( self.rota.model.worker_weekend_count[(worker.id)].value), "target: {},".format( sum([ worker.shift_target_number[shift.name] for shift in self.rota.get_shifts() if not set(("Sat", "Sun")).isdisjoint(shift.shift_days) ])), "target_diff: {},".format( self.rota.model.weekend_shift_count_t1[(worker.id)].value + self.rota.model.weekend_shift_count_t2[(worker.id)].value + 1), "balance: {},".format( self.rota.model.weekend_shift_count_w[worker.id].value), )