[Pyhton] SimPy 離散事件模擬框架詳解 —— 以一個簡單的汽車充電排隊模擬為例

一、背景知識
二、SimPy 講解
三、后續(xù)
參考鏈接
附錄二

一、背景知識

人們在生產(chǎn)活動和社會活動中，經(jīng)常遇到一類復(fù)雜的系統(tǒng)，這類系統(tǒng)中有許多事件時而出現(xiàn)，時而消失，時而動作，時而停止，而啟動和停止都發(fā)生在一些離散的時刻，并帶有一定的隨機性。例如，港口中船舶的停靠碼頭、生產(chǎn)線上機床的啟停、電話的接通和斷開、計算機系統(tǒng)中某項作業(yè)的進(jìn)行和退出，凡此種種，都帶有上述特點，這類系統(tǒng)叫做離散事件動態(tài)系統(tǒng)(DEDS)。隨著生產(chǎn)和科技的日益發(fā)展，以及人類社會交往的日趨頻繁，這類系統(tǒng)的數(shù)量和種類也越來越多。

設(shè)計此類系統(tǒng)時，往往需要仿真來評估算法或方案的性能，常見的軟件有：

商業(yè)軟件：

名字	簡介	圖示
AnyLogic	通用多方法建模工具。結(jié)合了基于代理、系統(tǒng)動力學(xué)和離散事件建模。
Arena	一種離散事件模擬程序，也允許對連續(xù)過程進(jìn)行建模。
Care pathway simulator	專門為服務(wù)行業(yè)（如醫(yī)療保健）設(shè)計的離散事件模擬程序。
Enterprise Dynamics	一個模擬軟件平臺，用于模擬和分析幾乎任何制造、材料處理和物流挑戰(zhàn)。
ExtendSim	通用仿真軟件包
DELMIA	3DEXPERIENCE 平臺的一部分
FlexSim	拖拖拉拉做離散事件模擬，3D
GoldSim	將動態(tài)離散事件模擬嵌入到 Monte Carlo 框架
GPSS	離散事件模擬語言。供應(yīng)商可以提供不同的實現(xiàn)
Micro Saint Sharp	通用離散事件建模工具，使用拖放界面和C#編程語言
MS4 Modeling Environment	基于離散事件和混合模型的通用DEVS方法的軟件環(huán)境
Plant Simulation	能夠模擬和優(yōu)化生產(chǎn)系統(tǒng)和流程的軟件
ProModel AutoCAD? Edition	在數(shù)字孿生Autodesk?工具集（AutoCAD?和Inventor?）中快速構(gòu)建空間精確的布局和過程仿真模型。
Simcad Pro	實時變化，零代碼有界面，支持 VR
SimEvents	向MATLAB / Simulink環(huán)境添加離散事件仿真。
SIMUL8	基于對象的仿真軟件
VisualSim	基于時序、功耗和功能的電子、嵌入式軟件和半導(dǎo)體的基于模型的系統(tǒng)架構(gòu)探索
WITNESS	可在桌面和云端使用 VR 進(jìn)行離散事件模擬

開源軟件：

名字	語言	類型	License	簡介
JaamSim	Java	App	Apasche 2.0	JaamSim是一款自由開源的離散事件模擬軟件，包括拖放式用戶界面、交互式3D圖形、輸入和輸出處理以及模型開發(fā)工具和編輯器。
CPN Tools	BETA	App	GPLv2	用于分析所有類型應(yīng)用中的物流/排隊模型的工具。
DESMO-J	Java	Lib	Apasche 2.0	Java離散事件模擬框架，支持混合事件/過程模型，并提供2D和3D動畫。
Facsimile	Scala	Lib	LGPLv3	離散事件模擬/仿真庫。
PowerDEVS	C++	App	AFL, GPLv2	基于DEVS形式的混合系統(tǒng)建模和仿真集成工具。
Ptolemy II	Java	Lib	BSD	支持面向角色設(shè)計實驗的軟件框架。
SIM.JS	JavaScript	Lib	LGPL	JS是一個完全用JavaScript編寫的通用離散事件模擬庫。在瀏覽器中運行，支持基于GUI的建模工具。
SimPy	Python	Lib	MIT	SimPy是基于標(biāo)準(zhǔn)Python的基于過程的離散事件模擬框架。
Simula	Simula	Language		一種專門為模擬而設(shè)計的編程語言。
SystemC	C++	Lib	Apache 2.0	提供事件驅(qū)動模擬內(nèi)核。

