Python SimPy 模擬系列 (1)

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

在物流行業/工廠製造業/餐飲服務業存在大量急需優化的場景， 例如：

• 如何最優化快遞分揀人員的排班表以滿足雙十一突發的快遞件量
• 如何估算餐廳在用餐高峰的排隊時長
• 估算特定工序下，工廠生產所需要的物料成本/人力成本/時間成本

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

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

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

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

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

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

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

官方資料

SimPy 是一個基於標準 Python 以進程為基礎的離散事件模擬框架。

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

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

• SimPy 的官方文檔可以查詢自：https://simpy.readthedocs.io/en/latest/
• SimPy 代碼託管網址：https://bitbucket.org/simpy/simpy/src/default/docs/index.rst?fileviewer=file-view-default
• PyPy 網址：https://pypi.python.org/pypi/simpy
• 原理性介紹的ppy: https://stefan.sofa-rockers.org/downloads/simpy-ep14.pdf

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, randintseed(23)import simpyclass 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 randomimport simpyRANDOM_SEED = 42NEW_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 simulationprint(Bank renege)random.seed(RANDOM_SEED)env = simpy.Environment()# Start processes and runcounter = 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 simpyimport randomPROCESS_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(fitem_{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=fm_{i})) env.run()

結語

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

接下來的文章計劃是：

• 介紹如何構建一個基於時間動態的人力資源排班表，來模擬不同時段人力的分布，比如流水線上不同崗位在不同的時間段需求的人力資源是不均等，我們可以通過調崗的形式來達到人力資源調優，讓人力資源在時間分布上最優。
• 介紹如何一個隊列，同時實現時間先後優先和優先順序上優先，來模擬比如銀行櫃檯客戶排隊 vip 進行插隊的情景
• 啟動一個翻譯 SimPy 官方文檔的眾包計劃，希望在國內降低大眾學習 SimPy 的語言成本

本文作者：

• 趙鵬 @Pong
• 韓藍毅 @韓藍毅