使用Python的SimPy进行制造仿真-技术圈

使用Python和SimPy创建一个吉他工厂仿真

仿真是一种基于模型的活动。它通过对系统模型的试验达到分析与研究系统的目的。

仿真技术是再现系统动态行为、分析系统配置与参数是否合理、预测瓶颈工序、判断系统性能是否满足规定要求、为制造系统的设计和运行提供决策支持。

在本文中，我们将使用SimPy来建立一个吉他工厂，介绍一些非常精彩的东西，你可以将这些用到你自己的仿真案例中。

1.SimPy

首先，什么是SimPy？SimPy文档中将其定义为：“SimPy是基于过程的离散事件的标准Python模拟框架”。如果您没有安装SimPy，用下面的代码进行SimPy的安装：

pip install simpy

2.吉他工厂

首先我们将从头开始建立一个吉他工厂，从非常简单的东西到更完善的仿真系统。在此示例中我们生产一种吉他，吉他的木质主体分为两部分：琴身和琴颈，这两部分是分别生产的，但使用的都是相同类型的木材，然后将这些半成品送到给喷涂工序车间进行喷涂。最后，将喷涂好的琴身、琴颈和电子元件组合在一起，从而完成一个吉他的生产。
我们先看看生产业务流程图：

先解释一下业务流程：

有2个主要容器（库存）：木材仓库和电子元件仓库。这些容器中有N单位数量木材/电子元件，这些原材料将在生产流程中使用。
琴身部件从木材仓库取出1单位木材，生产成1个琴身，然后将其存储在琴身仓库中。琴颈也一样，但是从1单位木材上得到2个琴颈。琴颈存储在琴颈仓库中，琴身和琴颈都在等待喷涂。
喷涂车间给琴身和琴颈上漆，然后将它们存储在已喷涂的琴身仓库（待组装琴身仓库）和已喷涂的琴颈仓库（待组装琴颈仓库）中。
1个琴身和1个琴颈和1单位电子元件在组装车间就组装出来1个成品吉他，组装完成后存放在成品仓库中。
生产完成一定量的吉他成品，商店安排人来取货。
当木材或电子元件的原材料库存低到一定水平时，会联系供应商进行原材料供货。T天后，供应商送货到达工厂的仓库，原材料库存增加。

3.循序渐进仿真

1）简单例子

我们从易到难，先看看最简单的业务模型，琴身和琴颈车间分别从木材仓库取1单位木材，分别生产出1个琴身和2个琴颈，放在成品仓库中（我们暂时称之为产品仓库），如下图：

代码如下：

import simpywood_capacity = 1000initial_wood = 500dispatch_capacity = 500
class Guitar_Factory:    def __init__(self, env):        self.wood = simpy.Container(env, capacity = wood_capacity, init = initial_wood)        self.dispatch = simpy.Container(env ,capacity = dispatch_capacity, init = 0)

我们开始导入SimPy，之后创建Guitar_Factory类，并添加两个容器（仓库），一个是木材仓库，最大库存为上面设置的1000，初始化库存为500，另一个是成品仓库，最大库存为上面设置的500，初始化库存为0。注意env参数，这个是SimPy的一个环境，我们后面会说明。

现在我们创建琴身和琴颈的生产过程。

def body_maker(env, guitar_factory):    while True:        yield guitar_factory.wood.get(1)        body_time = 1        yield env.timeout(body_time)        yield guitar_factory.dispatch.put(1)
def neck_maker(env, guitar_factory):    while True:        yield guitar_factory.wood.get(1)        neck_time = 1        yield env.timeout(neck_time)        yield guitar_factory.dispatch.put(2)

我们创建两个生产过程，该函数有两个参数：SimPy环境和guitar_factory类（注意guitar_factory与是我们定义的Guitar_Factory不的实例化）。过程是这样的：

