`aioble` --- 非同期 BLE#

aioble は bluetooth モジュールを高レベルで asyncio に対応させたラッパーです。スキャン、接続、アドバタイズ、GATT サービス、L2CAP チャネルを扱うためのクリーンなコルーチンを提供します。

すべてのリモート操作（接続、切断、クライアントの読み書き、サーバーの indicate、l2cap の recv/send、ペアリング）は await 可能で、タイムアウトをサポートします。

サポートされるロール:

Broadcaster（アドバタイザ） --- 一般的なフィールド向けにアドバタイジングおよびスキャンレスポンスのペイロードを生成し、ペイロードをアドバタイジングとスキャンレスポンスに自動的に分割し、無期限または一定時間アドバタイズします。
Peripheral --- セントラルからの接続を待機し、MTU の交換を待機します。
Observer（スキャナ） --- パッシブスキャンとアクティブスキャン、同一デバイスのアドバタイジングとスキャンレスポンスのペイロードの結合、アドバタイジングペイロードからの一般的なフィールドの解析を行います。
Central --- ペリフェラルに接続し、MTU の交換を開始します。
GATT クライアント --- サービス／キャラクタリスティック／ディスクリプタの探索（必要に応じて UUID 指定）、キャラクタリスティックとディスクリプタの read／write／write-with-response、（CCCD を介した）通知および indication への登録、通知および indication の待機を行います。
GATT サーバー --- サービス／キャラクタリスティック／ディスクリプタの登録、キャラクタリスティックとディスクリプタへの write の待機、read 要求のインターセプト、通知および indication の送信（および応答の待機）を行います。
L2CAP --- L2CAP コネクション指向チャネルの accept と connect、チャネルのフロー制御の管理を行います。
Security --- JSON ベースの鍵／シークレット管理、ペアリングの開始、暗号化／認証状態の照会を行います。

例#

近くの BLE デバイスをスキャンし、見つけるたびにそれぞれを表示します:

import aioble
import asyncio

async def find_devices():
    async with aioble.scan(duration_ms=5000, active=True) as scanner:
        async for result in scanner:
            print(result.device.addr_hex(), result.rssi, result.name())

asyncio.run(find_devices())

central として Heart Rate サービスをアドバタイズしているペリフェラルに接続し、その測定値の通知に登録します:

import aioble
import asyncio
import bluetooth

_HR_SERVICE = bluetooth.UUID(0x180D)
_HR_MEASUREMENT = bluetooth.UUID(0x2A37)

async def connect_and_read():
    device = None
    async with aioble.scan(duration_ms=5000, active=True) as scanner:
        async for result in scanner:
            if _HR_SERVICE in result.services():
                device = result.device
                break
    if device is None:
        return

    async with await device.connect() as conn:
        service = await conn.service(_HR_SERVICE)
        char = await service.characteristic(_HR_MEASUREMENT)
        await char.subscribe(notify=True)
        while True:
            data = await char.notified()
            print("notify:", data)

asyncio.run(connect_and_read())

peripheral として動作します。GATT サービスを登録し、それをアドバタイズし、接続してきた相手に通知をプッシュします:

import aioble
import asyncio
import bluetooth
import struct

_ENV_SERVICE = bluetooth.UUID(0x181A)
_TEMP_CHAR = bluetooth.UUID(0x2A6E)

def encode_temperature(deg_c):
    # Bluetooth Temperature (0x2A6E) is sint16 little-endian, 0.01 degC units.
    return struct.pack("<h", round(deg_c * 100))

service = aioble.Service(_ENV_SERVICE)
temp_char = aioble.Characteristic(service, _TEMP_CHAR, read=True, notify=True)
aioble.register_services(service)

async def peripheral_task():
    while True:
        connection = await aioble.advertise(
            interval_us=250000,
            name="openmv-sensor",
            services=[_ENV_SERVICE],
            appearance=0x0300,
        )
        print("connected:", connection.device.addr_hex())
        async with connection:
            while connection.is_connected():
                temp_char.write(encode_temperature(23.68), send_update=True)
                await asyncio.sleep(1)

