投稿者: 
言語伝播をシミュレーションする 2
以前の投稿で流行伝播モデルと地図を組み合わせて言語伝播のシミュレーションを行ってみた．
今回はさらに例として中心を東京に設定して行なって見たいと思う．
コードは下記．(感染者と表現しているが，これは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.5, 139.5), (35.9, 139.9)),
initial_infected_coord: tuple = (35.681236, 139.767125),
initial_infected_radius: float = 0.005) -> gpd.GeoDataFrame:
"""
指定範囲内でランダムに人口（地理空間情報付き）を生成し、初期感染者を
東京駅付近（指定座標・半径内）の個体に設定する。
Parameters:
num_points (int): 個体数。
bounds (tuple): ((min_lat, min_lon), (max_lat, max_lon)) で表す生成範囲。
initial_infected_coord (tuple): (latitude, longitude) の形式で初期感染の中心座標（東京駅）。
initial_infected_radius (float): 初期感染の半径（度単位）。
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)
# 全個体を感受性者（S）とする
statuses = np.array(['S'] * num_points)
population = pd.DataFrame({
'latitude': latitudes,
'longitude': longitudes,
'status': statuses
})
# GeoDataFrameに変換
gdf = gpd.GeoDataFrame(population,
geometry=gpd.points_from_xy(population.longitude, population.latitude))
# 東京駅付近の点を初期感染者 (I) に設定
tokyo_station = Point(initial_infected_coord[1], initial_infected_coord[0]) # (lon, lat)
# 各個体と東京駅との距離を計算し、一定範囲内のものを感染状態に
distances = gdf.geometry.distance(tokyo_station)
gdf.loc[distances < initial_infected_radius, 'status'] = 'I'
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.5, 139.5), (35.9, 139.9)),
initial_infected_coord: tuple = (35.681236, 139.767125),
initial_infected_radius: float = 0.005,
beta: float = 0.05,
gamma: float = 0.01,
radius: float = 0.01) -> list:
"""
指定ステップ数にわたってSIRシミュレーションを実行し、各時点の状態を記録する。
Parameters:
steps (int): シミュレーションステップ数。
num_points (int): 人口の総数。
bounds (tuple): 人口生成の空間的境界。
initial_infected_coord (tuple): 初期感染の中心座標。
initial_infected_radius (float): 初期感染者の半径（度単位）。
beta (float): 感染率。
gamma (float): 回復率。
radius (float): 感染伝播の接触半径。
Returns:
list: 各ステップの状態を示す GeoDataFrame のリスト。
"""
population_gdf = create_population(num_points=num_points,
bounds=bounds,
initial_infected_coord=initial_infected_coord,
initial_infected_radius=initial_infected_radius)
history = [population_gdf.copy()]
for _ 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'}
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')
features.append({
'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
}
}
})
center_lat = history[0]['latitude'].mean()
center_lon = history[0]['longitude'].mean()
m = folium.Map(location=[center_lat, center_lon], zoom_start=12)
TimestampedGeoJson({
'type': 'FeatureCollection',
'features': features
}, period='P1D', add_last_point=True).add_to(m)
return m
def main():
"""
東京駅を起点として都内でのSIRシミュレーションの実行と可視化を行うメイン関数。
"""
steps = 30
num_points = 1000
bounds = ((35.5, 139.5), (35.9, 139.9))
initial_infected_coord = (35.681236, 139.767125) # 東京駅の座標
initial_infected_radius = 0.005 # 東京駅周辺の半径（度単位）
beta = 0.05
gamma = 0.01
radius = 0.01
simulation_history = run_simulation(steps=steps,
num_points=num_points,
bounds=bounds,
initial_infected_coord=initial_infected_coord,
initial_infected_radius=initial_infected_radius,
beta=beta,
gamma=gamma,
radius=radius)
simulation_map = visualize_simulation(simulation_history)
simulation_map.save('tokyo_language_spread_simulation.html')
print("シミュレーション完了。'tokyo_language_spread_simulation.html' に地図を保存しました。")
if __name__ == '__main__':
main()
