`aioble` — BLE Assíncrono¶

aioble é um invólucro de alto nível compatível com asyncio para o módulo bluetooth. Fornece corrotinas limpas para digitalização, ligação, publicidade, serviços GATT e canais L2CAP.

Todas as operações remotas (ligar, desligar, leitura/escrita do cliente, indicação do servidor, receção/envio l2cap, emparelhamento) são aguardáveis e suportam tempos limite.

Funções suportadas:

Difusor (anunciante) — gera payloads de publicidade e de resposta a digitalização para campos comuns, divide automaticamente o payload entre publicidade e resposta a digitalização, anuncia indefinidamente ou por uma duração fixa.
Periférico — aguarda ligação de um central, aguarda troca de MTU.
Observador (scanner) — digitalização passiva e ativa, combina payloads de publicidade e de resposta a digitalização para o mesmo dispositivo, analisa campos comuns a partir dos payloads de publicidade.
Central — liga-se a um periférico, inicia troca de MTU.
Cliente GATT — descobre serviços / características / descritores (opcionalmente por UUID); leitura / escrita / escrita-com-resposta em características e descritores; subscreve notificações e indicações (via CCCD); aguarda notificações e indicações.
Servidor GATT — regista serviços / características / descritores; aguarda escritas em características e descritores; interceta pedidos de leitura; envia notificações e indicações (e aguarda resposta).
L2CAP — aceita e liga canais L2CAP orientados à ligação, gere o controlo de fluxo do canal.
Segurança — gestão de chaves/segredos com suporte JSON, inicia emparelhamento, consulta estado de encriptação / autenticação.

Exemplos¶

Digitalizar dispositivos BLE próximos e imprimir cada um à medida que é detetado:

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

Ligar a um periférico que anuncia o serviço de Frequência Cardíaca como central e subscrever as 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 periférico: registar um serviço GATT, anunciá-lo e enviar notificações a quem se ligar:

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 ao nível do módulo¶

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

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

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

aioble.stop() → None¶: Desativa o rádio BLE subjacente e executa quaisquer manipuladores de encerramento de sub-módulos registados. Após chamar isto, scanners, anunciantes, ligações e canais L2CAP são todos encerrados.

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

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

duration_ms: Duração da digitalização, em milissegundos. Passe 0 para digitalizar indefinidamente até o gestor de contexto sair.
interval_us: Intervalo de digitalização em microssegundos. Predefinição: 1.280.000.
window_us: Janela de digitalização em microssegundos (deve ser menor ou igual a interval_us). Predefinição: 11.250.
active: Se True, realiza uma digitalização ativa (solicita dados de resposta a digitalização). Predefiniçã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¶

Corrotina assíncrona que inicia a publicidade e aguarda uma ligação central de entrada. Devolve uma DeviceConnection representando o central ligado, ou lança asyncio.TimeoutError em caso de tempo limite.

interval_us: Intervalo de publicidade, em microssegundos.
adv_data: Payload de publicidade em bruto. Se não definido, adv_data é construído a partir dos restantes argumentos de palavra-chave.
resp_data: Payload de resposta a digitalização em bruto. Preenchido automaticamente com o excesso de adv_data se necessário.
connectable: Se True, este é um anúncio ligável.
limited_disc: Usa a flag de descoberta limitada em vez da geral.
br_edr: Define a flag de suporte a BR/EDR.
name: Nome local completo opcional a incluir.
services: Iterável de bluetooth.UUID a anunciar.
appearance: Valor de aparência de 16 bits (ver números atribuídos Bluetooth).
manufacturer: Tuplo de (company_id, data_bytes) a anunciar como dados específicos do fabricante.
timeout_ms: Pára a publicidade após este número de milissegundos sem uma ligação. None significa anunciar até haver ligação.

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

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

services: Uma ou mais instâncias Service.

Constantes ao nível do 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¶: Lançada quando uma operação GATT remota (leitura / escrita / indicação) termina com um estado não nulo. O código de estado está disponível no atributo _status.

exception aioble.DeviceDisconnectedError¶: Lançada dentro de uma operação assíncrona (p.ex. leitura, escrita, notificação) quando a ligação subjacente cai enquanto aguarda.

exception aioble.L2CAPDisconnectedError¶: Lançada quando se tenta uma operação de envio/receção/descarga num canal L2CAP sobre um canal desligado (ou interrompida por ele).

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

Classes¶

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

Representa um dispositivo BLE remoto pelo endereço. Duas instâncias Device são iguais se tanto addr_type como addr coincidirem. Usado como identificador para iniciar ligações.

