3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@sakutto-panda(サクッとぱんだ)

J-SHIS・地理院APIで住所から地震ハザードをPythonで取得する──政令市ジオコードの罠と非同期実装パターン

3
Posted at

住所から地震リスクを取得する処理を実装した。使うAPIは地理院の住所検索APIとJ-SHISの地震ハザードステーションAPI。どちらも申請不要・APIキーなしで使えるが、ドキュメントに書いてないことが多くて実測で確認しながら進めることになった。

全体の流れ

住所文字列（例: "横浜市青葉区青葉台1-1-1"）
  ↓ 地理院 住所検索API
緯度経度 (lat, lng)
  ↓ J-SHIS PSHM Meshinfo API
30年超過確率（震度5弱以上・6弱以上・6強以上）
  ↓ J-SHIS sstrct Meshinfo API
表層地盤（AVS30・地盤増幅率ARV・微地形区分）
  ↓ avs30_to_liquefaction_risk()
液状化リスク 0〜5

J-SHISは250mメッシュ単位で値を返す。2ステップのAPIコール（PSHM + sstrct）で地震リスクに必要な情報が揃う。

地理院住所検索API：政令市の罠

エンドポイントは https://msearch.gsi.go.jp/address-search/AddressSearch?q=<住所> で、GeoJSONのFeatureCollectionが返ってくる。

class GsiFetcher:
    ADDRESS_SEARCH_URL = "https://msearch.gsi.go.jp/address-search/AddressSearch"

    async def search_address(self, query: str) -> list[GeocodeResult]:
        r = await self.client.get(self.ADDRESS_SEARCH_URL, params={"q": query})
        r.raise_for_status()
        features = r.json()  # list[Feature]
        results = []
        for f in features:
            coords = f.get("geometry", {}).get("coordinates", [])
            if len(coords) < 2:
                continue
            lng, lat = float(coords[0]), float(coords[1])
            results.append(GeocodeResult(
                query=query,
                title=f["properties"].get("title", ""),
                lat=lat,
                lng=lng,
            ))
        return results

一見シンプルだが、政令市の区を含む住所でハマった。

横浜市青葉区のように「〇〇区」が区名だけだと、全国に同名の区が存在する場合に誤った座標が返ることがある。「青葉区」で検索すると仙台市青葉区や名古屋市守山区の一部（かつての青葉区）が候補に混入するケースがあった。

centroid（町丁目の代表座標）の計算結果を確認していて、横浜市青葉区の複数の町丁目が神奈川県ではなく全く別の場所にプロットされていることで気づいた。

修正はgeocode_queryの組み立て方を変えるだけ。政令市の区は「市名 + 区名 + 町名」でクエリを作る。

def build_geocode_query(pref_name: str, city_name: str, area_name: str) -> str:
    """ジオコーディング用クエリを組み立てる。

    政令市の区（横浜市青葉区等）は市名を省くと全国の同名区にヒットする。
    city_nameに区名が含まれる場合（"青葉区"等）はpref_nameを前置する。
    """
    if city_name.endswith("区") and "市" not in city_name:
        # 政令市の区：pref + city + area で検索
        return f"{pref_name}{city_name}{area_name}"
    return f"{city_name}{area_name}"

city_name.endswith("区") and "市" not in city_name という判定で政令市の区かどうかを判別している。「新宿区」のような特別区（東京23区）はそのまま区名だけで十分一意なので該当しない。

J-SHIS PSHM API：プロパティキーはドキュメントにない

J-SHISのAPIドキュメントにはエンドポイントの説明はあるが、レスポンスのプロパティキーがほとんど記載されていない。

PSHM Meshinfo（確率論的地震動予測値）のエンドポイントは：

GET https://www.j-shis.bosai.go.jp/map/api/pshm/{year}/AVR/TTL_MTTL/meshinfo.geojson
    ?position={lng},{lat}&epsg=4326

実際に叩いて確認したプロパティキーが以下。

キー 内容
T30_I45_PS 30年で計測震度4.5以上（≒震度5弱以上）の超過確率
T30_I55_PS 30年で計測震度5.5以上（≒震度6弱以上）の超過確率
T30_I60_PS 30年で計測震度6.0以上（≒震度6強以上）の超過確率
meshcode 250mメッシュコード

「計測震度4.5以上」が「震度5弱以上」に対応するというのも、気象庁の計測震度と震度階の対応表から読み取る必要があった。APIのレスポンスに説明はない。

return JshisHazard(
    intensity_5_lower_30y=_to_float(props.get("T30_I45_PS")),
    intensity_6_lower_30y=_to_float(props.get("T30_I55_PS")),
    intensity_6_upper_30y=_to_float(props.get("T30_I60_PS")),
    mesh_code=str(props.get("meshcode") or "") or None,
    raw_pshm=props,  # 未知のキーも含めて生データを保存しておく
)

raw_pshm に生のpropertiesを丸ごと持っておくのは、後から別のキーが必要になったときにAPIを再取得せずに済むようにするため。

データなしとリスク0を混同しない

J-SHISが features を返さないケースが2種類ある：海上・国境付近などAPIの対応範囲外と、座標系の渡し間違い。どちらも「リスク0」ではなく「データ取得不能」なので、例外を投げて呼び出し元で判断させる。

class JshisDataUnavailable(RuntimeError):
    """J-SHISが該当地点のfeatureを返さない。「データなし」≠「リスク0」。"""

async def fetch_pshm_meshinfo(self, lat: float, lng: float) -> JshisHazard:
    for year in self.PSHM_DATASET_YEARS:  # ["Y2024", "Y2023", ...]
        data = await self._get_pshm_geojson(year, lat, lng)
        features = data.get("features") or []
        if not features:
            continue  # 次の年度でリトライ
        props = features[0].get("properties") or {}
        return JshisHazard(...)

    raise JshisDataUnavailable(
        f"全年度でfeatureが取得できませんでした (lat={lat}, lng={lng})"
    )

