雨が降りそうになったら自動でお知らせ!Yahoo! YOLP天気APIで作る降雨アラートシステム

Google Home

「雨が降るのに気づかず傘を持たずに外出してしまった」という経験はありませんか?
本記事では、Yahoo! YOLP(Yahoo! Open Local Platform)天気APIを使って自宅周辺の降雨データをリアルタイムに取得し、雨が降りそうになった際に自動でGoogle Homeが音声でお知らせするシステムの実装方法を解説します。
10分先の降水予報まで取得できるYOLPならではの精度で、「もうすぐ雨が降りそうです」「雨が降り始めました」をタイムリーに通知できます。

なぜこの機能を作ったのか?

我が家は4人家族で、子供が学校に行く朝の時間帯に「今日は傘いる?」という質問が飛び交います。
スマホで天気アプリを確認すればわかるものの、毎朝それをやるのが手間ですし、子供にはスマホを渡したくない場面もあります。
そこで、自宅のリビングに置いてあるGoogle Homeが雨の予報を自動で音声でお知らせしてくれる仕組みを作りました。特に「もうすぐ降るよ」という事前通知は傘の持ち忘れを防ぐのに非常に役立っています。

家族が意識しなくても、雨のタイミングをちゃんと教えてほしい!

実現したいこと

  • 自宅周辺の降水情報をリアルタイムで取得する
  • 雨が降る前(10〜20分前)にGoogle Homeで音声通知する
  • 雨が降り始めたタイミングでもGoogle Homeで音声通知する
  • 一度通知したら2時間は再通知しない(通知が連続して煩わしくならないようにする)
  • Raspberry Piのサービスとして常時起動し、自動的に監視し続ける

この記事でわかること

  • Yahoo! YOLP天気APIの使い方と降水データの取得方法
  • PythonでYOLPのAPIレスポンスを解析して降雨を判定する方法
  • Google Homeで音声通知を送る方法(別記事の自作ライブラリを使用)
  • Raspberry PiでPythonプログラムをsystemdサービスとして常時起動する方法

必要な準備と用意するもの

ハードウェア
  • Raspberry Pi(例:Raspberry Pi 5)
  • Google Home(Nest)
ソフトウェア/サービス
  • Python
    • requests
  • Yahoo! YOLP(Yahoo! Open Local Platform)天気API
    • Yahoo! JAPAN Developer Network でアプリケーションを登録してAPIキーを取得する必要あり
  • Google Home通知ライブラリ(自作 googlehome.py)
    • 別記事「Google Homeに任意のフレーズを発話させる方法」を参照

Google HomeへのPython経由の音声通知の実装については、別記事「Google Homeに任意のフレーズを発話させる方法」を参照してください。本記事ではgooglehome.pyという自作ライブラリが用意されている前提で解説します。

完成イメージ

完成すると、次のような状況でGoogle Homeが自動で音声通知を行います。

  • 雨が10〜20分以内に降り始める予報のとき:「もうすぐ雨が降りそうです。」
  • 雨が降り始めたとき:「雨が降り始めました。」

プログラムはRaspberry PiのsystemdサービスとしてOS起動と同時に自動起動し、1分ごとに降水状況を確認し続けます。
一度通知を出した後は2時間は再通知しないため、同じ雨に対して何度も通知が鳴ってうるさくなることはありません。

システムの仕組み

全体の流れは下記の通りです。

  1. Yahoo! YOLP天気API:指定した緯度経度の降水強度(mm/h)を10分間隔で3時点分(10分前実績・現在・10分後予報)取得できる
  2. Raspberry Pi(rain_forecast_alert.py):1分ごとにYOLPのAPIを叩いて降水データを取得し、降雨の開始を検知したらGoogle Homeに通知
  3. Google Home(リビング):Raspberry Piからの指示を受けて音声で通知

実装のポイント

Yahoo! YOLP 天気APIについて

Yahoo! YOLP(Yahoo! Open Local Platform)が提供する天気APIは、指定した地点の降水強度(mm/h)を10分単位で取得できます。APIレスポンスには下記の3時点のデータが含まれています。

  • weather_data[0]:10分前の実績降水強度(過去の実測値)
  • weather_data[1]:現在の降水強度(実質的な現在値)
  • weather_data[2]:10分後の降水予報

APIのレスポンスはデータ更新タイミングの関係で、weather_data[0]が「10分前の実績」、weather_data[1]が「現在」として扱われます。実装時にはAPIドキュメントで最新の仕様を確認してください。

