NSDT工具推荐： Three.js AI纹理开发包 - YOLO合成数据生成器 - GLTF/GLB在线编辑 - 3D模型格式在线转换 - 可编程3D场景编辑器 - REVIT导出3D模型插件 - 3D模型语义搜索引擎 - Three.js虚拟轴心开发包 - 3D模型在线减面 - STL模型在线切割

离散事件仿真 (DES) 往往是专门产品的领域，例如 SIMUL8  和 MatLab/Simulink 。然而，当我在 Python 中执行过去使用 MatLab 的分析时，我很想测试 Python 是否也有 DES 的解决方案。

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

本文将首先介绍一个场景并展示如何在 SimPy 中实现它。然后，我们将研究三种不同的可视化结果的方法：Python 原生解决方案（使用 Matplotlib  和 Tkinter ）、基于 HTML5 画布的方法和交互式 AR/VR 可视化。最后，我们将使用我们的 SimPy 模型来评估替代配置。

1、仿真场景简介

我们将使用之前的一些工作中的一个示例作为仿真场景：公交服务入口队列。但是，遵循类似模式的其他示例可能是杂货店或接受在线订单的餐厅、电影院、药房或火车站的队列。

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

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

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

  • 巴士平均每 3 分钟一班。我们将使用 λ 为 1/3 的指数分布来表示
  • 每辆巴士将包含 100 +/- 30 名访客，使用正态分布确定（μ = 100，σ = 30）
  • 访客将使用正态分布（μ = 2.25，σ = 0.5）形成 2.25 +/- 0.5 人的小组。我们将把它四舍五入到最接近的整数
  • 我们假设固定比例的 40% 的访客需要在卖家展位购买门票，另外 40% 将使用已在线购买的门票到达，20% 将使用工作人员凭证到达
  • 访客平均需要一分钟下车步行到卖家展位（正态，μ = 1，σ = 0.25），另外半分钟从卖家步行到扫描仪（正态，μ = 0.5，σ = 0.1)。对于那些跳过卖家（预购票或有徽章的工作人员）的人，我们假设平均步行 1.5 分钟（正态，μ = 1.5，σ = 0.35）
  • 访客到达时会选择最短的线路，每条线路都有一个卖家或扫描仪
  • 一次销售需要 1 +/- 0.2 分钟才能完成（正态，μ = 1，σ = 0.2）
  • 完成一次扫描需要 0.05 +/- 0.01 分钟（正态，μ = 0.05，σ = 0.01）

考虑到这一点，让我们从输出开始，然后从那里向后推进：

左侧的图表代表每分钟到达的访客数量，右侧的图表代表当时退出队列的访客需要等待服务的平均时间。

2、SimPy 设置

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

首先，让我们从模拟的参数开始。分析最有趣的变量是卖家行数 ( SELLER_LINES ) 和每行卖家数 ( SELLERS_PER_LINE ) 以及扫描仪的等价物 ( SCANNER_LINES和SCANNERS_PER_LINE )。另外，请注意两种可能的队列/卖家配置之间的区别：虽然最流行的配置是有多个不同的队列供访客选择并停留直到他们得到服务，但在零售业中看到多个队列也变得更加主流一条线的卖家（例如，一般商品大卖场零售商的快速结账线）。

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

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

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, seller_lines, scanner_lines)) 

env.run(until = 60)

请注意，我们正在创建一个RealtimeEnvironment，它旨在近乎实时地运行模拟，特别是为了我们在运行时可视化它的意图。随着环境的建立，我们生成了我们的卖家和扫描线资源（队列），然后我们将依次传递给巴士到达的“主事件”。env.process ()命令将开始下面描述的bus_arrival()函数中描述的过程。此函数是调度所有其他事件的顶级事件。它模拟每BUS_ARRIVAL_MEAN分钟到达的公交车，有BUS_OCCUPANCY_MEAN人在车上，然后相应地触发销售和扫描过程。

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

        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
            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))
            else:
                env.process(purchasing_customer(env, people_processed, seller_lines, scanner_lines))

