～「SpeeDBee Hive」でBluetooth対応のセンサーコレクタを開発する(SwichBot温湿度計) ～

概要

• この記事では、「SpeeDBee Hive」を使用して、SWITCHBOT社の温湿度計からBluetooth通信を用いた温湿度のデータ収集（センサーコレクタ開発）・可視化（グラフツール利用）の方法について解説します。
• 「SpeeDBee Hive」のPythonカスタムコレクタを用いたデータ収集方法の解説となり、他のBluetooth通信対応のセンサーなどを利用したデータ収集プログラム開発の参考になります。

はじめに

「SpeeDBee Hive」は多様なデータを取り込み、分析、制御、可視化をGUIで簡単に実現する実用的IoTミドルウェア、兼アプリケーションプラットフォームです。
ソルティスターのHPから評価版をダウンロードすることが可能です。

「SpeeDBee Hive」のカスタムコレクタとしてサンプルプログラムを使用していただくことでSWITCHBOT社の温湿度計からBluetooth通信を用いて温湿度のデータ収集・可視化が可能となります。

本記事ではサンプルプログラムとSWITCHBOT社の温湿度計を使用してデータの収集・可視化を行います。

サンプルプログラムは本記事の最後のまとめに貼り付けてあります。

環境

本記事でサンプルプログラムを動作させた環境は以下のようになっています。

• Raspberry Pi 3 Model B
  OS：Raspbian GNU/Linux 11 (bullseye)
• SwitchBot 温湿度計
  https://www.switchbot.jp/products/switchbot-meter
• SpeeDBee Hive
  バージョン：3.7.2
• Grafana
  バージョン：8.3.3

SpeeDBee Hiveのインストール

SpeeDBee Hiveの体験版はソルティスターのHPからダウンロードすることが可能です。

『ダウンロードデータ』のRaspberry Pi版をダウンロードし、Raspberry Piにインストールを行います。
インストール方法はSpeeDBee Hive マニュアルのインストール手順をご参照ください。

製品マニュアルにはSpeeDBee Hiveのインストールをはじめに起動方法、動作確認方法などが記載されていますので、ご一読ください。

SpeeDBee Hive カスタムコレクタ

SpeeDBee Hiveでは、データの収集を行うモジュールのことを"コレクタ"と呼びます。
PLC等のデータを収集する"コレクタ"も標準で用意されていますが、ユーザーが独自に開発、運用することも可能となっています。
ユーザーが独自に開発、運用する"コレクタ"が"カスタムコレクタ"となります。
使用できる言語としてはC言語、もしくはPythonとなっています。

この記事ではSpeeDBee HiveでBluetooth通信可能なセンサーからデータを取得し登録するためにPythonで開発したカスタムコレクタを使用しています。

PythonカスタムコレクタについてはSpeeDBee Hive マニュアルのPythonカスタムコレクタをご参照ください。

センサーやSpeeDBee Hive、カスタムコレクタ等の各関係は下図のような構成となっています。

カスタムコレクタの追加

カスタムコレクタに必要となる、サンプルプログラムの追加方法とその実行に必要なPythonモジュールの追加方法についてとなります。

SpeeDBee Hiveの設定画面で「システム」タブのサブメニュー「コレクタ関連設定」→「カスタム(Python)」で、以下の項目を設定してください。

• SwitchBot温湿度計(ble_collector_switchbot.py)のカスタムコレクタを追加
  サンプルプログラム(ble_collector_switchbot.py)は本記事の最後のまとめに貼り付けてあります。

1. 「システム」タブ→「コレクタ関連設定」→「カスタム(Python)」より、Pythonカスタムコレクタ管理画面を開きます。
   

2. 「追加」をクリックし、ble_collector_switchbot.pyを選択します。
   

3. 登録が完了するとカスタムコレクタの一覧に表示されます。
   

• Pythonモジュールに「bluepy」を追加
  サンプルプログラムではPythonでBluetooth通信を制御するためにbluepyモジュールを使用しています。
  bluepyモジュールの追加方法についての説明となります。

