SwitchBot APIで家中のデータを自動収集！消費電力・気温・開閉状態をInfluxDBに蓄積する

InfluxDB

「各部屋の気温や湿度を時系列で見たい」「テレビや冷蔵庫が今どれだけ電気を使っているか知りたい」「玄関の鍵が閉まっているか確認したい」——SwitchBot デバイスはスマートフォンアプリで単体確認できますが、それをデータとして蓄積・可視化するには API でラズパイに定期収集する仕組みが必要です。
本記事では、SwitchBot API v1.1 を使ってプラグ（消費電力）・温湿度計・開閉センサー・鍵・カーテンなどのデータを定期取得し、InfluxDB に蓄積する実装方法を解説します。

なぜこの機能を作ったのか？

我が家のスマートホームでは「今この瞬間の状態」だけでなく「過去の推移」も重要な情報です。
例えば「今日の冷蔵庫の電力消費はいつもより多くないか」「夜中に気温が下がりすぎていないか」「昨日の何時にカーテンが開いたか」——こういった時系列分析は、デバイスのデータを継続的に蓄積していなければできません。
そこで各 SwitchBot デバイスのステータスを定期的にラズパイで取得し、時系列データベース（InfluxDB）に保存することで、Grafana などのダッシュボードで可視化できる基盤を作りました。

SwitchBot のデータをただ見るだけでなく、時系列で蓄積して傾向分析に活かしたい！

実現したいこと

  • SwitchBot プラグミニから現在の消費電力（W）・当日積算電力量（Wh）・電圧・電流を収集
  • 温湿度計から各部屋の気温・湿度を定期収集
  • 開閉センサーからドアの開閉状態・照度・人感センサーの検知状態を収集
  • ハブミニ3から気温・湿度・照度・人感センサーを収集
  • スマートロック（玄関の鍵）の施錠状態・ドア開閉状態を収集
  • カーテンの開閉位置（スライド位置）を収集
  • 収集したデータを InfluxDB に保存し、Grafana で可視化できるようにする
  • crontab で定期実行し、常に最新のデータが蓄積され続ける状態を維持する

この記事でわかること

  • SwitchBot API v1.1 の HMAC-SHA256 署名認証の実装方法
  • SwitchBot API でデバイスの一覧とステータスを取得する方法
  • プラグから消費電力の差分・累積 Wh を計算して InfluxDB に保存する方法
  • 温湿度計・開閉センサー・鍵・カーテン・ハブミニのステータスを InfluxDB に保存する方法
  • 収集頻度に応じて複数のスクリプトに分け、crontab で間隔を変えて実行する方法

必要な準備と用意するもの

ハードウェア
  • Raspberry Pi（例：Raspberry Pi 5）
  • SwitchBot デバイス（以下のうち収集したいもの）
    • SwitchBot プラグミニ（消費電力・電圧・電流の取得に対応）
    • SwitchBot 温湿度計（各部屋の気温・湿度）
    • SwitchBot 開閉センサー（ドア開閉・照度・人感）
    • SwitchBot Hub Mini 3（気温・湿度・照度・人感）
    • SwitchBot スマートロック（鍵の施錠状態・ドア開閉）
    • SwitchBot カーテン（スライド位置）
ソフトウェア/サービス
  • Python 3
    • requests
  • SwitchBot アカウント
    • API トークン（v1.1 用）
    • クライアントシークレット（v1.1 署名用）
  • InfluxDB（データ保存先の時系列データベース）

完成イメージ

完成後のシステムは crontab によって下記の 2 つのスクリプトが定期実行され、常に最新のデータが InfluxDB に蓄積されます。

  • device_status_record_short.py（1 分ごと）：変化が頻繁なデータを高頻度で収集
    • プラグミニの消費電力（W）・当日積算電力量（Wh）・電圧・電流・差分 Wh・累積 Wh
    • 開閉センサーの開閉状態・照度・人感検知
  • device_status_record_long.py（10 分ごと）：変化がゆるやかなデータを低頻度で収集
    • 各部屋の温湿度計（気温・湿度）
    • Hub Mini 3（気温・湿度・照度・人感）
    • 玄関の鍵（施錠状態・ドア開閉）
    • カーテンのスライド位置

本記事では InfluxDB への書き込みは自作ライブラリ（influxwriter）を通じて行います。InfluxDB のセットアップや書き込みライブラリの詳細は別記事を参照してください。

システムの仕組み