由于这是顶层事件函数，我们看到该函数中的所有工作都在一个无限循环中进行。在循环中，我们使用env.timeout() “让出”我们的等待时间。SimPy 广泛使用生成器函数，这些函数将返回生成值的迭代器。有关 Python 生成器的更多信息可以在 [10] 中找到。

在循环结束时，我们将调度两个事件之一，具体取决于我们是直接前往扫描仪还是我们随机决定该组需要先购买门票。请注意，我们不会屈服于这些过程，因为这会指示 SimPy 按顺序完成这些操作中的每一个；取而代之的是，所有离开公共汽车的游客将同时进入队列。

请注意，正在使用people_ids列表，以便为每个人分配一个唯一 ID 以用于可视化目的。我们使用people_ids列表作为剩余待处理人员的队列；当访客被派往他们的目的地时，他们会从people_ids队列中移除。

purchase_customer ()函数模拟三个关键事件：步行到排队，排队等候，然后将控制权传递给scanning_customer()事件（对于那些绕过卖家并直接进入的人， bus_arrival()调用的函数相同扫描仪）。此功能根据选择时最短的线来选择线。

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)
    with seller_line[0].request() as req:
        yield req # Wait in line

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

        env.process(scanning_customer(env, people_processed, scanner_lines, TIME_TO_WALK_TO_SCANNERS_MEAN, TIME_TO_WALK_TO_SCANNERS_STD))

最后，我们需要实现scanning_customer()的行为。这与purchase_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 
        for person in people_processed:
            yield env.timeout(random.gauss(SCANNER_MEAN, SCANNER_STD)) # Scan their ticket

我们将步行持续时间和标准差传递给scanning_customer()函数，因为这些值会根据访问者是直接走到扫描仪前还是先停在卖家处而有所不同。

3、用 Tkinter 可视化数据

为了可视化数据，我们添加了一些全局列表和字典来跟踪关键指标。例如，arrivals 字典按分钟跟踪到达的数量，而 Seller_waits 和 scan_waits 字典将模拟的分钟映射到那些分钟内退出队列的等待时间列表。还有一个 event_log 列表，我们将在下一节的 HTML5 Canvas 动画中使用它。当关键事件发生时（例如，访问者退出队列），将调用simpy example.py文件中ANALYTICAL_GLOBALS标题下的函数以使这些字典和列表保持最新。

我们使用辅助 SimPy 事件向 UI 发送tick事件，以更新时钟、更新当前等待平均值并重绘 Matplotlib 图表。完整的代码可以在 GitHub 存储库中找到；但是，以下代码片段提供了如何从 SimPy 分派这些更新的框架视图。

class ClockAndData: 
    def __init__(self, canvas, x1, y1, x2, y2, time): 
        # Draw the initial state of the clock and data on the canvas 
        self.canvas.update() 

    def tick(self, time): 
        # Re-draw the the clock and data fields on the canvas. Also update the Matplotlib charts. 

# ... 

clock = ClockAndData(canvas, 1100, 320, 1290, 400, 0)  

# ... 

def create_clock(env):
    while True: 
        yield env.timeout(0.1) 
        clock.tick(env.now) 

# ... 

env.process(create_clock(env))

用户进出卖方和扫描仪队列的可视化使用标准 Tkinter 逻辑表示。我们创建了QueueGraphics类来抽象卖方和扫描仪队列的公共部分。此类中的方法被编码到上一节中描述的 SimPy 事件函数中以更新画布（例如，sellers.add_to_line(1)，其中 1 是卖家编号，以及Sellers.remove_from_line(1)）。作为未来的工作，我们可以在流程的关键点使用事件处理程序，这样 SimPy 模拟逻辑就不会与特定于该分析的 UI 逻辑紧密耦合。

4、使用 HTML5 Canvas 动画数据

