`aioble` — BLE Assíncrono¶

aioble é um wrapper de alto nível compatível com asyncio em torno do módulo bluetooth. Ele fornece coroutines limpas para escaneamento, conexão, anúncio (advertising), serviços GATT e canais L2CAP.

Todas as operações remotas (connect, disconnect, leitura/escrita do cliente, indicate do servidor, recv/send L2CAP, pair) são aguardáveis (awaitable) e suportam timeouts.

Papéis suportados:

Broadcaster (anunciante) — gera payloads de anúncio (advertising) e de resposta de escaneamento para campos comuns, divide automaticamente o payload entre anúncio e resposta de escaneamento, anuncia indefinidamente ou por uma duração fixa.
Peripheral (periférico) — aguarda a conexão de um central, aguarda a troca de MTU.
Observer (scanner) — escaneamento passivo e ativo, combina os payloads de anúncio e de resposta de escaneamento para o mesmo dispositivo, analisa campos comuns dos payloads de anúncio.
Central — conecta-se a um periférico, inicia a troca de MTU.
Cliente GATT — descobre serviços / características / descritores (opcionalmente por UUID); lê / escreve / escreve-com-resposta em características e descritores; assina notificações e indicações (via o CCCD); aguarda notificações e indicações.
Servidor GATT — registra serviços / características / descritores; aguarda escritas em características e descritores; intercepta requisições de leitura; envia notificações e indicações (e aguarda a resposta).
L2CAP — aceita e conecta canais L2CAP orientados a conexão, gerencia o controle de fluxo do canal.
Segurança — gerenciamento de chaves/segredos baseado em JSON, inicia o pareamento, consulta o estado de criptografia / autenticação.

Exemplos¶

Escanear dispositivos BLE próximos e imprimir cada um conforme é visto:

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

Conectar-se a um periférico que anuncia o serviço Heart Rate como um central e assinar suas notificações de medição:

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

Atuar como um periférico: registrar um serviço GATT, anunciá-lo e enviar notificações a quem se conectar:

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

Funções de nível de módulo¶

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

Encaminha para bluetooth.BLE.config(), garantindo que o rádio BLE esteja ativo primeiro.

args: Nome opcional de um único parâmetro a consultar.
kwargs: Argumentos nomeados para definir valores de configuração.

aioble.stop() → None¶: Desativa o rádio BLE subjacente e executa quaisquer handlers de encerramento de submódulos registrados. Após chamar isto, scanners, anunciantes, conexões e canais L2CAP são todos desmontados.

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

Retorna um gerenciador de contexto assíncrono / iterador assíncrono scan que produz instâncias de ScanResult para cada dispositivo único descoberto (ou para cada novo dado de anúncio de um dispositivo conhecido).

duration_ms: Por quanto tempo escanear, em milissegundos. Passe 0 para escanear indefinidamente até que o gerenciador de contexto seja encerrado.
interval_us: Intervalo de escaneamento em microssegundos. O padrão é 1.280.000.
window_us: Janela de escaneamento em microssegundos (deve ser menor ou igual a interval_us). O padrão é 11.250.
active: Se True, realiza um escaneamento ativo (requisita dados de resposta de escaneamento). O padrão é 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 assíncrona que começa a anunciar e aguarda uma conexão de central recebida. Retorna um DeviceConnection representando o central conectado, ou levanta asyncio.TimeoutError em caso de timeout.

interval_us: Intervalo de anúncio, em microssegundos.
adv_data: Payload de anúncio bruto. Se não definido, adv_data é construído a partir dos argumentos nomeados restantes.
resp_data: Payload de resposta de escaneamento bruto. Preenchido automaticamente com o transbordamento de adv_data se necessário.
connectable: Se True, este é um anúncio conectável.
limited_disc: Usa a flag de descoberta limitada (limited-discoverable) em vez da geral.
br_edr: Define a flag de suporte a BR/EDR.
name: Nome local completo opcional a incorporar.
services: Iterável de bluetooth.UUID a anunciar.
appearance: Valor de aparência (appearance) de 16 bits (veja os números atribuídos Bluetooth).
manufacturer: Tupla de (company_id, data_bytes) a anunciar como dados específicos do fabricante.
timeout_ms: Para de anunciar após esta quantidade de milissegundos sem uma conexão. None significa anunciar até que haja conexão.

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

Registra um ou mais objetos Service (e suas características e descritores) no servidor GATT. Deve ser chamado uma vez antes de iniciar advertise. Chamadas subsequentes substituem o registro anterior.

