タグアーカイブ: 地震情報

出張中に地震もあったりで、P2P地震情報 API の通知を使えないかと考えてみた。

• https://www.p2pquake.net/

緊急地震速報が出たときに、Raspberry Pi の cec-client で TV などを点けるなど処理をさせたいな。

ということで、Gemini でプログラムのひな形を作らせてみた。Systemd で起動するようにちょい修正だな。

import asyncio
import json
import websockets

# あなたが住んでいる、または通知を受け取りたい地域（例：東京都）
MY_PREF = "東京都"
MY_CITIES = ["千代田区", "新宿区"]  # 必要に応じて市区町村まで絞る場合

async def listen_eew():
  uri = "wss://api.p2pquake.net/v2/ws"
  async with websockets.connect(uri) as websocket:
    print("地震情報の受信を待機中...")
    async for message in websocket:
      data = json.loads(message)

      # --- パターンA: 震度情報（code: 551 など）の場合 ---
      if data.get("code") == 551:
        # 各地の震度データから自分の地域を探す
        for pref_data in data.get("pref", []):
          if pref_data.get("name") == MY_PREF:
            max_scale = pref_data.get("maxScale")
            print(f"【地元で地震を検知】{MY_PREF} 予想/観測震度: {max_scale}")
            # TODO: 通知処理をここに書く

      # --- パターンB: 緊急地震速報（code: 554 / 556 など）の場合 ---
      elif data.get("code") in [554, 556]:
        # 警報・予報の対象地域に自分の地域が含まれているかチェック
        # (※実際のJSON構造に合わせたキー指定が必要です)
        areas = data.get("areas", [])
        for area in areas:
          if MY_PREF in area.get("pref", ""):
            print(f"【緊急地震速報】お住まいの地域が含まれています: {area}")
            # TODO: 強い揺れへの警戒・音声アラートなど

asyncio.run(listen_eew())