乐趣区

关于算法:运筹优化工具ortools解读与实践CPSAT应用案例

上一篇文章咱们介绍了 CP-SAT 的简略用法,以及 CP-SAT 蕴含的不同变量、束缚函数。实际出真知,本篇文章将用几个例子加深对 CP-SAT 应用的了解。

1. 员工排班

问题形容

医院主管须要为四名护士创立一个为期三天的时间表,但必须满足以下条件:

每天被分成三个 8 小时的班次。
每天,每个轮班调配给一名护士,没有护士的工作超过一班。
在三天的工作期间,每个护士至多要上两班。

代码及解读

from ortools.sat.python import cp_model

class NursesPartialSolutionPrinter(cp_model.CpSolverSolutionCallback):

    def __init__(self, shifts, num_nurses, num_days, num_shifts, sols):
        cp_model.CpSolverSolutionCallback.__init__(self)
        self._shifts = shifts
        self._num_nurses = num_nurses
        self._num_days = num_days
        self._num_shifts = num_shifts
        self._solutions = set(sols)
        self._solution_count = 0

    #回调函数,每产生一个 solution 会调用一次回调函数。这里对回调函数进行重写。def on_solution_callback(self):
        self._solution_count += 1
        if self._solution_count in self._solutions:
            print('Solution %i' % self._solution_count)
            for d in range(self._num_days):
                print('Day %i' % d)
                for n in range(self._num_nurses):
                    is_working = False
                    for s in range(self._num_shifts):
                        if self.Value(self._shifts[(n, d, s)]):
                            is_working = True
                            print('Nurse %i works shift %i' % (n, s))
                    if not is_working:
                        print('Nurse {} does not work'.format(n))
            print()

    def solution_count(self):
        return self._solution_count


def main():
    # Data.
    num_nurses = 4    # 护士数量
    num_shifts = 3    # 每天班次
    num_days = 3      # 一共要排多少天班
    all_nurses = list(range(num_nurses))
    all_shifts = list(range(num_shifts))
    all_days = list(range(num_days))

    # 创立 cp-sat 求解器
    model = cp_model.CpModel()

    # 创立 0 - 1 布尔变量变量,shifts[(n, d, s)]:示意护士 n 被调配到第 d 天的第 s 个班次
    shifts = {}
    for n in all_nurses:
        for d in all_days:
            for s in all_shifts:
                shifts[(n, d, s)] = model.NewBoolVar('shift_n%id%is%i' % (n, d, s))

    # 增加束缚,每个班次都必须有一名护士
    for d in all_days:
        for s in all_shifts:
            model.Add(sum(shifts[(n, d, s)] for n in all_nurses) == 1)

    # 增加束缚,每个护士一天最多只能做一个班次
    for n in all_nurses:
        for d in all_days:
            model.Add(sum(shifts[(n, d, s)] for s in all_shifts) <= 1)

    # 如何尽可能均匀地调配护士轮班,因为三天有九班,咱们能够给四名护士每人调配两班。# 之后还会有一个班次,能够调配给任何护士。# 以下代码确保每位护士在三天的工作期间至多上两班。# min_shifts_per_nurse 每个护士至多调配的班次数量
    # max_shifts_per_nurse 每个护士至少调配的班次数量
    min_shifts_per_nurse = (num_shifts * num_days) // num_nurses
    max_shifts_per_nurse = min_shifts_per_nurse + 1
    for n in all_nurses:
        num_shifts_worked = sum(shifts[(n, d, s)] for d in all_days for s in all_shifts)
        model.Add(min_shifts_per_nurse <= num_shifts_worked)
        model.Add(num_shifts_worked <= max_shifts_per_nurse)

    # 求解问题
    solver = cp_model.CpSolver()

    # 展现后果(只展现 5 条后果)a_few_solutions = range(5)
    solution_printer = NursesPartialSolutionPrinter(shifts, num_nurses, num_days, num_shifts, a_few_solutions)
    solver.SearchForAllSolutions(model, solution_printer)

    # 统计分析.
    print()
    print('Statistics')
    print('- conflicts       : %i' % solver.NumConflicts())
    print('- branches        : %i' % solver.NumBranches())
    print('- wall time       : %f ms' % solver.WallTime())
    print('- solutions found : %i' % solution_printer.solution_count())


