Fitbit x GCP x Pythonで作る！活動量リアルタイム観測アプリの作成手順まとめ

はじめに

今回は、FitbitのデータをAPI経由で取得し、GoogleDataStudio上でビジュアライズするアプリを開発しました。Fitbitは、API経由で大変に詳細なデータを取得する事ができます。

通常ですと、専用のスマホアプリでライフログを確認するのですが、より細かなデータを見てみたかったので、今回は1時間に1回更新される、ダッシュボードを開発してみました。

動機

元々Fitbitは所有していたので、APIを使った開発には興味があったのですが、最近高齢化や孤独死などの問題がメディアでもよく取り沙汰されている事がきっかけになりました。

これまで以上に散り散りに生活するようになっていく中で、"自分の両親や、遠く離れた人が元気にしているか確認する用途で使えそう"と思い、今回の開発を行いました。

「老人 見守り iot」などでググると、この手の遠隔監視(？)機能を提供するサービス・製品は多く出てきますが、

• Fitbitの本体代のコストのみで実装可能！
• 歩数 / カロリー消費量 / 睡眠の質 など詳細な活動量を確認できる！
• Fitbitを腕に巻いてもらうだけで導入できる障壁の低さ!

などの点から、取っ掛かりとしては、結構ありかもしれないと思っています 笑

対象読者に含まれる方であれば、当記事を最初から最後まで読んで頂ければ、誰でも実装できるように書いているつもりではありますが、もし分からない点などあれば、お気軽にTwitterで絡んでください！

参考にさせて頂いた記事

• Pythonでfitbit APIから心拍数を取得してみよう！ (
  @fujit33
  )
  • FitbitAPIの環境設定について、そのまま手順を真似させて頂きました!

対象読者

• Python/GCPにある程度詳しいエンジニア
• IoTデバイスを使った開発に興味があるエンジニア
• Fitbit所有者で、何かやってみたいな〜と思っている人
• 遠くにいるあの人の健康状態を把握したいと思っている人

自分で実装する時に必要なもの

• Fitbit本体とアカウント
  • 持っていない人でも是非読んで行って下さい!
• GCPのプロジェクト
  • 無償の範囲に余裕で収まりますが、課金に関しては自己責任でお願い致しますmm

完成したもの

• Fitbitダッシュボードテンプレート
  • こちらのURLから飛べます。BigQuery上に対応するデータソースを配置すると、そのまま使用する事ができます。

• 指標の一覧
  • 色々表示していますが、これでもほんの一部です..! APIで取得可能な指標の一覧はこちらのAPIドキュメントを参照。