crontab から定期実行されたスクリプトが、SwitchBot API にリクエストを送ってデバイスのステータスを取得し、InfluxDB に書き込みます。SwitchBot API のアクセスは自作ライブラリ（switchbot.py）にまとめており、各スクリプトからはデバイス ID を渡すだけでデータを取得できます。

SwitchBot API v1.1 ではリクエストごとに HMAC-SHA256 署名が必要です。トークン・タイムスタンプ・nonce の組み合わせで署名を生成し、リクエストヘッダーに付与します。

実装のポイント

SwitchBot API v1.1 の HMAC-SHA256 署名認証

SwitchBot API v1.1 以降は、API トークンだけでなくリクエストごとに HMAC-SHA256 署名を生成してヘッダーに付与する必要があります。
署名の生成には「トークン ＋ タイムスタンプ（ミリ秒）＋ nonce（UUID）」を連結した文字列をクライアントシークレットで署名します。

v1.0 の API トークンのみの認証は廃止予定です。新規実装は v1.1 の署名認証を使ってください。

収集頻度に応じたスクリプト分割

データの変化頻度に合わせてスクリプトを 2 つに分けることで、API コール数を最小限に抑えつつ必要な解像度でデータを収集できます。

  • 短期（1 分）：電力・開閉状態のように刻々と変わるデータ
  • 長期（10 分）：気温・湿度・鍵の状態のようにゆっくり変わるデータ

API コール数の節約
SwitchBot API は無料枠の API コール数に上限があります。変化の少ないデータを高頻度で取得しても無駄なので、デバイスの特性に合わせた間隔を設定することが重要です。

プラグの消費電力：差分と累積 Wh の計算

プラグの API レスポンスは「現在の電力（W）」と「当日の積算電力量（Wh）」を返しますが、時系列グラフで「どの時間帯に何 Wh 消費したか」を見たい場合は差分計算が必要です。
直近の DB 保存値との差分を計算し、さらに起動してからの累積 Wh も保存することで、Grafana で様々な集計が可能になります。

electricityOfDay は日付が変わるとリセットされます。差分がマイナスになった場合（日またぎ）は当日の値をそのまま差分として扱います。

事前準備

SwitchBot API トークンとシークレットの取得

  1. SwitchBot アプリを開き「プロフィール」→「環境設定」を選択
  2. 「アプリのバージョン」を 10 回タップして開発者向けオプションを表示
  3. 「トークン」と「クライアントシークレット」を取得してメモしておく

デバイス ID の取得

SwitchBot API の GET /v1.1/devices にリクエストを送ると登録済みデバイスの一覧と deviceId が取得できます。switchbot.py の devices() メソッドを使って一覧を取得し、各デバイスの ID を settings.json に登録してください。

ライブラリのインストール

pip install requests

実装方法

設定ファイル（settings.json）

API トークン・シークレット・デバイス ID を JSON で管理します。

{
    "switchbot": {
        "api": "<APIトークン>",
        "secret": "<クライアントシークレット>",
        "api_url": "https://api.switch-bot.com/v1.0/",
        "api_url_v11": "https://api.switch-bot.com/v1.1/",
        "device_id": {
            "floor_heating_plug":      "<デバイスID>",
            "tv_plug":                 "<デバイスID>",
            "nucbox_plug":             "<デバイスID>",
            "strage_room_plug":        "<デバイスID>",
            "miyu_aircon_plug":        "<デバイスID>",
            "bedroom_aircon_plug":     "<デバイスID>",
            "tv_around_plug":          "<デバイスID>",
            "refrigerator_plug":       "<デバイスID>",
            "charging_around_plug":    "<デバイスID>",
            "rice_cooker_plug":        "<デバイスID>",
            "dining_plug":             "<デバイスID>",
            "living_door_sensor":      "<デバイスID>",
            "living_aircon_sensor":    "<デバイスID>",
            "hall_lighting_sensor":    "<デバイスID>",
            "strage_room_thermometer": "<デバイスID>",
            "living_thermometer":      "<デバイスID>",
            "bedroom_thermometer":     "<デバイスID>",
            "miyu_room_thermometer":   "<デバイスID>",
            "living_hubmini3":         "<デバイスID>",
            "front_door_lock":         "<デバイスID>",
            "living_curtain":          "<デバイスID>",
            "living_curtain_8B":       "<デバイスID>"
        }
    }
}

SwitchBot API ライブラリ（switchbot.py）

