冠狀病毒傳播模擬器的原理和實現(Python版)【附原始碼】
本文摘要:
本文首先會解釋一下到底什麼是"冠狀病毒",以及殺死"冠狀病毒"的方法。然後會利用Python實現一個"冠狀病毒"傳播模擬器,來演示一下為何“不出門“ +“瘋狂建醫院”會間接殺死病毒(動態模擬了從發生疫情,到疫情結束的整個過程)。以及如果控制不好,會有什麼後果(一定是很嚴重的)。
相信大家現在一定和我一樣,無比鬱悶。好不容易盼到過年了,買了一大堆好吃的,結果哪也去不了,只能在家消耗這些美食了。而且很多人宅在家裡的每一天都會做同一個偉大的計劃!!
那麼我們為什麼不能出門呢?答案大家都已經知道了,這就是一種被稱為“冠狀病毒”的東西在搗亂。據鍾南山院士和其他一些專家推測,這種“冠狀病毒”的潛伏期是14天左右,所以需要儘可能避免在14天內接觸更多的人。而宅在家裡是最好的選擇。不過這是頭一次放這麼長的假,而且還必須呆在家裡,也是相當鬱悶,恍惚間竟想起了猴哥,500年你是怎麼熬過來的呢!
下面就先來了解下到底什麼是冠狀病毒!
1. 什麼是冠狀病毒
如果詳細解釋什麼是冠狀病毒,先要從病毒講起,估計要寫一本書了,不過完全沒有必要。大家只要知道一點就好。病毒不是生物(類生物,有生物的部分特徵),不能獨立生存,需要依賴於宿主,也就是人體的細胞。換句話說,病毒需要攻擊人體的細胞才能生存。病毒可以利用細胞中的成分複製自身,從而達到繁殖的目的。如果受到感染的細胞過多,人體就會生病。
那麼"冠狀病毒"呢?當然也屬於病毒的一種,不過它不是普通的病毒。首先看看"冠狀病毒"的樣子,病毒表面有很多凸起的冠(看起來還挺漂亮的),所以稱為"冠狀病毒"。
"冠狀病毒"除了外形上與普通病毒有差異,在本質上與普通病毒也是不同的。普通病毒通常都是DNA病毒,也就是雙螺旋結構。而"冠狀病毒"屬於RNA病毒,也就是單螺旋結構。雙螺旋結構的DNA病毒更穩定,不容易變異。但RNA病毒只有一個鏈,非常不穩定,容易變異。這才是"冠狀病毒"的殺手鐗,通過變異抵抗人類的反擊。目前"冠狀病毒"是否變異,還很難說。
2. 如何殺死"冠狀病毒"
現在的問題是,如何殺死"冠狀病毒"。當然,最直接的方法是將"冠狀病毒"趕出細胞,或直接從細胞中清除,不過目前人類的技術還做不到這一點。這應該屬於比細胞醫學更高階的醫學:分子醫學或原子醫學。目前人類的科技水平甚至還沒完全達到細胞醫學的高度(最多0.7個細胞醫學),因為癌症等細胞疾病還無法完全治癒。
既然病毒目前還無法直接殺死,那麼這麼多痊癒的病人是如何做到的呢?其實可以採用如下兩種方法間接殺死"冠狀病毒",也包括其他大多數病毒。
1. 利用人體內的抗體細胞(如T細胞),將被感染的細胞連同"冠狀病毒"一起消滅
2. 干擾"冠狀病毒"在細胞內的複製過程(例如,RNA干擾),讓"冠狀病毒"無法正常複製自身,也就是讓"冠狀病毒"沒有後代,這樣"冠狀病毒"就會由於自身的生命終結而死亡。
目前絕大多數痊癒的病人屬於第一種情況,也就是通過自身的抗體細胞(如T細胞)檢測人體內被感染的細胞,然後通知這些被感染的細胞啟動自毀程式,當這些細胞被銷燬後,那麼病毒也就被消滅了。
不過可能有的同學會問,既然病毒可以被抗體細胞消滅,那麼這麼多醫護人員和醫療裝置豈不是多餘了。其實並不是多於的,而是非常必要。這是因為病毒在攻擊正常細胞的同時,也讓人體的抵抗力開始下降,人體內的抗體細胞是有限的,這些抗體細胞會到處救火,顧不上對付"冠狀病毒"了。正是由於這些醫療器械(如呼吸機),儘可能讓人體機能恢復到接近正常人的水平,這樣各種抗體細胞就可以集中力量對付"冠狀病毒"了。所以說,這些痊癒的病人其實是通過自身的抗體細胞消滅了"冠狀病毒",而醫療器械、各種藥物,醫務人員的護理,其實是抗體細胞的援軍。另外,由於不同人的體質不同。極個別的個體,抗體細胞非常強大,造成了"冠狀病毒"無法快速複製(但仍然在複製,只是增速變慢),這也是為什麼有的個體的潛伏期會超過14天的原因,但這畢竟是極少數。潛伏期是被病毒感染的細胞數量達到足以致病的時間(人體內如果只有少量的被病毒感染的細胞,是不會表現出任何症狀的)。
還有就是,為什麼"冠狀病毒"感染者在發病時大多會有肺炎的症狀,而且伴隨著發燒呢?其實這就是抗體細胞在和"冠狀病毒"進行較量呢!進醫院隔離,是為了增強抗體細胞對抗"冠狀病毒"的籌碼。不過由於一些人的抗體細胞的戰鬥力實在太差(可能有部分人是因為年齡太大的原因),所以就算抗體細胞等來了援軍,也於事無補,這些就是已經死亡的被感染者(大多是60歲以上的老年人)。所以等疫情結束後,好好鍛鍊身體吧,擁有強壯的身體,不能保證你不得病,但至少可以增加得病後活下來的機率。
對於目前正在研發的抗"冠狀病毒"的藥物主要是通過第2種方式消滅病毒的。也就是干擾病毒的複製過程,不過很遺憾,到現在為止,還沒有被證明非常有效的藥物可以做到這一點。
3. 對付"冠狀病毒"的手段
從生物學角度,我們已經瞭解了"冠狀病毒"的發病原理,但在現實中,如何操作呢?
其實對付"冠狀病毒"以及其他大多數病毒,基本上就是基於12個字: 有症狀趕快治,沒症狀要隔離。這也是國際上通用的原則。
前6個字容易理解,有症狀了,就直接進醫院了。如果沒症狀呢?沒症狀有兩種情況:疑似和正常人。疑似主要是指與被感染者近距離接觸,或從外地來本地的人員(輸入者),由於"冠狀病毒"的潛伏期是14天左右,所以這些疑似者至少需要被隔離14天才會確定是否真的被感染。而正常人只要沒和被感染者近距離接觸,就不會被感染。這些人之所以也需要隔離,是因為怕被別人感染。不過這裡的隔離通常是在自己的家中,不出門。通過隔離,可以大幅度減少病毒感染新的人群,也就是讓存量不再增加或少增加。而還有很多被感染者,這些人就需要在醫院裡接受治療了。不過由於被感染者太多,所以武漢等地區快速建起了很多臨時醫院。 這是用來減少存量的。 當存量不但不會增加、而且在不斷減少,直到被感染者為0,疑似者為0時,疫情才會徹底結束,這也是本文要介紹的病毒擴散模擬器的基本原理。
4. 用病毒擴散模擬器來演示病毒擴散和疫情結束的全過程
在實現這個模擬器之前,先來演示下這個模擬器。
模擬器可以對多個數據進行模擬,包括健康者人數、潛伏期人數、發病者人數、已經隔離的人數、已經死亡的人數、空餘床位、繼續床位、病毒傳播率、病毒潛伏期、醫院收治響應時間、醫院當前床位、安全距離、平均流動意向。
啟動程式,會利用初始值進行模擬,初始發病人數為50人,市民總數為5000人。如下圖所示。
中間區域的若干個點表示各種狀態的市民。白色的表示健康市民、黃色表示潛伏期市民、紅色表示發病市民、黑色表示死亡的市民。右側的豎條表示醫院的床位,初始值是100。如果用引數值進行模擬,100張床位很快就會被填滿,然後病毒在人群中就會大爆發,很快紅點就會遍佈人群,如下圖所示
當前天數已經顯示過了31天(耽誤了一個月),感染者已經接近1000了,這時政府開始採取緊急措施。主要有兩個:封城和關閉娛樂場所、增加醫院的床位。前者是為了避免感染更多的人,後者是為了消耗被感染者的存量。所以通過下面的設定來調整引數。例如,將流動意向調整為-1.71。並且增加床位323個。
這時總床位數變成了423。這裡的流動意向在-3和3之間,如果是3,表示市民的活動意願非常強烈,例如,正好是春節時期,市民逛商場,聚會非常頻繁。流動意願越小,流動意願就越低。這裡調成-1.76,表示市民不能參加聚會、不能出城、出門需要戴口罩,但市民仍然可以在市內流動。 流動意願遠低於春節正常的值。不過儘管政府採取了一定的措施,但由於是在疫情開始後一個月才採取了緊急措施,所以病毒已經擴散了,因此,疫情並沒有得到非常明顯的緩解。如下圖所示。 主要表現為市民仍然可以自由活動(儘管不能參加聚會),仍然存在一定的感染風險。 而且醫院床位明顯不足。
為了更進一步控制疫情,政府開始封閉小區,更進一步限制人員的活動,以及軍方開始干預,派出了大量的醫護人員以及各種醫療裝置,並且建立的多個方艙醫院。醫院床位得到了很大的緩解。這裡將引數設定成最大值來模擬這一過程,增加床位1200個,流動意向設定為-3.0,也就是說基本上市民不流動了。如下圖所示。
這時看到床位已經增加到了1623,比急需的床位多了不少,而且人員趨於不流動,發病人數不斷減少(都被送進了醫院),而且潛伏期人數逐漸轉換為發病人數,也被送進了醫院,最終,潛伏期人數和發病者人數都是0,疫情結束,如下圖所示。共耗費了60天。當然,實際情況沒這麼順利。模擬器可以立刻增加醫院床位數,可以立刻隔離人員,但在實際操作中,建立醫院需要時間,隔離也需要協調,尤其是上千萬人的大城市。
不過只要能做到隔離和及時就醫,冠狀病毒疫情結束也只是時間問題。當然,這要在這兩點做的比較好的情況下,如果處理失當,那麼模擬器就會呈現下圖的狀態,完全失控,人類將面臨一場浩劫。
5. 病毒傳播模擬器的實現
現在來談談模擬器實現的原理。模擬器使用Python和PyQt5實現。PyQt5是封裝了Qt library的跨平臺GUI開發庫,基於Python語言。
這裡主要涉及到模擬器效果繪製,以及如何模擬多個引數。先來說一下繪製市民的狀態。繪製的工作通過drawing.py檔案的Drawing類來完成。該類是QWidget的子類,這也就意味著Drawing類本身是PyQt5的一個元件。與按鈕、標籤類似。只是並不需要往Drawing上放置任何子元件。只要在Drawing上繪製各種圖形即可。
在PyQt5中,任何一個QWidget的子類,都可以實現一個paintEvent方法,當元件每次重新整理時,就會呼叫paintEvent方法重新繪製元件的內容。Drawing類中paintEvent方法的程式碼如下:
def paintEvent(self, event): qp = QPainter() qp.begin(self) # 繪製城市的各種狀態的市民 self.drawing(qp) qp.end()
在繪製圖像前,需要建立QPainter物件,然後呼叫QPainter物件的begin方法,結束繪製後,需要呼叫QPainter物件的end方法。上面程式碼中的drawing方法用於完成具體的繪製工作。
模擬器可以模擬5000個市民的狀態,所以需要用5000個小矩形來表示這5000個市民。也就是在drawing方法中需要繪製這5000個表示市民的小矩形。程式碼如下:
def drawing(self, event): ... ... # 繪製代表市民的小矩形 persons = Persons().persons if persons == None: return normal_person_count = 0 latency_person_count = 0 confirmed_person_count = 0 freeze_person_count = 0 death_person_count = 0 # 掃描內一個人的狀態 for person in persons: if person.state == NORMAL: # 健康人 qp.setPen(Qt.white) normal_person_count += 1 elif person.state == LATENCY: # 潛伏期感染者 qp.setPen(QColor(255,238,0)) latency_person_count += 1 elif person.state == CONFIRMED: # 確診患者 qp.setPen(Qt.red) confirmed_person_count += 1 elif person.state == FREEZE: # 已隔離者 qp.setPen(QColor(72, 255, 252)) freeze_person_count += 1 elif person.state == DEATH: # 死亡患者 qp.setPen(Qt.black) death_person_count += 1 person.update() # 更新每一個人的狀態 bed_half_size = Hospital().bed_size // 2 rect = QRect(person.x - bed_half_size, person.y - bed_half_size,Hospital().bed_size//2, Hospital().bed_size//2) brush = QBrush(Qt.SolidPattern) brush.setColor(qp.pen().color()) qp.setBrush(brush) qp.drawRect(rect) ... ...
在上面的程式碼中,通過 Persons物件的persons屬性獲取表示市民的物件(Person物件)列表。並在迴圈中根據Person物件的狀態設定小矩形的顏色,以及分別統計不同人群的數量,這些數量會顯示在模擬器右側的元件中。最後,使用drawRect方法繪製表示每一個市民的小矩形。這樣就繪製了當前狀態的5000個市民。
當然,這些狀態要不斷更新。這裡使用執行緒每100毫秒重新整理一次,這些功能在refresh.py檔案的Refresh類中,程式碼如下:
from PyQt5.QtCore import * from params import * class Refresh(QThread): def __init__(self, drawing): super(Refresh, self).__init__() self.drawing = drawing def run(self): while not Params.success: try: QThread.msleep(100) # 重新整理Drawing self.drawing.update() Params.current_time += 1 except: pass
每次重新整理Drawing,需要呼叫update方法,呼叫該方法後,Drawing就會呼叫自身的paintEvent方法重新繪製整個元件的內容。
在paintEvent方法中,還呼叫了Person物件的update方法,該方法是我們自己編寫的,用於不斷更新每一個人的狀態,這些狀態會根據多個引數進行協調。該方法屬於Person類,程式碼如下:
def update(self): # 如果已經隔離或者死亡了,就不需要處理了 if self.state == FREEZE or self.state == DEATH: return # 處理已經確診的感染者(即患者) if self.state == CONFIRMED and self.dead_time == 0: destiny = random.randrange(1,10001) # 幸運數字,[1,10000]隨機數 if destiny >= 1 and destiny <= int(Params.fatality_rate * 10000): # 幸運數字落在死亡區間 dt = int(sp.random.normal(Params.dead_time,Params.dead_variance)) self.dead_time = self.confirmed_time + self.dead_time else: self.dead_time = -1 # 逃過了死神的魔爪 if self.state == CONFIRMED and Params.current_time - self.confirmed_time >= Params.hospital_receive_time: # 如果患者已經確診,且(世界時刻-確診時刻)大於醫院響應時間,即醫院準備好病床了,可以抬走了 bed = Hospital().pick_bed() # 查詢空床位 if bed == None: # 沒有空床位,報告需求床位數 if not self.need_bed: Hospital().need_bed_count += 1 self.need_bed = True else: # 安置病人 self.used_bed = bed self.state = FREEZE self.x = bed.x + Hospital().bed_size // 2 self.y = bed.y + Hospital().bed_size // 2 if self.need_bed and Hospital().need_bed_count > 0: Hospital().need_bed_count -= 1 bed.is_empty = False # 處理病死者 if (self.state == CONFIRMED or self.state == FREEZE) and Params.current_time >= self.dead_time and self.dead_time > 0: self.state = DEATH # 患者死亡 personpool.Persons().latency_persons.remove(self) # 已經死亡,無法傳染別人,需要從確診者中刪除 Hospital().empty_bed(self.used_bed) # 騰出床位 if Hospital().need_bed_count > 0: Hospital().need_bed_count -= 1 # 增加一個正態分佈用於潛伏期內隨機發病時間 latency_symptom_time = sp.random.normal(Params.virus_latency / 2,25) # 處理髮病的潛伏期感染者 if Params.current_time - self.infected_time > latency_symptom_time and self.state == LATENCY: self.state = CONFIRMED # 潛伏者發病 self.confirmed_time = Params.current_time # 重新整理確診時間 # 處理未隔離者的移動問題 self.action() # 處理健康人被感染的問題 persons = personpool.Persons().persons # 不是健康人,返回 if self.state >= LATENCY: return # 通過一個隨機幸運值和安全距離決定感染其他人 latency_persons = personpool.Persons().latency_persons.copy() for person in latency_persons: random_value = random.random() if random_value < Params.broad_rate and self.distance(person) < Params.safe_distance: self.be_infected() break
update方法主要就是根據在params.py中的各種引數變數,以及隨機值,計算下一次狀態中潛伏期人數、感染人數、被隔離人數等資料,並且在每次重新整理頁面時更新這些資料。
以上的描述就是如何繪製表示5000個市民的狀態。右側各種資料並不是繪製在頁面上的,而是通過QtDesigner設計的右側的介面,然後將Drawing物件作為標準的元件放在了主介面的左側。設計介面如下圖所示:
然後通過pyuic將.ui檔案生成.py檔案,在程式中呼叫即可。這些元件的更新同樣是在前面給出的drawing方法中。
另外,這個模擬器還提供了動態設定引數的功能。這是通過另外一個程式實現的,兩個程式通過socket通訊。這個設定程式同樣是通過QtDesigner設計的,設計介面如下圖所示。
在設定程式中,通過Transmission類的send_command方法向模擬器釋出命令,例如,更新床位數的程式碼如下:
from PyQt5.QtWidgets import * from socket import * class Transmission: def __init__(self,ui): self.ui = ui self.host = 'localhost' self.port = 5678 self.addr = (self.host, self.port) # 向模擬器釋出命令 def send_command(self, command, value = None): tcp_client_socket = socket(AF_INET, SOCK_STREAM) tcp_client_socket.connect(self.addr) if value == None: value = 0 data = command + ':' + str(value) tcp_client_socket.send(('%s\r\n' % data).encode(encoding='utf-8')) data = tcp_client_socket.recv(1024) result = data.decode('utf-8').strip() tcp_client_socket.close() return result # 更新床位數 def update_bed_count(self): print(self.ui.horizontalSliderBedCount.value()) result = self.send_command('add_bed_count',self.ui.horizontalSliderBedCount.value()) if result == 'ok': QMessageBox.information(self.ui.centralwidget, '訊息', f'成功添加了{self.ui.horizontalSliderBedCount.value()}張床位', QMessageBox.Ok) 在模擬器端,通過Receiver以及TCPServer來接收設定程式發過來的命令,如果成功設定,返回ok。Receiver類以及相關的程式碼如下: from socketserver import (TCPServer as TCP,StreamRequestHandler as SRH) from common import * from params import * from hospital import * from PyQt5.QtCore import * import sys # 響應客戶端請求事件的類 class MyRequestHandler(SRH): def handle(self): # 讀取客戶端傳送的資料 data = str(self.rfile.readline(),'utf-8') index = data.find(':') command = data[:index] value = data[index + 1:] value = int(value) # 執行具體的命令 if command == 'add_bed_count': Params.hospital_bed_count += value Hospital().free_bed_count = Hospital().free_bed_count + value Hospital().compute(value) elif command == 'set_flow_intention': Params.average_flow_intention = value / 100 elif command == 'set_broad_rate': Params.broad_rate = value / 100 elif command == 'set_latency': Params.virus_latency = value * 10 elif command == 'close': Params.app.quit() self.wfile.write(b'ok\r\n') # 線上程中監聽客戶端的請求 class Receiver(QThread): tcp_server = None def __init__(self): super(Receiver,self).__init__() self.host = '' self.port = 5678 self.addr = (self.host,self.port) Receiver.tcp_server = TCP(self.addr, MyRequestHandler) def run(self): Receiver.tcp_server.serve_forever()
以上就是這個病毒傳播模擬器的基本實現方法,其中涉及到了大量PyQt5的知識,如果大家想詳細瞭解PyQt5技術,可以參考我的《PyQt5(Python)開發與實戰視訊課程》課程。另外,《冠狀病毒傳播模擬器的原理和實現(Python版)》視訊課程即將推出,歡迎關注。
原始碼下載(github)
技術支援請關注“極客起源”公眾號。
《Python爬蟲技術:深入理解原理、技術與開發》已經出版,敬請關注!
《冠狀病毒傳播模擬器(Python版)》(掃描進入課程頁面)
https://ke.qq.com/course/1045964?tuin=a22a65ce
《PyQt5實戰視訊課程》(掃描進入課程頁面)
https://ke.qq.com/course/374285?tuin=a22a65ce
&n