ラズベリーパイPico WでBluetoothのセントラル機器を構築してみる
ラズベリーパイPico WでBluetoothを使ってみるでラズベリーパイPico Wを子機(ペリフェラル)として、親機(セントラル)をAndroidのSerial Bluetooth Terminalにしてブルートゥースで接続しました。
今回は親機(セントラル)の方もラズベリーパイPico Wで構築してみます。
はじめにコードをかけるパソコンが一台しかない場合を考慮して、子機の方を電池やUSBを介してコンセントに直接挿して動作するように改修します。
でRaspberry Pi Picoの方にmain.pyを作成して、下記のコードを入力して保存します。
main.py
import bluetooth
from ble_simple_peripheral import BLESimplePeripheral
import time
ble = bluetooth.BLE()
p = BLESimplePeripheral(ble)
def on_rx(v):
print("RX", v)
p.on_write(on_rx)
i = 0
while True:
if p.is_connected():
# Short burst of queued notifications.
for _ in range(3):
data = str(i) + "_"
print("TX", data)
p.send(data)
i += 1
time.sleep_ms(100)
このコードはble_simple_peripheral.pyのdemo関数のコードをそのまま使用しています。
これでラズベリーパイPico Wはパソコンに繋がっていなくても動作するようになりました。
続いて、親機(セントラル)側のコードを作成します。
子機同様、マイコン(Raspberry Pi Pico)に直接コードを保存します。
下記URLを開き、エディタ(Thonny)にコードをペーストしble_advertising.pyというファイル名でラズベリーパイPico Wに保存します。
一応必要箇所を転機しておきます。
from micropython import const
import struct
import bluetooth
_ADV_TYPE_FLAGS = const(0x01)
_ADV_TYPE_NAME = const(0x09)
_ADV_TYPE_UUID16_COMPLETE = const(0x3)
_ADV_TYPE_UUID32_COMPLETE = const(0x5)
_ADV_TYPE_UUID128_COMPLETE = const(0x7)
_ADV_TYPE_UUID16_MORE = const(0x2)
_ADV_TYPE_UUID32_MORE = const(0x4)
_ADV_TYPE_UUID128_MORE = const(0x6)
_ADV_TYPE_APPEARANCE = const(0x19)
_ADV_MAX_PAYLOAD = const(31)
def advertising_payload(limited_disc=False, br_edr=False, name=None, services=None, appearance=0):
payload = bytearray()
def _append(adv_type, value):
nonlocal payload
payload += struct.pack("BB", len(value) + 1, adv_type) + value
_append(
_ADV_TYPE_FLAGS,
struct.pack("B", (0x01 if limited_disc else 0x02) + (0x18 if br_edr else 0x04)),
)
if name:
_append(_ADV_TYPE_NAME, name)
if services:
for uuid in services:
b = bytes(uuid)
if len(b) == 2:
_append(_ADV_TYPE_UUID16_COMPLETE, b)
elif len(b) == 4:
_append(_ADV_TYPE_UUID32_COMPLETE, b)
elif len(b) == 16:
_append(_ADV_TYPE_UUID128_COMPLETE, b)
# See org.bluetooth.characteristic.gap.appearance.xml
if appearance:
_append(_ADV_TYPE_APPEARANCE, struct.pack("<h", appearance))
if len(payload) > _ADV_MAX_PAYLOAD:
raise ValueError("advertising payload too large")
return payload
def decode_field(payload, adv_type):
i = 0
result = []
while i + 1 < len(payload):
if payload[i + 1] == adv_type:
result.append(payload[i + 2 : i + payload[i] + 1])
i += 1 + payload[i]
return result
def decode_name(payload):
n = decode_field(payload, _ADV_TYPE_NAME)
return str(n[0], "utf-8") if n else ""
def decode_services(payload):
services = []
for u in decode_field(payload, _ADV_TYPE_UUID16_COMPLETE):
services.append(bluetooth.UUID(struct.unpack("<h", u)[0]))
for u in decode_field(payload, _ADV_TYPE_UUID32_COMPLETE):
services.append(bluetooth.UUID(struct.unpack("<d", u)[0]))
for u in decode_field(payload, _ADV_TYPE_UUID128_COMPLETE):
services.append(bluetooth.UUID(u))
return services
続いて、下記URLを開き、エディタにコードをペーストしble_simple_central.pyというファイル名でラズベリーパイPico Wに保存します。
一応必要箇所を転機しておきます。
import bluetooth
import random
import struct
import time
import micropython
from ble_advertising import decode_services, decode_name
from micropython import const
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_GATTS_WRITE = const(3)
_IRQ_GATTS_READ_REQUEST = const(4)
_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)
_UART_SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
_UART_RX_CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")
_UART_TX_CHAR_UUID = bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E")
class BLESimpleCentral:
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
# 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._tx_handle = None
self._rx_handle = None
def _irq(self, event, data):
if event == _IRQ_SCAN_RESULT:
addr_type, addr, adv_type, rssi, adv_data = data
if adv_type in (_ADV_IND, _ADV_DIRECT_IND) and _UART_SERVICE_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:
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)
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)
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
print("service", data)
if conn_handle == self._conn_handle and uuid == _UART_SERVICE_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 uart 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 == _UART_RX_CHAR_UUID:
self._rx_handle = value_handle
if conn_handle == self._conn_handle and uuid == _UART_TX_CHAR_UUID:
self._tx_handle = value_handle
elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
# Characteristic query complete.
if self._tx_handle is not None and self._rx_handle is not None:
# We've finished connecting and discovering device, fire the connect callback.
if self._conn_callback:
self._conn_callback()
else:
print("Failed to find uart rx characteristic.")
elif event == _IRQ_GATTC_WRITE_DONE:
conn_handle, value_handle, status = data
print("TX complete")
elif event == _IRQ_GATTC_NOTIFY:
conn_handle, value_handle, notify_data = data
if conn_handle == self._conn_handle and value_handle == self._tx_handle:
if self._notify_callback:
self._notify_callback(notify_data)
# Returns true if we've successfully connected and discovered characteristics.
def is_connected(self):
return (
self._conn_handle is not None
and self._tx_handle is not None
and self._rx_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 self._conn_handle is None:
return
self._ble.gap_disconnect(self._conn_handle)
self._reset()
# Send data over the UART
def write(self, v, response=False):
if not self.is_connected():
return
self._ble.gattc_write(self._conn_handle, self._rx_handle, v, 1 if response else 0)
# Set handler for when data is received over the UART.
def on_notify(self, callback):
self._notify_callback = callback
def demo():
ble = bluetooth.BLE()
central = BLESimpleCentral(ble)
not_found = False
def on_scan(addr_type, addr, name):
if addr_type is not None:
print("Found peripheral:", addr_type, addr, name)
central.connect()
else:
nonlocal not_found
not_found = True
print("No peripheral found.")
central.scan(callback=on_scan)
# Wait for connection...
while not central.is_connected():
time.sleep_ms(100)
if not_found:
return
print("Connected")
def on_rx(v):
print("RX", v)
central.on_notify(on_rx)
with_response = False
i = 0
while central.is_connected():
try:
v = str(i) + "_"
print("TX", v)
central.write(v, with_response)
except:
print("TX failed")
i += 1
time.sleep_ms(400 if with_response else 30)
print("Disconnected")
if __name__ == "__main__":
demo()
今回のコードにはデモ用のコードがありますのでThonny上で実行してみます。
実行するとコンソールに
MPY: soft reboot
Found peripheral: 0 b'\xd8:\xddZ\xbc\xb9' mpy-uart
service (64, 1, 3, UUID(0x1800))
service (64, 4, 6, UUID(0x1801))
service (64, 7, 12, UUID('6e400001-b5a3-f393-e0a9-e50e24dcca9e'))
Connected
TX 0_
RX
RX
RX
TX 1_
のように出力されました。