SwitchBot API の認証・デバイス一覧取得・ステータス取得・コマンド送信をまとめた共通ライブラリです。

# -*- coding: utf-8 -*-
import requests
import json
import time
import hashlib
import hmac
import base64
import uuid

class ApiVersion:
    VERSION_10 = 'v1.0'
    VERSION_11 = 'v1.1'

# API v1.1 用の HMAC-SHA256 署名ヘッダーを生成
def get_header_v11(token, secret):
    nonce = str(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())

    header = {}
    header["Authorization"] = token
    header["sign"] = str(sign, 'utf-8')
    header["t"] = str(t)
    header["nonce"] = nonce
    return header

header_v11 = get_header_v11(settings['switchbot']['api'], settings['switchbot']['secret'])

header_v10 = {
    "Authorization": settings['switchbot']['api'],
    'Content-Type': 'application/json; charset: utf8'
}

class Switchbot:
    def __init__(self, api_version: ApiVersion = ApiVersion.VERSION_11):
        if api_version == ApiVersion.VERSION_10:
            self.switchbot_api_url = settings['switchbot']['api_url']
            self.header = header_v10
        elif api_version == ApiVersion.VERSION_11:
            self.switchbot_api_url = settings['switchbot']['api_url_v11']
            self.header = header_v11

        self.device_list = None
        self.infrared_remote_List = None

    def _get_devices(self):
        if self.device_list is None:
            url = self.switchbot_api_url + "devices"
            body = json.loads(requests.get(url, headers=self.header).text)['body']
            self.device_list = body['deviceList']
            self.infrared_remote_List = body['infraredRemoteList']

    def devices(self):
        self._get_devices()
        return self.device_list

    # デバイスのステータスを返却
    def get_device_status(self, device_id):
        url = self.switchbot_api_url + "devices/" + device_id + '/status'
        device_status = json.loads(requests.get(url, headers=self.header).text)['body']
        return device_status

    # デバイスにコマンドを送信
    def execute_command(self, device_id, json_data):
        url = self.switchbot_api_url + "devices/" + device_id + '/commands'
        response = requests.post(url, headers=self.header, json=json_data)
        return response

短期収集スクリプト：プラグ・開閉センサー（device_status_record_short.py）

1 分ごとに実行され、プラグの電力情報と開閉センサーのステータスを InfluxDB に保存します。

# -*- coding: utf-8 -*-
__group__ = '機器記録'
__description__ = '家庭内のIoT機器の状況を記録(1分おき)'

from influx.v2 import client as influxclient
from influx.v2 import writer as influxwriter
import smarthomelog
import common

def insert_sensor_info_by_switchbot():
    influx = influxwriter.Writer()

    # ---- プラグミニ（消費電力・電流・電圧など）----
    influx.insert_plug_points(settings['switchbot']['device_id']['floor_heating_plug'])  # 床暖房
    influx.insert_plug_points(settings['switchbot']['device_id']['tv_plug'])             # テレビ
    influx.insert_plug_points(settings['switchbot']['device_id']['nucbox_plug'])         # NUCBOX
    influx.insert_plug_points(settings['switchbot']['device_id']['strage_room_plug'])    # 物置部屋
    influx.insert_plug_points(settings['switchbot']['device_id']['miyu_aircon_plug'])    # 子供部屋エアコン
    influx.insert_plug_points(settings['switchbot']['device_id']['bedroom_aircon_plug']) # 寝室エアコン
    influx.insert_plug_points(settings['switchbot']['device_id']['tv_around_plug'])      # テレビ周り
    influx.insert_plug_points(settings['switchbot']['device_id']['refrigerator_plug'])   # 冷蔵庫
    influx.insert_plug_points(settings['switchbot']['device_id']['charging_around_plug'])# スマホ充電
    influx.insert_plug_points(settings['switchbot']['device_id']['rice_cooker_plug'])    # 炊飯器
    influx.insert_plug_points(settings['switchbot']['device_id']['dining_plug'])         # ダイニング

    # ---- 開閉センサー（開閉状態・照度・人感）----
    influx.insert_opensensor_points(settings['switchbot']['device_id']['living_door_sensor'])    # リビングドア
    influx.insert_opensensor_points(settings['switchbot']['device_id']['living_aircon_sensor'])  # エアコン送風口
    influx.insert_opensensor_points(settings['switchbot']['device_id']['hall_lighting_sensor'])  # 廊下照度

