環境センサRS-BTEVS1のBLE AdvertisingDataを受信する

IoTセンサデバイスでは、長期間のバッテリ動作を実現するために通信手段として省電力タイプのBLE(Bluetooth Low Enagy)が使用されます。通信Protocolも簡略化と省電力化のためにBeacon(ビーコン)と同様なAdvertising Packetを定期的に発信するような仕様となっています。これらは放送電波のように相手かまわず発信される(Broadcast)ので、受信側で待ち受け、受信したPacketからAdvertizingDataを取り出し、そこから詳細なデータを取り出す必要があります。AdvertisingDataにはManufacturer Code(製造者ID)が含まれており、そのIDとBluetoothアドレスで送信元の機器を特定します。Bluetoothアドレスは機器に貼り付けられているステッカーに表記されています。Manufacturer Codeは下記のURL(RS-BTEVS1のAdvertisingDataのFormat説明)で公表しています。

当社にはBLE Advertising Packetを発信するIoT Deviceがいくつかあります。そのうちの一つ、環境センサ(CO2濃度、PM2.5など、温度、湿度) RS-BTEVS1 からのAdvertisingDataを受信してWindows11のデスクトップにCO2,湿度、温度を表示するWidgetの作成例を紹介します。本当はタスクバーの左端に表示される「お天気」Widgetの横にCO2濃度、湿度などのデータを表示したかったのですが、方法がわからず今回はあきらめました。
RS-BTEVS1が発信するAdvertisingDataのFormat説明は当社のHP https://www.ratocsystems.com/topics/newfeature/btevs1fmt/ からdownloadすることができます。

Python 3.11 BLEライブラリ bleak を使う

いつものように、Windows11上でPython 3.11を使用したProgram例を示します。BLE用のライブラリは bleak が提供されていますが、Versionにより関数名が異なる場合や削除されている場合があるため、Web上のProgram例がそのままでは動作しないことがあります。本稿では bleak 0.20.2を使用しました。
bleakの解説はhttps://bleak.readthedocs.io/en/latest/index.htmlを参照して下さい。

bleakライブラリはコマンドプロンプトから事前にinstallしておいて下さい。

>python -m -pip install bleak

今回はAdvertisingDataを受信するだけでよいのでライブラリ中のBleakScanner Classのみを使用します。したがってProgramの最初の部分で

from bleak import BleakScanner

が必要です。

bleakのonlineドキュメント(上記URL)では、Advertising Packetを受信する(待受ける)方法について2通りの方法を解説しています。

• Easy Method
  • BleakScanner.discover()を使用する方法
  • Advertise専用Ch.(37,38,39)をScanしながらAdvertising Packetを受信するのを待つ
• Recommended Method
  • BleakScanner.start()、BleakScanner.stop()を使用する方法
  • start()とstop()の間で受信したAdvertisingパケットをDict型のpropertyとして取り出す

本稿のProgram例では、widgetのためのtkinter、timerなどの複数のEventを一つのcontext managerで管理する必要があり、そのためRecommended Methodを使用しています。その方法にしたがってAdvertisingDataを受信する部分は下のListのような処理を行っています。

async def get_advdata():
    global scanner      # = BleakScanner()

    await scanner.start()
    await asyncio.sleep(10.0)
    await scanner.stop()

    # Advertising Dataから必要な情報を取得する処理

    return dsp_txt

前回のWi-SUNのBルート接続例とは違って、Advertising Packetの受信はconnectionが不要ですので通信部分は簡単になります。受信待機・検出・完了のevent処理はasyncio — awaitで行います。

実際の受信待機とAdvertisingDataの取り出しは、次の手順で行われます。

1. BleakScanner.start()を呼び出し、Advertising Packetを待ちます。
2. bleakのDocumentのおすすめの5秒でstop()を呼び出すと、時たま空のPropertyが返ってくることがあるため、10秒待機しています。
3. Advertising Packetを受信したら、BleakScanner.stop()を呼び出します。
4. BleakScanner.stop()を実行した後で、BleakScanner.discovered_devices_and_advertisement_dataプロパティからDict(Dictionaryの略)型の構造体を取り出します。

Dict型の構造体では、BTアドレスをKeyとしてAdvertisingDataを含んだTuple(タプル)型の構造体が紐づけされています。

    btevs1 = 'F5:7A:BC:3E:B5:0A' 

BTアドレスはWindows11 PCの場合、PCのBLuetooth I/Fがきちんと受信できているかどうかの確認を兼ねて知る方法があります。

これはdebugを進めるうえでとても大切なことです。
 
 

Advertising Packetの受信、data取り出し部の関数は下記です。

#  Get AdvertisementData
 
