ラズパイで天気予報・気象警報を喋らせる（Voicevox）備忘録

初めに

気象庁の仕様変更を機に春日部つむぎボイスでしゃべらせたくなり作りました．
コードは前回とほぼ変わりませんが記録として残します．
ちなみにNHKのRSS仕様変更によりニュースは途中で止まります．

使用環境

・Raspberry pi 4B
・OS:Raspberry pi OS(trixie 64bit)

事前準備

・Voicevoxを利用できるようにしてください(Raspberry Pi単体では非常に重かったためエンジンは別のPCで実行しました)
・mpg321をインストールしてください(参考)
・時報ボイスをそれぞれ0～23.mp3として保存
・市町村区のコード,アメダスNo.を以下を参考に取得してください。

市町村区のコードの取得

このページから地方エリアのコードと市町村区のコードを取得しておきます。
例えば東京都千代田区の場合、関東甲信地方の東京都東京地方であり、そのあと千代田区を選択した際URLのclass20s&area_code=""の部分が市町村区のコードとなります。東京都千代田区は1310100になります。

アメダスNo.の取得

ここからもっとも近い場所のアメダスを選択してください。選択した際のURLのamdno=""がアメダスNo.となります。東京都千代田区の場合,44132です。
https://www.jma.go.jp/bosai/amedas/const/amedastable.json から探すこともできます。

実装

・適当な場所に以下のファイルを作ってください。

talk_function.py

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

import requests
import argparse
import json
import shlex
import subprocess

#話者ID(デフォルト8：春日部つむぎ)
speaker = 8
VOLUME = 3.0 #ボリューム
output_path = '/path/to/音声ファイルの出力先'

def voicevox(number, input_texts):
        HOSTNAME='10.207.171.20'#エンジンを起動しているPC
        #HOSTNAME='localhost'
        #audio_query (音声合成用のクエリを作成するAPI)
        res1 = requests.post('http://' + HOSTNAME + ':50021/audio_query',
        params={'text': input_texts, 'speaker': speaker}).json()
        res1["volumeScale"] = VOLUME
        #synthesis (音声合成するAPI)
        res2 = requests.post('http://' + HOSTNAME + ':50021/synthesis',
        params={'speaker': speaker},
        data=json.dumps(res1))
        # wavファイルに書き込み
        with open(output_path + '/' + str(number) + f'.wav', mode='wb') as f:
                f.write(res2.content)

        return

def trans_warning(code):
        trans_warning = {
                "02":"暴風雪警報",
                "03":"大雨警報",
                "04":"洪水警報",
                "05":"暴風警報",
                "06":"大雪警報",
                "07":"波浪警報",
                "08":"高潮警報",
                "09":"土砂災害警報",
                "10":"大雨注意報",
                "12":"大雪注意報",
                "13":"風雪注意報",
                "14":"雷注意報",
                "15":"強風注意報",
                "16":"波浪注意報",
                "17":"融雪注意報",
                "18":"洪水注意報",
                "19":"高潮注意報",
                "20":"濃霧注意報",
                "21":"乾燥注意報",
                "22":"なだれ注意報",
                "23":"低温注意報",
                "24":"霜注意報",
                "25":"着氷注意報",
                "26":"着雪注意報",
                "29":"土砂災害注意報",
                "32":"暴風雪特別警報",
                "33":"大雨特別警報",
                "35":"暴風特別警報",
                "36":"大雪特別警報",
                "37":"波浪特別警報",
                "38":"高潮特別警報",
                "39":"土砂災害特別警報",
                "43":"大雨危険警報",
                "48":"高潮危険警報",
                "49":"土砂災害危険警報"}
        return trans_warning[f"{code}"]
        
def WDR_INFO(code):
        WDR_INFO = {
                "0":"静穏",
                "1":"北北東",
                "2":"北東",
                "3":"東北東",
                "4":"東",
                "5":"東南東",
                "6":"南東",
                "7":"南南東",
                "8":"南",
                "9":"南南西",
                "10":"南西",
                "11":"西南西",
                "12":"西",
                "13":"西北西",
                "14":"北西",
                "15":"北北西",
                "16":"北"}
        return WDR_INFO[f"{code}"]