asyncio.run(peripheral_task())

モジュールレベルの関数#

aioble.config(*args, **kwargs) → Any#

bluetooth.BLE.config() に転送し、まず BLE 無線がアクティブであることを保証します。

args: 照会する単一のパラメータ名（任意）。
kwargs: 設定値を指定するためのキーワード引数。

aioble.stop() → None#: 基盤となる BLE 無線を非アクティブ化し、登録されているサブモジュールのシャットダウンハンドラを実行します。これを呼び出した後は、スキャナ、アドバタイザ、接続、L2CAP チャネルがすべて破棄されます。

aioble.scan(duration_ms: int, interval_us: int | None = None, window_us: int | None = None, active: bool = False) → scan#

発見された一意なデバイスごと（または既知のデバイスから新たなアドバタイジングデータを受信するたび）に ScanResult インスタンスを生成する scan の非同期コンテキストマネージャ／非同期イテレータを返します。

duration_ms: スキャンする時間（ミリ秒）。0 を渡すと、コンテキストマネージャが終了するまで無期限にスキャンします。
interval_us: スキャン間隔（マイクロ秒）。デフォルトは 1,280,000 です。
window_us: スキャンウィンドウ（マイクロ秒、interval_us 以下である必要があります）。デフォルトは 11,250 です。
active: True の場合、アクティブスキャン（スキャンレスポンスデータを要求）を実行します。デフォルトは False です。

aioble.advertise(interval_us: int, adv_data: bytes | None = None, resp_data: bytes | None = None, connectable: bool = True, limited_disc: bool = False, br_edr: bool = False, name: str | None = None, services: list | None = None, appearance: int = 0, manufacturer: tuple | None = None, timeout_ms: int | None = None) → DeviceConnection#

アドバタイズを開始し、セントラルからの着信接続を待機する非同期コルーチンです。接続されたセントラルを表す DeviceConnection を返すか、タイムアウト時に asyncio.TimeoutError を送出します。

interval_us: アドバタイジング間隔（マイクロ秒）。
adv_data: 生のアドバタイジングペイロード。指定しない場合、adv_data は残りのキーワード引数から構築されます。
resp_data: 生のスキャンレスポンスペイロード。必要に応じて adv_data からあふれた分が自動的に格納されます。
connectable: True の場合、これは接続可能なアドバタイズメントです。
limited_disc: general の代わりに limited-discoverable フラグを使用します。
br_edr: BR/EDR サポートフラグを設定します。
name: 埋め込む完全なローカル名（任意）。
services: アドバタイズする bluetooth.UUID のイテラブル。
appearance: 16 ビットの appearance 値（Bluetooth assigned numbers を参照）。
manufacturer: manufacturer specific data としてアドバタイズする (company_id, data_bytes) のタプル。
timeout_ms: 接続がないまま指定したミリ秒数が経過したらアドバタイズを停止します。None は接続されるまでアドバタイズし続けることを意味します。

aioble.register_services(*services: Service) → None#

1 つ以上の Service オブジェクト（およびそのキャラクタリスティックとディスクリプタ）を GATT サーバーに登録します。advertise を開始する前に一度呼び出す必要があります。以降の呼び出しは前回の登録を置き換えます。

services: 1 つ以上の Service インスタンス。

モジュールレベルの定数#

aioble.ADDR_PUBLIC#: パブリック BLE デバイスアドレスタイプ（0）。

aioble.ADDR_RANDOM#: ランダム BLE デバイスアドレスタイプ（1）。

例外#

exception aioble.GattError#: リモートの GATT 操作（read／write／indicate）が 0 以外のステータスで完了したときに送出されます。ステータスコードは _status 属性で参照できます。

exception aioble.DeviceDisconnectedError#: 待機中に基盤となる接続が切断されたとき、非同期操作（例: read、write、notified）の内部で送出されます。

