[AWS][SwitchBot] SwitchBotのCO2センサデータをCloudWatchで監視
目的
子供の健康のため、CO2濃度が1000ppmを超えた時に検知したい、SwitchBotのCO2センサを買ったので、ここから情報取得してアラートをあげたい、というのが今回の目的です。SwitchBotアプリからアラートもあげることができるのだが（そしてそれが一番楽でまっとうだが）、自分でAmazon CloudWatchにメトリクスを登録しアラートをあげる仕組みを作ってみたいと考えていました。
利用デバイス
家では、この2つのデバイスを使ってます。前者の方がCO2を取得できるデバイスですが、単体ではSwitchBotのAPIから取得できず、後者のハブデバイスが必要になります。
室内のCO2濃度基準
今回、CO2濃度が1000ppm以上で通知としています。この基準に関しては、下記、SwitchBot社のブログで説明されていたり、厚生労働省の資料などに記載があります。
センサーで検知された二酸化炭素（CO2）濃度は、ppm（パーツ・パー・ミリオン）という単位で表記され、通常の室内は450～700ppm、許容できる二酸化炭素濃度は700～1,000ppm、そして1,000ppm以上の値が出ると眠気や不活発性などを感じやすくなると言われています。
1000ppm 程度の低濃度の二酸化炭素そのものによる労働生産性（意思決定能力や問題解決能力）への影響が示唆
室内の二酸化炭素濃度は全般的な室内空気の汚染度や換気の状況を評価する１つの指標としても用いられており、二酸化炭素濃度の基準値は 1000ppm 以下と定められている。
利用するSwitchBot API
SwitchBotのAPIは、下記のサイトの手順にしたがってTokenやSecretを取得することで、呼び出せるようになります。
どんなAPIがあるのかは下記ドキュメントがありますので参考にしてください。
なお今回使用するのは、下記2つのエンドポイントです。
https://api.switch-bot.com/v1.1/devices
https://api.switch-bot.com/v1.1/devices/{deviceId}/status
devicesは目的のデバイスのdeviceIdの取得のため、devices/{deviceId}/statusは実際のメトリクス取得のためになります。
今回は SwitchBot CO2センサー（温湿度計） が対象なのでdeviceTypeは MeterPro(CO2) になり、status APIでは、CO2濃度含む、下記の情報が取得できます。
AWS上構成図
下記のような構成とします。
流れとしては下記のような動きになります。
- EventBridgeで5分ごとにLambdaをキック
- LambdaはSwitchBotのSecret, TokenをSecret Managerから取得
- LambdaはPutMetricDataを呼び出し、CloudWatchにメトリクスを登録
- CloudWatch AlarmでSNS経由でメールにてユーザに通知
Lambdaの実装
エラー処理等は雑になってますが、下記で取得しています。ちょっと長いので折りたたんでおきます。
Secretの取得には、AWS Parameters and Secrets Lambda Extensionを使用しています。
実装コード
import os
import json
import urllib3
import hmac
import base64
import uuid
import time
import hashlib
import boto3
import datetime
switchbot_api_url = 'https://api.switch-bot.com/v1.1'
http = urllib3.PoolManager()
cloudwatch_client = boto3.client('cloudwatch')
NAMESPACE="switchbot/MeterPro(CO2)"
TARGET_METRICS=["temperature", "battery", "humidity", "CO2"]
DIMENSIONS=['deviceType', 'deviceId']
def genApiHeaders():
apiHeader = {}
headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
secrets_extension_endpoint = "http://localhost:2773/secretsmanager/get?secretId=switchbot/api-key"
response = http.request(
"GET",
secrets_extension_endpoint,
headers=headers
)
response_json = json.loads(response.data)
token = json.loads(response_json['SecretString'])['token']
secret = json.loads(response_json['SecretString'])['secret']
nonce = uuid.uuid4()
t = int(round(time.time() * 1000))
string_to_sign = '{}{}{}'.format(token, t, nonce)
string_to_sign = bytes(string_to_sign, 'utf-8')
secret = bytes(secret, 'utf-8')
sign = base64.b64encode(hmac.new(secret, msg=string_to_sign, digestmod=hashlib.sha256).digest())
print ('Authorization: {}'.format(token))
print ('t: {}'.format(t))
print ('sign: {}'.format(str(sign, 'utf-8')))
print ('nonce: {}'.format(nonce))
apiHeader['Authorization']=token
apiHeader['Content-Type']='application/json'
apiHeader['charset']='utf8'
apiHeader['t']=str(t)
apiHeader['sign']=str(sign, 'utf-8')
apiHeader['nonce']=str(nonce)
return apiHeader
def getDevices(apiHeaders):
switchbot_url = '{}/devices'.format(switchbot_api_url)
response = http.request(
'GET',
switchbot_url,
headers=apiHeaders
)
if (response.status != 200):
# TODO EXCEPTION
pass
devices_json = json.loads(response.data.decode('utf-8'))
print(devices_json)
return devices_json['body']['deviceList']
def getSpecificTypeOfDevices(devices, deviceType):
device_array = []
print(devices)
for device in devices:
if device['deviceType'] == deviceType:
device_array.append(device)
else :
pass
return device_array
def getMetricsFromMeterProCo2Device(device, apiHeaders):
switchbot_url = '{}/devices/{}/status'.format(switchbot_api_url, device['deviceId'])
response = http.request(
'GET',
switchbot_url,
headers=apiHeaders
)
if (response.status != 200):
# TODO EXCEPTION
pass
device_status_json = json.loads(response.data.decode('utf-8'))
print(device_status_json)
return device_status_json['body']
def getMetricsFromMeterProCo2Devices(devices, apiHeaders):
metrics_array = []
for device in devices:
status = getMetricsFromMeterProCo2Device(device, apiHeaders)
metrics_array.append(status)
print(metrics_array)
return metrics_array
def convertMetricsIntoCloudWatchMetricData(metrics):
metricDataArray = []
for metric in metrics:
dimensions = []
for k,v in metric.items():
if k in DIMENSIONS:
dimension = {}
dimension['Name'] = k
dimension['Value'] = v
dimensions.append(dimension)
for k,v in metric.items():
metricData = {}
if k in TARGET_METRICS:
metricData['MetricName'] = k
metricData['Value'] = v
metricData['Dimensions'] = dimensions
metricData['Timestamp'] = datetime.datetime.now()
metricDataArray.append(metricData)
return metricDataArray
def putMetricsOnCloudWatch(metrics):
metricData = convertMetricsIntoCloudWatchMetricData(metrics)
print(metricData)
cloudwatch_client.put_metric_data(
Namespace=NAMESPACE,
MetricData=metricData
)
def handler(event, context):
apiHeaders = genApiHeaders()
devices = getDevices(apiHeaders)
meterProCo2Devices = getSpecificTypeOfDevices(devices, "MeterPro(CO2)")
metrics = getMetricsFromMeterProCo2Devices(meterProCo2Devices, apiHeaders)
putMetricsOnCloudWatch(metrics)
return {
'statusCode': 200,
'body': {"message": "put metrics has been completed successfully."}
}
結果
Metrics表示
Amazon CloudWatchでMetricsを確認してみます。
CO2濃度を含むメトリクスがCloudWatch Metrics上で表示されている
AWS Lambdaで登録したメトリクスが正しくAmazon CloudWatchに登録されていることが確認できました。
Alarm発出
AlarmのE-mailが正しく発出されることも確認しました。
AWS Management Console上のAmazon CloudWatch Alarmでも、正しくIn Alarm状態となっています。
まとめ
SwitchBotのデータをCloudWatch上で表示することができました。CloudWatch Dashboardには共有機能もあるので、このメトリクスから自分でDashboardを作って、どこかに埋め込んで常に表示するなど今後どこかでやりたいと思っています。
Discussion