if __name__ == '__main__':
    main()

2. 车间调度

问题形容

作业车间中,在多台机器上解决多个作业。每个作业由一系列工作组成,这些工作必须依照给定的程序执行,并且每个工作必须在特定的机器上进行解决。例如,该工作能够是生产单个消费品,例如汽车。问题是将工作安顿在机器上,以最小化打算的长度——所有作业实现所需的工夫。

作业车间问题有几个限度:

  • 在前一个作业工作实现之前,不能启动该作业的任何工作。
  • 一台机器一次只能解决一项工作。
  • 工作一旦启动,就必须始终运行到实现。

上面是工作车间问题一个简略的例子,每个工作由一对数字 (m,p) 示意,m 是机器的数量,p 示意工作的解决工夫。(作业和机器的编号从 0 开始。)

job 0 = [(0, 3), (1, 2), (2, 2)]
job 1 = [(0, 2), (2, 1), (1, 4)]
job 2 = [(1, 4), (2, 3)]

job 0 有三个工作,第一个 (0,3) 必须在机器 0 上以 3 个单位的工夫进行解决,第二个 (1,2) 必须在机器 1 上以 2 个单位的工夫解决,以此类推,总共有八个工作。

作业车间问题的后果是为每个任务分配一个启动工夫,该工夫满足下面给出的约束条件。下图显示了该问题的一种可能的解决方案:

问题的变量和束缚

决策变量:让 task(i, j)示意作业 i 序列中的第 j 个工作,定义为 t_i,j,问题的决策变量为特定工作 t 的开始工夫。

作业车间问题有两种类型的束缚:

优先束缚:这是因为对于同一作业中的任意两个间断工作,必须在启动第二个工作之前实现第一个工作。例如,task(0,2)和 task(0,3)是 job 0 的间断工作。因为 task(0,2)的解决工夫为 2,因而 task(0,3)的开始工夫必须在 task 2 的开始工夫之后至多为 2 个单位的工夫。后果失去如下束缚:

\(t_{0,2}+2<=t_{0,3} \)

没有重叠的束缚:这是因为机器不能同时解决两个工作。例如,task(0,2)和 task(2,1)都在机器 1 上解决。因为它们的解决工夫别离为 2 和 4,因而必须满足以下条件之一:

\(t_{0,2}+2<=t_{2,1} \),如果 task(0,2)在 task(2,1)之前被调度

\(t_{2,1}+4<=t_{0,2} \),如果 task(2,1)在 task(0,2)之前被调度

指标函数

作业车间问题的指标是最小化实现工夫:从作业的最早开始工夫到最近完结工夫的工夫长度。

代码及解读

"""Minimal jobshop example."""
import collections
from ortools.sat.python import cp_model