降雨判定のロジック

降水強度(Rainfall)が 1.0 mm/h 以上を「雨が降っている」として判定します。

もうすぐ雨が降りそうです(事前通知)
現在または直近は雨が降っていないが、10〜20分後に降水予報がある場合に通知
(rain_fall_now < 1 or rain_fall_after_10 < 1) and (rain_fall_after_10 >= 1 or rain_fall_after_20 >= 1)

雨が降り始めました(開始通知)
直前のチェックでは雨が降っていなかったが、今回のチェックで降り始めた場合に通知
rain_fall_latest < 1 and (rain_fall_now > 1 or rain_fall_after_10 > 1)

通知クールダウンの仕組み

一度通知を出した後、restart_datetimeという変数に「現在時刻 + 2時間」を設定します。次回の通知判定ではrestart_datetimeを経過していない場合はAPI呼び出し自体をスキップします。
これにより、同じ雨に対して何度もGoogole Homeが鳴ることを防ぎます。

無限ループ + systemdサービスで常時監視

プログラムはwhile Trueの無限ループで動き続け、1分ごとに降水データを取得します。プロセスが何らかの理由で落ちても、systemdのRestart=always設定により60秒後に自動で再起動されます。

事前準備

Yahoo! YOLP APIキーの取得

  1. Yahoo! JAPAN IDを取得・ログインする
  2. Yahoo! JAPAN Developer Network(developer.yahoo.co.jp)にアクセスする
  3. 「アプリケーション管理」→「新しいアプリケーションを開発する」からアプリを登録する
  4. 「Client ID(アプリケーションID)」をメモしておく(これがAPIキーになる)

APIキーは外部に公開しないよう注意してください。settings.jsonに保存し、プログラム内にハードコードしないようにしましょう。

自宅の緯度経度を調べる

Yahoo! YOLP天気APIは緯度・経度で地点を指定します。Googleマップで自宅周辺の座標を確認してメモしておきます。
フォーマットは「経度,緯度」の順(例:139.650740,35.484642)です。

settings.jsonの設定

{
    "weather": {
        "base_url": "https://map.yahooapis.jp/weather/V1/place",
        "url_parameter": {
            "appid": "xxxxxxxxxxxxxxxxxxxxxxxxxxx",
            "Coordinates": "139.xxxxxx,35.xxxxxx",
            "interval": "10",
            "json": "json"
        }
    }
}
  • appid:Developer Networkで取得したAPIキー
  • Coordinates:自宅周辺の「経度,緯度」
  • interval:取得する時間間隔(分)。10で10分間隔の降水データを取得
  • json:出力形式。json固定

実装方法

降雨アラートプログラム (rain_forecast_alert.py)

# -*- coding: utf-8 -*-
__group__       = '情報通知'
__description__ = '自宅周辺で降雨がある際に通知'
from pathlib import Path
import sys
PROJECT_ROOT = Path(__file__).resolve()
while PROJECT_ROOT.name != "smarthome":
    if PROJECT_ROOT.parent == PROJECT_ROOT:
        raise RuntimeError("smarthome project root not found")
    PROJECT_ROOT = PROJECT_ROOT.parent
sys.path.insert(0, str(PROJECT_ROOT))
from libs.bootstrap import Bootstrap
bootstrap = Bootstrap(__file__)
settings = bootstrap.settings
# ------------------------------------------------------------
import requests
from influx.v2 import client as influxclient
from influx.v2 import writer as influxwriter
import googlehome
import time
from datetime import datetime, timedelta
import smarthomelog