async def get_advdata():
    global label
    global scanner      # = BleakScanner()
    global prv_dsptxt

    btevs1 = 'F5:7A:BC:3E:B5:0A'
        
    print("wait AdvertisementData")
    
    await scanner.start()
    await asyncio.sleep(10.0)
    await scanner.stop()

    devices = scanner.discovered_devices_and_advertisement_data
    print("Discover device: ", devices )
    # Property: Dict[ str, Tuple[BLEdevice,AdvertisementData]]
   
   　if devices != {}:
        if btevs1 in devices:
            tp = devices[ btevs1 ]
            print("Tuple: ", tp[1])
            if tp != None:
                advd = tp[1]       # get AdvertisingData
                
                # 2912 = 0x0B60 : RATOC Systems
                vd = advd.manufacturer_data.get(2912)    
                if vd != None:
                    #print("v:", vd )
                    co2d = vd[0:0+2]
                    co2 =int.from_bytes(co2d, byteorder='little', 　　　　　　　　　　　　　　　　　　　
　　　　　　　　　　　　　　　signed=False)
                    tempd = vd[6:6+2]
                    temp =int.from_bytes(tempd, byteorder='little', 
　　　　　　　　　　　　　　　　signed=False)
                    temp = temp/10
                    hm = vd[8]
                    rsval = advd.rssi
                    dev_name = advd.local_name
                    print("RSSI:", rsval, dev_name )
                    print("CO2:", co2, "ppm Temp:", temp,"℃ ",hm, "%")
                    prv_dsptxt = "CO2 : "+str(co2)+"ppm" + "\r\n"
　　　　　　　　　　　　　　　　　　　+ " 温度:"+str(temp)+"℃ 湿度:"+str(hm) + "%"
                    now = time.strftime("%H:%M:%S")
                    print( now )
                    # dsp_txt = prv_dsptxt
                    # retuen dsp_txt                
    dsp_txt = prv_dsptxt 
    return dsp_txt

Advertising Packetを受信した後、BTアドレスをKeyにDict型のPropertyからAdvertisingDataを含んだTuple型Propertyを取り出します。そして、manufactureer_data.get()という関数を使って、上記のURLで公開されているData部を取り出します。

Byte列か何かでスッと取り出せれば簡単なのですが、Pythonならではのデータ処理や文字列処理のルールに従わねばならないので、少々複雑になっています。取り出した温度、湿度、CO2濃度などのデータは、Print()で扱えるような文字列データとして戻り値としています。途中に入っているprint()はデバッグ用です。デバッグが終わったら、頭に”#”を付けてコメントアウトしておいてください。

    
def tick():
    global root
    global label
    global prv_dsptxt

    now = time.strftime("%H:%M:%S")
    now_sec = now[6:6+2]

    if now_sec == "00":
        dsp_txt= asyncio.run( get_advdata())
        prv_dsptxt = dsp_txt
    else:
        dsp_txt = prv_dsptxt

    now = time.strftime("%H:%M:%S")
    label.config(text=dsp_txt, font=("Times",'20'))
    root.title("BTEVS1 環境データ      " + now )
    label.after(1000, tick)
    return now

1分おきにget_advdata()をcallbackしてBTEVS1からのデータを取り出す関数tick()は上のリストを参照してください。Windows 11のClockの秒が”00″になるたびにAdvertising Packet受信関数をcallbackします。BTEVS1も1分間隔でAdvertising Packetを3回(ch.37,38,39でそれぞれ1回ずつ)送信しますが、その送信タイミングがWindows 11のClockと同期していません。しかし、asyncioとawaitで受信タイミングを合わせることができます。この方法であれば、秒が”00″、分が”x0″という条件式で10分おきの温度、湿度、CO2濃度を知ることができます。

最後にGUI部を含めた全体のコントロール部を示します。context managerはtkinterの

root.mainloop()です。

# Set up Main Window
root=tk.Tk()

w = root.winfo_screenwidth()
h = root.winfo_screenheight()

w = w-450
h = h-250

now = time.strftime("%H:%M:%S")
root.title("BTEVS1 環境データ      " + now )
root.geometry("400x150+"+str(w)+"+"+str(h))

# set up main frame and put it
frame = ttk.Frame(root)
#label_frame = tk.Label(frame)
#button_frame = tk.Button(frame)

frame.pack(fill = tk.BOTH, padx=20,pady=10)

scanner = BleakScanner()
dsp_txt= asyncio.run( get_advdata())
prv_dsptxt = dsp_txt

now = time.strftime("%H:%M:%S")

label = Label(frame, text=dsp_txt, font=("Times", '20'))

print("Loop Start","\r\n")

now = tick()
label.pack()
# main loop
root.mainloop()
print("end")

BLEのAdvertising を使用したIoTセンサデバイスは、低消費電力で、connectionの必要がないので、受信側の任意のタイミングで受信することができるというメリットがあります。Data部の長さが31バイトという制約がありますが、温度や湿度、CO2濃度などのデータであれば31バイトに収めることができます。しかし、IoTデバイスからの一方通行なのでIoTデバイスの設定などを行うことができないという欠点があります。設定などを行う必要がある場合には一時的にconnectionを成立させ、GATTなどのprotocolを使用することにより行うことができます。

同じようにBLE Advertising を使用した当社製品RS-BTTHM1のAdvertisingData formatも公開していますのでお試しください。

この記事で紹介した製品

RS-BTEVS1 － Bluetooth 環境センサー