【SwitchBot】温湿度計のデータをPython × InfluxDB × Grafanaでダッシュボード化する

概要

この記事では、SwitchBot温湿度計のデータをAPI経由で取得し、時系列DBのInfluxDBに保存した上で、Grafanaでダッシュボード化する方法を解説します。

Grafanaで作成したダッシュボード

SwitchBot温湿度計について

SwitchBot温湿度計は、温度と湿度を計測するセンサーです。温度と湿度をデバイスの画面で確認する以外にも、他のSwitchBotデバイスと連携し、「温度が低下したら暖房をつける」といったトリガー起動に活用できます。

2023年3月現在は2種類の温湿度計が販売されており、この記事では両方に対応しています。「プラス」の方が液晶を見やすいですが、とくにこだわりがなければ通常版でも十分だと思います。

https://amzn.to/3laBpvZ
https://amzn.to/3YFTQXa

また、2023年3月に発売された「Hub2」でも温湿度情報をAPI経由で取得できました。
https://amzn.to/3KntsNY

温湿度計から送信されたデータはアプリ上で閲覧できます。これだけでも十分な情報が確認できますが、表示形式が限定されるのと、1つのデバイスの情報しか一度に閲覧できないという制約があります。

スマホアプリの表示例

そこで今回は、API経由で生データを収集することで、自分の好きな形式でダッシュボード化することを目指します。

InfluxDBとGrafanaについて

データの可視化にはOSSのInfluxDBとGrafanaを利用します。

InfluxDB
InfluxDBは、時系列データを保存するためのオープンソースのデータベースです。高速なデータ書き込みと柔軟なクエリ言語を備えており、IoT、監視、ログなどの用途に適しています。後述するGrafanaのデータソースとして指定できます。ちなみに、InfluxDBそのものにも可視化ツールが備わっています。

InfluxDB

Grafana
Grafanaは、データの可視化を行うためのオープンソースのツールです。InfluxDBやPrometheusなどのデータソースからデータを取得し、ダッシュボードを作成できます。

Grafana

Grafanaには以下のような特徴があります。

• 今回利用するInflucDB以外にも、さまざまなデータソースに対応している
• 線グラフやヒートマップなど、豊富な種類のグラフが用意されている
• ダッシュボードのカスタマイズが簡単かつ柔軟に実施できる
• さまざまな条件でアラート機能を実装できる

さらに、Grafanaコミュニティから多数のプラグインやパネルが提供されており、自由度の高いカスタマイズが可能です。

システム構成

この記事で紹介するシステムの構成は、以下のようなイメージです。3つのツールを組み合わせているので、Dockerを活用して、環境構築が一発で完了するようにしました。

システム構成イメージ

---

Dockerで環境を構築する

ここからは具体的な実装方法を紹介します。ソースコードは以下のリポジトリで公開しています。

https://github.com/tanny-pm/switchbot-dashboard

はじめに、Dockerで実行環境を構築します。

Dockerを設定する

Pythonスクリプトを実行するためのイメージとして、以下のようなDockerfileを用意します。RUN python switchbot.pyの箇所は、あとで説明するPythonスクリプトを実行するためのものです。

Dockerfile

FROM python:3.11

WORKDIR /app

COPY ./requirements.txt /app/requirements.txt
RUN pip install -r requirements.txt

COPY ./app /app
RUN python switchbot.py

ENTRYPOINT [ "python", "main.py" ]

次に、各コンテナを起動するようにdocker-compose.ymlを作成します。

docker-compose.yml

version: "3"

services:

  python:
    build:
      context: .
      dockerfile: Dockerfile
    tty: true

  influxdb:
    image: influxdb:2.6.1
    ports:
      - "8086:8086"
    volumes:
      - ./docker/influxdb/data:/var/lib/influxdb2
      - ./docker/influxdb/config:/etc/influxdb2
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=user
      - DOCKER_INFLUXDB_INIT_PASSWORD=password
      - DOCKER_INFLUXDB_INIT_ORG=org
      - DOCKER_INFLUXDB_INIT_BUCKET=switchbot

  grafana:
    image: grafana/grafana-oss:9.4.1
    ports:
      - "3000:3000"
    volumes:
      - ./docker/grafana/data:/var/lib/grafana
    depends_on:
      - influxdb

InfluxDBとGrafanaのデータはdockerディレクトリ内に保存されます。

ここでDOCKER_INFLUXDB_INIT_XXXにInflux DBの設定を記載しておくことで、起動後の初期設定が不要になります。（パスワードなどは必要に応じて変更してください）

Dockerを初回起動する

docker-compose up -dコマンドでDockerを起動します。以下のURLからInfluxDBとGrafanaにアクセスできれば成功です。細かい設定方法は後述します。

• InfluxDB：http://localhost:8086/
• Grafana：http://localhost:3000/