addr_type: Pode ser ADDR_PUBLIC ou ADDR_RANDOM.
addr: Endereço de seis bytes como bytes, ou uma cadeia hexadecimal separada por dois-pontos (p.ex. "aa:bb:cc:dd:ee:ff").

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

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

addr_hex() → str¶: Devolve o endereço formatado como uma cadeia 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 ligação GAP a este dispositivo e devolve a DeviceConnection resultante. Cancela qualquer digitalização em curso.

timeout_ms: Tempo a aguardar para a ligação ser concluída.
scan_duration_ms: Duração de digitalização inicial antes de ligar (específico do controlador).
min_conn_interval_us / max_conn_interval_us: Limites opcionais do intervalo de ligação, em microssegundos.

class aioble.DeviceConnection¶

Uma ligação GAP ativa a um Device. Devolvida por Device.connect() ou advertise. Suporta utilização como gestor de contexto async with que desliga automaticamente ao sair.

Não construir diretamente.

device¶: O Device subjacente.

encrypted¶: True assim que a ligação está encriptada (p.ex. após emparelhamento).

authenticated¶: True se a ligação foi autenticada (emparelhamento com proteção MITM).

bonded¶: True se o emparelhamento produziu chaves de vinculação.

key_size¶: Tamanho negociado da chave de encriptação em bytes, ou False se não estiver encriptada.

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

is_connected() → bool¶: Devolve se a ligação ainda está ativa.

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

Assíncrono. Desliga e aguarda o IRQ de desligação.

timeout_ms: Tempo máximo a aguardar pela desligação.

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

Assíncrono. Aguarda que a ligação seja terminada por qualquer um dos lados. Se disconnect for True, desliga ativamente primeiro.

timeout_ms: Tempo máximo a aguardar. None significa aguardar indefinidamente.
disconnect: Se True, inicia a desligação.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Devolve um gestor de contexto que cancela o seu corpo se o tempo limite expirar (lançando asyncio.TimeoutError) ou se o dispositivo se desligar (lançando DeviceDisconnectedError).

timeout_ms: Tempo limite em milissegundos, ou None sem tempo limite.

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

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

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

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

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

Devolve um iterador assíncrono de objetos ClientService remotos. Use com async for e execute o ciclo até à conclusão.

uuid: Filtro UUID opcional. None devolve todos os serviços.
timeout_ms: Tempo limite 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 emparelhamento nesta ligação. Atualiza os atributos encrypted / authenticated / bonded / key_size quando concluído.

bond: Persiste as chaves de emparelhamento.
le_secure: Usa Ligações Seguras LE.
mitm: Requer proteção contra ataques man-in-the-middle.
io: Constante de capacidade de E/S (p.ex. 3 para sem entrada/saída).
timeout_ms: Tempo limite de emparelhamento.

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

Assíncrono. Escuta no PSM indicado e devolve um L2CAPChannel assim que o remoto o abrir.

psm: Multiplexador de Protocolo/Serviço a escutar.
mtu: Tamanho máximo de receção, em bytes.
timeout_ms: Tempo máximo a aguardar que o remoto se ligue.

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

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

psm: Multiplexador de Protocolo/Serviço a ligar.
mtu: Tamanho máximo de receção, em bytes.
timeout_ms: Tempo limite da ligação.

class aioble.ScanResult¶

Um único dispositivo descoberto durante scan. A mesma instância é re-produzida à medida que chegam novos dados de publicidade.

Não construir diretamente.

device¶: O Device subjacente.

rssi¶: Último RSSI reportado, em dBm.

adv_data¶: Payload de publicidade em bruto (bytes ou None).

resp_data¶: Payload de resposta a digitalização em bruto (bytes ou None), se a digitalização ativa estiver ativada.

connectable¶: True se o anúncio mais recente foi ligável.

name() → str | None¶: Descodifica o nome local anunciado completo (ou abreviado) a partir 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 tuplos (company_id, data) a partir de campos de publicidade específicos do fabricante.

filter: Se fornecido, apenas produz entradas cujo ID de empresa corresponda.

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

Um serviço GATT local. Crie um serviço com uma ou mais instâncias Characteristic, depois passe-o para register_services.

uuid: O UUID do serviço.

uuid¶: O UUID do serviço.

characteristics¶: Lista de objetos Characteristic associados 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 appende-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 colocados em fila (até 10) para que escritas consecutivas rápidas não se percam. Cada chamada written devolve então um tuplo (connection, data).

uuid¶: O UUID da característica.

flags¶: Máscara de bits das flags de propriedade GATT construída a partir do construtor.

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

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

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

Atualiza o valor na base de dados GATT local.

data: Novos bytes de valor.
send_update: Se True, também notifica/indica cada ligação subscrita.

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

Envia uma Notificação GATT para connection.

