SimPy: Simulating Real-World Processes With Python
仿真環境：電影院仿真
目標：減少顧客的平均等待時間，少于10分鐘
在開始仿真前，先思考這個仿真過程，顧客在坐下來看電影前需要經過哪些步驟

• 到達影院
• 排隊買票
• 買到票
• 排隊檢票
• 檢查完票
• 決定是否買零食
• 買零食或者直接入場坐下

這些步驟中又一些是可以控制的，比如有多少雇員在賣票或者賣小零食，有一些步驟需要依賴之前的數據進行預測，比如有多少顧客到達，接下來開始仿真過程，首先導入需要的庫

import random import simpy import statistics

記錄優化目標：找到雇員的最佳數量，使所有顧客的平均等待時間小于10分鐘，使用列表存儲顧客等待時間

wait_times = []

1.Creating the Environment: Class Definition

構建仿真第一步是分析系統藍圖，也就是你的整個環境中會發生的事，會從某個地方移動到另一個地方的人或物，環境可以是任何類型的系統，如銀行，洗車所，安全檢查等，在本例中，環境是一個電影院，因此定義類名：Theater

class Theater(object):def __init__(self):pass

現在開始思考電影院的組成部分，首先肯定有theater本身，也就是你的environment，之后，你會用simpy的一些函數補充theater使其更像真實的環境，現在只需要簡單將其添加到類定義中

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

接下來繼續思考電影院還會有什么，通過之前分析的步驟可以知道，當顧客到達的時候，他們需要在指定位置排隊，然后有收銀員幫助顧客購票，也就是說，環境里有以下兩件事：

• 收銀員cashiers
• 顧客可以從收銀員手中買票purchase tickets

cashiers可以看成電影院提供給顧客的資源resource，他們幫助顧客完成買票這個進程process，但是目前你不知道在仿真環境里有多少cashiers，實際上，這就是你需要解決的問題，顧客等待時間也取決于cashier數量，你可以把這個未知數量稱為num_cashiers，實際取值可以之后篩選，現在你只知道cashier是theater環境必不可少的部分，將其添加到類定義中

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

這樣就增加了一個新參數num_cashiers，創建了一個資源self.cashier，使用simpy.Resource()表示在某時刻有多少資源。還需要考慮的是cashier幫顧客買票是需要時間的，可以通過歷史數據得出買一張票大概需要1-3分鐘，那么如何在simpy中表示這個過程？只需要使用timeout這個事件

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

env.timeout()告訴simpy在經過指定時間后去觸發某個事件，在本例中這個事件就是顧客買到票。現在把這個事件封裝到函數里

class Theater(object):def __init__(self, env, num_cashiers):self.env = envself.cashier = simpy.Resource(env, num_cashiers)def purchase_ticket(self, moviegoer):yield self.env.timeout(random.randint(1, 3))

觸發purchase_ticket()事件的事顧客，因此需要傳入moviegoer。目前為止就定義了一個有時間限制的資源，以及與它相關的進程，除了cashier，還有兩類資源需要定義

• 用來檢票的Ushers
• 用來賣食物的Servers

假設經過歷史數據得知，一位Usher檢查一次票只需要3秒，一位Server交易一次需要1-5分鐘，這倆類資源的處理過程如同上面的cashier，代碼類似

class Theater(object):def __init__(self, env, num_cashiers, num_ushers, num_servers):self.env = envself.cashier = simpy.Resource(env, num_cashiers)self.usher = simpy.Resource(env, num_ushers)self.server = simpy.Resource(env, num_servers)def purchase_ticket(self, moviegoer):yield self.env.timeout(random.randint(1, 3))def check_ticket(self, moviegoer):yield self.env.timeout(3 / 60)def sell_food(self, moviegoer):yield self.env.timeout(random.randint(1, 5))

2.Moving Through the Environment: Function Definition

目前為止已經通過定義類組建完了environment，有資源和進程，現在開始模擬一位顧客使用它們，當一位moviegoer到達影院時，就會開始申請各種資源，直到顧客要做的事件全部完成。

def go_to_movies(env, moviegoer, theater):# Moviegoer到達theaterarrival_time = env.now

需要傳入三個參數：

• env：moviegoer的行為在環境中指定
• moviegoer：可以當成是顧客編號
• theater：調用theater中定義好的進程

使用env.now可以獲取每位moviegoer到達時間。moviegoer在theater中的每個process都有對應的requests來申請資源，例如第一個process是purchase_ticket()，需要使用一個cashier資源，moviegoer需要申請獲取cashier才能執行process