環境変数を登録する

.envファイルに環境変数を登録します。Switchbotのtokenとsecretの取得方法の詳細は過去の記事を参照してください。

.env

SWITCHBOT_ACCESS_TOKEN=
SWITCHBOT_SECRET=
INFLUXDB_TOKEN=

なお、INFLUXDB_TOKENは、dockerの起動後に/docker/influxdb/config/influx-configsへ出力されています。

influx-configs

[default]
  url = "http://localhost:8086"
  token = "YOUR_API_TOKRN"
  org = "org"
  active = true

環境変数を登録した後にdockerを再起動すると、準備は完了です。

このコンテナを起動すると、自動的にデータの収集を開始します。データの収集を停止する場合は、docker-compose downコマンドを実行してください。

PythonでSwitchBotのデータを取得する

SwitchBotのAPIをコールするPythonスクリプトを作成します。SwitchBot APIの公式ドキュメントを参照しながら実装します。

SwithBot APIのドキュメント
https://github.com/OpenWonderLabs/SwitchBotAPI

デバイスIDを取得する

まずは温湿度計のデバイスIDを取得します。SwitchBot APIを利用したデバイスIDの取得方法は、過去の記事で説明しているため、詳細はこちらを参照してください。

今回は以下のように、Switchbot APIの操作をまとめたSwithbotクラスを作成しました。このPythonスクリプト自体を実行することで、自分が登録しているデバイスのリストをjsonファイルで取得できます。

switchbot.py

import base64
import hashlib
import hmac
import json
import os
import time

import requests
from dotenv import load_dotenv
from requests.exceptions import HTTPError, RequestException

load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))

API_BASE_URL = "https://api.switch-bot.com"

ACCESS_TOKEN: str = os.environ["SWITCHBOT_ACCESS_TOKEN"]
SECRET: str = os.environ["SWITCHBOT_SECRET"]

class Switchbot:
    def __init__(self, access_token=None, secret=None):
        self.access_token = access_token or ACCESS_TOKEN
        self.secret = secret or SECRET

    def __generate_request_headers(self) -> dict:
        """SWITCH BOT APIのリクエストヘッダーを生成する"""

        nonce = ""
        t = str(round(time.time() * 1000))
        string_to_sign = "{}{}{}".format(self.access_token, t, nonce)
        string_to_sign_b = bytes(string_to_sign, "utf-8")
        secret_b = bytes(self.secret, "utf-8")
        sign = base64.b64encode(
            hmac.new(secret_b, msg=string_to_sign_b, digestmod=hashlib.sha256).digest()
        )

        headers = {
            "Authorization": self.access_token,
            "t": t,
            "sign": sign,
            "nonce": nonce,
        }

        return headers

    def get_device_list(self) -> dict:
        """SWITCH BOTのデバイスリストを取得する"""

        url = f"{API_BASE_URL}/v1.1/devices"
        try:
            r = requests.get(url, headers=self.__generate_request_headers())
            r.raise_for_status()
        except HTTPError as e:
            raise HTTPError(f"HTTP error: {e}")
        except RequestException as e:
            raise RequestException(e)
        else:
            return r.json()["body"]["deviceList"]

    def get_device_status(self, device_id: str) -> dict:
        """Switchbotデバイスのステータスを取得する"""

        url = f"{API_BASE_URL}/v1.1/devices/{device_id}/status"

        try:
            r = requests.get(url, headers=self.__generate_request_headers())
            r.raise_for_status()
        except HTTPError as e:
            raise HTTPError(f"HTTP error: {e}")
        except RequestException as e:
            raise RequestException(e)
        else:
            return r.json()["body"]

if __name__ == "__main__":
    # デバイスリストをjsonファイルに出力する
    bot = Switchbot()
    device_list = bot.get_device_list()

    with open("./device_list.json", "w") as f:
        f.write(json.dumps(device_list, indent=2, ensure_ascii=False))

これをスクリプトとして直接実行すると、以下のようなjsonファイルを出力します。この処理はdockerの起動時に1回だけ実行することで、APIのコール数を減らしています。

device_list.json

[
  {
    "deviceId": "yyyyy",
    "deviceName": "キッチン",
    "deviceType": "Color Bulb",
    "enableCloudService": true,
    "hubDeviceId": "zzzzz"
  },
  {
    "deviceId": "xxxxx",
    "deviceName": "温湿度計プラス",
    "deviceType": "MeterPlus",
    "enableCloudService": true,
    "hubDeviceId": "yyyyy"
  }
]

今回はdeviceTypeがMeterPlusのデバイスを対象として、温湿度情報を取得します。

温湿度計のデータを定期的に取得する

次に、SwitchBot APIを使用して、温湿度計のデータを定期的に取得します。

先ほど出力したjsonファイルを読み込み、deviceTypeがMeterPlusの場合はデータを取得します。

