はじめに
つい最近、台湾で大きな地震がありましたね。
一人でも多くの無事と一日も早い復興を願ってます。
さて、今回は台湾の最新の地震情報を取得できるAPIについてご紹介したいと思います。
台湾には中央気象署という気象業務を行う公的機関があり、日本でいう気象庁のような役割を果たしています。
中央気象署は、台湾国内の最新の気象情報を入手できるオープンデータを提供しており、会員登録をすれば誰でも無料で利用することができます。
まずは会員登録を済ませましょう。
中央気象署会員になる
1. 上記のサイトにジャンプし、利用規約を確認したら「同意」をクリックします。
2. 登録情報を入力します。 Google翻訳を掛ければ読めると思いますが、下記に解説します。
- 電子郵件 - メールアドレス
- 密碼 - パスワード（6文字以上、12文字以下）（スペース、「+」「%」「&」は使用不可）
- 再次輸入密碼 - パスワードの再入力
-
訂閱 - 受け取りたいメールの種類をチェック
- 每日氣象電子報 - 天気日報
- 天氣警、特報 - 気象警報
- 地震報告 - 地震情報
そして最後の「驗證碼」が確認コードです。
全て入力し終わったら「送出」を押しましょう。
パスワードを自動生成に頼っていて、パスワード欄をコピー&ペーストで入力した場合は認証に失敗する場合があります。
最後まで入力したのに「驗證碼失效」が出る場合は、パスワードを手入力してみてください。または、別のパスワードも試してみてください。
3. この段階では仮会員です。
入力したメールアドレス宛に確認メールが届くので、URLをクリックします。
これで本登録は完了です！
アクセスキーの発行
APIにアクセスするには、まずアクセスキーを入手する必要があります。
1. 上記のオープンデータ用HPにアクセスし、右上の「登入/註冊」をクリックします。
2. 「氣象會員登入」を選択し、メールアドレスとパスワードを入力、最後にCAPTCHAチェックしてログインします。
3. 自動でアクセスキーを入手するページに飛ぶので、「取得授權碼」をクリックして、アクセスキーを発行します。右に出てくる「CWA-」で始まる文字がアクセスキーです。
発行したアクセスキーは忘れずにメモしておきましょう。
もしアクセスキーを紛失した場合は、「更新授權碼」で再発行できます。
しかし発行済みのアクセスキーは無効となる上、7日間は再発行できなくなるので、注意しましょう。
APIにアクセスする
1. 上記リンクをクリックすると、地震津波情報のオープンデータ一覧が確認できます。
今回は震度3級以上の地震情報を取得したいので、「顯著有感地震報告」を選択します。
2. 「資料擷取API服務說明網址」項目の「API」をクリックします。
3. Swaggerの選択した項目にジャンプしました。右に「Try it out」をクリックします。
4. 各パラメータを入力します。一番上の「Authorization」が必須項目で、先ほど入手したアクセスキーをコピペしておきます。
5. 最後に「Execute」をクリックすると結果が返されます。あとは煮るなり焼くなり好きにしましょう。
LINEに通知する
LINE Notifyを使って通知します。
ソースコード
pip install requests
pip install googletrans==3.1.0a0
import requests
import time
import datetime
from googletrans import Translator
import re
import os
import urllib.request
def cst_to_jst(dt): # 台湾時間から日本時間への変換
cst = datetime.datetime.strptime(dt, "%Y-%m-%d %H:%M:%S")
jst = cst + datetime.timedelta(hours=1)
return jst
def translate(str_zh): # 翻訳
translator = Translator()
trans_jp = translator.translate(str_zh, dest="ja")
return trans_jp.text
def notifysend(message, img): # 通知系APIへのPOST
notify_url = "https://notify-api.line.me/api/notify"
if img:
with open(img, "rb") as f:
r = requests.post(notify_url,
headers={"Authorization": "Bearer " + token},
data={"message": message},
files={"imageFile": f.read()},
params={"notificationDisabled": True}
)
else:
r = requests.post(notify_url,
headers={"Authorization": "Bearer " + token},
data={"message": message},
params={"notificationDisabled": True}
)
return r
def notify_postlimit(p): # APIレート制限確認用
ratelimit = p.headers.get("X-RateLimit-Limit")
ratelimit_re = p.headers.get("X-RateLimit-Remaining")
imgratelimit = p.headers.get("X-RateLimit-ImageLimit")
imgratelimit_re = p.headers.get("X-RateLimit-ImageRemaining")
ratereset = p.headers.get("X-RateLimit-Reset")
reset_time = int(ratereset) - int(time.time())
m, s = divmod(reset_time, 60)
print(f"LINE Notify送信完了。 {datetime.datetime.now()}")
print(f"残りAPIコール回数：{ratelimit_re} / {ratelimit}")
print(f"残りImage Upload回数；{imgratelimit_re} / {imgratelimit}")
print(f"上限リセットまで、あと{m}m{s}s\n")
token = input("LINE Notifyのトークンを入力")
checkNum = 0
url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001?Authorization={Authkey}&limit=1&format=JSON&AreaName="
jsonData = requests.get(url).json()
eqNo1 = jsonData["records"]["Earthquake"][0]["EarthquakeNo"]
while True:
time.sleep(10)
jsonData = requests.get(url).json()
eqNo2 = jsonData["records"]["Earthquake"][0]["EarthquakeNo"]
if eqNo2 != eqNo1:
eqNo1 = eqNo2
checkNum += 1
print(f"EarthQuakeNoの変化を検知。メッセージを送信します。 日時：{datetime.datetime.now()} 回数：{checkNum}\n")
reportType = jsonData["records"]["Earthquake"][0]["ReportType"]
reportColor = jsonData["records"]["Earthquake"][0]["ReportColor"] # 程度の強さ
# 色の区分は仮のもの。エラーが出たら疑うべき
if reportColor == "綠色": # 下の基準に満たない
eqDeal = "" # 表現の度合い
expression = "。"
elif reportColor == "黃色": # M5.5かつ震度4級以上
eqDeal = "やや強い"
expression = "。"
elif reportColor == "橙色": # M6.0かつ震度5弱以上
eqDeal = "強い"
expression = "！"
elif reportColor == "紅色": # M6.5かつ震度6弱以上
eqDeal = "非常に強い"
expression = "！！"
else:
raise ValueError("想定外のreportColor")
riURI = jsonData["records"]["Earthquake"][0]["ReportImageURI"] # 震度地図
imgName = riURI.split("/")[-1]
os.makedirs("eqmap", exist_ok=True)
dl_path = f"./eqmap/{imgName}"
urllib.request.urlretrieve(riURI, dl_path)
eqInfo = jsonData["records"]["Earthquake"][0]["EarthquakeInfo"]
originTime = eqInfo["OriginTime"] # 発生時刻
otFormat = f"{cst_to_jst(originTime):%d日 %H時%M分頃、}" # フォーマット済み発生時刻（JST）
source = eqInfo["Source"] # 情報元
epiCenter = eqInfo["Epicenter"]["Location"] # 震源地
epiCenter = translate(epiCenter)
depth = str(eqInfo["FocalDepth"]) + "km" # 深さ
mType = eqInfo["EarthquakeMagnitude"]["MagnitudeType"] # マグニチュード
mValue = eqInfo["EarthquakeMagnitude"]["MagnitudeValue"]
if mType == "芮氏規模": # ローカルマグニチュード
mUnit = "ML"
else:
mUnit = "M"
shakingArea = jsonData["records"]["Earthquake"][0]["Intensity"]["ShakingArea"]
shakingArea = sorted(shakingArea, key=lambda a: a["AreaIntensity"], reverse=True)
maxInt = shakingArea[0]["AreaIntensity"]
skaList = []
for ska in shakingArea:
if re.findall("最大震度.*地區", ska["AreaDesc"]):
areaInt = ska["AreaIntensity"]
county = translate(ska["CountyName"])
res = f"【震度{areaInt}】\n{county}"
skaList.append(res)
mainHead = f"\n【🇹🇼CWA{reportType}】\n{otFormat}台湾で{eqDeal}地震があったピヨ{expression}"
mainBody = f"震源地：{epiCenter}\n震源の深さ：{depth}\n最大震度：{maxInt}\n推定マグニチュード：{mUnit}{mValue}"
mainFoot = f"ソース：{source}"
mainAns = f"{mainHead}\n\n{mainBody}\n\n{mainFoot}"
notifysend(mainAns, dl_path)
skaHead = "\n各地の震度は次の通りピヨ🐤"
skaBody = "\n".join(skaList)
skaAns = f"{skaHead}\n\n{skaBody}"
post = notifysend(skaAns, None)
notify_postlimit(post)
else:
checkNum += 1
print(f"更新完了。 更新日時：{datetime.datetime.now()} 回数：{checkNum}\n")
最後に
気象庁の防災情報もJSON化してほしい（切実）