在仿真运行期间，车间从仓库取出1单位木材 guitar_factory.wood.get(1)
在车间加工，一段时间后env.timeout(body_time)，body_time定义为1个时间单位，将会产出一个琴身或琴颈，它模拟了车间的生产时间。
在该时间单位（在我们的示例中为1）过去之后，车间会将产出放入dispatch的容器（仓库）中。琴身车间将用1单位木材制成1个吉他琴身，而琴颈车间将用1单位木材制成2个琴颈。

hours = 8days = 5#定义仿真的时间长度total_time = hours * days
env = simpy.Environment()guitar_factory = Guitar_Factory(env)
body_maker_process = env.process(body_maker(env, guitar_factory))neck_maker_process = env.process(neck_maker(env, guitar_factory))print('仿真开始：')env.run(until = total_time)print(f'仓库Dispatch中分别有%d 琴身和琴颈的库存!' % guitar_factory.dispatch.level)print('仿真结束。')

最后，我们创建一个仿真环境，在这个环境中实例化一个吉他工厂guitar_factory = Guitar_Factory(env)，
琴身和琴颈的生产过程会在env的环境中通过env.process来实例化，通过传递我们定义好的until=total_time给到env.run的运行仿真环境函数，将仿真运行40小时（5天*8小时）。
我们通过guitar_factory.dispatch.level获取成品的库存水平。
运行结果如下：

2）增加一些工序

现在我们增加喷涂工序和组装工序到我们的业务模型。为了达到目的，我们增加：

喷涂前和喷涂后的容器（仓库），和喷涂工序。
电子元件容器（仓库）
组装工序

electronic_capacity = 100initial_electronic = 100
pre_paint_capacity = 100post_paint_capacity = 200
class Guitar_Factory:    def __init__(self, env):        self.wood = simpy.Container(env, capacity = wood_capacity, init = initial_wood)        self.electronic = simpy.Container(env, capacity = electronic_capacity, init = initial_electronic)        self.pre_paint = simpy.Container(env, capacity = pre_paint_capacity, init = 0)        self.post_paint = simpy.Container(env, capacity = post_paint_capacity, init = 0)        self.dispatch = simpy.Container(env ,capacity = dispatch_capacity, init = 0)

以上内容没有什么新的内容，我们增加了电子元件、喷涂前和喷涂后的仓库，并设定他们的最大库存水平和初始化库存水平。

接下来我们定义喷涂和组装工序。

def painter(env, guitar_factory):    while True:        yield guitar_factory.pre_paint.get(10)        paint_time = 4        yield env.timeout(paint_time)        yield guitar_factory.post_paint.put(10)
def assembler(env, guitar_factory):    while True:        yield guitar_factory.post_paint.get(2)        yield guitar_factory.electronic.get(1)        assembling_time = 1        yield env.timeout(assembling_time)        yield guitar_factory.dispatch.put(1)

正如我们在生产琴身和琴颈的一样，我们创建喷涂和组装的工序，喷涂需要4小时，每次能同时喷涂10个，放到一个名为

post_paint的容器（仓库）中 guitar_factory.post_paint.put(10)。

组装工序是使用1个喷涂后的琴身和1个喷涂后的琴颈，这里我们不区分琴颈和琴身，直接提取两个库存

guitar_factory.post_paint.get(2)（这里稍后需要优化），再加上一份电子元件，经过1个小时的组装就得到一个成品。

然后我们添加一些打印输出，创建环境并运行仿真。

env = simpy.Environment()guitar_factory = Guitar_Factory(env)
body_maker_process = env.process(body_maker(env, guitar_factory))neck_maker_process = env.process(neck_maker(env, guitar_factory))painter_process = env.process(painter(env, guitar_factory))assembler_process = env.process(assembler(env, guitar_factory))
print(f'仿真开始：')
env.run(until = total_time)
print(f'喷涂前有%d 琴身和琴颈准备喷涂' % guitar_factory.pre_paint.level)print(f'喷涂后有 %d 琴身和琴颈准备组装' % guitar_factory.post_paint.level)print(f'有 %d 吉他成品!' % guitar_factory.dispatch.level)print(f'----------------------------------')print(f'仿真完成。')