main.py（部分）

def task():
    """定期実行するタスク"""
    bot = Switchbot(ACCESS_TOKEN, SECRET)

    with open("device_list.json", "r") as f:
        device_list = json.load(f)

    for d in device_list:
        device_type = d.get("deviceType")
        if device_type == "MeterPlus":
            try:
                status = bot.get_device_status(d)
            except Exception as e:
                print(f"Request error: {e}")
                continue

            try:
                save_device_status(status)
            except Exception as e:
                print(f"Save error: {e}")

scheduleモジュールを利用して、このタスクを5分間に1回実行するように設定します。

main.py（部分）

if __name__ == "__main__":
    schedule.every(5).minutes.do(task)

    while True:
        schedule.run_pending()
        sleep(1)

取得したデータをInfluxDBに保存する

influxdb_clientモジュールを利用して、取得したデータをInfluxDBに保存します。タグにはデバイスIDを、フィールドには温度と湿度を保存します。タイムスタンプには書き込み日時が自動的に設定されます。

main.py（部分）

def save_device_status(status: dict):
    """SwitchbotデバイスのステータスをInfluxDBに保存する"""

    device_type = status.get("deviceType")

    if device_type == "MeterPlus":
        p = (
            Point("MeterPlus")
            .tag("device_id", status["deviceId"])
            .field("humidity", float(status["humidity"]))
            .field("temperature", float(status["temperature"]))
        )

        write_api.write(bucket=bucket, record=p)
        print(f"Saved:{status}")

このスクリプトを実行すると、以下のようにデータが記録されます。温湿度計を複数登録している場合は、同じバケットに異なるdevice_idで記録することになります。

Bucket: switchbotのデータの例

time	measurement	device_id	humidity	temperature
2023-01-01 12:00:00	MeterPlus	yyyyy	45.0	22.0
2023-01-01 12:05:00	MeterPlus	yyyyy	50.0	20.0
2023-01-01 12:10:00	MeterPlus	yyyyy	55.0	18.0

Grafanaでダッシュボードを作る

最後に、InfluxDBに保存したデータをGrafanaで可視化し、ダッシュボードを作成します。ここからは画面の操作になるので、手順を大まかに説明します。

データソースの設定

• ブラウザでGrafanaにアクセスし、adminユーザーでログインします。

  • デフォルトのID/PWはadmin/adminです。初回ログイン後は任意のPWに変更してください。

• 左側のサイドバーから「Configuration」を選択し、「Data Sources」→「Add data source」→「InfluxDB」の順にクリックします。InfuluxDBの設定画面が開きます。

• 設定画面でInfluxDBの接続情報を入力します。

  • Name: 任意の名前
  • Query Language: Flux
  • URL: http://influxdb:8086
  • Auth: 全てOFF
  • Organization: org
  • Token:「環境変数を取得する」で取得したInfluxDBのトークン
  • Default Bucket: switchbot

InfuluxDBの接続設定画面

• 「Save & Test」をクリックし、接続が成功したことを確認します。
• 左側のサイドバーから「Dashboards」→「New dashboard」→「Add a new panel」の順にクリックします。パネルの編集画面が開きます。
• クエリの入力欄に、以下のクエリを入力します。これは湿度を取得するクエリです。温度を取得する場合は、r._fieldの箇所を修正します。

from(bucket: "switchbot")
  |> range(start: v.timeRangeStart, stop:v.timeRangeStop)
  |> filter(fn: (r) =>
    r._measurement == "MeterPlus" and
    r._field == "humidity"
  )

これで、データを可視化する準備ができました。

データの可視化

次に、可視化の方法を選択します。シンプルに時系列で表示する場合は「Time series」を選択します。

Time Series

Statsは最新情報をわかりやすく表示したい時に便利です。

Stats

今回のように、周期的に変化する値を可視化する場合にオススメなのが、「Hourly heatmap」です。24時間単位で横に並べたヒートマップを作成できます。

Hourly heatmap

Houtly heatmapはプラグインとして提供されているので、追加でインストールしてください。（インストール方法の説明は割愛します。）

https://grafana.com/grafana/plugins/marcusolsson-hourly-heatmap-panel/

同様に他のパネルも追加していけば、ダッシュボードの完成です！

完成したダッシュボード

なお、今回はダッシュボードの作成までにしましたが、Grafanaではアラートを設定することもできます。データをしばらく眺めてみて、設定を入れていこうと思っています。

まとめ

この記事では、SwitchBotの温湿度計からデータを取得し、Pythonで処理してInfluxDBに保存し、Grafanaでダッシュボードを作成する方法を解説しました。

同じような情報はSwitchBotの公式アプリからも取得できますが、データをAPI経由で取得することで、もっと自由に可視化・分析できます。ぜひ試してみてください！