exception aioble.L2CAPDisconnectedError#: 切断されたチャネルに対して（またはそれによって中断されて）L2CAP チャネルの send／recv／flush 操作が試みられたときに送出されます。

exception aioble.L2CAPConnectionError#: DeviceConnection.l2cap_connect でチャネルの確立に失敗したときに送出されます。Bluetooth ステータスコードが最初の引数です。

クラス#

class aioble.Device(addr_type: int, addr: bytes | str)#

リモート BLE デバイスをアドレスで表します。2 つの Device インスタンスは、addr_type と addr の両方が一致する場合に等しいと比較されます。接続を開始するためのハンドルとして使用されます。

addr_type: ADDR_PUBLIC または ADDR_RANDOM のいずれか。
addr: bytes としての 6 バイトアドレス、またはコロン区切りの 16 進文字列（例: "aa:bb:cc:dd:ee:ff"）。

addr_type#: デバイスを構築したときのアドレスタイプ。

addr#: 生の 6 バイトデバイスアドレス。

addr_hex() → str#: アドレスをコロン区切りの 16 進文字列として整形して返します。

connect(timeout_ms: int = 10000, scan_duration_ms: int | None = None, min_conn_interval_us: int | None = None, max_conn_interval_us: int | None = None) → Awaitable[DeviceConnection]#

非同期。このデバイスへの GAP 接続を開始し、結果として得られる DeviceConnection を返します。進行中のスキャンがあればキャンセルします。

timeout_ms: 接続が完了するまで待機する時間。
scan_duration_ms: 接続前の初期スキャン時間（コントローラ依存）。
min_conn_interval_us / max_conn_interval_us: 接続間隔の境界値（マイクロ秒、任意）。

class aioble.DeviceConnection#

Device へのアクティブな GAP 接続。Device.connect() または advertise によって返されます。終了時に自動的に切断する async with コンテキストマネージャとして使用できます。

直接構築しないでください。

device#: 基盤となる Device。

encrypted#: リンクが暗号化されると（例: ペアリング後）True になります。

authenticated#: リンクが認証された（MITM 保護されたペアリング）場合に True。

bonded#: ペアリングによってボンディング鍵が生成された場合に True。

key_size#: ネゴシエートされた暗号鍵のサイズ（バイト単位）、または暗号化されていない場合は False。

mtu#: exchange_mtu 後にネゴシエートされた ATT MTU、または設定されるまでは None。

is_connected() → bool#: 接続がまだアクティブかどうかを返します。

disconnect(timeout_ms: int = 2000) → Awaitable[None]#

非同期。切断し、切断 IRQ を待機します。

timeout_ms: 切断を待機する最大時間。

disconnected(timeout_ms: int | None = None, disconnect: bool = False) → Awaitable[None]#

非同期。いずれかの側によって接続が終了されるのを待機します。disconnect が True の場合、まず能動的に切断します。

timeout_ms: 待機する最大時間。None は永久に待機することを意味します。
disconnect: True の場合、切断を開始します。

timeout(timeout_ms: int | None) → DeviceTimeout#

タイムアウトが経過した場合（asyncio.TimeoutError を送出）、またはデバイスが切断された場合（DeviceDisconnectedError を送出）に本体をキャンセルするコンテキストマネージャを返します。

timeout_ms: タイムアウト（ミリ秒）、またはタイムアウトなしの場合は None。

exchange_mtu(mtu: int | None = None, timeout_ms: int = 1000) → Awaitable[int]#

非同期。ATT MTU の交換を開始し、ネゴシエートされた MTU を返します。

mtu: 交換前に基盤となる BLE インターフェースに設定する優先 MTU（任意）。
timeout_ms: 交換のタイムアウト。

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]#: 非同期。uuid に一致する単一のリモートサービスを探索します。見つからない場合は None を返します。