1. Pythonモジュールのテキスト欄に"bluepy"を入力し「追加」をクリックします。
   

2. ダイアログにて「OK」をクリックします。
   

3. インストールが完了すると、"bluepy"が一覧されます。
   

4. 権限の設定を行います。
   SpeeDBee Hiveの実行ユーザではbluepyモジュールの実行権限がなく、そのままではデータの収集ができないので権限を変更します。
   Raspberry Piで以下を実行してください。

$ sudo setcap "cap_net_admin+ep" /var/speedbeehive/dynlibs/pyvenv/lib/python/site-packages/bluepy/bluepy-helper

上記のカスタムコレクタの追加、Pythonモジュールの追加について詳しくはSpeeDBee Hive マニュアルのPythonカスタムコレクタの"6.2.2.3 カスタムコレクタ登録"をご参照ください。

カスタムコレクタ設定

「コレクタ」タブのサブメニュー「カスタム」の をクリックすると新規登録画面が表示されます。

各項目を入力し、「保存」をクリックし設定を保存します。

設定が必要な項目は以下の通りです。

入力項目	説明
カスタム名	データを収集するコレクタ名 ※コレクタ全体で重複不可
タイプ	pythonを選択
ライブラリ	使用するカスタムコレクタ(Python)のスクリプトファイル名 ※保存後は変更不可
パラメータ	カスタムコレクタ内で利用される引数（文字列） デバイス名やデバイスのアドレス、インサート間隔等をJSON形式で指定します。 入力欄は改行出来ませんので、一行で入力してください。

• パラメータについて
  パラメータの例を以下に記載します。
  パラメータはJSON形式で指定します。

※見やすいように改行とインデントを入れています。
{
  "devices":[
    {
      "name":"sensor1",
      "address":"aa:bb:cc:dd:ee:ff"
    },
    {
      "name":"sensor2",
      "address":"11:22:33:44:55:66"
    }
  ],
  "interval":5
}

パラメータの各要素は以下のようになっています。

プロパティ	説明
devices	センサーの情報 nameとaddressをデータ取得したいセンサ個数分設定
name	センサーの名前(任意の名前)
address	センサーのアドレス
interval	データのインサート間隔(秒)

SwitchBot 温湿度計のアドレスの確認

SwitchBotの公式アプリでセンサーのアドレスを確認をします。

1. アドレスを確認したいセンサーを選択

1. 歯車のマークをタップ

1. 「デバイス情報」をタップ

1. "BLE MAC"の値がセンサーのアドレスとなり、パラメータの"address"に入力する情報となります。

SpeeDBee Hiveの起動

「システム」タブにある「制御」からHiveの「起動」ボタンをクリックします。

ここまでに設定した条件でセンサーデータ収集の開始されます。

Grafanaによる温湿度データの可視化

SpeeDBee HiveはGrafana Labs社のオープンソース可視化ツールである「Grafana」に対応をしており、コレクタが収集した各種データをWeb ブラウザ上で折れ線グラフなどを用いて視覚的にわかりやすく閲覧することができます。

Grafanaでの可視化方法について詳しくはSpeeDBee Hive マニュアルのGrafana連携をご参照ください。

下の例はSwitchBot 温湿度計より収集した、現在時刻から過去15分間の温度データの推移を折れ線グラフで表示したものとなります。
温度データは10秒間隔で収集しています。

サンプルプログラムの解説

サンプルプログラムの全体像はプログラムをダウンロードしてご確認ください。
以下、解説となります。

必要となるライブラリ

まず、Pythonカスタムコレクタを開発する際にはかならずHiveCollectorBaseクラスと HiveColumnクラスをインポートする必要があります。
また、Bluetooth通信を行うためbluepyよりbtleクラスのインポートが必要となります。
パラメータにJSON形式を利用しているためjsonもインポートしておきます。

ble_collector_switchbot.py

from hive_collector import HiveCollectorBase, HiveColumn
from bluepy import btle
import json

HiveCollectorクラス

Pythonカスタムコレクタで必須となる、HiveCollectorBaseクラスを継承したHiveCollectorクラスです。

