ラズパイ:Yahoo!天気・災害の警報や注意報を取得してGメールで通知する

スポンサーリンク
Raspberry Pi

Yahoo! 天気・災害のサイトから指定した地域の警報や注意報を取得して自分のGmailアドレスから通知するPythonスクリプト
ラズパイのcrontabに設定して5分おきにチェックすれば十分実用的なアプリとなる。

前回の記事で情報を取得するスクレイピングの部分の説明はしているので参考にしてください。

事前に準備しておく項目

Gmailのセキュリティ設定の変更

Googleアカウントの設定の変更が必要になる。
以下を参考に設定しておく。

スクレイピング用にPythonのbeautifulsoup4モジュールをインストール

ラズパイのターミナルから以下を実行。

pip3 install bs4

ソースコード

27,29~31行のハイライト部分は各自のPythonスクリプトのディレクトリ、Gメールアドレス、Gメールパスワード、送信先メールアドレスに置き換えてください。
32行目と57行目の都道府県取得したい市町村の設定は、前回記事(ラズパイ:Yahoo!災害情報から指定した地域の警報や注意報を取得してテキスト化する)の部分を参考にしてください。

sendSaigaiMail.py

#!/usr/bin python3
# -*- coding: utf-8 -*-

"""
	ラズパイ(Raspberry Pi)でYahoo!天気・災害サイトから取得した情報をGmailで通知する
		2021-08-16
		
		python 3.5.3
		OS: Raspbian 9.13 stretch(Raspberry Pi2)にて確認
"""
# スクレイピング関連
import urllib.request
from bs4 import BeautifulSoup
import os, sys

# メール送信のためのSMTPセッションクライアントを作成するため
import smtplib

# メールの送信日付をインターネットで送信できるフォーマットで指定するため
from email.utils import formatdate

# メール本文に日本語を送信できるようにするためのMIME規格に変換するため
from email.mime.text import MIMEText

# 定数関連
keihoFlg = False							# 警報・注意報が出ているか? true: 出ている false: 出ていない
DATA_PATH = "/home/pi/"						# Pythonプログラムとデータファイルのディレクトリ
DATA_FILE = "sendSaigaiMail.txt"			# 送信履歴データファイル
GMAIL_ADDR = "(メールアドレス)"		# Gメールアドレス(メール送信元となる)
GMAIL_PASS = "(パスワード)"				# Gメールのパスワード
TO_ADDR = "(送信先メールアドレス)"	# 送信先メールアドレス
url = "https://typhoon.yahoo.co.jp/weather/jp/warn/8/"		# 例)茨城県
														# Yahoo!の災害トップ > 警報・注意報のサイト > 地域のURL

# 災害情報のサイトにアクセスしHTMLタグを取得
data = urllib.request.urlopen(url)

# HTMLを解析して取得
soup = BeautifulSoup(data, 'html.parser')

# 表示用文字列
texts = ""

# 地域タイトル取得
title = soup.find("title").text
texts += "{}\n\n".format(title)

# classが 'warnArea_table' のタグを取得(表全体のデータとなる)
saigaiTable = soup.find(class_="warnArea_table")

# テーブルの1行ずつを順に処理する
for tr in saigaiTable.find_all("tr"):
	# 市町村名を取得
	city = tr.find("a").text		# aタグに市町村名が入っている
	
	# 指定した市町村名の情報のみ取得する → 【 and city != "(市町村名)"】 で取得したい市町村名を追加
	if city != "水戸市" and city != "日立市":
		continue
	else:
		texts += "{}\n".format(city);	# 市町村名を文字列に追加

	# 特別警報、警報、注意報を取得(全てリストとして取得される)
	emgWarnings = tr.find_all(class_="icoEmgWarning")	# 特別警報
	warnings = tr.find_all(class_="icoWarning")				# 警報
	advisorys = tr.find_all(class_="icoAdvisory")			# 注意報

	# 特別警報・警報・注意報が全て出ていない場合
	if len(emgWarnings) == 0 and len(warnings) == 0 and len(advisorys) == 0:
		texts += "\t警報・注意報はありません\n"

	# どれか一つでも出ていた場合
	else:
		keihoFlg = True
		# 特別警報・警報・注意報ごとに文字列保存
		if len(emgWarnings) != 0:		# 特別警報
			texts += "\t"
			for emgWarning in emgWarnings:
				texts += "{}\t".format(emgWarning.text)
			texts += "\n"
	
		if len(warnings) != 0:			# 警報
			texts += "\t"
			for warning in warnings:
				texts += "{}\t".format(warning.text)
			texts += "\n"
	
		if len(advisorys) != 0:			# 注意報
			texts += "\t"
			for advisory in advisorys:
				texts += "{}\t".format(advisory.text)
			texts += "\n"
	texts += "\n"	# 市町村ごとに改行しておく(見やすくするため)