def wspeed(ws):
        if 0 <= ws <= 0.3:
                return (0, u"静穏")
        elif 0.3 <= ws < 1.6:
                return (1, u"至軽風")
        elif 1.6 <= ws < 3.4:
                return (2, u"軽風")
        elif 3.4 <= ws < 5.5:
                return (3, u"軟風")
        elif 5.5 <= ws < 8.0:
                return (4, u"和風")
        elif 8.0 <= ws < 10.8:
                return (5, u"疾風")
        elif 10.8 <= ws < 13.9:
                return (6, u"雄風")
        elif 13.9 <= ws < 17.2:
                return (7, u"強風")
        elif 17.2 <= ws < 20.8:
                return (8, u"疾強風")
        elif 20.8 <= ws < 24.5:
                return (9, u"大強風")
        elif 24.5 <= ws < 28.5:
                return (10, u"暴風")
        elif 28.5 <= ws < 32.7:
                return (11, u"烈風")
        elif 32.7 <= ws:
                return (12, u"颶風")

上記のファイルと同じディレクトリに以下のファイルを作ってください。

jma_talkweather.py

#!/usr/bin/env python3
# -*- coding:utf-8 -*-
import shlex
import subprocess
from datetime import datetime
import time
import urllib.request
import feedparser
import os.path
import shutil
import json
import threading
import sys
import talk_function


LOAD_MP3 = "mpg321 -q"

#東京都千代田区
CLASS_AREA_CODE = "1310100" # 市町村区のコード
AMDNO = "44132" #アメダスNo.

da = datetime.now()
area_data = urllib.request.urlopen('https://www.jma.go.jp/bosai/common/const/area.json')
area_data = json.loads(area_data.read())
area = area_data["class20s"][CLASS_AREA_CODE]["name"]
class15s_area_code = area_data['class20s'][CLASS_AREA_CODE]['parent']
class10s_area_code = area_data['class15s'][class15s_area_code]['parent']
offices_area_code = area_data['class10s'][class10s_area_code]['parent']

say_counter_end = -1

if os.path.isdir('/path/to/音声ファイルの出力先'):
        shutil.rmtree('/path/to/音声ファイルの出力先')
os.mkdir('/path/to/音声ファイルの出力先')

def main():
        say_datetime()
        say_weather()
        say_news()
        return

