経緯

埼玉県熊谷市に住んでいます。2018/7に中型バイクの免許を取得して、休みの日は近所を乗り回していました。
しかし、北関東は突発的な夕立が多く、バイク乗りとしては、夕立に巻き込まれそうだったら一刻も早く帰宅したい。そんな理由から、今回のスクリプトを作りました。専門はサーバ・ネットワークなので、pythonは見よう見まねで書いています。ダメ出し、改善点などありましたらご指摘お願い致します。記事を書くにあたって見直してて気づいたのですが変数名がぐちゃぐちゃですね…。

夕立の判断基準

夏場の熊谷市では、群馬県で発生した雨雲が発達しながら南下することにより、夕立が発生します。経験則ですが、前橋市、高崎市、太田市辺りで降っていたら、熊谷でも高確率で夕立になります。秩父方面でも雨雲が発達しますが、秩父の雨雲は熊谷市には流れず、東京都へ流れることが多いです。以上のことから、熊谷市近辺及び群馬県南部で雨が降っていた場合、LINEに通知するbotを作りました。

ボツ案(各拠点のアメダスから降雨情報を判断する)
前橋・高崎・太田のアメダス情報から、降雨の有無が判断出来たのですが、1時間おきにしか更新されないため、急速に発達する夕立雲では間に合わないためボツになりました。こちらの場合はテキスト処理のみで済むのでbashで作成していました。
採用案(レーダー・ナウキャストの画像から降雨を判断する)
レーダー・ナウキャストは分単位で情報が更新されており、最新の降雨情報を把握出来るため、こちらを画像解析することにより、降雨の有無を判断することにしました。画像解析が必要なためpythonで書くことにしました。

仕組み

レーダー・ナウキャストから最新の降雨情報の画像を取得します。
指定された範囲で画像を切り出します(今回は群馬南部から熊谷にかけて)
切り出した画像の範囲内で、色を解析し、降雨のカラーチャートと同じ色がある場合は雨判定とします。
雨判定とされた場合はLINE APIを叩いてLINEのグループに通知をします。

サーバの準備

さくらVPSのサーバを使いました。CentOS7です。
crontabで夕立の可能性がある午後のみ、毎分動かしています。
毎分通知はさすがに鬱陶しいので、スクリプト内でフラグを立てて30分毎の通知としています。

$ crontab -l
*/1 12-18 * * * /usr/bin/python3 /home/arkey22/git/weather/weather.py

コード

# -*- coding: utf-8 -*-
# coding:utf-8

import cv2
import datetime
import time
import urllib
import numpy as np
import requests
import sys
import os
import logging

# 変数設定
BASE_URL = "https://www.jma.go.jp/jp/radnowc/imgs/radar/206/"
image_kantou = "/home/arkey22/git/weather/image_kantou.png"
image_saihoku = "/home/arkey22/git/weather/image_saihoku.png"
token = "########################################"
flag = "/tmp/weather_flag.tmp"
FLAG_INTERVAL_MINUTES = 30  # 分指定。
message = "雨近い！"

# ログ設定
log_path = "/var/log/weather/weather.log"
log_format = '%(asctime)s- %(levelname)s - %(message)s'
logging.basicConfig(filename=log_path, format=log_format, level=logging.INFO)

def compare_flag_time():
    """ flagの最終更新日時と現在時刻を比較する """
    # flagのタイムスタンプ取得
    flag_utime = os.path.getmtime(flag)

    # 現時刻取得
    now = time.time()

    # 前回時刻よりn分経過しているか
    return (now - flag_utime) > FLAG_INTERVAL_MINUTES * 60


def is_invoked_too_often():
    """ flagが前回時刻よりn分経過していた場合,flagの更新日時を今に書き換える """
    too_often = compare_flag_time()

    if too_often:
        os.utime(flag, None)
    return too_often

def get_url():
    """ 最新の気象データ画像のurlを作成 """
    today = datetime.datetime.today()
    rounded_minutes = today.minute // 5 * 5
    if rounded_minutes == 0:
        rounded_minutes = "00"

    # image_url = BASE_URL + re.sub(r'.$', '', date) + date_minutes + "-00.png"
    image_url = BASE_URL + today.strftime("%Y%m%d%H") + str(rounded_minutes) + "-00.png"
    return image_url


def save_image(url):
    """ 作成したURLから画像を保存 """
    urllib.request.urlretrieve(url, image_kantou)
    return image_kantou


def image_cut(image):
    """ 画像を指定の座標で切り出す """
    x = 190  # cols
    y = 130  # rows
    width = 45
    height = 45
    src = cv2.imread(image, 1)
    dst = src[y:y+height, x:x+width]
    cv2.imwrite(image_saihoku, dst)
    return image_saihoku


def compare_colour(org_image):
    """ 切り出した画像から雨の色を判断 """
    img = cv2.imread(image_saihoku, 1)
    # img = cv2.imread(image_blued, 1) # DEBUG用途
    b, g, r = cv2.split(img)

    # 気象レーダー規定の色を指定(上から時間辺りの雨量が1, 5, 10, 20, 30, 50, 80mm/h)
    mask = (b == 255) & (g == 210) & (r == 160) | \
           (b == 255) & (g == 140) & (r == 33) | \
           (b == 255) & (g == 65) & (r == 0) | \
           (b == 0) & (g == 245) & (r == 250) | \
           (b == 0) & (g == 153) & (r == 255) | \
           (b == 0) & (g == 40) & (r == 255) | \
           (b == 104) & (g == 0) & (r == 180)
    img_masked = cv2.bitwise_and(img, img, mask=mask.astype(np.uint8))
    return img_masked


def notify_line(img_masked, url):
    """ 切り取った画像の範囲で降雨が観測された場合はLINE APIを叩いて通知 """
    headers = {
        'Authorization': 'Bearer ' + token,
    }
    files = {
        'message': (None, message + " " + url),
    }
    api_status = requests.post('https://notify-api.line.me/api/notify', headers=headers, files=files)
    if api_status.status_code != requests.codes.ok:
        logging.error("Failed to call LINE API.")
        return False

if __name__ == '__main__':
    compare_flag_time()
    if is_invoked_too_often():
        logging.info("Continue script...")
    else:
        logging.debug("Script stopped due to reached the interval limit.")
        sys.exit(1)
    url = get_url()
    image = save_image(url)
    org_image = image_cut(image)
    img_masked = compare_colour(org_image)
    if img_masked.any():
        logging.info("It's about to rain!")
        notify_line(img_masked, url)
    else:
        logging.info("Not raining now.")

LINEの通知画面

反省点

画像の座標指定で雨判定をしているため、どこからどこまで指定しているか分かりづらい。
終日雨の場合は30分ごとに通知が来てしまう。朝から雨の場合は通知をしないという処理を入れてもいいかも
雨の強さで通知メッセージを変えたら、夕立の判断がしやすかったかもしれない。

家の近くで雨が降ったらLINEに通知するbotを作った

経緯

夕立の判断基準

仕組み

サーバの準備

コード

LINEの通知画面

反省点