# 取得した災害情報を表示
print(texts)		# これがメール本文となる

"""
	メール件名の文字列設定
"""

if keihoFlg:		# 警報データあり
	subjectText = "【警報発令@raspi】{}".format(title)
else:			# 警報データなし
	subjectText = "【警報解除@raspi】{}".format(title)	

"""
	前回の送信データと照合
"""
if not os.path.exists(DATA_PATH + DATA_FILE):			# データファイルが存在しない
	print("{} が見つからないためファイルを新規作成します。".format(DATA_PATH + DATA_FILE))
	f = open(DATA_PATH + DATA_FILE, "w")
	f.write(texts)
	f.close()
	print("【書き込んだデータ】\n" + texts)
else:												# データファイルが存在する
	print("現在の災害情報とファイル {} を照合します。".format(DATA_PATH + DATA_FILE))
	f = open(DATA_PATH + DATA_FILE, "r")
	lastMailBody = f.read()
	print("【前回のデータ】\n" + lastMailBody)
	
	# 前回の送信データと一致するか?
	if texts == lastMailBody:
		print("前回とデータが一致しています。\nメール送信はしません。")
		f.close()
		sys.exit()
	else:
		print("前回から変更がありました。\n")
		f.close()
		f = open(DATA_PATH + DATA_FILE, "w")
		f.write(texts)
		f.close()
		print("【書き込んだデータ】\n" + texts)
		print("ファイル{}に書き込みました。".format(DATA_PATH + DATA_FILE))

"""
	Gmail送信(警報・注意報あり または 警報・注意報が解除された場合)
"""

# メール用にデータ提供元URLを本文に追加
texts += "\n\nデータ提供元: {} {}\n".format(title, url)

# Gmailアカウント設定
gmail_addr = GMAIL_ADDR
gmail_pass = GMAIL_PASS
SMTP = "smtp.gmail.com"
PORT = 587

# 送信メール情報
from_addr = gmail_addr						# 送信元メールアドレス(Gmailでなくてもよい)
to_addr = TO_ADDR							# 送信先メールアドレス
subject = subjectText							# 件名
body = texts									# 本文

# メールメッセージを作成
msg = MIMEText(body, "plain", "utf-8")	# ラズパイのpython3では3つの引数でエンコードをutf-8にしないと
										# 'ascii' codec can't encode characters in position 0-14: ordinal not in range(128) というエラーがでる
msg["From"] = from_addr
msg["To"] = to_addr
msg["Date"] = formatdate()						# 日付をインターネットで送信できるフォーマットで指定
msg["Subject"] = subject

# メール送信
try:
	print("メール送信中...")
	send = smtplib.SMTP(SMTP, PORT)				# GmailのSMTPを利用してSMTPオブジェクトを生成
	send.ehlo()
	send.starttls()
	send.ehlo()
	send.login(gmail_addr, gmail_pass)				# Gmailにログイン
	send.send_message(msg)							# メールを送信
	send.close()
except Exception as e:
	print("except: " + str(e))										# エラー送出時のメッセージ表示
else:
	print("{0}へメール送信したよ!".format(to_addr))	# メールが正常に送信されたときのメッセージ


実行方法とイメージ

ターミナルから以下のコマンドで実行できる。

sudo python3 sendSaigaiMail.py 

取得した情報は、ターミナル上に表示し、ファイル sendSaigaiMail.txtに書き込みを行う。
ファイル書き込み後にGメールで送信する。

