ortools中修改的公交车调度问题

Modified bus scheduling problem in ortools

我想修改 bus scheduling problem from ortools 以便每个 driver 的班次在插槽方面是连续的,并且 driver 可以根据需要同时共享一个班次.

例如,假设我们有以下 half-hour 班次(格式类似于来自 ortools 的原始 bus_scheduling_problem):

shifts = [
[0, '07:00', '07:30', 420, 450, 30],
[1, '07:30', '08:00', 450, 480, 30],
[2, '08:00', '08:30', 480, 510, 30],
[3, '08:30', '09:00', 510, 540, 30],
[4, '09:00', '09:30', 540, 570, 30],
[5, '09:30', '10:00', 570, 600, 30],
[6, '10:00', '10:30', 600, 630, 30],
[7, '10:30', '11:00', 630, 660, 30],
[8, '11:00', '11:30', 660, 690, 30],
[9, '11:30', '12:00', 690, 720, 30],
[10, '12:00', '12:30', 720, 750, 30],
[11, '12:30', '13:00', 750, 780, 30],
[12, '13:00', '13:30', 780, 810, 30],
[13, '13:30', '14:00', 810, 840, 30],
[14, '14:00', '14:30', 840, 870, 30],
[15, '14:30', '15:00', 870, 900, 30],
[16, '15:00', '15:30', 900, 930, 30],
[17, '15:30', '16:00', 930, 960, 30],
[18, '16:00', '16:30', 960, 990, 30],
[19, '16:30', '17:00', 990, 1020, 30],
[20, '17:00', '17:30', 1020, 1050, 30],
[21, '17:30', '18:00', 1050, 1080, 30],
[22, '18:00', '18:30', 1080, 1110, 30],
[23, '18:30', '19:00', 1110, 1140, 30],
[24, '19:00', '19:30', 1140, 1170, 30],
[25, '19:30', '20:00', 1170, 1200, 30],
[26, '20:00', '20:30', 1200, 1230, 30],
[27, '20:30', '21:00', 1230, 1260, 30],
[28, '21:00', '21:30', 1260, 1290, 30],
[29, '21:30', '22:00', 1290, 1320, 30],
[30, '22:00', '22:30', 1320, 1350, 30],
[31, '22:30', '23:00', 1350, 1380, 30],
[32, '23:00', '23:30', 1380, 1410, 30],
[33, '23:30', '24:00', 1410, 1440, 30]
]

我成功执行了 this version of the bus_scheduling code,我发现我需要 2 driver 来满足上述计划的需要。工作时间范围从07:00 am to 24:00 (midnight).

因此,如果我们有 2 辆公共汽车 drivers 用于此时间表,我更喜欢基于 12-h driver 轮班涵盖日常职责的分配,如下所示:

