问题描述¶
在烤箱调度问题 (OSP) 中，一组作业必须被分配到烤箱、组合成批次，并在有限的时间范围内调度。每个作业具有释放日期、截止日期、尺寸、处理时间范围以及一个属性。每台烤箱具有初始属性、容量以及可用时间段。
同一批次中的作业必须具有相同的属性。此外，同一烤箱上的连续批次可能需要依赖于其属性的准备时间和准备成本。
目标是最小化四个准则的加权和：使用批次的总处理时间、延迟作业的数量、准备成本以及准备时间。
建模要点¶
- 使用 set 决策变量 建模批次的内容
- 添加 区间决策变量 建模批次
- 使用 lambda 函数 和
step_array建模可用时间段 - 使用 ‘distinct’ 算子 和 lambda 函数 确保同一批次中的所有作业具有相同类型
建模思路¶
烤箱调度问题 (OSP) 的 OptAgent 模型使用 set 和区间决策变量。对每台烤箱，它创建固定数量的潜在批次。批次变量存储在一个扁平 list 中，然后按烤箱和批次位置被看作二维结构，这使得排序、准备和容量约束更容易表达。
每个批次有一个 set 变量表示分配给它的作业，以及一个区间变量表示其处理时间。set 变量构成一个 partition，以确保我们将每个作业分配到恰好一个批次。find 算子用于识别每个作业所属的批次，并将其链接到相应的区间。
对每个批次，模型计算其总尺寸和属性。批次尺寸不得超过烤箱容量，批次中的所有作业必须具有相同的属性，并且每个作业只能分配到一台可用的烤箱。
在每台烤箱上，连续批次必须遵守依赖于序列的准备时间。空批次放在非空批次之后。
烤箱的可用性由 step_array 表示，这些数组根据可用时间段构建。对每个使用的批次，其准备和处理过程都必须适合一个可用的时间段。
目标是使用批次的总处理时间、延迟作业的数量、准备成本以及准备时间的加权和。
Python 实现¶
from pathlib import Path
from optagent import OptModel, solve
def read_integers(filename):
return [
int(x)
for x in Path(filename).read_text(encoding="utf-8").split()
]
#
# Create a step array from a dense array
#
def create_step_array(model, dense_array, horizon):
indexes = []
values = []
value = dense_array[0]
for t in range(1, horizon):
if dense_array[t] != value:
indexes.append(t)
values.append(value)
value = dense_array[t]
indexes.append(horizon)
values.append(value)
return model.step_array(model.array(indexes), model.array(values))
class _OptAgentRunner:
"""Keep the original model block's lifecycle while using OptAgent."""
def __init__(self, time_limit):
self.model = OptModel()
self.time_limit = time_limit
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return False
def solve(self):
return solve(self.model, time_limit_s=float(self.time_limit))
#
# Read instance data from the input file
#
def main(input_file, output_file=None, time_limit=60):
values = iter(read_integers(input_file))
horizon = next(values)
nb_attributes = next(values)
# Attribute 0 is reserved for empty batches
empty_attribute = 0
# Input only provides setup values for real destination attributes;
# transitions to empty batches are set to zero
setup_costs_matrix = [
[0 for _ in range(nb_attributes + 1)] for _ in range(nb_attributes + 1)
]
for a1 in range(nb_attributes + 1):
for a2 in range(nb_attributes + 1):
setup_costs_matrix[a1][a2] = (
0 if a2 == empty_attribute else next(values)
)
setup_times_matrix = [
[0 for _ in range(nb_attributes + 1)] for _ in range(nb_attributes + 1)
]
for a1 in range(nb_attributes + 1):
for a2 in range(nb_attributes + 1):
setup_times_matrix[a1][a2] = (
0 if a2 == empty_attribute else next(values)
)
nb_machines = next(values)
machine_min_capacity = [next(values) for _ in range(nb_machines)]
machine_max_capacity = [next(values) for _ in range(nb_machines)]
machine_initial_attribute = [next(values) for _ in range(nb_machines)]
nb_availabilities = next(values)
availability_start = [
[next(values) for _ in range(nb_availabilities)]
for _ in range(nb_machines)
]
availability_end = [
[next(values) for _ in range(nb_availabilities)]
for _ in range(nb_machines)
]
nb_jobs = next(values)
machine_eligibility = [
[False for _ in range(nb_machines)] for _ in range(nb_jobs)
]
for j in range(nb_jobs):
nb_eligible = next(values)
for _ in range(nb_eligible):
machine_id = next(values)
# The machine IDs in the input start at 1
machine_eligibility[j][machine_id - 1] = True
job_release_date = [next(values) for _ in range(nb_jobs)]
job_due_date = [next(values) for _ in range(nb_jobs)]
job_min_processing_time = [next(values) for _ in range(nb_jobs)]
job_max_processing_time = [next(values) for _ in range(nb_jobs)]
job_size = [next(values) for _ in range(nb_jobs)]
job_attribute = [next(values) for _ in range(nb_jobs)]
upper_bound_integer_objective = next(values)
weight_batch_processing_time = next(values)
weight_tardy_jobs = next(values)
weight_setup_times = next(values)
weight_setup_costs = next(values)
#
# Preprocess instance data
#
# Create availability arrays from availability windows
availabilities_dense = [
[0 for _ in range(horizon)] for _ in range(nb_machines)
]
for m in range(nb_machines):
for a in range(nb_availabilities):
for t in range(availability_start[m][a], availability_end[m][a]):
availabilities_dense[m][t] = 1
# Worst case: one job per batch and per machine
nb_max_batch = nb_jobs
nb_total_batch = nb_machines * nb_max_batch
with _OptAgentRunner(time_limit) as optimizer:
#
# Declare the optimization model
#
model = optimizer.model
setup_costs_array = model.array(setup_costs_matrix)
setup_times_array = model.array(setup_times_matrix)
job_size_array = model.array(job_size)
job_attribute_array = model.array(job_attribute)
job_release_date_array = model.array(job_release_date)
job_due_date_array = model.array(job_due_date)
job_min_processing_time_array = model.array(job_min_processing_time)
job_max_processing_time_array = model.array(job_max_processing_time)
machine_max_capacity_array = model.array(machine_max_capacity)
machine_initial_attribute_array = model.array(machine_initial_attribute)
# Convert availability arrays into step arrays to save space
availabilities = [
create_step_array(model, availabilities_dense[m], horizon)
for m in range(nb_machines)
]
# Set decisions: jobs assigned to each batch
batch_content = [model.set(nb_jobs) for _ in range(nb_total_batch)]
# Interval decisions: time range of each batch
batch_interval = [
model.interval(0, horizon) for _ in range(nb_total_batch)
]
# Each job must be assigned to exactly one batch
model.constraint(model.partition(batch_content))
# View batches by machine and batch position
machine_batch_content = [
[batch_content[m * nb_max_batch + b] for b in range(nb_max_batch)]
for m in range(nb_machines)
]
# View time ranges by machine and batch position
machine_batch_interval = [
[batch_interval[m * nb_max_batch + b] for b in range(nb_max_batch)]
for m in range(nb_machines)
]
# Total size of each batch
size_lambda = model.lambda_function(lambda j: job_size_array[j // 1])
machine_batch_size = [
[
model.sum(machine_batch_content[m][b], size_lambda)
for b in range(nb_max_batch)
]
for m in range(nb_machines)
]
# Identify non-empty batches
machine_batch_used = [
[
model.count(machine_batch_content[m][b]) > 0
for b in range(nb_max_batch)
]
for m in range(nb_machines)
]
# Identify common attribute of each batch
attribute_lambda = model.lambda_function(
lambda j: job_attribute_array[j // 1]
)
machine_batch_type = [
[
model.iif(
machine_batch_used[m][b],
model.min(machine_batch_content[m][b], attribute_lambda),
empty_attribute,
)
for b in range(nb_max_batch)
]
for m in range(nb_machines)
]
# Batch selected for each job
batch_content_array = model.array(batch_content)
job_index = [model.find(batch_content_array, j) for j in range(nb_jobs)]
# Jobs in the same batch share the same time range
batch_interval_array = model.array(batch_interval)
job_interval = [
batch_interval_array[job_index[j]] for j in range(nb_jobs)
]
for j in range(nb_jobs):
# Jobs cannot start before their release date
model.constraint(
model.start(job_interval[j]) >= job_release_date_array[j]
)
# Batch length must satisfy each assigned job
model.constraint(
model.length(job_interval[j])
>= job_min_processing_time_array[j]
)
model.constraint(
model.length(job_interval[j])
<= job_max_processing_time_array[j]
)
for j in range(nb_jobs):
for m in range(nb_machines):
if not machine_eligibility[j][m]:
for b in range(nb_max_batch):
# Jobs can only be assigned to eligible machines
model.constraint(
model.not_(
model.contains(machine_batch_content[m][b], j)
)
)
for b in range(nb_total_batch):
# Each batch must consist of jobs with the same type
model.constraint(
model.count(model.distinct(batch_content[b], attribute_lambda))
<= 1
)
for m in range(nb_machines):
for b in range(nb_max_batch):
# Batch size cannot exceed the machine capacity
model.constraint(
machine_batch_size[m][b] <= machine_max_capacity_array[m]
)
for m in range(nb_machines):
for b in range(1, nb_max_batch):
prev_attribute = machine_batch_type[m][b - 1]
next_attribute = machine_batch_type[m][b]
setup_time = setup_times_array[prev_attribute][next_attribute]
# Non-overlap between batches, with sequence-dependent setup times
model.constraint(
model.end(machine_batch_interval[m][b - 1]) + setup_time
<= model.start(machine_batch_interval[m][b])
)
# Empty batches must be last to keep setups between non-empty batches
model.constraint(
model.or_(
model.not_(machine_batch_used[m][b]),
machine_batch_used[m][b - 1],
)
)
for m in range(nb_machines):
setup_time = setup_times_array[machine_initial_attribute_array[m]][
machine_batch_type[m][0]
]
# Account for the setup time before the first batch for each machine
model.constraint(
model.start(machine_batch_interval[m][0]) >= setup_time
)
# Setup costs between each batch
machine_setup_costs = [None for _ in range(nb_machines)]
for m in range(nb_machines):
machine_setup_costs[m] = model.sum(
[
setup_costs_array
[machine_batch_type[m][b - 1]]
[machine_batch_type[m][b]]
for b in range(1, nb_max_batch)
]
)
# Setup times between each batch
machine_setup_times = [None for _ in range(nb_machines)]
for m in range(nb_machines):
machine_setup_times[m] = model.sum(
[
setup_times_array
[machine_batch_type[m][b - 1]]
[machine_batch_type[m][b]]
for b in range(1, nb_max_batch)
]
)
for m in range(nb_machines):
availability = availabilities[m]
availability_lambda = model.lambda_function(
lambda t: availability[t // 1]
)
for b in range(nb_max_batch):
next_interval = machine_batch_interval[m][b]
prev_attribute = model.iif(
b == 0,
empty_attribute,
machine_batch_type[m][b - 1],
)
next_attribute = machine_batch_type[m][b]
setup_time = setup_times_array[prev_attribute][next_attribute]
# Setup and processing must fit in a single machine availability
model.constraint(
model.min(
model.range(
model.start(next_interval) - setup_time,
model.end(next_interval),
),
availability_lambda,
)
>= 1
)
# Total batch processing time
batch_processing_time = model.sum(
[
machine_batch_used[m][b]
* model.length(machine_batch_interval[m][b])
for m in range(nb_machines)
for b in range(nb_max_batch)
]
)
# Number of tardy jobs
tardy_jobs = model.sum(
[
model.end(job_interval[j]) > job_due_date_array[j]
for j in range(nb_jobs)
]
)
# Total setup costs
setup_costs = model.sum(machine_setup_costs)
# Total setup times
setup_times = model.sum(machine_setup_times)
# Minimize weighted processing time, tardy jobs, and setup costs
objective = (
weight_batch_processing_time * batch_processing_time
+ weight_tardy_jobs * tardy_jobs
+ weight_setup_costs * setup_costs
+ weight_setup_times * setup_times
)
model.minimize(objective)
solution = optimizer.solve()
if not solution.feasible:
print(f"No feasible schedule found; Status = {solution.feasible}")
return solution
write_solution(
output_file,
nb_machines,
nb_max_batch,
machine_batch_used,
machine_batch_interval,
machine_batch_content,
objective,
)
print(f"Objective = {objective.value}; Status = {solution.feasible}")
return solution
#
# Write the solution in a file
#
def write_solution(
output_file,
nb_machines,
nb_max_batch,
machine_batch_used,
machine_batch_interval,
machine_batch_content,
objective,
):
if output_file is None:
return
with open(output_file, "w") as sol_file:
sol_file.write("Objective: %s\n" % objective.value)
sol_file.write("Machine Start End Jobs\n")
for m in range(nb_machines):
for b in range(nb_max_batch):
if not machine_batch_used[m][b].value:
continue
sol_file.write(
"Machine: %d | Start: %d | End: %d | Jobs: "
% (
m + 1,
machine_batch_interval[m][b].value.start(),
machine_batch_interval[m][b].value.end(),
)
)
for j in machine_batch_content[m][b].value:
sol_file.write("%d " % (j + 1))
sol_file.write("\n")
本地运行¶
在 notebook 所在目录执行以下 cell，即可调用一个小型炉窑排程实例。
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_osp_01 = main(INSTANCE_DIR / "01-n10-k2-a2.dat", time_limit=1)