ラズパイPicoW同士で簡単Bluetooth(BLE)通信

# ble_peripheral_led.py
# BLEで受信したコマンド（文字列）に応じて処理を実行（ここではLEDの点灯／消灯）

from machine import Pin     # PicoのGPIOモジュール
import uasyncio as asyncio  # 非同期処理ライブラリ
import aioble               # MicroPythonのBLEライブラリ
import bluetooth            # MicroPythonのBLEモジュール

UART_SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")  # UARTサービスUUID
UART_RX_CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")  # UART受信用キャラクタリスティックUUID
UART_TX_CHAR_UUID = bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E")  # UART送信用キャラクタリスティックUUID

# サービスとキャラクタリスティックの定義
uart_service = aioble.Service(UART_SERVICE_UUID)  # UARTサービスを定義
tx_char = aioble.Characteristic(uart_service, UART_TX_CHAR_UUID, read=True, notify=True)  # UART送信用キャラクタリスティックを定義
rx_char = aioble.Characteristic(uart_service, UART_RX_CHAR_UUID, write=True, capture=True)  # UART受信用キャラクタリスティックを定義

aioble.register_services(uart_service)  # GATTサービスを登録

led = Pin("LED", Pin.OUT)  # LED端子をledとして出力に設定

# 受信コマンド処理完了メッセージ送信（通知：Notify）
async def write_result(connection, message):
    print("処理結果送信:", message)
    try:
        if connection and connection.is_connected():  # 接続が有効な場合のみ実行
            await tx_char.notify(connection, message.encode('utf-8')) # セントラルへメッセージを送信（通知）
        else:
            print("通知対象が無効です:", connection)  # 接続が無効な場合の処理
    except Exception as e:  # エラー処理
        if "'NoneType' object isn't iterable" in str(e):  # このエラーは無視
            pass
        else:
            print("Notifyエラー", e)
        
# 非同期で受信タスクを実行
async def handle_rx(connection):
    # 受信待ちループ
    while connection.is_connected():  # 接続が有効な間ループ
        try:
            connection, value = await rx_char.written() # 受信処理（受信データが来るまでここで待機する）
            string = value.decode ('utf-8')  # バイト文字を文字列に変換
            print("受信コマンド:", string)
            
            if string == "LED_ON":  # 受信した文字列が「LED_ON」なら
                led.value(1)  # LED点灯
                await write_result(connection, "ON!")  # 受信コマンド処理完了メッセージ送信（通知：Notify）
            elif string == "LED_OFF":  # 受信した文字列が「LED_OFF」なら
                led.value(0)  # LED消灯
                await write_result(connection, "OFF!")  # 受信コマンド処理完了メッセージ送信（通知：Notify）
            else:
                await write_result(connection, "Unknown command!")  # 未登録コマンドの場合のメッセージ送信（書込）
                
        except Exception as e:  # エラー処理
            print("エラー:", e) 
            break

# BLE Peripheralとして動作するメインループ
async def ble_uart_echo():
    while True:
        led.value(0)  # 接続待機中はLED消灯
        try:
            async with await aioble.advertise(  # BLEアドバタイズを開始
                250_000,                        # アドバタイズ間隔（マイクロ秒）
                name = "RPi-Pico",              # アドバタイズ名（BLE接続デバイス名）
                services = [UART_SERVICE_UUID], # サービスUUIDを指定
            ) as connection: # 接続待機
                print("接続中:", connection.device)  # 接続されたデバイス情報を表示

                # 受信処理を別タスクで開始
                rx_task = asyncio.create_task(handle_rx(connection))

                # 接続が切れるまで待機
                while connection.is_connected():
                    await asyncio.sleep_ms(100)  # 100ms待機

                # 接続解除（タスクキャンセル）
                rx_task.cancel()
                print("接続解除されました")
        except Exception as e:
            print("アドバタイズエラー:", repr(e))
        await asyncio.sleep_ms(200)  # 200ms待機してから再アドバタイズ

# PicoのBLE Peripheralとして動作開始
asyncio.run(ble_uart_echo())

# ble_central_button.py
# ボタンを押すとBLEでコマンド（文字列）送信（ここではBTN1を押すと「LED ON」送信、BTN1を押すと「LED OFF」送信）

from machine import Pin     # PicoのGPIO（標準入出力）モジュール
import uasyncio as asyncio  # 非同期処理ライブラリ
import aioble               # MicroPythonのBLEライブラリ
import bluetooth            # MicroPythonのBLEモジュール

device_name = "RPi-Pico"  # 接続したいデバイス名

# UUIDの定義
UART_SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")  # UARTサービスUUID
UART_RX_CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")  # UART受信用キャラクタリスティックUUID
UART_TX_CHAR_UUID = bluetooth.UUID("6E400003-B5A3-F393-E0A9-E50E24DCCA9E")  # UART送信用キャラクタリスティックUUID

# 入力端子設定
BTN1 = Pin(20, Pin.IN, Pin.PULL_UP)  # GP20をBTN1として入力端子（プルアップ）に設定
BTN2 = Pin(16, Pin.IN, Pin.PULL_UP)  # GP16をBTN2として入力端子（プルアップ）に設定

led = Pin("LED", Pin.OUT)  # LED端子をledとして出力に設定

# デバイススキャン実行
async def find_device():
    async with aioble.scan(5000, interval_us=30000, window_us=30000, active=True) as scanner: 
        async for result in scanner:  # スキャン結果を順番に処理
            print("見つかったデバイス:", result.name(), result.device)
            if result.name() == device_name:  # デバイス名が見つかったら
                return result.device  # デバイス情報を返す
    return None  # 見つからなかったらNoneを返す