Driver 1: 07:00 - 19:00 with a break at 13:00
Driver 2: 12:00 - 24:00 with a break at 14:00 (basically no overlap with Driver 1's break)

我所说的连续小时的意思是满足 12-h driver 形式的 07:00-11:00 + 14:00-15:00 + 17:00-24:00 移位解决方案的解决方案应该 是可以接受的。具有更多 drivers 的解决方案还应确保中断不重叠,以便 而不是 所有 drivers 都处于中断状态。此外,由于工作量大,休息时间可能会被阻塞。

我在 or-tools 讨论 中得到一个 answer 说我需要在每个节点维护自从转变的开始,但我很难编码,假设它解决了我的问题。

对我来说,bus scheduling problem from ortools is an overkill for your task since you mentioned that shift durations are always 30 minutes, and setup/cleanup time is not needed. Moreover, drivers must work exactly 11 hours and have a contiguous break. Instead, we could write a script similar to nurse scheduling problem可能更容易理解(对我来说,这是第一次用or-tools写东西而且很清楚).

准备

首先,总班次可以计算如下:

num_shifts = len(shifts)

需要 driver 的数量:

num_drivers = ceil(float(num_shifts) / working_time)

在您的情况下,driver 必须刚好行驶 11 小时,因此需要 22 班次(每个班次固定为 30 分钟):

working_time = 22

休息时间是 1 小时,所以:

break_time = 2

正如您在评论中提到的,每个 driver 必须在驾驶 4 小时后休息,但不得迟于 8 小时后:

break_interval = [8, 16]

一个driver可以开始工作的最后班次:

latest_start_shift = num_shifts - working_time - break_time

真的,如果 he/she 晚一点开始工作,那么 driver 就不会在整个工作时间内工作。

构建模型

让我们为 drivers 定义一个班次数组:

driver_shifts = {}
for driver_id in range(num_drivers):
    for shift_id in range(num_shifts):
        driver_shifts[(driver_id, shift_id)] = model.NewBoolVar('driver%ishift%i' % (driver_id, shift_id))

driver_shifts[(d, s)] 等于 1 如果 shift s 分配给 driver d,否则 0

此外,为 drivers 创建一个开始班次数组:

start_time = {}
for driver_id in range(num_drivers):
    for shift_id in range(latest_start_shift + 1):
        start_time[(driver_id, shift_id)] = model.NewBoolVar('driver%istart%i' % (driver_id, shift_id))

start_time[(d, s)] 等于 1 如果 driver d 在轮班 s 开始工作日,否则 0

Driver 每天正好开车 11 小时

每个driver必须在一天内准确驾驶规定的驾驶时间:

for driver_id in range(num_drivers):
    model.Add(sum(driver_shifts[(driver_id, shift_id)] for shift_id in range(num_shifts)) == working_time)

但是,这还不够,因为 driver 必须连续进行,中间有一个中断。我们稍后会看到如何做到这一点。

所有班次都由 drivers

涵盖

每个班次必须至少由一名司机负责 driver:

for shift_id in range(num_shifts):
    model.Add(sum(driver_shifts[(driver_id, shift_id)] for driver_id in range(num_drivers)) >= 1)

Driver连续驱动

这里 start_time 发挥作用。基本思想是,对于 driver 的每个可能开始时间,我们强制 driver 在该时间工作(从物理上讲,driver 每天只能开始工作一次!)。

所以,driver每天只能开始工作一次:

for driver_id in range(num_drivers):
    model.Add(sum(start_time[(driver_id, start_shift_id)] for start_shift_id in range(latest_start_shift + 1)) == 1)

对于driver的每个开始时间,连续working_time + break_time内的工作时间为working_time

for driver_id in range(num_drivers):
     for start_shift_id in range(latest_start_shift + 1):
         model.Add(sum(driver_shifts[(driver_id, shift_id)] for shift_id in
                   range(start_shift_id, start_shift_id + working_time + break_time)) == working_time) \
             .OnlyEnforceIf(start_time[(driver_id, start_shift_id)]) 

中断是连续的

为此,我们需要一个额外的数组 break_ind[(d, s, b)],它表示给定的 driver d 和给定的工作班次开始 s 是否在班次 b。因此,在这种情况下,休息时间的 driver_shifts 值应为 0

     l = start_shift_id + break_interval[0]
     r = start_shift_id + break_interval[1]
     for s in range(l, r):
        break_ind[(driver_id, start_shift_id, s)] = model.NewBoolVar("d%is%is%i"%(driver_id, start_shift_id, s))
        model.Add(sum(driver_shifts[(driver_id, s1)] for s1 in range(s, s + break_time)) == 0)\
        .OnlyEnforceIf(start_time[(driver_id, start_shift_id)])\
        .OnlyEnforceIf(break_ind[(driver_id, start_shift_id, s)]) 

另外,driver每天只能休息一次:

model.Add(sum(break_ind[(driver_id, start_shift_id, s)] for s in range(l, r)) == 1)

完整代码

您可以查看下面的完整代码或here(我添加它以供将来参考)。在那里您还可以找到 drivers 不休息的情况的简化版本。

from ortools.sat.python import cp_model
from math import ceil

shifts = [
    [0, '07:00', '07:30', 420, 450, 30],
    [1, '07:30', '08:00', 450, 480, 30],
    [2, '08:00', '08:30', 480, 510, 30],
    [3, '08:30', '09:00', 510, 540, 30],
    [4, '09:00', '09:30', 540, 570, 30],
    [5, '09:30', '10:00', 570, 600, 30],
    [6, '10:00', '10:30', 600, 630, 30],
    [7, '10:30', '11:00', 630, 660, 30],
    [8, '11:00', '11:30', 660, 690, 30],
    [9, '11:30', '12:00', 690, 720, 30],
    [10, '12:00', '12:30', 720, 750, 30],
    [11, '12:30', '13:00', 750, 780, 30],
    [12, '13:00', '13:30', 780, 810, 30],
    [13, '13:30', '14:00', 810, 840, 30],
    [14, '14:00', '14:30', 840, 870, 30],
    [15, '14:30', '15:00', 870, 900, 30],
    [16, '15:00', '15:30', 900, 930, 30],
    [17, '15:30', '16:00', 930, 960, 30],
    [18, '16:00', '16:30', 960, 990, 30],
    [19, '16:30', '17:00', 990, 1020, 30],
    [20, '17:00', '17:30', 1020, 1050, 30],
    [21, '17:30', '18:00', 1050, 1080, 30],
    [22, '18:00', '18:30', 1080, 1110, 30],
    [23, '18:30', '19:00', 1110, 1140, 30],
    [24, '19:00', '19:30', 1140, 1170, 30],
    [25, '19:30', '20:00', 1170, 1200, 30],
    [26, '20:00', '20:30', 1200, 1230, 30],
    [27, '20:30', '21:00', 1230, 1260, 30],
    [28, '21:00', '21:30', 1260, 1290, 30],
    [29, '21:30', '22:00', 1290, 1320, 30],
    [30, '22:00', '22:30', 1320, 1350, 30],
    [31, '22:30', '23:00', 1350, 1380, 30],
    [32, '23:00', '23:30', 1380, 1410, 30],
    [33, '23:30', '24:00', 1410, 1440, 30]
]

class VarArraySolutionPrinter(cp_model.CpSolverSolutionCallback):
    def __init__(self, driver_shifts, num_drivers, num_shifts, solutions):
        cp_model.CpSolverSolutionCallback.__init__(self)
        self.driver_shifts = driver_shifts
        self.num_drivers = num_drivers
        self.num_shifts = num_shifts
        self.solutions = solutions
        self.solution_id = 0

    def on_solution_callback(self):
        if self.solution_id in self.solutions:
            self.solution_id += 1
            print ("Solution found!")
            for driver_id in range(self.num_drivers):
                print ("*************Driver#%s*************" % driver_id)
                for shift_id in range(self.num_shifts):
                    if (self.Value(self.driver_shifts[(driver_id, shift_id)])):
                        print('Shift from %s to %s' %
                              (shifts[shift_id][1],
                               shifts[shift_id][2]))
            print()

    def solution_count(self):
        return self.solution_id

solver = cp_model.CpSolver()
model = cp_model.CpModel()

num_shifts = len(shifts)

working_time = 22
break_time = 2

# when take a break within the working time
break_interval = [8, 16]

latest_start_shift = num_shifts - working_time - break_time
num_drivers = ceil(float(num_shifts) / working_time)

# create an array of assignments of drivers
driver_shifts = {}
for driver_id in range(num_drivers):
    for shift_id in range(num_shifts):
        driver_shifts[(driver_id, shift_id)] = model.NewBoolVar('driver%ishift%i' % (driver_id, shift_id))

# driver must work exactly {working_time} shifts
for driver_id in range(num_drivers):
    model.Add(sum(driver_shifts[(driver_id, shift_id)] for shift_id in range(num_shifts)) == working_time)

# each shift must be covered by at least one driver
for shift_id in range(num_shifts):
    model.Add(sum(driver_shifts[(driver_id, shift_id)] for driver_id in range(num_drivers)) >= 1)

# create an array of start times for drivers
start_time = {}
for driver_id in range(num_drivers):
    for shift_id in range(latest_start_shift + 1):
        start_time[(driver_id, shift_id)] = model.NewBoolVar('driver%istart%i' % (driver_id, shift_id))

break_ind = {}
for driver_id in range(num_drivers):
     for start_shift_id in range(latest_start_shift + 1):
         model.Add(sum(driver_shifts[(driver_id, shift_id)] for shift_id in
                   range(start_shift_id, start_shift_id + working_time + break_time)) == working_time) \
             .OnlyEnforceIf(start_time[(driver_id, start_shift_id)])

         l = start_shift_id + break_interval[0]
         r = start_shift_id + break_interval[1]
         for s in range(l, r):
            break_ind[(driver_id, start_shift_id, s)] = model.NewBoolVar("d%is%is%i"%(driver_id, start_shift_id, s))
            model.Add(sum(driver_shifts[(driver_id, s1)] for s1 in range(s, s + break_time)) == 0)\
            .OnlyEnforceIf(start_time[(driver_id, start_shift_id)])\
            .OnlyEnforceIf(break_ind[(driver_id, start_shift_id, s)])
         model.Add(sum(break_ind[(driver_id, start_shift_id, s)] for s in range(l, r)) == 1)

for driver_id in range(num_drivers):
    model.Add(sum(start_time[(driver_id, start_shift_id)] for start_shift_id in range(latest_start_shift + 1)) == 1)

solution_printer = VarArraySolutionPrinter(driver_shifts, num_drivers, num_shifts, range(2))
status = solver.SearchForAllSolutions(model, solution_printer)