aioble --- Async BLE

aioble là một wrapper asyncio cấp cao thân thiện bọc quanh module bluetooth. Nó cung cấp các coroutine sạch cho việc quét, kết nối, quảng bá, dịch vụ GATT và kênh L2CAP.

Tất cả các thao tác từ xa (kết nối, ngắt kết nối, đọc/ghi client, server indicate, l2cap recv/send, ghép đôi) đều có thể await và hỗ trợ timeout.

Các vai trò được hỗ trợ:

  • Broadcaster (advertiser) --- tạo payload quảng bá và scan-response cho các trường phổ biến, tự động chia payload giữa advertising và scan response, quảng bá vô thời hạn hoặc trong thời gian cố định.

  • Peripheral --- chờ kết nối từ central, chờ trao đổi MTU.

  • Observer (scanner) --- quét thụ động và tích cực, kết hợp payload advertising và scan-response cho cùng một thiết bị, phân tích các trường phổ biến từ payload quảng bá.

  • Central --- kết nối đến peripheral, khởi tạo trao đổi MTU.

  • GATT Client --- khám phá dịch vụ / đặc trưng / bộ mô tả (tùy chọn theo UUID); đọc / ghi / ghi-có-phản-hồi trên các đặc trưng và bộ mô tả; đăng ký nhận thông báo và chỉ thị (qua CCCD); chờ thông báo và chỉ thị.

  • GATT Server --- đăng ký dịch vụ / đặc trưng / bộ mô tả; chờ ghi vào các đặc trưng và bộ mô tả; chặn các yêu cầu đọc; gửi thông báo và chỉ thị (và chờ phản hồi).

  • L2CAP --- chấp nhận và kết nối các kênh L2CAP hướng kết nối, quản lý kiểm soát luồng kênh.

  • Security --- quản lý khóa/bí mật dựa trên JSON, khởi tạo ghép đôi, truy vấn trạng thái mã hóa / xác thực.

Ví dụ

Quét các thiết bị BLE lân cận và in ra từng thiết bị khi phát hiện:

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

Kết nối đến peripheral đang quảng bá dịch vụ Heart Rate với vai trò central và đăng ký nhận thông báo đo lường:

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

Hoạt động như một peripheral: đăng ký dịch vụ GATT, quảng bá nó và đẩy thông báo đến ai kết nối:

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

Các hàm cấp module

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

Chuyển tiếp đến bluetooth.BLE.config(), đảm bảo radio BLE đang hoạt động trước.

args

Tên tham số đơn tùy chọn để truy vấn.

kwargs

Các đối số từ khóa để đặt giá trị cấu hình.

aioble.stop() None

Vô hiệu hóa radio BLE bên dưới và chạy mọi trình xử lý tắt máy sub-module đã đăng ký. Sau khi gọi hàm này, tất cả scanner, advertiser, kết nối và kênh L2CAP đều bị đóng lại.

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

Trả về một scan async context-manager / async-iterator tạo ra các instance ScanResult cho mỗi thiết bị duy nhất được phát hiện (hoặc cho mỗi dữ liệu quảng bá mới từ một thiết bị đã biết).

duration_ms

Thời gian quét, tính bằng mili giây. Truyền 0 để quét vô thời hạn cho đến khi context manager thoát.

interval_us

Khoảng thời gian quét tính bằng micro giây. Mặc định là 1.280.000.

window_us

Cửa sổ quét tính bằng micro giây (phải nhỏ hơn hoặc bằng interval_us). Mặc định là 11.250.

active

Nếu True, thực hiện quét tích cực (yêu cầu dữ liệu scan response). Mặc định là 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

Coroutine async bắt đầu quảng bá và chờ kết nối đến từ central. Trả về DeviceConnection đại diện cho central đã kết nối, hoặc ném asyncio.TimeoutError khi hết timeout.

interval_us

Khoảng thời gian quảng bá, tính bằng micro giây.

adv_data

Payload quảng bá thô. Nếu không đặt, adv_data được xây dựng từ các đối số từ khóa còn lại.

resp_data

