【BLE】RaspberryPi Pico W MicroPythonでBluetoothを使う方法【チュートリアル セントラル編】

import bluetooth
import random
import struct
import time
import micropython
from ble_advertising import decode_services, decode_name
from micropython import const
from machine import Pin

_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)

# 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,),
)


class BLETemperatureCentral:

    # メンバを初期化して、bleのオブジェクトをアクティブにします。
    def __init__(self, ble):
        self._ble = ble
        self._ble.active(True)
        self._ble.irq(self._irq)
        self._reset()
        self._led = Pin('LED', Pin.OUT)

    # メンバを空にします
    def _reset(self):
        
        self._name = None
        self._addr_type = None
        self._addr = None
        self._value = None
        
        self._scan_callback = None
        self._conn_callback = None
        self._read_callback = None
        self._notify_callback = None

        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
            
            if adv_type in (_ADV_IND, _ADV_DIRECT_IND):
                
                # アドバタイズのデータ内容をデコードします
                type_list = decode_services(adv_data)
                
                # スキャンしたペリフェラルが、Pico WのUUIDかどうかチェックします
                if _ENV_SENSE_UUID in type_list:

                    # アドレスのタイプを取得します
                    self._addr_type = addr_type

                    #
                    # Bluetoothアドレスを取得します。データは変わることがあるので、
                    # 実体をコピー(deep copy)が必要です。
                    #
                    self._addr = bytes(addr)
                    
                    # サービス名を取得します。
                    self._name = decode_name(adv_data) or "?"

                    # スキャンを停止します。
                    self._ble.gap_scan(None)

        #
        # スキャン完了
        #
        elif event == _IRQ_SCAN_DONE:
            
            if self._scan_callback:
                
                #　Bluetoothアドレスが取得されている(タイムアウトではない)
                if self._addr:
                    self._scan_callback(self._addr_type, self._addr, self._name)
                    self._scan_callback = None
                
                # タイムアウト
                else:
                    self._scan_callback(None, None, None)

        #
        # ペリフェラルへの接続成功
        #
        elif event == _IRQ_PERIPHERAL_CONNECT:
                       
            # dataの取得
            conn_handle, addr_type, addr = data
            
            # アドレスタイプ・MACアドレスがペリフェラルのPicoWと同じ
            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:

            # 切断したのが、接続していたペリフェアルだった場合
            conn_handle, _, _ = data
            if conn_handle == self._conn_handle:
                # メンバ変数で保持していた、接続先の情報をリセットします
                self._reset()
                
        #
        # ペリフェアルのサービスを取得
        #
        elif event == _IRQ_GATTC_SERVICE_RESULT:
            
            # dataの取得
            conn_handle, start_handle, end_handle, uuid = data
            
            # ペリフェラルのPico WのサービスのUUIDと同じ場合
            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:
            
            if self._start_handle and self._end_handle:
                
                # サービスの特徴(Characteristics)を取得する。
                self._ble.gattc_discover_characteristics(
                    self._conn_handle, self._start_handle, self._end_handle
                )
            else:
                print("Failed to find environmental sensing service.")

        #
        # ペリフェラルの特徴(Characteristic)を取得
        #
        elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
            #dataの取得           
            conn_handle, def_handle, value_handle, properties, uuid = data
            
            # 特徴(Characteristics)のUUIDが、ペリフェラルのPicoWのUUIDと同じだったら。
            if conn_handle == self._conn_handle and uuid == _TEMP_UUID:
                # 特性のValueのATTへのハンドルを保持する。
                self._value_handle = value_handle

        #
        # ペリフェラルの特徴(Characteristic)を取得完了
        #
        elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
            
            # ペリフェラルのPicoWの特性情報を取得できていたら、
            if self._value_handle:
                if self._conn_callback:
                    # コールバック関数を呼び出します。
                    self._conn_callback()
            else:
                print("Failed to find temperature characteristic.")

        #
        # 特徴(Characteristic)へのReadの習得
        # demo関数からRead実行後にイベントが発生します。
        #
        elif event == _IRQ_GATTC_READ_RESULT:
            
            # dataの取得
            conn_handle, value_handle, char_data = data

            # Read元がPico Wであったら場合。            
            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

        # READが完了した場合
        elif event == _IRQ_GATTC_READ_DONE:
            
            #　dataを取得します
            conn_handle, value_handle, status = data

        # ペリフェラルからNotifyが送られてきた場合
        elif event == _IRQ_GATTC_NOTIFY:
                        
            # dataを取得します
            conn_handle, value_handle, notify_data = data
            
            # Readと同様に、接続済みのPico Wからの物であれば、
            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)


    def is_connected(self):
        # ペリフェラルへの接続ハンドルと、特徴のValueのATTへのハンドルが取得済みであれば
        # ペリフェラルへ接続していると判定します。
        return self._conn_handle is not None and self._value_handle is not None

    # アドバタイズしているペリフェラルをスキャンします
    def scan(self, callback=None):
        self._addr_type = None
        self._addr = None
        self._scan_callback = callback
        self._ble.gap_scan(2000, 30000, 30000)

    # スキャンで取得した情報で、ペリフェラルに接続します。
    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

    # ペリフェラルと切断します
    def disconnect(self):
        if not self._conn_handle:
            return
        self._ble.gap_disconnect(self._conn_handle)
        self._reset()

    # Readを行います
    def read(self, callback):
        if not self.is_connected():
            return
        self._read_callback = callback
        try:
            self._ble.gattc_read(self._conn_handle, self._value_handle)
        except OSError as error:
            print(error)

    # notifyが着た際のコールバックを設定します(未使用)
    def on_notify(self, callback):
        self._notify_callback = callback

    # 温度情報の変換を行います
    def _update_value(self, data):
        try:
            self._value = struct.unpack("<h", data)[0] / 100
        except OSError as error:
            print(error)

    # メンバ変数Valueを取得する関数です
    def value(self):
        return self._value