try:
    BASE_URL    = settings['weather']['base_url']
    APP_ID      = settings['weather']['url_parameter']['appid']
    COORDINATES = settings['weather']['url_parameter']['Coordinates']
    INTERVAL    = settings['weather']['url_parameter']['interval']
    OUTPUT      = settings['weather']['url_parameter']['json']

    influx_client = influxclient
    influx        = influxwriter.Writer()
    success_log   = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
    log           = smarthomelog.Log(__file__, smarthomelog.LogType.YLOP)

    # Google Homeの初期設定(リビング)
    google_notify = googlehome.Notify(googlehome.GoogleHome.LIVING)

    rain_fall_latest = 9999  # 直前の降水強度(初期値は大きい値で「雨」扱いを回避)
    ret_value        = 0     # 0:未通知 / 1:通知済み

    # 前回通知してから2時間は再通知しない(初期値は現在時刻)
    restart_datetime = datetime.now().strftime('%Y/%m/%d %H:%M:%S')

    def weather_alert():
        global rain_fall_latest
        global ret_value

        # YOLP天気APIのURLを組み立て
        yolp_url = BASE_URL + '?appid=' + APP_ID + '&coordinates=' + COORDINATES \
                             + '&interval=' + INTERVAL + '&output=' + OUTPUT

        # 天気情報をJSON形式で取得
        weather_json = requests.get(yolp_url).json()

        # APIのログ記録(APIキーはダミーに置き換えて記録)
        yolp_url_log = BASE_URL + '?appid=dummy&coordinates=' + COORDINATES \
                                + '&interval=' + INTERVAL + '&output=' + OUTPUT
        log.write(yolp_url_log)
        influx.insert_execution_api_log(__file__, influxwriter.LogApiType.YLOP, COORDINATES, BASE_URL, None, None, None)

        # 降水データを取得
        weather_data      = weather_json['Feature'][0]['Property']['WeatherList']['Weather']
        rain_fall_now     = weather_data[0]['Rainfall']   # 10分前の実績
        rain_fall_after_10 = weather_data[1]['Rainfall']  # 現在
        rain_fall_after_20 = weather_data[2]['Rainfall']  # 10分後の予報

        # InfluxDBに取得データを記録
        current_time = datetime.now().strftime('%Y/%m/%d %H:%M:%S')
        influx.insert_execution_log(__file__,
            '現在時刻=' + current_time
            + ' | ' + weather_data[0]['Date'][8:10] + ':' + weather_data[0]['Date'][10:12] + 'の降水実績=' + str(weather_data[0]['Rainfall'])
            + ' | ' + weather_data[1]['Date'][8:10] + ':' + weather_data[1]['Date'][10:12] + 'の降水予想=' + str(weather_data[1]['Rainfall'])
            + ' | ' + weather_data[2]['Date'][8:10] + ':' + weather_data[2]['Date'][10:12] + 'の降水予想=' + str(weather_data[2]['Rainfall']))

        # 事前通知:現在は雨が降っていないが、10〜20分後に降雨予報がある
        if (rain_fall_now < 1 or rain_fall_after_10 < 1) and (rain_fall_after_10 >= 1 or rain_fall_after_20 >= 1):
            ret_value = 1
            google_notify.speak('もうすぐ雨が降りそうです。')

        # 開始通知:前回チェック時は雨が降っていなかったが、今回は降り始めた
        elif rain_fall_latest < 1 and (rain_fall_now > 1 or rain_fall_after_10 > 1):
            ret_value = 1
            google_notify.speak('雨が降り始めました。')

        # 直前の降水強度を更新
        rain_fall_latest = rain_fall_now

    # メインループ(1分ごとに降水状況を確認し続ける)
    while True:
        current_time = datetime.now().strftime('%Y/%m/%d %H:%M:%S')
        if restart_datetime <= current_time:
            if ret_value == 0:
                weather_alert()

            if ret_value == 1:
                # 通知済みの場合は2時間後までスキップ
                restart_datetime = (datetime.now() + timedelta(hours=2)).strftime('%Y/%m/%d %H:%M:%S')
                ret_value = 0
                time.sleep(7200)
        else:
            influx.insert_execution_log(__file__, '現在時刻=' + current_time + ' | リスタート時刻=' + restart_datetime)

        success_log.write('処理が正常に終了しました。', __file__)
        time.sleep(60)  # 1分待機

except Exception as e:
    log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
    log.write_error(sys.exc_info())

systemdサービスファイルの作成

プログラムをRaspberry PiのsystemdサービスとしてOS起動時に自動で開始し、常時稼働させます。

[Unit]
Description=Weather Alert
After=nginx.service influxdb.service webapi.service
Wants=nginx.service influxdb.service webapi.service

[Service]
ExecStart=/home/<username>/Projects/venv/bin/python /home/<username>/Projects/smarthome/core/alert/rain_forecast/rain_forecast_alert.py
Restart=always
RestartSec=60
Type=simple
User=<username>

[Install]
WantedBy=multi-user.target

サービスファイルを/etc/systemd/system/に配置して有効化します。

# サービスファイルをコピー
sudo cp smh-rain-forecast-alert.service /etc/systemd/system/

# systemdに読み込ませる
sudo systemctl daemon-reload

