建模思路¶
柔性作业车间调度问题 (FJSP) 的 OptAgent 模型使用两种决策变量:区间变量和列表变量。一方面,我们使用区间变量来建模工序的时间范围。另一方面,我们使用列表变量来表示每台机器上调度的任务顺序。
利用 partition 算子,我们确保每个任务被分配到恰好一台机器。由于任务和机器之间存在兼容性约束,并非所有分配都是可行的。对于每道工序,我们利用 contains 算子过滤掉不兼容的机器。然后使用 find 算子获取被选中处理每个任务的机器索引,并直接用该索引访问加工时间数组。
优先关系约束很容易写出:对于每个作业,其每道工序必须在前一道工序结束后才开始。析取资源约束可以表述如下:对所有 i,在位置 i+1 上加工的任务必须等到在位置 i 上加工的任务结束后才能开始。为了建模这个约束,我们定义一个 lambda 函数 来表达两个相邻活动之间的关系。该函数随后在一个可变参 and 算子内对每台机器上加工的所有任务求值。注意,这些 and 表达式中项的数量在搜索过程中是变化的,列表的大小(每台机器上分配的任务数)也随之变化。
目标是最小化最大完工时间(makespan),即所有任务完成的时间。
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
# Trivial upper bound for the end times of the tasks
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
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,
) = 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)
# Each task is scheduled on a machine
model.constraint(model.partition(machines))
# Only compatible machines can be selected for 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)))
# For each task, the selected machine
task_machine = [model.find(machines, task) for task in range(nb_tasks)]
task_processing_time = model.array(task_processing_time_data)
# Interval decisions: time range of each task
tasks = [model.interval(0, max_end) for _ in range(nb_tasks)]
# The task duration depends on the selected 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)
# Precedence constraints between the operations of a job
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[before] < tasks[after])
# Disjunctive resource constraints between the tasks on a machine
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"
solution = main(INSTANCE_DIR / "Mk01.fjs", time_limit=5)