季節の変わり目ということで、自宅の気温と湿度をCloudWatchに保存したくなりました。

そこで最小限の労力でPythonを書いて、Switchbot ハブ2の気温と湿度（とついでに照度）をCloudWatchに送信しようと思います。

さらにEventBridgeとLambdaで5分ごとに定期実行します。

GitHubリポジトリ

全てのソースコードを下記で公開しています。

#1 Switchbot API用のトークンを取得する

まずはコーディング前の準備です。

公式サポートページの手順で「トークン」と「クライアントシークレット」を取得します。Switchbotアプリでバージョンを10回タップすると現れます。

#2 Switchbot Hub2のデバイスIDを取得する

SwitchbotアプリでHub2を選び、「設定」→「デバイス情報」で「BLE MAC」の値を確認します。そこからコロンを除いた値がデバイスIDです。

例: A1:BC:23:D4:5E:F6 → A1BC23D45EF6

▼参考

#3 AWS Secrets Managerにトークン類を保存する

ソースコードにトークン類を書くのはよくないので、AWS Secrets Managerに保存します。

AWS マネジメントコンソール → Secrets Manager → Store a new secret

シークレット名: switchbot

Key	説明
`token`	#1で取得したトークン
`secret`	#1で取得したクライアントシークレット
`device_id`	#2で取得したデバイスID

#4 Switchbotからデータを取得する

ここからコーディングです。この章では次のことをします。

Secrets Managerからトークン類を取得する
1のトークン類を使ってSwitchbot APIを叩き、データを取得する
手元で実行してみる

#4-1 Secrets Managerからトークン類を取得する

公式ドキュメントに従ってSDKを使います。

token, secret, device_idを取得する関数を雑に作りました。キャッシュの設定はデフォルトのままです。

一部抜粋 app/switchbot.py

def _get_secrets():

  client = botocore.session.get_session().create_client('secretsmanager')
  cache_config = SecretCacheConfig()
  cache = SecretCache( config = cache_config, client = client)

  secret_string = cache.get_secret_string('switchbot')
  secret_json = json.loads(secret_string)
  device_id = secret_json['device_id']
  token = secret_json['token']
  secret = secret_json['secret']

  return device_id, token, secret

#4-2 Switchbot APIを叩いてHub2の気温・湿度・照度を取得する

執筆時点でSwitchbot APIはv1.0とv1.1の両方が使えますが、認証がよりセキュアになったv1.1を使いましょう。

READMEの「How to Sign?」の章にPython3のサンプルがあり、コピペするだけで動くので難しくはありません。

今回使うAPIは GET https://api.switch-bot.com/v1.1/devices/{device_id}/status です。

一部抜粋 app/switchbot.py

def get_data():

  device_id, token, secret = _get_secrets()

  url = f'https://api.switch-bot.com/v1.1/devices/{device_id}/status'

  apiHeader = {}

  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())

  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)

  response = requests.get(url, headers=apiHeader)
  response_body = response.json()['body']

  return response_body

ここまでのコードは app/switchbot.py (GitHub) にまとめました。

#4-3 手元で実行してみる

先にターミナルにAWSの認証情報を設定します。

もっとも手っ取り早くIAM Userを使う例は下記です。

export AWS_ACCESS_KEY_ID=AKIAIOSFODNN7EXAMPLE
export AWS_SECRET_ACCESS_KEY=wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY
export AWS_DEFAULT_REGION=ap-northeast-1 # 東京リージョン

さっそく実行してみます。

python app/switchbot.py                 
{'deviceId': 'XXX', 'deviceType': 'Hub 2', 'hubDeviceId': 'C3CC01A04BF3', 'humidity': 73, 'temperature': 28.2, 'lightLevel': 1, 'version': 'V1.0-0.8'}

無事に気温・湿度・照度が取れました！

「aws_secretsmanager_cachingがないよ」などと出たらpip installしてください。

#5 CloudWatchにカスタムメトリクスを送信する

次に、気温・湿度・照度を引数で受け取ってCloudWatchに送る関数を作ります。

ベースにした公式ドキュメントはこちらです。

一部抜粋 app/cloudwatch.py

def post_data(temperature, humidity, light_level):

  namespace = 'Switchbot'
  dimentions = [
    {
      'Name': 'DeviceName',
      'Value': 'hub2'
    }
  ]
  metric_data = [
    {
      'MetricName': 'Temperature',
      'Value': temperature,
      'Unit': 'Count',
      'Dimensions': dimentions
    },
    {
      'MetricName': 'Humidity',
      'Value': humidity,
      'Unit': 'Percent',
      'Dimensions': dimentions
    },
    {
      'MetricName': 'LightLevel',
      'Value': light_level,
      'Unit': 'Count',
      'Dimensions': dimentions
    },
  ]

  cloudwatch = boto3.client('cloudwatch')

  cloudwatch.put_metric_data(Namespace = namespace, MetricData = metric_data)

この関数を app/cloudwatch.py (GitHub) のようにまとめ、実行してみます。

python app/cloudwatch.py
success

AWS マネジメントコンソール → CloudWatch → All Metrics を開き、送信したメトリクスを探します。ちゃんと送信されてますね！

#6 lambda_handlerを作る

あとは#4と#5を繋げる app/main.py (GitHub) を作ります。

このあとLambdaで動かすときに、関数の引数でevent, contextを受け取る必要があるので、今のうちに入れておきます。

app/main.py

