本系列文章旨在介紹 SimPy 在工業(yè)仿真中的應(yīng)用。

在物流行業(yè)/工廠制造業(yè)/餐飲服務(wù)業(yè)存在大量急需優(yōu)化的場(chǎng)景， 例如：如何最優(yōu)化快遞分揀人員的排班表以滿足雙十一突發(fā)的快遞件量

如何估算餐廳在用餐高峰的排隊(duì)時(shí)長(zhǎng)

估算特定工序下，工廠生產(chǎn)所需要的物料成本/人力成本/時(shí)間成本

這類場(chǎng)景無(wú)法通過(guò)常規(guī)算法求出最優(yōu)解, 但是我們可以通過(guò)大量業(yè)務(wù)實(shí)踐中總結(jié)出一些接近的次優(yōu)解。

實(shí)際生產(chǎn)中，隨時(shí)調(diào)整廠房的生產(chǎn)線來(lái)試驗(yàn)最優(yōu)解是非常昂貴的。引進(jìn)仿真技術(shù)，可以給業(yè)務(wù)研究員無(wú)限的自由度去調(diào)整驗(yàn)證不同的優(yōu)化方案。仿真的成本無(wú)非是計(jì)算機(jī)的算力，以及程序員編寫(xiě)業(yè)務(wù)邏輯的時(shí)間。

行業(yè)上其實(shí)已經(jīng)存在一些工業(yè)仿真軟件。但這類仿真軟件往往針對(duì)某些特定場(chǎng)景高度定制化，數(shù)據(jù)埋點(diǎn)往往不全，缺乏通用的數(shù)據(jù)庫(kù)接口，難以結(jié)合真實(shí)業(yè)務(wù)產(chǎn)生的數(shù)據(jù)進(jìn)行仿真，這樣就失去與真實(shí)業(yè)務(wù)進(jìn)行比較的可能。

利用 SimPy 我們可以構(gòu)建一套完全開(kāi)源的仿真方案，可以完全私有定制業(yè)務(wù)場(chǎng)景。利用 Python 強(qiáng)大的生態(tài)，仿真數(shù)據(jù)從來(lái)源到輸出分析，可以銜接所有開(kāi)源流行的數(shù)據(jù)分析框架。

目前，我們已經(jīng)利用 SimPy 仿真模擬物流核心分揀業(yè)務(wù)，結(jié)合 MySQL，Tableau，Pandas，Spark 構(gòu)建一整套完整的報(bào)表可視化分析體系，已經(jīng)能夠成功應(yīng)用于現(xiàn)代物流中，為分揀業(yè)務(wù)提供持續(xù)優(yōu)化改良方案。

我們創(chuàng)造性地解決了一些原有軟件仿真中欠缺的環(huán)節(jié)，這些內(nèi)容將會(huì)在接下來(lái)的文章中分享。

作為本系列文章的開(kāi)篇，我們將簡(jiǎn)要地介紹 SimPy 框架的基本理念。

官方資料

SimPy 是一個(gè)基于標(biāo)準(zhǔn) Python 以進(jìn)程為基礎(chǔ)的離散事件仿真框架。

SimPy 中的進(jìn)程是由 Python 生成器構(gòu)成，生成器的特性可以模擬具有主動(dòng)性的物件，比如客戶、汽車、或者中介等等。SimPy也提供多種類的共享資源(shared resource)來(lái)描述擁擠點(diǎn)(比如服務(wù)器、收銀臺(tái)和隧道)。

仿真運(yùn)行速度非常快，仿真中的模擬時(shí)間長(zhǎng)短不影響仿真運(yùn)行效率，仿真中的模擬時(shí)間單位可以任意指定，一秒、一年、一小時(shí)都是允許的。

SimPy 安裝

SimPy 可以同時(shí)在 Python 2 (>=2.7)以及 Python 3(>=3.2)上運(yùn)行。只要有pip，輕松安裝。“$ pip install simpy”

手動(dòng)安裝 SimPy 也非常方便。提取存檔，打開(kāi)存放 SimPy 的 terminal 窗口，然后輸入：“$ python setup.py install”

你可以選擇性地運(yùn)行 SimPy 測(cè)試文件以了解軟件是否可行。前提是要安裝 pytest 包。并在 SimPy 的安裝路徑下運(yùn)行下列命令行：“$ py.test --pyargs simpy”

