ws网络下的SIR模型代码
时间: 2023-06-14 22:06:56 浏览: 51
以下是一个基于SIR模型的简单传染病传播模拟的示例代码,使用Python和websocket实现:
```python
import asyncio
import json
import random
import websockets
# 模拟人口数量
POPULATION = 1000
# 初始感染人数
INIT_INFECTED = 10
# SIR参数(beta, gamma)
BETA = 0.2
GAMMA = 0.1
# 初始化人口
population = [{'id': i, 'state': 'S'} for i in range(POPULATION)]
# 随机选取INIT_INFECTED个人作为初始感染者
for i in random.sample(range(POPULATION), INIT_INFECTED):
population[i]['state'] = 'I'
# 根据传染病模型进行模拟
async def simulate():
# 初始状态
state = {'S': POPULATION - INIT_INFECTED, 'I': INIT_INFECTED, 'R': 0}
# 循环模拟
while True:
# 计算每个人的下一个状态
next_population = []
for person in population:
if person['state'] == 'S':
# 计算被感染的概率
infected_prob = sum([1 for neighbor in population
if neighbor['state'] == 'I'
and neighbor['id'] != person['id']]) / POPULATION * BETA
# 根据概率随机决定是否被感染
if random.random() < infected_prob:
person['state'] = 'I'
elif person['state'] == 'I':
# 根据gamma概率恢复
if random.random() < GAMMA:
person['state'] = 'R'
next_population.append(person)
population[:] = next_population
# 统计状态数量
state = {'S': 0, 'I': 0, 'R': 0}
for person in population:
state[person['state']] += 1
# 将状态发送给所有连接的客户端
await asyncio.wait([websocket.send(json.dumps(state)) for websocket in websockets])
# 与客户端建立连接并发送初始状态
async def handler(websocket, path):
await websocket.send(json.dumps({'S': POPULATION - INIT_INFECTED, 'I': INIT_INFECTED, 'R': 0}))
# 启动websocket服务器
start_server = websockets.serve(handler, 'localhost', 8765)
# 启动模拟
asyncio.get_event_loop().run_until_complete(asyncio.gather(start_server, simulate()))
```
这个代码使用了Python的asyncio模块来实现协程,使用websockets库来实现websocket服务器。SIR模型中的beta和gamma参数可以根据实际情况进行调整。