HiveCollector.__init__メソッドはコンストラクタでコレクタの開始時に最初に実行されるメソッドです。
引数として自分自身のインスタンスと"param"を受け取っています。
サンプルプログラムでは"param"はセンサーのアドレスなどが記述されたJSON形式のパラメータが渡ってきます。
パラメータを解析し、SwitchbotCollecterインスタンスの作成とデータ収集間隔を保持します。
SwitchbotCollecterクラスについては後に説明します。

ble_collector_switchbot.py

class HiveCollector(HiveCollectorBase):
    def __init__(self, param):
        self.logger.info("init")
        # JSON形式のパラメータの読込
        param_dict = json.loads(param)

        # パラメータを解析し、SwitchbotCollecterインスタンス作成やデータ収集間隔を保持
        self.collectors = {}
        device_list = param_dict['devices']
        for device_info in device_list:
            device_info['address'] = str(device_info['address']).lower()
            self.collectors[device_info['address']] = SwitchbotCollecter(self, device_info['name'])
        
        self.interval = param_dict['interval']

HiveCollector.mainloopメソッドはカスタムコレクタのデータ収集のメイン処理を行うメソッドです。
Bluetoothセンサーのスキャンでセンサーが見つかった場合にコールバックされるScanDelegateインスタンスを作成します。
ScanDelegateクラスについては後に説明します。
作成したScanDelegateをbtle.Scanner().withDelegateメソッドによってdelegate関数として指定することでスキャン時にコールバックされます。

最後にself.intervalCall()を呼び出しています。これにより、引数に与えられたproc関数が定期的にコールされます。

ble_collector_switchbot.py

    def mainloop(self):
        self.logger.debug("main loop execute")
        # ScanDelegateインスタンス作成
        self.delegate = ScanDelegate(self.collectors)
        # delegate関数として指定
        self.scanner = btle.Scanner().withDelegate(self.delegate)
        # データ収集、登録
        self.intervalCall(self.interval*1000*1000, self.proc)

HiveCollector.procメソッドはself.intervalCall()の内部から、定期的にコールされる関数です。
このprocメソッド内でscanner.scanメソッドを呼び出しBluetoothセンサーのスキャンを行っています。
スキャンでセンサーが見つかった場合はdelegate関数として指定された、ScanDelegateクラスのhandleDiscoveryメソッドが実行されます。
後のScanDelegateクラスの説明で記載しますが、handleDiscoveryメソッド内ではBluetoothセンサーから取得した温湿度データの解析、登録を実行しています。
scanner.scanメソッドの引数にパラメータで設定された実行周期(self.interval)から0.1(秒)少ない値を渡しています。
これはHiveCollector.procメソッド自体の処理が実行周期(self.interval)内で終わるようにするためとなっています。

ble_collector_switchbot.py

    def proc(self, ts, skip):
        try:
            # 設定した実行周期内で処理が終了するように0.1秒だけ少なくscanを実行
            self.scanner.scan(self.interval - 0.1)
        except btle.BTLEManagementError as e:
            self.logger.error(e)

ScanDelegateクラス

Bluetoothセンサーのスキャンでコールバックされるdelegate関数となるScanDelegateクラスです。
bluepyライブラリではdelegate関数にはDefaultDelegateクラスが用意してあり、このクラスのサブクラスを作成しています。
後述する、handleDiscoveryメソッドをオーバーライドすることで、 独自のdelegate関数を定義することができます。

ScanDelegate.__init__メソッドはコンストラクタで、内部でSwitchbotCollecterインスタンスが必要となるため保持しています。

ble_collector_switchbot.py

class ScanDelegate(btle.DefaultDelegate):
    def __init__(self, collectors):
        super().__init__()
        self.collectors = collectors

