SimPy 离散事件仿真入门-yangzy's blog

DES（Discrete event simulation）是一种使用统计函数对现实事件进行建模的方法，通常用于医疗保健、制造、物流等领域的队列和资源使用。最终目标是获得关键运营指标，例如资源使用情况和平均等待时间，以便评估和优化各种现实生活中的配置。SimPy库支持在Python中描述和运行DES模型。与SIMUL8等软件包不同，SimPy不是用于构建、执行和报告模拟的完整图形环境，不过它的确提供了执行仿真及输出数据给可视化和分析环节的基础组件。

Python方法为编程留下了更多的余地。从头开始组装模拟逻辑并构建UI和测量支持对于快速分析来说可能很笨拙；但是，它确实提供了很大的灵活性，并且对于已经熟悉Python的人来说应该相对简单。

一个例子

一、仿真场景简介

仿真场景：公交服务入口队列。

将模拟一个完全由公共交通服务的入口：一辆公共汽车将定期送走几名顾客，然后他们需要在进入活动之前扫描其门票。一些游客将有他们提前预购的徽章或门票，而另一些游客则需要先到卖家摊位购买门票。更复杂的是，当访客接近卖家摊位时，他们会成群结队地这样做（模拟家庭/团体购票）；但是，每个人都需要单独扫描他们的门票。

下面描述了此场景的顶层布局：

为了模拟这一点，我们需要决定如何使用概率分布来表示这些不同的事件。我在实施中所做的假设包括：

巴士平均3min一班，使用$\lambda = \frac{1}{3}$的指数分布来表示
每辆巴士将包含100(+/-30)位游客，使用正态分布确定，$\mu = 100, \sigma = 30$
访客将分为2.25(+/-0.5)人的小组，使用正态分布确定然后将其四舍五入，$\mu=2.25,\sigma=0.5$
假设固定比例的访客($40\%$)在售票处购买门票，另外($40\%$)提前在线购买门票，剩余($20\%$)将使用工作人员凭证直接进入展区
访客平均需要1min时间下车步行到售票处购票，使用正态分布，$\mu=1,\sigma=0.25$；另外需要0.5min从售票处步行到检票处，使用正态分布，$\mu=0.5,\sigma=0.1$；对于那些已经提前在线购买好票的访客或者那些工作人员，假设平均步行时间为1.5min，正态分布，$\mu=1.5,\sigma=0.35$
访客到达时会选择最短的线路，每条线路有售票处或检票处其中之一
售票处一次销售需要1(+/-0.2)min完成，正态分布，$\mu=1,\sigma=0.2$
完成一次扫描检票需要0.05(+/-0.01)min完成，正态分布，$\mu=0.05,\sigma=0.01$

二、SimPy设置

可以在这里找到具有完整可运行源的存储库，其中包含从简单的simpy_example.py文件中提取的以下片段。在本文中，我们将聚焦于特定于SimPy的设置；但是请注意，为了关注SimPy的DES功能，省略了连接到Tkinter进行可视化的部分。

2.1 基础参数设置

BUS_ARRIVAL_MEAN = 3
BUS_OCCUPANCY_MEAN = 100
BUS_OCCUPANCY_STD = 30

PURCHASE_RATIO_MEAN = 0.4
PURCHASE_GROUP_SIZE_MEAN = 2.25
PURCHASE_GROUP_SIZE_STD = 0.50

TIME_TO_WALK_TO_SELLERS_MEAN = 1
TIME_TO_WALK_TO_SELLERS_STD = 0.25
TIME_TO_WALK_TO_SCANNERS_MEAN = 0.5
TIME_TO_WALK_TO_SCANNERS_STD = 0.1

# 售票处的列数
SELLER_LINES = 6
# 每列的售票处数量，反映一次能处理售票事件数目
SELLERS_PER_LINE = 1
SELLER_MEAN = 1
SELLER_STD = 0.2

# 检票处的列数
SCANNER_LINES = 4
# 每列的检票处数量，反映一次能处理检票事件数目
SCANNERS_PER_LINE = 1
SCANNER_MEAN = 1 / 20
SCANNER_STD = 0.01

2.2 启动模拟

配置完成后，让我们通过首先创建一个“环境”、所有队列（资源）来启动SimPy过程，然后运行模拟（在本例中，直到60分钟时将结束模拟）：