年度ごとにフォールバックしているのは、最新年度のデータが未反映の地域があるため。

J-SHIS sstrct API：表層地盤と液状化リスク

表層地盤情報は別エンドポイント（sstrct）から取得する。

GET https://www.j-shis.bosai.go.jp/map/api/sstrct/{version}/meshinfo.geojson
    ?position={lng},{lat}&epsg=4326

返ってくるプロパティ：

キー 内容
AVS 30m平均S波速度（m/s）。地盤が軟弱なほど小さい
ARV 表層地盤増幅率。揺れやすい地盤ほど大きい
JCODE 微地形区分コード（"9"=ローム台地 等）
JNAME 微地形区分名（"埋立地"、"谷底低地" 等）
meshcode メッシュコード

バージョンは V3（新計算）と V2（旧計算）があり、V3 → V2 の順にフォールバックする。meshcodeの末尾仕様が違う点に注意が必要で、V3は末尾に N がなく、V2は末尾 N ありで返ってくる。meshcodeを他の処理に使う場合は統一してから使う。

for version in ["V3", "V2"]:
    data = await self._get_sstrct_geojson(version, lat, lng)
    features = data.get("features") or []
    if features:
        props = features[0].get("properties") or {}
        # V2のmeshcodeは末尾にNが付く場合がある
        mesh = str(props.get("meshcode") or "").rstrip("N") or None
        return props

液状化APIは存在しない：AVS30から自前で算出

J-SHISの公開APIには液状化リスクのメッシュ情報が含まれていない。液状化は地盤の柔らかさと強く相関するため、sstrctから取得できるAVS30（30m平均S波速度）を使って液状化リスクを算出した。

def avs30_to_liquefaction_risk(avs30: float | None) -> int | None:
    """AVS30（m/s）から液状化リスク 0〜5 に変換。

    閾値は国土交通省「宅地液状化防止事業技術指針」と
    地盤工学会の標準的な区分に準拠。
    """
    if avs30 is None:
        return None
    if avs30 >= 300: return 0  # 岩盤・密実な礫層（ほぼリスクなし）
    if avs30 >= 200: return 1  # 洪積台地・ローム層
    if avs30 >= 150: return 2  # 沖積平野・粘性土
    if avs30 >= 100: return 3  # 沖積低地・河川沿い
    if avs30 >= 50:  return 4  # 埋立地・旧河道・砂州
    return 5                   # 極軟弱地盤（非常に高リスク）

AVS30が低い（地盤が軟らかい）ほど液状化リスクが高い。埋立地・旧河道・砂州などはAVS30が低く出る傾向があり、微地形区分（JNAME）と組み合わせると現実に即した判定ができる。

非同期・リトライ・キャッシュの実装パターン

J-SHISもGSIも1アドレスにつき複数回のAPIコールが発生する（PSHM + sstrct の2回など）。複数の住所を並行処理するとリクエストが重なるので、httpx の非同期クライアントに tenacity のリトライを組み合わせた。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import httpx

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(httpx.HTTPError),
    reraise=True,
)
async def _get_pshm_geojson(self, year: str, lat: float, lng: float) -> dict:
    url = f"{self.BASE_URL}/pshm/{year}/AVR/TTL_MTTL/meshinfo.geojson"
    r = await self.client.get(url, params={"position": f"{lng},{lat}", "epsg": "4326"})
    r.raise_for_status()
    return r.json()

wait_exponential(min=2, max=10) で2秒→4秒→8秒（上限10秒）のバックオフ。共通HTTPクライアントの設定（lib/http.py）で httpx.Limits(max_connections=10) を指定し、同時接続数を制御している。

同一メッシュのキャッシュ

250mメッシュ単位でデータが返ってくるため、同じメッシュ内の複数住所を処理する場合はAPIを再取得しない。ETLで東京23区3000件超を処理したときに効果が大きかった。

@staticmethod
def _mesh_key(lat: float, lng: float) -> tuple[int, int]:
    """250mメッシュ単位のキャッシュキーを生成する。
    
    0.002083° × 0.003125° の格子（≒250m）で丸める。
    同一メッシュ内の複数地点は同じキーになる。
    """
    return (int(lat * 480), int(lng * 320))

async def fetch_meshinfo(self, lat: float, lng: float) -> JshisHazard:
    key = self._mesh_key(lat, lng)
    if key in self._mesh_cache:
        return self._mesh_cache[key]

    hazard = await self.fetch_pshm_meshinfo(lat, lng)
    sstrct = await self.fetch_sstrct_meshinfo(lat, lng)
    if sstrct:
        hazard.surface_amp_factor = _to_float(sstrct.get("ARV"))
        hazard.surface_avs30 = _to_float(sstrct.get("AVS"))
        hazard.liquefaction_risk = avs30_to_liquefaction_risk(hazard.surface_avs30)

    self._mesh_cache[key] = hazard
    return hazard

まとめ

J-SHIS + 地理院のハマりポイントをまとめる：

  • 政令市のジオコーディング：区名だけで検索すると全国の同名区にヒットする。市名を含めたクエリを組み立てる
  • J-SHISのプロパティキー：公式ドキュメントに記載がない。実際にAPIを叩いてレスポンスのキーを確認する必要がある
  • 液状化データは存在しない：AVS30から自前で算出する。閾値は国交省指針に準拠
  • データなしとリスク0は別：features が空のときは例外を投げて呼び出し元で判断させる
  • 同一メッシュのキャッシュ：大量の住所を処理するときに効果が大きい

シリーズ記事：

この記事は Zenn にも同じ内容を投稿しています。

3
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?