ScanDelegate.handleDiscoveryメソッドでDefaultDelegate.handleDiscoveryメソッドをオーバーライドし、使用するセンサーの仕様に合わせたcallback関数を定義しています。
センサーのスキャンでデータ取得ができた場合、取得したデータが温湿度データであるかを判定するため、getScanDataメソッドで得られたタプル内の説明(desc)の値を使用しています。
SwitchBot 温湿度計の場合、説明(desc)が"16b Service Data"であるデータが温湿度データとなります。
説明(desc)が"16b Service Data"であり、パラメータで設定してあるセンサーのアドレスからのデータであった場合はSwitchbotCollecter.extractメソッドを呼び出し、データの解析とSpeeDBee Hiveへの登録を行います。

ble_collector_switchbot.py

    def handleDiscovery(self, dev, isNewDev, isNewData):
        if isNewData:
            for (adtype, desc, value) in dev.getScanData():
                if desc == '16b Service Data' and dev.addr in self.collectors:
                    self.collectors[dev.addr].insert(value)

SwitchbotCollecterクラス

SwitchbotCollecterクラスは"SwitchBot 温湿度計"からのデータの解析や、SpeeDBee Hiveへの登録などの処理を定義したクラスとなります。
複数の"SwitchBot 温湿度計"をパラメータで設定し、データ収集が可能なようにクラス化しています。

SwitchbotCollecter.__init__メソッドでは処理に必要となる変数を定義しています。
また、_cleate_columnメソッドを呼ぶことでSpeeDBee Hiveの"カラム"の作成を行います。

ble_collector_switchbot.py

class SwitchbotCollecter():
    def __init__(self, hivecollector, device_name):
        # カスタムコレクタインスタンス
        self._hivecollector = hivecollector
        # センサー名
        self._device_name = device_name
        # Hiveカラム
        self._columns = {}
        
        # Hiveカラムの作成
        self._cleate_column()

SwitchbotCollecter._cleate_columnメソッドでは"カラム"の作成を行っています。
makeOutputColumn()メソッドをコールすることにより、 コレクタが出力するデータを登録する"カラム"を定義することができます。
前述のHiveCollector.__init__メソッド内でSwitchbotCollecterインスタンスの作成時点で実行され、"カラム"が定義されます。
今回は温度、湿度データの登録を行うため2つのカラムを作成しています。
判別をしやすくするためにパラメータで設定したセンサーの名前と温度(temp)または湿度(humi)を組み合わせたカラム名を指定し、データ値がFloatであるため型指定として"HiveColumn.TypeFloat"を与えています。

ble_collector_switchbot.py

    def _cleate_column(self):
        self._columns['temp'] = self._hivecollector.makeOutputColumn(self._device_name + "_temp",  HiveColumn.TypeFloat)
        self._columns['humi'] = self._hivecollector.makeOutputColumn(self._device_name + "_humi",  HiveColumn.TypeFloat)

SwitchbotCollecter.insertメソッドではセンサーから取得したデータを温度、湿度のデータに解析し、その後データをSpeeDBee Hiveへ登録しています。
まず、SwitchBot 温湿度計のデータ仕様に則ってデータ解析を行い、温度、湿度データとしています。
その後、SpeeDBee Hiveへ温度データ、湿度データとして登録を行っています。

ble_collector_switchbot.py

    def insert(self, value):
        # 取得したデータ全体からセンサデータのみを取り出す
        valueBinary = bytes.fromhex(value[4:])
        # 温度を数値データに変換
        isTemperatureAboveFreezing = valueBinary[4] & 0b10000000
        temp = ( valueBinary[3] & 0b00001111 ) / 10 + ( valueBinary[4] & 0b01111111 )
        
        if not isTemperatureAboveFreezing:
            temp = -temp
        
        # 湿度を数値データに変換
        humi = valueBinary[5] & 0b01111111
        
        # Hiveカラムへデータのインサート処理
        self._columns['temp'].insert(temp)
        self._columns['humi'].insert(humi)

サンプルプログラムの応用

サンプルプログラム内で

• ScanDelegate.handleDiscoveryメソッド内の温湿度データの判定を行う説明(desc)の文字列
• SwitchbotCollecter.insertメソッドでのデータ解析、登録

上記をお持ちのBluetooth接続可能な温湿度センサーの仕様に合わせて変更していただくことで、他のセンサーからのデータ取得も可能となっています。

