建模思路¶
开放车间调度问题 (OSSP) 的 OptAgent 模型使用区间决策变量来表示活动的时间范围。区间的长度由每个活动的处理时间约束。
除了区间变量外，我们还使用 list 决策变量。与 Job Shop 示例类似，list 对机器上或作业内的活动进行排序。通过使用 count 算子约束 list 的大小，我们确保每个作业在每台机器上都被处理。
析取资源约束（即每台机器一次只能处理一个活动）可以重新表述如下：对所有 i，在位置 i+1 处理的活动必须在其位置 i 处理的活动结束后才开始。为了建模这些约束，我们将区间决策（时间范围）与 list 决策（作业排序）配对。我们编写一个 lambda 函数 来表达两个连续活动之间的关系。该函数用于每台机器所处理的所有活动的可变参数 and 算子中。
我们使用相同的策略来建模活动的析取约束。对所有作业和所有 i，作业中位置 i+1 的活动必须在该作业中位置 i 的活动结束后才开始。与析取资源约束类似，我们使用一个 lambda 函数结合可变参数 and 算子来建模这些约束，覆盖构成每个作业的所有活动。
目标是最小化 makespan，即所有活动处理完成的时间。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
def read_instance(filename):
# The input files follow the "Taillard" format
with open(filename, 'r') as f:
lines = f.readlines()
first_line = lines[1].split()
nb_jobs = int(first_line[0])
nb_machines = int(first_line[1])
# Processing times for each job on each machine
# (given in the task order, the processing order is a decision variable)
processing_times_task_order = [[int(proc_time) for proc_time in line.split()]
for line in lines[3:3 + nb_jobs]]
# Index of machines for each task
machine_index = [[int(machine_i) - 1 for machine_i in line.split()]
for line in lines[4 + nb_jobs:4 + 2 * nb_jobs]]
# Reorder processing times: processingTime[j][m] is the processing time of the
# task of job j that is processed on machine m
processing_times = [[processing_times_task_order[j][machine_index[j].index(m)]
for m in range(nb_machines)] for j in range(nb_jobs)]
# Trivial upper bound for the end time of tasks
max_end = sum(map(lambda processing_times_job: sum(processing_times_job), processing_times))
return nb_jobs, nb_machines, processing_times, max_end
def main(instance_file, output_file=None, time_limit=60):
nb_jobs, nb_machines, processing_times, max_end = read_instance(instance_file)
model = OptModel()
tasks = [
[model.interval(0, max_end)
for machine in range(nb_machines)]
for job in range(nb_jobs)
]
for job in range(nb_jobs):
for machine in range(nb_machines):
model.constraint(
tasks[job][machine].length() == processing_times[job][machine],
)
task_array = model.array(tasks)
jobs_order = [
model.list(nb_jobs)
for machine in range(nb_machines)
]
for machine, sequence in enumerate(jobs_order):
model.constraint(sequence.count() == nb_jobs)
model.constraint(
model.and_(model.range(0, nb_jobs - 1),
model.lambda_function(
lambda position: task_array[sequence[position // 1], machine]
< task_array[sequence[(position + 1) // 1], machine]
)),
)
machines_order = [
model.list(nb_machines)
for job in range(nb_jobs)
]
for job, sequence in enumerate(machines_order):
model.constraint(sequence.count() == nb_machines)
model.constraint(
model.and_(model.range(0, nb_machines - 1),
model.lambda_function(
lambda position: task_array[job, sequence[position // 1]]
< task_array[job, sequence[(position + 1) // 1]]
)),
)
makespan = model.max(
[tasks[job][machine].end() for job in range(nb_jobs) for machine in range(nb_machines)]
)
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
print(f"Jobs = {nb_jobs}; Machines = {nb_machines}; Makespan = {makespan.value}; Status = {solution.feasible}")
lines = []
for machine, sequence in enumerate(jobs_order):
text = " ".join(str(job + 1) for job in sequence.value)
print(f"Machine {machine + 1}: {text}")
lines.append(text)
if output_file is not None:
Path(output_file).write_text(f"{makespan.value}\n" + "\n".join(lines) + "\n", encoding="utf-8")
return solution
本地运行¶
以下代码格演示如何调用 OptAgent 的开放车间调度模型。
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_tai44 = main(INSTANCE_DIR / "tai44_0.txt", time_limit=1)