带资源生产与消耗的项目调度问题¶

类别： 排程

使用 OptAgent 的 Python 接口描述变量、约束与目标。

问题与原始示例来源：Hexaly Code Templates。

问题描述¶

在 Project Scheduling Problem with Production and Consumption of Resources 中，一个项目由一组需要调度的任务组成。每个任务都有一个给定的持续时间，且不能被中断。任务之间存在优先级约束：每个任务必须在其所有后继任务开始之前结束。

该问题还涉及两组资源。一方面，可再生资源被任务占用，并在任务结束时立即释放。另一方面，库存资源由任务消耗和/或产生。每个任务对每种可再生资源都有一个给定的资源需求或权重。可再生资源具有给定的容量：只要它们的累积权重不超过容量，它们就可以同时处理多个任务。任务在开始时消耗一定数量的每种库存资源，并在结束时产生一定数量的每种库存资源。每种库存资源都有一个初始水平。当任务开始并消耗资源时，水平会下降；当任务结束并产生资源时，水平会上升。

目标是找到一个调度方案，使完工时间（makespan）最小：即所有任务处理完成的时间。

学习要点¶

添加区间决策变量来建模任务
定义符号 Lambda 函数来建模可再生资源和库存资源约束

数据¶

我们提供的 Project Scheduling with Production and Consumption of Resources 实例来自 Koné et al。数据文件的格式如下：

第一行：任务数量、可再生资源数量、库存资源数量
第二行：每种可再生资源的最大容量、每种库存资源的初始水平
从第三行开始，对于每个任务：
任务的持续时间
每种资源的可再生资源需求（权重）
每种库存资源在任务开始时的消耗量以及结束时的产生量
后继任务的数量
每个后继任务的 ID

建模思路¶

Project Scheduling Problem with Consumption and Production of Resources 的 OptAgent 模型使用区间决策变量来表示任务的起止时间。每个 interval 的长度等于相应任务的持续时间。

然后我们写出优先级约束：每个任务必须在其任何后继任务开始之前结束。

可再生资源约束可以表述如下：对于每种可再生资源以及每个时间槽 t，正在处理的任务所消耗的资源量不能超过该资源的容量。为了建模这些约束，我们对每种资源和每个时间槽的所有活跃任务的权重进行求和。我们使用可变参数的 and 公式结合符号 Lambda 函数，以确保资源容量在任何时刻都被满足。这种写法集中表达了各时间段的资源约束；实际求解开销取决于实例规模与求解配置。

我们可以类似地对库存资源约束进行建模。对于每种库存资源和每个时间槽 t，初始水平加上在 t 之前结束的任务所产生的资源总量，必须大于等于在 t 之前开始的任务所消耗的资源总量。我们使用另一个可变参数的 and 公式结合符号 Lambda 函数，以确保资源水平在任何时刻都保持非负。

需要最小化的完工时间（makespan）是所有任务结束的时间。

Python 实现¶

from pathlib import Path

from optagent import OptModel, solve


# The input files follow the "Patterson" format
def read_instance(filename):
    with open(filename) as f:
        lines = f.readlines()

    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 inventories
    nb_inventories = int(first_line[2])
    
    second_line = lines[1].split()

    # Maximum capacity of each resource
    capacity = [int(second_line[r]) for r in range(nb_resources)]

    # Initial level of each inventory
    init_level = [int(second_line[r + nb_resources]) for r in range(nb_inventories)]

    # Duration of each task
    duration = [0 for i in range(nb_tasks)]

    # Resource weight of resource r required for task i
    weight = [[] for i in range(nb_tasks)]

    # Inventory consumed at beginning of task i
    start_cons = [[] for i in range(nb_tasks)]

    # Inventory produced at end of task i
    end_prod = [[] for i in range(nb_tasks)]

    # Number of successors
    nb_successors = [0 for i in range(nb_tasks)]

    # Successors of each task i
    successors = [[] for i 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)]
        start_cons[i] = [int(line[2*r + nb_resources + 1]) for r in range(nb_inventories)]
        end_prod[i] = [int(line[2*r + nb_resources + 2]) for r in range(nb_inventories)]
        nb_successors[i] = int(line[2*nb_inventories + nb_resources + 1])
        successors[i] = [int(line[2*nb_inventories + nb_resources + 2 + s]) - 1 for s in range(nb_successors[i])]

    # 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_inventories, capacity, init_level, duration, weight, start_cons, end_prod, nb_successors, successors, horizon)


def main(instance_file, output_file=None, time_limit=60):
    nb_tasks, nb_resources, nb_inventories, capacity, init_level, duration, weight, start_cons, end_prod, nb_successors, successors, horizon = read_instance(
        instance_file)

    model = OptModel()

    # Each task is a non-preemptive interval in the trivial horizon.
    tasks = [model.interval(0, horizon) for i in range(nb_tasks)]
    for i in range(nb_tasks):
        model.constraint(tasks[i].length() == duration[i])

    # A task must finish before each of its successors starts.
    for i in range(nb_tasks):
        for successor in successors[i]:
            model.constraint(tasks[i] < tasks[successor])

    makespan = model.max(task.end() for task in tasks)

    # Renewable capacity is checked at every integer time in the makespan.
    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))

    # Inventory consumption happens at task start; production at task end.
    for r in range(nb_inventories):
        inventory_non_negative = model.lambda_function(
            lambda t: init_level[r]
            + model.sum(
                end_prod[i][r] * (tasks[i].end() <= t)
                - start_cons[i][r] * (tasks[i].start() <= t)
                for i in range(nb_tasks)
            )
            >= 0
        )
        model.constraint(model.and_(model.range(makespan + 1), inventory_non_negative))

    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

    lines = [f"Makespan = {makespan.value}; Status = {solution.feasible}"]
    for i, task in enumerate(tasks):
        lines.append(f"{i + 1} {task.start().value} {task.end().value}")
    result_text = "\n".join(lines)
    print(result_text)
    if output_file is not None:
        Path(output_file).write_text(result_text + "\n", encoding="utf-8")
    return solution

本地运行¶

下面的代码格使用仓库中提供的 Patterson 实例调用 main。可以按需更换实例或调整求解时间。

INSTANCE_DIR = Path.cwd() / "instances"
print("Instances:", INSTANCE_DIR)

solution_consprod = main(
    INSTANCE_DIR / "ConsProd_bl2005.rcp",
    time_limit=1,
)