這次文章是關于如何用 SimPy 來解決兩個仿真需求:

如何隨時中斷恢復 Process (進程)

如何動態設置 Resource (資源)的數量

相應地這兩個需求滿足的場景是:

仿真過程中, 某一工序被中斷, 中斷可以依據一個預先設定的時間或者是不確定時間

仿真過程中, 人力資源也是依據時間變化, 模擬現實中工人的排班安排

回顧資源和進程的概念

Resource 和 Process 是 SimPy 對人力資源和進程進行抽象的構造. Resource 好比一個隊列, 其長度就是提前設置好的資源數, 不同的工序就按照時間先后和賦予的優先級進入隊列. Process 從構造上來說就是一個生成器, 我們可以通過 send 方法傳入 Exception 對 Process 進行打斷.

比如某個工序需要占用一個工人, 耗時 30 min 來完成一個進程, 當前所有可以調用的工人數是 10, 代碼形式如下:

import simpy

import random

WORKERNUM = 10 # 工人數

PROCESS_TIME = 30 * 60 # 工序耗時, 使用秒作為單位

MEAN_ = 4 * 60 # 平均物件生成時間

def process(env, workers, store):

"""工序"""

while True:

with workers.request() as req:

yield req

item = yield store.get()

print(f"{env.now} - {item} - start")

yield env.timeout(PROCESS_TIME)

print(f"{env.now} - {item} - done")

def put_item(env, store):

"""每隔一段時間生成一個物品"""

for i in range(100):

item = f"{i}_item"

store.put(item)

yield env.timeout(random.expovariate(1 / MEAN_))

env = simpy.Environment()

workers = simpy.Resource(env, 10)

store = simpy.Store(env)

env.process(process(env, workers, store))

env.process(put_item(env, store))

env.run()

Process 進程的動態調整

存在以下兩種情景:

進程隨時中斷以及恢復

按照時間表對進程進行啟動或者終止

要區分一件事情, 中斷的時候是讓當前進程完成后再中斷, 還是立即中斷. 具體場景可以想象為一個工人被調離當前崗位, 他應該是先完成手頭上的工序, 或者他需要停下手頭的工作離開工位.

如果是必須實現進程的隨時中斷, 只能通過 process.interrupt() 中斷 process, 即第一種場景; 假若中斷是按照時間表進行, 就可以通過第二種場景, 構建多個不同時間開啟的進程來進行模擬.

進程中斷的實現

from simpy import interrupt, Environment

env = Environment()

def interrupter(env, victim_proc):

yield env.timeout(1)

victim_proc.interrupt('Spam')

def victim(env):

try:

yield env.timeout(10)

except Interrupt as interrupt:

cause = interrupt.cause

多段進程模擬按時間安排的開關

import simpy

PROCESS_TIME = 3

def put_item(env, store):

for i in range(20):

yield env.timeout(0.5)

store.put(f"{i}_item")

def process(i, env, store, start, end):

yield env.timeout(start)

while True:

item = yield store.get()

# 判斷 item 到達時間是否超出本進程關閉時間

if env.now > end:

print(f"{env.now} - process {i} - end")

store.put(item)

env.exit()

else:

print(f"{env.now} - {item} - start")

yield env.timeout(PROCESS_TIME)

print(f"{env.now} - {item} - end")

env = simpy.Environment()

store = simpy.Store(env)

env.process(put_item(env, store))

for i, (start, end) in enumerate([(20, 30), (40, 50), (60, 90)]):

env.process(process(i, env, store, start, end))

env.run()

Resource 資源的動態調整

資源人數按指定的排版表調配

由于Resource 在實例化后, 就沒辦法修改了. 為了滿足在仿真過程中對資源進行修改, 使用了一個反向的思路. 首先所有資源使用 PriorityResource 實例, 預先設置一個可以調節的最大資源數, 當需要調節資源數的時候, 使用一個優先級為 -1 的 request 去占用資源, 而正常的進程默認優先級是 0.

通過這樣的操作會使得, 我們調節資源的占用進程優先級更高, 正常進程可以調用的資源數會變成

:

可以調用資源 = 最大資源 - 占用資源

import simpy

PROCESS_TIME = 2

def put_item(env, store):

for i in range(20):

yield env.timeout(0.5)

store.put(f"{i}_item")

def process(env, store, resource):

while True:

item = yield store.get()

with resource.request() as req:

yield req

yield env.timeout(PROCESS_TIME)

def set_resource(env, resource, start_time, end_time):

"""占用資源，模擬資源減少的情況，

end_time 會出現 np.inf 無窮大，

simpy 只會用作為排序，可以放在timeout事件里。

"""

duration = end_time - start_time

yield env.timeout(start_time)

with resource.request(priority=-1) as req:

yield req

yield env.timeout(duration)

env = simpy.Environment()

store = simpy.Store(env)

res = simpy.PriorityResource(env, 10)

res_time_table = [(10, 20, 5), (20, 30, 6)]

env.process(put_item(env, store))

env.process(process(env, store, res))

for start, end, target_num in res_time_table:

place_holder = 10 - target_num

for _ in range(place_holder):

env.process(set_resource(env, res, start, end))

env.run()

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的python 仿真_Python SimPy 仿真系列 (2)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。