services(uuid: bluetooth.UUID | None = None, timeout_ms: int = 2000) → ClientDiscover#

リモートの ClientService オブジェクトの非同期イテレータを返します。async for とともに使用し、ループを最後まで実行してください。

uuid: UUID フィルタ（任意）。None はすべてのサービスを返します。
timeout_ms: 探索ごとのタイムアウト。

pair(bond: bool = True, le_secure: bool = True, mitm: bool = False, io: int = 3, timeout_ms: int = 20000) → Awaitable[None]#

非同期。この接続でペアリングを開始します。完了時に encrypted／authenticated／bonded／key_size 属性を更新します。

bond: ペアリング鍵を永続化します。
le_secure: LE Secure Connections を使用します。
mitm: man-in-the-middle 保護を要求します。
io: IO 機能の定数（例: 入出力なしの場合は 3）。
timeout_ms: ペアリングのタイムアウト。

l2cap_accept(psm: int, mtu: int, timeout_ms: int | None = None) → Awaitable[L2CAPChannel]#

非同期。指定した PSM で待ち受け、リモートが開いたら L2CAPChannel を返します。

psm: 待ち受ける Protocol/Service Multiplexer。
mtu: 最大受信サイズ（バイト単位）。
timeout_ms: リモートが接続するまで待機する最大時間。

l2cap_connect(psm: int, mtu: int, timeout_ms: int = 1000) → Awaitable[L2CAPChannel]#

非同期。指定した PSM でリモートへの L2CAP チャネルを開きます。

psm: 接続先の Protocol/Service Multiplexer。
mtu: 最大受信サイズ（バイト単位）。
timeout_ms: 接続のタイムアウト。

class aioble.ScanResult#

scan 中に発見された単一のデバイス。新しいアドバタイジングデータが到着すると同じインスタンスが再度生成されます。

直接構築しないでください。

device#: 基盤となる Device。

rssi#: 最後に報告された RSSI（dBm）。

adv_data#: 生のアドバタイジングペイロード（bytes または None）。

resp_data#: 生のスキャンレスポンスペイロード（bytes または None）。アクティブスキャンが有効な場合に得られます。

connectable#: 直近のアドバタイズメントが接続可能だった場合に True。

name() → str | None#: ペイロードから完全な（または短縮された）アドバタイズされたローカル名をデコードします。存在しない場合は None。

services() → Iterator[bluetooth.UUID]#: 16／32／128 ビットのサービスリストフィールドでアドバタイズされた各 bluetooth.UUID を生成するジェネレータ。

manufacturer(filter: int | None = None) → Iterator[tuple[int, bytes]]#

manufacturer specific のアドバタイジングフィールドから (company_id, data) タプルを生成するジェネレータ。

filter: 指定した場合、company ID が一致するエントリのみを生成します。

class aioble.Service(uuid: bluetooth.UUID)#

ローカル GATT サービス。1 つ以上の Characteristic インスタンスでサービスを構築し、それを register_services に渡します。

uuid: サービス UUID。

uuid#: サービス UUID。

characteristics#: このサービスにバインドされた Characteristic オブジェクトのリスト。

class aioble.Characteristic(service: Service, uuid: bluetooth.UUID, read: bool = False, write: bool = False, write_no_response: bool = False, notify: bool = False, indicate: bool = False, initial: bytes | None = None, capture: bool = False)#

ローカル GATT キャラクタリスティック。これを構築すると自動的に service に追加されます。

service: 所有する Service。
uuid: キャラクタリスティック UUID。
read, write, write_no_response, notify, indicate: サポートする GATT 操作を選択するブール値。
initial: 初期値（bytes、任意）。
capture: True の場合、書き込まれた値がキュー（最大 10 個）に保存され、立て続けの書き込みが失われないようになります。各 written の呼び出しはその後 (connection, data) タプルを返します。

uuid#: キャラクタリスティック UUID。

flags#: コンストラクタから構築された GATT プロパティフラグのビットマスク。

