本系列文章旨在介紹 SimPy 在工業仿真中的應用。

在物流行業/工廠制造業/餐飲服務業存在大量急需優化的場景， 例如：如何最優化快遞分揀人員的排班表以滿足雙十一突發的快遞件量

如何估算餐廳在用餐高峰的排隊時長

估算特定工序下，工廠生產所需要的物料成本/人力成本/時間成本

這類場景無法通過常規算法求出最優解, 但是我們可以通過大量業務實踐中總結出一些接近的次優解。

實際生產中，隨時調整廠房的生產線來試驗最優解是非常昂貴的。引進仿真技術，可以給業務研究員無限的自由度去調整驗證不同的優化方案。仿真的成本無非是計算機的算力，以及程序員編寫業務邏輯的時間。

行業上其實已經存在一些工業仿真軟件。但這類仿真軟件往往針對某些特定場景高度定制化，數據埋點往往不全，缺乏通用的數據庫接口，難以結合真實業務產生的數據進行仿真，這樣就失去與真實業務進行比較的可能。

利用 SimPy 我們可以構建一套完全開源的仿真方案，可以完全私有定制業務場景。利用 Python 強大的生態，仿真數據從來源到輸出分析，可以銜接所有開源流行的數據分析框架。

目前，我們已經利用 SimPy 仿真模擬物流核心分揀業務，結合 MySQL，Tableau，Pandas，Spark 構建一整套完整的報表可視化分析體系，已經能夠成功應用于現代物流中，為分揀業務提供持續優化改良方案。

我們創造性地解決了一些原有軟件仿真中欠缺的環節，這些內容將會在接下來的文章中分享。

作為本系列文章的開篇，我們將簡要地介紹 SimPy 框架的基本理念。

官方資料

SimPy 是一個基于標準 Python 以進程為基礎的離散事件仿真框架。

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

仿真運行速度非?？?#xff0c;仿真中的模擬時間長短不影響仿真運行效率，仿真中的模擬時間單位可以任意指定，一秒、一年、一小時都是允許的。

SimPy 安裝

SimPy 可以同時在 Python 2 (>=2.7)以及 Python 3(>=3.2)上運行。只要有pip，輕松安裝?！? pip install simpy”

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

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

SimPy 核心概念

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

process 表達為 generators (生成器)， 構建event(事件)并通過 yield 語句拋出事件。

當一個進程拋出事件，進程會被暫停，直到事件被激活(triggered)。多個進程可以等待同一個事件。 SimPy 會按照這些進程拋出的事件激活的先后， 來恢復進程。

其實中最重要的一類事件是 Timeout， 這類事件允許一段時間后再被激活， 用來表達一個進程休眠或者保持當前的狀態持續指定的一段時間。這類事件通過 Environment.timeout來調用。

Environment

Environment 決定仿真的起點/終點， 管理仿真元素之間的關聯, 主要 API 有simpy.Environment.process - 添加仿真進程

simpy.Environment.event - 創建事件

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

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

simpy.Environment.run - 仿真啟動

樣例代碼說明 API:

下面是來自官方文檔的兩個例子:第一個例子， 描述如何定義一個進程， 并添加到 env 內， 簡單展示啟動仿真的代碼結構

第二個例子， 描述一個汽車駕駛一段時間后停車充電， 汽車駕駛進程和電池充電進程通過事件的激活來相互影響

Example 1

import simpy

# 定義一個汽車進程

def car(env):

while True:

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

parking_duration = 5

yield env.timeout(parking_duration) # 進程延時 5s

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

trip_duration = 2

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

# 仿真啟動

env = simpy.Environment() # 實例化環境

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

env.run(until=15) # 設定仿真結束條件, 這里是 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):

"""駕駛進程"""

while True:

# 駕駛 20-40 分鐘

print("開始駕駛 時間: ", env.now)

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

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

# 停車 1-6 小時

print("開始停車 時間: ", env.now)

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

self.bat_ctrl_reactivate = env.event()

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

print("結束停車 時間:", env.now)

def bat_ctrl(self, env):

"""電池充電進程"""

while True:

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

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

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

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

print("充電程序結束 時間:", 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 也是另外一類重要的核心概念, 但凡仿真中涉及的人力資源以及工藝上的物料消耗都會抽象用 Resource 來表達, 主要的 method 是 request. Store 處理各種優先級的隊列問題, 表現跟 queue 一致, 通過 method get / put 存放 item

Store - 抽象隊列simpy.Store - 存取 item 遵循仿真時間上的先到后到

simpy.PriorityStore - 存取 item 遵循仿真時間上的先到后到同時考慮人為添加的優先級

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

simpy.Container - 表達連續/不可分的物質, 包括液體/氣體的存放, 存取的是一個 float 數值

Resource - 抽象資源simpy.Resource - 表達人力資源或某種限制條件, 例如某個工序可調用的工人數, 可以調用的機器數

simpy.PriorityResource - 兼容Resource的功能, 添加可以插隊的功能, 高優先級的進程可以優先調用資源, 但只能是在前一個被服務的進程結束以后進行插隊

simpy.PreemptiveResource - 兼容Resource的功能, 添加可以插隊的功能, 高優先級的進程可以打斷正在被服務的進程進行插隊

樣例代碼說明 API:

Example 3

"""

銀行排隊服務例子

情景:

一個柜臺對客戶進行服務, 服務耗時, 客戶等候過長會離開柜臺

"""

import random

import simpy

RANDOM_SEED = 42

NEW_CUSTOMERS = 5 # 客戶數

INTERVAL_CUSTOMERS = 10.0 # 客戶到達的間距時間

MIN_PATIENCE = 1 # 客戶等待時間, 最小

MAX_PATIENCE = 3 # 客戶等待時間, 最大

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

"""進程用于生成客戶"""

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):

"""一個客戶表達為一個協程, 客戶到達, 被服務, 然后離開"""

arrive = env.now

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

with counter.request() as req:

patience = random.uniform(MIN_PATIENCE, MAX_PATIENCE)

# 等待柜員服務或者超出忍耐時間離開隊伍

results = yield req | env.timeout(patience)

wait = env.now - arrive

if req in results:

# 到達柜臺

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:

# 沒有服務到位

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

"""

工廠工序和傳送帶

情景:

一個機器處理物件, 處理完畢后放上傳送帶, 傳送帶傳送一段時間后到達下個一個機器設備.

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

"""

import simpy

import random

PROCESS_TIME = 0.5 # 處理時間

CON_BELT_TIME = 3 # 傳送帶時間

WORKER_NUM = 2 # 每個機器的工人數/資源數

MACHINE_NUM = 2 # 機器數

MEAN_TIME = 0.2 # 平均每個物件的到達時間間距

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) # 傳送帶傳送時間

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):

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

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

def process(item):

"""模擬一個工人的工作進程"""

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):

"""模擬物件的到達"""

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__':

# 實例環境

env = simpy.Environment()

# 設備前的物件隊列

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()

結語

文章暫時結束，文章主要通過代碼來展示 SimPy 的仿真能力。

接下來的文章計劃是：介紹如何構建一個基于時間動態的人力資源排班表，來模擬不同時段人力的分布，比如流水線上不同崗位在不同的時間段需求的人力資源是不均等，我們可以通過調崗的形式來達到人力資源調優，讓人力資源在時間分布上最優。

介紹如何一個隊列，同時實現時間先后優先和優先級上優先，來模擬比如銀行柜臺客戶排隊 vip 進行插隊的情景

啟動一個翻譯 SimPy 官方文檔的眾包計劃，希望在國內降低大眾學習 SimPy 的語言成本

本文作者：

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。