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