资源受限项目调度程序RCPSP

问题介绍

资源受限项目调度问题（Resource-Constrained Project Scheduling Problem，RCPSP）是一个经典的优化问题，涉及在有限资源的情况下安排项目任务，以最大化某种指标，比如项目完成时间、资源利用率或成本最小化等。在许多实际应用中，资源受限是常见的，例如在制造业、建筑业、信息技术和项目管理等领域。

在资源受限项目调度问题中，通常会给定以下几个方面的限制和条件：

1. 任务： 项目被分解为一系列可执行的任务，每个任务都有一个开始时间和结束时间。
2. 资源：项目所需的资源包括人力、设备、资金等，这些资源是有限的。
3. 约束： 每个任务对资源的需求是不同的，同时存在任务之间的先后顺序和依赖关系。
4. 优化目标： 最常见的优化目标是最小化项目完成时间或最大化资源利用率，但也可能涉及其他目标，比如最小化成本或最大化利润等。

资源受限项目调度问题是一个NP-难问题，因此没有多项式时间的解法。解决该问题的方法通常包括：

1. 启发式算法： 基于经验或直觉设计的算法，如遗传算法、模拟退火等。
2. 精确算法： 尝试找到最优解的算法，如动态规划、分支定界等。但这些算法在大规模问题上的效率通常较低。 混合方法。
3. 结合启发式算法和精确算法，以在可接受的时间内找到较好的解决方案。

资源受限项目调度问题在许多实际应用中都有广泛的应用，包括但不限于：

1. 生产制造： 在生产线上安排任务以最大化产量并最小化成本。
2. 建筑业： 安排施工工序和资源以优化工程进度和资源利用率。
3. 信息技术：安排软件开发项目中的任务和团队资源。
4. 项目管理： 规划和安排复杂项目中的任务和资源分配。

示例

某个项目包含9个活动，活动间先后关系如图所示：

各活动工期如下：

活动	活动名称	活动持续时间	资源需求量
1	活动1	2	3
2	活动2	5	5
3	活动3	7	8
4	活动4	6	10
5	活动5	5	6
6	活动6	4	3
7	活动7	2	3
8	活动8	4	3
9	活动9	7	5

启发规则求解

采用最短工期活动最先开始调度规则，生成上述问题的结果如下：

# %%
"""
Author: TUUG
Date: April 26, 2024 20:59
Description: 求解资源受限项目调度问题并画图.
"""
import numpy as np
import pandas as pd
from collections import defaultdict
import matplotlib.pyplot as plt
import random
import math

plt.rcParams["axes.labelsize"]=14
plt.rcParams["xtick.labelsize"]=12
plt.rcParams["ytick.labelsize"]=12
plt.rcParams['font.sans-serif'] = ['SimHei']  # 指定默认字体
plt.rcParams['axes.unicode_minus'] = False  # 解决保存图像是负号'-'显示为方块的问题

# %%
class Activity(object):
    '''
    活动类：包含  1.活动ID  2.活动持续时间    3.活动资源需求量   4.活动紧前活动    5.活动最早开始时间  6.活动最晚开始时间  7.活动是否被访问
    '''
    def __init__(self, id, duration, resourceRequest, successor):
        self.id = id
        self.time_long = duration  # 活动总时长，固定不变
        self.duration = duration  # 活动剩余时长，动态变化
        self.resourceRequest = np.array(resourceRequest)[0]
        self.predecessor = None
        self.successor = successor
        self.visited = False
        self.start = None  # 这里的start是时点数据，不是时段数据
        self.end = None
        
class Day:
    def __init__(self,id,total_resource):
        self.id = id
        self.total_resource = total_resource
        self.resource_used = 0
        self.resource_left = total_resource
        self.act_list = []
        
    #================修改代码=====================
def start_act(act,day,days):
    """启动活动"""
    if act.visited == False:
        act.visited = True
        act.start = day.id-1
        act.end = act.start + act.time_long
        # 如果活动时长大于0
        for i in range(act.time_long):
            execute_act(act,days[days.index(day)+i])  # 这里替换为执行活动函数
        if act.time_long == 0:
            execute_act(act,day)
    
def execute_act(act,day):
    """执行活动"""
    act.duration -= 1
    day.resource_used += act.resourceRequest
    day.resource_left -= act.resourceRequest
    day.act_list.append(act.id)
        
def can_start1(act,day,days):
    # 判断一个活动是否能开始，条件1：资源是否足够
    a = days.index(day)
    for j in range(act.time_long):
        if days[min(a+j,len(days)-1)].resource_left < act.resourceRequest:
            return False
    return True
    
def can_start2(act,act_done):
    # 判断一个活动是否能开始，条件2：该活动的前序活动是否已经完成
    for preAct in act.predecessor:
        if preAct not in act_done:
            return False
    return True

def pre_done(act,act_done,activities):
    # 判断活动的所有前序活动是否都已完成，返回0则说明都已经完成
    ans = [0 if activities[i] in act_done else 1 for i in act.predecessor]
    return sum(ans) 
    #================修改代码=====================