if __name__ == '__main__':
    try:
        if common.is_active_service_status('influxdb.service'):
            insert_sensor_info_by_switchbot()
            success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
            success_log.write('処理が正常に終了しました。', __file__)
    except Exception as e:
        log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
        log.write_error(sys.exc_info())

長期収集スクリプト：温湿度計・鍵・カーテン（device_status_record_long.py）

10 分ごとに実行され、変化のゆるやかなデバイスのステータスを InfluxDB に保存します。

# -*- coding: utf-8 -*-
__group__ = '機器記録'
__description__ = '家庭内のIoT機器の状況を記録(10分おき)'

from influx.v2 import client as influxclient
from influx.v2 import writer as influxwriter
import smarthomelog
import common

def insert_sensor_info_by_switchbot():
    influx = influxwriter.Writer()

    # ---- 温湿度計（各部屋の気温・湿度）----
    influx.insert_thermometer_points(settings['switchbot']['device_id']['strage_room_thermometer']) # 物置部屋
    influx.insert_thermometer_points(settings['switchbot']['device_id']['living_thermometer'])      # リビング
    influx.insert_thermometer_points(settings['switchbot']['device_id']['bedroom_thermometer'])     # 寝室
    influx.insert_thermometer_points(settings['switchbot']['device_id']['miyu_room_thermometer'])   # 子供部屋

    # ---- Hub Mini 3（気温・湿度・照度・人感）----
    influx.insert_hubmini3_points(settings['switchbot']['device_id']['living_hubmini3'])

    # ---- スマートロック（施錠状態・ドア開閉）----
    influx.insert_lock_points(settings['switchbot']['device_id']['front_door_lock'])

    # ---- カーテン（スライド位置）----
    influx.insert_curtain_points(settings['switchbot']['device_id']['living_curtain'])
    influx.insert_curtain_points(settings['switchbot']['device_id']['living_curtain_8B'])

if __name__ == '__main__':
    try:
        if common.is_active_service_status('influxdb.service'):
            insert_sensor_info_by_switchbot()
            success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)
            success_log.write('処理が正常に終了しました。', __file__)
    except Exception as e:
        log = smarthomelog.Log(__file__, smarthomelog.LogType.SYSTEM_ERROR)
        log.write_error(sys.exc_info())

InfluxDB 書き込み処理（writer.py の主要メソッド）

各デバイスタイプに対応した書き込みメソッドを実装します。デバイス ID を受け取り、SwitchBot API でステータスを取得して InfluxDB に保存します。

# -*- coding: utf-8 -*-
import datetime
import switchbot
from influx.v2 import client as influxclient

# プラグの情報を取得する間隔(秒)
PLUG_INFO_GET_INTERVAL = 20