作为替代可视化，我们希望从 SimPy 模拟中导出事件并将它们拉入一个简单的 HTML5 Web 应用程序，以在 2D 画布上可视化场景。我们通过在 SimPy 事件发生时附加到event_log列表来实现这一点。特别是，公交车到达、步行到卖家、在卖家排队等候、买票、步行到扫描仪、在扫描仪排队等候和扫描车票事件都被记录为单独的字典，然后在模拟结束时导出到 JSON . 可以在这里看到一些示例输出。

我们开发了一个快速的概念验证来展示如何将这些事件转换为 2D 动画，您可以在在这里进行试验。可以在这里查看动画逻辑的源代码：

这种可视化受益于动画，然而，出于实际目的，基于 Python 的 Tkinter 界面组装起来更快，而且 Matplotlib 图形（可以说是这个模拟中最重要的部分）也更流畅，更熟悉在 Python 中设置. 话虽如此，看到行为动画是有价值的，特别是在寻求将结果传达给非技术利益相关者时。

5、使用VR动画

让画布动画更进一步，

马修斯·希梅内斯我一起使用 HTML5 画布也使用的相同 JSON 模拟数据将以下 AR/VR 3-D 可视化放在一起。我们使用我们已经熟悉的 React [11] 和 A-FRAME [12] 实现了这一点，它非常易于访问且易于学习。可以在这个网址测试模拟：

6、分析卖方/扫描队列配置备选方案

尽管这个例子已经被放在一起来演示如何创建和可视化 SimPy 模拟，我们仍然可以展示一些例子来展示平均等待时间如何依赖于队列的配置。

让我们从上面动画中演示的案例开始：六个卖家和四个扫描仪，每行一个卖家和一个扫描仪 (6/4)。60 分钟后，我们看到卖家平均等待时间为 1.8 分钟，扫描仪平均等待时间为 0.1 分钟。从下面的图表中，我们看到卖家时间在几乎 6 分钟的等待时间达到峰值。

我们可以看到卖家持续备份（虽然3.3分钟可能不会太不合理）；所以，让我们看看如果我们再增加四个卖家，将总数增加到 10 个，会发生什么。

正如预期的那样，平均卖家等待时间减少到 0.7 分钟，最长等待时间减少到刚刚超过 3 分钟。

现在，假设通过降低在线门票的价格，我们能够将持票到达的人数增加 35%。最初，我们假设 40% 的访客需要购买门票，40% 已在网上预购，20% 是持证件进入的员工和供应商。因此，随着持票人数增加 35%，我们将需要购买的人数减少到 26%。让我们用我们最初的 6/4 配置来模拟这个。

在这种情况下，平均卖家等待时间减少到 1.0 分钟，最长等待时间超过 4 分钟。在这种情况下，将在线销售额提高 35% 与在平均等待时间中增加更多卖家队列的效果相似；如果等待时间是我们最有兴趣减少的指标，那么可以考虑这两个选项中的哪一个具有更强的商业案例。

7、结束语

可用于 Python 的数学和分析工具的广度令人生畏，SimPy 完善了这些功能，还包括离散事件模拟。与 SIMUL8 等商业打包工具相比，Python 方法确实为编程留下了更多的余地。从头开始组装模拟逻辑并构建 UI 和测量支持对于快速分析来说可能很笨拙；但是，它确实提供了很大的灵活性，并且对于已经熟悉 Python 的人来说应该相对简单。如上所述，SimPy 提供的 DES 逻辑可以生成干净、易于阅读的代码。

如前所述，Tkinter 可视化是三种演示方法中最直接的一种，特别是在包含 Matplotlib 支持的情况下。HTML5 画布和 AR/VR 方法可以方便地组合可共享和交互式的可视化；然而，它们的发展并非微不足道。

比较队列配置时需要考虑的一项重要改进是卖方/扫描仪的利用率。减少排队时间只是分析的一个组成部分，因为在得出最佳解决方案时还应考虑卖家和扫描仪空闲时间的百分比。此外，如果有人看到队列太长，则添加一个概率来解释他们选择不进入的概率也会很有趣。

原文链接：Simulating and Visualizing Real-Life Events in Python with SimPy

BimAnt翻译整理，转载请标明出处