import cloudwatch
import switchbot

def lambda_handler(event, context):
  data = switchbot.get_data()
  temperature = data['temperature']
  humidity = data['humidity']
  light_level = data['lightLevel']
  cloudwatch.post_data(temperature, humidity, light_level)

if __name__ == "__main__":
  lambda_handler(None, None)
  print('success')

実行しましょう。

python app/main.py      
success

コーディングはここまでです。3つのファイルが完成しました。

.
└── app
    ├── cloudwatch.py
    ├── main.py
    └── switchbot.py

#7 Lambdaで動かす

この章の手順を一発で実行するスクリプトをリポジトリに置いています。お急ぎの方はご利用ください。

./zip.sh
./create_aws_resources.sh

#7-1 zip化する

作ったファイルをzip化します。あとでこれをLambda関数にします。

zip -j function.zip app/*.py

次に外部ライブラリをzip化します。boto3は最初からLambdaの実行環境に入っているので、requestsとaws_secretsmanager_cachingを-tでディレクトリを指定してpip installします。あとでこれをLambdaレイヤーにします。

pip install requests aws_secretsmanager_caching -t python
zip -r layer.zip python

▼ここから余談

なお、Lambda関数とLambdaレイヤーでzipコマンドを変えているのは意図的です。

今回Lambda関数はディレクトリを作る必要がないので、zipの中身は次のようにしています。

.
├── cloudwatch.py
├── main.py
└── switchbot.py

一方Lambdaレイヤーのzipはpythonディレクトリを含めています。

.
└── python
    ├── requests
    │   ├── __init__.py
    │   └── ほか中身たくさん
    └── ほかにも pip install で入ったものがたくさん

これはLambdaレイヤーの仕様でディレクトリ名が決まっているからです。

https://docs.aws.amazon.com/ja_jp/lambda/latest/dg/packaging-layers.html

▲余談ここまで

#7-2 IAM Roleを作る

Lambda関数につける実行ロールを作ります。

AWS マネジメントコンソール → IAM → Roles → Create Role

IAM Role名: switchbot-lambda-execution-role

Trust PolicyはLambdaだけで使えるように書きます。

assume-role-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

インラインのPermission Policyを次の内容で作ります。Secrets managerのREAD類とcloudwatch:PutMetricDataに加え、LambdaがログをCloudWatch Logsに送れるように権限を付けています。

permission-policy.json

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "secretsmanager:GetResourcePolicy",
        "secretsmanager:GetSecretValue",
        "secretsmanager:DescribeSecret",
        "secretsmanager:ListSecretVersionIds",
        "cloudwatch:PutMetricData",
        "logs:CreateLogGroup",
        "logs:CreateLogStream",
        "logs:PutLogEvents"
      ],
      "Resource": "*"
    }
  ]
}

#7-3 Lambdaレイヤーを作る

AWS マネジメントコンソール → Lambda → Layers → Create Layer

レイヤー名: switchbot

先ほど作ったlayer.zipをアップロードします。x86_64とPython 3.11にチェックを入れてレイヤーを作成しましょう。

#7-4 Lambda関数を作る

AWS マネジメントコンソール → Lambda → Functions → Create Function

関数名: switchbot

この画面ではzipをアップロードできないので、いったん空の関数を作ります。実行ロールは先ほど作ったIAM Roleを指定します。

作成後の画面でfunction.zipをアップロードします。

無事にアップロードされました🎉 続けて同じ画面の下の方でHandlerとレイヤーを設定します。

main.py のlambda_handlerを実行したいのでmain.lambda_handlerにします。

先ほど作ったレイヤーを指定します。

一度テスト実行してみましょう。エディタの上にある「Test」ボタンをクリックし、出てきたポップアップで何も編集せずに「Invoke」を押してみます。

すると関数がタイムアウトしてしまいました。

{
  "errorMessage": "2023-09-18T04:47:41.465Z {中略} Task timed out after 3.04 seconds"
}

そこでタイムアウトを15秒に伸ばします。ついでにメモリも256MBに増やしておきます。

再度関数を実行すると、今度は正常終了しました！

REPORT RequestId: {中略}	Duration: 2439.00 ms	Billed Duration: 2439 ms	Memory Size: 256 MB	Max Memory Used: 102 MB	Init Duration: 714.18 ms

上記ログのMax Memory Usedで最大メモリ使用量を確認できます。何回か実行してみたところ、この関数のメモリ使用量はおおむね100MB前後でした。余裕をもってメモリの設定は256MBのままにしておきます。

#7-5 EventBridge Ruleを作る

最後にこの関数を定期実行します。関数の画面で「Add trigger」をクリックし...

Eventbridge Ruleを選んで...

cron(0/5 * * * ? *)でルールを作成しましょう。

なお、このルールからの実行を許可する Resource-based policy は自動で追加されているので気にしなくて大丈夫です。

これで5分ごとに実行されるようになりました。動かない場合はCloudWatch Logsに出力されたログを確認してみてください。

結果

自宅の気温と湿度がCloudWatchに保存されている〜〜〜！🎉

Switchbotアプリでも「気温がN℃以上になったら」という条件は作れますが、このメトリクスを使ってCloudWatchアラームを作ることで、より高度な条件を実現できそうです。

最後までお読みいただきありがとうございました。よろしければ「いいね」もお願いします！

【Python】Switchbot Hub2の気温・湿度をCloudWatchに定期的に送信する【AWS Lamba】