くろねこ。の実験室

毎日の運動量と睡眠状態をTwitterに垂れ流してモニタリングする。

Date: 2021/05/23 13:54

Category: none

はじめに

最近、このようなご時世である故に、在宅勤務で一歩も外に出ない日も珍しくない。
意識しないと運動不足になりかねない。最近、体の健康管理を目的に fitbit というスマートウォッチを利用している。
この fitbit は API が公開されており、スマートウォッチのアプリを作成したり、デバイスの情報を取得することができる。
そこで、運動状況と睡眠状態のモニタリングのため、fitbit で取得した情報を毎日ツイートすることにした。

fitbit とは...

fitbit とは、アメリカの fitbit.inc が販売している、スマートウォッチである。これを手首に装着することで運動中の心拍数や酸素濃度などを測定できる。このような情報から、運動による消費カロリー、心拍数の変化、睡眠の質の記録などが可能なものである。私が使用しているFitbit Versa 3 では、それらに加えて、GPS でワークアウトを管理できたり、将来的に Suica に対応し、非接触決済端末としても使えるようになるそうだ。

fitbit の API について...

fitbit は API でデバイスの情報を取得したり、fitbit 専用の独自アプリを開発ができるようになっている。この API を使えば、fitbit で取得されている 1 日の消費カロリーや睡眠時間などを取得できる。既に fitbit を利用しているのであれば Swagger UI 形式で、Fitbit Web API を試すことができる。この API は、OAuth2.0 での認可を要求し、認可を取得できれば、アクセストークンとリフレッシュトークンが取得できる。

ClientID などが記載されたページの下部にある OAuth 2.0 tutorial page で実際に試すことができる。リフレッシュトークンの期限が切れると再度認可取得からやる必要があり面倒だが、OAuth2.0 では、期限内であればリフレッシュトークンからアクセストークンを取得することができる。

どうやって Tweet するんだっけ？

fitbit API で取得した情報をツイートするためには、少なくとも、以下の 4 点をクリアしないといけない。

Twitter の developer 登録を済ませて、Twitter の API を叩くためのアクセストークン(無期限)を取得していないといけない。
fitbit の API のアクセストークンとリフレッシュトークン(期限がある)を取得する必要がある。
fitbit の API のアクセストークンを定期的に更新する仕組みがある必要がある。
定期的に fitbitAPI を叩いて取得した値を用いてメッセージを生成し、Twitter に投稿する仕組みが必要になる。

Twitter の developer 登録を済ませて、Twitter の API を叩くためのアクセストークン(無期限)を取得していないといけない。

に関しては、Twitter の Developer ポータルからアプリの登録とトークンの発行は可能である。ただし、Twitter の API を悪用する者が多いからか、作成するアプリの概要、どんなエンドポイントを叩くかなどを英語で作文し、Twitter の審査を受ける必要がある。私も頑張って英語で作りたいものを説明したが、もっと詳しく教えて欲しいとの旨のメールを受け取っている...

そのメールが Google 翻訳で自動的に日本語になっていたのか、返信を日本語で返したら再度同じメールをもらう、ということもやらかしていたので、アプリが承認されるまで時間がかかってしまった。

最終的に以下のように、ただ毎日 00:10 に決まったフォーマットのツイートをしたいだけなんだ、というメールを送りつけたことにより、アプリは承認された。

Post to your own (@ myblackcat7112) account once a day, the result of hitting the Fitbit API to check your exercise habits, formatted as follows:

Today's exercise from fitbit
Sitting time: 662 minutes
Light exercise time: 127 minutes
Active exercise time: 49 minutes
Time for strenuous exercise: 78 minutes
Today's steps: 8650 steps (5.893km)
Basal metabolism: 1402 kcal
Calories burned: 2727 kcal

https://twitter.com/myblackcat7112/status/1377597776662470657?s=20

I will post to my account, @myblackcat7112  at 00:10:00 everydays.
I want to use the API to post an auto-generated message to Twitter.

この承認を得る作業が、今回、最も難しかったと言っても過言じゃない笑

fitbit の API のアクセストークンとリフレッシュトークン(期限がある)を取得する必要がある。

に関しては、上述したOAuth 2.0 tutorial pageで簡単に取得できるので問題ない。

実際の認可取得画面のURLを生成するページ

詳細に関してはこちらの記事も参考になると思う。

Python で fitbit API から心拍数を取得してみよう！

