Python中的高效排班

Question

我目前正在为一家出租车公司做一些排班模拟。该公司有350辆出租车，每天都在使用。司机每班工作5班，每班12小时，每天有4个重叠的班次。班次为3:00-15:00、15:00-3:00、16:00-4:00和4:00-16:00。我最初是用Python开发的，因为需要快速开发，我认为性能可以接受。最初的参数只需要一天两班（3:00-15:00和15:00-3:00），虽然性能不是很好，但对我来说已经足够了。它可以用一个简单的暴力算法，在大约8分钟内为司机制定一个每周的时间表（评估所有潜在的交换，看看情况是否可以改善）

由于四个轮班重叠，表现绝对糟糕。做一个每周的时间表需要一个多小时。我用cProfile做了一些分析，看起来罪魁祸首是两种方法。一种方法是确定驾驶员换班时是否存在冲突。它确保他们不是在同一天轮班，或在前一班或后一班。每天只有两班，这很简单。人们只需确定驾驶员是否已经被安排在轮班之前或之后工作。随着四个相互重叠的转变，这变得更加复杂。第二个罪魁祸首是确定轮班是白班还是夜班的方法。同样，对于最初的两个班次，这就像确定班次号是偶数还是奇数一样简单，班次号从0开始。第一班（0班）被指定为夜班，下一班是白班，依此类推。前两个是夜，下两个是，等等。这些方法互相调用，所以我会把他们的身体放在下面。在

def conflict_exists(shift, driver, shift_data):
    next_type = get_stype((shift+1) % 28)
    my_type = get_stype(shift)

    nudge = abs(next_type - my_type)

    if driver in shift_data[shift-2-nudge] or driver in shift_data[shift-1-nudge] or driver in shift_data[(shift+1-(nudge*2)) % 28] or driver in shift_data[(shift+2-nudge) % 28] or driver in shift_data[(shift+3-nudge) % 28]:
        return True
    else:
        return False

注意get_stype返回轮班的类型，0表示是夜班，1表示是白班。在

为了确定轮班类型，我使用以下方法：

^{pr2}$

下面是cProfile的相关输出：

     ncalls  tottime  percall  cumtime  percall
     57662556   19.717    0.000   19.717    0.000 sim8.py:241(get_stype)
     28065503   55.650    0.000   77.591    0.000 sim8.py:247(in_conflict)

有人对我如何改进这个脚本的性能有什么明智的建议或建议吗？任何帮助将不胜感激！在

干杯

蒂姆

编辑：对不起，我应该澄清一下，每个班次的数据都存储为一个集合，即班次数据[k]属于集合数据类型。在

编辑2：

根据下面的请求添加主循环以及调用的其他方法。有点乱，我为此道歉。在