def main():
    # 创立数据源
    jobs_data = [# task = (machine_id, processing_time).
        [(0, 3), (1, 2), (2, 2)],  # Job0
        [(0, 2), (2, 1), (1, 4)],  # Job1
        [(1, 4), (2, 3)]  # Job2
    ]

    #机器数
    machines_count = 1 + max(task[0] for job in jobs_data for task in job)
    #机器 ID list
    all_machines = range(machines_count)
    #计算耗费时长的最大值
    horizon = sum(task[1] for job in jobs_data for task in job)

    model = cp_model.CpModel()

    # 定义一个带有 'start end interval' 三个名称域的 tuple 子类
    task_type = collections.namedtuple('task_type', 'start end interval')
    # 定义一个带有 'start job index duration' 四个名称域的 tuple 子类
    assigned_task_type = collections.namedtuple('assigned_task_type',
                                                'start job index duration')

    # 工作字典,job_id, task_id 示意 key,value 为 task_type 的实例
    all_tasks = {}
    machine_to_intervals = collections.defaultdict(list)

    for job_id, job in enumerate(jobs_data):
        for task_id, task in enumerate(job):
            machine = task[0]
            duration = task[1]
            suffix = '_%i_%i' % (job_id, task_id)
            # 决策变量
            start_var = model.NewIntVar(0, horizon, 'start' + suffix)
            end_var = model.NewIntVar(0, horizon, 'end' + suffix)
            # 创立一个带有起起点窗口的窗口变量,依赖于决策变量 start_var、end_var
            interval_var = model.NewIntervalVar(start_var, duration, end_var,
                                                'interval' + suffix)
            # 工作字典,job_id, task_id 示意 key,value 为 task_type 的实例
            all_tasks[job_id, task_id] = task_type(start=start_var,
                                                   end=end_var,
                                                   interval=interval_var)
            #将雷同机器的工作放在同一个 list 中
            machine_to_intervals[machine].append(interval_var)

    #束缚:机器外部不能存在工作工夫窗重合.
    for machine in all_machines:
        model.AddNoOverlap(machine_to_intervals[machine])

    #束缚:job 内 task 程序束缚
    for job_id, job in enumerate(jobs_data):
        for task_id in range(len(job) - 1):
            model.Add(all_tasks[job_id, task_id +
                                1].start >= all_tasks[job_id, task_id].end)

    # 定义指标函数,这里用一个变量 + 束缚的模式定义指标函数.
    obj_var = model.NewIntVar(0, horizon, 'makespan')
    #束缚:obj_var 与所有工作完结工夫 list 的最大值雷同
    model.AddMaxEquality(obj_var, [all_tasks[job_id, len(job) - 1].end
        for job_id, job in enumerate(jobs_data)
    ])
    model.Minimize(obj_var)

    # Creates the solver and solve.
    solver = cp_model.CpSolver()
    status = solver.Solve(model)

    if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
        print('Solution:')
        # Create one list of assigned tasks per machine.
        assigned_jobs = collections.defaultdict(list)
        for job_id, job in enumerate(jobs_data):
            for task_id, task in enumerate(job):
                machine = task[0]
                assigned_jobs[machine].append(
                    assigned_task_type(start=solver.Value(all_tasks[job_id, task_id].start),
                                       job=job_id,
                                       index=task_id,
                                       duration=task[1]))

        # Create per machine output lines.
        output = ''
        for machine in all_machines:
            # Sort by starting time.
            assigned_jobs[machine].sort()
            sol_line_tasks = 'Machine' + str(machine) + ':'
            sol_line = ' '

            for assigned_task in assigned_jobs[machine]:
                name = 'job_%i_task_%i' % (assigned_task.job,
                                           assigned_task.index)
                # Add spaces to output to align columns.
                sol_line_tasks += '%-15s' % name

                start = assigned_task.start
                duration = assigned_task.duration
                sol_tmp = '[%i,%i]' % (start, start + duration)
                # Add spaces to output to align columns.
                sol_line += '%-15s' % sol_tmp

            sol_line += '\n'
            sol_line_tasks += '\n'
            output += sol_line_tasks
            output += sol_line

        # Finally print the solution found.
        print(f'Optimal Schedule Length: {solver.ObjectiveValue()}')
        print(output)
    else:
        print('No solution found.')

    # Statistics.
    print('\nStatistics')
    print('- conflicts: %i' % solver.NumConflicts())
    print('- branches : %i' % solver.NumBranches())
    print('- wall time: %f s' % solver.WallTime())


if __name__ == '__main__':
    main()

最终,新的调度计划如下方所示,整体耗时从 12 小时升高为 11 小时。

退出移动版