services: Uma ou mais instâncias de Service.

Constantes de nível de módulo¶

aioble.ADDR_PUBLIC¶: Tipo de endereço de dispositivo BLE público (0).

aioble.ADDR_RANDOM¶: Tipo de endereço de dispositivo BLE aleatório (1).

Exceções¶

exception aioble.GattError¶: Levantada quando uma operação GATT remota (read / write / indicate) é concluída com um status diferente de zero. O código de status está disponível no atributo _status.

exception aioble.DeviceDisconnectedError¶: Levantada dentro de uma operação assíncrona (por exemplo, read, write, notified) quando a conexão subjacente cai durante a espera.

exception aioble.L2CAPDisconnectedError¶: Levantada quando uma operação de send/recv/flush de um canal L2CAP é tentada em (ou interrompida por) um canal desconectado.

exception aioble.L2CAPConnectionError¶: Levantada por DeviceConnection.l2cap_connect quando o estabelecimento do canal falha. O código de status Bluetooth é o primeiro argumento.

Classes¶

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

Representa um dispositivo BLE remoto por endereço. Duas instâncias de Device são consideradas iguais se tanto addr_type quanto addr coincidirem. Usado como o handle para iniciar conexões.

addr_type: Ou ADDR_PUBLIC ou ADDR_RANDOM.
addr: Endereço de seis bytes como bytes, ou uma string hexadecimal separada por dois-pontos (por exemplo, "aa:bb:cc:dd:ee:ff").

addr_type¶: O tipo de endereço com o qual o dispositivo foi construído.

addr¶: O endereço bruto de seis bytes do dispositivo.

addr_hex() → str¶: Retorna o endereço formatado como uma string hexadecimal separada por dois-pontos.

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

Assíncrono. Inicia uma conexão GAP com este dispositivo e retorna o DeviceConnection resultante. Cancela qualquer escaneamento em andamento.

timeout_ms: Por quanto tempo aguardar a conclusão da conexão.
scan_duration_ms: Duração do escaneamento inicial antes de conectar (específico do controlador).
min_conn_interval_us / max_conn_interval_us: Limites opcionais do intervalo de conexão, em microssegundos.

class aioble.DeviceConnection¶

Uma conexão GAP ativa com um Device. Retornada por Device.connect() ou advertise. Suporta uso como um gerenciador de contexto async with que desconecta automaticamente ao sair.

Não construa diretamente.

device¶: O Device subjacente.

encrypted¶: True assim que o link estiver criptografado (por exemplo, após o pareamento).

authenticated¶: True se o link foi autenticado (pareamento protegido contra MITM).

bonded¶: True se o pareamento produziu chaves de vínculo (bonding).

key_size¶: Tamanho da chave de criptografia negociado em bytes, ou False se não estiver criptografado.

mtu¶: MTU ATT negociado após exchange_mtu, ou None até ser definido.

is_connected() → bool¶: Retorna se a conexão ainda está ativa.

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

Assíncrono. Desconecta e aguarda a IRQ de desconexão.

timeout_ms: Tempo máximo de espera pela desconexão.

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

Assíncrono. Aguarda que a conexão seja encerrada por qualquer um dos lados. Se disconnect for True, ele desconecta ativamente primeiro.

timeout_ms: Tempo máximo de espera. None significa esperar para sempre.
disconnect: Se True, inicia a desconexão.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Retorna um gerenciador de contexto que cancela seu corpo se o timeout expirar (levantando asyncio.TimeoutError) ou se o dispositivo desconectar (levantando DeviceDisconnectedError).

timeout_ms: Timeout em milissegundos, ou None para nenhum timeout.

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

Assíncrono. Inicia uma troca de MTU ATT e retorna o MTU negociado.

mtu: MTU preferido opcional a definir na interface BLE subjacente antes da troca.
timeout_ms: Timeout para a troca.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Assíncrono. Descobre um único serviço remoto que corresponde a uuid, ou None se não encontrado.

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

Retorna um iterador assíncrono de objetos ClientService remotos. Use com async for e execute o laço até a conclusão.

uuid: Filtro de UUID opcional. None retorna todos os serviços.
timeout_ms: Timeout por descoberta.

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

Assíncrono. Inicia o pareamento nesta conexão. Atualiza os atributos encrypted / authenticated / bonded / key_size quando concluído.

bond: Persiste as chaves de pareamento.
le_secure: Usa LE Secure Connections.
mitm: Requer proteção man-in-the-middle.
io: Constante de capacidade de IO (por exemplo, 3 para nenhuma entrada/saída).
timeout_ms: Timeout do pareamento.

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

