はじめに
今回はESP32でBLEの実装作業の最終回として、本格的なBLEでESP32同士の送受信を紹介します。本文にはubluetoothを使って一つのESP32をペリフェラル(peripheral)ともう一つのESP32をセントラル(central)でBLE通信を構築しました。
前回:
ESP32でBLEの実装作業【スキャン】
ESP32でBLEの実装作業【ブロードキャスト】
環境
- IoTデバイス（ESP-WROOM-32）
- ファームウェア（MicroPython1.13）
この記事の作業を始める前にmicropythonのバージョンが1.13であることを確認してください。確認手順は下にある図を参考してください。
※バージョンがずっと更新しているので, 異なるバージョンで機能しない可能性があります．
ソースコード及び結果（ESP32）
今回はペリフェラルとなるにするプログラムble_peripheralとセントラルとなるプログラムble_central.pyを分けて説明していきます。ESP32でBLEの実装作業【ブロードキャスト】で使用したble_advertising.pyをインストールされているのが前提条件です。
ペリフェラル
ペリフェラルはここで送受信の送信の役割を持っているロールと考えられば良いです。セントラルに対するデータをパケットで送信するだけだから、内容も比較的に簡単です。
コード
ペリフェラルとしてデータをブロードキャストする時のコードは下記の通りです。ESP32でBLEの実装作業【ブロードキャスト】を見たことがあるなら分かると思いますが、内容的にble_broadcastより大きく変更がないですから、主にindicateのイベントを追加しました。indicateはサーバからクライアントに通知後、クライアントに返答が必要です[1]。
ble_peripheral.py
# This example demonstrates a simple temperature sensor peripheral.
#
# The sensor's local value updates every second, and it will notify
# any connected central every 10 seconds.
import bluetooth
import random
import struct
import time
from ble_advertising import advertising_payload
from micropython import const
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_GATTS_INDICATE_DONE = const(20)
_FLAG_READ = const(0x0002)
_FLAG_NOTIFY = const(0x0010)
_FLAG_INDICATE = const(0x0020)
# 環境センシングサービス
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
# 温度特性
_TEMP_CHAR = (
bluetooth.UUID(0x2A6E),
_FLAG_READ | _FLAG_NOTIFY | _FLAG_INDICATE,
)
_ENV_SENSE_SERVICE = (
_ENV_SENSE_UUID,
(_TEMP_CHAR,),
)
# 検測された画像様式
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
class BLETemperature:
def __init__(self, ble, name="好きな名前"):
self._ble = ble
self._ble.active(True)
self._ble.irq(self._irq)
((self._handle,),) = self._ble.gatts_register_services((_ENV_SENSE_SERVICE,))
self._connections = set()
self._payload = advertising_payload(
name=name, services=[_ENV_SENSE_UUID], appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER
)
self._advertise()
def _irq(self, event, data):
# Track connections so we can send notifications.
if event == _IRQ_CENTRAL_CONNECT:
conn_handle, _, _ = data
self._connections.add(conn_handle)
elif event == _IRQ_CENTRAL_DISCONNECT:
conn_handle, _, _ = data
self._connections.remove(conn_handle)
# Start advertising again to allow a new connection.
self._advertise()
elif event == _IRQ_GATTS_INDICATE_DONE:
conn_handle, value_handle, status = data
def set_temperature(self, temp_deg_c, notify=False, indicate=False):
# Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
# Write the local value, ready for a central to read.
self._ble.gatts_write(self._handle, struct.pack("<h", int(temp_deg_c * 100)))
if notify or indicate:
for conn_handle in self._connections:
if notify:
# Notify connected centrals.
self._ble.gatts_notify(conn_handle, self._handle)
if indicate:
# Indicate connected centrals.
self._ble.gatts_indicate(conn_handle, self._handle)
def _advertise(self, interval_us=500000):
self._ble.gap_advertise(interval_us, adv_data=self._payload)
def demo():
ble = bluetooth.BLE()
temp = BLETemperature(ble)
t = 25
i = 0
count = 0
# 二回データを出す
#while count < 2:
# Write every second, notify every 10 seconds.
#i = (i + 1) % 10
#temp.set_temperature(t, notify=i == 0, indicate=False)
# Random walk the temperature.
#t += random.uniform(-0.5, 0.5)
#print(t)
#time.sleep_ms(1000)
#count += 1
# 無限ループのオプション
while True:
# Write every second, notify every 10 seconds.
i = (i + 1) % 10
temp.set_temperature(t, notify=i == 0, indicate=False)
# Random walk the temperature.
t += random.uniform(-0.5, 0.5)
print(t)
time.sleep_ms(1000)
if __name__ == "__main__":
demo()
結果
結果は一秒ごとにパケットを発信しています。ターミナルでは下図のように表示しているはずです。
セントラル
セントラルはここで送受信の受信の役割を持っているロールと考えられば良いです。一般的セントラルを担うのはPCやスマホなどのデバイスと思いますが、今度でESP32でやることにしました。
コード
ble_scanner.py
# This example finds and connects to a BLE temperature sensor (e.g. the one in ble_temperature.py).
import ubluetooth
import random
import ustruct
import utime
import ubinascii
import micropython
from ble_advertising import decode_services, decode_name
from micropython import const
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_SERVICE_RESULT = const(9)
_IRQ_GATTC_SERVICE_DONE = const(10)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
_IRQ_GATTC_DESCRIPTOR_RESULT = const(13)
_IRQ_GATTC_DESCRIPTOR_DONE = const(14)
_IRQ_GATTC_READ_RESULT = const(15)
_IRQ_GATTC_READ_DONE = const(16)
_IRQ_GATTC_WRITE_DONE = const(17)
_IRQ_GATTC_NOTIFY = const(18)
_IRQ_GATTC_INDICATE = const(19)
_ADV_IND = const(0x00)
_ADV_DIRECT_IND = const(0x01)
_ADV_SCAN_IND = const(0x02)
_ADV_NONCONN_IND = const(0x03)
# org.bluetooth.service.environmental_sensing
_ENV_SENSE_UUID = bluetooth.UUID(0x181A)
# org.bluetooth.characteristic.temperature
_TEMP_UUID = bluetooth.UUID(0x2A6E)
_TEMP_CHAR = (
_TEMP_UUID,
bluetooth.FLAG_READ | bluetooth.FLAG_NOTIFY,
)
_ENV_SENSE_SERVICE = (
_ENV_SENSE_UUID,
(_TEMP_CHAR,),
)
# org.bluetooth.characteristic.gap.appearance.xml
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
def form_mac_address(addr: bytes) -> str:
return ":".join('{:02x}'.format(b) for b in addr)
class BLETemperatureCentral:
def __init__(self, ble):
self._ble = ble
self._ble.active(True)
self._ble.irq(self._irq)
self._reset()
def _reset(self):
# Cached name and address from a successful scan.
self._name = None
self._addr_type = None
self._addr = None
# Cached value (if we have one)
self._value = None
# Callbacks for completion of various operations.
# These reset back to None after being invoked.
self._scan_callback = None
self._conn_callback = None
self._read_callback = None
# Persistent callback for when new data is notified from the device.
self._notify_callback = None
# Connected device.
self._conn_handle = None
self._start_handle = None
self._end_handle = None
self._value_handle = None
def _irq(self, event, data):
if event == _IRQ_SCAN_RESULT:
addr_type, addr, adv_type, rssi, adv_data = data
print('type:{} addr:{} rssi:{} data:{}'.format(addr_type, ubinascii.hexlify(addr), rssi, ubinascii.hexlify(adv_data)))
if adv_type in (_ADV_IND, _ADV_DIRECT_IND) and _ENV_SENSE_UUID in decode_services(
adv_data
):
# Found a potential device, remember it and stop scanning.
self._addr_type = addr_type
self._addr = bytes(
addr
) # Note: addr buffer is owned by caller so need to copy it.
self._name = decode_name(adv_data) or "?"
self._ble.gap_scan(None)
elif event == _IRQ_SCAN_DONE:
print('Scan compelete')
if self._scan_callback:
if self._addr:
# Found a device during the scan (and the scan was explicitly stopped).
self._scan_callback(self._addr_type, self._addr, self._name)
print("callbask is:", self._scan_callback)
self._scan_callback = None
else:
# Scan timed out.
self._scan_callback(None, None, None)
elif event == _IRQ_PERIPHERAL_CONNECT:
# Connect successful.
conn_handle, addr_type, addr = data
if addr_type == self._addr_type and addr == self._addr:
self._conn_handle = conn_handle
self._ble.gattc_discover_services(self._conn_handle)
print('peripheral discovered')
elif event == _IRQ_PERIPHERAL_DISCONNECT:
# Disconnect (either initiated by us or the remote end).
conn_handle, _, _ = data
if conn_handle == self._conn_handle:
# If it was initiated by us, it'll already be reset.
self._reset()
elif event == _IRQ_GATTC_SERVICE_RESULT:
# Connected device returned a service.
conn_handle, start_handle, end_handle, uuid = data
if conn_handle == self._conn_handle and uuid == _ENV_SENSE_UUID:
self._start_handle, self._end_handle = start_handle, end_handle
elif event == _IRQ_GATTC_SERVICE_DONE:
# Service query complete.
if self._start_handle and self._end_handle:
self._ble.gattc_discover_characteristics(
self._conn_handle, self._start_handle, self._end_handle
)
else:
print("Failed to find environmental sensing service.")
elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
# Connected device returned a characteristic.
conn_handle, def_handle, value_handle, properties, uuid = data
if conn_handle == self._conn_handle and uuid == _TEMP_UUID:
self._value_handle = value_handle
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
# Characteristic query complete.
if self._value_handle:
# We've finished connecting and discovering device, fire the connect callback.
if self._conn_callback:
self._conn_callback()
else:
print("Failed to find temperature characteristic.")
elif event == _IRQ_GATTC_READ_RESULT:
# A read completed successfully.
conn_handle, value_handle, char_data = data
if conn_handle == self._conn_handle and value_handle == self._value_handle:
self._update_value(char_data)
if self._read_callback:
self._read_callback(self._value)
self._read_callback = None
elif event == _IRQ_GATTC_READ_DONE:
# Read completed (no-op).
conn_handle, value_handle, status = data
elif event == _IRQ_GATTC_NOTIFY:
# The ble_temperature.py demo periodically notifies its value.
conn_handle, value_handle, notify_data = data
if conn_handle == self._conn_handle and value_handle == self._value_handle:
self._update_value(notify_data)
if self._notify_callback:
self._notify_callback(self._value)
# Returns true if we've successfully connected and discovered characteristics.
def is_connected(self):
return self._conn_handle is not None and self._value_handle is not None
# Find a device advertising the environmental sensor service.
def scan(self, callback=None):
self._addr_type = None
self._addr = None
self._scan_callback = callback
self._ble.gap_scan(2000, 30000, 30000)
# Connect to the specified device (otherwise use cached address from a scan).
def connect(self, addr_type=None, addr=None, callback=None):
self._addr_type = addr_type or self._addr_type
self._addr = addr or self._addr
self._conn_callback = callback
if self._addr_type is None or self._addr is None:
return False
self._ble.gap_connect(self._addr_type, self._addr)
return True
# Disconnect from current device.
def disconnect(self):
if not self._conn_handle:
return
self._ble.gap_disconnect(self._conn_handle)
self._reset()
# Issues an (asynchronous) read, will invoke callback with data.
def read(self, callback):
if not self.is_connected():
return
self._read_callback = callback
self._ble.gattc_read(self._conn_handle, self._value_handle)
# Sets a callback to be invoked when the device notifies us.
def on_notify(self, callback):
self._notify_callback = callback
def _update_value(self, data):
# Data is sint16 in degrees Celsius with a resolution of 0.01 degrees Celsius.
self._value = struct.unpack("<h", data)[0] / 100
return self._value
def value(self):
return self._value
def demo():
ble = bluetooth.BLE()
central = BLETemperatureCentral(ble)
not_found = False
def on_scan(addr_type, addr, name):
if addr_type is not None:
print("Found sensor:", addr_type, addr, name)
central.connect()
else:
nonlocal not_found
not_found = True
print("No sensor found.")
central.scan(callback=on_scan)
# Wait for connection...
while not central.is_connected():
time.sleep_ms(100)
if not_found:
return
print("Connected")
# Explicitly issue reads, using "print" as the callback.
while central.is_connected():
central.read(callback=print)
time.sleep_ms(2000)
# Alternative to the above, just show the most recently notified value.
#while central.is_connected():
# print(central.value())
# time.sleep_ms(2000)
print("Disconnected")
if __name__ == "__main__":
demo()
結果
結果は下図のように接続されているペリフェラルからのデータを引き続き表示します。
最後に
今回はBLEのペリフェラルとなるプログラムble_peripheral.pyとペリフェラルから送信されたパケットを受信するセントラルプログラムble_central.pyを紹介しました。基本的にこの三つの記事のコードを使えばちゃんとESP32同士をBLEで送受信できると思います。