建模思路¶
带机器相关切换时间的柔性作业车间调度问题的 OptAgent 模型使用区间决策变量来建模工序的时间范围,并使用列表决策变量来表示每台机器上调度的任务顺序。
利用 partition 算子,我们确保每个任务被分配到恰好一台机器。对于每道工序,我们利用 contains 算子过滤掉不兼容的机器。使用 find 算子,我们可以获取被选中处理每个任务的机器索引。这使我们能够推导出每道工序的加工时间(取决于所选机器),并相应地对每个区间的长度进行约束。
优先关系约束很容易写出。对于每个作业,其每道工序必须在前一道工序结束之后(加上切换时间)才能开始,其中切换时间取决于为这些工序选择的机器。析取资源约束可以表述如下:对所有 i,在位置 i+1 上加工的任务必须等到在位置 i 上加工的任务结束后才能开始。为了建模这个约束,我们定义一个 lambda 函数 来表达两个相邻活动之间的关系。该函数随后在一个可变参 and_ 算子内对每台机器上加工的所有任务求值。注意,这些 and 表达式中项的数量在搜索过程中是变化的,列表的大小(每台机器上分配的任务数)也随之变化。
目标是最小化最大完工时间(makespan),即所有任务完成的时间。
[1] P. Brandimarte (1993). Routing and scheduling in a flexible job shop by tabu search. Annals of Operations Research 22, 157-183.
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
# Constant for incompatible machines
INFINITE = 1000000
def read_instance(filename):
lines = Path(filename).read_text().splitlines()
first_line = lines[0].split()
# Number of jobs
nb_jobs = int(first_line[0])
# Number of machines
nb_machines = int(first_line[1])
# Number of operations for each job
nb_operations = [int(lines[j + 1].split()[0]) for j in range(nb_jobs)]
# Number of tasks
nb_tasks = sum(nb_operations[j] for j in range(nb_jobs))
# Processing time for each task, for each machine
task_processing_time = [[INFINITE] * nb_machines for _ in range(nb_tasks)]
# For each job, for each operation, the corresponding task id
job_operation_task = [[0] * nb_operations[j] for j in range(nb_jobs)]
task_id = 0
for j in range(nb_jobs):
line = lines[j + 1].split()
field_offset = 0
for o in range(nb_operations[j]):
nb_machines_operation = int(line[field_offset + o + 1])
for i in range(nb_machines_operation):
machine = int(line[field_offset + o + 2 * i + 2]) - 1
time = int(line[field_offset + o + 2 * i + 3])
task_processing_time[task_id][machine] = time
job_operation_task[j][o] = task_id
task_id += 1
field_offset += 2 * nb_machines_operation
# Changeover time between two machines
machine_changeover_time = [list(map(int, lines[nb_jobs + 1 + machine].split())) for machine in range(nb_machines)]
# Trivial upper bound for the end times of the tasks
# This intentionally matches the Hexaly example. Because changeover times are
# omitted from the bound, it may be too tight for some other instances.
max_end = sum(
max(task_processing_time[i][m] for m in range(nb_machines) if task_processing_time[i][m] != INFINITE)
for i in range(nb_tasks)
)
return (
nb_jobs,
nb_machines,
nb_tasks,
task_processing_time,
job_operation_task,
nb_operations,
max_end,
machine_changeover_time,
)
def main(instance_file, output_file=None, time_limit=20):
(
nb_jobs,
nb_machines,
nb_tasks,
task_processing_time_data,
job_operation_task,
nb_operations,
max_end,
machine_changeover_time_data,
) = read_instance(instance_file)
model = OptModel()
# Sequence of tasks assigned to each machine.
jobs_order = [model.list(nb_tasks) for m in range(nb_machines)]
machines = model.array(jobs_order)
model.constraint(model.partition(machines))
# Exclude machines that cannot process a task.
for task in range(nb_tasks):
for machine in range(nb_machines):
if task_processing_time_data[task][machine] == INFINITE:
model.constraint(model.not_(jobs_order[machine].contains(task)))
task_machine = [model.find(machines, task) for task in range(nb_tasks)]
task_processing_time = model.array(task_processing_time_data)
tasks = [model.interval(0, max_end) for _ in range(nb_tasks)]
# A task duration is selected from the row for its assigned machine.
durations = [task_processing_time.at(task, task_machine[task]) for task in range(nb_tasks)]
for task in range(nb_tasks):
model.constraint(tasks[task].length() == durations[task])
task_array = model.array(tasks)
machine_changeover_time = model.array(machine_changeover_time_data)
# Precedence constraints between the operations of a job with machine-dependent changeover times
for job in range(nb_jobs):
for operation in range(nb_operations[job] - 1):
before = job_operation_task[job][operation]
after = job_operation_task[job][operation + 1]
model.constraint(tasks[after].start() >= tasks[before].end() + machine_changeover_time[task_machine[before]][task_machine[after]])
# Consecutive tasks in each machine list must not overlap.
for sequence in jobs_order:
adjacent_positions = model.range(0, sequence.count() - 1)
precedes_next = model.lambda_function(
lambda i: task_array[sequence[i]] < task_array[sequence[i + 1]]
)
model.constraint(model.and_(adjacent_positions, precedes_next))
makespan = model.max([task.end() for task in tasks])
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}
for task in range(nb_tasks):
expressions[f"task_{task}_machine"] = task_machine[task]
expressions[f"task_{task}_start"] = tasks[task].start()
expressions[f"task_{task}_end"] = tasks[task].end()
values = {name: expression.value for name, expression in expressions.items()}
schedule_rows = []
for job in range(nb_jobs):
for operation in range(nb_operations[job]):
task = job_operation_task[job][operation]
schedule_rows.append(
(
job + 1,
operation + 1,
values[f"task_{task}_machine"] + 1,
values[f"task_{task}_start"],
values[f"task_{task}_end"],
)
)
print(f"Makespan = {values['makespan']}; Status = {solution.feasible}")
print("Job\tOperation\tMachine\tStart\tEnd")
for row in schedule_rows:
print("\t".join(map(str, row)))
if output_file is not None:
Path(output_file).write_text(
"\n".join("\t".join(map(str, row)) for row in schedule_rows) + "\n",
encoding="utf-8",
)
return solution
本地运行¶
INSTANCE_DIR = Path.cwd() / "instances"
# Tips: This arithmetic form intentionally mirrors the Hexaly model. Current
# OptAgent scheduling bootstrap cannot use a dynamically indexed lag as
# a first-class precedence, so short runs may find no feasible solution.
solution = main(INSTANCE_DIR / "Mk01.fjsc", time_limit=10)