No.	指標名	説明
1	歩数	Fitbitを装着した状態で歩いた歩数。体感だと、結構多めにカウントされる印象がある。
2	運動した時間	文字通り運動した時間。激しい運動と軽い運動に分けて表示させているが、軽い方は↑に同じく多めにカウントされる..(以下略
3	消費カロリー	1日の消費カロリー。基礎代謝は身長/体重のデータから算出されるので、運動した分が上乗せされて計算される。
4	ベッドにいた時間	ベッドに滞在した時間。覚醒状態と睡眠状態を分けて表示しているが、ベッドにいて起きた後もベッドでダラダラしてた時間なんかも計測されている。
5	寝返り回数	寝返りを打った回数。気付かぬ内に結構...。
6	心拍数	文字通り心拍数。

• 更新の仕組み
  • CloudFunctionのスケジュール実行機能を使って、定期的にデータを取得->更新します。

実装 - 全体像

全体の概要は以下のようになっています。

• 最初に、厄介な事を考えずにデータ取得を行えるように、__python-fitbitを使用する__事に決めました。
• そして、スクリプトを定期実行したい + サーバーコストを掛けたくないという理由から、CloudFunctionを実行環境に採用。
• DataStudioを使いたかったので、データの保存先としてBigQueryを採用しました。

※ もっと手軽にやるならば、GASを使ってスプレッドシートに引っ張ってくるとかでも良さそうです。

• ステップとしては以下4つに分かれるので、順番に説明していきます。

STEP⓪ : FitbitAPIの利用登録をする

STEP① : CloudFunctionからAPIを叩いてデータ取得

STEP② : BigQueryに結果をテーブル保存

STEP③ : DataStudioからBigQueryを参照してビジュアライズ

実装 - ステップ別詳細

STEP⓪ : FitbitAPIの利用登録をする

• いきなり丸投げしていますが、参考記事で挙げた@fujit33さんの、Pythonでfitbit APIから心拍数を取得してみよう！のフローに従って、環境セットアップを行ってください。

• ClientID / Client Secret / Access Token / Refresh Tokenは、後で使用するので保存しておいて下さい。

STEP① : CloudFunctionからAPIを叩いてデータ取得

• ここからはひたすらPythonのスクリプトについて、ポイントを説明していきますが、最終的にCloudFunction上で動作するソースコードは以下になります。(個人開発なので動けば良いやの精神)
• もし、Fitbitをお持ちなら、STEP⓪の認証情報を入力すると動作するはずです。

get_fitbit_data.py

import fitbit
import pandas
import json
from datetime import datetime, timedelta, timezone

def build_fitbit_authed_client(CLIENT_ID,CLIENT_SECRET,ACCESS_TOKEN,REFRESH_TOKEN):
    # ID等の設定
    authed_client = fitbit.Fitbit(CLIENT_ID, CLIENT_SECRET
                             ,access_token=ACCESS_TOKEN, refresh_token=REFRESH_TOKEN)
    return authed_client

# 直近7日間の日付リストを作成する
def build_date_list():
    date_array = []    
    # タイムゾーンの生成
    JST = timezone(timedelta(hours=+9),'JST')
    for i in range(0,7):
        date = datetime.now(JST).date() - timedelta(days = i)
        date_array.append(str(date))
    return date_array

# 実行日の分リストを作成する
def build_minutes_list():
    minutes_array = []
    # タイムゾーンの生成
    JST = timezone(timedelta(hours=+9),'JST')    
    seed_timestamp = datetime.now(JST).replace(hour=0,minute=0,second=0,microsecond=0)         
    for minute in range(1,1440):
        minutes_array.append( (seed_timestamp + timedelta(minutes = minute)).strftime("%H:%M:%S"))
    return minutes_array

# 直近7日間のデータを日別で取得し、辞書を返却する(Activity系とSleep系の2つを含む)
def build_days_metrics_dict(authed_client,dates_list):
    days_result_dict = {}
    
    for date in dates_list:
        singleday_activity_metrics = []        
        
        # 該当日のActivity系の指標を取得
        activity_metrics = ['caloriesOut','steps','lightlyActiveMinutes','veryActiveMinutes']
        activity_response = authed_client.activities(date=date)
        for metrics_name in activity_metrics:
            try:
                singleday_activity_metrics.append(activity_response['summary'][metrics_name])
            except:
                singleday_activity_metrics.append(0)                

        # 該当日のSleep系の指標を取得
        sleep_metrics = ['timeInBed','minutesAwake','minutesAsleep','restlessCount']
        sleep_response = authed_client.sleep(date=date)
        
        for metrics_name in sleep_metrics:
            try:
                singleday_activity_metrics.append(authed_client.sleep(date=date)["sleep"][0][metrics_name])
            except:
                singleday_activity_metrics.append(0)

        # 該当日の指標を辞書に格納
        days_result_dict[date] = singleday_activity_metrics
        
    return days_result_dict

# 実行日のデータを分単位で取得し、辞書を返却する
def build_intraday_metrics_dict(authed_client,minutes_list):
    intraday_minutes_result_dict = {}
    JST = timezone(timedelta(hours=+9),'JST')    
    
    # intra_day APIで、実行日の①歩数 ②心拍数 ③消費カロリー を取得する
    resources_list = ['steps','heart','calories']

    per_minutes_steps = authed_client.intraday_time_series('activities/steps', base_date=str((datetime.now(JST) - timedelta(days = 25)).date()), detail_level='1min', start_time="0:00", end_time="23:59")
    per_minutes_heart = authed_client.intraday_time_series('activities/heart', base_date=str((datetime.now(JST) - timedelta(days = 25)).date()), detail_level='1min', start_time="0:00", end_time="23:59")    
    per_minutes_calories = authed_client.intraday_time_series('activities/calories', base_date=str((datetime.now(JST) - timedelta(days = 25)).date()), detail_level='1min', start_time="0:00", end_time="23:59")    
    
    # リクエストから、分単位のデータを取得し、辞書に整形する
    for minute in minutes_list:
        per_minute_result = []  

        #分刻みのStepsを配列に追加
        steps_values = [x['value'] for x in per_minutes_steps['activities-steps-intraday']['dataset'] if x['time'] == minute]
        step_value = steps_values[0] if len(steps_values) else ''
        per_minute_result.append(step_value)

        #分刻みのHeartを配列に追加
        heart_values = [x['value'] for x in per_minutes_heart['activities-heart-intraday']['dataset'] if x['time'] == minute]
        heart_value = heart_values[0] if len(heart_values) else ''
        per_minute_result.append(heart_value)

        #分刻みのCaloriesを配列に追加
        calories_values = [x['value'] for x in per_minutes_calories['activities-calories-intraday']['dataset'] if x['time'] == minute]
        calories_value = calories_values[0] if len(calories_values) else ''
        per_minute_result.append(calories_value)        

        intraday_minutes_result_dict[minute] = per_minute_result
    
    return intraday_minutes_result_dict

# DataFrameをBigQueryの任意のプロジェクト / 保存先にエクスポートする
def export_df_to_bq(df,project_id,dataset_name,table_name):
    df.to_gbq(dataset_name + '.' + table_name, project_id, if_exists = 'replace')
    
# DictをDataFrameに変換する
def convert_dict_to_dataframe(dic,column_names,index_name):
    converted_df = pandas.DataFrame.from_dict(dic,orient = 'index',columns = column_names).reset_index()
    return converted_df    

def get_fitbit_data(event, context):
    # 認証情報の用意
    ## FitbitAPI関連(★★自身の情報に変更します★★)
    CLIENT_ID =  "XXXXXXX"    
    CLIENT_SECRET  = "XXXXXXX"
    ACCESS_TOKEN =  "XXXXXXX"
    REFRESH_TOKEN =  "XXXXXXX"
    
    ## BigQuery関連(★★自身の情報に変更します★★)
    project_id = "XXXXXXX"
    dataset_name = "XXXXXXX"
    
    # 認証済みクライアントの作成
    authed_client = build_fitbit_authed_client(CLIENT_ID,CLIENT_SECRET,ACCESS_TOKEN,REFRESH_TOKEN)
    
    # 時系列リストの生成
    dates_list = build_date_list()
    minutes_list = build_minutes_list()    

    # 日別データの取得 -> DataFrameに変換
    ## データの取得
    days_result_dict = build_days_metrics_dict(authed_client,dates_list)
    
    ## DataFrameに変換
    days_clumns_name = ['caloriesOut','steps','lightlyActiveMinutes','veryActiveMinutes','timeInBed','minutesAwake','minutesAsleep','restlessCount']    
    days_result_df = convert_dict_to_dataframe(days_result_dict,days_clumns_name,'date')
    print('--------日別の数値--------')
    print(days_result_df)
    print('------------------------')
    # BigQueryに書き込み
    days_table_name = 'days_metrics'
    export_df_to_bq(days_result_df,project_id,dataset_name,days_table_name)    
    
    #DataFrameに変換    
    minute_result_dict = build_intraday_metrics_dict(authed_client,minutes_list)
    minute_clumns_name = ['steps','heart','calories']   
    minute_result_df = convert_dict_to_dataframe(minute_result_dict,minute_clumns_name,'minute')    
    print('--------時間別の数値--------')
    print(minute_result_df)
    print('------------------------')
    # BigQueryに書き込み    
    minute_table_name = 'minutes_metrics'
    export_df_to_bq(minute_result_df,project_id,dataset_name,minute_table_name)    

PythonからAPIを叩く方法をざっくり

• python-fitbitを使用してデータを取得します。今回は、日単位のトレンドと、分単位のトレンドの2種類欲しかったので、それぞれメソッドを使い分けています。
• 取得するデータの種類によって、微妙にメソッドの仕様が異なりますが、ドキュメントを読み解きながら実装していきます。

.setup_fitbit_client.py

import fitbit

# 認証済みクライアントの生成
def build_fitbit_authed_client(CLIENT_ID,CLIENT_SECRET,ACCESS_TOKEN,REFRESH_TOKEN):
    authed_client = fitbit.Fitbit(CLIENT_ID, CLIENT_SECRET
                             ,access_token=ACCESS_TOKEN, refresh_token=REFRESH_TOKEN)
    return authed_client

## 認証情報の設定
CLIENT_ID =  "XXXXXXX"    
CLIENT_SECRET  = "XXXXXXX"
ACCESS_TOKEN =  "XXXXXXX"
REFRESH_TOKEN =  "XXXXXXX"

authed_client = build_fitbit_authed_client(CLIENT_ID,CLIENT_SECRET,ACCESS_TOKEN,REFRESH_TOKEN)

'''
取得パターン①: intraday_time_series()で特定の日付のデータを指定の粒度で取得
↓は実行日の歩数を0:00-23:59の間で1分置きに取得
'''
today_steps = authed_client.intraday_time_series('activities/steps', base_date=str((datetime.now()).date()), detail_level='1min', start_time="0:00", end_time="23:59")

'''
主なパターン②: sleep()で特定の日付の睡眠データを取得
↓は実行日の睡眠データを取得
'''
today_sleep = authed_client.sleep(date=str((datetime.now()).date()))

'''
主なパターン③: activities()特定の日付の行動データを取得
↓は実行日の活動量データを取得
'''
today_activities = authed_client.activities(date=str((datetime.now()).date()))

• 今回、単純に"日別の歩数/睡眠時間を取得する"メソッドがパッと見つけられなかったので、トレンドデータに関しては、authed_client.activities(date=date)を日付をズラしながら複数リクエストするという力技で実装しました..w
  • __1ユーザー当たり1Hに150リクエストまで__というAPIの呼び出し制限があるので、開発中に何度か引っ掛かりました。

Pub/SubトリガーのCloudFunctionを設定

• 上記のメソッドを用いて、データを取得するスクリプトをFunctionsにデプロイします。今回は、CloudSchedulerを用いて定期実行したいので、Pub/Subトリガーに設定します。
• requirement.txtには、pandas /pandas-gbq / fitbitを記述します。

requirement.txt

# Function dependencies, for example:
# package>=version
fitbit == 0.3.1
pandas == 0.24.2
pandas-gbq == 0.10.0

• CloudFunctionの定期実行は先日、Pub/Subトピックをわざわざ設定しなくても、実行できるように機能アップデートがあったばかりですが、今回は以下のようにPub/SubトピックをCloudSchedulerで定期パブリッシュする設定を行いました。

STEP② : BigQueryに結果をテーブル保存

• FitbitAPIのレスポンスは、dictionaryで返却されるので、DataFrameに変換後、今回はpandas-gbqを使って、BigQueryのテーブルに投入しています。

export_dataframe_to_bigquery.py

# DictをDataFrameに変換する
def convert_dict_to_dataframe(dic,column_names,index_name):
    converted_df = pandas.DataFrame.from_dict(dic,orient = 'index',columns = column_names).reset_index()
    return converted_df

# DataFrameをBigQueryの任意のプロジェクト / 保存先にエクスポートする
def export_df_to_bq(df,project_id,dataset_name,table_name):
    response = df.to_gbq(dataset_name + '.' + table_name, project_id, if_exists = 'replace')
    return response

# Fitbitから取得したデータは、dictionary型で返却される。
hogehoge_dict = authed_client.get_fitbit_data()

# DataFrameに変換
df = convert_dict_to_dataframe(hogehoge_dict,['aaa','bbb','ccc'],'daily')

# pands-gbqでDataFrameをBigQueyのテーブルにエクスポート
response = export_df_to_bq(df,`project_name`,`dataset_name`,`table_name`)

↓ authed_client.activities()などで取得したデータが最終的に、BigQueryに反映されます。

STEP③ : DataStudioからBigQueryを参照してビジュアライズ

- ここまでで①/②が完了したので、GoogleDataStudioを用いたビジュアライズに進みます。 - BigQueryテーブルを参照先として選択できるので、先ほど作成した`days_metrics(過去日次データ)`と`minutes_metrics(当日分刻みデータ)`を選択。 - 良くあるBIツールのノリでガシガシビジュアルを作っていきます。

(色々説明をすっ飛ばしている気がしないでもないが)完成!🎉🎉

積み残し

• より少ないAPI呼び出しで時系列表示する方法を調べる
• 急な変化があった際に、Slackに通知を飛ばすような機能もあるとより安心
• コード部分はPulumiやTerraformで構成管理したい
• 今は自分用になっているので、心配な人にFitbit渡す

おわり

思いつきで始めた開発ですが、短時間で実際に動くものを作ることができました。Fitbitは本当に可能性のあるデバイスだと感じました！

もし、少しでも役に立ったようでしたら、いいね宜しくお願いします！