def jma():
                global jma_data
                #気象警報・注意報
                #情報の取得
                warning_url = "https://www.jma.go.jp/bosai/warning/data/r8/%s.json" % (offices_area_code)
                warning_info = urllib.request.urlopen(url=warning_url)
                warning_info = json.loads(warning_info.read())
                warning_codes = [warning["code"]
                        for dataTypeCode in warning_info
                        for class_area in dataTypeCode["warning"]['class20Items']
                        if class_area['areaCode'] == CLASS_AREA_CODE
                        for warning in class_area["kinds"]
                        if warning["status"] != "解除" and warning["status"] != "発表警報・注意報はなし"]
                warning_texts = [talk_function.trans_warning(code) for code in warning_codes]
                warning_headlinetexts = [dataTypeCode["headlineText"]
                        for dataTypeCode in warning_info
                        if dataTypeCode["headlineText"] != "注意報を解除します。"]

                #現在の気温・湿度
                amedas_da = da.timestamp() - 600
                amedas_da = datetime.fromtimestamp(amedas_da)
                amedas_time = f"{(amedas_da.hour//3)*3:02}"
                temp_url = "https://www.jma.go.jp/bosai/amedas/data/point/%s/%s_%s.json" % (AMDNO, amedas_da.strftime("%Y%m%d"), amedas_time)
                temp_data = urllib.request.urlopen(url=temp_url)
                temp_data = json.loads(temp_data.read())
                last_time = [timelist for timelist in temp_data][-1]
                now_temp = temp_data[last_time]["temp"][0]
                now_humidity = temp_data[last_time]["humidity"][0]
                if temp_data[last_time]["windDirection"][0] == 0:
                        now_windDirection = []
                else:
                        now_windDirection = transweather.WDR_INFO(temp_data[last_time]["windDirection"][0]) + "の風"
                now_wind_speed = temp_data[last_time]["wind"][0]

                #明日の風速
                tommorow_wind_speed_url = "https://www.jma.go.jp/bosai/jmatile/data/wdist/VPFD/%s.json" % (class10s_area_code)
                tommorow_wind_speed_data = urllib.request.urlopen(url=tommorow_wind_speed_url)
                tommorow_wind_speed_data = json.loads(tommorow_wind_speed_data.read())
                for counter,weather_time in enumerate(tommorow_wind_speed_data["areaTimeSeries"]["timeDefines"]):
                        if  da.strftime("%Y-%m-%d") not in weather_time["dateTime"]:
                                weather_time = counter
                                break
                tommorow_wind_speed = 0
                for counter,tommorow_wind_speeds in enumerate(tommorow_wind_speed_data["areaTimeSeries"]["wind"][weather_time::]):
                        idx = tommorow_wind_speeds["range"].find(" ")
                        tommorow_wind_speed = tommorow_wind_speed + int(tommorow_wind_speeds["range"][idx+1:])
                tommorow_wind_speed = round(tommorow_wind_speed/counter + 1)

                #天気＆明日の風向き
                weather_url = "https://www.jma.go.jp/bosai/forecast/data/forecast/%s.json" % (offices_area_code)
                weather_data =urllib.request.urlopen(url=weather_url)
                weather_data = json.loads(weather_data.read())
                if da.strftime("%Y-%m-%d") in weather_data[0]["timeSeries"][0]["timeDefines"][0]:
                        today = 0
                        tommorow = today + 1
                else:
                        today = 1
                        tommorow = today + 1
                weathers = [(area_number,weather) for area_number,weather in enumerate(weather_data[0]["timeSeries"][0]["areas"])
                        if weather["area"]["code"] == class10s_area_code]
                today_weather = weathers[0][1]["weathers"][today]
                tommorow_weather = weathers[0][1]["weathers"][tommorow]
                tommorow_windDirection = weathers[0][1]["winds"][tommorow]
                today = ""
                for counter,i in enumerate(weather_data[0]["timeSeries"][2]["timeDefines"]):
                        if da.strftime("%Y-%m-%d") in i and today == "":
                                today = counter
                                tommorow = today + 2
                if today == "":
                        tommorow = 0
                if tommorow > counter:
                        tommorow = []
                temps = [area["temps"] for area in weather_data[0]["timeSeries"][2]["areas"]
                        if area["area"]["code"] == AMDNO]
                if tommorow != 0:
                        today_min = temps[0][today]
                        today_max = temps[0][today + 1]
                        if today_min == today_max:
                                today_min = []
                else:
                        today_min = today_max = []
                if tommorow !=[]:
                        tommorow_min = temps[0][tommorow]
                        tommorow_max = temps[0][tommorow + 1]
                else:
                        tommorow_min = tommorow_max = []

                #降水確率
                last_da = da.timestamp() - 21600
                last_da = datetime.fromtimestamp(last_da)
                last_time = f"{((last_da.hour)//6)*6:02}"
                now_da = da.timestamp() + 21600
                now_da = datetime.fromtimestamp(now_da)
                now_time = f"{((now_da.hour)//6)*6:02}"
                now_pop = 0
                tommorow_pop = 0
                for counter,timeDegines in enumerate(weather_data[0]["timeSeries"][1]["timeDefines"]):
                        if now_da.strftime("%Y-%m-%d") + "T" + now_time in timeDegines:
                                now_pop = weather_data[0]["timeSeries"][1]["areas"][weathers[0][0]]["pops"][counter]
                        if now_da.strftime("%Y-%m-%d") not in timeDegines and last_da.strftime("%Y-%m-%d") + "T" + last_time not in timeDegines:
                                break
                if counter + 1 == len(weather_data[0]["timeSeries"][1]["timeDefines"]):
                        tommorow_pop = []
                else:
                        tommorow_pop = max(weather_data[0]["timeSeries"][1]["areas"][weathers[0][0]]["pops"][counter::])

                #出力
                jma_data = {
                        "warning_texts":warning_texts,
                        "warning_headlinetexts":warning_headlinetexts,
                        "now_temp":now_temp,
                        "now_humidity":now_humidity,
                        "now_windDirection":now_windDirection,
                        "now_wind_speed":now_wind_speed,
                        "tommorow_windDirection":tommorow_windDirection,
                        "tommorow_wind_speed":tommorow_wind_speed,
                        "today_weather":today_weather,
                        "tommorow_weather":tommorow_weather,
                        "today_min":today_min,
                        "today_max":today_max,
                        "tommorow_min":tommorow_min,
                        "tommorow_max":tommorow_max,
                        "now_pop":now_pop,
                        "tommorow_pop":tommorow_pop
                }
                return jma_data