3）库存预警、供应链供货

现在我们添加一些非常酷的东西，到目前为止，我们已经为每一个工序设定了固定的时间，比如我们组装车间需要一个小时来组织吉他。用这个参数，我们生产一个吉他始终都是1个小时，不会有任何的偏差。
实际上，工序生产产品的时间都是在一个平均时间上下波动的，我们假设时间服从正态分布（实际上你应该通过实际的统计数据得到加工时间的分布特征来生成随机数。）

另外我们定义每种工序的数量（你可以理解为工序员工数量或者工序设备数量）。我们从导入随机库和定义参数开始。

import random
num_body = 2mean_body = 1std_body = 0.1
num_neck = 1mean_neck = 1std_neck = 0.2
num_paint = 1mean_paint = 4std_paint = 0.3
num_ensam = 4mean_ensam = 1std_ensam = 0.2

num_body设置了琴身的车间数量，设置为2，mean_body设置了生存一个琴身平均需要的时间，std_body设置了生存琴身的时间的标准差，我们修改我们的代码：

def body_maker(env, guitar_factory):    while True:        yield guitar_factory.wood.get(1)        body_time = random.gauss(mean_body, std_body)        yield env.timeout(body_time)        yield guitar_factory.pre_paint.put(1)
def neck_maker(env, guitar_factory):    while True:        yield guitar_factory.wood.get(1)        neck_time = random.gauss(mean_neck, std_neck)        yield env.timeout(neck_time)        yield guitar_factory.pre_paint.put(2)
def painter(env, guitar_factory):    while True:        yield guitar_factory.pre_paint.get(10)        paint_time = random.gauss(mean_paint, std_paint)        yield env.timeout(paint_time)        yield guitar_factory.post_paint.put(10)
def assembler(env, guitar_factory):    while True:        yield guitar_factory.post_paint.get(1)        yield guitar_factory.electronic.get(1)        assembling_time = max(random.gauss(mean_ensam, std_ensam), 1)        yield env.timeout(assembling_time)        yield guitar_factory.dispatch.put(1)之前的例子timeout的参数我们总是设定为一个固定值，现在我们将这几个工序的生产时间定义为一个随机数。
random.gauss(mean_body, std_body)

之前的例子timeout的参数我们总是设定为一个固定值，现在我们将这几个工序的生产时间定义为一个随机数。

random.gauss(mean_body, std_body)

意思是生成一个平均数为mean_body，标准差为std_body的正态分布随机数（正态分布也叫高斯分布，gauss）。
另外注意，我们的组装时间随机数取为1到1之间的最大值。换句话说，我们说组装吉他的时间永远不会少于一小时。很多情况下都需要这种设定，不然有可能生成的随机数可能是负数（仿真过程将会出错）。

我们必须通过创建新功能来更改工序数量，该功能允许我们创建多个工序。

def body_maker_gen(env, guitar_factory):    for i in range(num_body):        env.process(body_maker(env, guitar_factory))        yield env.timeout(0)
body_gen = env.process(body_maker_gen(env, guitar_factory))

当然，这里另外还有

neck_maker_gen, paint_maker_gen和assembler_maker_gen

三个工序有类似的代码。

通过for循环我们创建2个琴身工序（我们定义了num_body=2），因此，我们有2个琴身工序、1个琴颈生产工序，1个喷涂工序和4个装配工序。

现在我们将创建库存监控和联系供应商的函数。我们不断监控原材料的库存量（level），如果当前库存量低于我们定义的水平，它将致电供应商进行送货。经过一定的供应周期后，一定量的原材料将到达我们原材料仓库。

首先我们先定义我们的库存预警水平。

wood_critial_stock = (((8/mean_body) * num_body +(8/mean_neck) * num_neck) * 3) electronic_critical_stock = (8/mean_ensam) * num_ensam * 2