Payload scan response thô. Tự động điền phần tràn từ adv_data nếu cần.

connectable

Nếu True, đây là quảng bá có thể kết nối.

limited_disc

Sử dụng cờ limited-discoverable thay vì general.

br_edr

Đặt cờ hỗ trợ BR/EDR.

name

Tên cục bộ đầy đủ tùy chọn để nhúng vào.

services

Iterable của bluetooth.UUID để quảng bá.

appearance

Giá trị appearance 16-bit (xem Bluetooth assigned numbers).

manufacturer

Tuple (company_id, data_bytes) để quảng bá dưới dạng dữ liệu đặc thù nhà sản xuất.

timeout_ms

Dừng quảng bá sau số mili giây này mà không có kết nối. None nghĩa là quảng bá cho đến khi có kết nối.

aioble.register_services(*services: Service) None

Đăng ký một hoặc nhiều đối tượng Service (cùng với các đặc trưng và bộ mô tả của chúng) với GATT server. Phải được gọi một lần trước khi khởi động advertise. Các lần gọi tiếp theo sẽ thay thế đăng ký trước đó.

services

Một hoặc nhiều instance Service.

Hằng số cấp module

aioble.ADDR_PUBLIC

Loại địa chỉ thiết bị BLE công khai (0).

aioble.ADDR_RANDOM

Loại địa chỉ thiết bị BLE ngẫu nhiên (1).

Ngoại lệ

exception aioble.GattError

Được ném khi một thao tác GATT từ xa (đọc / ghi / indicate) hoàn thành với trạng thái khác không. Mã trạng thái có sẵn trên thuộc tính _status.

exception aioble.DeviceDisconnectedError

Được ném bên trong một thao tác async (ví dụ: đọc, ghi, được thông báo) khi kết nối bên dưới bị mất trong khi đang chờ.

exception aioble.L2CAPDisconnectedError

Được ném khi thao tác send/recv/flush trên kênh L2CAP được thực hiện trên (hoặc bị gián đoạn bởi) một kênh đã ngắt kết nối.

exception aioble.L2CAPConnectionError

Được ném bởi DeviceConnection.l2cap_connect khi thiết lập kênh thất bại. Mã trạng thái Bluetooth là đối số đầu tiên.

Các lớp

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

Đại diện cho một thiết bị BLE từ xa theo địa chỉ. Hai instance Device bằng nhau nếu cả addr_typeaddr đều khớp. Được sử dụng làm handle để khởi tạo kết nối.

addr_type

ADDR_PUBLIC hoặc ADDR_RANDOM.

addr

Địa chỉ sáu byte dưới dạng bytes, hoặc chuỗi hex phân cách bằng dấu hai chấm (ví dụ: "aa:bb:cc:dd:ee:ff").

addr_type

Loại địa chỉ mà thiết bị được tạo với.

addr

Địa chỉ thiết bị sáu byte thô.

addr_hex() str

Trả về địa chỉ được định dạng dưới dạng chuỗi hex phân cách bằng dấu hai chấm.

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]

Async. Khởi tạo kết nối GAP đến thiết bị này và trả về DeviceConnection kết quả. Hủy mọi lần quét đang diễn ra.

timeout_ms

Thời gian chờ kết nối hoàn thành.

scan_duration_ms

Thời gian quét ban đầu trước khi kết nối (tùy thuộc controller).

min_conn_interval_us / max_conn_interval_us

Giới hạn khoảng thời gian kết nối tùy chọn, tính bằng micro giây.

class aioble.DeviceConnection

Một kết nối GAP đang hoạt động đến Device. Được trả về bởi Device.connect() hoặc advertise. Hỗ trợ sử dụng như context manager async with tự động ngắt kết nối khi thoát.

Không tạo trực tiếp.

device

Device bên dưới.

encrypted

True khi liên kết được mã hóa (ví dụ: sau khi ghép đôi).

authenticated

True nếu liên kết đã được xác thực (ghép đôi có bảo vệ MITM).

bonded

True nếu quá trình ghép đôi tạo ra khóa bonding.