Assíncrono. Escuta no PSM fornecido e retorna um L2CAPChannel assim que o remoto o abrir.

psm: Multiplexador de Protocolo/Serviço (PSM) no qual escutar.
mtu: Tamanho máximo de recebimento, em bytes.
timeout_ms: Tempo máximo de espera para que o remoto se conecte.

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

Assíncrono. Abre um canal L2CAP para o remoto no PSM fornecido.

psm: Multiplexador de Protocolo/Serviço (PSM) ao qual conectar.
mtu: Tamanho máximo de recebimento, em bytes.
timeout_ms: Timeout de conexão.

class aioble.ScanResult¶

Um único dispositivo descoberto durante o scan. A mesma instância é reproduzida (re-yielded) conforme novos dados de anúncio chegam.

Não construa diretamente.

device¶: O Device subjacente.

rssi¶: Último RSSI reportado, em dBm.

adv_data¶: Payload de anúncio bruto (bytes ou None).

resp_data¶: Payload de resposta de escaneamento bruto (bytes ou None), se o escaneamento ativo estiver habilitado.

connectable¶: True se o anúncio mais recente era conectável.

name() → str | None¶: Decodifica o nome local anunciado completo (ou abreviado) do payload, ou None se não estiver presente.

services() → Iterator[bluetooth.UUID]¶: Gerador que produz cada bluetooth.UUID anunciado nos campos de lista de serviços de 16/32/128 bits.

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

Gerador que produz tuplas (company_id, data) a partir dos campos de anúncio específicos do fabricante.

filter: Se fornecido, produz apenas as entradas cujo ID de fabricante (company ID) corresponde.

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

Um serviço GATT local. Construa um serviço com uma ou mais instâncias de Characteristic e então passe-o para register_services.

uuid: O UUID do serviço.

uuid¶: O UUID do serviço.

characteristics¶: Lista de objetos Characteristic vinculados a este serviço.

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

Uma característica GATT local. Construir uma anexa-a automaticamente a service.

service: O Service proprietário.
uuid: O UUID da característica.
read, write, write_no_response, notify, indicate: Booleanos que selecionam as operações GATT suportadas.
initial: Valor inicial opcional (bytes).
capture: Se True, os valores escritos são enfileirados (até 10 de profundidade) para que escritas rápidas consecutivas não sejam perdidas. Cada chamada written então retorna uma tupla (connection, data).

uuid¶: O UUID da característica.

flags¶: Bitmask das flags de propriedade GATT construída a partir do construtor.

descriptors¶: Lista de objetos Descriptor vinculados a esta característica.

read() → bytes¶: Lê o valor atual do banco de dados GATT local.

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

Atualiza o valor no banco de dados GATT local.

data: Novos bytes de valor.
send_update: Se True, também notifica/indica todas as conexões assinadas.

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

Envia um Notify GATT para connection.

connection: A conexão de cliente alvo.
data: Payload a enviar. Se None, o valor local atual é enviado.

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

Assíncrono. Envia um Indicate GATT para connection e aguarda a confirmação do cliente. Levanta GattError em caso de status diferente de zero.

connection: A conexão de cliente alvo.
data: Payload a indicar, ou None para enviar o valor local.
timeout_ms: Tempo máximo de espera pela confirmação.

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

Assíncrono. Aguarda uma escrita remota. Retorna o DeviceConnection que escreveu, ou (connection, data) se a característica foi criada com capture=True.

timeout_ms: Tempo máximo de espera. None aguarda para sempre.

on_read(connection: DeviceConnection) → int¶: Hook de sobrescrita invocado de forma síncrona quando uma leitura remota é recebida. Retorne 0 para permitir a leitura ou um código de erro ATT diferente de zero para rejeitá-la. A implementação padrão retorna 0.

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

Uma Characteristic cujo buffer GATT de apoio pode ser configurado. Útil para receber valores maiores que o tamanho de atributo padrão, ou para enfileirar escritas consecutivas.

max_len: Tamanho do buffer, em bytes.
append: Se True, escritas sequenciais são anexadas ao buffer em vez de sobrescrevê-lo.

Outros argumentos são encaminhados para Characteristic.

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

Um descritor GATT local. Construir um anexa-o automaticamente a characteristic. Herda read, write e written de Characteristic.

characteristic: A Characteristic proprietária.
uuid: O UUID do descritor.
read, write: Booleanos que selecionam as operações GATT suportadas.
initial: Valor inicial opcional (bytes).

