问题描述¶
在 Resource Availability Cost Problem (RACP) 中，一个项目由一组需要调度的任务组成。每个任务都有一个给定的持续时间，且不能被中断。任务之间存在优先级约束：每个任务必须在其所有后继任务开始之前结束。问题涉及一组可再生资源。每个任务对每种资源都有一个给定的资源需求或权重（可能为零），表示该任务在执行过程中消耗的资源量。每种资源都有一个需要设定的最大容量。每单位容量都有一个给定的成本，对每种资源有所不同。被处理任务的权重之和不能超过该最大容量。目标是在确保所有任务在给定截止日期之前完成的前提下，找到一个使所需容量的总成本最小的调度方案。
学习要点¶
- 添加 区间决策变量 来建模任务
- 添加 整数决策变量 来建模容量
- 定义 符号 Lambda 函数 来建模累积资源约束
数据¶
我们提供的 Resource Availability Cost Problem (RACP) 实例来自 RACP instances，并遵循 Patterson 格式：
第一行：
任务数量（包括两个额外的持续时间为 0 的虚拟任务：源和汇）
可再生资源的数量
第二行：每种资源的单位容量成本
从第三行开始，对于每个任务：
任务的持续时间
每种资源的资源需求（权重）
后继任务的数量
每个后继任务的 ID
建模思路¶
Resource Availability Cost Problem (RACP) 的 OptAgent 模型使用 区间决策变量 来表示任务。每个 interval 的长度等于相应任务的持续时间。我们还定义了 整数决策变量 来表示分配给每种资源的容量。然后我们写出优先级约束：每个任务必须在其任何后继任务开始之前结束。完工时间（makespan）是所有任务完成的时间，它必须保持小于截止日期。
累积资源约束可以表述如下：对于每种资源以及每个时间槽 t，正在处理的任务所消耗的资源量不能超过该资源的容量。为了建模这些约束，我们对每种资源和每个时间槽的所有活跃任务的权重进行求和。我们使用可变参数的 and 公式结合 符号 Lambda 函数，以确保资源容量在任何时刻都被满足。这种写法集中表达了各时间段的资源约束；实际求解开销取决于实例规模与求解配置。
最后，我们将总成本计算为每种资源的单位成本与其所选容量的乘积之和。这就是我们希望最小化的目标。
Python 实现¶
# Copyright (c) Hexaly. Permission is hereby granted to use, copy,
# and modify this code for applications developed with Hexaly.
from pathlib import Path
from optagent import OptModel, solve
# The input files follow the "Patterson" format
def read_instance(filename):
lines = Path(filename).read_text(encoding="utf-8").splitlines()
first_line = lines[0].split()
nb_tasks = int(first_line[0])
nb_resources = int(first_line[1])
resource_cost = [int(lines[1].split()[r]) for r in range(nb_resources)]
duration = [0 for _ in range(nb_tasks)]
weight = [[] for _ in range(nb_tasks)]
nb_successors = [0 for _ in range(nb_tasks)]
successors = [[] for _ in range(nb_tasks)]
for i in range(nb_tasks):
line = lines[i + 2].split()
duration[i] = int(line[0])
weight[i] = [int(line[r + 1]) for r in range(nb_resources)]
nb_successors[i] = int(line[nb_resources + 1])
successors[i] = [int(line[nb_resources + 2 + s]) - 1 for s in range(nb_successors[i])]
horizon = sum(duration)
return (nb_tasks, nb_resources, resource_cost, duration, weight, nb_successors, successors, horizon)
def main(instance_file, deadline=None, output_file=None, time_limit=60):
nb_tasks, nb_resources, resource_cost, duration, weight, nb_successors, successors, horizon = read_instance(instance_file)
deadline = horizon if deadline is None else int(deadline)
model = OptModel()
tasks = [model.interval(0, horizon) for i in range(nb_tasks)]
capacity = [
model.int(0, sum(weight[i][r] for i in range(nb_tasks)))
for r in range(nb_resources)
]
for i in range(nb_tasks):
model.constraint(tasks[i].length() == duration[i])
for i in range(nb_tasks):
for successor in successors[i]:
model.constraint(tasks[i].end() <= tasks[successor].start())
makespan = model.max([task.end() for task in tasks])
model.constraint(makespan <= deadline)
for r in range(nb_resources):
capacity_respected = model.lambda_function(
lambda t: model.sum(weight[i][r] * model.contains(tasks[i], t // 1) for i in range(nb_tasks)) <= capacity[r]
)
model.constraint(model.and_(model.range(makespan), capacity_respected))
total_cost = model.sum(resource_cost[r] * capacity[r] for r in range(nb_resources))
model.minimize(total_cost)
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"Total cost = {total_cost.value}; Makespan = {makespan.value}; Status = {solution.feasible}")
if output_file is not None:
lines = [str(total_cost.value), str(makespan.value), " ".join(str(item.value) for item in capacity)]
lines.extend(f"{i + 1} {tasks[i].value.start()} {tasks[i].value.end()}" for i in range(nb_tasks))
Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
return solution
实例调用¶
下面使用仓库提供的 RACP 实例运行模型。
INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)
solution_racp = main(INSTANCE_DIR / "racp1.rcp", time_limit=1)
solution_racp.feasible