key_size

Kích thước khóa mã hóa đã thỏa thuận tính bằng byte, hoặc False nếu không mã hóa.

mtu

ATT MTU đã thỏa thuận sau exchange_mtu, hoặc None cho đến khi được đặt.

is_connected() bool

Trả về trạng thái kết nối còn hoạt động hay không.

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

Async. Ngắt kết nối và chờ IRQ ngắt kết nối.

timeout_ms

Thời gian tối đa để chờ ngắt kết nối.

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

Async. Chờ kết nối bị chấm dứt bởi một trong hai bên. Nếu disconnectTrue, nó sẽ chủ động ngắt kết nối trước.

timeout_ms

Thời gian tối đa để chờ. None nghĩa là chờ mãi.

disconnect

Nếu True, khởi tạo ngắt kết nối.

timeout(timeout_ms: int | None) DeviceTimeout

Trả về một context manager hủy thân của nó nếu timeout hết hạn (ném asyncio.TimeoutError) hoặc thiết bị ngắt kết nối (ném DeviceDisconnectedError).

timeout_ms

Timeout tính bằng mili giây, hoặc None nếu không có timeout.

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

Async. Khởi tạo trao đổi ATT MTU và trả về MTU đã thỏa thuận.

mtu

MTU ưa thích tùy chọn để đặt trên giao diện BLE bên dưới trước khi trao đổi.

timeout_ms

Timeout cho việc trao đổi.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) Awaitable[ClientService | None]

Async. Khám phá một dịch vụ từ xa duy nhất khớp với uuid, hoặc None nếu không tìm thấy.

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

Trả về async iterator của các đối tượng ClientService từ xa. Sử dụng với async for và chạy vòng lặp đến hoàn thành.

uuid

Bộ lọc UUID tùy chọn. None trả về mọi dịch vụ.

timeout_ms

Timeout cho mỗi lần khám phá.

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

Async. Khởi tạo ghép đôi trên kết nối này. Cập nhật các thuộc tính encrypted / authenticated / bonded / key_size khi hoàn thành.

bond

Lưu trữ khóa ghép đôi.

le_secure

Sử dụng LE Secure Connections.

mitm

Yêu cầu bảo vệ man-in-the-middle.

io

Hằng số khả năng IO (ví dụ: 3 cho không có đầu vào/đầu ra).

timeout_ms

Timeout ghép đôi.

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

Async. Lắng nghe trên PSM đã cho và trả về L2CAPChannel khi remote mở nó.

psm

Protocol/Service Multiplexer để lắng nghe.

mtu

Kích thước nhận tối đa, tính bằng byte.

timeout_ms

Thời gian tối đa để chờ remote kết nối.

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

Async. Mở kênh L2CAP đến remote trên PSM đã cho.

psm

Protocol/Service Multiplexer để kết nối.

mtu

Kích thước nhận tối đa, tính bằng byte.

timeout_ms

Timeout kết nối.

class aioble.ScanResult

Một thiết bị duy nhất được phát hiện trong scan. Cùng instance được tạo lại khi dữ liệu quảng bá mới đến.

Không tạo trực tiếp.

device

Device bên dưới.

rssi

RSSI được báo cáo lần cuối, tính bằng dBm.

adv_data

Payload quảng bá thô (bytes hoặc None).

resp_data

Payload scan response thô (bytes hoặc None), nếu bật quét tích cực.

connectable

True nếu quảng bá gần nhất là connectable.

name() str | None

Giải mã tên cục bộ quảng bá đầy đủ (hoặc rút gọn) từ payload, hoặc None nếu không có.

services() Iterator[bluetooth.UUID]

Generator tạo ra mỗi bluetooth.UUID được quảng bá trong các trường service-list 16/32/128-bit.

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

Generator tạo ra các tuple (company_id, data) từ các trường quảng bá đặc thù nhà sản xuất.

filter

Nếu được cung cấp, chỉ tạo ra các mục có company ID khớp.

class aioble.Service(uuid: bluetooth.UUID)