class aioble.ClientService¶

Um serviço GATT remoto descoberto em um par (peer). Retornado por DeviceConnection.service() ou iterado a partir de DeviceConnection.services().

Não construa diretamente.

connection¶: O DeviceConnection proprietário.

uuid¶: O UUID do serviço remoto.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Assíncrono. Descobre uma única característica por UUID, ou None se não encontrada.

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

Retorna um iterador assíncrono de objetos ClientCharacteristic. Use com async for e execute o laço até a conclusão.

uuid: Filtro de UUID opcional.
timeout_ms: Timeout por descoberta.

class aioble.ClientCharacteristic¶

Uma característica GATT remota descoberta em um par (peer). Retornada por ClientService.characteristic() ou iterada a partir de ClientService.characteristics().

Não construa diretamente.

service¶: O ClientService proprietário.

uuid¶: O UUID da característica.

properties¶: Bitmask das operações GATT suportadas conforme reportado pelo par (peer).

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

Assíncrono. Emite uma Read GATT e retorna o valor. Levanta GattError em caso de status diferente de zero.

timeout_ms: Timeout de leitura.

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

Assíncrono. Emite uma Write GATT.

data: Valor a escrever.
response: True para exigir uma resposta de escrita (write-response) (e levantar GattError em caso de falha). False para escrita-sem-resposta (write-without-response). None (padrão) seleciona automaticamente com base no que o par (peer) anuncia.
timeout_ms: Timeout de escrita (relevante apenas se response for True).

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

Assíncrono. Aguarda a próxima notificação nesta característica e retorna seu payload. Retorna imediatamente se uma notificação já estiver enfileirada.

timeout_ms: Tempo máximo de espera. None aguarda para sempre.

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

Assíncrono. Aguarda a próxima indicação nesta característica e retorna seu payload.

timeout_ms: Tempo máximo de espera.

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

Assíncrono. Escreve o Client Characteristic Configuration Descriptor (CCCD) para assinar (ou cancelar a assinatura de) notificações e/ou indicações.

notify: Habilita notificações.
indicate: Habilita indicações.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Assíncrono. Descobre um único descritor por UUID, ou None se não encontrado.

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: Retorna um iterador assíncrono de objetos ClientDescriptor. Use com async for e execute o laço até a conclusão.

class aioble.ClientDescriptor¶

Um descritor GATT remoto descoberto em um par (peer). Herda read e write de ClientCharacteristic.

Não construa diretamente.

characteristic¶: A ClientCharacteristic proprietária.

uuid¶: O UUID do descritor.

class aioble.L2CAPChannel¶

Um canal L2CAP orientado a conexão ativo. Retornado por DeviceConnection.l2cap_accept() ou DeviceConnection.l2cap_connect(). Suporta uso como um gerenciador de contexto async with que desconecta automaticamente ao sair.

Não construa diretamente.

our_mtu¶: Tamanho máximo, em bytes, que o par (peer) pode nos enviar em uma única SDU.

peer_mtu¶: Tamanho máximo, em bytes, que podemos enviar ao par (peer) em uma única SDU.

available() → bool¶: Retorna de forma síncrona True se houver dados de recebimento em buffer prontos (ou seja, recvinto não bloqueará).

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

Assíncrono. Recebe dentro de buf, retornando o número de bytes lidos. Aguarda novos dados se o canal estiver vazio.

buf: Buffer pré-alocado a preencher.
timeout_ms: Tempo máximo de espera. None aguarda para sempre.

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

Assíncrono. Envia buf no canal, fragmentando payloads maiores em pedaços do tamanho do MTU. Aguarda créditos de controle de fluxo conforme necessário.

buf: Objeto do tipo bytes a enviar.
timeout_ms: Tempo máximo de espera por pedaço.
chunk_size: Sobrescrita opcional para o tamanho de pedaço por chamada. Limitado a min(our_mtu * 2, peer_mtu).

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

Assíncrono. Aguarda até que qualquer send estagnado tenha sido drenado pelo controlador.

timeout_ms: Tempo máximo de espera.

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

Assíncrono. Desconecta o canal e aguarda a IRQ de desconexão.

timeout_ms: Tempo máximo de espera.

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

Assíncrono. Aguarda até que o canal seja desconectado por qualquer um dos lados.

timeout_ms: Tempo máximo de espera.

aioble — BLE Assíncrono¶

Exemplos¶

Funções de nível de módulo¶

Constantes de nível de módulo¶

Exceções¶

Classes¶

`aioble` — BLE Assíncrono¶