[游戏开发]【python】SimPy库

官网

添加链接描述

作用

使用SimPY进行离散事件仿真
SimPY是一个Python下的第三方库，可以方便的进行离散事件的仿真。仿真速度比较快。下面记录一下我的一点心得，不保证完全正确，供参考。

主要概念

Environment
Process
Event
Resource
SimPY使用Environment，Process，Event，Resource四大概念来进行离散事件的仿真。

Environment就是整体仿真所在的时间，主要用于提取时间。
Process就是仿真过程中的实体，如：顾客， 设备， 车辆等。 Process本质上也是一个event。源代码里面可以看到是继承Event的一个类。
Event是仿真中触发的事件，可以理解为一个定时器。当定时器到时时，触发事件。
Resource是仿真中的资源，如ATM机，服务器等。

示例

第一个例子是car流程。汽车会交替行驶和停放一段时间。当它开始行驶(或停车)时，它会打印当前的模拟时间。

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

流程交互

由Environment.process()返回的流程实例可用于流程交互。两个最常见的例子是等待另一个进程完成，以及在等待事件时中断另一个进程。
SimPy允许您通过调用interrupt()方法来中断正在运行的进程

>>> class Car(object):
...     def __init__(self, env):
...         self.env = env
...         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 may get interrupted while charging the battery
...             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 ...')
...
...             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)

共享资源

simpy.Resource(env, capacity=2)
请求资源bcs.request()

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


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

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

服务站示例

"""
服务站示例
场景介绍:
  一个有特定服务提供工作站，客户服务时长不一，工作机器数有限。
  Client接受服务步骤：Client到达工作站，若有空闲的机器就立刻接受服务，如果没有，就等待直到其他机器空闲下来。
  每个接受过服务的Client都有一个完成满意度（或者为进度）实时统计服务客户数和完成满意进度。
"""
import random
import simpy
 
# 可接受输入参数
RANDOM_SEED = 0         # 不设置
NUM_MACHINES = 2        # 可以同时处理的机器数（类似工作工位数）
TIME_CONSUMING = 5      # 单任务耗时 (可以设计成随机数)
TIME_INTERVAL = 5       # 来车的间隔时间约5分钟   (可以设计成随机数)
SIM_TIME = 1000         # 仿真总时间
CLIENT_NUMBER = 2       # 初始时已经占用机器数
 
 
class WorkStation(object):
    """
    一个工作站，拥有特定数量的机器数。 一个客户首先申请服务。在对应服务时间完成后结束并离开工作站
    """
    def __init__(self, env, num_machines, washtime):
        self.env = env
        self.machine = simpy.Resource(env, num_machines)
        self.washtime = washtime
        self.allClient = 0
        self.accomplishClient = 0
 
    def wash(self, car):
        """服务流程"""
        yield self.env.timeout(random.randint(2, 10))  # 假设服务时间为随机数（2~10）
        self.allClient += 1
        per = random.randint(50, 99)
        print("%s's 任务完成度：%d%%." % (car, per))
        if per > 80:
            self.accomplishClient += 1
 
        print("工作站服务客户数：%d,"
              "工作站服务达标率：%.2f。" % (self.allClient, float(self.accomplishClient) / float(self.allClient)))
 
 
def Client(env, name, cw):
    """
    客户到达动作站接受服务，结束后离开
    """
 
    print('%s 到达工作站 at %.2f.' % (name, env.now))
    with cw.machine.request() as request:
        yield request
        print('%s 接受服务   at %.2f.' % (name, env.now))
        yield env.process(cw.wash(name))
        print('%s 离开服务站 at %.2f.' % (name, env.now))
 
 
def setup(env, num_machines, washtime, t_inter, clientNumber):
    """创建一个工作站，几个初始客户，然后持续有客户到达. 每隔t_inter - 2, t_inter + 3分钟（可以自定义）."""
    # 创建工作站
    workstation = WorkStation(env, num_machines, washtime)
 
    # 创建clientNumber个初始客户
    for i in range(clientNumber):
        env.process(Client(env, 'Client_%d' % i, workstation))
 
    # 在仿真过程中持续创建客户
    while True:
        yield env.timeout(random.randint(t_inter - 2, t_inter + 3))  # 3-8分钟
        i += 1
        env.process(Client(env, 'Client_%d' % i, workstation))
 
 
# 初始化并开始仿真任务
print('开始仿真')
 
# 初始化seed，指定数值的时候方正结果可以复现
random.seed()
 
# 创建一个环境并开始仿真
env = simpy.Environment()
env.process(setup(env, NUM_MACHINES, TIME_CONSUMING, TIME_INTERVAL, CLIENT_NUMBER))
 
# 开始执行!
env.run(until=SIM_TIME)

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