tsunami_union を保存 → 交差チェックをそれに切替
上位N→ネットワーク最短の選定を導入（経路品質UP）
アイソクローンのレイヤをGPKG/地図に追加（“どこまで行けるか”の可視化）
警告フラグをUIに表示
GraphDB CSV出力で家族アプリへの橋渡し
を行う
1) tsunami_union をGPKGにキャッシュ（毎回のunion計算を省略）
これで気になったのが
このtsunami_union をGPKGにキャッシュしたものは現在地、もしくはすでに計測したもののみ有効？
もし新規の場所で避難することがあれば対応したいので
すでにあるルートならキャッシュ、
そうでないのなら新規ルートを使うようにしたい
結論：tsunami_union は“現在地に依存しない”全域キャッシュです。
静岡県の「津波浸水想定（A40）」レイヤ全体を union したものなので、同じデータ（A40）がソースならどの場所でも再利用可能。新規の場所でもそのまま使えます。再生成が必要なのは、A40を更新・差し替えたときだけです。
A) tsunami_union（全域）のキャッシュ方針
* 有効範囲：A40レイヤ（静岡版）全域。現在地・ルートに非依存。
* 再生成条件：A40が変わった可能性があるとき（例：ファイル更新、feature数やBBOXが変化）。
* 高速化：ルート検証時は tsunami_union をその場のAOIでクリップして使うと軽いです。
例：キャッシュの検証＆必要なら再生成
import geopandas as gpd
from fiona import listlayers
GPKG = "outputs/shizuoka_hazard.gpkg"
tsunami = gpd.read_file(GPKG, layer="tsunami_inundation")
feat_count = len(tsunami)
bbox = tuple(tsunami.total_bounds) # (xmin,ymin,xmax,ymax)
# 既存 union があるか＆メタデータが一致するか
need_rebuild = True
if "tsunami_union" in listlayers(GPKG):
union = gpd.read_file(GPKG, layer="tsunami_union")
# 簡易同一性チェック（必要なら厳格化）
if "meta_feat" in union.columns and "meta_bbox" in union.columns:
need_rebuild = not (int(union.loc[0,"meta_feat"]) == feat_count and
tuple(map(float, union.loc[0,"meta_bbox"].split(","))) == bbox)
if need_rebuild:
flood_union = gpd.GeoDataFrame(
{"meta_feat":[feat_count], "meta_bbox":[",".join(map(str,bbox))]},
geometry=[tsunami.geometry.union_all()], crs=tsunami.crs
)
flood_union.to_file(GPKG, layer="tsunami_union", driver="GPKG")
else:
flood_union = gpd.read_file(GPKG, layer="tsunami_union")
例：ルート検証時は AOI で軽量化
# route_gdf: ルート LineString (EPSG:4326)
AOI_m = 15000 # 15km
aoi = route_gdf.to_crs("EPSG:32654").buffer(AOI_m).to_crs(4326)
flood_union_aoi = gpd.overlay(flood_union, gpd.GeoDataFrame(geometry=aoi, crs=4326), how="intersection")
# 以後は flood_union_aoi で交差・近接判定（軽い）
B) 「既に計測したルート」はルート側をキャッシュ（任意）
tsunami_union は“全域・一回”でOKですが、ルートの結果は現在地や避難所・パラメータに依存するので、キー付きでGPKGに保存して再利用が有効です。
例：ルート結果のキャッシュ（キー = 端点・モード・半径）
import hashlib, json, geopandas as gpd
def route_key(home_wgs, dest_wgs, mode="walk", radius_m=12000):
payload = {"home":home_wgs, "dest":dest_wgs, "mode":mode, "r":radius_m}
return hashlib.md5(json.dumps(payload, sort_keys=True).encode()).hexdigest()
key = route_key(home_wgs, (dest_pt.y, dest_pt.x), mode="walk", radius_m=GRAPH_RADIUS_M)
# 取り出し（あれば使う）
try:
cache = gpd.read_file(GPKG, layer="route_cache")
hit = cache.loc[cache["rid"]==key]
if len(hit):
route_gdf = hit.set_crs(4326) # そのまま使える
else:
raise IOError
except Exception:
# 計算 → route_gdf を作ったら保存
route_gdf["rid"] = key
route_gdf.to_file(GPKG, layer="route_cache", driver="GPKG", mode="a") # 追記
A40をアップデートしたら、route_cacheは無効化（レイヤを削除、またはキーにA40のメタ情報を混ぜる）するのが安全です。
A40をアップデートしたら、route_cacheは無効化（レイヤを削除、またはキーにA40のメタ情報を混ぜる）するのが安全です。
ということで
ルート再利用性を高めたい場合は、ルート結果をキー付きでGPKGにキャッシュする機能を追加したい
import hashlib, json, geopandas as gpd
def route_key(home_wgs, dest_wgs, mode="walk", radius_m=12000):
payload = {"home":home_wgs, "dest":dest_wgs, "mode":mode, "r":radius_m}
return hashlib.md5(json.dumps(payload, sort_keys=True).encode()).hexdigest()
key = route_key(home_wgs, (dest_pt.y, dest_pt.x), mode="walk", radius_m=GRAPH_RADIUS_M)
# 取り出し（あれば使う）
try:
cache = gpd.read_file(GPKG, layer="route_cache")
hit = cache.loc[cache["rid"]==key]
if len(hit):
route_gdf = hit.set_crs(4326) # そのまま使える
else:
raise IOError
except Exception:
# 計算 → route_gdf を作ったら保存
route_gdf["rid"] = key
route_gdf.to_file(GPKG, layer="route_cache", driver="GPKG", mode="a") # 追記
これはどこに追記する？
入れる場所は**「目的地（dest_pt）が決まった直後、経路計算（nx.shortest_path）に入る前」です。
ヒットしたらそのまま使い、ミスなら今までどおり計算→検証→保存しつつキャッシュへ追記**します
# ==== ルートキャッシュ（GPKG: route_cache） ====
import hashlib, json, geopandas as gpd
from fiona import listlayers
GPKG = "outputs/shizuoka_hazard.gpkg"
def route_key(home_wgs, dest_wgs, mode="walk", radius_m=GRAPH_RADIUS_M, hazard_meta=None):
payload = {"home": home_wgs, "dest": dest_wgs, "mode": mode, "r": radius_m, "hz": hazard_meta}
return hashlib.md5(json.dumps(payload, sort_keys=True).encode()).hexdigest()
# A40が変わったら無効化できるように簡易メタを混ぜる（件数＋BBOX）
hazard_meta = None
try:
if "tsunami_union" in listlayers(GPKG):
# union保存時にメタ列を持たせていればそれを使う／無ければ tsunami_inundation から作る
tu = gpd.read_file(GPKG, layer="tsunami_union")
if {"meta_feat","meta_bbox"} <= set(tu.columns):
hazard_meta = f'{int(tu.loc[0,"meta_feat"])}|{tu.loc[0,"meta_bbox"]}'
if hazard_meta is None:
ti = gpd.read_file(GPKG, layer="tsunami_inundation")
hazard_meta = f'{len(ti)}|{",".join(map(str, ti.total_bounds))}'
except Exception:
pass
key = route_key(home_wgs, (dest_pt.y, dest_pt.x), mode="walk", radius_m=GRAPH_RADIUS_M, hazard_meta=hazard_meta)
route_gdf_cached = None
if "route_cache" in listlayers(GPKG):
cache = gpd.read_file(GPKG, layer="route_cache")
hit = cache.loc[cache["rid"] == key]
if len(hit):
route_gdf_cached = hit.set_crs(4326)
print("↩︎ ルート: キャッシュヒット")
# ヒット時の扱い：後続の検証/表示にそのまま使えるよう最低限の変数を整備
if route_gdf_cached is not None:
route_gdf = route_gdf_cached[["meters","eta_min_walk","eta_min_walk_safe","eta_min_drive","geometry"]].copy()
# route_latlon / meters 等は既存列から復元（既に保存してある前提）
meters = float(route_gdf["meters"].iloc[0])
eta_walk_min = float(route_gdf["eta_min_walk"].iloc[0])
eta_walk_safe = float(route_gdf["eta_min_walk_safe"].iloc[0])
eta_drive_min = float(route_gdf["eta_min_drive"].iloc[0])
# folium描画用
coords = list(route_gdf.geometry.iloc[0].coords)
route_latlon = [(lat, lon) for lon, lat in coords]
# 以降の「堅牢チェック→保存」はスキップしてOK（または最新A40で再チェックしたいならそのまま通しても良い）
use_cached_route = True
else:
use_cached_route = False
その後のフローはこうしてください：
* use_cached_route == True のとき：
* すでに route_gdf / meters / eta_* / route_latlon が揃っているので、堅牢チェックと保存をスキップして Folium 表示だけ行う（必要なら最新 tsunami_union で再チェックだけ通してもOK）。
* use_cached_route == False のとき：
* これまでどおり nx.shortest_path → 堅牢チェック → ETA計算 → route_latest 保存。
* さらに route_cache に追記します（下の1行を“保存直後”に追加）。
# route_latest を保存した直後にキャッシュへ追記
route_gdf_cache = route_gdf.copy()
route_gdf_cache["rid"] = key
# append（GeoPandas 0.14+）：if_exists="append" が使えます
try:
route_gdf_cache.to_file(GPKG, layer="route_cache", driver="GPKG", if_exists="append")
except TypeError:
# 古い版は append がないので自前で結合して上書き
if "route_cache" in listlayers(GPKG):
old = gpd.read_file(GPKG, layer="route_cache")
gpd.GeoDataFrame(pd.concat([old, route_gdf_cache], ignore_index=True), crs=4326)\
.to_file(GPKG, layer="route_cache", driver="GPKG")
else:
route_gdf_cache.to_file(GPKG, layer="route_cache", driver="GPKG")
まとめ（差し込む位置の目安）
1. 目的地 dest_pt 決定直後に「キャッシュ照会ブロック」を挿入
2. use_cached_route が False なら従来の「計算→検証→保存」ブロックへ進む
3. 保存直後に route_cache へ追記
4. True なら保存はスキップし、地図表示へ直行（または再検証のみ実施）
これで「既知の組み合わせは即復帰／未知は新規計算」の運用ができます。
これはモジュールにして読み込んだ方が良い？
結論：はい。モジュール化して読み込むのが吉です。
理由は「再利用・テスト・差し替え（例：県を静岡→他県）」が一気に楽になるから。