# Pico本体のLEDを点滅させる関数です
def sleep_ms_flash_led(self, flash_count, delay_ms):
    self._led.off()
    while(delay_ms > 0):
        for i in range(flash_count):            
            self._led.on()
            time.sleep_ms(100)
            self._led.off()
            time.sleep_ms(100)
            delay_ms -= 200
        time.sleep_ms(1000)
        delay_ms -= 1000

def print_temp(result):
    print("read temp: %.2f degc" % result)

# 中心となる処理を行う関数です。
def demo(ble, central):
    not_found = False

    #スキャン完了時に呼び出して貰う関数です。
    #ペリフェラルの名前を表示して、接続を行います。
    def on_scan(addr_type, addr, name):
        if addr_type is not None:
            print("Found sensor: %s" % name)
            central.connect()
        else:
            nonlocal not_found
            not_found = True
            print("No sensor found.")
    
    # ペリフェラルのスキャンを行います。
    central.scan(callback=on_scan)

    # ライブラリ側が接続処理をしてくれるまで待ちます。
    while not central.is_connected():
        time.sleep_ms(100)
        if not_found:
            return

    print("Connected")

    # 接続が続いている間、ペリフェラルの温度の値を読み取ります。
    while central.is_connected():
        central.read(callback=print_temp)
        sleep_ms_flash_led(central, 2, 2000)

    print("Disconnected")

#メイン関数です
if __name__ == "__main__":
    ble = bluetooth.BLE()
    central = BLETemperatureCentral(ble)
    while(True):
        demo(ble, central)
        sleep_ms_flash_led(central, 1, 10000)

# 中心となる処理を行う関数です。
def demo(ble, central):
    not_found = False

    #スキャン完了時に呼び出して貰う関数です。
    #ペリフェラルの名前を表示して、接続を行います。
    def on_scan(addr_type, addr, name):
        if addr_type is not None:
            print("Found sensor: %s" % name)
            central.connect()
        else:
            nonlocal not_found
            not_found = True
            print("No sensor found.")
    
    # ペリフェラルのスキャンを行います。
    central.scan(callback=on_scan)

    # ライブラリ側が接続処理をしてくれるまで待ちます。
    while not central.is_connected():
        time.sleep_ms(100)
        if not_found:
            return

    print("Connected")

    # 接続が続いている間、ペリフェラルの温度の値を読み取ります。
    while central.is_connected():
        central.read(callback=print_temp)
        sleep_ms_flash_led(central, 2, 2000)

    print("Disconnected")

