工作/时隙置换/约束过滤算法

Question

希望你能帮我解决这些问题。这对工作没有帮助——这是为一个非常努力工作的志愿者的慈善机构，他们真的可以使用一个比他们目前拥有的更少令人困惑/烦人的时间表系统。在

如果有人知道一个好的第三方应用程序，它（当然）可以自动化这一点，那也差不多。只是。。。请不要建议随意安排时间表，比如预定教室的时间表，因为我认为他们做不到。在

提前感谢你的阅读，我知道这是一个很大的帖子。不过，我正尽力把这件事记录清楚，并表明我自己已经做出了努力。在

问题

我需要一个工人/时间段调度算法，它可以为工人生成轮班，它满足以下条件：

输入数据

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 )

加工

综上所述，位移和工人是处理的两个主要输入变量

每个班次都有最少和最多需要的工人人数。满足轮班的最低要求是成功的关键，但如果所有其他方法都失败了，用手工填补空缺的轮班表比“错误”要好：主要的算法问题是，当有足够的工人时，不应该有不必要的空缺。在

理想情况下，一个班次的最大工人数量将被填补，但相对于其他限制，这是最低优先级，因此，如果有任何事情必须给予，它应该是这样的。在

灵活约束

这些方法有点灵活，如果找不到“完美”的解决方案，它们的边界可能会被推得稍微大一点。不过，这种灵活性应该是最后的手段，而不是被随意利用。理想情况下，灵活性可以通过“fudge_factor”变量或类似变量进行配置。在

• 有一个最短的时间段 两个班次之间。所以，一个工人 不应该安排两班倒 比如在同一天。在
• 有一个最大轮班数a 工人可以在给定的时间段内完成 （比如说，一个月）
• 有一个最大数量的 可以在一个月内完成的轮班 （比如说，夜班）

很高兴有，但不是必须的

如果你能想出一个算法来实现上述功能并包含其中的任何一个，我将非常感激。即使是一个单独完成这些工作的附加脚本也会很棒。在

• 轮班重叠。例如， 如果能够具体说明 “前台”轮班和“后台” 两者同时发生的移位。 这可以通过单独的调用来完成 使用不同的移位数据， 除了关于日程安排的限制 在给定时间内进行多次轮班的人员 周期将被错过。

• 可指定工人的最短重新安排时间段 以每个工人（而不是全球）为基础。例如， 如果乔觉得工作过度或处理个人问题， 或者是一个初学者，我们可以安排他 比其他工人少。

• 一些自动/随机/公平选择员工以填补最低限度的方法 当没有合适的工人时轮班。

• 一些处理突然取消的方法，只是填补空白 无需重新安排其他班次。

输出试验

可能，算法应该生成尽可能多的匹配解，每个解都是这样的：

^{pr2}$

下面是一个针对单个解决方案的测试函数，给出了上述数据。我认为这是对的，但我也希望能有同行的评论。在

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

  # 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))

尝试

我试过用遗传算法来实现这一点，但似乎不能很好地调整它，所以尽管基本原理似乎适用于单班制，但它甚至不能解决几个轮班和几个工人的简单情况。在

我最近的尝试是生成每一个可能的排列作为一个解决方案，然后削减不满足约束的排列。这似乎可以更快地工作，并使我走得更远，但我使用的是python2.6itertools.product（）来帮助生成排列，但我不能完全正确。如果有很多bug我也不会感到惊讶，因为老实说，这个问题并不适合我的头脑：）

目前我的代码在两个文件中：模型.py以及旋转木马. 模型.py看起来像：

# -*- 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

以及旋转木马看起来像：

#!/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()

在结果之前截取调试输出，当前将给出：

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>
                >
            >
        >
    >

Answer 1

我没有算法选择，但我可以联系一些实际的考虑。在

由于该算法处理的是取消，所以每当发生调度异常时都必须运行它来重新安排所有人。在