def load_mp3():
        text = "/時報ボイスのディレクトリのパス/%s.mp3" % (da.hour)
        if 1<= da.hour <=4:
                text = LOAD_MP3 + " -g 40 " + text
        else:
                text = LOAD_MP3 + " " + text
        print(text)
        proc = subprocess.Popen(shlex.split(text))
        proc.communicate()
        return

def load_wav():
        counter = 0
        n = 0
        global say_counter
        global say_counter_end
        while 100 >= n:
                n += 1
                path = "/path/to/音声ファイルの出力先/%s.wav" % (counter)
                if say_counter >= counter and os.path.isfile(path):
                        text ="aplay " + path
                        proc = subprocess.Popen(shlex.split(text))
                        proc.communicate()
                        counter += 1
                elif say_counter_end == counter:
                        da = datetime.now()
                        text = "voicevox -t %s時%s分です。" % (da.hour, da.minute)
                        print(text)
                        proc = subprocess.Popen(shlex.split(text))
                        proc.communicate()
                        shutil.rmtree('/path/to/音声ファイルの出力先')
                        exit()
                else:
                        time.sleep(1)
        return

def say_datetime():
        global say_counter
        global say_counter_end
        if 6<= da.hour <=23:
                say_counter = 0
                text = "天気予報です。%s年%s月%s日、%s時をお知らせします。" % (da.year, da.month, da.day, da.hour)
                print(text)
                talk_function.voicevox(say_counter, text)
                return
        else:
                say_counter = -1
                say_counter_end = 0
                exit()