サンプルで指定した茨城県のURLから「水戸市」と「日立市」の情報を取り出した場合のイメージ。

pi@raspberrypi ~ $ sudo python3 sendSaigaiMail.py
茨城県の警報・注意報 - Yahoo!天気・災害

水戸市
        警報・注意報はありません

日立市
        大雨注意報      波浪注意報      高潮注意報


/home/pi/sendSaigaiMail.txt が見つからないためファイルを新規作成します。
【書き込んだデータ】
茨城県の警報・注意報 - Yahoo!天気・災害

水戸市
        警報・注意報はありません

日立市
        大雨注意報      波浪注意報      高潮注意報


メール送信中...
hoge@yahoo.co.jp へメール送信したよ!

受信メールしたメールはこんな感じ。(ちなみにメール本文の最後にはデータ提供元のURLを追加して送信している)

再度実行した際、前回保存したファイル内容と取得した情報を照らし合わせて、同じ内容であればGメール送信はしない仕組みになっている。

例)前回と同じ情報だった場合

pi@raspberrypi ~ $ sudo python3 sendSaigaiMail.py
茨城県の警報・注意報 - Yahoo!天気・災害

水戸市
        警報・注意報はありません

日立市
        大雨注意報      波浪注意報      高潮注意報


現在の災害情報とファイル /home/pi/python/sendSaigaiMail.txt を照合します。
【前回のデータ】
茨城県の警報・注意報 - Yahoo!天気・災害

水戸市
        警報・注意報はありません

日立市
        大雨注意報      波浪注意報      高潮注意報


前回とデータが一致しています。
メール送信はしません。

crontabに設定して定期実行させる

ターミナルから実行確認が出来たら、crontabに定期実行させるようにすれば全て自動化できる。

sudo crontab -e

でcrontab編集画面を表示させる。

例として5分おきに災害情報をチェックするのであれば、以下のようにコマンドを追加する。(Pythonスクリプトが /home/pi に存在する場合)

*/5 * * * * sudo python3 /home/pi/sendSaigaiMail.py

Ctrl + O で上書き保存。Ctrl + X で終了。

これで警報・注意報が出た時や変化した時警報・注意報が解除された時にのみメール送信がされるようになる。

更に音声合成でしゃべらせるには

AquesTalk Piをつかって音声合成で災害情報をしゃべらせてみる
ラズパイがスピーカーにつながっていることが条件。

AquesTalk Piを、以下のリンク先からダウンロードして所定のディレクトリに展開しておく。
展開したディレクトリ名はaquestalkpiとなり、AquesTalkPiが実行アプリケーション名となる。

AquesTalk Pi
https://www.a-quest.com/products/aquestalkpi.html

先ほどのPythonスクリプトに音声合成でしゃべる処理を加えたものが以下のスクリプト。
ハイライト部分は各自の環境に合わせて編集してください。

sendSaigaiMailTalk.py

#!/usr/bin python3
# -*- coding: utf-8 -*-

"""
	ラズパイ(Raspberry Pi)でYahoo!天気・災害サイトから取得した情報をしゃべらせてGmail通知する
		2021-08-16
		
		python 3.5.3
		OS: Raspbian 9.13 stretch(Raspberry Pi2)にて確認
"""
# スクレイピング関連
import urllib.request
from bs4 import BeautifulSoup
import os, sys
import re

# メール送信のためのSMTPセッションクライアントを作成するため
import smtplib

# メールの送信日付をインターネットで送信できるフォーマットで指定するため
from email.utils import formatdate

# メール本文に日本語を送信できるようにするためのMIME規格に変換するため
from email.mime.text import MIMEText

# 定数関連
keihoFlg = False							# 警報・注意報が出ているか? true: 出ている false: 出ていない
DATA_PATH = "/home/pi/"		# Pythonプログラムとデータファイルのディレクトリ
DATA_FILE = "sendSaigaiMail.txt"			# 送信履歴データファイル
GMAIL_ADDR = "(Gメールアドレス)"	# Gメールアドレス(メール送信元となる)
GMAIL_PASS = "(Gメールパスワード)"			# Gメールのパスワード
TO_ADDR = "(送信先アドレス)"		# 送信先メールアドレス
url = "https://typhoon.yahoo.co.jp/weather/jp/warn/8/"		# 例)茨城県
														# Yahoo!の災害トップ > 警報・注意報のサイト > 地域のURL
