気象庁の天気予報追跡Webアプリを作った

概要

以前，こんな記事を書きました．

当時は，Google Colaboratoryを実行することでモニタリング結果が見られるようにしていました．

しかし最近，Streamlitというフレームワークを使って超簡単にWebアプリを作る&デプロイできることを知り，これをWebアプリ化することにしました．

本記事はその実装の解説です．

Streamlitの便利さはこちらの記事で簡単に紹介したことがあるので，興味のある方はぜひご覧ください．

処理フロー

内部処理はほぼColab版と同じです．

基本的な処理フローは以下のような形です．

1. スプレッドシートから値を取得し，データを整形
2. 「天気コード：天気アイコン」の対応表，「地域コード：地域名」の対応表を読み込む
3. 指定した予報対象日時と地域にマッチするレコードを抽出
4. 信頼度，降水確率，最低気温，最高気温のグラフを表示

1. スプレッドシートから値を取得し，データを整形

こちらの記事を参考にして取得したJSONから認証情報を読み込み，スプレッドシートからデータを読めるようにします．

ただ今回はStreamlit Shareにデプロイする関係上，ソースコードはPublicにしておく必要があるため，認証情報をオープンにするわけにはいきません．

なので下の実装では（credentials=...の部分），環境変数で認証情報を取得するようにしています．

import gspread
from oauth2client.service_account import ServiceAccountCredentials
import os

# 認証
scope = ['https://spreadsheets.google.com/feeds','https://www.googleapis.com/auth/drive']

credentials = ServiceAccountCredentials.from_json_keyfile_dict({
    "type": os.environ.get("TYPE"),
    "project_id": os.environ.get("PROJECT_ID"),
    "private_key_id": os.environ.get("PRIVATE_KEY_ID"),
    "private_key": os.environ.get("PRIVATE_KEY"),
    "client_email": os.environ.get("CLIENT_EMAIL"),
    "client_id": os.environ.get("CLIENT_ID"),
    "auth_uri": os.environ.get("AUTH_URI"),
    "token_uri": os.environ.get("TOKEN_URI"),
    "auth_provider_x509_cert_url": os.environ.get("AUTH_PROVIDER_X509_CERT_URL"),
    "client_x509_cert_url": os.environ.get("CLIENT_X509_CERT_URL"),
    "universe_domain": os.environ.get("UNIVERSE_DOMAIN")
}, scope)
gc = gspread.authorize(credentials)

# データ取得
if 'df' not in st.session_state:
    df = None
    with st.spinner('データを取得しています...'):
        for i in range(40):
            sheet = gc.open_by_key('-----スプレッドシートのファイルID-----').worksheet(f'data{i}')
            data = sheet.get_all_values()

            # データを整形
            tmp = pd.DataFrame(data[1:], columns=data[0])
            tmp = tmp.replace('', np.nan)
            tmp = tmp.astype({'pop':'float','tempMin':'float','tempMinUpper':'float','tempMinLower':'float','tempMax':'float','tempMaxUpper':'float','tempMaxLower':'float'})
            if tmp.shape[0] == 0:
                break

            # 結合
            if df is None:
                df = tmp
            else:
                df = pd.concat([df,tmp])
    st.session_state.df = df
else:
    df = st.session_state.df

2. 対応表を読み込む

気象庁から送信される天気予報のレスポンスには，地域名および天気は「京都府」とか「晴」の形で記述されているのではなく，「260000」とか「100」というコードで記述されています．

そこで対応表が必要になるので，JSON形式の対応表を読み込みます．

地域名－地域コードの対応表についてはこちらの記事を，

天気－天気コードの対応表については，こちらの記事の「5-2.お天気アイコンを表示する」を参考にしてください．

# 地域名と地域コードの対応表
with open('./areaCode.json', 'r', encoding='utf-8') as f:
    areaCode_dict = json.load(f)

# 天気コードと天気の対応表
with open('./weatherCode.json', 'r', encoding='utf-8') as f:
    weatherCode_dict = json.load(f)

3. 指定した予報対象日時と地域にマッチするレコードを抽出

変数areaに地域名，変数targetDateに予報対象日時をYYYY-MM-DD形式で格納されているとし，条件に合致するレコードをdfから取り出してsubとします．

# 抽出
areaCode = areaCode_dict[area]
sub = df[df['targetDate']+df['areaCode'] == targetDate+'T00:00:00+09:00'+areaCode].copy()

4. 信頼度，降水確率，最低気温，最高気温のグラフを表示

あとはデータsubにもとづいて，Streamlitで表示させるだけです．

# 予報対象日時
cols0 = st.columns(2)
cols0[0].metric(label='予報対象日', value=targetDate)
cols0[1].metric(label='予報対象地域', value=area)

# 天気
st.divider()
st.write('**天気**')
cols = [st.columns(4) for i in range(4)]
for i,(date,weatherCode) in enumerate(zip(sub['reportDatetime'].values,sub['weatherCode'].values)):
    with cols[i//4][i%4]:
        st.write(date[:10] + '(' + date[11:13] + '時)時点')
        if weatherCode is not np.nan:
            st.image(f'https://www.jma.go.jp/bosai/forecast/img/{weatherCode_dict[str(weatherCode)][0]}')

cols1 = st.columns(2)
cols2 = st.columns(2)
sub['reportDatetime'] = pd.to_datetime(sub['reportDatetime'])

# 最低気温
cols1[0].write('**最低気温**')
cols1[0].line_chart(data=sub.dropna(), x='reportDatetime', y=['tempMin','tempMinLower','tempMinUpper'], color=['#0000ff','#a9ceec','#a9ceec'])
# 最高気温
cols1[1].write('**最高気温**')
cols1[1].line_chart(data=sub.dropna(), x='reportDatetime', y=['tempMax','tempMaxLower','tempMaxUpper'], color=['#ff0000','#ffb6c1','#ffb6c1'])
# 信頼度
cols2[0].write('**信頼度**')
cols2[0].line_chart(data=sub.dropna(), x='reportDatetime', y='reliability')
# 降水確率
cols2[1].write('**降水確率**')
cols2[1].area_chart(data=sub.dropna(), x='reportDatetime', y='pop')

Streamlitの実装ポイントとしては，

• st.columns
  • st.columns(2)で2カラムのブロックを生成できます．
• st.divider
  • st.divider()と書くだけで，その部分に区切り線を書けます．
• st.metric
  • ちょっとこじゃれた感じになります．
• st.image
  • 引数に画像URLを貼れば表示してくれました．
• st.line_chart
  • data=にPandasDataFrameを，x=やy=に列名を指定するだけでインタラクティブなグラフが書けます．

完成品

こうしてできたのがこちらのWebアプリです．

予報対象日と地域を選択するだけで，天気予報の履歴を見ることができます．