问题描述¶
在带强度参数的作业车间调度问题中，一组作业必须在车间中的每台机器上进行处理。每个作业由有序的任务序列（称为活动）组成，每个活动代表该作业在一台机器上的处理。每个作业在每台机器上都有一个活动，并且必须在前一个活动完成后才能开始下一个活动。每个活动都有一个给定的处理时间，并且每台机器一次只能处理一个活动。机器处理活动的强度随时间变化。具体而言，强度为 0 意味着机器处于关闭状态。在每个时间步，每个正在进行活动的进度按其机器的当前强度递增。在时间 t 开始的活动被视为已完成，当自时间 t 起其机器的强度之和达到其处理时间时即为完成。
目标是找到一个使 makespan（即所有作业处理完成的时间）最小化的作业序列。
学习要点¶
- 添加 区间决策变量 来建模活动
- 添加 列表决策变量 来建模每台机器上活动的顺序
- 定义 符号 Lambda 函数 根据机器的强度来约束 interval 的长度
数据¶
我们提供来自 FT and LA datasets 的作业车间调度问题实例，并附带额外的强度矩阵。数据文件的格式如下：
- 第一行：作业数、机器数和时间跨度
- 对每个作业：在每台机器上的处理时间（按处理顺序给出）
- 对每个作业：处理顺序（访问机器的有序列表）
- 对每台机器：其在每个时间步的强度。
建模思路¶
带强度参数的作业车间调度问题的 OptAgent 模型与 Job Shop 调度问题的模型非常相似。原始的决策变量保持不变。一方面，我们使用 区间决策变量 来建模活动的时间区间。另一方面，我们使用 列表决策变量 来表示调度在机器上的活动的顺序。
强度约束定义了活动时间区间与机器随时间变化的强度之间的关系：活动持续时间内强度之和必须大于其处理时间。为了对这些约束建模，我们在 'sum' 算子中使用一个 符号 Lambda 函数。
紧前和析取资源约束的建模方式与作业车间调度问题相同。需要最小化的 makespan 是所有作业被处理完成的时间。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
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])
# Time horizon: number of time steps
time_horizon = int(first_line[2])
# 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][i] 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)
]
# Intensity for each machine for each time step
intensity = [
[int(lines[i].split()[j]) for j in range(time_horizon)]
for i in range(5 + 2 * nb_jobs, 5 + 2 * nb_jobs + nb_machines)
]
return nb_jobs, nb_machines, time_horizon, processing_time, machine_order, intensity
def main(instance_file, output_file=None, time_limit=10):
(
nb_jobs,
nb_machines,
time_horizon,
processing_time,
machine_order,
intensity,
) = read_instance(instance_file)
model = OptModel()
tasks = [[model.interval(0, time_horizon) for _ in range(nb_machines)] for _ in range(nb_jobs)]
task_array = model.array([model.array(job_tasks) for job_tasks in tasks])
intensity_array = model.array(intensity)
# Sum the machine intensity over every integer time point in the task.
# The dynamic [start, end) range is the OptAgent equivalent of applying
# the original lambda aggregate directly to an interval.
for machine in range(nb_machines):
intensity_lambda = model.lambda_function(lambda time: intensity_array[machine, time])
for job in range(nb_jobs):
active_times = model.range(tasks[job][machine].start(), tasks[job][machine].end())
model.constraint(
model.sum(active_times, intensity_lambda) >= processing_time[job][machine],
)
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],
)
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,
)
precedes_next = model.lambda_function(
lambda i: task_array[sequence[i], machine] < task_array[sequence[i + 1], machine]
)
# Keep the original list-indexed disjunctive formulation. Current
# OptAgent scheduling bootstrap does not recognize this or the
# intensity aggregate as first-class scheduling declarations.
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
values = {'makespan': makespan.value, **{f'machine_{machine}': jobs_order[machine].value for machine in range(nb_machines)}}
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 / "i01_ft06.txt", time_limit=5)