def optimize_schedule(shift_data, driver_shifts, recheck):
    skip = set()

    if len(recheck) == 0:
        first_run = True
        recheck = []
        for i in xrange(28):
            recheck.append(set())
    else:
        first_run = False

    for i in xrange(28):

        if (first_run):
            targets = shift_data[i]
        else:
            targets = recheck[i]

        for j in targets:
            o_score = eval_score = opt_eval_at_coord(shift_data, driver_shifts, i, j)

            my_type = get_stype(i)
            s_type_fwd = get_stype((i+1) % 28)

            if (my_type == s_type_fwd):
                search_direction = (i + 2) % 28
                end_direction = i
            else:
                search_direction = (i + 1) % 28 
                end_direction = (i - 1) % 28 

            while True:
                if (end_direction == search_direction):
                    break
                for k in shift_data[search_direction]:

                    coord = search_direction * 10000 + k 

                    if coord in skip:
                        continue

                    if k in shift_data[i] or j in shift_data[search_direction]:
                        continue

                    if in_conflict(search_direction, j, shift_data) or in_conflict(i, k, shift_data):
                        continue

                    node_a_prev_score = o_score
                    node_b_prev_score = opt_eval_at_coord(shift_data, driver_shifts, search_direction, k)

                    if (node_a_prev_score == 1) and (node_b_prev_score == 1):
                        continue

                    a_type = my_type
                    b_type = get_stype(search_direction)

                    if (node_a_prev_score == 1):
                        if (driver_shifts[j]['type'] == 'any') and (a_type != b_type):
                            test_eval = 2
                        else:
                            continue
                    elif (node_b_prev_score == 1):
                        if (driver_shifts[k]['type'] == 'any') and (a_type != b_type):
                            test_eval = 2
                        else:
                            test_eval = 0
                    else:
                        if (a_type == b_type):
                            test_eval = 0
                        else:
                            test_eval = 2

                    print 'eval_score: %f' % test_eval

                    if (test_eval > eval_score):

                        cand_coords = [search_direction, k]
                        eval_score = test_eval
                        if (test_eval == 2.0):
                            break
                else:
                    search_direction = (search_direction + 1) % 28
                    continue

                break

            if (eval_score > o_score):
                print 'doing a swap: ',
                print cand_coords,

                shift_data[i].remove(j)
                shift_data[i].add(cand_coords[1])

                shift_data[cand_coords[0]].add(j)   
                shift_data[cand_coords[0]].remove(cand_coords[1])

                if j in recheck[i]:
                    recheck[i].remove(j)

                if cand_coords[1] in recheck[cand_coords[0]]:               
                    recheck[cand_coords[0]].remove(cand_coords[1])

                recheck[cand_coords[0]].add(j)
                recheck[i].add(cand_coords[1])

            else:
                coord = i * 10000 + j
                skip.add(coord)

    if first_run:
        shift_data = optimize_schedule(shift_data, driver_shifts, recheck)

    return shift_data



def opt_eval_at_coord(shift_data, driver_shifts, i, j):
    node = j
    if in_conflict(i, node, shift_data):
        return float('-inf')
    else:
        s_type = get_stype(i)

        d_pref = driver_shifts[node]['type']

        if (s_type == 0 and d_pref == 'night') or (s_type == 1 and d_pref == 'day') or (d_pref == 'any'):
            return 1
        else:
            return 0

Answer 1

嗯。有趣又有趣的问题。我得多看一看。现在，我有一个建议：为什么要引入浮动？我将按如下方式获取_stype（）：

def get_stype(k):
    if k % 4 < 2:
        return 0
    return 1

这不是一个巨大的加速，但它更快（更简单）。而且，你不必在喂食get_stype时执行mod 28，因为这已经由get_stype中的mod 4处理。在

如果有重大的改进，它们将以更好的算法的形式出现。（我不是说你的算法不好，或者还有更好的算法。我真的没有花足够的时间看它。但如果找不到更好的算法，那么使用PyPy、Cython、Shed Skin或完全用另一种（更快的）语言重写，就必须进一步显著提高速度

Answer 2

没有什么能明显减慢这些函数的速度，事实上它们并不慢。他们只是经常接到电话。你说你使用的是蛮力算法-你能写一个不尝试所有可能的组合的算法吗？或者有没有一种更有效的方法，比如用司机而不是轮班来存储数据？在

当然，如果您需要即时加速，那么在PyPy这样的解释器中运行，或者使用Cython将关键部分转换为C可能会有好处

Answer 3

我不认为你的问题是运行这两个函数所需的时间。请注意，函数的percall值是0.000。这意味着每次调用函数时，所用时间不到1毫秒。在

我想你的问题是函数被调用的次数。python中的函数调用非常昂贵。例如，在我的机器上调用一个不执行任何操作的函数57662556次需要7.15秒：

>>> from timeit import Timer
>>> t = Timer("t()", setup="def t(): pass")
>>> t.timeit(57662556)
7.159075975418091

我想知道的是shift_data变量。值是列表还是dicts？在

^{pr2}$

如果是列表，in将花费O(N)时间，而如果是dict，O(1)时间

编辑：由于shift_数据值是设置的，所以应该可以