预警库存的定义，取决于创建琴身或琴颈的平均时间，琴身和琴颈的制造工序数量。当然这里我们也可以直接设定某一个数值。

木材供应周期为2天。
我们在我们的Guitar_Factory类定义里面添加一个预警操作过程。

class Guitar_Factory:    def __init__(self, env):        self.wood = simpy.Container(env, capacity = wood_capacity, init = initial_wood)        self.wood_control = env.process(self.wood_stock_control(env))        self.electronic = simpy.Container(env, capacity = electronic_capacity, init = initial_electronic)        self.electronic_control = env.process(self.electronic_stock_control(env))        self.pre_paint = simpy.Container(env, capacity = pre_paint_capacity, init = 0)        self.post_paint = simpy.Container(env, capacity = post_paint_capacity, init = 0)        self.dispatch = simpy.Container(env ,capacity = dispatch_capacity, init = 0)

    def wood_stock_control(self, env):        yield env.timeout(0)        while True:            if self.wood.level <= wood_critial_stock:                print(f'在第{0}日 第{1}小时，木材库存 ({2})低于预警库存水平下 '.format(                    int(env.now/8), env.now % 8,self.wood.level))                print('联系供应商')                print('----------------------------------')                yield env.timeout(16)                print('在第{0}天 第{1}小时，木材送达'.format(int(env.now/8), env.now % 8))                yield self.wood.put(300)                print('当前库存是：{0}'.format(                    self.wood.level))                print('----------------------------------')                yield env.timeout(8)            else:                yield env.timeout(1)

我们在Guitar_Factory类中的__init__函数中增加wood_stock_control和electronic_stock_control两个过程。我们看看wood_stock_control的过程是怎么样工作的，上述代码没有electronic_stock_control过程，但原理是一样的，你可以自己创建，也可以下载我的代码。

首先yield env.timeout(0)，意味着木材监控进程在仿真开始时就启动，后面的yield env.timeout(16)和yield env.timeout(8)表示如果预警产生供货请求后，16小时后送货到达，然后再等待8小时后，我们恢复库存监控。
while True表示该过程将在模拟运行的所有时间内执行。
然后，它将检查库存水平是否等于或小于先前定义的临界水平。如果库存大于该水平，下一个时间单位再监控yield env.timeout(1)。
当库存水平等于或低于临界水平时，将执行一些打印输出时间和当前库存量，并且联系供应商送货。
2天（16小时）后，木材原材料到达，木材原材料增加300到我们的仓库中。yield self.wood.put(300)。
最后，将打印新库存水平，并且警报将关闭1天（yield env.timeout(8)）。

我们将仿真运行5天，结果如下：

4）独立仓库和商场配送

制作完琴身和琴颈后，我们将琴身和琴颈看作同一种东西，将它们存储在同一个仓库中（等待喷涂），这有点不合理。现在，我们将为琴身和琴颈分别实现单独的容器（仓库），我们可以对其进行适当处理。
流程图如下：

guitars_made = 0
body_pre_paint_capacity = 60neck_pre_paint_capacity = 60body_post_paint_capacity = 120neck_post_paint_capacity = 120
class Guitar_Factory:    def __init__(self, env):        self.wood = simpy.Container(env, capacity = wood_capacity, init = initial_wood)        self.wood_control = env.process(self.wood_stock_control(env))        self.electronic = simpy.Container(env, capacity = electronic_capacity, init = initial_electronic)        self.electronic_control = env.process(self.electronic_stock_control(env))        self.body_pre_paint = simpy.Container(env, capacity = body_pre_paint_capacity, init = 0)        self.neck_pre_paint = simpy.Container(env, capacity = neck_pre_paint_capacity, init = 0)        self.body_post_paint = simpy.Container(env, capacity = body_post_paint_capacity, init = 0)        self.neck_post_paint = simpy.Container(env, capacity = neck_post_paint_capacity, init = 0)        self.dispatch = simpy.Container(env ,capacity = dispatch_capacity, init = 0)        self.dispatch_control = env.process(self.dispatch_guitars_control(env))

