上一篇文章我们介绍了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小时。