Cloud FunctionsでSwitchbotのWebhookを受け取る
この記事では、Google CloudのCloud Functionsを活用してSwitchbotのデータをWebhookで受け取る方法を紹介します。
はじめに
Web APIを経由してSwitchbotのデバイス情報を取得したい場合、「Get device status」のAPIを利用できます。
このAPIを一定間隔でコールすれば、温湿度計などの値を継続的に取得できます。この方法を利用して、温湿度計のデータを可視化する仕組みを以下の記事で紹介しています。
Swicthbot APIには、1日あたり10,000回（1時間あたりだと約400回）のリクエスト上限があります。そのため、取得するデバイスの数が多い場合はこの上限に引っかかる可能性があります。
デバイスの情報を取得するもうひとつの手段として、Webhookの仕組みが用意されています。Switchbot APIからのPOSTリクエストを受け付けるサーバーを自分で用意することで、そのサーバーでデータを随時受信できるようになります。
この記事では、受信サーバーとしてGoogle CloudのCloud Functionsを利用する方法を紹介します。
作成したソースコードは以下のリポジトリに保存しています。
環境
この記事では、Switchbot APIとGoogle Cloudを利用準備が完了していることを前提として説明します。くわしい手順は筆者の過去記事を参照してください。
- Switchbot APIの利用方法
- Cloud Functionsの利用方法
Webhookの設定
Switchbot APIのWebhook機能を利用するためには、API経由でON/OFF設定を行う必要があります。具体的な手順は公式のAPI仕様書に記載されています。
今回はPythonスクリプトでAPIを操作しました。Web APIが操作できれば、Postmanなど他の手段で操作しても問題ありません。
import argparse
import base64
import hashlib
import hmac
import os
import time
import requests
from dotenv import load_dotenv
load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))
API_BASE_URL = "https://api.switch-bot.com/v1.1"
ACCESS_TOKEN: str = os.environ["SWITCHBOT_ACCESS_TOKEN"]
SECRET: str = os.environ["SWITCHBOT_SECRET"]
WEBHOOK_URL: str = os.environ["WEBHOOK_URL"]
def __generate_request_headers() -> dict:
"""SWITCH BOT APIのリクエストヘッダーを生成する"""
nonce = ""
t = str(round(time.time() * 1000))
string_to_sign = "{}{}{}".format(ACCESS_TOKEN, t, nonce)
string_to_sign_b = bytes(string_to_sign, "utf-8")
secret_b = bytes(SECRET, "utf-8")
sign = base64.b64encode(
hmac.new(secret_b, msg=string_to_sign_b, digestmod=hashlib.sha256).digest()
)
return {
"Content-Type": "application/json",
"Authorization": ACCESS_TOKEN,
"t": t,
"sign": sign,
"nonce": nonce,
}
def post_webhook(path: str, body: dict) -> str:
url = f"{API_BASE_URL}/webhook/{path}"
response = requests.post(url, headers=__generate_request_headers(), json=body)
return response.json()
def setup_webhook(webhook_url: str):
"""
Configure webhook
Configure the url that all the webhook events will be sent to
"""
body = {
"action": "setupWebhook",
"url": webhook_url,
"deviceList": "ALL",
}
return post_webhook("setupWebhook", body)
def query_webhook():
"""
Get webhook configuration
Get the current configuration info of the webhook
"""
body = {"action": "queryUrl"}
return post_webhook("queryWebhook", body)
def delete_webhook(webhook_url: str):
"""
Delete webhook
Delete the configuration of the webhook
"""
body = {"action": "deleteWebhook", "url": webhook_url}
return post_webhook("deleteWebhook", body)
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument(
"function", choices=["setup", "query", "delete"], help="Function to call"
)
args = parser.parse_args()
if args.function == "setup":
print(setup_webhook(WEBHOOK_URL))
elif args.function == "query":
print(query_webhook())
elif args.function == "delete":
print(delete_webhook(WEBHOOK_URL))
このスクリプトでは、「setupWebhook」「queryWebhook」「deleteWebhook」の3つのAPIをコマンドラインで利用できるようにしています。APIキーや送信先URLは環境変数で入力します。
以下のようにして、送信先URLのセットアップや削除を実行できます。（今回はPythonプロジェクト管理にuvを利用しているため、uv runでスクリプトを実行しています）
❯ uv run webhook.py setup
{'statusCode': 100, 'body': {}, 'message': 'success'}
❯ uv run webhook.py query
{'statusCode': 100, 'body': {'urls': ['https://ADDRESS/test-webhook']}, 'message': 'success'}
❯ uv run webhook.py delete
{'statusCode': 100, 'body': {}, 'message': 'success'}
送信先のURLは、この後の手順で作成したものを登録します。URLは1つしかセットできないようなので、ここでダミーのURLをセットした場合は、deleteで削除しておいてください。
Cloud Functionsの作成
次に、Cloud Functionsを使用して、SwitchbotのWebhookを受け取るサーバーレスなエンドポイントを作成する手順を説明します。
ここでは、Google Cloudのアカウント作成やCloud Functionsの基本的は利用設定に関する説明は割愛します。
関数の作成
Google Cloudのコンソール画面を開き、Cloud Functionsのセクションに移動し、「関数を作成」をクリックします。
関数の名前を入力し、トリガータイプとして「HTTPS」を選択します。
認証は「未認証の呼び出しを許可」を選択します。これで関数が公開APIとなり、Switchbotからのデータを受信できるようになります。
Cloud Functionsの設定画面（1/2）
コードのデプロイ
次に「コード」画面の設定を行います。ランタイムはPython 3.xxを選択します。
今回はSwitchbotの情報を受信することだけを試したいので、ソースコードはサンプルのコードを流用しました。Switchbot APIがPOSTしたJsonの中身をそのままログに出力します。
import functions_framework
@functions_framework.http
def hello_http(request):
request_json = request.get_json(silent=True)
request_args = request.args
print(request_json)
print(request_args)
return "200"
インラインエディタで上記のソースコードを入力し、「デプロイ」を実行します。
Cloud Functionsの設定画面（2/2）
デプロイが完了したら、関数のコンソール画面に表示されているURLを控えておきます。
データの受信
最後に、Webhookの受信URLを設定します。
Webhook設定用のPythonスクリプトの環境変数に先ほど取得したURLを記載し、setupを実行します。queryで正しく設定されたことを確認できます。
❯ uv run webhook.py setup
{'statusCode': 100, 'body': {}, 'message': 'success'}
❯ uv run webhook.py query
{'statusCode': 100, 'body': {'urls': ['作成した関数のURL']}, 'message': 'success'}
この状態でしばらく待っていると、データを受信できていることをログから確認できます。
Cloud Functionsのログ
今回はWebhook機能を試したかったので、取得したデータはログ出力するだけにしました。
データをストレージに保存しておけば、BigQuery→Lookerで可視化することもできそうです。機会があれば実装してみます。
クローズ対応について
Cloud Functionsを利用しているため、今回の設定を放置していると課金が発生する可能性があります。
この機能を使わない場合は、作成した関数を削除し、Switchbot APIからのデータ送信設定も削除しておいてください。
まとめ
このブログでは、Google Cloud の Cloud Functionsを使用してSwitchbotのWebhookを受け取る方法について解説しました。
この機能を使う場合はサーバーを用意する必要があるため、やや敷居が高い印象でした。しかしクラウドサービスを活用すれば、思ったよりシンプルに実装できることがわかりました。
ただ、APIの利用制限にかからず、データ取得のリアルタイム性を求めない場合は、シンプルにGETのAPIを活用した方が簡単かもしれません。データ取得のユースケースに応じて選択していきたいところですね。
Discussion