`aioble` — асинхронный BLE¶

aioble — это высокоуровневая, дружелюбная к asyncio обёртка вокруг модуля bluetooth. Она предоставляет удобные корутины для сканирования, подключения, объявления, GATT-сервисов и L2CAP-каналов.

Все удалённые операции (connect, disconnect, чтение/запись на стороне клиента, indicate на стороне сервера, l2cap recv/send, pair) являются awaitable и поддерживают тайм-ауты.

Поддерживаемые роли:

Broadcaster (рекламодатель) — формирование полезной нагрузки рекламных и scan-response пакетов для распространённых полей, автоматическое разделение полезной нагрузки между рекламным пакетом и scan response, реклама бесконечно или в течение заданного времени.
Peripheral — ожидание подключения от центрального устройства, ожидание обмена MTU.
Observer (сканер) — пассивное и активное сканирование, объединение полезной нагрузки рекламы и scan-response для одного устройства, разбор распространённых полей из рекламной полезной нагрузки.
Central — подключение к периферийному устройству, инициирование обмена MTU.
GATT Client — обнаружение сервисов / характеристик / дескрипторов (опционально по UUID); чтение / запись / запись с ответом для характеристик и дескрипторов; подписка на уведомления и индикации (через CCCD); ожидание уведомлений и индикаций.
GATT Server — регистрация сервисов / характеристик / дескрипторов; ожидание записей в характеристики и дескрипторы; перехват запросов на чтение; отправка уведомлений и индикаций (с ожиданием ответа).
L2CAP — приём и установление L2CAP-каналов, ориентированных на соединение, управление потоком данных в канале.
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¶

Возвращает асинхронный контекстный менеджер / асинхронный итератор scan, выдающий экземпляры ScanResult для каждого уникального обнаруженного устройства (или для каждой новой порции рекламных данных от уже известного устройства).

duration_ms: Как долго сканировать, в миллисекундах. Передайте 0, чтобы сканировать бесконечно, пока не завершится контекстный менеджер.
interval_us: Интервал сканирования в микросекундах. По умолчанию 1 280 000.
window_us: Окно сканирования в микросекундах (должно быть меньше или равно interval_us). По умолчанию 11 250.
active: Если True, выполняется активное сканирование (запрос данных scan response). По умолчанию 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: Сырая полезная нагрузка scan response. Автоматически заполняется переполнением из adv_data при необходимости.
connectable: Если True, это реклама с возможностью подключения.
limited_disc: Использовать флаг ограниченной обнаруживаемости вместо общей.
br_edr: Установить флаг поддержки BR/EDR.
name: Опциональное полное локальное имя для встраивания.
services: Итерируемый набор bluetooth.UUID для рекламы.
appearance: 16-битное значение appearance (см. назначенные номера Bluetooth).
manufacturer: Кортеж (company_id, data_bytes) для рекламы в виде данных, специфичных для производителя.
timeout_ms: Прекратить рекламу после указанного количества миллисекунд без подключения. None означает рекламировать до подключения.

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

Регистрирует один или несколько объектов Service (вместе с их характеристиками и дескрипторами) на GATT-сервере. Должен быть вызван один раз перед запуском 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¶: Вызывается при попытке выполнить операцию send/recv/flush для L2CAP-канала на отключённом канале (или при её прерывании отключением канала).

exception aioble.L2CAPConnectionError¶: Вызывается методом DeviceConnection.l2cap_connect, когда установление канала завершается неудачей. Код статуса Bluetooth передаётся первым аргументом.

Классы¶

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

Представляет удалённое BLE-устройство по адресу. Два экземпляра Device считаются равными, если совпадают и addr_type, и addr. Используется как дескриптор для инициирования подключений.

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¶

Активное GAP-подключение к Device. Возвращается методом Device.connect() или advertise. Поддерживает использование в качестве контекстного менеджера async with, который автоматически отключается при выходе.

Не создавайте напрямую.

device¶: Базовое устройство Device.

encrypted¶: True, как только канал зашифрован (например, после сопряжения).

authenticated¶: True, если канал был аутентифицирован (сопряжение с защитой от MITM).

bonded¶: True, если сопряжение создало ключи привязки.

key_size¶: Согласованный размер ключа шифрования в байтах или False, если соединение не зашифровано.

mtu¶: Согласованный ATT MTU после exchange_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 Secure Connections.
mitm: Требовать защиту от атаки «человек посередине».
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]¶

Асинхронный. Открывает L2CAP-канал к удалённой стороне на заданном PSM.

psm: Protocol/Service Multiplexer для подключения.
mtu: Максимальный размер приёма, в байтах.
timeout_ms: Тайм-аут подключения.

class aioble.ScanResult¶

Одно устройство, обнаруженное во время scan. Один и тот же экземпляр повторно выдаётся по мере поступления новых рекламных данных.

Не создавайте напрямую.

device¶: Базовое устройство Device.

rssi¶: Последний зарегистрированный RSSI, в дБм.

adv_data¶: Сырая рекламная полезная нагрузка (bytes или None).

resp_data¶: Сырая полезная нагрузка scan response (bytes или None), если включено активное сканирование.

connectable¶: True, если последняя реклама допускала подключение.

name() → str | None¶: Декодирует полное (или сокращённое) объявленное локальное имя из полезной нагрузки или None, если оно отсутствует.

services() → Iterator[bluetooth.UUID]¶: Генератор, выдающий каждый bluetooth.UUID, объявленный в полях списка сервисов 16/32/128-бит.

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

Генератор, выдающий кортежи (company_id, data) из полей рекламы, специфичных для производителя.

filter: Если задан, выдаются только записи, чей идентификатор производителя совпадает.

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¶

Отправляет GATT Notify в connection.

connection: Целевое клиентское соединение.
data: Полезная нагрузка для отправки. Если None, отправляется текущее локальное значение.

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

Асинхронный. Отправляет GATT Indicate в connection и ожидает подтверждения от клиента. Вызывает GattError при ненулевом статусе.

connection: Целевое клиентское соединение.
data: Полезная нагрузка для indicate или None для отправки локального значения.
timeout_ms: Максимальное время ожидания подтверждения.

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

Асинхронный. Ожидает удалённой записи. Возвращает записывающее DeviceConnection или (connection, data), если характеристика была создана с capture=True.

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

Characteristic, чей резервный GATT-буфер можно настроить. Полезно для приёма значений больше размера атрибута по умолчанию или для постановки в очередь последовательных записей.

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. Наследует read, write и written от Characteristic.

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-дескриптор, обнаруженный на одноранговом устройстве. Наследует read и write от ClientCharacteristic.

Не создавайте напрямую.

characteristic¶: Характеристика-владелец ClientCharacteristic.

uuid¶: UUID дескриптора.

class aioble.L2CAPChannel¶

Активный L2CAP-канал, ориентированный на соединение. Возвращается методом DeviceConnection.l2cap_accept() или DeviceConnection.l2cap_connect(). Поддерживает использование в качестве контекстного менеджера async with, который автоматически отключается при выходе.

Не создавайте напрямую.

our_mtu¶: Максимальный размер в байтах, который одноранговое устройство может отправить нам в одном SDU.

peer_mtu¶: Максимальный размер в байтах, который мы можем отправить одноранговому устройству в одном SDU.

available() → bool¶: Синхронно возвращает True, если буферизованные данные приёма готовы (т. е. recvinto не будет блокироваться).

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¶