带状态约束的柔性资源受限项目调度问题 (FRCPSPS)¶
类别： 排程
使用 OptAgent 的 Python 接口描述变量、约束与目标。
问题与原始示例来源：Hexaly Code Templates。
数据¶
为了说明此示例,我们生成了随机实例。带状态约束的柔性资源受限项目调度问题 (FRCPSPS) 实例的格式如下:
第一行:
任务数
资源数
状态数
第二行:每种资源的最大容量
从第三行起:
对每对状态 (i, j),给出从状态 i 改变到状态 j 之前的相应等待时间。
从下一行起,对每个任务:
任务的持续时间
任务的状态
建模思路¶
带状态约束的柔性资源受限项目调度问题 (FRCPSPS) 的 OptAgent 模型使用区间决策变量来表示任务,使用集合决策变量来表示每种资源上调度的任务集合。
利用 partition 算子,我们确保每个任务被分配到恰好一种资源。
资源约束可以表述如下:对于每种资源和每个时间槽 t,正在被处理的任务所消耗的资源量不能超过该资源的容量。为了建模这些约束,我们对每种资源和每个时间槽,将处于活跃状态的任务数量累加起来。我们使用可变参 and 形式结合一个 lambda 函数 来确保任何时刻的资源容量均得到满足。这种写法集中表达了各时间段的资源约束；实际求解开销取决于实例规模与求解配置。
最后,我们通过以下方式来表述状态约束:对于分配给同一资源的每对任务,要么任务具有相同的类型,要么它们的执行区间是不相交的,从而防止同一资源上同时执行两个不同类型的任务。此外,对于不同类型的任务,在满足区间不相交的条件时,我们引入状态之间每次转移所对应的等待时间。这确保了如果一个类型为 i 的任务结束,那么一个类型为 j 的任务在从状态 i 到状态 j 的等待时间结束之前不能开始。
需要最小化的最大完工时间(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[0].split()
# Number of tasks
nb_tasks = int(first_line[0])
# Number of resources
nb_resources = int(first_line[1])
# Number of states
nb_states = int(first_line[2])
# Maximum capacity of each resource
capacity = [int(lines[1].split()[r]) for r in range(nb_resources)]
#Delay after change of state
delay_state = [[] for i in range(nb_states)]
for i in range(nb_states):
delay_state[i] = [int(lines[2+i].split()[j]) for j in range(nb_states)]
# Duration of each task
duration = [0 for i in range(nb_tasks)]
# State of each task
state = [0 for i in range(nb_tasks)]
for i in range(nb_tasks):
task_line = lines[i + 2 + nb_states].split()
duration[i] = int(task_line[0])
state[i] = int(task_line[1])
# Trivial upper bound for the end times of the tasks
horizon = sum(duration[i] for i in range(nb_tasks))
return (nb_tasks, nb_resources, nb_states, capacity, delay_state, duration, state , horizon)
def main(instance_file, output_file=None, time_limit=60):
nb_tasks, nb_resources, nb_states, capacity, delay_state, duration, state , horizon = read_instance(instance_file)
model = OptModel()
# Interval decisions: time range of each task.
tasks = [model.interval(0, horizon) for _ in range(nb_tasks)]
# Task duration constraints.
for i in range(nb_tasks):
model.constraint(tasks[i].length() == duration[i])
# Set of tasks done by each resource.
resources_tasks = [model.set(nb_tasks) for r in range(nb_resources)]
resources = model.array(resources_tasks)
# All tasks must be scheduled on exactly one resource.
model.constraint(model.partition(resources))
tasks_array = model.array(tasks)
state_array = model.array(state)
delay_array = model.array(delay_state)
makespan = model.max(*(task.end() for task in tasks))
# Resource capacity must hold at every time slot before the makespan.
for r in range(nb_resources):
capacity_respected = model.lambda_function(
lambda t: model.sum(
resources_tasks[r],
model.lambda_function(lambda i: model.contains(model.at(tasks_array, i), t)),
) <= capacity[r]
)
model.constraint(model.and_(model.range(makespan), capacity_respected))
# Tasks sharing a resource need a compatible order and changeover delay.
for r in range(nb_resources):
state_respected = model.lambda_function(
lambda i: model.and_(
resources_tasks[r],
model.lambda_function(
lambda j: model.or_(
model.at(state_array, i) == model.at(state_array, j),
model.end(model.at(tasks_array, i))
+ model.at(delay_array, model.at(state_array, i), model.at(state_array, j))
<= model.start(model.at(tasks_array, j)),
model.end(model.at(tasks_array, j))
+ model.at(delay_array, model.at(state_array, j), model.at(state_array, i))
<= model.start(model.at(tasks_array, i)),
)
),
)
)
model.constraint(model.and_(resources_tasks[r], state_respected))
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 i in range(nb_tasks):
expressions[f"task_{i}_resource"] = model.find(resources, i)
expressions[f"task_{i}_start"] = tasks[i].start()
expressions[f"task_{i}_end"] = tasks[i].end()
values = {name: expression.value for name, expression in expressions.items()}
# Write the solution in a file with the following format:
# total makespan, number of tasks, then task id/start/end/resource rows.
print(f"Makespan = {values['makespan']}; Status = {solution.feasible}")
if output_file is not None:
lines = [f"{values['makespan']} {nb_tasks}"]
for i in range(nb_tasks):
lines.append(f"{i + 1} {values[f'task_{i}_start']} {values[f'task_{i}_end']} {values[f'task_{i}_resource']}")
Path(output_file).write_text("\n".join(lines) + "\n", encoding="utf-8")
print("Solution written in file", output_file)
return solution
# if __name__ == '__main__':
# import sys
# if len(sys.argv) < 2:
# print("Usage: python frcpsps.py instance_file [output_file] [time_limit]")
# sys.exit(1)
# instance_file = sys.argv[1]
# output_file = sys.argv[2] if len(sys.argv) >= 3 else None
# time_limit = int(sys.argv[3]) if len(sys.argv) >= 4 else 60
# main(instance_file, output_file, time_limit)
INSTANCE_DIR = Path.cwd() / "instances"
solution = main(INSTANCE_DIR / "instance1.txt", time_limit=10)