次回はセンサーをINKBIRD社の温湿度計(IBS-TH2)に変更し、データ収集を行うためのサンプルプログラムの修正について記載します。

次回の記事:
「SpeeDBee Hive」でINKBIRD社の温湿度計(IBS-TH2)からデータを取得する(サンプルプログラムの応用)

まとめ

今回はソルティスターの「SpeeDBee Hive」を使用して、SWITCHBOT社の温湿度計からBluetooth通信を用いた温湿度のデータ収集・可視化を行いました。
お手元に環境をご用意できる方は是非、データ収集と可視化をお試しください。

SpeeDBee Hiveの関連記事：
・AE2100でソルティスターの「SpeeDBee Hive」を使用してみよう(1) ―SmartHopセンサーデータの収集・可視化編―
・AE2100でソルティスターの「SpeeDBee Hive」を使用してみよう(2) ―パトランプ制御編―
・AE2100でソルティスターの「SpeeDBee Hive」を使用してみよう(3) ―AWS連携編―

最後に本記事で使用しているサンプルプログラムを以下に貼りますのでご活用ください。

ble_collector_switchbot.py

from hive_collector import HiveCollectorBase, HiveColumn
from bluepy import btle
import json

class HiveCollector(HiveCollectorBase):
    def __init__(self, param):
        self.logger.info("init")
        # JSON形式のパラメータの読込
        param_dict = json.loads(param)

        # パラメータを解析し、SwitchbotCollecterインスタンス作成やデータ収集間隔を保持
        self.collectors = {}
        device_list = param_dict['devices']
        for device_info in device_list:
            device_info['address'] = str(device_info['address']).lower()
            self.collectors[device_info['address']] = SwitchbotCollecter(self, device_info['name'])
        
        self.interval = param_dict['interval']

    def mainloop(self):
        self.logger.debug("main loop execute")
        # ScanDelegateインスタンス作成
        self.delegate = ScanDelegate(self.collectors)
        # delegate関数として指定
        self.scanner = btle.Scanner().withDelegate(self.delegate)
        # データ収集、登録
        self.intervalCall(self.interval*1000*1000, self.proc)
        
    def proc(self, ts, skip):
        try:
            # 設定した実行周期内で処理が終了するように0.1秒だけ少なくscanを実行
            self.scanner.scan(self.interval - 0.1)
        except btle.BTLEManagementError as e:
            self.logger.error(e)

class ScanDelegate(btle.DefaultDelegate):
    def __init__(self, collectors):
        super().__init__()
        self.collectors = collectors

    def handleDiscovery(self, dev, isNewDev, isNewData):
        if isNewData:
            for (adtype, desc, value) in dev.getScanData():
                if desc == '16b Service Data' and dev.addr in self.collectors:
                    self.collectors[dev.addr].insert(value)

class SwitchbotCollecter():
    def __init__(self, hivecollector, device_name):
        # カスタムコレクタインスタンス
        self._hivecollector = hivecollector
        # センサー名
        self._device_name = device_name
        # Hiveカラム
        self._columns = {}
        
        # Hiveカラムの作成
        self._cleate_column()
        
    def _cleate_column(self):
        self._columns['temp'] = self._hivecollector.makeOutputColumn(self._device_name + "_temp",  HiveColumn.TypeFloat)
        self._columns['humi'] = self._hivecollector.makeOutputColumn(self._device_name + "_humi",  HiveColumn.TypeFloat)

    def insert(self, value):
        # 取得したデータ全体からセンサデータのみを取り出す
        valueBinary = bytes.fromhex(value[4:])
        # 温度を数値データに変換
        isTemperatureAboveFreezing = valueBinary[4] & 0b10000000
        temp = ( valueBinary[3] & 0b00001111 ) / 10 + ( valueBinary[4] & 0b01111111 )
        
        if not isTemperatureAboveFreezing:
            temp = -temp
        
        # 湿度を数値データに変換
        humi = valueBinary[5] & 0b01111111
        
        # Hiveカラムへデータのインサート処理
        self._columns['temp'].insert(temp)
        self._columns['humi'].insert(humi)