[Pyhton] SimPy 离散事件模拟框架详解 —— 以一个简单的汽车充电排队模拟为例

一、背景知识

人们在生产活动和社会活动中，经常遇到一类复杂的系统，这类系统中有许多事件时而出现，时而消失，时而动作，时而停止，而启动和停止都发生在一些离散的时刻，并带有一定的随机性。例如，港口中船舶的停靠码头、生产线上机床的启停、电话的接通和断开、计算机系统中某项作业的进行和退出，凡此种种，都带有上述特点，这类系统叫做离散事件动态系统(DEDS)。随着生产和科技的日益发展，以及人类社会交往的日趋频繁，这类系统的数量和种类也越来越多。

设计此类系统时，往往需要仿真来评估算法或方案的性能，常见的软件有：

商业软件：

名字	简介	图示
AnyLogic	通用多方法建模工具。结合了基于代理、系统动力学和离散事件建模。
Arena	一种离散事件模拟程序，也允许对连续过程进行建模。
Care pathway simulator	专门为服务行业（如医疗保健）设计的离散事件模拟程序。
Enterprise Dynamics	一个模拟软件平台，用于模拟和分析几乎任何制造、材料处理和物流挑战。
ExtendSim	通用仿真软件包
DELMIA	3DEXPERIENCE 平台的一部分
FlexSim	拖拖拉拉做离散事件模拟，3D
GoldSim	将动态离散事件模拟嵌入到 Monte Carlo 框架
GPSS	离散事件模拟语言。供应商可以提供不同的实现
Micro Saint Sharp	通用离散事件建模工具，使用拖放界面和C#编程语言
MS4 Modeling Environment	基于离散事件和混合模型的通用DEVS方法的软件环境
Plant Simulation	能够模拟和优化生产系统和流程的软件
ProModel AutoCAD Edition	在数字孪生Autodesk工具集（AutoCAD和Inventor）中快速构建空间精确的布局和过程仿真模型。
Simcad Pro	实时变化，零代码有界面，支持 VR
SimEvents	向MATLAB / Simulink环境添加离散事件仿真。
SIMUL8	基于对象的仿真软件
VisualSim	基于时序、功耗和功能的电子、嵌入式软件和半导体的基于模型的系统架构探索
WITNESS	可在桌面和云端使用 VR 进行离散事件模拟

开源软件：

名字	语言	类型	License	简介
JaamSim	Java	App	Apasche 2.0	JaamSim是一款自由开源的离散事件模拟软件，包括拖放式用户界面、交互式3D图形、输入和输出处理以及模型开发工具和编辑器。
CPN Tools	BETA	App	GPLv2	用于分析所有类型应用中的物流/排队模型的工具。
DESMO-J	Java	Lib	Apasche 2.0	Java离散事件模拟框架，支持混合事件/过程模型，并提供2D和3D动画。
Facsimile	Scala	Lib	LGPLv3	离散事件模拟/仿真库。
PowerDEVS	C++	App	AFL, GPLv2	基于DEVS形式的混合系统建模和仿真集成工具。
Ptolemy II	Java	Lib	BSD	支持面向角色设计实验的软件框架。
SIM.JS	JavaScript	Lib	LGPL	JS是一个完全用JavaScript编写的通用离散事件模拟库。在浏览器中运行，支持基于GUI的建模工具。
SimPy	Python	Lib	MIT	SimPy是基于标准Python的基于过程的离散事件模拟框架。
Simula	Simula	Language		一种专门为模拟而设计的编程语言。
SystemC	C++	Lib	Apache 2.0	提供事件驱动模拟内核。

注：软件示意图见《附录二》

二、SimPy 讲解

2.1 SimPy 概述

1）SimPy 是 python 的离散事件模拟框架。

2）SimPy 中的流程由 Python 生成器函数定义。例如，可以用于为客户、车辆或代理等活动组件建模。SimPy 还提供各种类型的共享资源来模拟容量有限的拥塞点（如服务器、结账柜台和隧道）。

3）模拟可以要多快有多快、实时、或者手动步进三种方式进行。

3）尽管理论上 SimPy 可以用于连续模拟，但它没有任何功能来帮助您实现这一点。另一方面，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 是一个离散事件仿真库。活动组件（如车辆、客户或消息）的行为是用流程建模的。所有进程都存在于一个环境中。它们通过事件与环境和彼此交互。

2）流程由简单的 Python generator 描述。你可以将他们称为过程函数或过程方法，取决于它是函数还是类的方法。在其整个生命周期内，他们产生事件等待被触发。

3）当一个过程产生一个事件时，该进程就会被挂起。当事件发生时（我们说事件被触发），SimPy 恢复该过程。多个进程可以等待同一个事件。SimPy 以它们产生该事件的相同顺序恢复它们。

4）一个最重要的事件类型就是 Timeout 类事件。它允许在进程给定的时间内休眠（或保持其他状态）。

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 在走走停停过程中增加充电过程（过程交互）

我们在上面汽车例子基础上引入充电的过程：车走一段时间，停下来充电，电充好了，才能继续走。

这里引入了 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

如果我们不想等充电结束，而是想中断充电过程并开始驾驶，可以使用 SimPy 的 interrupt() 方法来中断正在运行的进程：

>>> 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 提供三种类型的资源，用于解决建模中多个进行希望使用有限资源的问题（例如：加油站汽车场景中的燃油泵）或典型的生产者-消费者问题。

我们还用汽车充电的例子：汽车开到充电桩旁 a battery charging station (BCS)，向两个充电桩申请使用其一进行充电，如果两个桩都在被使用，它将会等待直到可用，然后开始充电，然后开走。

>>> 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() 将会产生一个事件，该事件会阻塞直到资源可用，一般情况下使用资源后需要调用 release 对资源进行释放，这里的 with 语句意味着自动释放。

我们创建有两个充电桩的资源：

>>> import simpy

>>> env = simpy.Environment()

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

然后我们创建 4 辆车：

>>> for i in range(4):

...     env.process(car(env, 'Car %d' % i, bcs, i*2, 5))

最后，我们可以开始模拟了。由于汽车进程在此模拟中都自行终止，因此我们无需指定直到时间——当没有更多事件时，模拟将自动停止：

>>> 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

注意到前两辆车到达BCS后可以立即开始充电，而2号车和3号车需要等待，符合预期。

三、后续

之后我将用 SimPy 模拟射频节点的数据收发，进一步做一个 MESH 通信的模拟程序，用于验证不同的算法对 MESH 网络带来的性能差异。

敬请期待！！！

注：BLUETOOTH MESH 是利用蓝牙广播链路，基于洪范算法做的一种简单的组网协议（这里可以将广播理解为喊话、UDP广播等）

参考链接

[1]. SimPy 主页

[2]. 百度百科离散事件动态系统

[3]. List of discrete event simulation software

[4]. 离散事件系统仿真（原书第5版）

---

: 如果觉得不错，帮忙点个支持哈～

附录二

AnyLogic：

Arena：

FlexSim：

GoldSim：

Plant Simulation：

ProModel AutoCAD Edition：

VisualSim：