SimPy：用 Python 模拟真实世界的进程

模拟的工作原理

在 Python 中，您可以使用该simpy框架进行事件模拟。首先，快速了解一下模拟过程如何在 Python 中运行。下面是模拟安全检查点系统的代码片段。以下三行代码设置环境，传递所有必要的功能，并运行模拟：

# Set up the environment
env = simpy.Environment()

# Assume you've defined checkpoint_run() beforehand
env.process(checkpoint_run(env, num_booths, check_time, passenger_arrival))

# Let's go!
env.run(until=10)

上面的第一行代码建立了环境。您将通过分配simpy.Environment()给所需的变量来完成此操作。在这里，它被简单地命名为env。这告诉simpy创建一个名为的环境对象env，该对象将管理模拟时间并在每个后续时间步长中移动模拟。

建立环境后，您将传入所有将作为参数的变量。这些是您可以更改的内容，以查看系统将如何对更改做出反应。对于此安全检查点系统，您使用以下参数：

1. env：调度和处理事件的环境对象
2. num_booths:身份证检查站的数量
3. check_time: 检查乘客身份证所需的时间
4. passenger_arrival：乘客到达队列的速度

然后，是时候运行模拟了！您可以通过调用env.run()并指定您希望模拟运行多长时间来完成此操作。模拟在几分钟内运行，因此此示例代码将实时运行模拟 10 分钟。

回顾一下，以下是在 Python 中运行模拟的三个步骤：

1. 建立环境。
2. 传入参数。
3. 运行模拟。

但在引擎盖下还有很多事情要做！您需要了解如何选择这些参数，并且必须定义在运行模拟时将调用的所有函数。

让我们开始吧！

如何开始 simpy

在 Python 中创建模拟之前，您应该核对一些待办事项。您需要做的第一件事是确保您对Python 基础知识有充分的了解。特别是，您需要很好地掌握类和生成器。

您要做的下一件事是安装所需的软件包。您将使用的主要框架是simpy. 这是将创建、管理和运行模拟的核心包。您可以使用以下命令安装它pip：

$ python3 -m pip install simpy

您还需要一些内置的 Python 模块。您将使用该statistics模块计算平均等待时间并使用该random模块生成随机数。这些是 Python 标准库的一部分，因此您无需安装任何新东西。

最后，您需要选择运行模拟的方式。通常，您可以选择以下两个选项之一：

1. 以交互方式运行它：使用Jupyter Notebook，其中每个代码块将包含自己的类或函数定义。输出将显示在笔记本的底部。
2. 在 shell 中运行它：将您的模拟保存为一个.py文件并告诉 Python 在您的终端中运行它。输出将直接打印到控制台。

选择您最满意的方法！结果应该是一样的。

在本教程中，您将看到对名为simulate.py. 在您学习本教程时，将参考代码块simulate.py以帮助您跟踪所有部分如何组合在一起。供您参考，您可以通过simulate.py以下链接访问完整代码：

头脑风暴模拟算法

在编写一行代码之前，重要的是首先要弄清楚您的流程在现实生活中的运行方式。这是为了确保当您将其传递给机器时，该过程准确反映了客户将真正体验到的内容。以下是您可能会如何思考电影观众可能会采取的步骤来编写您的算法：

1. 到达剧院，排队，等待购票。
2. 购买从票房一票。
3. 排队等候检票。
4. 获取门票由一个招待员检查。
5. 选择是否在特许摊位排队：
   • 如果他们排队，那么他们就会购买食物。
   • 如果他们没有排队，那么他们会跳到最后一步。
6. 去找到自己的座位。

对于在剧院售票处购买门票的电影观众来说，这是一个循序渐进的过程。您已经可以看到该流程的哪些部分是可以控制的。您可以通过在售票处增加更多的收银员来影响客户等待的时间。

过程中也有一些部分是无法控制的，比如第一步。您无法控制有多少客户到达，或者他们到达的速度有多快。您可以进行猜测，但您不能简单地选择一个数字，因为这不能很好地反映现实。对于此参数，您可以做的最好的事情是使用可用数据来确定合适的到达时间。