考虑到有些算法不是很线性的，可能会从根本上重新安排每个人的时间。你可能想避免这种情况，人们喜欢提前知道他们的日程安排。在

您可以在不重新运行算法的情况下处理某些取消，因为它可以预先安排下一个或两个可用的人。在

总是生成最理想的解决方案可能是不可能或不可取的，但您可以为每个工作线程保留“低于最佳”事件的运行计数，并且在必须分配另一个“错误选择”时，始终选择计数最低的工作线程。这就是人们通常关心的（一些“糟糕”的日程安排决策经常/不公平）。在

Answer 2

好吧，我不知道具体的算法，但这里是我要考虑的。在

评估

不管你用什么方法，你都需要一个函数来计算你的解满足约束的程度。你可以采取“比较”的方法（没有全局分数，但可以比较两个解决方案），但我建议进行评估。在

真正好的是，如果你能在较短的时间内得到一个分数，例如每天，如果你能从一个部分解中“预测”最终得分的范围（例如，7天中的前3天），这对算法非常有帮助。这样，如果部分解已经太低而无法满足您的期望，则可以中断基于此部分解的计算。在

对称性

很可能在这200个人中，你有着相似的特征：即具有相同特征（可用性、经验、意愿等）。如果你拿两个有相同侧面的人，他们是可以互换的：

• （1）其他工人：1。。。在
• 解决方案二：（1班：吉姆）（2班：乔）……仅限其他工人。。。在

从你的角度来看，实际上是相同的解决方案。在

好消息是，通常情况下，你的个人资料比个人资料少，这对花在计算上的时间有很大帮助！在

例如，假设您基于解决方案1生成了所有解决方案，那么就不需要基于解决方案2计算任何问题。在

迭代

你可以考虑一周一次生成一个完整的时间表。净收益是减少了一周的复杂性（可能性更小）。在

然后，一旦你有了这个星期，你就要计算第二个，当然要注意第一个要考虑到第一个约束条件。在

这样做的好处是，你可以显式地设计你的算法来考虑已经使用过的解决方案，这样对于下一代的日程生成来说，它将确保不会让一个人连续工作24小时！在

序列化

您应该考虑解决方案对象的序列化（根据您的选择，pickle对于Python来说非常好）。在生成新的计划时，您将需要以前的计划，而且我打赌您不希望为200人手动输入它。在

详尽

现在，在所有这些之后，我实际上更倾向于一个详尽的搜索，因为使用对称性和求值的可能性可能不那么多（问题仍然是NP完全的，没有银弹）。在

你可以试试看Backtracking Algorithm。在

另外，你应该看看以下处理类似问题的链接：

• Python Sudoku Solver
• Java Sudoku Solver using Knuth Dancing Links（使用回溯算法）

两者都讨论了在实现过程中遇到的问题，因此检查它们应该会对您有所帮助。在

Answer 3

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.

简而言之，不要！除非你在遗传算法方面有丰富的经验，否则你不会做对。

• 它们是近似的方法，不能保证收敛到一个可行的解决方案。在
• 只有当您能够合理地确定当前解决方案的质量（即未满足的标准数量）时，它们才会起作用。在
• 它们的质量在很大程度上取决于你用来将以前的解决方案组合成新的解决方案的操作员的质量。在

如果你对遗传算法的经验几乎为零，那么在小python程序中获得正确的结果是一件困难的事情。如果你有一小群人，彻底搜索并不是那么糟糕的选择。问题是它可能对n人起作用，对n+1人来说会很慢，对{}人来说会慢得让人无法忍受，而且很可能你的n最终会低至10。在

你正在研究一个NP完全问题，没有容易取胜的解决方案。如果你选择的python的时间表不够好的话，你的脚本不太可能有更好的时间表。在

如果您坚持通过自己的代码来实现这一点，那么使用min-max或模拟退火可以更容易地得到一些结果。在