我们现在已经知道如何制作一个容器（仓库）了。注意这里新增一个guitars_made变量和dispatch_control方法。当然，我们需要修改body_maker和neck_maker两个函数（工序），把生产出来的琴身和琴颈分开独立仓库存放：

def body_maker(env, guitar_factory):    while True:        yield guitar_factory.wood.get(1)        body_time = random.gauss(mean_body, std_body)        yield env.timeout(body_time)        yield guitar_factory.body_pre_paint.put(1)

现在，我们将琴身存储在body_pre_paint容器中（仓库）。neck_maker函数也一样处理。painter函数（工序）的提取半成品原料来源的仓库（容器）也需要做相应的修改，让painter工序分别在琴身仓库和琴颈仓库提取材料：

def painter(env, guitar_factory):    while True:        yield guitar_factory.body_pre_paint.get(5)        yield guitar_factory.neck_pre_paint.get(5)        paint_time = random.gauss(mean_paint, std_paint)        yield env.timeout(paint_time)        yield guitar_factory.body_post_paint.put(5)        yield guitar_factory.neck_post_paint.put(5)

喷涂好的琴身和琴颈，分别存放在body_post_paint和neck_post_paint容器（仓库）中。

现在，我们像在电子元件或木材上一样建立一个控制过程。这个过程会跟踪吉他成品的库存水平，并通知商店过来取货。

def dispatch_guitars_control(self, env):        global guitars_made        yield env.timeout(0)        while True:            if self.dispatch.level >= 50:                print('成品库存为：{0}, 在第{1}日 第{2}小时 联系了商场取货'.format(self.dispatch.level, int(env.now/8), env.now % 8))                print('----------------------------------')                yield env.timeout(4)                print('在第{0}日 第{1}小时，商场取走{2}吉他'.format(int(env.now/8), env.now % 8,self.dispatch.level))                guitars_made += self.dispatch.level                yield self.dispatch.get(self.dispatch.level)                print('----------------------------------')                yield env.timeout(8)            else:                yield env.timeout(1)

我们创建了一个名为guitars_made全局变量，记录我们的总产量。
如果成品库存水平等于或高于50，我们会通知商店过来取货。4小时后yield env.timeout(4)，他们到达仓库并拿走所有可用的吉他yield self.dispatch.get(self.dispatch.level)。

当他们取走吉他时，我们用语句guitars_made + = self.dispatch.level将吉他的数量累加到guitars_made变量，用于记录我们总共出货了多少吉他。
然后，控制过程将暂停8个单位时间yield env.timeout(8)，当然也可以不暂停，不暂停的话，这里也不会触发，因为成品库存水平是不足的，或者当天不会再来取货。

组装工序也需要修改为分别从两个待组装的仓库中提取半成品。

def assembler(env, guitar_factory):    while True:        yield guitar_factory.body_post_paint.get(1)        yield guitar_factory.neck_post_paint.get(1)        yield guitar_factory.electronic.get(1)        assembling_time = max(random.gauss(mean_ensam, std_ensam), 1)        yield env.timeout(assembling_time)        yield guitar_factory.dispatch.put(1)

最后，添加一些打印信息并运行仿真程序：

print('当前等待喷涂的琴身数量：{0} 和琴颈数量：{1}'.format(    guitar_factory.body_pre_paint.level, guitar_factory.neck_pre_paint.level))print('当前等待组装的琴身数量：{0} 和琴颈数量：{1}'.format(    guitar_factory.body_post_paint.level, guitar_factory.neck_post_paint.level))print(f'当前成品库存量：%d ' % guitar_factory.dispatch.level)print(f'----------------------------------')print('此周期的吉他总生产数量: {0}'.format(guitars_made + guitar_factory.dispatch.level))print(f'----------------------------------')print(f'仿真完成！')

5）监控