cashier是一類共享資源，這意味著許多moviegoer使用相同的cashier，但是同一時刻，一個cashier只能幫助一位moviegoer，因此需要一些等待時間。

def go_to_movies(env, moviegoer, theater):# Moviegoer到達theaterarrival_time = env.nowwith theater.cashier.request() as request:yield requestyield env.process(theater.purchase_ticket(moviegoer))

• theater.cashier.request():moviegoer申請一個cashier
• yield request:如果所有的cashier都在使用中，moviegoer需要等待直到一個cashier空閑
• yield env.process():moviegoer借助可用的cashier完成指定的process，也就是theater.purchase_ticket()

資源使用完后必須被釋放release()，此處使用with語句，資源就會自動釋放。當一個cashier空閑，moviegoer就會花一些時間買票，env.process()告訴仿真進入到Theater實例中運行purchase_ticket()。在檢票這里也是一樣的流程：request,use,release。但是顧客買零食是隨機可選的，這種不確定性的事件可以使用隨機數來表示。

• True:moviegoer申請server然后訂購食物
• False:moviegoer直接入座

def go_to_movies(env, moviegoer, theater):# Moviegoer到達theaterarrival_time = env.nowwith theater.cashier.request() as request:yield requestyield env.process(theater.purchase_ticket(moviegoer))with theater.usher.request() as request:yield requestyield env.process(theater.check_ticket(moviegoer)) if random.choice([True, False]):with theater.server.request() as request:yield requestyield env.process(theater.sell_food(moviegoer))

仿真的目的是為了確定cashier，usher，server數量，減少顧客等待時間，因此關鍵在于記錄顧客從到達到入座經過了多長時間

def go_to_movies(env, moviegoer, theater):# Moviegoer到達theaterarrival_time = env.nowwith theater.cashier.request() as request:yield requestyield env.process(theater.purchase_ticket(moviegoer))with theater.usher.request() as request:yield requestyield env.process(theater.check_ticket(moviegoer)) if random.choice([True, False]):with theater.server.request() as request:yield requestyield env.process(theater.sell_food(moviegoer)) wait_times.append(env.now - arrival_time)

這里也可以用單獨的列表departure_time存儲離開時間，但是沒有必要，這只是重復的代碼，DRP（do not repeat yourself）

3.Making Things Happen: Function Definition

現在，你需要定義一個函數來運行仿真，run_theater()會實例化theater，不停的生成moviegoers直到仿真停止

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

傳入之前定義的三個參數：

• num_cashiers
• num_ushers
• num_servers

這些參數決定了仿真環境的配置。假設在仿真開始的時候就有一些moviegoers在電影院等候，現實生活中也可能會有電影院還沒開門的時候就有人等著。設置初始時刻有3位moviegoer在排隊等待買票

def run_theater(env, num_cashiers, num_servers, num_ushers):theater = Theater(env, num_cashiers, num_ushers, num_servers)for moviegoer in range(3):env.process(go_to_movies(env, moviegoer, theater))

使用range()生成3位moviegoer，然后使用env.process()告訴仿真準備按流程流動moviegoer，剩下的moviegoer會在某時刻到達theater，因此這個函數應該能不停的在仿真運行期間生成新的moviegoer。假設每12秒（0.2分鐘）到達一位顧客，使用timeout表示時間間隔

def run_theater(env, num_cashiers, num_servers, num_ushers):theater = Theater(env, num_cashiers, num_ushers, num_servers)for moviegoer in range(3):env.process(go_to_movies(env, moviegoer, theater))while True:yield env.timeout(0.2)moviegoer += 1env.process(go_to_movies(env, moviegoer, theater))

4.Calculating the Wait Time: Function Definition

運行完wait_times列表記錄了所有顧客的等待時間，對數據進行處理，計算均值，輸出值用每分每秒表示

def calculate_wait_time(wait_times):average_wait = statistics.mean(wait_times)minutes, frac_minutes = divmod(average_wait, 1)seconds = frac_minutes * 60return round(minutes), round(seconds)

5.Choosing Parameters: User Input Function Definition

之前的環境取決于以下三個變量的值：

• num_cashiers
• num_servers
• num_ushers

仿真的優勢在于你可以隨意測試這些參數改變仿真場景，可以單獨設置函數來獲取仿真配置參數

