気象庁の緊急地震速報を受け取ってDiscordに通知する

はじめに

緊急地震速報をDiscordに通知するBotはかなり見かけますが、いろんなサーバーで使われてるだけあって遅延がありますよね...
あ、自分で動かせばいいんじゃね？

ということで...

緊急地震速報を通知するものを作っていきます。
他サーバーに送信する予定が無いのであればWebhookで大丈夫そう。

処理の流れ

1. 緊急地震速報を配信するTCPサーバーから緊急地震速報を受信
2. Pythonで人間が読みやすいように処理
3. Discordに送信

プログラム

webhook_urlを変更してから動かして下さい。

import time
import socket
import requests
import math
import base64
import json
import gzip
import threading
import xml.etree.ElementTree as ET
from datetime import datetime, timezone

nolog_types = ["PING", "PONG", "RECV"]

webhook_url = "" # ここにDiscordのWebhookのURL

s = socket.socket()
s.connect(("eew.siyukatu.com",4175)) # 必要に応じて変更
s.settimeout(30)
running = True

sindo_colors = {
    "1": "f2f2ff",
    "2": "00aaff",
    "3": "0041ff",
    "4": "fae696",
    "5-": "ffe600",
    "5+": "ff9900",
    "6-": "ff2800",
    "6+": "a50021",
    "7": "b40068",
}

def sindo2text(data):
    retxt = ""
    if data["from"] == data["to"]:
        retxt = data["from"]
    else:
        retxt = data["from"]+"以上"+data["to"]+"以下"
    if data["to"] == "over":
        retxt = data["from"]+"以上"
    return retxt.replace("-","弱").replace("+","強")

def sindo2color(data):
    color = "f2f2ff"
    max = "0"
    if data["from"] == data["to"]:
        max = data["from"]
    else:
        max = data["to"]
    if data["to"] == "over":
        max = data["from"]
    if max in sindo_colors:
        color = sindo_colors[max]
    return color

def processor(data):
    dfields = []
    head = data["head"]
    xml = gzip.decompress(base64.b64decode(data["body"].encode())).decode()
    root = ET.fromstring(xml)
    title = root.find('.//{http://xml.kishou.go.jp/jmaxml1/}Title').text
    date_string = root.find('.//{http://xml.kishou.go.jp/jmaxml1/}DateTime').text
    serial = root.find('.//{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}Serial').text
    EventID = root.find('.//{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}EventID').text
    dfields.append({"name":"識別ID","value":EventID})
    dtitle = title+" 第"+serial+"報"
    embed_color = "ffff00"
    if head["type"] == "VXSE43" or head["type"] == "VXSE45":
        description = ""
        NextAdvisory = root.find('.//{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}NextAdvisory')
        if NextAdvisory != None and NextAdvisory.text == "この情報をもって、緊急地震速報：最終報とします。":
            dtitle += "(最終報)"
        Headline = root.find('.//{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}Headline/{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}Text')
        InfoType = root.find('.//{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}InfoType').text
        if InfoType == "発表":
            try:
                if Headline != None:
                    discdata += Headline.text
            except:pass
            Shingen = "不明"
            Area = root.find('.//{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Earthquake/{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Hypocenter/{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Area')
            try:
                Shingen = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Name').text+"("+Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Code').text+")"
                Coordinate = Area.find('{http://xml.kishou.go.jp/jmaxml1/elementBasis1/}Coordinate')
                if Coordinate != None:
                    Shingen += " "+Coordinate.attrib["description"]
                dfields.append({"name":"震源地","value":Shingen})
            except:pass
            ForecastInt = root.find('.//{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}ForecastInt')
            try:
                dfields.append({"name":"震度","value":sindo2text({'from':ForecastInt.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}From').text,'to':ForecastInt.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}To').text})})
                embed_color = sindo2color({'from':ForecastInt.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}From').text,'to':ForecastInt.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}To').text})
            except:pass
            try:
                dfields.append({"name":"気象庁コメント","value":root.find('.//{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Comments/{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}WarningComment/{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Text').text})
            except:pass
            Forecast = root.find('.//{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Forecast')
            fir = True
            shindo = ""
            if Forecast != None:
                for Pref in Forecast.findall('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Pref'):
                    if not fir:shindo += "----------------------------------------\n"
                    fir = False
                    prname = Pref.find("{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Name").text
                    for Area in Pref.findall('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Area'):
                        Category = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Category/{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Kind/{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Name').text
                        Name = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Name').text
                        Code = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Code').text
                        ArrivalTime = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}ArrivalTime')
                        Condition = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}Condition')
                        ForecastInt = Area.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}ForecastInt')
                        shindo += Category+"\n"
                        shindo += prname+" - "+Name+"("+Code+")\n"
                        shindo += "震度: "+sindo2text({'from':ForecastInt.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}From').text,'to':ForecastInt.find('{http://xml.kishou.go.jp/jmaxml1/body/seismology1/}To').text})+"\n"
                        if ArrivalTime != None:
                            print(ArrivalTime.text.replace("+09:00", "+00:00"))
                            shindo += "主要動到達時刻: "+datetime.fromisoformat(ArrivalTime.text).strftime("%Y/%m/%d %H:%M:%S")+"\n"
                        if Condition != None:
                            shindo += Condition.text+"\n"
            if shindo != "":
                dfields.append({"name":"各地の震度情報","value":shindo})
        if InfoType == "取消":
            embed_color = "ffff00"
            dfields.append({"name":"気象庁コメント","value":root.find('.//{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}Body/{http://xml.kishou.go.jp/jmaxml1/informationBasis1/}Text').text})
        requests.post(webhook_url, json={"embeds": [{"title":dtitle, "description": description, "color": int(embed_color, 16), "timestamp": date_string, "fields": dfields}]})


while running:
    rd = b""
    while True:
        r = s.recv(1)
        if not r:running = False;break
        rd += r
        if b"\n" in rd:break
    if not running:break
    data = rd.replace(b"\n",b"").decode("utf8").split(" ")
    if data[0] == "PING":
        s.send(("PONG "+data[1]).encode("utf8"))
        s.send(("PING "+str(time.time())).encode("utf8"))
    if data[0] == "PONG":
        try:print("ping: "+str(math.floor((time.time()-float(data[1]))*1000))+"ms")
        except:pass
    if data[0] == "RECV":
        try:
            data = json.loads(base64.b64decode(data[1].encode()).decode())
            threading.Thread(target=processor,args=(data,)).start()
        except Exception as e:print(e)
print("プログラムが終了しました。")

実際に動かしてみて

このサーバーの地震情報チャンネルで実際に動作させています。
(「チャンネル一覧」から表示にチェックを入れないと表示されません)

今のところ特に問題なく正常に動いています。

さいごに

XMLフォーマットのドキュメントなど、かなり勉強になりました。
次回は日本地図に震度の情報を当てはめて色を付けた画像を生成してみようと思います。

参考資料

• 気象庁防災情報XMLフォーマット