はじめに
この記事は、生成AIによる支援を受けて作成しています。コーディングの実習目的に個人利用の範囲で作成され、地震の予報・予測等の実務のためのものではありません。
「世界中の地震も、日本国内の地震も、1つのダッシュボードで見たい」
地震の多い昨今、どの強さの地震が、毎日どれぐらい起きているのかはどなたでも興味のあるところだと思います。
そんな思いつきから、AWS Lambda + EventBridge で5分おきに地震情報を取得し、SORACOM Harvestに流し込む仕組みを作りました。データソースは2つ。
- USGS Earthquake Hazards Program API（アメリカ地質調査所、全世界カバー）
- 気象庁 防災情報XML（日本国内、体感震度つき）
どちらも無料で叩ける公開APIです。構想自体はこれらのソースを使うことを決めてから短時間で終わりました。ただし実際に本番相当のデータで動かしてみると、ドキュメントだけを見て実装した部分に思わぬ落とし穴があり、その対応を含めて共有します。
全体構成
EventBridge（5分おき）
│
▼
AWS Lambda ──── USGS API
│ └─── 気象庁 list.json
│
├─ SSM Parameter Store（前回取得時刻を保存）
│
▼
SORACOM Inventory /publish（device_secret方式）
│
▼
SORACOM Harvest（蓄積・可視化）
- Lambdaは5分おきにEventBridgeで起動
- 毎回「前回の実行からいままで」の時間ウィンドウで両APIを叩く
- 取得した地震データを整形して、SORACOM Inventoryの
/publishエンドポイントへdevice_secret方式で送信 - SORACOM Harvestに蓄積し、Lagoonなどで可視化する想定
コードの全体はこの記事の末尾のリポジトリを参照してください。ここでは設計上のポイントと、実装中にハマった点を紹介します。
ハマりどころ①: 気象庁APIの座標は「専用フィールド」に入っていない
気象庁の https://www.jma.go.jp/bosai/quake/data/list.json を叩くと、こんなレスポンスが返ってきます（一部抜粋）。
{
"eid": "20260922013910",
"at": "2026-09-22T01:39:00+09:00",
"anm": "熊本県熊本地方",
"cod": "+32.6+130.7+0/",
"mag": "3.5",
"maxi": "3"
}
最初は「latとかlonとかdepthみたいな個別フィールドがあるだろう」と思い込んで実装したのですが、そんなフィールドはどこにもありません。緯度・経度・深さは全部 cod という1つの文字列に詰め込まれています。
cod: "+32.2+130.4-10000/"
緯度 経度 深さ(m、地下方向がマイナス)
+32.2 +130.4 -10000 → 北緯32.2度, 東経130.4度, 深さ10km
これは ISO 6709 風の座標表記で、符号+緯度 符号+経度 符号+深さ(m) を並べて / で終端する形式です。深さの単位はメートルなので、キロメートルに直すには1000で割る必要があります。
存在しないフィールドをdict.get()で読むと、Pythonは例外を出さずに黙ってNoneを返します。今回のケースでは「緯度・経度が取れないレコードはスキップ」というロジックにしていたため、気象庁データが1件も送信されないまま、エラーも出ずに動き続けるという一番気づきにくいバグになっていました。ログ上も「フィルタ後: 0件」としか出ないので、本当に地震が無いのか、バグで0件なのか区別がつきません。
教訓として、公開APIのレスポンス構造は、想定で実装せず必ず実際に叩いて確認するようにしましょう。特に「無くても例外にならず静かにスキップされる」実装は、テストで気づきにくいので要注意です。
パース処理はこんな感じにしました。
import re
COD_RE = re.compile(r"^([+-]\d+(?:\.\d+)?)([+-]\d+(?:\.\d+)?)([+-]\d+(?:\.\d+)?)/?$")
def parse_cod(cod: str):
"""例: '+32.2+130.4-10000/' → (32.2, 130.4, 10.0)"""
if not cod:
return None, None, None
m = COD_RE.match(cod)
if not m:
return None, None, None
lat_str, lon_str, depth_m_str = m.groups()
depth_km = abs(float(depth_m_str)) / 1000
return float(lat_str), float(lon_str), depth_km
ハマりどころ②: 「取りこぼし防止」のつもりが取りこぼす設計だった
定期実行するLambdaでよくあるパターンとして、「前回どこまで取得したか」をどこかに保存して、次回はその続きから取得する、という設計にしました。保存先はSSM Parameter Storeです。
last_fetched = get_last_fetched_time() # SSMから読む
start_str = last_fetched if last_fetched else フォールバック時刻
# ... USGS・気象庁を取得してSORACOM Harvestへ送信 ...
save_last_fetched_time(end_str) # 最初はここを無条件で呼んでいた
一見自然な実装ですが、これだと送信が一部失敗しても、取得時刻は無条件で先に進んでしまいます。SORACOM側が一時的に不調だったり、USGS・気象庁のAPIがタイムアウトしたりすると、そのウィンドウのデータは二度と再取得されず、静かに欠損します。「取りこぼし防止」を謳いながら、実際には取りこぼす設計になっていたわけです。
対策は単純で、送信・取得のどちらかが1件でも失敗したらSSMを更新しないようにしました。
usgs_result = fetch_usgs(start_str, end_str) # 失敗時は None
jma_result = fetch_jma(start_dt, end_dt) # 失敗時は None
fetch_failed = usgs_result is None or jma_result is None
# ...送信ループで error をカウント...
if error == 0 and not fetch_failed:
save_last_fetched_time(end_str)
else:
logger.warning("失敗があったためSSMを更新しません → 次回同じ期間を再取得")
ポイントは、fetch_usgs/fetch_jma が**「取得失敗」と「フィルタ後0件」を明確に区別して返す**ことです。どちらも空リストを返してしまうと、この判定ができません。
def fetch_usgs(start_str, end_str) -> list[dict] | None:
try:
...
except Exception as e:
logger.error(f"USGS取得失敗: {e}")
return None # ← [] ではなく None
...
return results # 0件でも正常なら [] を返す
ハマりどころ③: リトライしても直らないエラーをリトライしていた
SORACOM Harvestへの送信が失敗したときのリトライも、最初は雑に「例外が出たら5回・10秒間隔でリトライ」としていました。しかし、device_secretが誤っている・ペイロードの形式が不正、といった4xx系のエラーはリトライしても絶対に直りません。それを5回×10秒待ってから諦める設計だと、1件あたり最大50秒消費してしまい、地震が同時多発したときにLambdaのタイムアウトを圧迫しかねません。
そこで、4xxは専用の例外でリトライせず即失敗、5xx・ネットワークエラーだけリトライする形にしました。
class PermanentSendError(Exception):
"""4xxなど、リトライしても解決しない送信エラー"""
def send_to_harvest(data, retry_count=5, delay=10):
for attempt in range(retry_count):
try:
response = requests.post(url, headers=headers, data=json.dumps(data), timeout=10)
if 400 <= response.status_code < 500:
raise PermanentSendError(f"恒久的エラー (HTTP {response.status_code})")
response.raise_for_status() # 5xxはここで例外 → リトライへ
return
except PermanentSendError:
raise # 即座に呼び出し元へ
except Exception as e:
logger.warning(f"送信失敗、リトライ ({attempt + 1}/{retry_count}): {e}")
if attempt < retry_count - 1:
time.sleep(delay)
raise Exception(f"{retry_count}回失敗しました")
文字列マッチで恒久的エラーを判定する方法も動くには動きますが、except PermanentSendError のように型で判定したほうが後から読んだときの見通しが良くなります。
まとめ
- 気象庁の
list.jsonは緯度・経度・深さがcodフィールドにISO6709風でまとめて入っています。個別フィールドは無いので、想定で実装せず必ず実データで確認しましょう - 定期実行バッチの「前回取得時刻」を保存する設計は、失敗時にどう振る舞うかまで設計しないと、簡単に「取りこぼし防止のつもりで取りこぼす」実装になってしまいます
- リトライは「リトライすれば直る可能性があるエラーか」を区別しましょう。4xxのような恒久的エラーは機械的にリトライしないようにします
どれも「動くには動くが、失敗時に静かにデータが消える」タイプの不具合で、実運用に投入してから気づくと発見が遅れがちです。定期実行バッチを書くときは、正常系だけでなく「取得が失敗したら」「送信が失敗したら」の2パターンを必ず洗い出すのがおすすめです。
Notice
本記事のコードは USGS Earthquake Hazards Program および 気象庁 防災情報XMLフォーマット の公開APIを利用しています。気象庁データの二次利用にあたっては、利用規約・著作権表示のルールを事前にご確認ください。本記事は個人の検証目的での実装例であり、両機関との提携・承認関係はありません。
SORACOM Inventory / Harvest は株式会社ソラコムのサービスです。本記事は同社のInventory APIを利用した非公式・独立した実装例であり、公式な提携・承認関係はありません。