#メイン関数です
if __name__ == "__main__":
    ble = bluetooth.BLE()
    central = BLETemperatureCentral(ble)
    while(True):
        demo(ble, central)
        sleep_ms_flash_led(central, 1, 10000)

        #
        # スキャンの結果
        #
        if event == _IRQ_SCAN_RESULT:

            # アドバタイズの各種データを取得します
            addr_type, addr, adv_type, rssi, adv_data = data
            
            if adv_type in (_ADV_IND, _ADV_DIRECT_IND):
                
                # アドバタイズのデータ内容をデコードします
                type_list = decode_services(adv_data)
                
                # スキャンしたペリフェラルが、Pico WのUUIDかどうかチェックします
                if _ENV_SENSE_UUID in type_list:

                    # アドレスのタイプを取得します
                    self._addr_type = addr_type

                    #
                    # Bluetoothアドレスを取得します。データは変わることがあるので、
                    # 実体をコピー(deep copy)が必要です。
                    #
                    self._addr = bytes(addr)
                    
                    # サービス名を取得します。
                    self._name = decode_name(adv_data) or "?"

                    # スキャンを停止します。
                    self._ble.gap_scan(None)

  if adv_type in (_ADV_IND, _ADV_DIRECT_IND):

            # アドバタイズの各種データを取得します
            addr_type, addr, adv_type, rssi, adv_data = data
            
            if adv_type in (_ADV_IND, _ADV_DIRECT_IND):
                
                # アドバタイズのデータ内容をデコードします
                type_list = decode_services(adv_data)
                
                # スキャンしたペリフェラルが、Pico WのUUIDかどうかチェックします
                if _ENV_SENSE_UUID in type_list:

                    # アドレスのタイプを取得します
                    self._addr_type = addr_type

                    #
                    # Bluetoothアドレスを取得します。データは変わることがあるので、
                    # 実体をコピー(deep copy)が必要です。
                    #
                    self._addr = bytes(addr)
                    
                    # サービス名を取得します。
                    self._name = decode_name(adv_data) or "?"

                    # スキャンを停止します。
                    self._ble.gap_scan(None)

        #
        # スキャン完了
        #
        elif event == _IRQ_SCAN_DONE:
            
            if self._scan_callback:
                
                #　Bluetoothアドレスが取得されている(タイムアウトではない)
                if self._addr:
                    self._scan_callback(self._addr_type, self._addr, self._name)
                    self._scan_callback = None
                
                # タイムアウト
                else:
                    self._scan_callback(None, None, None)

        #
        # ペリフェラルへの接続成功
        #
        elif event == _IRQ_PERIPHERAL_CONNECT:
                       
            # dataの取得
            conn_handle, addr_type, addr = data
            
            # アドレスタイプ・MACアドレスがペリフェラルのPicoWと同じ
            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:

            # 切断したのが、接続していたペリフェアルだった場合
            conn_handle, _, _ = data
            if conn_handle == self._conn_handle:
                # メンバ変数で保持していた、接続先の情報をリセットします
                self._reset()

        #
        # ペリフェアルのサービスを取得
        #
        elif event == _IRQ_GATTC_SERVICE_RESULT:
            
            # dataの取得
            conn_handle, start_handle, end_handle, uuid = data
            
            # ペリフェラルのPico WのサービスのUUIDと同じ場合
            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:
            
            if self._start_handle and self._end_handle:
                
                # サービスの特徴(Characteristics)を取得する。
                self._ble.gattc_discover_characteristics(
                    self._conn_handle, self._start_handle, self._end_handle
                )
            else:
                print("Failed to find environmental sensing service.")

        #
        # ペリフェラルの特徴(Characteristic)を取得
        #
        elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
            #dataの取得           
            conn_handle, def_handle, value_handle, properties, uuid = data
            
            # 特徴(Characteristics)のUUIDが、ペリフェラルのPicoWのUUIDと同じだったら。
            if conn_handle == self._conn_handle and uuid == _TEMP_UUID:
                # 特性のValueのATTへのハンドルを保持する。
                self._value_handle = value_handle

        #
        # ペリフェラルの特徴(Characteristic)を取得完了
        #
        elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
            
            # ペリフェラルのPicoWの特性情報を取得できていたら、
            if self._value_handle:
                if self._conn_callback:
                    # コールバック関数を呼び出します。
                    self._conn_callback()
            else:
                print("Failed to find temperature characteristic.")

        #
        # 特徴(Characteristic)へのReadの習得
        # demo関数からRead実行後にイベントが発生します。
        #
        elif event == _IRQ_GATTC_READ_RESULT:
            
            # dataの取得
            conn_handle, value_handle, char_data = data

            # Read元がPico Wであったら場合。            
            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

        # READが完了した場合
        elif event == _IRQ_GATTC_READ_DONE:
            
            #　dataを取得します
            conn_handle, value_handle, status = data

#                         ~~~~~ 中略 ~~~~~

 # Readを行います
    def read(self, callback):
        if not self.is_connected():
            return
        self._read_callback = callback
        try:
            self._ble.gattc_read(self._conn_handle, self._value_handle)
        except OSError as error:
            print(error)

# ペリフェラルからNotifyが送られてきた場合
        elif event == _IRQ_GATTC_NOTIFY:
                        
            # dataを取得します
            conn_handle, value_handle, notify_data = data
            
            # Readと同様に、接続済みのPico Wからの物であれば、
            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)