AQUES_TALK_PATH = "/home/pi/aquestalkpi/"	# AquesTalk Piを展開したディレクトリ

# 災害情報のサイトにアクセスしHTMLタグを取得
data = urllib.request.urlopen(url)

# HTMLを解析して取得
soup = BeautifulSoup(data, 'html.parser')

# 表示用文字列
texts = ""

# 地域タイトル取得
title = soup.find("title").text
texts += "{}\n\n".format(title)

# classが 'warnArea_table' のタグを取得(表全体のデータとなる)
saigaiTable = soup.find(class_="warnArea_table")

# テーブルの1行ずつを順に処理する
for tr in saigaiTable.find_all("tr"):
	# 市町村名を取得
	city = tr.find("a").text		# aタグに市町村名が入っている
	
	# 指定した市町村名の情報のみ取得する → 【 and city != "(市町村名)"】 で取得したい市町村名を追加
	if city != "水戸市" and city != "日立市":
		continue
	else:
		texts += "{}\n".format(city);	# 市町村名を文字列に追加

	# 特別警報、警報、注意報を取得(全てリストとして取得される)
	emgWarnings = tr.find_all(class_="icoEmgWarning")	# 特別警報
	warnings = tr.find_all(class_="icoWarning")				# 警報
	advisorys = tr.find_all(class_="icoAdvisory")			# 注意報

	# 特別警報・警報・注意報が全て出ていない場合
	if len(emgWarnings) == 0 and len(warnings) == 0 and len(advisorys) == 0:
		texts += "\t警報・注意報はありません\n"

	# どれか一つでも出ていた場合
	else:
		keihoFlg = True
		# 特別警報・警報・注意報ごとに文字列保存
		if len(emgWarnings) != 0:		# 特別警報
			texts += "\t"
			for emgWarning in emgWarnings:
				texts += "{}\t".format(emgWarning.text)
			texts += "\n"
	
		if len(warnings) != 0:			# 警報
			texts += "\t"
			for warning in warnings:
				texts += "{}\t".format(warning.text)
			texts += "\n"
	
		if len(advisorys) != 0:			# 注意報
			texts += "\t"
			for advisory in advisorys:
				texts += "{}\t".format(advisory.text)
			texts += "\n"
	texts += "\n"	# 市町村ごとに改行しておく(見やすくするため)

# 取得した災害情報を表示
print(texts)		# これがメール本文となる

"""
	メール件名の文字列設定
"""
if keihoFlg:		# 警報データあり
	subjectText = "【警報発令@raspi】{}".format(title)
else:			# 警報データなし
	subjectText = "【警報解除@raspi】{}".format(title)	

"""
	前回の送信データと照合
"""
if not os.path.exists(DATA_PATH + DATA_FILE):			# データファイルが存在しない
	print("{} が見つからないためファイルを新規作成します。".format(DATA_PATH + DATA_FILE))
	f = open(DATA_PATH + DATA_FILE, "w")
	f.write(texts)
	f.close()
	print("【書き込んだデータ】\n" + texts)
else:												# データファイルが存在する
	print("現在の災害情報とファイル {} を照合します。".format(DATA_PATH + DATA_FILE))
	f = open(DATA_PATH + DATA_FILE, "r")
	lastMailBody = f.read()
	print("【前回のデータ】\n" + lastMailBody)
	
	# 前回の送信データと一致するか?
	if texts == lastMailBody:
		print("前回とデータが一致しています。\nメール送信はしません。")
		f.close()
		sys.exit()
	else:
		print("前回から変更がありました。\n")
		f.close()
		f = open(DATA_PATH + DATA_FILE, "w")
		f.write(texts)
		f.close()
		print("【書き込んだデータ】\n" + texts)
		print("ファイル{}に書き込みました。".format(DATA_PATH + DATA_FILE))

