`aioble` --- 非同步 BLE#

aioble 是對 bluetooth 模組所做的高階、對 asyncio 友善的封裝。它為掃描、連線、廣播、GATT 服務以及 L2CAP 通道提供了清晰的協程介面。

所有遠端操作（連線、斷線、用戶端讀取/寫入、伺服端 indicate、l2cap recv/send、配對）皆可 await 並支援逾時設定。

支援的角色：

Broadcaster（廣播者） --- 為常見欄位產生廣告與掃描回應酬載、自動將酬載拆分到廣告與掃描回應之間、無限期或固定時長進行廣播。
Peripheral（周邊端） --- 等待來自 central 的連線、等待 MTU 交換。
Observer（掃描者） --- 被動與主動掃描、為同一裝置合併廣告與掃描回應酬載、從廣告酬載解析常見欄位。
Central（中央端） --- 連線至 peripheral、發起 MTU 交換。
GATT 用戶端 --- 探索服務／特徵／描述子（可選擇以 UUID 篩選）；對特徵與描述子進行讀取／寫入／寫入並回應；訂閱通知與指示（透過 CCCD）；等待通知與指示。
GATT 伺服端 --- 註冊服務／特徵／描述子；等待對特徵與描述子的寫入；攔截讀取請求；傳送通知與指示（並等待回應）。
L2CAP --- 接受並連接 L2CAP 連線導向通道、管理通道的流量控制。
安全性 --- 以 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 服務的 peripheral，並訂閱其測量值通知：：

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#

傳回一個 scan 非同步情境管理器／非同步迭代器，會為每個探索到的唯一裝置（或來自已知裝置的每筆新廣告資料）產生 ScanResult 實例。

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#

非同步協程，開始廣播並等待傳入的 central 連線。傳回代表已連線 central 的 DeviceConnection，或在逾時時引發 asyncio.TimeoutError。

interval_us: 廣播間隔，以微秒為單位。
adv_data: 原始廣告酬載。若未設定，adv_data 會從其餘的關鍵字引數建構而成。
resp_data: 原始掃描回應酬載。若有需要，會自動填入由 adv_data 溢位的內容。
connectable: 若為 True，則此為可連線的廣告。
limited_disc: 使用受限可探索（limited-discoverable）旗標，而非一般旗標。
br_edr: 設定 BR/EDR 支援旗標。
name: 可選的完整本地名稱，用以嵌入。
services: 要廣播的 bluetooth.UUID 可迭代物件。
appearance: 16 位元外觀值（請參閱 Bluetooth 指定編號）。
manufacturer: 要作為製造商特定資料廣播的 (company_id, data_bytes) 元組。
timeout_ms: 在這麼多毫秒後若仍無連線，便停止廣播。None 表示持續廣播直到連線為止。

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

向 GATT 伺服端註冊一個或多個 Service 物件（及其特徵與描述子）。必須在開始 advertise 之前呼叫一次。後續的呼叫會取代先前的註冊。

services: 一個或多個 Service 實例。

模組層級常數#

aioble.ADDR_PUBLIC#: 公開 BLE 裝置位址類型（0）。

aioble.ADDR_RANDOM#: 隨機 BLE 裝置位址類型（1）。

例外#

exception aioble.GattError#: 當遠端 GATT 操作（讀取／寫入／indicate）以非零狀態完成時引發。狀態碼可由 _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 裝置。若 addr_type 與 addr 皆相符，則兩個 Device 實例比較為相等。用作發起連線的控制代碼。

addr_type: 為 ADDR_PUBLIC 或 ADDR_RANDOM 其中之一。
addr: 以 bytes 表示的六位元組位址，或以冒號分隔的十六進位字串（例如 "aa:bb:cc:dd:ee:ff"）。

addr_type#: 建構此裝置時所用的位址類型。

addr#: 原始的六位元組裝置位址。

addr_hex() → str#: 傳回以冒號分隔的十六進位字串格式的位址。

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]]#

從製造商特定廣告欄位產生 (company_id, data) 元組的產生器。

filter: 若有提供，則僅產生公司 ID 相符的項目。

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

本地 GATT 服務。以一個或多個 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 並等待用戶端確認。狀態為非零時引發 GattError。

connection: 目標用戶端連線。
data: 要 indicate 的酬載，或 None 以傳送本地值。
timeout_ms: 等待確認的最長時間。

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

非同步。等待一筆遠端寫入。傳回進行寫入的 DeviceConnection，若該特徵建立時帶有 capture=True 則傳回 (connection, data)。

timeout_ms: 等待的最長時間。None 表示永久等待。

on_read(connection: DeviceConnection) → int#: 當收到遠端讀取時同步呼叫的可覆寫掛鉤。傳回 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 並傳回該值。狀態為非零時引發 GattError。

timeout_ms: 讀取逾時時間。

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

非同步。發出一個 GATT Write。

data: 要寫入的值。
response: True 表示要求寫入回應（失敗時引發 GattError）。False 表示不需回應的寫入。None（預設）會依對端所廣播的內容自動選擇。
timeout_ms: 寫入逾時時間（僅在 response 為 True 時才相關）。

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

非同步。等待此特徵上的下一筆通知並傳回其酬載。若已有通知排入佇列則立即傳回。

timeout_ms: 等待的最長時間。None 表示永久等待。

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

非同步。等待此特徵上的下一筆指示並傳回其酬載。

timeout_ms: 等待的最長時間。

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

非同步。寫入用戶端特徵組態描述子（CCCD），以訂閱（或取消訂閱）通知與／或指示。

notify: 啟用通知。
indicate: 啟用指示。

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#: 對端在單一 SDU 中可傳送給我們的最大大小，以位元組為單位。

peer_mtu#: 我們在單一 SDU 中可傳送給對端的最大大小，以位元組為單位。

available() → bool#: 若已緩衝的接收資料就緒（即 recvinto 不會阻塞），則同步傳回 True。

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

非同步。接收資料至 buf，傳回讀取的位元組數。若通道為空則 await 新資料。

buf: 用以填入資料的預先配置緩衝區。
timeout_ms: 等待的最長時間。None 表示永久等待。

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

非同步。在通道上傳送 buf，將較大的酬載分段為 MTU 大小的區塊。會視需要 await 流量控制額度。

buf: 要傳送的類位元組物件。
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#