人流データから滞在地点判定（scikit-mobility）

import pandas as pd
import re
import folium # 地図上へのマッピング用
import skmob
from skmob.preprocessing import detection # 滞在場所検出

# 位置情報データ読み込み
file_path = '../00_data/GPS_241109_SJIS_LF.CSV'
df_raw = pd.read_csv(file_path, encoding='cp932')

# 前処理
df = df_raw.copy()
# 日付と時刻を結合
df['datetime'] = '20' + df['日付'] + ' ' + df['時刻']
df['datetime'] = pd.to_datetime(df['datetime'])

# 抽出する時間範囲
start = '2024-11-09 09:45:00'
end = '2024-11-09 12:00:00'
dt_start = pd.to_datetime(start)
dt_end = pd.to_datetime(end)

# 時間範囲でフィルタリング
print('抽出前行数：', df.shape[0])
df = df[(df['datetime'] >= dt_start) & (df['datetime'] <= dt_end)]
print('抽出後行数：', df.shape[0])

# あらかじめpandas.DataFrameをTrajDataFrameに変換
tdf = skmob.TrajDataFrame(df, latitude='緯度', longitude='経度', datetime='datetime')

# パラメータの指定
cf_stoptime = 10 # 分
cf_radius = 0.2 # km
# 滞在地点判定
stdf = detection.stay_locations(
    tdf, 
    stop_radius_factor=0.5, 
    minutes_for_a_stop=cf_stoptime, # 〇〇分以上居ると滞在判定
    spatial_radius_km=cf_radius, # 半径〇〇km以内に居続けると滞在判定
    leaving_time=True # 滞在地点を出る時間をカラムとして追加するか
)

import pandas as pd
import geopandas as gpd
import skmob
# 駅情報データ
file_path = '../../00_common_data/01_国土数値情報/02_鉄道ライン/2023/N05-23_GML/N05-23_Station2.shp'
# シェープファイルの読み込み
gdf = gpd.read_file(file_path, encoding='shift-jis')
# カラム名を修正
gdf = gdf.rename(columns={
    'N05_001':'事業者種別', 'N05_002':'路線名', 'N05_003':'運営会社', 'N05_004':'供用開始年', 'N05_005b':'設置期間（設置開始）', 'N05_005e':'設置期間（設置終了）', 'N05_006':'関係ID', 'N05_007':'変遷ID', 'N05_008':'変遷備考', 'N05_009':'備考', 'N05_011':'駅名',
    })
# 前処理
tdf_sta = skmob.TrajDataFrame(gdf, latitude='lat', longitude='lng')
tdf_sta = tdf_sta.rename(columns={'lon':'lng'})

# 距離測定用
from shapely.geometry import Point, LineString, Polygon
from geopy.distance import geodesic

# 関数定義：滞在地点１点に対して駅マートの各駅に対して距離計算を実施し、最寄り駅と距離を返す
def calc_dist(stay_loc, df_spot, geometry_col):
    '''
    stay_loc: Point, ユーザーの滞在地点１点の座標
    df_spot: DataFrame, 駅の位置情報データ
    geometry_col : str, 座標データのカラム名
    '''
    # 距離測定用前処理
    df_spot['経度緯度'] = df_spot.apply(lambda x: Point(x['lng'], x['lat']), axis=1)

    # 距離測定
    df_spot['distance'] = [geodesic((stay_loc.y, stay_loc.x), (spot_loc.y, spot_loc.x)).meters for spot_loc in df_spot['経度緯度']]
    ic(df_spot['distance'])
    # 最小値とそのindex取得
    min_dist = df_spot['distance'].min()
    min_idx = df_spot['distance'].idxmin()
    # 最寄り駅取得
    min_sta = df_spot.loc[min_idx, '駅名']

    return min_sta, min_dist

tdf_dist = stdf.copy()
# 距離測定用前処理
tdf_dist['経度緯度'] = tdf_dist.apply(lambda x: Point(x['lng'], x['lat']), axis=1)

# 距離測定
tdf_dist['最寄駅'] = [calc_dist(stay_loc, tdf_sta, geometry_col='geometry')[0] for stay_loc in tdf_dist['経度緯度']]
tdf_dist['最寄駅距離'] = [calc_dist(stay_loc, tdf_sta, geometry_col='geometry')[1] for stay_loc in tdf_dist['経度緯度']]

# 滞在判定
cf_stay_dist = 200 # m
tdf_dist['滞在判定'] = tdf_dist['最寄駅距離'] < cf_stay_dist

# 関数定義：foliumで地図上に位置情報をプロット(iterrows)
def mapping(df_loc, df_spot):
    '''
    df_loc: DataFrame, 位置情報
    df_spot: DataFrame, スポットの位置座標
    '''
    # foliumで地図を表示
    # デフォルトの中心とズーム
    default_location = [df_loc['lat'].mean(), df_loc['lng'].mean()]
    default_zomm = 12
    folium_map = folium.Map(location=default_location, zoom_start=default_zomm)
    # スポット情報をプロット
    for idx, row in df_spot.iterrows():
        folium.Circle(
            location=[row['lat'], row['lng']],
            popup=row['駅名'],
            color='blue',
            radius=200,
            ).add_to(folium_map)
    # 位置情報をプロット
    for idx, row in df_loc.iterrows():
        folium.Circle(
            location=[row['lat'], row['lng']],
            popup=row['datetime'],
            color = 'red',            
            radius=200, # マーカーサイズを指定
            ).add_to(folium_map)
    return folium_map

# 地図上に位置情報をマッピング
folium_map = mapping(df_loc=tdf_dist, df_spot=tdf_sta)
display(folium_map)