`aioble` --- 异步 BLE#

aioble 是围绕 bluetooth 模块构建的高级、对 asyncio 友好的封装。它为扫描、连接、广播、GATT 服务和 L2CAP 通道提供了简洁的协程。

所有远程操作（connect、disconnect、客户端读/写、服务器 indicate、l2cap recv/send、pair）都是可等待的，并支持超时。

支持的角色：

广播者（advertiser） --- 为常见字段生成广播和扫描响应负载，自动将负载拆分到广播和扫描响应中，可无限期或按固定时长进行广播。
外围设备（Peripheral） --- 等待来自中心设备的连接，等待 MTU 交换。
观察者（scanner） --- 被动和主动扫描，将同一设备的广播和扫描响应负载组合起来，从广播负载中解析常见字段。
中心设备（Central） --- 连接到外围设备，发起 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())

作为 中心设备 连接到广播心率服务的外围设备，并订阅其测量值通知：：

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

作为 外围设备：注册一个 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#

异步协程，开始广播并等待传入的中心设备连接。返回表示已连接中心设备的 DeviceConnection，或在超时时引发 asyncio.TimeoutError。

interval_us: 广播间隔，以微秒为单位。
adv_data: 原始广播负载。如果未设置，则 adv_data 由其余关键字参数构建。
resp_data: 原始扫描响应负载。如有需要，会自动填充从 adv_data 溢出的部分。
connectable: 如果为 True，则这是一个可连接的广播。
limited_disc: 使用受限可发现标志而非通用标志。
br_edr: 设置 BR/EDR 支持标志。
name: 可选的待嵌入的完整本地名称。
services: 要广播的 bluetooth.UUID 可迭代对象。
appearance: 16 位外观值（参见蓝牙分配编号）。
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 引发。蓝牙状态码是第一个参数。

类#

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: 可选的首选 MTU，在交换之前设置到底层 BLE 接口上。
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 安全连接。
mitm: 要求中间人保护。
io: IO 能力常量（例如 3 表示无输入/输出）。
timeout_ms: 配对的超时时间。

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

异步。在给定的 PSM 上监听，一旦远端打开通道便返回一个 L2CAPChannel。

psm: 要监听的协议/服务多路复用器（PSM）。
mtu: 最大接收大小，以字节为单位。
timeout_ms: 等待远端连接的最长时间。

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

异步。在给定的 PSM 上向远端打开一个 L2CAP 通道。

psm: 要连接的协议/服务多路复用器（PSM）。
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: 如果给定，则只生成 company 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(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: 要指示的负载，或为 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 中，返回读取的字节数。如果通道为空，则等待新数据。

buf: 用于填充的预分配缓冲区。
timeout_ms: 等待的最长时间。None 表示永远等待。

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

异步。在通道上发送 buf，将较大的负载分片为 MTU 大小的块。按需等待流控制信用额度。

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#