fitbit の API のアクセストークンを定期的に更新する仕組みがある必要がある。

定期的に fitbitAPI を叩いて取得した値を用いてメッセージを生成し、Twitter に投稿する仕組みが必要になる。

に関しては少しコードを書く必要がある。今回は AWS の Lambda を用いて、Tweet することにした。

また、アクセストークンの更新は、AWS の Secret Manager に保存されているアクセストークン、リフレッシュトークンを、Tweet を実行する前のタイミングに更新する Lambda を実行することで解決することにした。

Secret Manager ?

AWS の Secret Manager は DB の認証情報や API のアクセストークンなどを安全に保管しておくことができるものである。

シークレットあたり $0.40 / 月

10,000 回の API コールあたり $0.05

とあるので、あまり多くを保存するのはコストが増えるだけではあるが、賢く使えば認証情報を安全に保管したりアクセストークンなどのローテション(自動更新)ができるものである。

今回は、TwitterAPI と FitbitAPI の API 情報を格納することにする。

使い方は簡単で、保存したい値の名前(key)と値(value)を保存するだけである。トークンなどのローテーションは、SecretManager の更新権限を、持ち実際に更新する処理が書かれている AWS Lambda を指定してあげることで可能になる。(一応設定してあるが、タイミングの問題かあまりうまくいっていないので、ここで設定している Lambda を定期実行するというださいことをやってしまっている...)

SecretManagerの画面

ローテーションの設定画面

SecretManager から Lambda を呼び出すのに、以下のような設定を画面から行なったがうまくいかなかった。(原因はいまだによくわからないが、権限が足らなかったのかもしれない？)