def say_weather():
        global say_counter
        #say_counter += 1
        try:
                title_text = u'%sの天気' % (area)
                weather_text = u'%sは%s、風力%s　%s　天気は%s。'
                pop_text = '降水確率は%s％です。'
                now_text = u'現在の気温は%s度、湿度%s％です。'
                temperature_text = u'%sの予想最高気温は%s度　予想最低気温は%s度です。'

                #気象警報
                if jma_data["warning_texts"] == []:
                        warning_text = "現在発表されている気象警報・注意報はありません。"
                else:
                        warning_text = "現在、%sが発表されています。" % ("、".join(jma_data["warning_texts"]))
                warning_headlinetext = "、".join(jma_data["warning_headlinetexts"])

                #今日
                now_pop = jma_data["now_pop"]
                wind_deg_now = jma_data["now_windDirection"]
                (wind_speed_now,wind_speed_name) = talk_function.wspeed(jma_data["now_wind_speed"])
                today_w = jma_data["today_weather"]
                today_pop_txt = pop_text % (now_pop)
                today_weather_txt = weather_text % (u"今日", jma_data["now_windDirection"], wind_speed_now, wind_speed_name, today_w)
                if jma_data["today_min"] == [] and jma_data["today_min"] != jma_data["today_max"]:
                        today_temp_txt = u'今日の予想最高気温は、%s度です。' % (jma_data["today_max"])
                elif jma_data["today_min"] == jma_data["today_max"]:
                        today_temp_txt = ""
                else:
                        today_temp_txt = temperature_text % (u"今日", jma_data["today_max"], jma_data["today_min"])
                today_now_txt = now_text % (jma_data["now_temp"], jma_data["now_humidity"])
                
                #明日
                tommorow_pop = jma_data["tommorow_pop"]
                wind_deg_tommorow = jma_data["tommorow_windDirection"]
                (wind_speed_tommorow,wind_speed_name) = talk_function.wspeed(jma_data["tommorow_wind_speed"])
                tommorow_weather = jma_data["tommorow_weather"]
                if tommorow_pop == []:
                        tommorow_pop_txt = ""
                else:
                        tommorow_pop_txt = pop_text % (tommorow_pop)
                tommorow_weather_txt = weather_text % (u"明日", wind_deg_tommorow, wind_speed_tommorow, wind_speed_name, tommorow_weather)
                if jma_data["tommorow_max"] == jma_data["tommorow_min"]:
                        tommorow_temp_txt = ""
                else:
                        tommorow_temp_txt = temperature_text % (u"明日", jma_data["tommorow_max"], jma_data["tommorow_min"])

                #実行
                #say_text = (title_text + ' ' + today_weather_txt + ' ' + today_pop_txt + ' ' + warning_text + ' ' + today_now_txt + ' ' + today_temp_txt + ' ' + tommorow_weather_txt + ' ' + tommorow_pop_txt + ' ' + tommorow_temp_txt).r>
                say_text = {
                        1:title_text,
                        2:today_weather_txt,
                        3:today_pop_txt,
                        4:warning_text,
                        5:warning_headlinetext,
                        6:today_now_txt.replace("-", "マイナス"),
                        7:today_temp_txt.replace("-", "マイナス"),
                        8:tommorow_weather_txt,
                        9:tommorow_pop_txt,
                        10:tommorow_temp_txt.replace("-", "マイナス")
                }
        except:
                print("エラーが出ました。:", sys.exc_info()[0])
                say_text = "atalk エラー発生。ログを確認してください"
        finally:
                for say in range(len(say_text)):
                        say_counter += 1
                        print(say_text[say + 1])
                        talk_function.voicevox(say_counter, say_text[say + 1])
        return

def say_news():
        global say_counter
        global say_counter_end #
        news_text = ""
        news = ["0", "4"]
        for loop in news:
                RSS_URL = 'https://www.nhk.or.jp/rss/news/cat%s.xml' % (loop)
                news_data = feedparser.parse(RSS_URL)
                for i, entry in enumerate(news_data.entries):
                        newstime = ",".join(map(str, entry.published_parsed[:5]))
                        newstime = datetime.strptime(newstime, "%Y,%m,%d,%H,%M")
                        if i == 0 and loop == news[0]:
                                say_text = "続いてニュースです。" + entry.summary
                        else:
                                say_text = "次のニュースです。" +  entry.summary
                        if da.timestamp() - newstime.timestamp() + 32400 <= 86400 and entry.summary not in news_text:
                                say_counter += 1
                                print(say_text)
                                talk_function.voicevox(say_counter, say_text)
                                news_text = news_text + entry.summary
        say_counter_end = say_counter + 1 #
        return

def say_datetime2():
        da = datetime.now()
        text = "%s時%s分です。" % (da.hour, da.minute)
        text = CMD_SAY + ' ' + text
        print(text)
        proc = subprocess.Popen(shlex.split(text))
        proc.communicate()
        return

### Execute
if __name__ == "__main__":
        th1 = threading.Thread(target=jma)
        th2 = threading.Thread(target=main)
        th1.start()
        th2.start()
        load_mp3()
        load_wav()

実際に実行してみるとこんな感じになると思います。

.venv/bin/python jma_talkweather.py

天気予報です。2023年6月10日、13時をお知らせします。
千代田区の天気 今日は南南東の風、風力2。軽風。天気はくもり　所により　夕方　から　夜のはじめ頃　雨。 降水確率は20％です。 現在発表されている気象警報・注意報はありません。 現在の気温は24.2度、湿度80％です。 今日の予想最高 気温は、26度です。 明日は南の風、風力5。疾風。天気はくもり　時々　雨。 降水確率は70％です。 明日の予想最高気温は、22度。予想最低気温は、20度です。

参考文献

Raspberry Piに現在時刻、天気予報、ニュースを喋らせる
現在出ている気象警報・注意報を取得する
新しい気象庁サイトからJSONデータが取得できる件