Go to Qiita Advent Calendar 2024 Top
LoginSignup
3
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@compo031

自作の防災情報Slack通知botが便利な話

Posted at

この記事は 防災アプリ開発 Advent Calendar 2024 の1日目です。

好きなおにぎりの具は、鮭いくらです。美味いものと美味いものが両方あったらそりゃ美味い。

はじめに

防災アプリを開発していると、自分の居住地域に関係なく、全国での大雨や地震津波情報の発表状況を知りたくなります。
著名な防災アプリでも全国向け通知があるにはありますが、カスタマイズ性が少ないことも多く、かゆいところに手が届かない…！ということがありました。

なければ作れば良いじゃない精神で、自分が欲しい情報を自分向けに通知してもらいます。

設計

プラットフォーム

データ元

データ元には DMDATA.JP を利用しました。

dmdataは気象庁発信の防災情報をWebSocketで受け取れる有償サービスで、筆者が現時点で利用中であることから、そのまま使います。

処理

Pythonでのこのこ処理します。

Python Scriptを作成したら、それを さくらのVPS に配置して稼働させることで、自宅PCでの稼働に比べて可用性を担保します。

通知先

今回はSlackを利用します。

たまたまAPIの扱いに慣れているのがSlackだったというもので、Slackにこだわりはなく、Discordでも何でも同じことができると思います。

通知対象

今回botを運用するにあたって重視した点は、「重要度の高い情報に絞る」でした。
何でもかんでも通知されると通知を気にしなくなってしまうので情報を絞ることで、通知されたということは何かヤバいことが起きているんだな、という状況を作り出します。

執筆時点では、以下の情報に絞っています。

ジャンル 情報名 備考
気象 特別警報報知
気象 顕著な大雨に関する全般気象情報
気象 氾濫発生情報
気象 大雨危険度通知 災害切迫（警戒レベル5相当）のみ
地震 緊急地震速報（予報） 震度5弱以上またはM6.5以上のみ
地震 震度速報 震度5弱以上のみ

津波、火山は将来的に実装を考えていますが、現時点で未実装です。
また、記録的短時間大雨情報は10月まで通知対象でしたが、あまりにも数が多いことから「通知を気にしなくなってしまう」現象が起こり、現在は対象除外としています。

実装

DMDATA.JP契約

以下の区分を契約します。

  • 地震・津波関連
  • 気象警報・注意報関連
  • 定時報・その他関連
  • 緊急地震（予報）

image.png

Pythonコード

dmdataのwebsocketに接続し、電文コード（例えば震度速報だったら VXSE51）別に処理を振り分けています。

XML電文を解析する際は、xmltodictライブラリを利用しています。その名のとおり、xmlをdictの感覚で扱えて便利です。
ただ、dmdataの機能でXMLをjsonに変換して渡してくれる機能があるので、それを利用すると同じくらい簡単になるかもしれません（自分は過去のXMl電文で色々テスト稼働させることが多いので、XMLを扱えたほうが便利で、XMLのまま受信しています）。

あまりにも自分用の雑コードですが、せっかくQiitaなので貼っておきます。

import base64
import gzip
import json
import re
import time
import unicodedata
from enum import Enum

import iso6709
import numpy as np
import requests
import websocket
import xmltodict

from config import *

eew_eventid_set = set()
vxse51_dict = {}
special_warning_prefs = set() # 特別警報発表中の府県


class ChannelId(Enum):
    EARTHQUAKE = CHANNEL_ID_OF_EARTHQUAKE
    WEATHER = CHANNEL_ID_OF_WEATHER


def main():
    run_websocket()


def run_websocket():
    data = {"classifications": ["telegram.earthquake", "telegram.weather", "telegram.scheduled", "eew.forecast"]}
    res = requests.post(
        url=f"https://api.dmdata.jp/v2/socket",
        data=json.dumps(data),
        auth=(DMDATA_API_KEY, "")
    )

    url = json.loads(res.text)["websocket"]["url"]
    ws = websocket.WebSocketApp(
        url=url,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close)
    ws.run_forever()


def on_message(ws: websocket.WebSocketApp, message: str):
    print("message: " + message, flush=True)

    try:
        message_json = json.loads(message)
        message_type = message_json["type"]

        if message_type == "ping":
            ping_id = message_json["pingId"]
            pong = json.dumps(
                {
                    "type": "pong",
                    "pingId": ping_id
                }
            )
            ws.send(pong)
        elif message_type == "data":
            compress_data(message_json)
    except:
        print("[invalid json]", message, flush=True)