"""
	AquesTalk Piでしゃべらせる
"""

# 文章をしゃべりやすくする
talkText = texts
talkText = re.sub('\s+', ' ', talkText)	# 複数の改行・タブ・空白文字がつづく場合は半角スペースに置換
talkText = re.sub(' ', '。', talkText)		# 半角スペースを「。」に置換
talkText = "ラズパイ災害警報をお伝えします。{}以上です。".format(talkText)
print("【しゃべるテキスト】" + talkText)

# AquesTalk Piでしゃべらせる
cmd = "{}AquesTalkPi '{}' | aplay".format(AQUES_TALK_PATH, talkText)
result = os.popen(cmd).readline().strip()

"""
	Gmail送信(警報・注意報あり または 警報・注意報が解除された場合)
"""

# メール用にデータ提供元URLを本文に追加
texts += "\n\nデータ提供元: {} {}\n".format(title, url)

# Gmailアカウント設定
gmail_addr = GMAIL_ADDR
gmail_pass = GMAIL_PASS
SMTP = "smtp.gmail.com"
PORT = 587

# 送信メール情報
from_addr = gmail_addr						# 送信元メールアドレス(Gmailでなくてもよい)
to_addr = TO_ADDR							# 送信先メールアドレス
subject = subjectText							# 件名
body = texts									# 本文

# メールメッセージを作成
msg = MIMEText(body, "plain", "utf-8")	# ラズパイのpython3では3つの引数でエンコードをutf-8にしないと
										# 'ascii' codec can't encode characters in position 0-14: ordinal not in range(128) というエラーがでる
msg["From"] = from_addr
msg["To"] = to_addr
msg["Date"] = formatdate()						# 日付をインターネットで送信できるフォーマットで指定
msg["Subject"] = subject

# メール送信
try:
	print("メール送信中...")
	send = smtplib.SMTP(SMTP, PORT)				# GmailのSMTPを利用してSMTPオブジェクトを生成
	send.ehlo()
	send.starttls()
	send.ehlo()
	send.login(gmail_addr, gmail_pass)				# Gmailにログイン
	send.send_message(msg)							# メールを送信
	send.close()
except Exception as e:
	print("except: " + str(e))										# エラー送出時のメッセージ表示
else:
	print("{0}へメール送信したよ!".format(to_addr))	# メールが正常に送信されたときのメッセージ

しゃべる部分のスクリプト説明

Gメール送信前に音声合成でしゃべらせている。

"""
	AquesTalk Piでしゃべらせる
"""

# 文章をしゃべりやすくする
talkText = texts
talkText = re.sub('\s+', ' ', talkText)	# 複数の改行・タブ・空白文字がつづく場合は半角スペースに置換
talkText = re.sub(' ', '。', talkText)		# 半角スペースを「。」に置換
talkText = "ラズパイ災害警報をお伝えします。{}以上です。".format(talkText)
print("【しゃべるテキスト】" + talkText)

# AquesTalk Piでしゃべらせる
cmd = "{}AquesTalkPi '{}' | aplay".format(AQUES_TALK_PATH, talkText)
result = os.popen(cmd).readline().strip()

メール本文の文字列をそのままつかってしまうと改行やタブ、余計な空白などがはいってしまいAquesTalk Piに渡した場合、誤動作の原因となりやすい。
そこで改行、タブ、空白文字は全て一旦半角スペース1文字に置換する。その後、半角スペースを「。」(句点)に置換している。

例えば

大雨注意報    波浪注意報 

などの文字列が

大雨注意報。波浪注意報。 

と置換される。この方が音声合成した時に「間」ができて聞き取りやすい。

実際にしゃべらせる部分はPythonからシェルを起動している。

# AquesTalk Piでしゃべらせる
cmd = "{}AquesTalkPi '{}' | aplay".format(AQUES_TALK_PATH, talkText)
result = os.popen(cmd).readline().strip()

上記は、実際にはターミナルから

/home/pi/AquesTalkPi "しゃべる文字列" | aplay

などと実行したのと同じ意味となる。

コメント

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