descriptors#: このキャラクタリスティックにバインドされた Descriptor オブジェクトのリスト。

read() → bytes#: ローカル GATT データベースから現在の値を読み取ります。

write(data: bytes, send_update: bool = False) → None#

ローカル GATT データベース内の値を更新します。

data: 新しい値のバイト列。
send_update: True の場合、登録されているすべての接続に notify／indicate も行います。

notify(connection: DeviceConnection, data: bytes | None = None) → None#

connection に GATT Notify を送信します。

connection: 対象のクライアント接続。
data: 送信するペイロード。None の場合、現在のローカル値が送信されます。

indicate(connection: DeviceConnection, data: bytes | None = None, timeout_ms: int = 1000) → Awaitable[None]#

非同期。connection に GATT Indicate を送信し、クライアントの確認を待機します。0 以外のステータスの場合 GattError を送出します。

connection: 対象のクライアント接続。
data: indicate するペイロード、またはローカル値を送信する場合は None。
timeout_ms: 確認を待機する最大時間。

written(timeout_ms: int | None = None) → Awaitable[DeviceConnection | tuple[DeviceConnection, bytes]]#

非同期。リモートからの write を待機します。書き込んできた DeviceConnection を返します。キャラクタリスティックが capture=True で作成されていた場合は (connection, data) を返します。

timeout_ms: 待機する最大時間。None は永久に待機します。

on_read(connection: DeviceConnection) → int#: リモートの read を受信したときに同期的に呼び出されるオーバーライドフック。read を許可する場合は 0 を、拒否する場合は 0 以外の ATT エラーコードを返します。デフォルトの実装は 0 を返します。

class aioble.BufferedCharacteristic(service: Service, uuid: bluetooth.UUID, max_len: int = 20, append: bool = False, **kwargs)#

バックエンドの GATT バッファを構成できる Characteristic。デフォルトの属性サイズより大きい値を受信したい場合や、立て続けの書き込みをキューに入れたい場合に便利です。

max_len: バッファサイズ（バイト単位）。
append: True の場合、連続した書き込みは上書きではなくバッファに追記されます。

その他の引数は Characteristic に転送されます。

class aioble.Descriptor(characteristic: Characteristic, uuid: bluetooth.UUID, read: bool = False, write: bool = False, initial: bytes | None = None)#

ローカル GATT ディスクリプタ。これを構築すると自動的に characteristic に追加されます。Characteristic から read、write、written を継承します。

characteristic: 所有する Characteristic。
uuid: ディスクリプタ UUID。
read, write: サポートする GATT 操作を選択するブール値。
initial: 初期値（bytes、任意）。

class aioble.ClientService#

ピアで発見されたリモート GATT サービス。DeviceConnection.service() によって返されるか、DeviceConnection.services() から反復処理されます。

直接構築しないでください。

connection#: 所有する DeviceConnection。

uuid#: リモートサービス UUID。

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]#: 非同期。UUID で単一のキャラクタリスティックを探索します。見つからない場合は None。

characteristics(uuid: bluetooth.UUID | None = None, timeout_ms: int = 2000) → ClientDiscover#

ClientCharacteristic オブジェクトの非同期イテレータを返します。async for とともに使用し、ループを最後まで実行してください。

uuid: UUID フィルタ（任意）。
timeout_ms: 探索ごとのタイムアウト。

class aioble.ClientCharacteristic#

ピアで発見されたリモート GATT キャラクタリスティック。ClientService.characteristic() によって返されるか、ClientService.characteristics() から反復処理されます。

直接構築しないでください。

service#: 所有する ClientService。

uuid#: キャラクタリスティック UUID。

properties#: ピアから報告されたサポート対象の GATT 操作のビットマスク。

read(timeout_ms: int = 1000) → Awaitable[bytes]#

非同期。GATT Read を発行して値を返します。0 以外のステータスの場合 GattError を送出します。

timeout_ms: read のタイムアウト。