SimPy 核心概念

SimPy 是離散事件驅(qū)動(dòng)的仿真庫(kù)。所有活動(dòng)部件，例如車輛、顧客,、即便是信息，都可以用 process (進(jìn)程) 來(lái)模擬。這些 process 存放在 environment (環(huán)境) 。所有 process 之間，以及與environment 之間的互動(dòng)，通過(guò) event (事件) 來(lái)進(jìn)行.

process 表達(dá)為 generators (生成器)， 構(gòu)建event(事件)并通過(guò) yield 語(yǔ)句拋出事件。

當(dāng)一個(gè)進(jìn)程拋出事件，進(jìn)程會(huì)被暫停，直到事件被激活(triggered)。多個(gè)進(jìn)程可以等待同一個(gè)事件。 SimPy 會(huì)按照這些進(jìn)程拋出的事件激活的先后， 來(lái)恢復(fù)進(jìn)程。

其實(shí)中最重要的一類事件是 Timeout， 這類事件允許一段時(shí)間后再被激活， 用來(lái)表達(dá)一個(gè)進(jìn)程休眠或者保持當(dāng)前的狀態(tài)持續(xù)指定的一段時(shí)間。這類事件通過(guò) Environment.timeout來(lái)調(diào)用。

Environment

Environment 決定仿真的起點(diǎn)/終點(diǎn)， 管理仿真元素之間的關(guān)聯(lián), 主要 API 有simpy.Environment.process - 添加仿真進(jìn)程

simpy.Environment.event - 創(chuàng)建事件

simpy.Environment.timeout - 提供延時(shí)(timeout)事件

simpy.Environment.until - 仿真結(jié)束的條件(時(shí)間或事件)

simpy.Environment.run - 仿真啟動(dòng)

樣例代碼說(shuō)明 API:

下面是來(lái)自官方文檔的兩個(gè)例子:第一個(gè)例子， 描述如何定義一個(gè)進(jìn)程， 并添加到 env 內(nèi)， 簡(jiǎn)單展示啟動(dòng)仿真的代碼結(jié)構(gòu)

第二個(gè)例子， 描述一個(gè)汽車駕駛一段時(shí)間后停車充電， 汽車駕駛進(jìn)程和電池充電進(jìn)程通過(guò)事件的激活來(lái)相互影響

Example 1

import simpy

# 定義一個(gè)汽車進(jìn)程

def car(env):

while True:

print('Start parking at %d' % env.now)

parking_duration = 5

yield env.timeout(parking_duration) # 進(jìn)程延時(shí) 5s

print('Start driving at %d' % env.now)

trip_duration = 2

yield env.timeout(trip_duration) # 延時(shí) 2s

# 仿真啟動(dòng)

env = simpy.Environment() # 實(shí)例化環(huán)境

env.process(car(env)) # 添加汽車進(jìn)程

env.run(until=15) # 設(shè)定仿真結(jié)束條件, 這里是 15s 后停止

Example 2

from random import seed, randint

seed(23)

import simpy

class EV:

def __init__(self, env):

self.env = env

self.drive_proc = env.process(self.drive(env))

self.bat_ctrl_proc = env.process(self.bat_ctrl(env))

self.bat_ctrl_reactivate = env.event()

self.bat_ctrl_sleep = env.event()

def drive(self, env):

"""駕駛進(jìn)程"""

while True:

# 駕駛 20-40 分鐘

print("開(kāi)始駕駛 時(shí)間: ", env.now)

yield env.timeout(randint(20, 40))

print("停止駕駛 時(shí)間: ", env.now)

# 停車 1-6 小時(shí)

print("開(kāi)始停車 時(shí)間: ", env.now)

self.bat_ctrl_reactivate.succeed() # 激活充電事件

self.bat_ctrl_reactivate = env.event()

yield env.timeout(randint(60, 360)) & self.bat_ctrl_sleep # 停車時(shí)間和充電程序同時(shí)都滿足

print("結(jié)束停車 時(shí)間:", env.now)

def bat_ctrl(self, env):

"""電池充電進(jìn)程"""

while True:

print("充電程序休眠 時(shí)間:", env.now)

yield self.bat_ctrl_reactivate # 休眠直到充電事件被激活