注：軟件示意圖見《附錄二》

二、SimPy 講解

2.1 SimPy 概述

1）SimPy 是 python 的離散事件模擬框架。
2）SimPy 中的流程由 Python 生成器函數(shù)定義。例如，可以用于為客戶、車輛或代理等活動組件建模。SimPy 還提供各種類型的共享資源來模擬容量有限的擁塞點（如服務(wù)器、結(jié)賬柜臺和隧道）。
3）模擬可以要多快有多快、實時、或者手動步進(jìn)三種方式進(jìn)行。
3）盡管理論上 SimPy 可以用于連續(xù)模擬，但它沒有任何功能來幫助您實現(xiàn)這一點。另一方面，SimPy 對于具有固定步長的模擬來說是過火的，在這種情況下，您的流程不會相互交互或共享資源。

模擬兩個時鐘在不同時間間隔滴答作響的簡短示例如下：

>>> import simpy
>>>
>>> def clock(env, name, tick):
...     while True:
...         print(name, env.now)
...         yield env.timeout(tick)
...
>>> env = simpy.Environment()
>>> env.process(clock(env, 'fast', 0.5))
<Process(clock) object at 0x...>
>>> env.process(clock(env, 'slow', 1))
<Process(clock) object at 0x...>
>>> env.run(until=2)
fast 0
slow 0
fast 0.5
slow 1
fast 1.0
fast 1.5

2.2 基本概念

1）SimPy 是一個離散事件仿真庫。活動組件（如車輛、客戶或消息）的行為是用流程建模的。所有進(jìn)程都存在于一個環(huán)境中。它們通過事件與環(huán)境和彼此交互。
2）流程由簡單的 Python generator 描述。你可以將他們稱為過程函數(shù)或過程方法，取決于它是函數(shù)還是類的方法。在其整個生命周期內(nèi)，他們產(chǎn)生事件等待被觸發(fā)。
3）當(dāng)一個過程產(chǎn)生一個事件時，該進(jìn)程就會被掛起。當(dāng)事件發(fā)生時（我們說事件被觸發(fā)），SimPy 恢復(fù)該過程。多個進(jìn)程可以等待同一個事件。SimPy 以它們產(chǎn)生該事件的相同順序恢復(fù)它們。
4）一個最重要的事件類型就是 Timeout 類事件。它允許在進(jìn)程給定的時間內(nèi)休眠（或保持其他狀態(tài)）。

2.3 一個汽車開開停停的例子

下面是一個簡單的汽車走走停停的例子，打印其走停的時間戳：

>>> def car(env):
...     while True:
...         print('Start parking at %d' % env.now)
...         parking_duration = 5
...         yield env.timeout(parking_duration)
...
...         print('Start driving at %d' % env.now)
...         trip_duration = 2
...         yield env.timeout(trip_duration)

>>> import simpy
>>> env = simpy.Environment()
>>> env.process(car(env))
<Process(car) object at 0x...>
>>> env.run(until=15)
Start parking at 0
Start driving at 5
Start parking at 7
Start driving at 12
Start parking at 14

2.4 在走走停停過程中增加充電過程（過程交互）

我們在上面汽車?yán)踊A(chǔ)上引入充電的過程：車走一段時間，停下來充電，電充好了，才能繼續(xù)走。
這里引入了 charge_duration 過程，在該過程中簡單寫了一個超過的掛起事件：

