2

我在 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)
4

3 回答 3

4

我尝试了其他方法，重载了 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% 有信心。

于 2017-04-18T14:58:13.050 回答
0

SimPy 相关

也许你可以使用PreemptiveResource（见这个例子）。有了这个，每个资源只需要一个阻塞进程，因为它可以“踢”不太重要的进程。

Python相关

  • 记录您的代码。take_res()和的目的是request_res()什么？（为什么这两个函数都使用priority=-100，无论如何？）
  • 使用更好的名称。the_list或者the_tuple不是很有帮助。
  • 而不是the_list.append(row[0], row[1])你可以做的the_list.append(row[:2])。
  • 为什么要将列表列表转换为元组的元组？据我所知，好处。但它增加了额外的代码，因此增加了额外的混乱和编程错误的可能性。
  • 您应该尽快离开该with open(file)块（在您的情况下，在前四行之后）。没有必要让文件保持打开的时间超过必要的时间，当您完成对所有行的迭代时，您不再需要它。
于 2015-10-26T18:41:27.023 回答
0

这就是我为我的应用程序解决它的方法。这并不完美，但考虑到我对 Python 和 SimPy 的基本技能水平，这是我能做的最好的事情。

结果是在所需时间有正确数量的顾问可用。

首先，我定义了一个存储并将容量设置为等于模拟中将存在的顾问实例的总数。

self.adviser_store = simpy.FilterStore(self.env,
capacity=self.total_ad_instances)

所需的 Adviser 类的实例是在初始化步骤中创建的，为简洁起见，我没有包括在内。我实际上使用 JSON 文件来自定义各个顾问实例，然后将它们放在一个列表中。

下面类定义中的运行参数实际上是另一个类，其中包含与当前模拟运行相关的所有信息 - 例如，它包含模拟的开始和结束日期。self.start_date 因此定义了顾问开始工作的日期。self.run.start_date 是模拟的开始日期。

class Adviser(object):

    def __init__(self, run, id_num, start_time, end_time, start_date,  end_date):

    self.env = run.env
    self.run = run
    self.id_num = id_num        
    self.start_time = start_time
    self.end_time = end_time
    self.start_date = datetime.datetime.strptime(start_date, '%Y, %m, %d')
    self.end_date = datetime.datetime.strptime(end_date, '%Y, %m, %d')
    self.ad_type = ad_type

    self.avail = False
    self.run.env.process(self.set_availability())

如您所见，创建顾问类也启动了设置可用性的过程。在下面的示例中，我将其简化为在给定的日期范围内每天设置相同的可用性。您当然可以根据日期/日期等设置不同的可用性。

def set_availability(self):

    start_delay = self.start_time + (self.start_date - self.run.start_date).total_seconds()/3600 # this returns the time in hours until the resource becomes available and is applied below.
    end_delay = self.end_time + (self.start_date - self.run.start_date).total_seconds()/3600
    repeat = (self.end_date - self.start_date).days + 1  # defines how man days to repeat it for

    for i in range(repeat):

        start_delayed(self.run.env, self.add_to_store(), start_delay)
        start_delayed(self.run.env, self.remove_from_store(), end_delay)
        start_delay += 24
        end_delay += 24

    yield self.run.env.timeout(0)


def add_to_store(self):

    self.run.ad_avail.remove(self)  # take adviser from a list
    self.run.adviser_store.put(self)  # and put it in the store
    yield self.run.env.timeout(0)

def remove_from_store(self):        

    current_ad = yield self.run.adviser_store.get(lambda item: item.id_num == self.id_num)  # get itself from the store 
    self.run.ad_avail.append(current_ad) # and put it back in the list
    yield self.run.env.timeout(0)

因此，基本上客户只能向商店请求顾问，并且顾问只会在特定时间出现在商店中。其余时间它们都在附加到当前模拟运行的列表中。

我认为这里仍然存在一个陷阱。当顾问对象不可用时，它可能正在使用中。我还没有注意到这是否会发生，或者如果发生会产生什么影响。

于 2016-04-21T09:48:16.923 回答