I hope you can help me with these guys. This does not help in the work - it is for the charity of very hardworking volunteers who really can use a less confusing / annoying schedule than what they currently have.
If someone knows about a good third-party application that (of course) automates this, it will be almost as good. Just ... please do not offer random schedules, for example, for booking classrooms, as I do not think they can do this.
Thanks in advance for reading; I know this is a great post. I try my best to fix it clearly, and to show that I made an effort on my own.
Problem
I need a work / time schedule scheduling algorithm that generates shifts for workers that meet the following criteria:
Input data
import datetime.datetime as dt class DateRange: def __init__(self, start, end): self.start = start self.end = end class Shift: def __init__(self, range, min, max): self.range = range self.min_workers = min self.max_workers = max tue_9th_10pm = dt(2009, 1, 9, 22, 0) wed_10th_4am = dt(2009, 1, 10, 4, 0) wed_10th_10am = dt(2009, 1, 10, 10, 0) shift_1_times = Range(tue_9th_10pm, wed_10th_4am) shift_2_times = Range(wed_10th_4am, wed_10th_10am) shift_3_times = Range(wed_10th_10am, wed_10th_2pm) shift_1 = Shift(shift_1_times, 2,3)
Treatment
Shifts and workers above are the two main input variables for processing.
Each shift has a minimum and maximum number of workers. Filling the minimum shift requirements is critical to success, but if all else fails, then a company with spaces that must be filled in manually is better than an “error” :). The main problem of the algorithm is that there should be no unnecessary spaces when enough workers are available.
Ideally, the maximum number of workers for the shift will be filled, but this is the lowest priority compared to other restrictions, so if something should give, it should be so.
Flexible restrictions
They are a little flexible, and their boundaries can be pushed a little if the “perfect” solution cannot be found. However, this flexibility should be the last resort, and not randomly used. Ideally, flexibility would be adjusted using the fudge_factor variable or similar.
- There is a minimum period of time between two shifts. So, a worker should not be scheduled for two shifts on the same day, for example.
- The employee can make the maximum number of shifts for a certain period of time (say, a month).
- The maximum number of defined shifts that can be made per month (say, night shifts)
Nice to have, but not necessarily
If you can come up with an algorithm that does the above and includes all / all of them, I will be seriously impressed and grateful. Even a script add-in to do these bits separately would also be great.
Overlapping shifts. For example, it would be nice to indicate a shift of the “reception desk” and “back office,” a shift that occurs simultaneously. This can be done using separate program calls with different shift data, except that restrictions on planning people for several shifts at a given time period will be skipped.
Minimum redistribution time period for workers based on workers (rather than global). For example, if Joe feels overwhelmed or is dealing with personal problems, or a beginner learning the ropes, we can plan it less often than other workers.
Some automated / random / honest ways to select personnel to fill the minimum shift number when there are no workers available.
Some ways to handle sudden deviations and just fill in the gaps without rearranging other shifts.
Exit test
Probably, the algorithm should generate as many suitable solutions as possible, where each solution looks like this:
class Solution: def __init__(self, shifts_workers): """shifts_workers -- a dictionary of shift objects as keys, and a a lists of workers filling the shift as values.""" assert isinstance(dict, shifts_workers) self.shifts_workers = shifts_workers
Here's a test function for a separate solution, given the above data. I think this is correct, but I would appreciate that he, too, deserves approval.
def check_solution(solution): assert isinstance(Solution, solution) def shift_check(shift, workers, workers_allowed): assert isinstance(Shift, shift): assert isinstance(list, workers): assert isinstance(list, workers_allowed) num_workers = len(workers) assert num_workers >= shift.min_workers assert num_workers <= shift.max_workers for w in workers_allowed: assert w in workers shifts_workers = solution.shifts_workers
Attempts
I tried to implement this using the genetic algorithm, but it seems that I didn’t set it up correctly, therefore, although the basic principle seems to work in single shifts, it cannot solve even simple cases with several shifts and several workers.
My last attempt is to generate every possible permutation as a solution, and then squeeze the permutations that do not meet the constraints. This seems to be much faster, and I got even more, but I use python 2.6 itertools.product () to help generate the permutations, and I cannot figure it out correctly. I would not be surprised if there were a lot of mistakes, as, frankly, the problem does not fit my head :)
Currently, my code for this file consists of two files: models.py and rota.py. models.py looks like this:
# -*- coding: utf-8 -*- class Shift: def __init__(self, start_datetime, end_datetime, min_coverage, max_coverage): self.start = start_datetime self.end = end_datetime self.duration = self.end - self.start self.min_coverage = min_coverage self.max_coverage = max_coverage def __repr__(self): return "<Shift %s--%s (%r<x<%r)" % (self.start, self.end, self.min_coverage, self.max_coverage) class Duty: def __init__(self, worker, shift, slot): self.worker = worker self.shift = shift self.slot = slot def __repr__(self): return "<Duty worker=%r shift=%r slot=%d>" % (self.worker, self.shift, self.slot) def dump(self, indent=4, depth=1): ind = " " * (indent * depth) print ind + "<Duty shift=%s slot=%s" % (self.shift, self.slot) self.worker.dump(indent=indent, depth=depth+1) print ind + ">" class Avail: def __init__(self, start_time, end_time): self.start = start_time self.end = end_time def __repr__(self): return "<%s to %s>" % (self.start, self.end) class Worker: def __init__(self, name, availabilities): self.name = name self.availabilities = availabilities def __repr__(self): return "<Worker %s Avail=%r>" % (self.name, self.availabilities) def dump(self, indent=4, depth=1): ind = " " * (indent * depth) print ind + "<Worker %s" % self.name for avail in self.availabilities: print ind + " " * indent + repr(avail) print ind + ">" def available_for_shift(self, shift): for a in self.availabilities: if shift.start >= a.start and shift.end <= a.end: return True print "Worker %s not available for %r (Availability: %r)" % (self.name, shift, self.availabilities) return False class Solution: def __init__(self, shifts): self._shifts = list(shifts) def __repr__(self): return "<Solution: shifts=%r>" % self._shifts def duties(self): d = [] for s in self._shifts: for x in s: yield x def shifts(self): return list(set([ d.shift for d in self.duties() ])) def dump_shift(self, s, indent=4, depth=1): ind = " " * (indent * depth) print ind + "<ShiftList" for duty in s: duty.dump(indent=indent, depth=depth+1) print ind + ">" def dump(self, indent=4, depth=1): ind = " " * (indent * depth) print ind + "<Solution" for s in self._shifts: self.dump_shift(s, indent=indent, depth=depth+1) print ind + ">" class Env: def __init__(self, shifts, workers): self.shifts = shifts self.workers = workers self.fittest = None self.generation = 0 class DisplayContext: def __init__(self, env): self.env = env def status(self, msg, *args): raise NotImplementedError() def cleanup(self): pass def update(self): pass
and rota.py looks like this:
#!/usr/bin/env python2.6 # -*- coding: utf-8 -*- from datetime import datetime as dt am2 = dt(2009, 10, 1, 2, 0) am8 = dt(2009, 10, 1, 8, 0) pm12 = dt(2009, 10, 1, 12, 0) def duties_for_all_workers(shifts, workers): from models import Duty duties = [] # for all shifts for shift in shifts: # for all slots for cov in range(shift.min_coverage, shift.max_coverage): for slot in range(cov): # for all workers for worker in workers: # generate a duty duty = Duty(worker, shift, slot+1) duties.append(duty) return duties def filter_duties_for_shift(duties, shift): matching_duties = [ d for d in duties if d.shift == shift ] for m in matching_duties: yield m def duty_permutations(shifts, duties): from itertools import product # build a list of shifts shift_perms = [] for shift in shifts: shift_duty_perms = [] for slot in range(shift.max_coverage): slot_duties = [ d for d in duties if d.shift == shift and d.slot == (slot+1) ] shift_duty_perms.append(slot_duties) shift_perms.append(shift_duty_perms) all_perms = ( shift_perms, shift_duty_perms ) # generate all possible duties for all shifts perms = list(product(*shift_perms)) return perms def solutions_for_duty_permutations(permutations): from models import Solution res = [] for duties in permutations: sol = Solution(duties) res.append(sol) return res def find_clashing_duties(duty, duties): """Find duties for the same worker that are too close together""" from datetime import timedelta one_day = timedelta(days=1) one_day_before = duty.shift.start - one_day one_day_after = duty.shift.end + one_day for d in [ ds for ds in duties if ds.worker == duty.worker ]: # skip the duty we're considering, as it can't clash with itself if duty == d: continue clashes = False # check if dates are too close to another shift if d.shift.start >= one_day_before and d.shift.start <= one_day_after: clashes = True # check if slots collide with another shift if d.slot == duty.slot: clashes = True if clashes: yield d def filter_unwanted_shifts(solutions): from models import Solution print "possibly unwanted:", solutions new_solutions = [] new_duties = [] for sol in solutions: for duty in sol.duties(): duty_ok = True if not duty.worker.available_for_shift(duty.shift): duty_ok = False if duty_ok: print "duty OK:" duty.dump(depth=1) new_duties.append(duty) else: print "duty **NOT** OK:" duty.dump(depth=1) shifts = set([ d.shift for d in new_duties ]) shift_lists = [] for s in shifts: shift_duties = [ d for d in new_duties if d.shift == s ] shift_lists.append(shift_duties) new_solutions.append(Solution(shift_lists)) return new_solutions def filter_clashing_duties(solutions): new_solutions = [] for sol in solutions: solution_ok = True for duty in sol.duties(): num_clashing_duties = len(set(find_clashing_duties(duty, sol.duties()))) # check if many duties collide with this one (and thus we should delete this one if num_clashing_duties > 0: solution_ok = False break if solution_ok: new_solutions.append(sol) return new_solutions def filter_incomplete_shifts(solutions): new_solutions = [] shift_duty_count = {} for sol in solutions: solution_ok = True for shift in set([ duty.shift for duty in sol.duties() ]): shift_duties = [ d for d in sol.duties() if d.shift == shift ] num_workers = len(set([ d.worker for d in shift_duties ])) if num_workers < shift.min_coverage: solution_ok = False if solution_ok: new_solutions.append(sol) return new_solutions def filter_solutions(solutions, workers): # filter permutations ############################ # for each solution solutions = filter_unwanted_shifts(solutions) solutions = filter_clashing_duties(solutions) solutions = filter_incomplete_shifts(solutions) return solutions def prioritise_solutions(solutions): # TODO: not implemented! return solutions # prioritise solutions ############################ # for all solutions # score according to number of staff on a duty # score according to male/female staff # score according to skill/background diversity # score according to when staff last on shift # sort all solutions by score def solve_duties(shifts, duties, workers): # ramify all possible duties ######################### perms = duty_permutations(shifts, duties) solutions = solutions_for_duty_permutations(perms) solutions = filter_solutions(solutions, workers) solutions = prioritise_solutions(solutions) return solutions def load_shifts(): from models import Shift shifts = [ Shift(am2, am8, 2, 3), Shift(am8, pm12, 2, 3), ] return shifts def load_workers(): from models import Avail, Worker joe_avail = ( Avail(am2, am8), ) sam_avail = ( Avail(am2, am8), ) ned_avail = ( Avail(am2, am8), ) bob_avail = ( Avail(am8, pm12), ) max_avail = ( Avail(am8, pm12), ) joe = Worker("joe", joe_avail) sam = Worker("sam", sam_avail) ned = Worker("ned", sam_avail) bob = Worker("bob", bob_avail) max = Worker("max", max_avail) return (joe, sam, ned, bob, max) def main(): import sys shifts = load_shifts() workers = load_workers() duties = duties_for_all_workers(shifts, workers) solutions = solve_duties(shifts, duties, workers) if len(solutions) == 0: print "Sorry, can't solve this. Perhaps you need more staff available, or" print "simpler duty constraints?" sys.exit(20) else: print "Solved. Solutions found:" for sol in solutions: sol.dump() if __name__ == "__main__": main()
Discarding debug output before the result, this currently gives:
Solved. Solutions found: <Solution <ShiftList <Duty shift=<Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2<x<3) slot=1 <Worker joe <2009-10-01 02:00:00 to 2009-10-01 08:00:00> > > <Duty shift=<Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2<x<3) slot=1 <Worker sam <2009-10-01 02:00:00 to 2009-10-01 08:00:00> > > <Duty shift=<Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2<x<3) slot=1 <Worker ned <2009-10-01 02:00:00 to 2009-10-01 08:00:00> > > > <ShiftList <Duty shift=<Shift 2009-10-01 08:00:00--2009-10-01 12:00:00 (2<x<3) slot=1 <Worker bob <2009-10-01 08:00:00 to 2009-10-01 12:00:00> > > <Duty shift=<Shift 2009-10-01 08:00:00--2009-10-01 12:00:00 (2<x<3) slot=1 <Worker max <2009-10-01 08:00:00 to 2009-10-01 12:00:00> > > > >