def on_error(ws: websocket.WebSocketApp, error):
    print(f"error: {error}", flush=True)
    ws.close()


def on_close(ws: websocket.WebSocketApp, close_status_code, close_msg):
    time.sleep(10)

    run_websocket()


def compress_data(message_json: dict):
    type_code = message_json["head"]["type"]

    if type_code[:4] == "VXKO":
        # 指定河川洪水予報

        if "氾濫発生情報" in message_json["xmlReport"]["head"]["title"]:
            headline = str(message_json["xmlReport"]["head"]["headline"])
            headline = headline.replace("【警戒レベル５相当情報［洪水］】", "")
            headline = headline.replace("では、", "で、")
            headline = headline.replace("（", "").replace("）", "")
            headline = headline.strip()

            editional_office = message_json["xmlReport"]["control"]["editorialOffice"]

            post_message(ChannelId.WEATHER, "氾濫発生情報", f"{headline}（{editional_office}）")
    elif type_code == "VPZJ50":
        # 全般気象情報

        if message_json["xmlReport"]["head"]["title"] == "顕著な大雨に関する全般気象情報":
            headline = str(message_json["xmlReport"]["head"]["headline"])
            headline = headline.replace(
                "が同じ場所で降り続いています。命に危険が及ぶ土砂災害や洪水による災害発生の危険度が急激に高まっています。", "")
            headline = headline.replace("では、", "で、")

            post_message(ChannelId.WEATHER, "顕著な大雨に関する全般気象情報", headline)
    elif type_code == "VPNO50":
        # 特別警報報知

        headline = str(message_json["xmlReport"]["head"]["headline"])
        if "特別警報を発表しました" in headline:
            m = re.match(
                r"【特別警報（(?P<warnings>.+?)）】(?P<pref>.+?)に特別警報を発表しました。", headline)
            if m:
                warning = m.group("warnings")
                pref = m.group("pref")

                if pref in special_warning_prefs:
                    # 既に発表中の場合は処理しない
                    return

                special_warning_prefs.add(pref)
                post_message(ChannelId.WEATHER, "気象特別警報", f"{pref}に{warning.replace('、', '・')}特別警報")

        elif "警報に切り替えました" in headline:
            m = re.match(r"【警報に切り替え】(?P<pref>.+?)の特別警報を警報に切り替えました。", headline)
            if m:
                pref = m.group("pref")
                if pref not in special_warning_prefs:
                    # 発表中でない場合は処理しない
                    return

                special_warning_prefs.remove(pref)
                post_message(ChannelId.WEATHER, "気象特別警報", f"{pref}の特別警報を警報に切り替え")

    elif type_code == "VPRN50":
        # 大雨危険度通知

        xml = body2xmldict(message_json["body"])
        pref_info = xml["Report"]["Body"]["MeteorologicalInfos"][0]
        level_info = pref_info["MeteorologicalInfo"][1]

        texts = []
        items = level_info["Item"]
        for item in items:
            kinds = []
            sigs = item["Kind"]["Property"]["SignificancyPart"]["Base"]["Significancy"]
            for idx, sig in enumerate(sigs):
                code = sig["Code"]
                condition = sig["Condition"]

                # レベル5相当（コード52）かつ新規出現時
                if code == "52" and condition == "上昇":
                    kinds.append(["土砂災害危険度", "浸水危険度", "洪水危険度"][idx])

            # 対象があれば
            if kinds:
                area_name = item["Area"]["Name"]
                texts.append(f"{area_name} {'、'.join(kinds)}でレベル5相当の格子が出現")

        # 対象があれば
        if texts:
            post_message(ChannelId.WEATHER, "大雨危険度通知", "\n".join(texts))

    elif type_code == "VXSE45":
        # 緊急地震速報（地震動予報）

        xml = body2xmldict(message_json["body"])
        xmlhead = xml["Report"]["Head"]
        xmlbody = xml["Report"]["Body"]

        # キャンセル報
        if "Text" in xmlbody:
            text = xmlbody["Text"]
            if "取り消し" in text:
                # キャンセル報は処理しない
                return

        # イベントID
        event_id = xmlhead["EventID"]

        # 震央地名
        epiname = xmlbody["Earthquake"]["Hypocenter"]["Area"]["Name"]

        # 深さ
        coord = xmlbody["Earthquake"]["Hypocenter"]["Area"]["jmx_eb:Coordinate"]["#text"]
        depth = iso6709.Location(coord).alt / -1000

        # マグニチュード
        magnitude = float(xmlbody["Earthquake"]["jmx_eb:Magnitude"]["#text"])

        # 震度
        if "Intensity" in xmlbody:
            maxint = xmlbody["Intensity"]["Forecast"]["ForecastInt"]["From"]
            maxint = str(maxint).replace("-", "弱").replace("+", "強")
        else:
            maxint = "未推定"

        # 最終報
        is_final = "NextAdvisory" in xmlbody

        # 整形
        text = f"{epiname}で地震 深さ{depth}km M{'未推定' if np.isnan(magnitude) else magnitude} 予想最大震度{maxint}"

        if event_id in eew_eventid_set:
            if is_final:
                post_message(ChannelId.EARTHQUAKE, "緊急地震速報（予報最終）", text)
        else:
            if maxint in ["5弱", "5強", "6弱", "6強", "7"] or magnitude >= 6.5:
                eew_eventid_set.add(event_id)
                post_message(ChannelId.EARTHQUAKE, "緊急地震速報（予報）", text)

    elif type_code == "VXSE51":
        # 震度速報

        xml = body2xmldict(message_json["body"], ("Item", "Area"))
        xmlhead = xml["Report"]["Head"]

        # イベントID
        event_id = xmlhead["EventID"]
        print(event_id)

        print(xmlhead["Headline"]["Information"]["Item"][0]["Areas"]["Area"])

        # 最大震度と地域名
        maxint = xmlhead["Headline"]["Information"]["Item"][0]["Kind"]["Name"]
        maxint = unicodedata.normalize('NFKC', maxint)
        areas = [x["Name"] for x in xmlhead["Headline"]["Information"]["Item"][0]["Areas"]["Area"]]

        # 整形
        text = f"{maxint} {' '.join(areas)}"

        # 投稿
        if event_id in vxse51_dict:
            if vxse51_dict[event_id] != text:
                vxse51_dict[event_id] = text
                post_message(ChannelId.EARTHQUAKE, "震度速報", text)
            pass
        else:
            if maxint in ["震度5弱", "震度5強", "震度6弱", "震度6強", "震度7"]:
                vxse51_dict[event_id] = text
                post_message(ChannelId.EARTHQUAKE, "震度速報", text)