print("充電程序激活 時(shí)間:", env.now)

yield env.timeout(randint(30, 90))

print("充電程序結(jié)束 時(shí)間:", env.now)

self.bat_ctrl_sleep.succeed()

self.bat_ctrl_sleep = env.event()

def main():

env = simpy.Environment()

ev = EV(env)

env.run(until=300)

if __name__ == '__main__':

main()

Resource 和 Store

Resource/Store 也是另外一類重要的核心概念, 但凡仿真中涉及的人力資源以及工藝上的物料消耗都會(huì)抽象用 Resource 來(lái)表達(dá), 主要的 method 是 request. Store 處理各種優(yōu)先級(jí)的隊(duì)列問(wèn)題, 表現(xiàn)跟 queue 一致, 通過(guò) method get / put 存放 item

Store - 抽象隊(duì)列simpy.Store - 存取 item 遵循仿真時(shí)間上的先到后到

simpy.PriorityStore - 存取 item 遵循仿真時(shí)間上的先到后到同時(shí)考慮人為添加的優(yōu)先級(jí)

simpy.FilterStore - 存取 item 遵循仿真時(shí)間上的先到后到, 同時(shí)隊(duì)列中存在分類, 按照不同類別進(jìn)行存取

simpy.Container - 表達(dá)連續(xù)/不可分的物質(zhì), 包括液體/氣體的存放, 存取的是一個(gè) float 數(shù)值

Resource - 抽象資源simpy.Resource - 表達(dá)人力資源或某種限制條件, 例如某個(gè)工序可調(diào)用的工人數(shù), 可以調(diào)用的機(jī)器數(shù)

simpy.PriorityResource - 兼容Resource的功能, 添加可以插隊(duì)的功能, 高優(yōu)先級(jí)的進(jìn)程可以優(yōu)先調(diào)用資源, 但只能是在前一個(gè)被服務(wù)的進(jìn)程結(jié)束以后進(jìn)行插隊(duì)

simpy.PreemptiveResource - 兼容Resource的功能, 添加可以插隊(duì)的功能, 高優(yōu)先級(jí)的進(jìn)程可以打斷正在被服務(wù)的進(jìn)程進(jìn)行插隊(duì)

樣例代碼說(shuō)明 API:

Example 3

"""

銀行排隊(duì)服務(wù)例子

情景:

一個(gè)柜臺(tái)對(duì)客戶進(jìn)行服務(wù), 服務(wù)耗時(shí), 客戶等候過(guò)長(zhǎng)會(huì)離開(kāi)柜臺(tái)

"""

import random

import simpy

RANDOM_SEED = 42

NEW_CUSTOMERS = 5 # 客戶數(shù)

INTERVAL_CUSTOMERS = 10.0 # 客戶到達(dá)的間距時(shí)間

MIN_PATIENCE = 1 # 客戶等待時(shí)間, 最小

MAX_PATIENCE = 3 # 客戶等待時(shí)間, 最大

def source(env, number, interval, counter):

"""進(jìn)程用于生成客戶"""

for i in range(number):

c = customer(env, 'Customer%02d' % i, counter, time_in_bank=12.0)

env.process(c)

t = random.expovariate(1.0 / interval)

yield env.timeout(t)

def customer(env, name, counter, time_in_bank):

"""一個(gè)客戶表達(dá)為一個(gè)協(xié)程, 客戶到達(dá), 被服務(wù), 然后離開(kāi)"""

arrive = env.now

print('%7.4f %s: Here I am' % (arrive, name))

with counter.request() as req:

patience = random.uniform(MIN_PATIENCE, MAX_PATIENCE)

# 等待柜員服務(wù)或者超出忍耐時(shí)間離開(kāi)隊(duì)伍

results = yield req | env.timeout(patience)

wait = env.now - arrive

if req in results:

# 到達(dá)柜臺(tái)

print('%7.4f %s: Waited %6.3f' % (env.now, name, wait))

tib = random.expovariate(1.0 / time_in_bank)

yield env.timeout(tib)

print('%7.4f %s: Finished' % (env.now, name))

else:

# 沒(méi)有服務(wù)到位

print('%7.4f %s: RENEGED after %6.3f' % (env.now, name, wait))

# Setup and start the simulation

