Databases
StackOverflow2013
Postsdbo
POWorker/Timeslot permutation/constraint filtering algorithm
 

Note that there are some explanatory texts on larger screens.

plurals
  1. POWorker/Timeslot permutation/constraint filtering algorithm
    primarykey
    data
    text
    <p>Hope you can help me out with this guys. It's not help with work -- it's for a charity of very hard working volunteers, who could really use a less confusing/annoying timetable system than what they currently have.</p> <p>If anyone knows of a good third-party app which (certainly) automate this, that would almost as good. Just... please don't suggest random timetabling stuff such as the ones for booking classrooms, as I don't think they can do this.</p> <p>Thanks in advance for reading; I know it's a big post. I'm trying to do my best to document this clearly though, and to show that I've made efforts on my own.</p> <h2>Problem</h2> <p>I need a worker/timeslot scheduling algorithm which generates shifts for workers, which meets the following criteria:</p> <p><strong>Input Data</strong></p> <pre><code>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) # allows 3, requires 2, but only 2 available shift_2 = Shift(shift_2_times, 2,2) # allows 2 shift_3 = Shift(shift_3_times, 2,3) # allows 3, requires 2, 3 available shifts = ( shift_1, shift_2, shift_3 ) joe_avail = [ shift_1, shift_2 ] bob_avail = [ shift_1, shift_3 ] sam_avail = [ shift_2 ] amy_avail = [ shift_2 ] ned_avail = [ shift_2, shift_3 ] max_avail = [ shift_3 ] jim_avail = [ shift_3 ] joe = Worker('joe', joe_avail) bob = Worker('bob', bob_avail) sam = Worker('sam', sam_avail) ned = Worker('ned', ned_avail) max = Worker('max', max_avail) amy = Worker('amy', amy_avail) jim = Worker('jim', jim_avail) workers = ( joe, bob, sam, ned, max, amy, jim ) </code></pre> <h2>Processing</h2> <p>From above, <em>shifts</em> and <em>workers</em> are the two main input variables to process</p> <p>Each shift has a minimum and maximum number of workers needed. Filling the minimum requirements for a shift is crucial to success, but if all else fails, a rota with gaps to be filled manually is better than "error" :) The main algorithmic issue is that there shouldn't be unnecessary gaps, when enough workers are available.</p> <p>Ideally, the maximum number of workers for a shift would be filled, but this is the lowest priority relative to other constraints, so if anything has to give, it should be this.</p> <p><strong>Flexible constraints</strong></p> <p>These are a little flexible, and their boundaries can be pushed a little if a "perfect" solution can't be found. This flexibility should be a last resort though, rather than being exploited randomly. Ideally, the flexibility would be configurable with a "fudge_factor" variable, or similar.</p> <ul> <li>There is a minimum time period between two shifts. So, a worker shouldn't be scheduled for two shifts in the same day, for instance.</li> <li>There are a maximum number of shifts a worker can do in a given time period (say, a month)</li> <li>There are a maximum number of certain shifts that can be done in a month (say, overnight shifts)</li> </ul> <p><strong>Nice to have, but not necessary</strong></p> <p>If you can come up with an algorithm which does the above and includes any/all of these, I'll be seriously impressed and grateful. Even an add-on script to do these bits separately would be great too.</p> <ul> <li><p>Overlapping shifts. For instance, it would be good to be able to specify a "front desk" shift and a "back office" shift that both occur at the same time. This could be done with separate invocations of the program with different shift data, except that the constraints about scheduling people for multiple shifts in a given time period would be missed.</p></li> <li><p>Minimum reschedule time period for workers specifiable on a per-worker (rather than global) basis. For instance, if Joe is feeling overworked or is dealing with personal issues, or is a beginner learning the ropes, we might want to schedule him less often than other workers.</p></li> <li><p>Some automated/random/fair way of selecting staff to fill minimum shift numbers when no available workers fit.</p></li> <li><p>Some way of handling sudden cancellations, and just filling the gaps without rearranging other shifts.</p></li> </ul> <h2>Output Test</h2> <p>Probably, the algorithm should generate as many matching Solutions as possible, where each Solution looks like this:</p> <pre><code>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 </code></pre> <p>Here's a test function for an individual solution, given the above data. I <em>think</em> this is right, but I'd appreciate some peer review on it too.</p> <pre><code>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 &gt;= shift.min_workers assert num_workers &lt;= shift.max_workers for w in workers_allowed: assert w in workers shifts_workers = solution.shifts_workers # all shifts should be covered assert len(shifts_workers.keys()) == 3 assert shift1 in shifts_workers.keys() assert shift2 in shifts_workers.keys() assert shift3 in shifts_workers.keys() # shift_1 should be covered by 2 people - joe, and bob shift_check(shift_1, shifts_workers[shift_1], (joe, bob)) # shift_2 should be covered by 2 people - sam and amy shift_check(shift_2, shifts_workers[shift_2], (sam, amy)) # shift_3 should be covered by 3 people - ned, max, and jim shift_check(shift_3, shifts_workers[shift_3], (ned,max,jim)) </code></pre> <h2>Attempts</h2> <p>I've tried implementing this with a Genetic Algorithm, but can't seem to get it tuned quite right, so although the basic principle seems to work on single shifts, it can't solve even easy cases with a few shifts and a few workers.</p> <p>My latest attempt is to generate every possible permutation as a solution, then whittle down the permutations that don't meet the constraints. This seems to work much more quickly, and has gotten me further, but I'm using python 2.6's itertools.product() to help generate the permutations, and I can't quite get it right. It wouldn't surprise me if there are many bugs as, honestly, the problem doesn't fit in my head that well :)</p> <p>Currently my code for this is in two files: models.py and rota.py. models.py looks like:</p> <pre><code># -*- 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 "&lt;Shift %s--%s (%r&lt;x&lt;%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 "&lt;Duty worker=%r shift=%r slot=%d&gt;" % (self.worker, self.shift, self.slot) def dump(self, indent=4, depth=1): ind = " " * (indent * depth) print ind + "&lt;Duty shift=%s slot=%s" % (self.shift, self.slot) self.worker.dump(indent=indent, depth=depth+1) print ind + "&gt;" class Avail: def __init__(self, start_time, end_time): self.start = start_time self.end = end_time def __repr__(self): return "&lt;%s to %s&gt;" % (self.start, self.end) class Worker: def __init__(self, name, availabilities): self.name = name self.availabilities = availabilities def __repr__(self): return "&lt;Worker %s Avail=%r&gt;" % (self.name, self.availabilities) def dump(self, indent=4, depth=1): ind = " " * (indent * depth) print ind + "&lt;Worker %s" % self.name for avail in self.availabilities: print ind + " " * indent + repr(avail) print ind + "&gt;" def available_for_shift(self, shift): for a in self.availabilities: if shift.start &gt;= a.start and shift.end &lt;= 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 "&lt;Solution: shifts=%r&gt;" % 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 + "&lt;ShiftList" for duty in s: duty.dump(indent=indent, depth=depth+1) print ind + "&gt;" def dump(self, indent=4, depth=1): ind = " " * (indent * depth) print ind + "&lt;Solution" for s in self._shifts: self.dump_shift(s, indent=indent, depth=depth+1) print ind + "&gt;" 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 </code></pre> <p>and rota.py looks like:</p> <pre><code>#!/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 &gt;= one_day_before and d.shift.start &lt;= 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 &gt; 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 &lt; 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() </code></pre> <p>Snipping the debugging output before the result, this currently gives:</p> <pre><code>Solved. Solutions found: &lt;Solution &lt;ShiftList &lt;Duty shift=&lt;Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2&lt;x&lt;3) slot=1 &lt;Worker joe &lt;2009-10-01 02:00:00 to 2009-10-01 08:00:00&gt; &gt; &gt; &lt;Duty shift=&lt;Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2&lt;x&lt;3) slot=1 &lt;Worker sam &lt;2009-10-01 02:00:00 to 2009-10-01 08:00:00&gt; &gt; &gt; &lt;Duty shift=&lt;Shift 2009-10-01 02:00:00--2009-10-01 08:00:00 (2&lt;x&lt;3) slot=1 &lt;Worker ned &lt;2009-10-01 02:00:00 to 2009-10-01 08:00:00&gt; &gt; &gt; &gt; &lt;ShiftList &lt;Duty shift=&lt;Shift 2009-10-01 08:00:00--2009-10-01 12:00:00 (2&lt;x&lt;3) slot=1 &lt;Worker bob &lt;2009-10-01 08:00:00 to 2009-10-01 12:00:00&gt; &gt; &gt; &lt;Duty shift=&lt;Shift 2009-10-01 08:00:00--2009-10-01 12:00:00 (2&lt;x&lt;3) slot=1 &lt;Worker max &lt;2009-10-01 08:00:00 to 2009-10-01 12:00:00&gt; &gt; &gt; &gt; &gt; </code></pre>
    singulars
    1. PO
      singulars
      1. PTAnswer
    1. This table or related slice is empty.
    1. PTQuestion
    1. USLee B
    1. USLee B
    plurals
    1. This table or related slice is empty.
    1. This table or related slice is empty.
    1. This table or related slice is empty.
    1. PO
      singulars
      1. PTAnswer
    2. PO
      singulars
      1. PTAnswer
    3. PO
      singulars
      1. PTAnswer
    1. VO
      singulars
      1. POWorker/Timeslot permutation/constraint filtering algorithm
      1. This table or related slice is empty.
      1. VTUpMod
    2. VO
      singulars
      1. POWorker/Timeslot permutation/constraint filtering algorithm
      1. This table or related slice is empty.
      1. VTUpMod
    3. VO
      singulars
      1. POWorker/Timeslot permutation/constraint filtering algorithm
      1. This table or related slice is empty.
      1. VTUpMod
    1. COHi Lee, have you found a working solution to this yet?
      singulars
      1. POWorker/Timeslot permutation/constraint filtering algorithm
      1. UShofnarwillie
 

Querying!

 
Guidance
A row detail

Detail views are divided into sections. All the information in the data section comes from columns in the selected row. The other sections display data from other, related rows.

Related data can be related in a to-one or a to-many fashion. Captions of data related in a to-many fashion link to a list view showing a filtered view of the table.

Try moving around until you find a non-empty to-many entry and click on the label to get to one. You can move back to the root by clicking on the database name in the header.

SQuiL has stopped working due to an internal error.

If you are curious you may find further information in the browser console, which is accessible through the devtools (F12).

Reload