このページでは、MESH ブロックを Python から扱うサンプルプログラムについて説明します。
必要なもの
Python から操作するための環境として、以下を使用します。
-
MESH ブロック
-
以下の説明ではボタンブロック(通知)、LED ブロック(書き込み)を使用します。
-
必要な MESH ブロックのソフトウェアバージョンは1.2.5です。詳細はブロックソフトウェアバージョンについてを参考にしてください。
-
-
Bluetooth 4.0 以上を搭載した端末
-
対応OSは、Windows 10、Mac OS、Linux です。(Python のライブラリ bleak に依存)
詳細については bleak のリポジトリを確認してください。
-
-
ソフトウェア
-
Python
-
バージョン 3.9 以上が必要です。
-
-
bleak
-
Bluetooth Low Energy を扱うための Python ライブラリです。
-
本ページでは、MESH ブロックとの通信に使用します。
-
-
準備
1. Python の実行環境がない場合、事前にセットアップを行ってください。
お手元の環境が指定のバージョン未満の場合は指定以上のバージョンにアップデートしてください。
2. お使いの端末のターミナル上で以下のコマンドを実行し、bleak ライブラリをインストールします。
pip install bleak
3. サンプルプログラムを実行する前に、MESH ブロックの電源を入れ、プログラムを動作させる PC 等のそばに置きます。電源の入れ方など、ブロックの操作については MESH ブロックの基本操作を確認してください。
サンプルプログラム 1:通知を受け取る
MESH ブロックから通知を受け取る例をボタンブロックを使って説明します。
サンプルプログラムを動かしてみる
1.以下のサンプルプログラムを「sampleButton.py」という名前で保存します。
import asyncio
from bleak import BleakClient, discover
from struct import pack
# UUID
CORE_INDICATE_UUID = ('72c90005-57a9-4d40-b746-534e22ec9f9e')
CORE_NOTIFY_UUID = ('72c90003-57a9-4d40-b746-534e22ec9f9e')
CORE_WRITE_UUID = ('72c90004-57a9-4d40-b746-534e22ec9f9e')
# Constant values
MESSAGE_TYPE_INDEX = 0
EVENT_TYPE_INDEX = 1
STATE_INDEX = 2
MESSAGE_TYPE_ID = 1
EVENT_TYPE_ID = 0
# Callback
def on_receive_notify(sender, data: bytearray):
if data[MESSAGE_TYPE_INDEX] != MESSAGE_TYPE_ID and data[EVENT_TYPE_INDEX] != EVENT_TYPE_ID:
return
if data[2] == 1:
print('Single Pressed.')
return
if data[2] == 2:
print('Long Pressed.')
return
if data[2] == 3:
print('Double Pressed.')
return
def on_receive_indicate(sender, data: bytearray):
data = bytes(data)
print('[indicate] ',data)
async def scan(prefix='MESH-100'):
while True:
print('scan...')
try:
return next(d for d in await discover() if d.name and d.name.startswith(prefix))
except StopIteration:
continue
async def main():
# Scan device
device = await scan('MESH-100BU')
print('found', device.name, device.address)
# Connect device
async with BleakClient(device, timeout=None) as client:
# Initialize
await client.start_notify(CORE_NOTIFY_UUID, on_receive_notify)
await client.start_notify(CORE_INDICATE_UUID, on_receive_indicate)
await client.write_gatt_char(CORE_WRITE_UUID, pack('<BBBB', 0, 2, 1, 3), response=True)
print('connected')
await asyncio.sleep(30)
# Finish
# Initialize event loop
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
2.以下のコマンドでサンプルプログラムを実行します。
python sampleButton.py
3.MESH ブロックとの接続が完了するとターミナル上に「connected」と表示されます。
4.ボタンブロックを 1 回押すと「Single Pressed.」、2 回連続で押すと「Double Pressed.」、長押しすると「Long Pressed.」とターミナル上に表示されることを確認します。
5.30 秒経過するとサンプルプログラムは終了します。
以降でそれぞれの処理について説明します。
スキャン(44 行目 - 46 行目、36 行目 - 41 行目)
近くにある MESH ブロックを探します。ボタンブロックは、デバイス名に「MESH-100BU」を含むかどうかで判定します。シリアルナンバーを指定することで任意のブロックを見つけることもできます。
44 行目 - 46 行目
# Scan device
device = await scan('MESH-100BU')
print('found', device.name, device.address)
36 行目 - 41 行目
while True:
print('scan...')
try:
return next(d for d in await discover() if d.name and d.name.startswith(prefix))
except StopIteration:
continue
初回通信(48 行目 - 54 行目)
スキャンで見つかったブロックと接続します。キャラクタリスティックの Notify と Indicate を有効化します。その後に、ブロック機能の有効化を行います。
# Connect device
async with BleakClient(device, timeout=None) as client:
# Initialize
await client.start_notify(CORE_NOTIFY_UUID, on_receive_notify)
await client.start_notify(CORE_INDICATE_UUID, on_receive_indicate)
await client.write_gatt_char(CORE_WRITE_UUID, pack('<BBBB', 0, 2, 1, 3), response=True)
print('connected')
通知を受け取る(18 行目 - 29 行目)
ブロックからの通知を受け取ることができます。ボタンブロックの場合は、1 回押し・長押し・2 回押しを検知すると通知が届きます。通知の内容についてはボタンブロック(MESH-100BU)をご参照ください。
def on_receive_notify(sender, data: bytearray):
if data[MESSAGE_TYPE_INDEX] != MESSAGE_TYPE_ID and data[EVENT_TYPE_INDEX] != EVENT_TYPE_ID:
return
if data[2] == 1:
print('Single Pressed.')
return
if data[2] == 2:
print('Long Pressed.')
return
if data[2] == 3:
print('Double Pressed.')
return
サンプルプログラム 2:コマンドを書き込む
MESH ブロックへの指示を書き込む例を LED ブロックを使って説明します。
サンプルプログラムを動かしてみる
1.以下の サンプルプログラムを「sampleLED.py」という名前で保存します。
import asyncio
from bleak import BleakClient, discover
from struct import pack
CORE_INDICATE_UUID = ('72c90005-57a9-4d40-b746-534e22ec9f9e')
CORE_NOTIFY_UUID = ('72c90003-57a9-4d40-b746-534e22ec9f9e')
CORE_WRITE_UUID = ('72c90004-57a9-4d40-b746-534e22ec9f9e')
# Callback
def on_receive(sender, data: bytearray):
data = bytes(data)
print(data)
async def scan(prefix = 'MESH-100'):
while True:
print('scan...')
try:
return next(d for d in await discover() if d.name and d.name.startswith(prefix))
except StopIteration:
continue
async def main():
# Scan device
device = await scan('MESH-100LE')
print('found', device.name, device.address)
# Connect device
async with BleakClient(device, timeout=None) as client:
# Initialize
await client.start_notify(CORE_NOTIFY_UUID, on_receive)
await client.start_notify(CORE_INDICATE_UUID, on_receive)
await client.write_gatt_char(CORE_WRITE_UUID, pack('<BBBB', 0, 2, 1, 3), response=True)
print('connected')
# Generate command
messagetype = 1
red = 2
green = 8
blue = 32
duration = 5 * 1000 # 5,000[ms]
on = 1 * 1000 # 1,000[ms]
off = 500 # 500[ms]
pattern = 1 # 1:blink, 2:firefly
command = pack('<BBBBBBBHHHB', messagetype, 0, red, 0, green, 0, blue, duration, on, off, pattern)
checksum = 0
for x in command:
checksum += x
command = command + pack('B', checksum & 0xFF) # Add check sum to byte array
print('command ',command)
try:
# Write command
await client.write_gatt_char(CORE_WRITE_UUID, command, response=True)
except Exception as e:
print('error', e)
return
await asyncio.sleep(duration / 1000)
# Finish
# Initialize event loop
if __name__ == '__main__':
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
2.以下のコマンドでサンプルプログラムを実行します。
python sampleLED.py
3.MESH ブロックとの接続が完了するとターミナル上に「connected」と表示されます。
4.LED が淡い青色で 1 秒点灯、0.5 秒消灯、1 秒点灯、0.5 秒消灯を繰り返します。
5.LED が点灯してから 5 秒経過するとサンプルプログラムは終了します。
以降では書き込み処理について説明します。スキャン、初期通信についてはサンプルプログラム 1 をご参照ください。
コマンドを作成する(35 行目 - 49 行目)
ブロックへ書き込むコマンドを作成します。コマンドの種類や設定できるパラメーターについては LED ブロック(MESH-100LE)をご参照ください。コマンドには必ずチェックサムを付ける必要があります。
# Generate command
messagetype = 1
red = 2
green = 8
blue = 32
duration = 5 * 1000 # 5,000[ms]
on = 1 * 1000 # 1,000[ms]
off = 500 # 500[ms]
pattern = 1 # 1:blink, 2:firefly
command = pack('<BBBBBBBHHHB', messagetype, 0, red, 0, green, 0, blue, duration, on, off, pattern)
checksum = 0
for x in command:
checksum += x
command = command + pack('B', checksum & 0xFF) # Add check sum to byte array
print('command ',command)
コマンドを書き込む(52 行目 - 53 行目)
作成したコマンドをブロックに対して書き込みます。
# Write command
await client.write_gatt_char(CORE_WRITE_UUID, command, response=True)