print('Bank renege')

random.seed(RANDOM_SEED)

env = simpy.Environment()

# Start processes and run

counter = simpy.Resource(env, capacity=1)

env.process(source(env, NEW_CUSTOMERS, INTERVAL_CUSTOMERS, counter))

env.run()

Example 4

# python 3.6 with SimPy

"""

工廠工序和傳送帶

情景:

一個(gè)機(jī)器處理物件, 處理完畢后放上傳送帶, 傳送帶傳送一段時(shí)間后到達(dá)下個(gè)一個(gè)機(jī)器設(shè)備.

[last_q][machine1] ----[con_belt]----> [next_q][machine2]

"""

import simpy

import random

PROCESS_TIME = 0.5 # 處理時(shí)間

CON_BELT_TIME = 3 # 傳送帶時(shí)間

WORKER_NUM = 2 # 每個(gè)機(jī)器的工人數(shù)/資源數(shù)

MACHINE_NUM = 2 # 機(jī)器數(shù)

MEAN_TIME = 0.2 # 平均每個(gè)物件的到達(dá)時(shí)間間距

def con_belt_process(env,

con_belt_time,

package,

next_q):

"""模擬傳送帶的行為"""

while True:

print(f"{round(env.now, 2)} - item: {package} - start moving ")

yield env.timeout(con_belt_time) # 傳送帶傳送時(shí)間

next_q.put(package)

print(f"{round(env.now, 2)} - item: {package} - end moving")

env.exit()

def machine(env: simpy.Environment,

last_q: simpy.Store,

next_q: simpy.Store,

machine_id: str):

"""模擬一個(gè)機(jī)器, 一個(gè)機(jī)器就可以同時(shí)處理多少物件 取決資源數(shù)(工人數(shù))"""

workers = simpy.Resource(env, capacity=WORKER_NUM)

def process(item):

"""模擬一個(gè)工人的工作進(jìn)程"""

with workers.request() as req:

yield req

yield env.timeout(PROCESS_TIME)

env.process(con_belt_process(env, CON_BELT_TIME, item, next_q))

print(f'{round(env.now, 2)} - item: {item} - machine: {machine_id} - processed')

while True:

item = yield last_q.get()

env.process(process(item))

def generate_item(env,

last_q: simpy.Store,

item_num: int=100):

"""模擬物件的到達(dá)"""

for i in range(item_num):

print(f'{round(env.now, 2)} - item: item_{i} - created')

last_q.put(f'item_{i}')

t = random.expovariate(1 / MEAN_TIME)

yield env.timeout(round(t, 1))

if __name__ == '__main__':

# 實(shí)例環(huán)境

env = simpy.Environment()

# 設(shè)備前的物件隊(duì)列

last_q = simpy.Store(env)

next_q = simpy.Store(env)

env.process(generate_item(env, last_q))

for i in range(MACHINE_NUM):

env.process(machine(env, last_q, next_q, machine_id=f'm_{i}'))

env.run()

結(jié)語(yǔ)

文章暫時(shí)結(jié)束，文章主要通過(guò)代碼來(lái)展示 SimPy 的仿真能力。

接下來(lái)的文章計(jì)劃是：介紹如何構(gòu)建一個(gè)基于時(shí)間動(dòng)態(tài)的人力資源排班表，來(lái)模擬不同時(shí)段人力的分布，比如流水線上不同崗位在不同的時(shí)間段需求的人力資源是不均等，我們可以通過(guò)調(diào)崗的形式來(lái)達(dá)到人力資源調(diào)優(yōu)，讓人力資源在時(shí)間分布上最優(yōu)。

介紹如何一個(gè)隊(duì)列，同時(shí)實(shí)現(xiàn)時(shí)間先后優(yōu)先和優(yōu)先級(jí)上優(yōu)先，來(lái)模擬比如銀行柜臺(tái)客戶排隊(duì) vip 進(jìn)行插隊(duì)的情景

啟動(dòng)一個(gè)翻譯 SimPy 官方文檔的眾包計(jì)劃，希望在國(guó)內(nèi)降低大眾學(xué)習(xí) SimPy 的語(yǔ)言成本

本文作者：

總結(jié)

以上是生活随笔為你收集整理的python的仿真效果好吗_Python SimPy 仿真系列 (1)的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。