class Writer:
    def __init__(self):
        self.conn = influxclient

    # 「温湿度計」の情報を保存
    def insert_thermometer_points(self, device_id):
        switchbot_instance = switchbot.Switchbot()
        data = switchbot_instance.get_device_status(device_id)
        measurement = 'meter'
        tags = {
            'hubDeviceId': data['hubDeviceId'],
            'deviceType': data['deviceType'],
            'deviceId': data['deviceId']
        }
        fields = {
            'temperature': float(data['temperature']),
            'humidity': int(data['humidity']),
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
        self.conn.write_dict(measurement, tags, fields)

    # 「プラグ」の情報を保存（差分・累積 Wh の計算含む）
    def insert_plug_points(self, device_id):
        # InfluxDB から直近の値を取得（差分計算用）
        res = self.conn.query_dict_by_influxql('plug', ['electricityOfDay'], {'deviceId': device_id}, influxclient.QueryMode.LAST, None, None)
        latestElectricityOfDay = res[0].get('electricityOfDay') if res else 0

        res = self.conn.query_dict_by_influxql('plug', ['wattCumulative'], {'deviceId': device_id}, influxclient.QueryMode.LAST, None, None)
        latest_watt_cumulative = res[0].get('wattCumulative') if res else 0

        switchbot_instance = switchbot.Switchbot()
        data = switchbot_instance.get_device_status(device_id)
        if data == {}:
            return

        # 当日積算電力量の差分を計算（日またぎでリセットされる場合は当日値をそのまま使用）
        electricityOfDayIncrement = int(data['electricityOfDay']) - latestElectricityOfDay
        if electricityOfDayIncrement < 0:
            electricityOfDayIncrement = int(data['electricityOfDay'])

        # 収集間隔（秒）から Wh に換算
        watt_hour_increment = float(data['weight']) / (60 * (60 / PLUG_INFO_GET_INTERVAL))
        watt_cumulative = latest_watt_cumulative + watt_hour_increment

        measurement = 'plug'
        tags = {
            'hubDeviceId': data['hubDeviceId'],
            'deviceType': data['deviceType'],
            'deviceId': device_id
        }
        fields = {
            'power': str(data['power']),
            'weight': float(data['weight']),              # 現在の消費電力 (W)
            'electricityOfDay': int(data['electricityOfDay']),  # 当日積算電力量 (Wh)
            'electricCurrent': float(data['electricCurrent']),  # 電流 (A)
            'voltage': float(data['voltage']),            # 電圧 (V)
            'isPowerOn': int(1 if data['power'] == 'on' else 0),
            'electricityOfDayIncrement': electricityOfDayIncrement,  # 差分 (Wh)
            'wattHourIncrement': watt_hour_increment,                # 区間換算 Wh
            'wattCumulative': watt_cumulative,                       # 累積 Wh
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
        self.conn.write_dict(measurement, tags, fields)

    # 「開閉センサー」の情報を保存
    def insert_opensensor_points(self, device_id):
        switchbot_instance = switchbot.Switchbot()
        data = switchbot_instance.get_device_status(device_id)
        measurement = 'contact_sensor'
        tags = {
            'hubDeviceId': data['hubDeviceId'],
            'deviceType': data['deviceType'],
            'deviceId': data['deviceId']
        }
        fields = {
            'brightness': str(data['brightness']),
            'openState': str(data['openState']),
            'moveDetected': data['moveDetected'],
            'isBright': int(1 if data['brightness'] == 'bright' else 0),
            'isOpen': int(0 if data['openState'] == 'close' else 1),
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
        self.conn.write_dict(measurement, tags, fields)

    # 「玄関の鍵」の情報を保存
    def insert_lock_points(self, device_id):
        switchbot_instance = switchbot.Switchbot()
        data = switchbot_instance.get_device_status(device_id)
        measurement = 'lock'
        tags = {
            'hubDeviceId': data['hubDeviceId'],
            'deviceType': data['deviceType'],
            'deviceId': data['deviceId']
        }
        fields = {
            'lockState': str(data['lockState']),
            'calibrate': data['calibrate'],
            'doorState': str(data['doorState']),
            'isLocked': int(1 if data['lockState'] == 'locked' else 0),
            'isClosed': int(1 if data['doorState'] == 'closed' else 0),
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
        self.conn.write_dict(measurement, tags, fields)

    # 「カーテン」の情報を保存
    def insert_curtain_points(self, device_id):
        switchbot_instance = switchbot.Switchbot()
        data = switchbot_instance.get_device_status(device_id)
        measurement = 'curtain'
        tags = {
            'hubDeviceId': data['hubDeviceId'],
            'deviceType': data['deviceType'],
            'deviceId': data['deviceId']
        }
        fields = {
            'group': data['group'],
            'slidePosition': int(data['slidePosition']),
            'moving': int(data['moving']),
            'calibrate': data['calibrate'],
            'isOpen': int(1 if data['slidePosition'] >= 99 else 0),  # 99以上で全開と判定
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
        self.conn.write_dict(measurement, tags, fields)

    # 「Hub Mini 3」の情報を保存
    def insert_hubmini3_points(self, device_id):
        switchbot_instance = switchbot.Switchbot()
        data = switchbot_instance.get_device_status(device_id)
        measurement = 'hubmini3'
        tags = {
            'hubDeviceId': data['hubDeviceId'],
            'deviceType': data['deviceType'],
            'deviceId': data['deviceId']
        }
        fields = {
            'version': data['version'],
            'temperature': float(data['temperature']),
            'lightLevel': int(data['lightLevel']),
            'humidity': float(data['humidity']),
            'moveDetected': data['moveDetected'],
            'onlineStatus': data['onlineStatus'],
            'timestamp': datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S.%f')
        }
        self.conn.write_dict(measurement, tags, fields)

crontab への登録

crontab -e で crontab を編集し、2 つのスクリプトを定期実行するエントリを追加します。

# 1分ごと：プラグ（電力）・開閉センサー
* * * * * /home/<username>/Projects/venv/bin/python /home/<username>/smarthome/core/record/device/device_status_record_short.py >> /home/<username>/smarthome/logs/crontab/device_status_short.log 2>&1

# 10分ごと：温湿度計・Hub Mini 3・鍵・カーテン
*/10 * * * * /home/<username>/Projects/venv/bin/python /home/<username>/smarthome/core/record/device/device_status_record_long.py >> /home/<username>/smarthome/logs/crontab/device_status_long.log 2>&1

コードの解説

HMAC-SHA256 署名の生成

get_header_v11() では「トークン ＋ タイムスタンプ（ミリ秒） ＋ nonce（UUID）」を連結した文字列を UTF-8 バイト列に変換し、クライアントシークレットで HMAC-SHA256 署名を生成します。
nonce には UUID を使うことで、同じタイムスタンプでも一意なリクエストになり、リプレイ攻撃を防ぎます。生成した署名はヘッダーの sign フィールドに Base64 エンコードして付与します。

API トークンとクライアントシークレットは絶対にソースコードにハードコードせず、settings.json や環境変数で管理してください。Git リポジトリに誤ってコミットすると第三者に SwitchBot デバイスを操作される恐れがあります。

プラグの差分 Wh 計算の考え方

SwitchBot プラグミニの API が返す electricityOfDay は「今日の 0 時からの積算電力量（Wh）」です。これを使うと日をまたいだ比較は難しいため、別途 2 つの独自フィールドを計算して保存しています。

  • electricityOfDayIncrement：前回保存時からの electricityOfDay の差分。直近の区間でどれだけ消費したかがわかる。
  • wattHourIncrement：現在の消費電力（W）をサンプリング間隔で積分した値（W × 時間 = Wh）。electricityOfDay が更新されていない場合のバックアップ値として使える。
  • wattCumulative：プログラム起動からの累積 Wh。Grafana でグラフとして可視化する際の基準値として使える。

InfluxDB が停止中は実行しない

common.is_active_service_status('influxdb.service') で InfluxDB が起動中かを確認してから実行することで、DB 停止中にエラーが大量発生するのを防いでいます。ラズパイの再起動直後や InfluxDB のメンテナンス時に特に有効です。

InfluxDB のデータ構造（measurement 一覧）

デバイスタイプごとに異なる measurement（テーブル相当）に保存します。

  • meter：温湿度計（temperature, humidity）
  • plug：プラグ（power, weight, electricityOfDay, electricCurrent, voltage, wattCumulative など）
  • contact_sensor：開閉センサー（brightness, openState, moveDetected, isBright, isOpen）
  • lock：スマートロック（lockState, doorState, isLocked, isClosed）
  • curtain：カーテン（slidePosition, moving, isOpen）
  • hubmini3：Hub Mini 3（temperature, humidity, lightLevel, moveDetected）

動作確認

  1. settings.json に API トークン・シークレット・デバイス ID を設定する
  2. switchbot.py の devices() でデバイス一覧を取得し、デバイス ID を確認する
  3. device_status_record_short.py を手動実行してエラーが出ないことを確認する
  4. device_status_record_long.py を手動実行してエラーが出ないことを確認する
  5. InfluxDB または Grafana でデータが保存されていることを確認する
  6. crontab にエントリを追加し、数分後にデータが増えていることを確認する

SwitchBot API は無料枠のリクエスト数に制限があります（1万回/日など）。プラグを 11 台 × 1 分ごと = 15,840 回/日になるため、デバイス数が多い場合は収集間隔を調整してください。

プラグのデバイスタイプが Plug Mini (JP) でない古いプラグは power（ON/OFF）しか取得できず、消費電力・電圧・電流は取得できません。get_device_status() の戻り値を確認し、フィールドが存在するか確認してください。

まとめ

SwitchBot API v1.1 と InfluxDB を組み合わせることで、家中のデバイスのデータを時系列で蓄積するスマートホームデータ基盤が実現できました。

  • SwitchBot API v1.1 の HMAC-SHA256 署名認証で安全に API を呼び出せる
  • プラグの差分 Wh・累積 Wh を計算して保存することで、時間帯別の電力消費分析ができる
  • 収集頻度に応じてスクリプトを分割し、API コール数を節約できる
  • InfluxDB 起動確認を入れることで、DB 停止中の無駄なエラーを防げる
  • デバイスタイプごとに measurement を分けることで、Grafana での可視化がしやすくなる

「InfluxDB に蓄積したデータを Grafana で可視化する」「異常な電力消費を検知して通知する」といった次のステップも、このデータ基盤があれば簡単に実装できます。ぜひ試してみてください。

タイトルとURLをコピーしました