[Pyhton] 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 0slow 0fast 0.5slow 1fast 1.0fast 1.52.2 基本概念1)SimPy 是一个离散事件仿真库 。活动组件(如车辆、客户或消息)的行为是用流程建模的 。所有进程都存在于一个环境中 。它们通过事件与环境和彼此交互 。2)流程由简单的 Python generator 描述 。你可以将他们称为过程函数或过程方法 , 取决于它是函数还是类的方法 。在其整个生命周期内 , 他们产生事件等待被触发 。3)当一个过程产生一个事件时 , 该进程就会被挂起 。当事件发生时(我们说事件被触发) , SimPy 恢复该过程 。多个进程可以等待同一个事件 。SimPy 以它们产生该事件的相同顺序恢复它们 。4)一个最重要的事件类型就是 Timeout 类事件 。它允许在进程给定的时间内休眠(或保持其他状态) 。
【[Pyhton] SimPy 离散事件模拟框架详解 —— 以一个简单的汽车充电排队模拟为例】

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

文章插图
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 0Start driving at 5Start parking at 7Start driving at 12Start parking at 142.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 0Start driving at 5Start parking and charging at 7Start driving at 12Start 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 ...')

经验总结扩展阅读