python - 简单的资源可用性计划

Question

我在 Python 中使用 SimPy 来创建一个离散事件模拟，该模拟需要根据用户在 csv 文件中输入的计划提供资源。目的是表示一天中不同时间可用的相同资源(例如员工)的不同数量。据我所知，这不是基本 SimPy 中可用的东西——比如资源优先级。

我已经设法让它工作，并包含下面的代码来展示如何。但是我想问问社区是否有更好的方法在 SimPy 中实现这个功能？

下面的代码的工作原理是在每天开始时在资源不应该可用的时间请求资源 - 具有更高的优先级以确保他们获得资源。然后在适当的时间释放资源以供其他事件/进程使用。正如我所说，它有效但似乎很浪费大量虚拟进程来确保资源的正确真实可用性。欢迎任何有助于改进的意见。

所以 csv 看起来像:

Number  time
0        23
50       22
100      17
50       10
20       8
5        6

其中 Number 表示在定义的时间可用的员工数量。例如:6 点到 8 点有 5 名员工，8 点到 10 点有 20 名员工，10 点到 17 点有 50 名员工，依此类推，直到一天结束。

代码:

import csv
import simpy

# empty list ready to hold the input data in the csv
input_list = []

# a dummy process that "uses" staff until the end of the current day
def take_res():
    req = staff.request(priority=-100)
    yield req  # Request a staff resource at set priority
    yield test_env.timeout(24 - test_env.now)

# A dummy process that "uses" staff for the time those staff should not 
# be available for the real processes     
def request_res(delay, avail_time):
    req = staff.request(priority=-100)
    yield req  # Request a staff resource at set priority
    yield test_env.timeout(delay)
    yield staff.release(req)
    # pass time it is avail for
    yield test_env.timeout(avail_time)
    test_env.process(take_res())

# used to print current levels of resource usage
def print_usage():
    print('At time %0.2f %d res are in use' % (test_env.now, staff.count))
    yield test_env.timeout(0.5)
    test_env.process(print_usage())

# used to open the csv and read the data into a list
with open('staff_schedule.csv', mode="r") as infile:
    reader = csv.reader(infile)
    next(reader, None)  # ignore header
    for row in reader:
        input_list.append(row[:2])

    # calculates the time the current number of resources will be 
    # available for and adds to the list 

    i = 0
    for row in the_list:
        if i == 0:
            row.append(24 - int(input_list[i][1]))
        else:
            row.append(int(input_list[i-1][1]) - int(input_list[i][1]))

        i += 1

    # converts list to tuple of tuples to prevent any accidental 
    # edits from this point in
    staff_tuple = tuple(tuple(row) for row in input_list)
    print(staff_tuple)

# define environment and creates resources   
test_env = simpy.Environment()
staff = simpy.PriorityResource(test_env, capacity=sum(int(l[0]) for l in staff_tuple))

# for each row in the tuple run dummy processes to hold resources 
# according to schedule in the csv
for item in the_tuple:
    print(item[0])
    for i in range(int(item[0])):
        test_env.process(request_res(int(item[1]), int(item[2])))

# run event to print usage over time
test_env.process(print_usage())

# run for 25 hours - so 1 day
test_env.run(until=25)

Answer 1

我尝试了其他方法，我重载了 Resource 类，只添加了一个方法，虽然我不完全理解源代码，但它似乎可以正常工作。您可以告诉资源更改模拟中某处的容量。

from simpy.resources.resource import Resource, Request, Release
from simpy.core import BoundClass
from simpy.resources.base import BaseResource

class VariableResource(BaseResource):
    def __init__(self, env, capacity):
        super(VariableResource, self).__init__(env, capacity)
        self.users = []
        self.queue = self.put_queue

    @property
    def count(self):
        return len(self.users)

    request = BoundClass(Request)
    release = BoundClass(Release)

    def _do_put(self, event):
        if len(self.users) < self.capacity:
            self.users.append(event)
            event.usage_since = self._env.now
            event.succeed()

    def _do_get(self, event):
        try:
            self.users.remove(event.request)
        except ValueError:
            pass
        event.succeed()

    def _change_capacity(self, capacity):
        self._capacity = capacity

我认为这应该可行，但我对触发器的工作方式不是 100% 有信心。