问题描述¶
在作业车间调度问题 (JSSP) 中，一组作业必须在车间中的每台机器上进行处理。作业由有序的任务序列（称为活动）组成。一个活动代表该作业在一台机器上的处理，并具有给定的处理时间。每个作业在每台机器上都有一个活动，且每个活动只能在前一个活动结束后才能开始。每台机器一次只能处理一个活动。目标是找到一个使 makespan（即所有作业处理完成的时间）最小化的作业序列。
学习要点¶
- 添加 区间决策变量 来建模活动
- 添加 列表决策变量 来建模每台机器上活动的顺序
- 定义一个 符号 Lambda 函数 将 interval 变量和 list 变量关联起来
数据¶
我们提供的作业车间调度问题 (JSSP) 实例遵循 Taillard 格式。数据文件的格式如下：
- 第一行：作业数、机器数、生成实例所用的种子，以及先前找到的上界和下界。
- 对每个作业：在每台机器上的处理时间（按处理顺序给出）。
- 对每个作业：处理顺序（访问机器的有序列表）。
建模思路¶
作业车间调度问题 (JSSP) 的 OptAgent 模型使用 区间决策变量 来建模活动的时间区间。我们将每个 interval 的长度约束为对应活动的处理时间。然后我们可以编写紧前约束：对于每个作业，该作业的每个活动必须在由前一台机器处理的活动结束之后才能开始。
除了表示活动时间区间的 区间决策变量 之外，我们还使用 列表决策变量。与 Flow Shop 示例一样，一个 list 对机器上活动的顺序进行建模。使用 count 算子约束列表的大小，我们确保每台机器处理每个作业。析取资源约束（每台机器一次只能处理一个活动）可表述如下：对所有 i，在位置 i+1 处理的活动必须位于位置 i 处理的活动结束之后开始。为了对这些约束建模，我们将 区间决策变量（时间区间）与 列表决策变量（作业顺序）配对。我们编写一个 符号 Lambda 函数 来表示两个连续活动之间的关系。该函数在每台机器所处理的所有活动上的变参 'and' 算子中使用。
目标是最小化 makespan，即所有活动都被处理完成的时间。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
# The input files follow the "Taillard" format
def read_instance(filename):
lines = Path(filename).read_text(encoding="utf-8").splitlines()
first_line = lines[1].split()
# Number of jobs
nb_jobs = int(first_line[0])
# Number of machines
nb_machines = int(first_line[1])
# Processing times for each job on each machine (given in the processing order)
processing_times_in_processing_order = [
[int(lines[i].split()[j]) for j in range(nb_machines)] for i in range(3, 3 + nb_jobs)
]
# Processing order of machines for each job
machine_order = [
[int(lines[i].split()[j]) - 1 for j in range(nb_machines)] for i in range(4 + nb_jobs, 4 + 2 * nb_jobs)
]
# Reorder processing times: processing_time[j][m] is the processing time of the
# task of job j that is processed on machine m
processing_time = [
[processing_times_in_processing_order[j][machine_order[j].index(m)] for m in range(nb_machines)]
for j in range(nb_jobs)
]
# Trivial upper bound for the end times of the tasks
max_end = sum(sum(processing_time[j]) for j in range(nb_jobs))
return nb_jobs, nb_machines, processing_time, machine_order, max_end
def main(instance_file, output_file=None, time_limit=10):
nb_jobs, nb_machines, processing_time, machine_order, max_end = read_instance(instance_file)
model = OptModel()
# tasks[job][machine] is the processing interval on that machine.
tasks = [[model.interval(0, max_end) for _ in range(nb_machines)] for _ in range(nb_jobs)]
for job in range(nb_jobs):
for machine in range(nb_machines):
model.constraint(
tasks[job][machine].length() == processing_time[job][machine],
)
task_array = model.array([model.array(job_tasks) for job_tasks in tasks])
# Respect the fixed machine-processing order of every job.
for job in range(nb_jobs):
for operation in range(nb_machines - 1):
before = machine_order[job][operation]
after = machine_order[job][operation + 1]
model.constraint(
tasks[job][before] < tasks[job][after],
)
# Each machine list is constrained to contain every job exactly once.
jobs_order = [model.list(nb_jobs) for machine in range(nb_machines)]
for machine in range(nb_machines):
sequence = jobs_order[machine]
model.constraint(
sequence.count() == nb_jobs,
)
machine_index = machine
precedes_next = model.lambda_function(
lambda position: task_array[sequence[position], machine_index]
< task_array[sequence[position + 1], machine_index]
)
model.constraint(
model.and_(model.range(0, nb_jobs - 1), precedes_next),
)
makespan = model.max([tasks[job][machine_order[job][nb_machines - 1]].end() for job in range(nb_jobs)])
model.minimize(makespan)
solution = solve(model, time_limit_s=float(time_limit))
if not solution.feasible:
print(f"No feasible schedule found; Status = {solution.feasible}")
return solution
expressions = {
"makespan": makespan,
**{f"machine_{machine}": jobs_order[machine] for machine in range(nb_machines)},
}
values = {name: expression.value for name, expression in expressions.items()}
lines = [f"Makespan = {values['makespan']}; Status = {solution.feasible}"]
machine_lines = []
for machine in range(nb_machines):
sequence_text = " ".join(str(job) for job in values[f"machine_{machine}"])
lines.append(f"Machine {machine}: {sequence_text}")
machine_lines.append(sequence_text)
print("\n".join(lines))
if output_file is not None:
Path(output_file).write_text("\n".join(machine_lines) + "\n", encoding="utf-8")
return solution
INSTANCE_DIR = Path.cwd() / "instances"
solution = main(INSTANCE_DIR / "ft06.txt", time_limit=5)