# %%
def read_data_from_RCP_file(file_name):
    '''
    读取标准化文件中的所有活动信息，包括  1.活动数   2.项目资源数 3.项目资源种类数   4.项目资源限量
    5.所有活动的ID，持续时间，资源需求，紧前活动
    :param fileName:
    :return: 标准化文件数据
    '''
    f = open(file_name)
    taskAndResourceType = f.readline().split('      ')  # 第一行数据包含活动数和资源数
    num_activities = int(taskAndResourceType[0])  # 得到活动数
    num_resource_type = int(taskAndResourceType[1])  # 得到资源数
    total_resource = np.array([int(value) for value in f.readline().split('      ')[:-1]])  # 获取资源限量
    # 将每个活动的所有信息存入到对应的Activity对象中去
    activities = {}
    preActDict = defaultdict(lambda: [])
    for i in range(num_activities):
        nextLine = [int(value) for value in f.readline().split('      ')[:-1]]
        # task = Activity(i + 1, nextLine[0], nextLine[1:5], nextLine[6:])
        task = Activity(i + 1, nextLine[0], nextLine[1:2], nextLine[3:])
        activities[task.id] = task
        # for act in nextLine[6:]:
        for act in nextLine[3:]:
            preActDict[act].append(i + 1)
    f.close()
    # 给每个活动加上紧前活动信息
    for actKey in activities.keys():
        activities[actKey].predecessor = preActDict[activities[actKey].id].copy()
    return num_activities, num_resource_type, total_resource, activities  

file_name = r'D:/python代码库/实验/mv9.rcp'
num_activities, num_resource_type, total_resource, activities = read_data_from_RCP_file(file_name)
total_resource = total_resource[0]


# %%
num_days = sum([i.duration for i in activities.values()]) # 总天数自动计算，不再需要手动指定
# num_days = 150
days = [Day(i,total_resource) for i in range(1,num_days+1)]
act_list = [i for i in activities.values()]
act_done = []
act_candidate = [activities[1]]

# %%
# ----------调度核心逻辑-------
for day in days:
    # 首先考虑工期为0的活动
    for act in act_candidate:
        if act.time_long == 0:
            act.start = day.id-1
            act.end = day.id-1
            act_done.append(act)
            act_candidate.remove(act)
            # act_candidate.append(act.successor)
            for i in act.successor:
                act_candidate.append(activities[i])
    
    # 遍历act_candidate列表,将能开启的工作全部启动
    for act in act_candidate:
        if can_start1(act,day,days):
            start_act(act,day,days)
    # -------更新act_done列表-----
    for act in act_list:
        if act.end:
            if act.end <= day.id and act not in act_done:
                act_done.append(act)
    # 更新act_candidate列表
    for act in act_candidate:
        for suc in act.successor:
            if pre_done(activities[suc],act_done,activities) == 0:
                act_candidate.append(activities[suc])       
    for act in act_candidate:
        if act in act_done:
            act_candidate.remove(act)

# %%
def plot_square(matrix):
    # plt.figure(figsize=(8, 8))
    fig, ax = plt.subplots()
    cmap = plt.get_cmap('viridis')  # 使用 'viridis' colormap，你可以根据需要选择其他colormap
    norm = plt.Normalize(vmin=0, vmax=matrix.max())  # 指定归一化范围
    for i in range(matrix.shape[0]):
        for j in range(matrix.shape[1]):
            color = cmap(norm(matrix[i, j]))
            if matrix[i, j] == 0:
                color = 'white'
            square = plt.Rectangle((j, total_resource-1-i), 1, 1, fill=True, color=color, edgecolor='black')
            ax.add_patch(square)
            if matrix[i, j] != 0:
                # 在正方形中心位置添加数字
                plt.text(j + 0.5, total_resource-1-i + 0.5, str(matrix[i, j]), color='black',fontsize=12, ha='center', va='center')
            
    ax.set_xlim(0, matrix.shape[1])
    ax.set_ylim(0, matrix.shape[0])
    ax.set_aspect('equal', adjustable='box')
    ax.set_xlabel('日期')
    ax.set_ylabel('资源')
    plt.title('项目调度图')
    plt.show()

# %%
ans = [days[i].act_list for i in range(num_days)]
ans.remove([])
b = {}
for i in range(len(activities)):
    b[i+1] = activities[i+1].resourceRequest
def copy(sub_ans,b):
    result = [item for item in sub_ans for _ in range(b.get(item, 1))]
    return result
ans_new = [[]] * len(ans)
for i in range(len(ans)):
    ans_new[i] = copy(ans[i],b)
# print(ans_new)
# print('--------------各活动开始时间----------------')
start_time = []
for i in range(1,len(activities)+1):
    # print('活动'+str(i)+'开始于'+str(activities[i].start))
    start_time.append(activities[i].start)
# print(start_time)
# print('-------------活动总工期----------------')
total_time = 0
for i in range(1,len(activities)+1):
    if activities[i].successor == []:
        total_time = activities[i].end+1
# print(total_time)
# 转成矩阵
a = np.zeros((total_resource,len(ans_new)),dtype=int)
for i in range(len(ans_new)):
    for j in range(len(ans_new[i])):
        a[total_resource-1-j,i] = int(ans_new[i][j])

plot_square(a)
# print(a)

求解结果

采用上述代码求解的项目调度图如图所示：

采用最短时间活动最先开始规则生成图如下：