# JMA防災情報XMLで津波関連の電文を監視する検証メモ
EEWや揺れ検知トリガーの仕組みを作ろうとすると、AIからは「端末の加速度センサーで揺れを検知する」「実際の揺れをハードウェアで拾う」といった提案が出ることがあります。
しかし今回やりたいことは、それではありません。
目的は、**端末の揺れを検知することではなく、気象庁の公式配信であるJMA防災情報XMLから、津波関連の電文を監視してトリガーを発火させること**です。
たとえば、以前の津波警報では、ロシアのカムチャッカ半島沖の地震により、日本沿岸に津波警報が発表されました。
このようなケースでは、自宅が揺れていなくても、日本向けの津波関連電文を検知できる必要があります。
つまり、今回作りたいのは以下のような仕組みです。
* JMA防災情報XMLのフィードを定期取得する
* 津波警報・津波注意報・津波情報などの電文を検知する
* 対象地域に該当する場合だけ通知する
* 取消・解除・更新も状態遷移として扱う
## JMA防災情報XMLの「実パーサ」とは何か
JMA防災情報XMLの「実パーサ」とは、単にXMLを読み込むだけではなく、気象庁XMLの仕様に沿って、必要な項目を確実に抽出する処理のことです。
たとえば津波関連であれば、次のような情報を取り出します。
* 電文タイトル
* 発表時刻
* 情報種別
* 発表・取消・訂正などの区分
* 見出し文
* 津波警報・津波注意報・津波予報などの種別
* 対象の津波予報区
* 対象地域コード
* 電文URL
今回の目的では、JMAの地震・火山系フィードである `eqvol.xml` を監視します。
EQVOL_FEED = "[https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml](https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml)"
## 最初に生成されたコード
最初に、GPTで以下のような常駐監視スクリプトを生成しました。
import time
import requests
import feedparser
import xml.etree.ElementTree as ET
EQVOL_FEED = "[https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml](https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml)"
NAMESP = {
"jmx": "[http://xml.kishou.go.jp/jmaxml1/](http://xml.kishou.go.jp/jmaxml1/)",
"jmx_ib": "[http://xml.kishou.go.jp/jmaxml1/informationBasis1/](http://xml.kishou.go.jp/jmaxml1/informationBasis1/)",
"jmx_eb": "[http://xml.kishou.go.jp/jmaxml1/elementBasis1/](http://xml.kishou.go.jp/jmaxml1/elementBasis1/)",
"jmx_seis": "[http://xml.kishou.go.jp/jmaxml1/body/seismology1/](http://xml.kishou.go.jp/jmaxml1/body/seismology1/)",
"jmx_tsunami": "[http://xml.kishou.go.jp/jmaxml1/body/tsunami1/](http://xml.kishou.go.jp/jmaxml1/body/tsunami1/)",
}
last_modified = None
seen_ids = set()
def fetch_feed():
headers = {}
if last_modified:
headers["If-Modified-Since"] = last_modified
```
r = requests.get(EQVOL_FEED, headers=headers, timeout=10)
if r.status_code == 304:
return None, None
r.raise_for_status()
return r.text, r.headers.get("Last-Modified")
```
def parse_entry_xml(url):
r = requests.get(url, timeout=10)
r.raise_for_status()
```
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
title = head.findtext("jmx_ib:Title", default="", namespaces=NAMESP)
report_dt = head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP)
info_kind = head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP)
info_type = head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP)
headline = head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP)
areas = []
kinds = []
for item in root.findall(
".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item",
NAMESP
):
kind_name = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
kinds.append(kind_name)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
areas.append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
return {
"title": title,
"report_dt": report_dt,
"info_kind": info_kind,
"info_type": info_type,
"headline": headline,
"kinds": list(set(kinds)),
"areas": areas,
"url": url,
}
```
def is_tsunami_trigger(rec):
if "津波" not in (rec["info_kind"] or rec["title"] or rec["headline"]):
return False
```
keywords = [
"大津波警報",
"津波警報",
"津波注意報",
"津波予報",
"津波情報",
]
if any(k in (rec["headline"] or "") for k in keywords):
return True
return any(k in keywords for k in rec["kinds"])
```
def loop():
global last_modified
```
while True:
try:
feed_text, lm = fetch_feed()
if feed_text:
d = feedparser.parse(feed_text)
for e in d.entries:
if e.id in seen_ids:
continue
seen_ids.add(e.id)
rec = parse_entry_xml(e.link)
if is_tsunami_trigger(rec):
print(
"[TSUNAMI TRIGGER]",
rec["report_dt"],
rec["headline"],
rec["kinds"][:3],
"areas=",
len(rec["areas"]),
)
if lm:
last_modified = lm
except Exception as ex:
print("error:", ex)
time.sleep(60)
```
if **name** == "**main**":
loop()
しかし、このコードをNotebook上で実行しても、期待したように結果が出ませんでした。
## 原因の切り分け
この時点で考えられる原因は、だいたい次のどれかです。
1. いま津波関連の電文が出ていない
2. `eqvol.xml` は取得できているが、津波関連ではない電文しか流れていない
3. XMLの取得やパースで失敗している
4. Notebook上で無限ループを回しているため、処理が終わらず状況が見えない
5. 処理は動いているが、条件にヒットしないため何も表示されていない
特にNotebookでは、最初から無限ループを動かすと、
「動いていない」のか、
「動いているが何もヒットしていない」のか、
「処理が詰まっている」のかが分かりにくくなります。
そのため、まずは **1回だけ実行するドライラン用コード** に切り替えました。
## 通信確認
まず、JMAのフィードに通信できるか確認します。
import requests
EQVOL_FEED = "[https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml](https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml)"
r = requests.get(EQVOL_FEED, timeout=15)
print("status:", r.status_code)
print(r.text[:1000])
結果は以下のようになりました。
status: 200 <?xml version="1.0" encoding="utf-8"?> <feed xmlns="http://www.w3.org/2005/Atom" lang="ja"> <title>é«é »åº¦ï¼å°éç«å±±ï¼</title> <subtitle>JMAXML publishing feed</subtitle> <updated>2025-11-06T05:01:14+09:00</updated> ...
`status: 200` なので、通信自体は成功しています。
文字化けしているように見えますが、この段階ではまず通信できていることを確認できれば十分です。
## feedparserでフィードを確認
次に、`feedparser` でAtomフィードとして読めているか確認します。
import feedparser
feed = feedparser.parse("[https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml](https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml)")
print("entries:", len(feed.entries))
for e in feed.entries[:5]:
print(e.title, e.link)
ここで `entries` が0でなければ、フィードの取得と基本的なパースはできています。
## Notebookの出力確認
次に、Notebook自体の出力が壊れていないか確認しました。
import sys
import time
import pathlib
import subprocess
print("STEP A1: print works?", flush=True)
sys.stdout.write("STEP A2: stdout.write works?\n")
sys.stdout.flush()
sys.stderr.write("STEP A3: stderr works?\n")
sys.stderr.flush()
for i in range(3):
print(f"tick {i}", flush=True)
time.sleep(0.5)
pathlib.Path("notebook_alive.txt").write_text("alive\n")
print("STEP A4: wrote notebook_alive.txt", flush=True)
ret = subprocess.run(
["echo", "STEP A5: subprocess echo"],
capture_output=True,
text=True,
)
print(ret.stdout, end="")
print("STEP A DONE", flush=True)
結果は以下です。
STEP A1: print works? STEP A2: stdout.write works? STEP A3: stderr works? tick 0 tick 1 tick 2 STEP A4: wrote notebook_alive.txt STEP A5: subprocess echo STEP A DONE
Notebookの出力自体は正常でした。
つまり、問題はNotebookの表示ではなく、最初の常駐監視コードの設計にあります。
## ドライラン用コード
次に、無限ループを使わず、最新5件だけを確認するドライランコードを実行しました。
import requests
import feedparser
import xml.etree.ElementTree as ET
EQVOL_FEED = "[https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml](https://www.data.jma.go.jp/developer/xml/feed/eqvol.xml)"
NAMESP = {
"jmx": "[http://xml.kishou.go.jp/jmaxml1/](http://xml.kishou.go.jp/jmaxml1/)",
"jmx_ib": "[http://xml.kishou.go.jp/jmaxml1/informationBasis1/](http://xml.kishou.go.jp/jmaxml1/informationBasis1/)",
"jmx_eb": "[http://xml.kishou.go.jp/jmaxml1/elementBasis1/](http://xml.kishou.go.jp/jmaxml1/elementBasis1/)",
"jmx_seis": "[http://xml.kishou.go.jp/jmaxml1/body/seismology1/](http://xml.kishou.go.jp/jmaxml1/body/seismology1/)",
"jmx_tsunami": "[http://xml.kishou.go.jp/jmaxml1/body/tsunami1/](http://xml.kishou.go.jp/jmaxml1/body/tsunami1/)",
}
def parse_entry_xml(url):
r = requests.get(url, timeout=15)
r.raise_for_status()
```
root = ET.fromstring(r.content)
head = root.find(".//jmx_ib:Head", NAMESP)
title = head.findtext("jmx_ib:Title", default="", namespaces=NAMESP)
report_dt = head.findtext("jmx_ib:ReportDateTime", default="", namespaces=NAMESP)
info_kind = head.findtext("jmx_ib:InfoKind", default="", namespaces=NAMESP)
info_type = head.findtext("jmx_ib:InfoType", default="", namespaces=NAMESP)
headline = head.findtext("jmx_ib:Headline/jmx_eb:Text", default="", namespaces=NAMESP)
areas = []
kinds = []
for item in root.findall(
".//jmx_eb:Body//jmx_eb:Information[@type='津波予報領域表現']/jmx_eb:Item",
NAMESP
):
kind_name = item.findtext("jmx_eb:Kind/jmx_eb:Name", default="", namespaces=NAMESP)
if kind_name:
kinds.append(kind_name)
for a in item.findall("jmx_eb:Areas/jmx_eb:Area", NAMESP):
areas.append({
"name": a.findtext("jmx_eb:Name", default="", namespaces=NAMESP),
"code": a.findtext("jmx_eb:Code", default="", namespaces=NAMESP),
})
return {
"title": title,
"report_dt": report_dt,
"info_kind": info_kind,
"info_type": info_type,
"headline": headline,
"kinds": list(set(kinds)),
"areas": areas,
"url": url,
}
```
def is_tsunami_trigger(rec):
text_pool = " ".join([
rec.get("title", ""),
rec.get("headline", ""),
rec.get("info_kind", ""),
])
```
if "津波" not in text_pool:
return False
keywords = [
"大津波警報",
"津波警報",
"津波注意報",
"津波予報",
"津波情報",
]
if any(k in text_pool for k in keywords):
return True
return any(k in rec.get("kinds", []) for k in keywords)
```
r = requests.get(EQVOL_FEED, timeout=15)
print("feed status:", r.status_code, "bytes:", len(r.content))
d = feedparser.parse(r.text)
print("entries:", len(d.entries))
for e in d.entries[:5]:
print("ENTRY:", e.get("published", ""), e.get("title", ""))
```
try:
rec = parse_entry_xml(e.link)
print(" InfoKind:", rec["info_kind"], "| InfoType:", rec["info_type"])
print(" Title :", rec["title"])
print(" Headline:", (rec["headline"] or "")[:120])
print(" kinds :", rec["kinds"], "| areas:", len(rec["areas"]))
print(" -> TRIGGER?", is_tsunami_trigger(rec))
except Exception as ex:
print(" ERROR parsing:", ex)
```
実行結果は以下です。
feed status: 200 bytes: 27328 entries: 37 ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 薩摩硫黄島 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 諏訪之瀬島 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 霧島山（新燃岳） 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 浅間山 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False ENTRY: 降灰予報（定時） InfoKind: 降灰予報 | InfoType: 発表 Title : 火山名 草津白根山（白根山（湯釜付近）） 降灰予報（定時） Headline: kinds : [] | areas: 0 -> TRIGGER? False
## この結果から分かったこと
この結果を見ると、次のことが分かります。
* JMAの `eqvol.xml` にはアクセスできている
* フィードは取得できている
* `feedparser` でエントリも取得できている
* XMLの個別電文もパースできている
* ただし、最新の電文は降灰予報であり、津波関連ではない
* そのため、津波トリガー条件にはヒットしていない
つまり、最初のコードが完全に失敗していたわけではありません。
**いま津波関連の電文が流れていないため、何も表示されなかった**と考えられます。
## 最初のコードの問題点
最初のコードは、常駐監視を前提としていました。
while True: ... time.sleep(60)
このような無限ループは、Notebook上では状況が分かりにくくなります。
今回のように、津波関連の電文が存在しない場合は、条件にヒットせず、何も表示されません。
その結果、見た目上は「動作していない」ように見えてしまいます。
そのため、開発初期は以下の順番が安全です。
1. 通信確認
2. フィード件数確認
3. 最新数件の中身確認
4. トリガー判定確認
5. 最後に常駐化
## 実運用に進める場合
今後、実運用に進めるなら、次のように拡張します。
### 1. 対象地域でフィルタする
たとえば静岡沿岸など、自分に関係する津波予報区コードだけに絞ります。
TARGET_AREA_CODES = {
"322", # 例: 実際のコードは公式コード表で確認
}
そして、`areas` に対象コードが含まれる場合だけ通知します。
def is_target_area(rec):
return any(
area.get("code") in TARGET_AREA_CODES
for area in rec.get("areas", [])
)
### 2. 取消・解除を状態遷移として扱う
`InfoType` が `取消` の場合は、単に無視するのではなく、通知状態を解除する処理にします。
if rec["info_type"] == "取消":
print("[TSUNAMI CANCEL]", rec["report_dt"], rec["headline"])
### 3. `extra.xml` も併用する
`eqvol.xml` だけでなく、随時情報系の `extra.xml` も監視すると、取りこぼしを減らせる可能性があります。
EXTRA_FEED = "[https://www.data.jma.go.jp/developer/xml/feed/extra.xml](https://www.data.jma.go.jp/developer/xml/feed/extra.xml)"
### 4. Notebookではなく常駐スクリプト化する
Notebookは検証には便利ですが、常時監視には向きません。
実運用では、以下のような形にするのがよいです。
* Pythonスクリプトとして保存
* systemdで常駐化
* ログファイルへ出力
* 例外時も停止しない
* 通知処理をLINEや音声読み上げへ接続する
## まとめ
今回の検証では、JMA防災情報XMLの `eqvol.xml` を取得し、最新電文をパースするところまで確認できました。
結果として、コードが動いていなかったのではなく、**現在の最新電文に津波関連情報が存在しなかったため、トリガー条件に一致しなかった**ことが分かりました。
Notebookで検証する場合は、最初から無限ループにせず、まずはドライランで以下を確認すると原因を切り分けやすくなります。
* フィードが取得できているか
* エントリが存在するか
* 個別XMLが読めているか
* `InfoKind` や `Title` が何になっているか
* 津波関連キーワードにヒットしているか
この方向であれば、端末の揺れを検知するハードウェアは不要です。
カムチャッカ沖のような国外震源でも、日本向けに津波警報・注意報・津波情報が発表されれば、それをトリガーとして検知できる可能性があります。
コメント