Python で SimPy を使用して、ユーザーが csv ファイルで入力したスケジュールに基づいてリソースを利用できるようにする必要がある離散イベント シミュレーションを作成しています。目的は、1 日の異なる時間帯に利用可能な同じリソース (スタッフなど) の異なる数を表すことです。私が知る限り、これはベースの 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)