"Statement": [
    {
      "Sid": "SecretsManagerAccess",
      "Effect": "Allow",
      "Principal": {
        "Service": "secretsmanager.amazonaws.com"
      },
      "Action": "lambda:InvokeFunction",
      "Resource": "arn:aws:lambda:ap-northeast-1:XXXXXXX:function:fitbit-api-token-refresh"
    },

結局、AWS Secrets Manager のシークレットローテーションに関するトラブルシューティングのシークレットのローテーションを設定しようとすると「アクセス拒否」が発生するの部分に書かれていたコマンドを実行したら一発で解決した。

aws lambda add-permission --function-name ARN_of_lambda_function --principal secretsmanager.amazonaws.com --action lambda:InvokeFunction --statement-id SecretsManagerAccess

アクセストークンを更新するために実装した Lambda

1import json
2import requests
3from datetime import datetime, timezone, timedelta
4from SecretManager import SecretManager
5
6def lambda_handler(event, context):
7    sm = SecretManager('Fitbit')
8    secret = sm.get()
9    headers = {
10        'Authorization':'Basic '+secret['BASIC_TOKEN'],
11        'Content-Type':'application/x-www-form-urlencoded'
12    }
13    data = {
14        'grant_type':'refresh_token',
15        'refresh_token':secret['REFRESH_TOKEN']
16    }
17    response = requests.post(
18        'https://api.fitbit.com/oauth2/token',
19        headers = headers,
20        data = data
21    )
22    params = json.loads(response.content)
23    secrets = {
24        'BASIC_TOKEN': secret['BASIC_TOKEN'],
25        'CLIENT_ID': secret['CLIENT_ID'],
26        'CLIENT_SECRET': secret['CLIENT_SECRET'],
27        'ACCESS_TOKEN': params['access_token'],
28        'REFRESH_TOKEN': params['refresh_token'],
29        'UPDATED': datetime.now(timezone(timedelta(hours=9))).strftime('%Y/%m/%d %H:%M:%S'),
30    }
31    sm.update(secrets)
32
33    return {
34        'statusCode': 200,
35        'body': 'update!!'
36    }
37

Lambda でやっていることは、大きく分けて 2 つ。

SecretManager に保存されているトークンの操作(取得と更新)
fitbit API の認証エンドポイントにリフレッシュトークンを post して、新たなアクセストークン、リフレッシュトークンを取得する

SecretManager class は、SecretManager のサンプルコードをもとに処理を class にまとめただけのもので別に特筆するものではないが一応コードを残しておこうと思う。boto3という AWS 操作のための Python ライブラリを使用している。

1from botocore.exceptions import ClientError
2import base64
3import boto3
4import json
5
6class SecretManager:
7
8    def __init__(self, secret_name, region_name = 'ap-northeast-1'):
9        self.region_name = region_name
10        self.secret_name = secret_name
11        self.client = self.get_client()
12
13    # secret manager client
14    def get_client(self):
15        # Create a Secrets Manager client
16        session = boto3.session.Session()
17        client = session.client(
18            service_name = 'secretsmanager',
19            region_name = self.region_name
20        )
21        return client
22
23    # secret manager トークン更新
24    def update(self, update_json):
25        try:
26            return self.client.update_secret(
27                SecretId = self.secret_name,
28                SecretString = json.dumps(update_json)
29            )
30        except ClientError as e:
31            print(e)
32
33    # Secrets Managerのサンプルコードを参考に
34    def get(self):
35        try:
36            get_secret_value_response = self.client.get_secret_value(
37                SecretId=self.secret_name
38            )
39        except ClientError as e:
40            print(e)
41            if e.response['Error']['Code'] == 'DecryptionFailureException':
42                # Secrets Manager can't decrypt the protected secret text using the provided KMS key.
43                # Deal with the exception here, and/or rethrow at your discretion.
44                raise e
45            elif e.response['Error']['Code'] == 'InternalServiceErrorException':
46                # An error occurred on the server side.
47                # Deal with the exception here, and/or rethrow at your discretion.
48                raise e
49            elif e.response['Error']['Code'] == 'InvalidParameterException':
50                # You provided an invalid value for a parameter.
51                # Deal with the exception here, and/or rethrow at your discretion.
52                raise e
53            elif e.response['Error']['Code'] == 'InvalidRequestException':
54                # You provided a parameter value that is not valid for the current state of the resource.
55                # Deal with the exception here, and/or rethrow at your discretion.
56                raise e
57            elif e.response['Error']['Code'] == 'ResourceNotFoundException':
58                # We can't find the resource that you asked for.
59                # Deal with the exception here, and/or rethrow at your discretion.
60                raise e
61        else:
62            # Decrypts secret using the associated KMS CMK.
63            # Depending on whether the secret is a string or binary, one of these fields will be populated.
64            if 'SecretString' in get_secret_value_response:
65                secret = get_secret_value_response['SecretString']
66                return json.loads(secret)
67            decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
68            return json.loads(decoded_binary_secret)
69

実際に Tweet するのに実装した Lambda

1import json
2from Twitter import Twitter
3from Fitbit import Fitbit
4
5def lambda_handler(event, context):
6
7    results = {}
8    fitbit = Fitbit()
9    for resource in fitbit.get_resources():
10        data = fitbit.intraday_time_series(resource)
11        print(data)
12        results[list(data.keys())[0]] = list(data.values())[0][0]['value']
13    results['datetime'] = list(data.values())[0][0]['dateTime']
14
15    distances = float(results['activities-tracker-distance']) * 1.61
16
17    message =  f"本日({results['datetime']})の運動 from Fitbit\n\n"
18    message += f"座っていた時間: {results['activities-tracker-minutesSedentary']}分\n"
19    message += f"軽い運動の時間: {results['activities-tracker-minutesLightlyActive']}分\n"
20    message += f"アクティブな運動の時間: {results['activities-tracker-minutesFairlyActive']}分\n"
21    message += f"激しい運動の時間: {results['activities-tracker-minutesVeryActive']}分\n\n"
22    message += f"本日の歩数: {results['activities-tracker-steps']}歩 ({distances:.3f}km)\n\n"
23    message += f"消費カロリー: {results['activities-tracker-calories']}kcal ("
24    message += f"基礎代謝: {results['activities-caloriesBMR']}kcal)"
25    print(message)
26
27    sleep = fitbit.sleep()['summary']
28    sleep_message = f"本日({results['datetime']})の睡眠時間 from Fitbit\n\n"
29    sleep_message += f"睡眠時間: {sleep['totalMinutesAsleep']/60 :.3f}時間"
30    print(sleep_message)
31
32    # twitterへメッセージを投稿する
33    twitter = Twitter()
34    twitter.status_update(message)
35    twitter.status_update(sleep_message)
36
37    return {
38        'statusCode': 200,
39        'body': json.dumps(results)
40    }
41
42

今回 Fitbit API で取得する対象は、以下のものであり、これらのエンドポイントから情報を取得する部分は、fitbit の Python ライブラリを用いている。
Activity & Exercise Logsに定義されている resource を

intraday_time_series(resource, base_date='today', detail_level='1min', start_time=None, end_time=None)

に指定してあげると、その情報を 1 日のサマリーを取得してくれる。

ただし、これが実行されるのは日付が変わってからなので、実際には以下のように前日の日付で取得している。

1    # 前日のサマリーを取得したいので。
2    def intraday_time_series(self, resource):
3        yesterday = datetime.now(timezone(timedelta(hours=9))) - timedelta(days=1)
4        return self.client.intraday_time_series(resource, base_date = yesterday)
5

1    # 取得対象
2    RESOURCES = [
3        'activities/caloriesBMR',
4        'activities/tracker/calories',
5        'activities/tracker/steps',
6        'activities/tracker/distance',
7        'activities/tracker/minutesSedentary',
8        'activities/tracker/minutesLightlyActive',
9        'activities/tracker/minutesFairlyActive',
10        'activities/tracker/minutesVeryActive',
11        'activities/tracker/activityCalories',
12    ]
13

cloud watch に吐き出されている実際のログ。加工が必要なのは移動距離くらいでそれ以外はそのまま出力しても問題ない値になっている。

{'activities-caloriesBMR': [{'dateTime': '2021-05-23', 'value': '1560'}]}
{'activities-tracker-calories': [{'dateTime': '2021-05-23', 'value': '1988'}]}
{'activities-tracker-steps': [{'dateTime': '2021-05-23', 'value': '1781'}]}
{'activities-tracker-distance': [{'dateTime': '2021-05-23', 'value': '0.75363626392912'}]}
{'activities-tracker-minutesSedentary': [{'dateTime': '2021-05-23', 'value': '551'}]}
{'activities-tracker-minutesLightlyActive': [{'dateTime': '2021-05-23', 'value': '90'}]}
{'activities-tracker-minutesFairlyActive': [{'dateTime': '2021-05-23', 'value': '4'}]}
{'activities-tracker-minutesVeryActive': [{'dateTime': '2021-05-23', 'value': '7'}]}
{'activities-tracker-activityCalories': [{'dateTime': '2021-05-23', 'value': '441'}]}

上述したデータから生成された実際のメッセージ。

本日(2021-05-23)の運動 from Fitbit
座っていた時間: 551分
軽い運動の時間: 90分
アクティブな運動の時間: 4分
激しい運動の時間: 7分
本日の歩数: 1781歩 (1.213km)
消費カロリー: 1988kcal (基礎代謝: 1560kcal)

睡眠情報に関しては、sleep()メソッドを呼び出している。

本日(2021-05-23)の睡眠時間 from Fitbit
睡眠時間: 12.433時間

あとはこの Lambda を Event Bridge(Cloud watch Events)で、アクセストークンの更新が終わったであろう、00:20 に呼び出してあげれば晴れて以下のようなツイートがされるというわけである。

<blockquote class="twitter-tweet">本日(2021-05-23)の運動 from Fitbit 座っていた時間: 551分 軽い運動の時間: 90分 アクティブな運動の時間: 4分 激しい運動の時間: 7分 本日の歩数: 1781歩 (1.213km) 消費カロリー: 1988kcal (基礎代謝: 1560kcal)— Blackcat🌔 (@myblackcat7112) <a href="https://twitter.com/myblackcat7112/status/1396486255504805888?ref_src=twsrc%5Etfw">May 23, 2021</a></blockquote> <blockquote class="twitter-tweet">本日(2021-05-23)の睡眠時間 from Fitbit 睡眠時間: 12.433時間— Blackcat🌔 (@myblackcat7112) <a href="https://twitter.com/myblackcat7112/status/1396486256725430277?ref_src=twsrc%5Etfw">May 23, 2021</a></blockquote>

ここまでのコードは、ここにすべてまとめてあります。

まとめ

Fitbit + Lambda + SecretManager + Twitter で、健康状況を常に意識する環境を整えた。
(この報告を深夜 2 時に書いている時点で、睡眠不足は必至なのだが...)

今後は、この値の平均値などが一定の閾値を下回ったり、上回ったりした時にも通知が飛ばせるようになれば、アラートとしても使えるような気がする。SRE の勉強会で読んだ、入門監視を再度読み直して、生かしてみるのもいいのかもしれない。

まさき。です。PHPエンジニアをやってます。

自分の課題を技術で乗り越えるの好きかもしれないです。

フロントエンドは苦手ですが、少しでもできるようになれたらな、ということでNextJSでこのブログサイトを作りました。