def post_message(channel_id: ChannelId, title: str, text: str, mention: bool = True):
    # メンションを付ける
    if mention:
        text += " \n<!channel>"

    # Slackに投稿
    data = {
        "token": SLACK_BOT_TOKEN,
        "channel": channel_id.value,
        "username": title,
        "text": text
    }
    requests.post(url="https://slack.com/api/chat.postMessage", data=data)


def body2xmldict(body: str, force_list: tuple = ()):
    # base64デコード
    decoded = base64.b64decode(body)

    # zip解凍
    decompressed = gzip.decompress(decoded)

    # byte列からUTF-8文字列にエンコード
    encoded = decompressed.decode('utf-8')

    # xmldictにパース
    parsed = xmltodict.parse(encoded, force_list=force_list)

    if parsed:
        return parsed
    else:
        return {}


if __name__ == "__main__":
    main()

また、configファイルにAPIキーなどを置いています。

DMDATA_API_KEY           = "(APIキー)"
SLACK_BOT_TOKEN          = "xoxb-なんとか"
CHANNEL_ID_OF_EARTHQUAKE = "Cなんとか"
CHANNEL_ID_OF_WEATHER    = "Cなんとか"

稼働状況

実装自体は2023年末に行い、2023年12月31日に稼働を開始しました。
2024年1月1日の能登半島地震でさっそく稼働とはいかず、チャンネルにSlackbotを招待し忘れたため通知はされず、実質1月2日から稼働が始まっています。

Slackの有料プランには入っていないため、90日前以降のログは見られず、直近のログのスクショをぺたぺたと貼ります。

気象

2024年9月21日の石川県能登地方の大雨事例です。

大雨危険度通知（「レベル5相当の格子が出現」）、顕著な大雨に関する全般気象情報（「線状降水帯による非常に激しい雨」）、記録的短時間大雨情報、特別警報報知と、重要度の高い情報が通知されています。
image.png

この際に通知量が多かったのもあり、記録的短時間大雨情報の投稿は現在ストップしています。

地震

9月以降のログです。
ときどきあるM6.5以上の緊急地震速報が流れたり、先日の石川県能登地方沖の緊急地震速報が投稿されてたりしています。
この、1ヶ月に1回以下程度の通知量が、ちょうど良いのではないかと思っています。

image.png

さいごに

昨年末に実装し、今年はじめに稼働を開始した防災情報Slack通知botの話でした。

ここぞという時に通知がくるため、自分で作って自分で助かっています。
image.png

3
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?