DataStudio + CloudRun + gspread で都道府県別の医療提供体制の状況を可視化する

はじめに

新型コロナウイルスにより、多くの都道府県の医療提供体制が逼迫する状況となっています。
新型コロナ 病床ひっ迫「ステージ4」5都道府県で指標超える, NHK, 2020/12/12

この状況を表す指標は、以下の厚生労働省のページにて、PDFおよびXLSXのデータで公開されています。
都道府県の医療提供体制等の状況（医療提供体制・監視体制・感染の状況）について（６指標）

このデータの可視化は、例えば、新型コロナ　感染状況のステージと６指標, yahoo japan で行われており、指標の解説とともに詳しくまとまっています。しかし、ソートや絞り込み、グラフ化などはできません。

改善にはまずは計測から、と言いますが、計測結果はすでにあるので、可視化から始めよう、ということで、DataStudioを使って、このデータを可視化してみようと思います。また、CloudRunなどを使って、低コストで自動的にデータを更新するようにしてみます。

できあがったもの

DataStudioで作ったダッシュボードと、そのデータソースとなるスプレッドシート、および実装を公開しています。

突貫で作ったため、もしかしたｒ，あ誤りがあるかもしれません。お気づきの点があれば、ダッシュボードに記載している問い合わせフォームや、Issue、こちらのコメントなどでご指摘いただければありがたいです。

• 都道府県の医療提供体制等の状況(非公式ダッシュボード)
  • 非常に重く、スマホで確認するのは難しいかもしれません。
• データソースとなるスプレッドシート
• 実装 (github:covid19-status-of-medical-care-provision-in-japan)

処理の流れ

厚生労働省のページを定期的に巡回し（1日1回程度)、XLSXのリンク先を取得し、更新があった場合にはデータを抽出し、スプレッドシートに投入します。これをDataStudioのデータソースとして利用しています。

更新の確認や抽出の処理はpythonで実装し、Cloud Run上で動いています。定期実行のために、Cloud Schedulerを使っています。また、デプロイは、github actionで自動化しました。

解説

実装を公開しているのでそちらを参考にしていただいたらと思いますが、簡単に処理やインフラを解説します。

データ抽出

厚生労働省のページで公開されているXLSXを、機械的に処理しやすくするため情報を取り出します。pandasのread_excelを使うことで、直接XLSXを読みこみデータフレームを作成しています。

xlsx_url = "https://www.mhlw.go.jp/content/10900000/000704818.xlsx"
df = pd.read_excel(xlsx_url)

あとは泥臭く、DataFrameを処理し、必要な情報を取り出していきます。可視化の際に利用しやすいように、都道府県の医療提供体制の現状を表す6指標の数値と、いくつかのメタデータ（データ集計時点、ステージの指標、注釈）に分けてデータを取得しました。

参考までに、6指標の数値を取り出すコードを抜粋します。列名がセルで結合されているため、ひとつひとつ書き直すことにしました。

