ラズベリーパイPico WでBluetoothを使ってみる

ラズベリーパイPico WでBluetooth(ブルートゥース)を使ってみます。

Bluetoothには大きく分けて二種類あり、Bluetooth ClassicとBluetooth Low Energy(BLE)があり、ラズベリーパイPico Wで使用できるのは後者のBLEになります。

Bluetooth Low Energy - Wikipedia


ラズベリーパイPico WのBluetoothの最低限の動作をする為に、Bluetoothに関する最低限の知識について触れておきます。

※以後はBluetoothをBLEと称します。



パソコン(スマホやタブレットを含む)とイヤホンをBLEで繋ぐ時、パソコンの方を親機(セントラル : Central)とし、イヤホンの方を子機(ペリフェラル : Peripheral)とします。


セントラル機器とペリフェラル機器がBLEで接続されたら、GATT(Generic Access Profile)通信でデータのやり取りをします。


セントラル機器とペリフェラル機器がBLEで接続をする時、ペリフェラル機器の方がアドバタイズ(Advertise)と呼ばれる無線信号を送り、セントラル機器がペリフェラル機器を発見して、接続の処理を始めます。


上記の内容で太字にした箇所を参考にすれば、公式で配布されているサンプルコードを使用することができるようになりますので、これからサンプルコードを試してみます。



micropython/examples/bluetooth at master · micropython/micropythonにBLEのサンプルコードがあるので試してみます。


エディタはラズベリーパイPico Wを使ってみるで紹介していますThonnyを使用します。


最初に下記URLを開き、エディタにコードをペーストしble_advertising.pyというファイル名でラズベリーパイPico Wに保存します。

ble_advertising.py


一応必要箇所を転機しておきます。

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


コードを保存する時は、Raspberry Pi Picoの方を選択して、マイコン内に直接保存します。

このコンピュータの方を選んでも良いのですが、後述するコード内のimportでble_advertising.pyを読み込む時、マイコンによってはこのコンピュータの方を選択してコードを保存した場合、外部ファイル読み込みを含むコードを実行した時にエラーになる現象に陥り、サンプルコードが試せないことがありました。



続いて、下記URLを開き、エディタにコードをペーストしble_simple_peripheral.pyというファイル名でラズベリーパイPico Wに保存します。

ble_simple_peripheral.py


一応必要箇所を転機しておきます。

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_WRITE = const(3)

_FLAG_READ = const(0x0002)
_FLAG_WRITE_NO_RESPONSE = const(0x0004)
_FLAG_WRITE = const(0x0008)
_FLAG_NOTIFY = const(0x0010)

_UART_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
_UART_TX = (
    bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E"),
    _FLAG_READ | _FLAG_NOTIFY,
)
_UART_RX = (
    bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E"),
    _FLAG_WRITE | _FLAG_WRITE_NO_RESPONSE,
)
_UART_SERVICE = (
    _UART_UUID,
    (_UART_TX, _UART_RX),
)


class BLESimplePeripheral:
    def __init__(self, ble, name="mpy-uart"):
        self._ble = ble
        self._ble.active(True)
        self._ble.irq(self._irq)
        ((self._handle_tx, self._handle_rx),) = self._ble.gatts_register_services((_UART_SERVICE,))
        self._connections = set()
        self._write_callback = None
        self._payload = advertising_payload(name=name, services=[_UART_UUID])
        self._advertise()

    def _irq(self, event, data):
        # Track connections so we can send notifications.
        if event == _IRQ_CENTRAL_CONNECT:
            conn_handle, _, _ = data
            print("New connection", conn_handle)
            self._connections.add(conn_handle)
        elif event == _IRQ_CENTRAL_DISCONNECT:
            conn_handle, _, _ = data
            print("Disconnected", conn_handle)
            self._connections.remove(conn_handle)
            # Start advertising again to allow a new connection.
            self._advertise()
        elif event == _IRQ_GATTS_WRITE:
            conn_handle, value_handle = data
            value = self._ble.gatts_read(value_handle)
            if value_handle == self._handle_rx and self._write_callback:
                self._write_callback(value)

    def send(self, data):
        for conn_handle in self._connections:
            self._ble.gatts_notify(conn_handle, self._handle_tx, data)

    def is_connected(self):
        return len(self._connections) > 0

    def _advertise(self, interval_us=500000):
        print("Starting advertising")
        self._ble.gap_advertise(interval_us, adv_data=self._payload)

    def on_write(self, callback):
        self._write_callback = callback


def demo():
    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)


if __name__ == "__main__":
    demo()

こちらのコードも後の改修を考えて、マイコンの方に直接保存しておきます。

今回のコードにはデモ用のコードがありますのでThonny上で実行してみます。


実行するとコンソールに

MPY: soft reboot
Starting advertising

のように出力されます。



ペリフェラル機器になりましたラズベリーパイPico Wに接続する準備をします。

Androidのスマホ(やタブレット)、クロームブックがある場合はGoogle Play経由でSerial Bluetooth Terminalをインストールします。

Serial Bluetooth Terminal - Google Playのアプリ


Serial Bluetooth Terminalをインストールできましたら、スマホ等のセントラル側のマシンでBluetoothをOnにした後にSerial Bluetooth Terminalを起動します。



画面左上の≡のアイコンをタップし、



Devicesをタップします。



Bluetooth LEの方を開き、表示されているペリフェラル機器を選択します。



ペリフェラル機器と無事に接続すると上のキャプチャのように数字の羅列が出力されます。


Thonnyのコンソールを確認してみると、

New connection 64
TX 0_
TX 1_
/** 途中省略 **/
TX 419_
Disconnected 64
Starting advertising

のような内容が出力されています。

同じカテゴリーの記事
マインクラフト用ビジュアルエディタを開発しています。
詳しくはinunosinsi/mcws_blockly - githubをご覧ください。

About

齋藤毅

Webデザイナとプログラマが共働できるフレームワークとCMSを開発しています。
SOY2フレームワーク

NPO法人でプログラミング教育に関わっています。
子どもと大人が学び合う現代の寺子屋 Tera school

個人事業主としての紹介については下記の記事に記載があります。
私にとっての農業とSOY Shop