# RealtimeEnvironment，旨在近乎实时地运行模拟
env = simpy.rt.RealtimeEnvironment(factor = 0.1, strict = False) 

# 生成售票处和检票处资源（队列），然后将依次传递给巴士到达的“主事件”
seller_lines = [ simpy.Resource(env, capacity = SELLERS_PER_LINE) for _ in range(SELLER_LINES) ] 
scanner_lines = [ simpy.Resource(env, capacity = SCANNERS_PER_LINE) for _ in range(SCANNER_LINES) ] 

# env.process()命令将创建一个新的进程bus_arrival()并将其与指定的函数关联起来，此函数是调度所有其他事件的顶级事件。
env.process(bus_arrival(env, seller_lines, scanner_lines)) 

env.run(until = 60)

2.3 bus_arrival()事件

def bus_arrival(env, seller_lines, scanner_lines):
    # Note that these unique IDs for busses and people are not required, but are included for eventual visualizations 
    next_bus_id = 0
    next_person_id = 0
    while True:
        next_bus = random.expovariate(1 / BUS_ARRIVAL_MEAN)        
        on_board = int(random.gauss(BUS_OCCUPANCY_MEAN, BUS_OCCUPANCY_STD))        
        
        # Wait for the bus 
        # 生成器函数，用于创建一个等待一段时间后再继续执行的事件
        yield env.timeout(next_bus)
        
        people_ids = list(range(next_person_id, next_person_id + on_board))
        next_person_id += on_board
        next_bus_id += 1

        # 每次循环中，从一个人员ID列表中取出一定数量的人员进行处理
        while len(people_ids) > 0:
            remaining = len(people_ids)
            group_size = min(round(random.gauss(PURCHASE_GROUP_SIZE_MEAN, PURCHASE_GROUP_SIZE_STD)), remaining)
            people_processed = people_ids[-group_size:] # Grab the last `group_size` elements
            people_ids = people_ids[:-group_size] # Reset people_ids to only those remaining

            # Randomly determine if this group is going to the sellers or straight to the scanners
            # 直接进入检票处，创建一个新的进程（Process）并将其与指定的函数关联起来，转移控制权
            if random.random() > PURCHASE_RATIO_MEAN:
                env.process(scanning_customer(env, people_processed, scanner_lines, TIME_TO_WALK_TO_SELLERS_MEAN + TIME_TO_WALK_TO_SCANNERS_MEAN, TIME_TO_WALK_TO_SELLERS_STD + TIME_TO_WALK_TO_SCANNERS_STD))
            # 去售票处买票，创建一个新的进程（Process）并将其与指定的函数关联起来，转移控制权
            else:
                env.process(purchasing_customer(env, people_processed, seller_lines, scanner_lines))

2.4 scanning_customer()事件

def purchasing_customer(env, people_processed, seller_lines, scanner_lines):
    # Walk to the seller
    yield env.timeout(random.gauss(TIME_TO_WALK_TO_SELLERS_MEAN, TIME_TO_WALK_TO_SELLERS_STD))
    # 选择最短的队列
    seller_line = pick_shortest(seller_lines)
    # 进程正在尝试获取卖票处的资源，然后使用yield req来等待获取资源。一旦资源可用（即排在队列中的进程可以得到执行），代码会继续执行下去
    with seller_line[0].request() as req:
        yield req # Wait in line

        yield env.timeout(random.gauss(SELLER_MEAN, SELLER_STD)) # Buy their tickets

        # 将控制权传递给scanning_customer()事件
        env.process(scanning_customer(env, people_processed, scanner_lines, TIME_TO_WALK_TO_SCANNERS_MEAN, TIME_TO_WALK_TO_SCANNERS_STD))

2.5 scanning_customer()事件

def scanning_customer(env, people_processed, scanner_lines, walk_duration, walk_std):
    # Walk to the seller 
    yield env.timeout(random.gauss(walk_duration, walk_std))

    # We assume that the visitor will always pick the shortest line
    scanner_line = pick_shortest(scanner_lines)
    with scanner_line[0].request() as req:
        yield req # Wait in line
        
        # Scan each person's tickets 
        # 与scanning_customer()事件的关键区别：
        # 虽然访客可能会成群结队地一起到达并步行，但每个人都必须单独扫描他们的票
        for person in people_processed:
            yield env.timeout(random.gauss(SCANNER_MEAN, SCANNER_STD)) # Scan their ticket