Một dịch vụ GATT cục bộ. Xây dựng dịch vụ với một hoặc nhiều instance Characteristic, sau đó truyền nó vào register_services.

uuid

UUID của dịch vụ.

uuid

UUID của dịch vụ.

characteristics

Danh sách các đối tượng Characteristic gắn với dịch vụ này.

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)

Một đặc trưng GATT cục bộ. Tạo một đặc trưng sẽ tự động thêm nó vào service.

service

Service sở hữu.

uuid

UUID của đặc trưng.

read, write, write_no_response, notify, indicate

Các boolean chọn các thao tác GATT được hỗ trợ.

initial

Giá trị ban đầu tùy chọn (bytes).

capture

Nếu True, các giá trị được ghi sẽ được xếp hàng đợi (tối đa 10 sâu) để các lần ghi back-to-back nhanh không bị mất. Mỗi lần gọi written sau đó trả về tuple (connection, data).

uuid

UUID của đặc trưng.

flags

Bitmask của các cờ thuộc tính GATT được xây dựng từ constructor.

descriptors

Danh sách các đối tượng Descriptor gắn với đặc trưng này.

read() bytes

Đọc giá trị hiện tại từ cơ sở dữ liệu GATT cục bộ.

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

Cập nhật giá trị trong cơ sở dữ liệu GATT cục bộ.

data

Bytes giá trị mới.

send_update

Nếu True, cũng notify/indicate mọi kết nối đã đăng ký.

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

Gửi GATT Notify đến connection.

connection

Kết nối client đích.

data

Payload để gửi. Nếu None, giá trị cục bộ hiện tại được gửi.

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

Async. Gửi GATT Indicate đến connection và chờ xác nhận từ client. Ném GattError khi trạng thái khác không.

connection

Kết nối client đích.

data

Payload để indicate, hoặc None để gửi giá trị cục bộ.

timeout_ms

Thời gian tối đa để chờ xác nhận.

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

Async. Chờ một lần ghi từ xa. Trả về DeviceConnection đang ghi, hoặc (connection, data) nếu đặc trưng được tạo với capture=True.

timeout_ms

Thời gian tối đa để chờ. None chờ mãi.

on_read(connection: DeviceConnection) int

Hook override được gọi đồng bộ khi nhận được yêu cầu đọc từ xa. Trả về 0 để cho phép đọc hoặc mã lỗi ATT khác không để từ chối. Triển khai mặc định trả về 0.

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

Một Characteristic có bộ đệm GATT backing có thể được cấu hình. Hữu ích để nhận các giá trị lớn hơn kích thước thuộc tính mặc định, hoặc để xếp hàng đợi các lần ghi back-to-back.

max_len

Kích thước bộ đệm, tính bằng byte.

append

Nếu True, các lần ghi tuần tự sẽ thêm vào bộ đệm thay vì ghi đè.

Các đối số khác chuyển tiếp đến Characteristic.

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

Một bộ mô tả GATT cục bộ. Tạo một bộ mô tả sẽ tự động thêm nó vào characteristic. Kế thừa read, writewritten từ Characteristic.

characteristic

Characteristic sở hữu.

uuid

UUID của bộ mô tả.

read, write

Các boolean chọn các thao tác GATT được hỗ trợ.

initial

Giá trị ban đầu tùy chọn (bytes).

class aioble.ClientService

Một dịch vụ GATT từ xa được khám phá trên peer. Được trả về bởi DeviceConnection.service() hoặc lặp từ DeviceConnection.services().

Không tạo trực tiếp.

connection

DeviceConnection sở hữu.

uuid

UUID dịch vụ từ xa.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) Awaitable[ClientCharacteristic | None]

Async. Khám phá một đặc trưng duy nhất theo UUID, hoặc None nếu không tìm thấy.

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

Trả về async iterator của các đối tượng ClientCharacteristic. Sử dụng với async for và chạy vòng lặp đến hoàn thành.

uuid

Bộ lọc UUID tùy chọn.

timeout_ms

Timeout cho mỗi lần khám phá.

class aioble.ClientCharacteristic