考虑到这些因素，是时候构建您的模拟了！

设置环境

在开始构建模拟之前，您需要确保正确配置了开发环境。您要做的第一件事就是导入必要的包。您可以通过import在文件顶部声明语句来做到这一点：

import simpy
import random
import statistics

这些是您将用于为剧院经理构建脚本的主要库。请记住，目标是找到平均等待时间少于 10 分钟的最佳员工人数。要做到这一点，您需要收集每个电影观众到达他们的座位所需的时间长度。下一步是声明一个列表来保存这些时间：

wait_times = []

该列表将包含每个电影观众从到达到坐在座位上花费在电影院中的总时间。您可以在文件的最顶部声明​​此列表，以便您可以在稍后定义的任何函数中使用它。

创建环境：类定义

您要构建的模拟的第一部分是系统的蓝图。这将是事情发生的整体环境，人或物体从一个地方移动到另一个地方。请记住，环境可以是许多不同系统之一，例如银行、洗车场或安全检查站。在这种情况下，环境是电影院，因此这将是您的类的名称：

class Theater(object):
    def __init__(self):
        # More to come!

现在是时候考虑一​​下电影院的各个部分了。当然，还有剧院本身，这就是您所说的环境。稍后，您将environment使用其中一个simpy函数明确地将剧院声明为实际剧院。现在，env简称它并将其添加到类定义中：

class Theater(object):
    def __init__(self, env):
        self.env = env

好吧，剧院里还有什么？您可以通过仔细考虑您之前计划的模拟算法来解决这个问题。当电影观众到达时，他们需要在售票处排队，收银员将在那里等待帮助他们。现在您已经发现了有关剧院环境的两件事：

1. 有收银员。
2. 电影观众可以从他们那里购买门票。

收银员是剧院向顾客提供的一种资源，他们在购票过程中帮助电影观众。现在，您不知道模拟剧院中有多少收银员可用。事实上，这正是您要解决的问题。等待时间如何变化，具体取决于特定晚上工作的收银员数量？

您可以继续调用这个未知变量num_cashiers。此变量将采用的确切值可以稍后整理出来。现在，只要知道它是剧院环境中不可或缺的一部分。将其添加到类定义中：

class Theater(object):
    def __init__(self, env, num_cashiers):
        self.env = env
        self.cashier = simpy.Resource(env, num_cashiers)

在这里，您将新参数添加num_cashiers到您的__init__()定义中。然后，您创建一个资源self.cashier并使用它simpy.Resource()来声明在任何给定时间此环境中可以有多少资源。

您还需要执行一个步骤。收银员不会自己买票吧？他们要帮助电影观众！同样，您知道购买机票的过程需要一定的时间。但时间是多少？

假设您向经理询问了剧院的历史数据，例如员工绩效评估或购票收据。根据此数据，您了解到在售票处出票平均需要 1 到 2 分钟。你如何simpy模仿这种行为？它只需要一行代码：

yield self.env.timeout(random.randint(1, 3))

env.timeout()告诉simpy在经过一定时间后触发事件。在这种情况下，事件是购买了票。

这需要的时间可能是一分钟、两分钟或三分钟。您希望每个电影观众在收银台上花费不同的时间。为此，您可以random.randint()在给定的低值和高值之间选择一个随机数。然后，对于每个电影观众，模拟将等待选定的时间量。

让我们把它包装在一个 tidy 函数中，并将它添加到类定义中：

class Theater(object):
    def __init__(self, env, num_cashiers):
        self.env = env
        self.cashier = simpy.Resource(env, num_cashiers)

    def purchase_ticket(self, moviegoer):
        yield self.env.timeout(random.randint(1, 3))

在 中启动事件的purchase_ticket()是moviegoer，因此它们必须作为必需的参数传递。

而已！您已经选择了一个有时限的资源，定义了它的相关流程，并将其编入了您的类定义中。对于本教程，您还需要声明两个资源：

1. 引导员查票
2. 服务器卖食品

检查经理发送过来的数据后，您确定服务器需要 1 到 5 分钟的时间来完成订单。此外，引座员查票速度非常快，平均3秒！

