GIS奮闘記

元GISエンジニアの技術紹介ブログ。主にPythonを使用。

スポンサーリンク

Python で気象庁防災情報XMLの地震情報をCSV出力する方法

さて、本日は気象庁防災情報XMLの地震情報をCSV出力しようと思います。

気象庁防災情報XMLとは

気象庁は過去長年にわたり、それぞれの防災情報毎に情報の性質・利用形態などを考慮し、気象庁独自の電文形式(フォーマット)を作成してきました。この方式は、防災情報の種類が少なく、情報の伝達がFAXや低速の通信回線の時代はそれぞれの情報に適していましたが、高度にICT化された社会において、より詳細で高度化された防災情報をより効果的に活用していただくために、新たな防災情報の提供様式を検討すべきと考え、「気象庁防災情報XMLフォーマット」を策定することとし、平成23年5月12日より運用を開始しました。(出典:気象庁

気象庁防災情報XMLの仕様について

気象庁のサイトにまとまっているのでこちらをご参照ください。

また、地震情報に関してはこちらにもまとまっています。

今回取得するデータ

こちらのサイトから地震情報を取得します。(今回は長期フィードを使用)

f:id:sanvarie:20220320070157p:plain

この中には火山情報も入っており、そこを抜いて地震情報のみを取得します。以下画像の赤枠が地震情報となっており、同じようにURLに「VXSE」を含んだXMLから地震情報を取得したいと思います。

f:id:sanvarie:20220321101929p:plain

取得する項目

さまざまな項目があるのですが、今回は重要そうな以下の項目を取得しようと思います。

  • イベントID
  • 地震発生場所
  • 地震発生時刻
  • 観測点で地震を検知した時刻
  • 緯度
  • 経度
  • マグニチュード
  • 県名称
  • 市町村名称
  • 最大震度
  • 震度観測点の情報(一つの地震に対して複数の情報があるため、セル内で半角スペースで区切ります)

実行環境

macOS 11.3.1
Python 3.9.6

サンプルコード

気象庁防災情報XMLから地震情報を取得するサンプルです。

# -*- coding: utf-8 -*-
import re
import csv
import urllib.request
from bs4 import BeautifulSoup

# CSV出力先
output_file = "eathquake.csv"

# CSVヘッダー
header = ["イベントID","地震発生場所","地震発生時刻","観測点で地震を検知した時刻","緯度","経度",
          "マグニチュード","県名称","市町村名称","最大震度","震度観測点の情報"]

with open(output_file, 'w', encoding='utf-8') as f:
    writer = csv.writer(f, lineterminator="\n")
    writer.writerow(header)

    # 対象URLにリクエストを送る
    url = "https://www.data.jma.go.jp/developer/xml/feed/eqvol_l.xml"
    res = urllib.request.urlopen(url).read()
    soup = BeautifulSoup(res)

    # 地震に関する情報のみを抽出
    search = re.compile('.*VXSE.*')

    for vxse in soup.find_all(text=search):

        # 対象URLにリクエストを送る
        res = urllib.request.urlopen(vxse).read()
        soup = BeautifulSoup(res)

        # イベントIDを取得
        event_id = soup.find("eventid").text

        for content in soup.find_all("body"):
            earthquake = content.find("earthquake")

            if earthquake is not None: # earthquakeの情報がない地震情報は対象外とする
                # 地震の各種情報を取得
                location = earthquake.find("name").text # 地震発生場所
                origintime = earthquake.find("origintime").text # 地震の発生した時刻
                arrivaltime = earthquake.find("arrivaltime").text # 観測点で地震を検知した時刻
                
                coordinate = earthquake.find("jmx_eb:coordinate").text # "北緯37.8度 東経141.6度 深さ 50km" +37.8+141.6-50000
                lat = coordinate[1:coordinate[1:].find("+") + 1]
                lon = coordinate[coordinate[1:].find("+") + 2:coordinate[1:].find("+") + 7]
                lat = int(lat[0:2])+(int(lat[3:])/60)+(0/3600)
                lon = int(lon[0:3])+(int(lon[4:])/60)+(0/3600)
                magnitude = earthquake.find("jmx_eb:magnitude").text 

            intensity = content.find("intensity")
            if intensity is not None: # intensityの情報がない地震情報は対象外とする:
                for pref in intensity.find_all("pref"):
                    
                    prefectuer = pref("name")[0].text                  

                    for city in pref.find_all("city"):
                        city_name = city("name")[0].text
                        maxint = city("maxint")[0].text #地震情報(震源・震度に関する情報)の各市町村での震度観測点の最大震度。

                        write_list = []
                        write_list.append(event_id)
                        write_list.append(location)
                        write_list.append(origintime)
                        write_list.append(arrivaltime)
                        write_list.append(lat)
                        write_list.append(lon)
                        write_list.append(magnitude)
                        write_list.append(prefectuer)
                        write_list.append(city_name)
                        write_list.append(maxint)

                        intensity_station = ""
                        for intensitystation in pref.find_all("intensitystation"): #地震情報(震源・震度に関する情報)の震度観測点の情報 *印は気象庁以外の震度観測点についての情報
                            intensity_station = intensity_station + intensitystation("name")[0].text + " " # 半角スペースで区切る
                        
                        write_list.append(intensity_station)

                        # CSV書き込み
                        writer.writerow(write_list)

出力結果は以下のようになりました。 f:id:sanvarie:20220321102555p:plain

さいごに

いかがでしたでしょうか?位置情報を含んでいるので出力結果を可視化してみてもいいかもしれないですね。また、気象庁防災情報XMLは地震情報以外にも様々な情報を提供しているので今後は他の情報でも遊んでみようと思います。本日は以上です。