Một đặc trưng GATT từ xa được khám phá trên peer. Được trả về bởi ClientService.characteristic() hoặc lặp từ ClientService.characteristics().

Không tạo trực tiếp.

service

ClientService sở hữu.

uuid

UUID của đặc trưng.

properties

Bitmask của các thao tác GATT được hỗ trợ theo báo cáo của peer.

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

Async. Gửi GATT Read và trả về giá trị. Ném GattError khi trạng thái khác không.

timeout_ms

Timeout đọc.

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

Async. Gửi GATT Write.

data

Giá trị để ghi.

response

True để yêu cầu write-response (và ném GattError khi thất bại). False cho write-without-response. None (mặc định) tự động chọn dựa trên những gì peer quảng bá.

timeout_ms

Timeout ghi (chỉ liên quan nếu responseTrue).

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

Async. Chờ thông báo tiếp theo trên đặc trưng này và trả về payload của nó. Trả về ngay lập tức nếu đã có thông báo trong hàng đợi.

timeout_ms

Thời gian tối đa để chờ. None chờ mãi.

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

Async. Chờ chỉ thị tiếp theo trên đặc trưng này và trả về payload của nó.

timeout_ms

Thời gian tối đa để chờ.

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

Async. Ghi Client Characteristic Configuration Descriptor (CCCD) để đăng ký (hoặc hủy đăng ký) nhận thông báo và/hoặc chỉ thị.

notify

Bật thông báo.

indicate

Bật chỉ thị.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) Awaitable[ClientDescriptor | None]

Async. Khám phá một bộ mô tả duy nhất theo UUID, hoặc None nếu không tìm thấy.

descriptors(timeout_ms: int = 2000) ClientDiscover

Trả về async iterator của các đối tượng ClientDescriptor. Sử dụng với async for và chạy vòng lặp đến hoàn thành.

class aioble.ClientDescriptor

Một bộ mô tả GATT từ xa được khám phá trên peer. Kế thừa readwrite từ ClientCharacteristic.

Không tạo trực tiếp.

characteristic

ClientCharacteristic sở hữu.

uuid

UUID của bộ mô tả.

class aioble.L2CAPChannel

Một kênh L2CAP hướng kết nối đang hoạt động. Được trả về bởi DeviceConnection.l2cap_accept() hoặc DeviceConnection.l2cap_connect(). Hỗ trợ sử dụng như context manager async with tự động ngắt kết nối khi thoát.

Không tạo trực tiếp.

our_mtu

Kích thước tối đa, tính bằng byte, mà peer có thể gửi đến chúng ta trong một SDU duy nhất.

peer_mtu

Kích thước tối đa, tính bằng byte, mà chúng ta có thể gửi đến peer trong một SDU duy nhất.

available() bool

Đồng bộ trả về True nếu dữ liệu nhận được đệm đã sẵn sàng (tức là recvinto sẽ không bị chặn).

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

Async. Nhận vào buf, trả về số byte đã đọc. Chờ dữ liệu mới nếu kênh trống.

buf

Bộ đệm được cấp phát trước để điền vào.

timeout_ms

Thời gian tối đa để chờ. None chờ mãi.

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

Async. Gửi buf trên kênh, phân mảnh các payload lớn hơn thành các chunk có kích thước MTU. Chờ tín dụng kiểm soát luồng khi cần.

buf

Đối tượng dạng bytes để gửi.

timeout_ms

Thời gian tối đa để chờ mỗi chunk.

chunk_size

Override tùy chọn cho kích thước chunk mỗi lần gọi. Bị giới hạn ở min(our_mtu * 2, peer_mtu).

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

Async. Chờ cho đến khi bất kỳ send đang bị trì hoãn nào đã được controller drain.

timeout_ms

Thời gian tối đa để chờ.

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

Async. Ngắt kết nối kênh và chờ IRQ ngắt kết nối.

timeout_ms

Thời gian tối đa để chờ.

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

Async. Chờ cho đến khi kênh bị ngắt kết nối bởi một trong hai bên.

timeout_ms

Thời gian tối đa để chờ.