您需要将这些资源添加到您的类中并定义相应的函数check_ticket()和sell_food(). 你能弄清楚代码应该是什么样子吗？当你有了一个想法时，你可以展开下面的代码块来检查你的理解：

仔细看看新的资源和功能。请注意它们如何遵循与上述相同的格式。sell_food()用于random.randint()生成 1 到 5 分钟之间的随机数，表示电影观众下订单和收到食物所需的时间。

的时间延迟check_ticket()有点不同，因为引导员只需要 3 秒。由于simpy以分钟为单位工作，因此需要将此值作为一分钟的一小部分或3 / 60.

让事情发生：功能定义

现在您需要定义一个函数来运行模拟。run_theater()将负责创建剧院的实例并生成电影观众，直到模拟停止。这个函数应该做的第一件事是创建一个剧院的实例：

def run_theater(env, num_cashiers, num_servers, num_ushers):
    theater = Theater(env, num_cashiers, num_servers, num_ushers)

由于这是主要过程，因此您需要传递迄今为止声明的所有未知数：

• num_cashiers
• num_servers
• num_ushers

这些都是模拟创建和控制环境所需的变量，因此将它们全部传递是绝对重要的。然后，您定义一个变量theater并告诉模拟设置具有一定数量的收银员、服务员和招待员的剧院。

您可能还想以一些在剧院等候的电影观众来开始您的模拟。估计一开门就有几个人准备走了！经理说，预计票房一开放，大约有 3 名观众排队准备买票。您可以告诉模拟继续并通过这个初始组，如下所示：

def run_theater(env, num_cashiers, num_servers, num_ushers):
    theater = Theater(env, num_cashiers, num_servers, num_ushers)

    for moviegoer in range(3):
        env.process(go_to_movies(env, moviegoer, theater))

您range()过去常常让 3 位电影观众在剧院里演出。然后，您使用env.process()告诉模拟准备移动它们穿过剧院。其余的电影观众将在他们自己的时间进入剧院。因此，只要模拟正在运行，该功能就应该不断地将新客户送入影院。

您不知道新观众需要多长时间才能进入影院，因此您决定查看过去的数据。使用带有时间戳的票房收据，您可以了解到电影观众平均每 12 秒就会到达剧院一次。现在你要做的就是告诉函数在生成一个新人之前等待这么长时间：

def run_theater(env, num_cashiers, num_servers, num_ushers):
    theater = Theater(env, num_cashiers, num_servers, num_ushers)

    for moviegoer in range(3):
        env.process(go_to_movies(env, moviegoer, theater))

    while True:
        yield env.timeout(0.20)  # Wait a bit before generating a new person

        # Almost done!...

请注意，您使用十进制数0.20来表示 12 秒。要获得这个数字，您只需将 12 秒除以 60 秒，即一分钟的秒数。

等待后，该函数应递增moviegoer1 并生成下一个人。该发电机的功能是您用来初始化第3名影迷同一个：

def run_theater(env, num_cashiers, num_servers, num_ushers):
    theater = Theater(env, num_cashiers, num_servers, num_ushers)

    for moviegoer in range(3):
        env.process(go_to_movies(env, moviegoer, theater))

    while True:
        yield env.timeout(0.20)  # Wait a bit before generating a new person

        moviegoer += 1
        env.process(go_to_movies(env, moviegoer, theater))

而已！当您调用此函数时，模拟将生成 3 个电影观众开始并使用 开始在影院中移动他们go_to_movies()。之后，新的电影观众将以12秒的间隔到达剧院，并在他们自己的时间穿过剧院。

计算等待时间：函数定义

此时，您应该有一个列表wait_times，其中包含每个电影观众到达他们的座位所花费的总时间。现在您需要定义一个函数来帮助计算moviegoer从他们到达到他们完成检查他们的票所花费的平均时间。get_average_wait_time()只是这样做：

def get_average_wait_time(wait_times):
    average_wait = statistics.mean(wait_times)

此函数将您的wait_times列表作为参数并用于statistics.mean()计算平均等待时间。

