使用SimPy进行离散事件仿真

在物流行业/工厂制造业/餐饮服务业存在大量急需优化的场景,但是在实际生产中随时调整生产线来试验最优解是非常昂贵的,这就需要引入仿真技术。SimPy是一个以进程为基础的离散事件仿真框架,可以给业务研究人员无限的自由度去调整验证不同的优化方案。

一、核心概念

1、进程、事件、环境

SimPy中所有的活动部件(例如顾客、工人、车辆,甚至是信息),都可以用进程来模拟。这些进程都存放在环境中。而进程之间的互动,以及进程与环境间的互动,是以事件的方式来进行。
当一个进程抛出事件,进程会被暂停,直到事件被激活(triggered)。多个进程可以等待同一个事件,SimPy会按照这些进程抛出的事件激活的先后来恢复进程。最重要的一类事件是 Timeout,这类事件允许一段时间后再被激活,用来表达一个进程休眠或者保持当前的状态持续指定的一段时间。

2、模拟资源

但凡仿真中涉及的人力资源以及生产物料消耗都会用抽象资源(Resource)来表达。通过给资源配置优先级可以实现插队功能。

3、模拟管道

对于具有工序、流水线性质的流程,可以使用抽象管道(Store)来表达。

二、学习进程和事件

1、定义进程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import simpy

# 定义一个汽车进程
def car(env):
while True:
# 停泊一段时间行驶一段时间
print('停泊在 %d' % env.now)
yield env.timeout(5) # timeout事件,延时5s
print('行驶到 %d' % env.now)
yield env.timeout(2) # timeout事件,延时2s

env = simpy.Environment() # 实例化环境
env.process(car(env)) # 添加汽车进程
env.run(until=15) # 仿真启动,设定仿真结束条件(这里是15s后停止)

2、进程通过事件交互

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from random import seed, randint
import simpy

class EV:
def __init__(self, env):
self.env = env
self.drive_proc = env.process(self.drive(env)) # 定义行驶进程
self.bat_ctrl_proc = env.process(self.bat_ctrl(env)) # 定义充电进程
self.bat_ctrl_reactivate = env.event() # 充电事件, 默认是休眠
self.bat_ctrl_sleep = env.event() # 充电结束事件, 默认是休眠

def drive(self, env):
# 行驶进程
while True:
print('开始行驶 时间: ', env.now)
yield env.timeout(randint(20, 40)) # 行驶 20-40 分钟

print('停止行驶 时间: ', env.now)
self.bat_ctrl_reactivate.succeed() # 激活充电事件
self.bat_ctrl_reactivate = env.event() # 发送一个激活信号后充电事件再次休眠
yield env.timeout(randint(60, 360)) & self.bat_ctrl_sleep # 停车时间(1-6小时)和充电结束信号同时都满足

def bat_ctrl(self, env):
# 电池充电进程
while True:
print('--充电程序休眠中')
yield self.bat_ctrl_reactivate # 休眠直到充电事件被激活
print('--充电程序激活 时间:', env.now)
yield env.timeout(randint(30, 90))
print('--充电程序结束 时间:', env.now)
self.bat_ctrl_sleep.succeed() # 激活充电结束事件
self.bat_ctrl_sleep = env.event() # 发送一个激活信号后充电结束事件再次休眠

env = simpy.Environment()
ev = EV(env)
env.run(until=300)

三、学习模拟资源

模拟银行接待客户。假设只有一个柜台对客户进行服务, 客户等候过长会离开柜台。

先进先出模式:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
from random import randint
import simpy

class Bank(object):
def __init__(self, env, capacity):
self.env = env
self.counter = simpy.Resource(self.env, capacity=capacity) # 争用柜台

self.success = 0 # 成功服务的顾客数量

def service(self, name):
# 银行提供的服务
print('-- %s 在 %.2f 时刻占用柜台' % (name, env.now))
yield self.env.timeout(randint(3, 10)) # 假设服务时间为随机数
self.success += 1
print('-- %s 在 %.2f 时刻离开柜台' % (name, env.now))

def customer(env, name, bank):
# 客户进程
print('%s 在 %.2f 时刻进入银行' % (name, env.now))
arrive = env.now # 到达时间
with bank.counter.request() as request:
patience = randint(1, 3) # 客户耐心
# 等待柜员服务或者超出忍耐时间离开队伍
results = yield request | env.timeout(patience)

if request in results:
# 到达柜台
yield env.process(bank.service(name))
else:
# 耐心耗尽
print('[%s 等待了 %.2f 后离开]' % (name, (env.now-arrive)))

def setup(env, capacity, number):
# 创建银行
bank = Bank(env, capacity)
i = 0
# 在仿真过程中持续创建客户
for i in range(number):
yield env.timeout(randint(3, 8)) # # 客户到达的间距时间
i += 1
env.process(customer(env, '客户_%d' % i, bank))

env = simpy.Environment()
env.process(setup(env, 1, 5))
env.run()

营业额优先模式:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from random import randint
import simpy

class Bank(object):
def __init__(self, env, capacity):
self.env = env
self.counter = simpy.PriorityResource(self.env, capacity=capacity) # 争用柜台

self.success = 0 # 成功服务的顾客数量

def service(self, name):
# 银行提供的服务
print('-- %s 在 %.2f 时刻占用柜台' % (name, env.now))
yield self.env.timeout(randint(5, 15)) # 假设服务时间为随机数
self.success += 1
print('-- %s 在 %.2f 时刻离开柜台' % (name, env.now))

def customer(env, name, bank, account):
# 客户进程
print('%s 在 %.2f 时刻进入银行' % (name, env.now))
arrive = env.now # 到达时间
# 以业绩作为优先级, priority越小, 优先级越大
with bank.counter.request(priority = 1/account) as request:
patience = randint(10, 15) # 客户耐心
# 等待柜员服务或者超出忍耐时间离开队伍
results = yield request | env.timeout(patience)

if request in results:
# 到达柜台
yield env.process(bank.service(name))
else:
# 耐心耗尽
print('[%s 等待了 %.2f 后离开]' % (name, (env.now-arrive)))

def setup(env, capacity, number):
# 创建银行
bank = Bank(env, capacity)
i = 0
# 在仿真过程中持续创建客户
for i in range(number):
yield env.timeout(randint(1, 3)) # # 客户到达的间距时间
i += 1
env.process(customer(env, '客户_%d' % i, bank, account=randint(5000, 100000)))

env = simpy.Environment()
env.process(setup(env, 1, 5))
env.run()

四、学习模拟管道

模拟一条通讯线路。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
import simpy

class Cable(object):
# 模拟通信线路
def __init__(self, env, delay):
self.env = env
self.delay = delay
self.store = simpy.Store(env)

def latency(self, value): # 延时
yield self.env.timeout(self.delay)
self.store.put(value)

def put(self, value):
self.env.process(self.latency(value))

def get(self):
return self.store.get()

def sender(env, cable):
# 消息发送方
while True:
yield env.timeout(5) # 等待下一次通讯
cable.put('在 %d 时刻发送的消息' % env.now)


def receiver(env, cable):
# 消息接收方
while True:
msg = yield cable.get()
print('在 %d 接收到 %s' % (env.now, msg))

env = simpy.Environment()
cable = Cable(env, 10)
env.process(sender(env, cable))
env.process(receiver(env, cable))

env.run(until=100)

官方文档

0%