毎日の運動量と睡眠状態を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で簡単に取得できるので問題ない。
詳細に関してはこちらの記事も参考になると思う。
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から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"><p lang="ja" dir="ltr">本日(2021-05-23)の運動 from Fitbit<br><br>座っていた時間: 551分<br>軽い運動の時間: 90分<br>アクティブな運動の時間: 4分<br>激しい運動の時間: 7分<br><br>本日の歩数: 1781歩 (1.213km)<br><br>消費カロリー: 1988kcal (基礎代謝: 1560kcal)</p>— Blackcat🌔 (@myblackcat7112) <a href="https://twitter.com/myblackcat7112/status/1396486255504805888?ref_src=twsrc%5Etfw">May 23, 2021</a></blockquote> <blockquote class="twitter-tweet"><p lang="ja" dir="ltr">本日(2021-05-23)の睡眠時間 from Fitbit<br><br>睡眠時間: 12.433時間</p>— 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でこのブログサイトを作りました。