def get_formated_df(unformat_df):
    df = (
        unformat_df.iloc[9:57, 1:]
        .reset_index(drop=True)
        .rename(
            columns={
                "（参考）都道府県の医療提供体制等の状況①（医療提供体制）": "都道府県",
                # 単位: 千人
                "Unnamed: 2": "人口",
                #
                # 医療提供体制
                #
                # ステージ3の指標: 25%  ステージ4の指標: なし
                "Unnamed: 3": "確保_病床使用率",
                "Unnamed: 4": "確保_病床使用率_前週差",
                "Unnamed: 5": "確保想定_病床使用率",
                "Unnamed: 6": "確保想定_病床使用率_前週差",
                "Unnamed: 7": "確保_病床使用率_重症患者",
                "Unnamed: 8": "確保_病床使用率_重症患者_前週差",
                "Unnamed: 9": "確保想定_病床使用率_重症患者",
                "Unnamed: 10": "確保想定_病床使用率_重症患者_前週差",
                # 単位: 対人口10万人
                "Unnamed: 11": "療養者数",
                "Unnamed: 12": "療養者数_前週差",
                "Unnamed: 13": "調整列",
                "（参考）都道府県の医療提供体制等の状況②（監視体制・感染の状況）": "都道府県_2",
                "Unnamed: 15": "人口_2",
                #
                # 監視体制
                #
                "Unnamed: 16": "陽性者数_PCR検査件数_最近1週間",
                "Unnamed: 17": "陽性者数_PCR検査件数_最近1週間_前週差",
                "Unnamed: 18": "調整列_2",
                #
                # 感染の状況
                #
                "Unnamed: 19": "直近1週間の陽性者数",
                "Unnamed: 20": "直近1週間の陽性者数_前週差",
                "Unnamed: 21": "直近1週間とその前1週間の比",
                "Unnamed: 22": "直近1週間とその前1週間の比_前週差",
                "Unnamed: 23": "感染経路不明な者の割合",
                "Unnamed: 24": "感染経路不明な者の割合_前週差",
            }
        )
        .drop(columns=["調整列", "調整列_2", "都道府県_2", "人口_2"])
    )
    df["人口"] = df["人口"] * 1000
    return df

gspread

抽出したデータは、gspreadを利用して、スプレッドシートに書き込みます。書き込む際には権限が必要なため、あらかじめ作成したサービスアカウントを利用します。この手順については、公式ドキュメントをご確認ください。

ローカルで動作し、かつCloudRun上でも実行できるようにしたいので、このサービスアカウントの指定は、GOOGLE_APPLICATION_CREDENTIALSを使って行いたいです。上記のドキュメントに記載されておらず少々苦労しましたが、このissueなどを参考に結果的に、google.auth.default を使いました。

scopes = [
    "https://www.googleapis.com/auth/spreadsheets",
    "https://www.googleapis.com/auth/drive"
]

def get_sh():
    credentials, _ = google.auth.default(scopes=scopes)
    gc = gspread.authorize(credentials)
    sh = gc.open_by_key(settings.spreadsheet_id)
    return sh

シートへの書き込みは、worksheetを取得してupdateに、書き込みたいデータを渡すだけと、非常に簡単に実装できます。

    worksheet = sh.worksheet("医療提供体制")
    worksheet.update([df.columns.values.tolist()] + df.values.tolist())

シートには、上で記載した抽出したデータに加えて、データ取得元のXLSXのURLや、ダッシュボードの説明となるテキストも合わせてに書き込みました。特にこのURLは、ダッシュボード上での表示だけではなく、後述のクロールの際にも利用します。

データ更新のためのクロール

XLSXが記載される厚生労働省のページは週に1回程度更新が入り、新たなXLSXへのリンクが追加されます。しかし、いつ更新されるかはわからないので、定期的なクロールにより、この更新チェックを自動化したいと考えました。取得したいURLの命名規則も明確だったので、requests.getを使い、正規表現でURLを取り出すような単純なものになっています。

import requests
import re

page_url = "https://www.mhlw.go.jp/stf/seisakunitsuite/newpage_00035.html"
base_url = baseurl = "https://www.mhlw.go.jp"


def get_latest_xlsx_url():
    response = requests.get(page_url)
    pattern = r"/content/10900000/[0-9]+\.xlsx"
    latest_xlsx_path = re.findall(pattern, response.text)[-1]
    xlsx_url = baseurl + latest_xlsx_path
    return xlsx_url

このようにクロールして取り出したURLが、シートに書き込まれたURLと同じ場合には処理を終了し、異なっている場合には、データの抽出や書き込みを行うか判断しています。

CloudRun

バックエンドにはCloudRunを利用します。そのため、上記処理をまとめ、flaskを使って、非常に簡単なwebアプリケーションにしています。
CloudRunは、URLの公開・非公開を決めることができます。誤って公開しないようにしましょう。デプロイのオプションにより、不要な処理を行えないようにするため、同時実行数や、最大インスタンス数なども最小に設定しています。CloudRunからスプレッドシートに書き込むため、デフォルトのサービスアカウントではなく、スプレッドシートを操作できる権限を与えたサービスアカウントを指定しています。