def get_user_input():num_cashiers = input('cashier number: ')num_ushers = input('ushers number: ')num_servers = input('servers number: ')params = [num_cashiers, num_ushers, num_servers]# 檢查輸入是否錯誤if all(str(i).isdigit() for i in params):params = [int(x) for x in params]else:print('input wrong, simulation start with default value')params = [1, 1, 1]return params

6.Finalizing the Setup: Main Function Definition

最后一步就是創建主函數

def main():# setuprandom.seed(42)num_cashiers, num_ushers, num_servers = get_user_input()# runenv = simpy.Environment()env.process(run_theater(env, num_cashiers, num_ushers, num_servers))env.run(until=90)# outputmins, secs = calculate_wait_time(wait_times)print( "Running simulation...",f"\nThe average wait time is {mins} minutes and {secs} seconds.")

• 設置隨機數，獲取環境配置參數
• 創建environment
• 告訴simpy運行run_theater()進程：創建theater環境和生成moviegoer
• 設置運行時間為90
• 計算并輸出平均等待時間

7.How to Run the Simulation

最后回顧一下定義的類和函數

• Theater:此類定義了你要模擬的環境的藍圖。它確定有關該環境的一些信息，例如可用的資源類型以及與它們相關聯的進程
• go_to_movies():此函數發出使用資源的明確請求，執行完相關的進程，然后將其釋放給下一個電影觀眾
• run_theater():該函數控制仿真過程，使用Theater藍圖創建一個劇院的實例，然后調用go_to_movies()來生成和移動人們完成所有步驟
• calculate_wait_time():計算并輸出等待時間
• get_user_input():獲取環境配置參數
• main():集成以上內容作為主函數

if __name__ == '__main__':main() cashier number: 2 ushers number: 2 servers number: 2 Running simulation... The average wait time is 36 minutes and 44 seconds.

8.Full code

class Theater(object):def __init__(self, env, num_cashiers, num_ushers, num_servers):self.env = envself.cashier = simpy.Resource(env, num_cashiers)self.usher = simpy.Resource(env, num_ushers)self.server = simpy.Resource(env, num_servers)def purchase_ticket(self, moviegoer):yield self.env.timeout(random.randint(1, 3))def check_ticket(self, moviegoer):yield self.env.timeout(3 / 60)def sell_food(self, moviegoer):yield self.env.timeout(random.randint(1, 5))def go_to_movies(env, moviegoer, theater):# Moviegoer到達theaterarrival_time = env.nowwith theater.cashier.request() as request:yield requestyield env.process(theater.purchase_ticket(moviegoer))with theater.usher.request() as request:yield requestyield env.process(theater.check_ticket(moviegoer)) if random.choice([True, False]):with theater.server.request() as request:yield requestyield env.process(theater.sell_food(moviegoer)) # Moviegoer離開theaterwait_times.append(env.now - arrival_time)def run_theater(env, num_cashiers, num_servers, num_ushers):theater = Theater(env, num_cashiers, num_ushers, num_servers)for moviegoer in range(3):env.process(go_to_movies(env, moviegoer, theater))while True:yield env.timeout(0.2)moviegoer += 1env.process(go_to_movies(env, moviegoer, theater))def calculate_wait_time(wait_times):average_wait = statistics.mean(wait_times)minutes, frac_minutes = divmod(average_wait, 1)seconds = frac_minutes * 60return round(minutes), round(seconds)def get_user_input():num_cashiers = input('cashier number: ')num_ushers = input('ushers number: ')num_servers = input('servers number: ')params = [num_cashiers, num_ushers, num_servers]# 檢查輸入是否錯誤if all(str(i).isdigit() for i in params):params = [int(x) for x in params]else:print('input wrong, simulation start with default value')params = [1, 1, 1]return paramsdef main():# setuprandom.seed(42)num_cashiers, num_ushers, num_servers = get_user_input()# runenv = simpy.Environment()env.process(run_theater(env, num_cashiers, num_ushers, num_servers))env.run(until=90)# outputmins, secs = calculate_wait_time(wait_times)print( "Running simulation...",f"\nThe average wait time is {mins} minutes and {secs} seconds.")if __name__ == '__main__':main() cashier number: 2 ushers number: 3 servers number: 1 Running simulation... The average wait time is 36 minutes and 43 seconds.

總結

以上是生活随笔為你收集整理的python实现电影院仿真（SimPy)的全部內容，希望文章能夠幫你解決所遇到的問題。

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