connection: A ligação do 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 uma Indicação GATT para connection e aguarda a confirmação do cliente. Lança GattError em caso de estado não nulo.

connection: A ligação do cliente alvo.
data: Payload a indicar, ou None para enviar o valor local.
timeout_ms: Tempo máximo a aguardar confirmação.

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

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

timeout_ms: Tempo máximo a aguardar. None aguarda indefinidamente.

on_read(connection: DeviceConnection) → int¶: Função hook invocada de forma síncrona quando é recebida uma leitura remota. Devolve 0 para permitir a leitura ou um código de erro ATT não nulo para rejeitá-la. A implementação predefinida devolve 0.

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

Uma Characteristic cujo buffer GATT de suporte pode ser configurado. Útil para receber valores maiores do que o tamanho padrão do atributo, ou para colocar em fila escritas consecutivas.

max_len: Tamanho do buffer, em bytes.
append: Se True, escritas sequenciais acrescentam ao buffer em vez de sobrescrever.

Outros argumentos são passados 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 appende-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 num par. Devolvido por DeviceConnection.service() ou iterado a partir de DeviceConnection.services().

Não construir diretamente.

connection¶: A DeviceConnection proprietária.

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 for encontrada.

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

Devolve um iterador assíncrono de objetos ClientCharacteristic. Use com async for e execute o ciclo até à conclusão.

uuid: Filtro UUID opcional.
timeout_ms: Tempo limite por descoberta.

class aioble.ClientCharacteristic¶

Uma característica GATT remota descoberta num par. Devolvida por ClientService.characteristic() ou iterada a partir de ClientService.characteristics().

Não construir diretamente.

service¶: O ClientService proprietário.

uuid¶: O UUID da característica.

properties¶: Máscara de bits das operações GATT suportadas conforme reportado pelo par.

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

Assíncrono. Emite uma Leitura GATT e devolve o valor. Lança GattError em caso de estado não nulo.

timeout_ms: Tempo limite de leitura.

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

Assíncrono. Emite uma Escrita GATT.

data: Valor a escrever.
response: True para exigir uma resposta de escrita (e lançar GattError em caso de falha). False para escrita-sem-resposta. None (predefinição) seleciona automaticamente com base no que o par anuncia.
timeout_ms: Tempo limite de escrita (apenas relevante se response for True).

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

Assíncrono. Aguarda a próxima notificação nesta característica e devolve o seu payload. Devolve imediatamente se já houver uma notificação em fila.

timeout_ms: Tempo máximo a aguardar. None aguarda indefinidamente.

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

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

timeout_ms: Tempo máximo a aguardar.

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

Assíncrono. Escreve o Descritor de Configuração de Característica do Cliente (CCCD) para subscrever (ou anular a subscrição) de notificações e/ou indicações.

notify: Ativa notificações.
indicate: Ativa 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 for encontrado.

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

class aioble.ClientDescriptor¶

Um descritor GATT remoto descoberto num par. Herda read e write de ClientCharacteristic.

Não construir diretamente.

characteristic¶: A ClientCharacteristic proprietária.

uuid¶: O UUID do descritor.

class aioble.L2CAPChannel¶

Um canal L2CAP orientado à ligação ativo. Devolvido por DeviceConnection.l2cap_accept() ou DeviceConnection.l2cap_connect(). Suporta utilização como gestor de contexto async with que desliga automaticamente ao sair.

Não construir diretamente.

our_mtu¶: Tamanho máximo, em bytes, que o par pode enviar para nós numa única SDU.

peer_mtu¶: Tamanho máximo, em bytes, que podemos enviar para o par numa única SDU.

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

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

Assíncrono. Recebe para buf, devolvendo 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 a aguardar. None aguarda indefinidamente.

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

Assíncrono. Envia buf pelo canal, fragmentando payloads maiores em blocos de tamanho MTU. Aguarda créditos de controlo de fluxo conforme necessário.

buf: Objeto semelhante a bytes a enviar.
timeout_ms: Tempo máximo a aguardar por bloco.
chunk_size: Substituição opcional para o tamanho do bloco 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 paralisado seja drenado pelo controlador.

timeout_ms: Tempo máximo a aguardar.

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

Assíncrono. Desliga o canal e aguarda o IRQ de desligação.

timeout_ms: Tempo máximo a aguardar.

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

Assíncrono. Aguarda até o canal ser desligado por qualquer um dos lados.

timeout_ms: Tempo máximo a aguardar.

aioble — BLE Assíncrono¶

Exemplos¶

Funções ao nível do módulo¶

Constantes ao nível do módulo¶

Exceções¶

Classes¶

`aioble` — BLE Assíncrono¶