我们希望监控每一个时间点上，仿真系统各个节点的状态，比如我们想记录每个时间点的待组装的琴身的库存量并实时作图。

我们增加一个监控过程函数，用来监控我们的等待组装的琴身和琴颈的半成品库存，看最后一行是我们新增的监控过程env_status。

class Guitar_Factory:    def __init__(self, env):        self.wood = simpy.Container(env, capacity = wood_capacity, init = initial_wood)        self.wood_control = env.process(self.wood_stock_control(env))        self.electronic = simpy.Container(env, capacity = electronic_capacity, init = initial_electronic)        self.electronic_control = env.process(self.electronic_stock_control(env))        self.body_pre_paint = simpy.Container(env, capacity = body_pre_paint_capacity, init = 0)        self.neck_pre_paint = simpy.Container(env, capacity = neck_pre_paint_capacity, init = 0)        self.body_post_paint = simpy.Container(env, capacity = body_post_paint_capacity, init = 0)        self.neck_post_paint = simpy.Container(env, capacity = neck_post_paint_capacity, init = 0)        self.dispatch = simpy.Container(env ,capacity = dispatch_capacity, init = 0)        self.dispatch_control = env.process(self.dispatch_guitars_control(env))        self.env_status_monitor = env.process(self.env_status(env))

我们看看env_status是如何实现的。

def env_status(self, env):  global status  status = pd.DataFrame(columns = ["datetime", "dispatch_level",'wood','electronic','body_pre_paint','neck_pre_paint','body_post_paint','neck_post_paint'])  status[["datetime", "dispatch_level",'wood','electronic','body_pre_paint','neck_pre_paint','body_post_paint','neck_post_paint']] = status[["datetime", "dispatch_level",'wood','electronic','body_pre_paint','neck_pre_paint','body_post_paint','neck_post_paint']].astype(int)  while True:    im = plt.plot(status['datetime'], status['neck_pre_paint'],color='#4D9221')    ims.append(im)    plt.title('neck_pre_paint')    plt.pause(0.001)    print('{0}在第{1}日 第{2}小时，成品库存量：{3}'.format(env.now,int(env.now/8), env.now % 8,self.dispatch.level))    tmp = {'datetime':env.now,         'dispatch_level':self.dispatch.level,         'wood':self.wood.level,         'electronic':self.electronic.level,         'body_pre_paint':self.body_pre_paint.level,         'neck_pre_paint':self.neck_pre_paint.level,         'body_post_paint':self.body_post_paint.level,         'neck_post_paint':self.neck_post_paint.level         }    status = status.append([tmp])
    yield env.timeout(1)

我们先定义一个全局变量status，是一个pandas的dataframe类型，并且定义好它的结构，第一列是时间，后面各列是各个节点（容器）)的库存，并把库存列定义为int整型。
这里我们只用body_pre_paint待组装的琴身库存举例。
我们通过stock_list变量组装好一个字典，然后把它追加到status变量中。看起来是这样子的：

然后我们通过matplotlib模块把它用动态图表现出来。
记得程序之前需要导入这个模块，并且用plt.ion()用做动态图。

import matplotlib.pyplot as plt
plt.ion()

在监控函数中我们用下面的代码来每一个时间间隔画一次图片。

plt.plot(status['datetime'], status['neck_pre_paint'],color='#4D9221')

得到neck_pre_paint的库存变化图，如下：

要监控其他库存，只需要修改stock_list，并追加到status中，然后plot作图即可。你也可以对status进行统计各个节点的平均库存。

根据上文的系统配置，我们很简单就发现了等待喷涂的琴颈库存和等待喷涂的琴身的库存不匹配，有点偏高，所以我们可以调整琴颈生产的一些参数，来得到更好的库存水平。

比如修改这个

或者修改这个

至此，我们完成了一个比较完整的仿真！希望您喜欢，并且可以从中获得有用的东西。

本文完整代码，请在公众号内回复“仿真”进行下载。