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

阅读原文时间：2023年07月09日阅读：2

一、背景知识
二、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

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号车需要等待，符合预期。