由于您正在创建将由电影院经理使用的脚本，因此您需要确保用户可以轻松读取输出。您可以添加一个调用函数calculate_wait_time()来执行此操作：

def calculate_wait_time(arrival_times, departure_times):
    average_wait = statistics.mean(wait_times)
    # Pretty print the results
    minutes, frac_minutes = divmod(average_wait, 1)
    seconds = frac_minutes * 60
    return round(minutes), round(seconds)

函数的最后一部分用于divmod()以分钟和秒为单位返回结果，因此管理人员可以轻松了解程序的输出。

如何运行模拟

只需再多写几行代码，您就可以亲眼目睹您的模拟变得栩栩如生。但首先，这里是您迄今为止定义的函数和类的概述：

• Theater：这个类定义用作您要模拟的环境的蓝图。它确定有关该环境的一些信息，例如可用的资源类型以及与它们相关联的进程。

• go_to_movies()：该函数明确请求使用资源，经过关联的过程，然后将其释放给下一个电影观众。

• run_theater()：此功能控制模拟。它使用Theater类蓝图创建剧院的实例，然后调用go_to_movies()生成并移动人们穿过剧院。

• get_average_wait_time()：此功能可计算出moviegoer通过剧院所需的平均时间。

• calculate_wait_time()：此功能可确保最终输出易于用户阅读。

• get_user_input()：这个函数允许用户定义一些参数，比如有多少可用的收银员。

• main()：此功能可确保您的脚本在命令行中正常运行。

现在，您只需要另外两行代码来调用您的 main 函数：

if __name__ == '__main__':
    main()

这样，您的脚本就可以运行了！打开您的终端，导航到您存储的位置simulate.py，然后运行以下命令：

$ python simulate.py
Input # of cashiers working:

系统将提示您选择仿真所需的参数。以下是使用默认参数的输出：

$ python simulate.py
Input # of cashiers working: 1
Input # of servers working: 1
Input # of ushers working: 1
Running simulation...
The average wait time is 42 minutes and 53 seconds.

哇！等了好久啊！

何时改变现状

请记住，您的目标是向经理提供解决方案，了解他需要多少员工才能将等待时间保持在 10 分钟以内。为此，您需要调整参数以查看哪些数字提供了最佳解决方案。

首先，尝试一些完全疯狂的事情并最大限度地利用资源！假设有 100 个收银员、100 个服务器和 100 个引导员在这个剧院工作。当然，这是不可能的，但是使用高得离谱的数字会很快告诉您系统的限制是多少。现在就试试：

$ python simulate.py
Input # of cashiers working: 100
Input # of servers working: 100
Input # of ushers working: 100
Running simulation...
The average wait time is 3 minutes and 29 seconds.

即使您最大限度地利用了资源，您的等待时间也只能缩短到 3 分半钟。现在尝试更改数字，看看是否可以像经理要求的那样将等待时间缩短到 10 分钟。你想出了什么解决方案？您可以展开下面的代码块以查看一种可能的解决方案：

此时，您将向经理展示您的结果并提出建议以帮助改进剧院。例如，为了降低成本，他可能希望在剧院前面安装 10 个售票亭，而不是每晚都安排 10 个收银员。

结论

在本教程中，您学习了如何使用该框架在 Python 中构建和运行模拟simpy。您已经了解系统如何让代理进行处理，以及如何创建这些系统的虚拟表示以加强它们免受拥塞和延迟的影响。虽然模拟的类型可能会有所不同，但整体执行是相同的！您将能够将在这里学到的知识应用到各种不同的场景中。

现在你可以：

• 逐步头脑风暴模拟算法
• 在 Python 中创建一个虚拟环境simpy
• 定义代表代理和进程的函数
• 更改模拟参数以找到最佳解决方案

你可以用它做很多事情simpy，所以不要让你的探索就此停止。将您学到的知识应用到新的场景中。您的解决方案可以帮助人们节省宝贵的时间和金钱，所以深入了解您可以优化的其他流程！您可以通过单击下面的链接下载您在本教程中构建的脚本的源代码：