# OS起動時の自動起動を有効化
sudo systemctl enable smh-rain-forecast-alert.service

# サービスを開始
sudo systemctl start smh-rain-forecast-alert.service

コードの解説

YOLP APIのリクエストURLの組み立て

APIのリクエストURLはクエリパラメータを連結して組み立てます。

yolp_url = BASE_URL + '?appid=' + APP_ID + '&coordinates=' + COORDINATES \
                     + '&interval=' + INTERVAL + '&output=' + OUTPUT

組み立てたURLの例(APIキーはダミー):

https://map.yahooapis.jp/weather/V1/place?appid=dummy&coordinates=139.xxxxxx,35.xxxxxx&interval=10&output=json

APIレスポンスの構造と降水データの取得

YOLPのレスポンスJSONから降水データを取り出す箇所です。

weather_data       = weather_json['Feature'][0]['Property']['WeatherList']['Weather']
rain_fall_now      = weather_data[0]['Rainfall']   # 10分前の実績降水強度
rain_fall_after_10 = weather_data[1]['Rainfall']   # 現在の降水強度
rain_fall_after_20 = weather_data[2]['Rainfall']   # 10分後の降水予報

YOLPのレスポンスJSONは次のような構造になっています。

{
  "Feature": [
    {
      "Property": {
        "WeatherList": {
          "Weather": [
            { "Date": "202605261200", "Rainfall": 0.0, "Type": "observation" },
            { "Date": "202605261210", "Rainfall": 0.0, "Type": "observation" },
            { "Date": "202605261220", "Rainfall": 2.5, "Type": "forecast" }
          ]
        }
      }
    }
  ]
}

通知クールダウンの仕組み

通知後にrestart_datetimeを2時間後に設定することで、同じ雨に対して何度も通知が鳴るのを防ぎます。

if ret_value == 1:
    restart_datetime = (datetime.now() + timedelta(hours=2)).strftime('%Y/%m/%d %H:%M:%S')
    ret_value = 0
    time.sleep(7200)  # 2時間待機

restart_datetimeは文字列として保持しており、current_time(現在時刻の文字列)と比較しています。同じフォーマット(%Y/%m/%d %H:%M:%S)なので文字列の辞書順比較で時刻前後が正しく判定できます。

systemdサービスの設定ポイント

  • Restart=always:プロセスが何らかの理由で終了した場合に自動で再起動する
  • RestartSec=60:再起動まで60秒待機する(API連続アクセスを防ぐため)
  • After=influxdb.service:InfluxDBが起動してから本サービスを起動する

動作確認

まずはコマンドラインから手動で実行して動作を確認します。

python /home/<username>/Projects/smarthome/core/alert/rain_forecast/rain_forecast_alert.py

プログラムが起動したら、ターミナルにAPIの取得ログが1分ごとに出力されることを確認します。
実際に雨が降り始めるタイミングでGoogle Homeが発話することを確認できれば完成です。

systemdサービスとして起動している場合は、下記コマンドで状態を確認できます。

# サービスの状態確認
sudo systemctl status smh-rain-forecast-alert.service

# ログの確認
journalctl -u smh-rain-forecast-alert.service -f

Google Homeへの通知が来ない場合は下記を確認してください。
・settings.jsonのAPIキーと座標が正しいか
・Yahoo! YOLP APIのリクエスト制限に達していないか(無料枠:1日50,000リクエスト)
・Google Homeがネットワークに接続されているか
・google-home-notifier(Node.js)がインストールされているか

まとめ

今回はYahoo! YOLP天気APIを使って自宅周辺の降水データをリアルタイム取得し、雨の前後をGoogle Homeで自動通知する仕組みを実装しました。
実装のポイントをまとめると下記の通りです。

  • YOLP天気APIは緯度経度を指定するだけで10分単位の降水強度(mm/h)を取得できる
  • 降水強度 1.0 mm/h を閾値として「雨が降っている」と判定する
  • 「前回雨なし → 今回雨あり」の変化を検知することで、降り始めのタイミングを正確に通知できる
  • 通知後2時間のクールダウンを設けることで、過剰通知を防いでいる
  • systemdサービスにすることでOS起動から常時監視が実現でき、障害時の自動復旧も対応できる

同じYOLPのAPIを応用して、強風・大雪・猛暑などの気象条件に応じた通知に拡張することもできます。ぜひ自分のスマートホームに合わせてカスタマイズしてみてください。

タイトルとURLをコピーしました