ラズパイPicoWペアリング不要簡単Bluetooth(BLE)通信の使い方

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

# UUIDの設定やキャラクタリスティックの設定、各サービスの開始
# --- 省略（サンプルプログラム参照）---

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

# データ送信（接続先の「connection」に対して「data」を送信）
await tx_char.notify(connection, data.encode())  # データをエンコードして送信

# データ受信（接続先の「connection」から「tx_char」に書き込まれたデータを「value」に格納）
connection, value = await rx_char.written() # 受信処理（受信データが来るまでここで待機する）

# pico_ble_temp.py (micropython)
# BLE接続の基本設定完了後、Pico基板内蔵の温度センサの値を取得して1秒ごとに送信

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

UART_SERVICE_UUID = bluetooth.UUID("6E400001-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, notify=True)  # UART送信用キャラクタリスティックを定義

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

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

# Pico内蔵温度センサーから温度を取得
def getTemp():
    sensor_temp = ADC(4)
    conversion_factor = 3.3 / 65535

    reading = sensor_temp.read_u16()
    voltage = reading * conversion_factor
    temperature = 27 - (voltage - 0.706) / 0.001721
    print("Temperature: {:.2f} °C".format(temperature))
    return temperature

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

                    try:  # データを送信（try:～内で実行しないと送信できない）
                        temp = getTemp()  # 温度センサーから温度を取得
                        data = f"{temp:.2f}℃\n"  # 小数点以下2桁にフォーマット
                        await tx_char.notify(connection, data.encode())  # データをエンコードして送信
                    except Exception:
                        pass  # エラー（Notify Error: 'NoneType' object isn't iterable）は無視して続行
                    
                    await asyncio.sleep(1)  # 1秒ごとにデータを送信
        except Exception as e:  # アドバタイズエラー処理
            print("Error:", e)  # エラーが発生した場合の処理
        await asyncio.sleep_ms(1000)  # 1秒待機して再試行

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

UART_SERVICE_UUID = bluetooth.UUID("6E400001-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, notify=True)  # UART送信用キャラクタリスティックを定義

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

# Pico内蔵温度センサーから温度を取得
def getTemp():

# BLE Peripheralとして動作するメインループ
async def ble_notify_loop():
    while True:
        try:  # BLE接続を待機
            .
            ここでアドバタイズ（広告）が実行され、接続が開始されたら処理を実行
            アドバタイズに失敗しても1秒後に再試行される
            .
        except Exception as e:  # アドバタイズエラー処理
            print("Error:", e)  # エラーが発生した場合の処理
        await asyncio.sleep_ms(1000)  # 1秒待機して再試行

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

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

while connection.is_connected():  # 接続が維持されている間ループ
    led.value(1)  # 接続されたらLED点灯

    try:  # データを送信（try:～内で実行しないと送信できない）
        temp = getTemp()  # 温度センサーから温度を取得
        data = f"{temp:.2f}℃\n"  # 小数点以下2桁にフォーマット
        await tx_char.notify(connection, data.encode())  # データをエンコードして送信
    except Exception:
        pass  # エラー（Notify Error: 'NoneType' object isn't iterable）は無視して続行
                    
   　await asyncio.sleep(1)  # 1秒ごとにデータを送信

# pico_ble_uart_echo.py
# BLEで受信したデータをそのまま返す（コールバック）

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

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として出力に設定

# 非同期で受信タスクを実行
async def handle_rx(connection):
    # 受信待ちループ
    while connection.is_connected():
        try:
            connection, value = await rx_char.written() # 受信処理（受信データが来るまでここで待機する）
            print("受信:", value)  # 受信したデータを表示

            # 受信したデータをそのまま通知（送信）
            try:
                if connection and connection.is_connected():  # 接続が有効な場合のみ送信
                    print("送信:", value) # 先に出力しておく（この後エラーになるため）
                    await tx_char.notify(connection, value) # 通知（送信）は成功するがエラーになる
                else:
                    print("通知対象が無効です:", connection)  # 接続が無効な場合の処理
            except Exception:
                pass  # エラー（Notify Error: 'NoneType' object isn't iterable）は無視して続行
        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)  # 接続されたデバイス情報を表示
                led.value(1)

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

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

                # 接続解除 → タスクキャンセル
                rx_task.cancel()
                led.value(0)  # LED消灯
                print("接続解除されました")
        except Exception as e:
            print("アドバタイズエラー:", repr(e))
        await asyncio.sleep_ms(200)
# PicoのBLE Peripheralとして動作開始
asyncio.run(ble_uart_echo())    

# 非同期で受信タスクを実行
async def handle_rx(connection):
    # 受信待ちループ
    while connection.is_connected():
        try:
            connection, value = await rx_char.written() # 受信処理（受信データが来るまでここで待機する）
            print("受信:", value)  # 受信したデータを表示

            # 受信したデータをそのまま通知（送信）
            try:
                if connection and connection.is_connected():  # 接続が有効な場合のみ送信
                    print("送信:", value) # 先に出力しておく（この後エラーになるため）
                    await tx_char.notify(connection, value) # 通知（送信）は成功するがエラーになる
                else:
                    print("通知対象が無効です:", connection)  # 接続が無効な場合の処理
            except Exception:
                pass  # エラー（Notify Error: 'NoneType' object isn't iterable）は無視して続行
        except Exception as e:
            print("受信/送信エラー:", e) 
            break

import asyncio
import csv
import datetime
import threading
import tkinter as tk
from tkinter import ttk, scrolledtext
from bleak import BleakClient, BleakScanner

# UUID定義
UART_SERVICE_UUID = "6E400001-B5A3-F393-E0A9-E50E24DCCA9E"
UART_RX_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
UART_TX_CHAR_UUID = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"

class BLEUARTApp:
    def __init__(self, root):
        self.client = None
        self.csv_file = None
        self.recording = False
        self.loop = asyncio.new_event_loop()
        threading.Thread(target=self.loop.run_forever, daemon=True).start()

        self.setup_gui(root)
        root.protocol("WM_DELETE_WINDOW", self.on_close)

    def setup_gui(self, root):
        root.title("BLE UART Communication")
        root.geometry("400x350")

        style = ttk.Style()
        style.configure("Right.TCombobox", justify='right')  # テキストを右寄せ

        
        # デバイス名選択用コンボボックス（setup_guiの中）
        tk.Label(root, text="デバイス名：", anchor="e").grid(row=0, column=0, sticky="e")
        self.device_name_var = tk.StringVar()
        self.device_combo = ttk.Combobox(root, textvariable=self.device_name_var)
        self.device_combo.grid(row=0, column=1, columnspan=2, sticky="ew")

        # 送信テキストの改行オプション
        self.send_append_lf = tk.BooleanVar(value=True)    # 初期状態でLFがオン
        self.send_append_cr = tk.BooleanVar(value=False)
        line_control_frame = tk.Frame(root)
        line_control_frame.grid(row=0, column=3, sticky="n", padx=5)

        tk.Checkbutton(line_control_frame, text="LF", variable=self.send_append_lf).pack(anchor="w")
        tk.Checkbutton(line_control_frame, text="CR", variable=self.send_append_cr).pack(anchor="w")

        # 送信テキスト
        self.send_entry = tk.Entry(root)
        self.send_entry.grid(row=1, column=0, columnspan=4, sticky="ew")
        self.send_entry.bind("<Return>", lambda e: self.send_data())

        # 受信テキストエリア + スクロール
        self.recv_text = scrolledtext.ScrolledText(root)
        self.recv_text.grid(row=2, column=0, columnspan=4, sticky="nsew")

        # 操作ボタン
        self.scan_btn = tk.Button(root, text="スキャン", width=10, command=self.scan_devices)
        self.scan_btn.grid(row=3, column=0)
        self.connect_btn = tk.Button(root, text="接続", width=10, command=self.connect_device)
        self.connect_btn.grid(row=3, column=1)
        self.disconnect_btn = tk.Button(root, text="切断", width=10, command=self.disconnect_device)
        self.disconnect_btn.grid(row=3, column=2)
        self.send_btn = tk.Button(root, text="送信", width=10, command=self.send_data)
        self.send_btn.grid(row=3, column=3)

        # 記録制御ボタン
        self.record_btn = tk.Button(root, text="記録", width=10, command=self.start_recording)
        self.record_btn.grid(row=4, column=0)
        self.stop_btn = tk.Button(root, text="停止", width=10, command=self.stop_recording)
        self.stop_btn.grid(row=4, column=1)
        self.clear_btn = tk.Button(root, text="削除", width=10, command=lambda: self.recv_text.delete(1.0, tk.END))
        self.clear_btn.grid(row=4, column=2)

        # レイアウト調整用の空白行
        tk.Frame(root, height=10).grid(row=5, column=0, columnspan=4)

        root.grid_columnconfigure(0, weight=1)
        root.grid_columnconfigure(1, weight=1)
        root.grid_columnconfigure(2, weight=1)
        root.grid_columnconfigure(3, weight=1)
        root.grid_rowconfigure(2, weight=1)

    def log_to_gui(self, message):
        timestamp = datetime.datetime.now().strftime("%H:%M:%S")
        formatted = f"[{timestamp}] {message}"
        self.recv_text.insert(tk.END, formatted + "\n")
        self.recv_text.yview(tk.END)
        print(message)  # コンソールへも出力

    async def handle_notifications(self, sender, data):
        try:
            text = data.decode()
            lines = text.split("\n")
            timestamp = datetime.datetime.now().strftime("%H:%M:%S") 

            for line in lines:
                if not line.strip():
                    continue

                formatted_line = f"[{timestamp}] {line.strip()}"
                self.recv_text.insert(tk.END, formatted_line + "\n")
                self.recv_text.yview(tk.END)

                if self.recording and self.csv_filename:
                    try:
                        with open(self.csv_filename, "a", newline="", encoding="utf-8") as f:
                            writer = csv.writer(f)
                            fields = [timestamp] + [float(x) if x.replace(".", "", 1).isdigit() else x for x in line.split(",")]
                            writer.writerow(fields)
                    except Exception as e:
                        self.log_to_gui(f"記録中のエラー: {e}")
        except Exception as e:
            self.log_to_gui(f"受信処理エラー: {e}")

    def scan_devices(self):
        asyncio.run_coroutine_threadsafe(self.async_scan(), self.loop) 

    async def async_scan(self):
        try:
            self.log_to_gui("BLEデバイスのスキャン開始")
            devices = await BleakScanner.discover(timeout=5.0)
            names = sorted({d.name for d in devices if d.name})  # 重複除去 + ソート
            if names:
                self.device_combo['values'] = names
                self.device_combo.set(names[0])  # 最初の項目を初期選択
                self.log_to_gui(f"見つかったデバイス一覧: {names}")
            else:
                self.device_combo.set("")
                self.device_combo['values'] = []
                self.log_to_gui(f"名前付きデバイスが見つかりません")
        except Exception as e:
            self.log_to_gui(f"スキャンエラー: {e}")

    async def connect(self, device_name):
        try:
            if self.client:
                await self.client.disconnect()

            self.log_to_gui(f" {device_name}へ接続中...")
            devices = await BleakScanner.discover()
            target = next((d for d in devices if d.name == device_name), None)
            if target is None:
                self.log_to_gui(f"デバイスが見つかりません")
                return

            self.client = BleakClient(target)
            await self.client.connect()
            await self.client.start_notify(UART_TX_CHAR_UUID, self.handle_notifications)
            self.log_to_gui(f"接続成功")
        except Exception as e:
            self.log_to_gui(f"接続エラー: {e}")

    def connect_device(self):
        device_name = self.device_name_var.get()
        if not device_name:
            return
        asyncio.run_coroutine_threadsafe(self.connect(device_name), self.loop)

    def send_data(self):
        data = self.send_entry.get()

        # 末尾追加処理
        suffix = ""
        if self.send_append_cr.get():
            suffix += "\r"
        if self.send_append_lf.get():
            suffix += "\n"
        data += suffix

        if not self.client or not self.client.is_connected:
            return

        asyncio.run_coroutine_threadsafe(
            self.client.write_gatt_char(UART_RX_CHAR_UUID, data.encode()),
            self.loop
        )
        # self.send_entry.delete(0, tk.END) # 送信後に入力欄をクリアしないようにコメントアウト

    def start_recording(self):
        timestamp = datetime.datetime.now().strftime("%y%m%d_%H%M%S")
        self.csv_filename = f"{timestamp}.csv"
        self.recording = True

        # 空のCSVファイルを作成
        try:
            with open(self.csv_filename, "a", newline="", encoding="utf-8") as f:
                pass  # 何も書き込まずに即閉じる（ヘッダーを追加する場合は無効に）
                # writer = csv.writer(f)  # ヘッダー行を追加する場合は以下を有効に
                # writer.writerow(["timestamp", "受信データ"])  # ヘッダーの内容
        except Exception as e:
            self.log_to_gui(f"記録ファイル作成エラー: {e}")
            return

        self.log_to_gui(f"記録開始: {self.csv_filename}")

    def stop_recording(self):
        self.recording = False
        self.csv_filename = None
        self.log_to_gui("記録停止")

    def disconnect_device(self):
        if self.client and self.client.is_connected:
            asyncio.run_coroutine_threadsafe(self.client.disconnect(), self.loop)
            self.log_to_gui("接続を解除しました")
            self.client = None
        else:
            self.log_to_gui("現在接続されていません")

    def on_close(self):
        asyncio.run_coroutine_threadsafe(self.cleanup(), self.loop)
        self.loop.call_soon_threadsafe(self.loop.stop)
        root.quit()  # GUIイベントループを終了

    async def cleanup(self):
        if self.client:
            try:
                await self.client.disconnect()
                print("接続解除")
            except Exception as e:
                print("切断時エラー:", e)
        self.stop_recording()
        root.destroy()

# アプリ起動
root = tk.Tk()
app = BLEUARTApp(root)
root.mainloop()

pip install bleak

cd Documents\python  

python BLE_tkinter.py