投稿者: 
言語伝播をシミュレーションする
以前の投稿で流行伝播モデルというものを紹介した．
今回は流行伝播モデルと地図を組み合わせて言語伝播のシミュレーションを行ってみたいと思う．
コードは下記．(感染者と表現しているが，これはSIRモデルの言葉を借りており，これは言語に影響を受けた対象と捉えることにする)
import geopandas as gpd
import numpy as np
import pandas as pd
import folium
from folium.plugins import TimestampedGeoJson
from shapely.geometry import Point
def create_population(num_points: int = 1000,
bounds: tuple = ((35.6, 139.6), (35.8, 139.8)),
initial_infected: int = 10) -> gpd.GeoDataFrame:
"""
指定範囲内でランダムな人口（地理空間情報付き）を生成し、初期状態を設定する。
Parameters:
num_points (int): 個体数。
bounds (tuple): ((min_lat, min_lon), (max_lat, max_lon)) で表される空間的境界。
initial_infected (int): 初期感染者数。
Returns:
gpd.GeoDataFrame: 'latitude', 'longitude', 'status', 'geometry' を含む GeoDataFrame。
"""
latitudes = np.random.uniform(bounds[0][0], bounds[1][0], num_points)
longitudes = np.random.uniform(bounds[0][1], bounds[1][1], num_points)
statuses = np.array(['S'] * num_points)
population = pd.DataFrame({
'latitude': latitudes,
'longitude': longitudes,
'status': statuses
})
# 初期感染者の割当
infected_indices = np.random.choice(num_points, initial_infected, replace=False)
population.loc[infected_indices, 'status'] = 'I'
# GeoDataFrameに変換
gdf = gpd.GeoDataFrame(population,
geometry=gpd.points_from_xy(population.longitude, population.latitude))
return gdf
def sir_step(population_gdf: gpd.GeoDataFrame,
beta: float = 0.05,
gamma: float = 0.01,
radius: float = 0.01) -> gpd.GeoDataFrame:
"""
SIRモデルに基づく1ステップのシミュレーションを実施する。
Parameters:
population_gdf (gpd.GeoDataFrame): 現在の人口状態。
beta (float): 感染率（接触ごと）。
gamma (float): 回復率。
radius (float): 感染の伝播が成立する接触半径。
Returns:
gpd.GeoDataFrame: 1ステップ後の状態を反映した GeoDataFrame。
"""
new_gdf = population_gdf.copy()
# 感染者と感受性者を抽出
infected_gdf = new_gdf[new_gdf['status'] == 'I']
susceptible_gdf = new_gdf[new_gdf['status'] == 'S']
# 感染者が存在する場合、空間インデックスを作成
if not infected_gdf.empty:
infected_sindex = infected_gdf.sindex
# 感受性者の状態更新（感染判定）
for idx, susceptible in susceptible_gdf.iterrows():
if not infected_gdf.empty:
# 接触範囲内の候補を空間インデックスで抽出
buffer = susceptible.geometry.buffer(radius)
candidate_indices = list(infected_sindex.intersection(buffer.bounds))
candidate_infected = infected_gdf.iloc[candidate_indices]
# 実際に半径内にある感染者を抽出
neighbors = candidate_infected[candidate_infected.geometry.distance(susceptible.geometry) < radius]
if not neighbors.empty:
infection_probability = 1 - (1 - beta) ** len(neighbors)
if np.random.rand() < infection_probability:
new_gdf.at[idx, 'status'] = 'I'
# 感染者の状態更新（回復判定）
for idx, _ in infected_gdf.iterrows():
if np.random.rand() < gamma:
new_gdf.at[idx, 'status'] = 'R'
return new_gdf
def run_simulation(steps: int = 30,
num_points: int = 1000,
bounds: tuple = ((35.6, 139.6), (35.8, 139.8)),
initial_infected: int = 10,
beta: float = 0.05,
gamma: float = 0.01,
radius: float = 0.01) -> list:
"""
指定ステップ数にわたりSIRシミュレーションを実行し、各時点の状態を記録する。
Parameters:
steps (int): シミュレーションステップ数。
num_points (int): 人口の総数。
bounds (tuple): 人口生成の空間的境界。
initial_infected (int): 初期感染者数。
beta (float): 感染率。
gamma (float): 回復率。
radius (float): 感染伝播の接触半径。
Returns:
list: 各ステップの状態を示す GeoDataFrame のリスト。
"""
population_gdf = create_population(num_points=num_points,
bounds=bounds,
initial_infected=initial_infected)
history = [population_gdf.copy()]
for step in range(steps):
population_gdf = sir_step(population_gdf, beta=beta, gamma=gamma, radius=radius)
history.append(population_gdf.copy())
return history
def visualize_simulation(history: list) -> folium.Map:
"""
Folium を用いてシミュレーション結果を可視化する。
Parameters:
history (list): 各シミュレーションステップの GeoDataFrame のリスト。
Returns:
folium.Map: タイムスタンプ付きの GeoJSON レイヤーを含む地図オブジェクト。
"""
features = []
status_colors = {'S': 'blue', 'I': 'red', 'R': 'green'}
# 各ステップのデータを GeoJSON フォーマットに変換
for step, gdf in enumerate(history):
timestamp = f'2025-01-{step+1:02d}'
for _, row in gdf.iterrows():
color = status_colors.get(row['status'], 'black')
feature = {
'type': 'Feature',
'geometry': {
'type': 'Point',
'coordinates': [row['longitude'], row['latitude']]
},
'properties': {
'time': timestamp,
'style': {'color': color},
'icon': 'circle',
'iconstyle': {
'fillColor': color,
'fillOpacity': 0.7,
'stroke': True,
'radius': 4
}
}
}
features.append(feature)
# 地図の中心を初期集団の平均位置に設定
center_lat = history[0]['latitude'].mean()
center_lon = history[0]['longitude'].mean()
m = folium.Map(location=[center_lat, center_lon], zoom_start=12)
# タイムスタンプ付き GeoJSON レイヤーを追加
TimestampedGeoJson({
'type': 'FeatureCollection',
'features': features
}, period='P1D', add_last_point=True).add_to(m)
return m
def main():
"""
シミュレーションの実行と可視化を行うメイン関数。
"""
# シミュレーションパラメータ
steps = 30
num_points = 1000
bounds = ((35.6, 139.6), (35.8, 139.8))
initial_infected = 10
beta = 0.05
gamma = 0.01
radius = 0.01
# シミュレーション実行
simulation_history = run_simulation(steps=steps,
num_points=num_points,
bounds=bounds,
initial_infected=initial_infected,
beta=beta,
gamma=gamma,
radius=radius)
# 結果の可視化
simulation_map = visualize_simulation(simulation_history)
simulation_map.save('language_spread_simulation.html')
print("シミュレーション完了。'language_spread_simulation.html' に地図を保存しました。")
if __name__ == '__main__':
main()