write(data: bytes, response: bool | None = None, timeout_ms: int = 1000) → Awaitable[None]#

非同期。GATT Write を発行します。

data: 書き込む値。
response: True の場合、write-response を要求します（失敗時に GattError を送出）。False は write-without-response です。None（デフォルト）はピアがアドバタイズする内容に基づいて自動選択します。
timeout_ms: write のタイムアウト（response が True の場合のみ関係します）。

notified(timeout_ms: int | None = None) → Awaitable[bytes]#

非同期。このキャラクタリスティックでの次の通知を待機し、そのペイロードを返します。通知がすでにキューに入っている場合は即座に返します。

timeout_ms: 待機する最大時間。None は永久に待機します。

indicated(timeout_ms: int | None = None) → Awaitable[bytes]#

非同期。このキャラクタリスティックでの次の indication を待機し、そのペイロードを返します。

timeout_ms: 待機する最大時間。

subscribe(notify: bool = True, indicate: bool = False) → Awaitable[None]#

非同期。Client Characteristic Configuration Descriptor（CCCD）に書き込んで、通知および／または indication の登録（または登録解除）を行います。

notify: 通知を有効にします。
indicate: indication を有効にします。

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]#: 非同期。UUID で単一のディスクリプタを探索します。見つからない場合は None。

descriptors(timeout_ms: int = 2000) → ClientDiscover#: ClientDescriptor オブジェクトの非同期イテレータを返します。async for とともに使用し、ループを最後まで実行してください。

class aioble.ClientDescriptor#

ピアで発見されたリモート GATT ディスクリプタ。ClientCharacteristic から read と write を継承します。

直接構築しないでください。

characteristic#: 所有する ClientCharacteristic。

uuid#: ディスクリプタ UUID。

class aioble.L2CAPChannel#

アクティブな L2CAP コネクション指向チャネル。DeviceConnection.l2cap_accept() または DeviceConnection.l2cap_connect() によって返されます。終了時に自動的に切断する async with コンテキストマネージャとして使用できます。

直接構築しないでください。

our_mtu#: ピアが 1 つの SDU で当方に送信できる最大サイズ（バイト単位）。

peer_mtu#: 当方がピアに 1 つの SDU で送信できる最大サイズ（バイト単位）。

available() → bool#: バッファされた受信データが準備できている場合（つまり recvinto がブロックしない場合）に同期的に True を返します。

recvinto(buf: bytearray, timeout_ms: int | None = None) → Awaitable[int]#

非同期。buf に受信し、読み取ったバイト数を返します。チャネルが空の場合は新しいデータを待機します。

buf: 埋めるための事前割り当て済みバッファ。
timeout_ms: 待機する最大時間。None は永久に待機します。

send(buf: bytes, timeout_ms: int | None = None, chunk_size: int | None = None) → Awaitable[None]#

非同期。buf をチャネルで送信し、大きなペイロードを MTU サイズのチャンクに分割します。必要に応じてフロー制御のクレジットを待機します。

buf: 送信する bytes-like オブジェクト。
timeout_ms: チャンクごとに待機する最大時間。
chunk_size: 呼び出しごとのチャンクサイズのオーバーライド（任意）。min(our_mtu * 2, peer_mtu) に制限されます。

flush(timeout_ms: int | None = None) → Awaitable[None]#

非同期。停止している send がコントローラによって排出されるまで待機します。

timeout_ms: 待機する最大時間。

disconnect(timeout_ms: int = 1000) → Awaitable[None]#

非同期。チャネルを切断し、切断 IRQ を待機します。

timeout_ms: 待機する最大時間。

disconnected(timeout_ms: int = 1000) → Awaitable[None]#

非同期。いずれかの側によってチャネルが切断されるまで待機します。

timeout_ms: 待機する最大時間。

aioble --- 非同期 BLE#

例#

モジュールレベルの関数#

モジュールレベルの定数#

例外#

クラス#

`aioble` --- 非同期 BLE#