>>> class Car(object):
...     def __init__(self, env):
...         self.env = env
...         # Start the run process everytime an instance is created.
...         self.action = env.process(self.run())
...
...     def run(self):
...         while True:
...             print('Start parking and charging at %d' % self.env.now)
...             charge_duration = 5
...             # We yield the process that process() returns
...             # to wait for it to finish
...             yield self.env.process(self.charge(charge_duration))
...
...             # The charge process has finished and
...             # we can start driving again.
...             print('Start driving at %d' % self.env.now)
...             trip_duration = 2
...             yield self.env.timeout(trip_duration)
...
...     def charge(self, duration):
...         yield self.env.timeout(duration)

>>> import simpy
>>> env = simpy.Environment()
>>> car = Car(env)
>>> env.run(until=15)
Start parking and charging at 0
Start driving at 5
Start parking and charging at 7
Start driving at 12
Start parking and charging at 14

如果我們不想等充電結(jié)束，而是想中斷充電過程并開始駕駛，可以使用 SimPy 的 interrupt() 方法來中斷正在運行的進(jìn)程：

>>> def driver(env, car):
...     yield env.timeout(3)
...     car.action.interrupt()

由于原來的充電過程被中斷會報異常，因此我們要對異常處理下：

...     		try:
...                 yield self.env.process(self.charge(charge_duration))
...             except simpy.Interrupt:
...                 # When we received an interrupt, we stop charging and
...                 # switch to the "driving" state
...                 print('Was interrupted. Hope, the battery is full enough ...')

再次運行：

>>> env = simpy.Environment()
>>> car = Car(env)
>>> env.process(driver(env, car))
<Process(driver) object at 0x...>
>>> env.run(until=15)
Start parking and charging at 0
Was interrupted. Hope, the battery is full enough ...
Start driving at 3
Start parking and charging at 5
Start driving at 10
Start parking and charging at 12

2.5 共享資源

SimPy 提供三種類型的資源，用于解決建模中多個進(jìn)行希望使用有限資源的問題（例如：加油站汽車場景中的燃油泵）或典型的生產(chǎn)者-消費者問題。

我們還用汽車充電的例子：汽車開到充電樁旁 a battery charging station (BCS)，向兩個充電樁申請使用其一進(jìn)行充電，如果兩個樁都在被使用，它將會等待直到可用，然后開始充電，然后開走。

>>> def car(env, name, bcs, driving_time, charge_duration):
...     # Simulate driving to the BCS
...     yield env.timeout(driving_time)
...
...     # Request one of its charging spots
...     print('%s arriving at %d' % (name, env.now))
...     with bcs.request() as req:
...         yield req
...
...         # Charge the battery
...         print('%s starting to charge at %s' % (name, env.now))
...         yield env.timeout(charge_duration)
...         print('%s leaving the bcs at %s' % (name, env.now))

備注： bcs.request() 將會產(chǎn)生一個事件，該事件會阻塞直到資源可用，一般情況下使用資源后需要調(diào)用 release 對資源進(jìn)行釋放，這里的 with 語句意味著自動釋放。

我們創(chuàng)建有兩個充電樁的資源：

>>> import simpy
>>> env = simpy.Environment()
>>> bcs = simpy.Resource(env, capacity=2)

然后我們創(chuàng)建 4 輛車：

>>> for i in range(4):
...     env.process(car(env, 'Car %d' % i, bcs, i*2, 5))

最后，我們可以開始模擬了。由于汽車進(jìn)程在此模擬中都自行終止，因此我們無需指定直到時間——當(dāng)沒有更多事件時，模擬將自動停止：

>>> env.run()
Car 0 arriving at 0
Car 0 starting to charge at 0
Car 1 arriving at 2
Car 1 starting to charge at 2
Car 2 arriving at 4
Car 0 leaving the bcs at 5
Car 2 starting to charge at 5
Car 3 arriving at 6
Car 1 leaving the bcs at 7
Car 3 starting to charge at 7
Car 2 leaving the bcs at 10
Car 3 leaving the bcs at 12

注意到前兩輛車到達(dá)BCS后可以立即開始充電，而2號車和3號車需要等待，符合預(yù)期。