Google Container Registory と github action による CD

CloudRunで動かすために、コンテナレジストリへの登録が必要です。Google Container Registory(GCR)を利用しました。
ビルド、登録、デプロイなどをまとめており、mainブランチへのマージですぐにビルド・リリースが実行されます。

こちらのページを参考にさせていいただきました。
https://qiita.com/szk3/items/38a3dba7fdfed189f4c9

gcloud run deployで、サービスアカウントを指定するため、github actionに利用するサービスアカウントには、 serviceAccounts.actAs 権限が与えられているロールを付与する必要があります。

name: Delivery

on:
  push:
    branches:
      - main

env:
  IMAGE: asia.gcr.io/${{ secrets.GCP_PROJECT_ID }}/covid19-status-of-medical-care-provision-in-japan

jobs:
  build:
    runs-on: ubuntu-latest
    steps:
      - name: checkout
        uses: actions/checkout@v2
      - name: Set outputs
        id: vars
        run: echo "::set-output name=sha_short::$(git rev-parse --short HEAD)"
      - name: gcp auth
        uses: google-github-actions/setup-gcloud@master
        with:
          project_id: ${{ secrets.GCP_PROJECT_ID }}
          service_account_key: ${{ secrets.GCP_SA_KEY }}
      - name: configure docker
        run: gcloud auth configure-docker --quiet
      - name: build image
        run: docker build . -t $IMAGE:${{ steps.vars.outputs.sha_short }}
      - name: push image
        run: docker push $IMAGE:${{ steps.vars.outputs.sha_short }}
      - name: deploy
        run: |
             gcloud run deploy \
               --project ${{ secrets.GCP_PROJECT_ID }} \
               --image $IMAGE:${{ steps.vars.outputs.sha_short }} \
               --region asia-northeast1 \
               --platform managed \
               --concurrency 1 \
               --max-instances 1 \
               --update-env-vars SPREADSHEET_ID=${{ secrets.SPREADSHEET_ID }} \
               --quiet \
               --service-account ${{ secrets.CLOUDRUN_SERVICE_ACCOUNT }} \
               covid19-status-of-medical-care-provision-in-japan

Cloud Scheduler

定期実行を行うために、CloudRunのURLを叩きます。CloudRunは非公開のAPIであるため、利用するサービスアカウントを指定します。(公式ドキュメントの手順)[https://cloud.google.com/run/docs/triggering/using-scheduler#create_job]通りに実装するだけで簡単に定期実行が可能になりました。

スプレッドシート

スプレットシートについてはデータの書き込み先としてだけ利用しており、はじめに、各シートを準備するくらいで、あとはいじらないようにしています。これにより、データの流れを明確にしています。

DataStudio

ここまでこれば、スプレッドシートにデータは投入されるようになるので、DataStudioで可視化するだけです。ここは腕の見せ所ですが、現状かなりごちゃごちゃしているので、もう少しスッキリさせたい気持ちがあります。

できてないこと

時間などの都合で、以下のようなことができてません。

• 過去のデータの蓄積・可視化
• ダッシュボード画像のtwitterへの自動投稿
• 表の列名などの整理・見直し
  • 少々イケてないところがあるのでおいおい直したいと思っています
• データフォーマットが変わったときの対応方法の検討
• スマホ対応

おわりに

定期的に実行するタスクも、CloudRunを使えば、ほとんどコストが掛からず運用できてとても良いです。簡単なバッチ処理に応用が効くシステム構成かなと思っています。また、厚生労働省が公開していたデータは、列名の設定は少々大変でしたが、それ以外の加工はほぼ必要なく、非常に扱いやすいデータフォーマットで助かりました。
今回、可視化した指標は、SLI/SLO/SLAとしても考えられるのではないかと思っています。「アラートが鳴っている状況」だと考えると、ソフトウェアエンジニアの私にとって、より自分ごとのように感じられます。一日も早い正常化を願っています。