# コマンド送信
async def write_command(command, rx_char):
    print("送信コマンド:", command)
    await rx_char.write(command.encode('utf-8'))  # コマンド送信実行

# Notifyを受信して表示
async def handle_notify(tx_char):
    while True:
        try:
            data = await tx_char.notified()  # Notifyを待機
            print("Notify受信:", data.decode('utf-8'))  # 受信データを表示
        except Exception as e:
            print("Notifyエラー:", e)
            break

# デバイスに接続実行（リトライあり）
async def connect_to_device():
    while True:
        led.value(0)  # LED消灯（未接続状態）

        device = await find_device()  # デバイススキャン実行
        if not device:  # デバイスが見つからなかったら
            print(device, "が見つかりません。再スキャンします...")
            await asyncio.sleep(3)  # 3s待機
            continue  # 再スキャンへ

        try:
            print(device_name, "に接続開始...")
            connection = await device.connect()  # デバイスに接続
            if connection:  # 接続に成功したら
                print("接続済")
                led.value(1)  # LED点灯（接続成功）
                return connection  # 接続成功したら返す
            else:
                print("接続に失敗しました。再試行します...")
        except Exception as e:  # 接続エラー処理
            print("接続エラー:", e)  # エラーメッセージ表示

        await asyncio.sleep(3)  # 少し待ってから再試行

# メイン処理
async def main():
    btn_state1 = False # BTN1ボタン状態格納用
    btn_state2 = False # BTN2ボタン状態格納用

    while True:
        led.value(0)  # LED消灯
        connection = await connect_to_device()  # デバイスに接続（リトライあり）

        async with connection:  # 接続コンテキスト内で処理
            try:
                uart_service = await connection.service(UART_SERVICE_UUID)     # UART接続のサービスを取得
                rx_char = await uart_service.characteristic(UART_RX_CHAR_UUID) # ペリフェラルのRX受信用キャラクタリスティック取得
                tx_char = await uart_service.characteristic(UART_TX_CHAR_UUID) # ペリフェラルのTX送信用キャラクタリスティック取得

                await tx_char.subscribe()  # Notifyを購読
                notify_task = asyncio.create_task(handle_notify(tx_char))  # Notify受信タスク開始

            except Exception as e:
                print("サービスとキャラクタリスティックの取得に失敗", e)
                return

            # 接続中の処理ループ
            while connection.is_connected():  # 接続が維持されている間
                if BTN1.value() == 0 and not btn_state1:  # BTN1が押されたら
                    btn_state1 = True  # ボタン1の状態をTrueへ
                    await write_command("LED_ON", rx_char)  # コマンド「LED_ON」送信
                if BTN1.value() == 1 and btn_state1:  # BTN1が離されたら
                    btn_state1 = False  # ボタン1の状態をFalseへ

                if BTN2.value() == 0 and not btn_state2:  # BTN2が押されたら
                    btn_state2 = True  # ボタン2の状態をTrueへ
                    await write_command("LED_OFF", rx_char)  # コマンド「LED_OFF」送信
                if BTN2.value() == 1 and btn_state2:  # BTN2が離されたら
                    btn_state2 = False  # ボタン2の状態をFalseへ

                await asyncio.sleep_ms(100)  # 100ms待機

            notify_task.cancel()  # 接続が切れたらNotifyタスクをキャンセル

asyncio.run(main())  # メイン処理実行

async with await aioble.advertise(  # BLEアドバタイズを開始
    250_000,                        # アドバタイズ間隔（マイクロ秒）
    name = "RPi-Pico",              # アドバタイズ名（BLE接続デバイス名）
    services = [UART_SERVICE_UUID], # サービスUUIDを指定
) as connection: # 接続待機
    print("接続中:", connection.device)  # 接続されたデバイス情報を表示

# 受信処理を別タスクで開始
rx_task = asyncio.create_task(handle_rx(connection))

connection, value = await rx_char.written() # 受信処理（受信データが来るまでここで待機する）

# 受信コマンド処理完了メッセージ送信（通知：Notify）
async def write_result(connection, message):
    print("処理結果送信:", message)
    try:
        if connection and connection.is_connected():  # 接続が有効な場合のみ実行
            await tx_char.notify(connection, message.encode('utf-8')) # セントラルへメッセージを送信（通知）
        else:
            print("通知対象が無効です:", connection)  # 接続が無効な場合の処理

except Exception as e:  # エラー処理
    if "'NoneType' object isn't iterable" in str(e):  # このエラーは無視
        pass
    else:
        print("Notifyエラー", e)

connection = await connect_to_device()  # デバイスに接続（リトライあり）

# デバイススキャン実行
async def find_device():
    async with aioble.scan(5000, interval_us=30000, window_us=30000, active=True) as scanner: 
        async for result in scanner:  # スキャン結果を順番に処理
            print("見つかったデバイス:", result.name(), result.device)
            if result.name() == device_name:  # デバイス名が見つかったら
                return result.device  # デバイス情報を返す
    return None  # 見つからなかったらNoneを返す

uart_service = await connection.service(UART_SERVICE_UUID)     # UART接続のサービスを取得
rx_char = await uart_service.characteristic(UART_RX_CHAR_UUID) # ペリフェラルのRX受信用キャラクタリスティック取得
tx_char = await uart_service.characteristic(UART_TX_CHAR_UUID) # ペリフェラルのTX送信用キャラクタリスティック取得

await tx_char.subscribe()  # Notifyを購読

notify_task = asyncio.create_task(handle_notify(tx_char))  # Notify受信タスク開始

data = await tx_char.notified()  # Notifyを待機

await write_command("LED_ON", rx_char)  # コマンド「LED_ON」送信