11.10. Atuando como central

O outro lado da conversa é o central – o dispositivo que escaneia por periféricos que estão anunciando, escolhe um para conversar, abre uma conexão, percorre o banco de dados GATT remoto e lê ou se inscreve em características nele. Uma câmera que coleta leituras de um sensor vestível, escuta um beacon ou conversa com um microcontrolador companheiro é um central.

O padrão central no aioble passa por quatro estágios: escanear, conectar, descobrir, operar.

11.10.1. Escaneando

aioble.scan() retorna um gerenciador de contexto assíncrono que também funciona como um iterador assíncrono sobre os dispositivos descobertos. O uso típico é escanear até que um dispositivo de interesse apareça e então sair da iteração:

import aioble
import asyncio
import bluetooth

HR_SERVICE = bluetooth.UUID(0x180D)

async def find_heart_rate():
    async with aioble.scan(duration_ms=5000, active=True) as scanner:
        async for result in scanner:
            if HR_SERVICE in result.services():
                return result.device
    return None

duration_ms=5000 limita por quanto tempo o escaneamento é executado; duration_ms=0 escaneia indefinidamente (até que o gerenciador de contexto saia). active=True solicita respostas de escaneamento, o que dobra o tamanho do payload por dispositivo ao custo de uma pequena transmissão adicional de ambos os lados. Os argumentos nomeados restantes interval_us / window_us ajustam o ciclo de trabalho do próprio rádio do scanner e raramente são alterados dos valores padrão.

Cada aioble.ScanResult expõe o endereço do dispositivo, o último RSSI, os bytes brutos de anúncio e de resposta de escaneamento, e helpers que analisam os campos padrão:

  • result.device – um aioble.Device pronto para chamar connect().

  • result.rssi – indicador de intensidade do sinal recebido em dBm, útil para a lógica de “escolher o mais próximo”.

  • result.name() – a string do nome local, ou None se não for anunciado.

  • result.services() – um gerador de bluetooth.UUID para cada serviço que o dispositivo anuncia.

  • result.manufacturer() – um gerador de tuplas (company_id, data) para os campos específicos do fabricante.

  • result.connectable – se o anúncio mais recente era conectável.

O mesmo ScanResult é reemitido conforme novos dados de anúncio chegam para o mesmo dispositivo, de modo que um ouvinte passivo que apenas deseja rastrear dispositivos indefinidamente pode executar o iterador assíncrono para sempre e despachar a cada evento.

11.10.2. Conectando

Uma vez identificado um dispositivo alvo, abrir uma conexão é um único await

async def talk_to(device):
    connection = await device.connect()           # 10 s timeout
    async with connection:
        # ... do GATT work ...
        pass

aioble.Device.connect() recebe timeout_ms (por quanto tempo aguardar até a conexão ser estabelecida; padrão 10 s) e min_conn_interval_us / max_conn_interval_us (a faixa de intervalo de conexão solicitada, de Conexões).

11.10.2.1. Reconectando a um peer conhecido sem escanear

Uma vez que exista um vínculo (bond) com um peer, o endereço já é conhecido e outra rodada de escanear-e-escolher é tempo de rádio desperdiçado. Construa um aioble.Device diretamente com o endereço salvo e pule direto para connect()

import aioble

KITCHEN_CAM = aioble.Device(aioble.ADDR_PUBLIC,
                            "aa:bb:cc:dd:ee:ff")

async def talk_to_kitchen():
    async with await KITCHEN_CAM.connect() as connection:
        # ... GATT work ...
        pass

O primeiro argumento é um de aioble.ADDR_PUBLIC (o endereço de fábrica de um controlador) ou aioble.ADDR_RANDOM (um endereço privado estático gerado ou resolvível); o segundo é um valor bytes de seis bytes ou uma string hexadecimal separada por dois-pontos. Os atributos addr_type e addr de qualquer Device (por exemplo, um obtido anteriormente de um ScanResult) podem ser persistidos e fornecidos de volta aqui.

A aioble.DeviceConnection retornada é aquilo sobre o que o restante do trabalho do central se apoia. async with garante que a conexão seja fechada quando o bloco sair – em caso de sucesso, de cancelamento ou de qualquer exceção, incluindo aioble.DeviceDisconnectedError causada pelo peer se desconectar.

Se o central precisar de um valor de característica maior do que o MTU padrão de 23 bytes permite, este é o lugar para negociá-lo:

await connection.exchange_mtu(512)

(exchange_mtu() retorna o MTU efetivamente negociado, que é o mínimo entre o valor solicitado e o que o peer suporta.)

11.10.3. Descoberta

A descoberta percorre o banco de dados GATT remoto para encontrar os serviços e características pelos seus UUIDs. Há dois sabores: direcionado (você conhece o UUID e quer uma coisa específica) e exaustivo (você quer tudo).

Direcionado – o caso comum:

service = await connection.service(HR_SERVICE)
if service is None:
    return                                        # no such service

char = await service.characteristic(HR_MEASUREMENT)
if char is None:
    return                                        # no such characteristic

aioble.DeviceConnection.service() e aioble.ClientService.characteristic() recebem, cada um, um bluetooth.UUID e retornam o objeto correspondente (ou None). Ambos têm um argumento nomeado timeout_ms por descoberta, cujo padrão é 2 s.

Exaustivo:

async for service in connection.services():
    print("service:", service.uuid)
    async for char in service.characteristics():
        print("  characteristic:", char.uuid, "properties:", hex(char.properties))

É isto que os apps genéricos de exploração de Bluetooth fazem – útil para desenvolvimento, menos para código de produção que sabe quais UUIDs espera.

11.10.3.1. Inspecionando o que uma característica suporta

A descoberta retorna a máscara de bits de propriedades GATT que o peer anunciou para cada característica como properties. Os bits são os definidos pelo GATT – read (0x02), write-without-response (0x04), write (0x08), notify (0x10), indicate (0x20) e afins. Inspecionar a máscara de bits antes de emitir uma operação permite que um cliente genérico se adapte a características cujas capacidades ele não conhece de antemão:

_PROP_READ = const(0x02)
_PROP_NOTIFY = const(0x10)

char = await service.characteristic(STATUS_UUID)
if char.properties & _PROP_NOTIFY:
    await char.subscribe(notify=True)
    value = await char.notified()
elif char.properties & _PROP_READ:
    value = await char.read()
else:
    value = None                                  # nothing the client can do

Código de produção que já conhece o perfil GATT do peer geralmente não precisa disso – os UUIDs foram documentados de antemão. Clientes genéricos / exploratórios (uma página de configurações que percorre um dispositivo desconhecido, um host de plugins) se apoiam nisso.

11.10.4. Operando

Uma vez que o central detenha uma ClientCharacteristic, cada operação GATT é uma única chamada de corrotina:

  • Read. Emita uma leitura GATT e obtenha o valor de volta:

    value = await char.read()
print("value:", value)

    Leituras longas (valores maiores que o MTU) são tratadas de forma transparente.

  • Write. Envie um novo valor para o servidor:

    await char.write(b"\\x01")

    response=True aguarda uma resposta de escrita e lança aioble.GattError se o servidor rejeitar a escrita. response=False é write-without-response: dispara e esquece. response=None (o padrão) escolhe automaticamente com base no que o peer anunciou.

  • Subscribe. Habilite notificações ou indicações escrevendo no CCCD da característica:

    await char.subscribe(notify=True)

    Depois que isto retorna, o central pode aguardar pushes de entrada.

  • Notified / indicated. Aguarde o próximo push do servidor:

    while True:
    data = await char.notified()
    print("push:", data)

    timeout_ms=None (o padrão) aguarda indefinidamente; passe um inteiro em milissegundos para desistir após um tempo.

Juntar os quatro resulta no programa central canônico de “conectar, inscrever, transmitir”:

async def stream_heart_rate():
    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
        else:
            return

    async with await device.connect() as connection:
        service = await connection.service(HR_SERVICE)
        char = await service.characteristic(HR_MEASUREMENT)
        await char.subscribe(notify=True)
        while connection.is_connected():
            data = await char.notified()
            print("hr push:", data)

asyncio.run(stream_heart_rate())

A coisa toda tem cerca de uma dúzia de linhas e cobre o fluxo de “nenhum Bluetooth em execução” até “transmissão de dados ao vivo”. O iterador de escaneamento corresponde ao padrão broadcaster/observer, connect abre a conexão GAP, service / characteristic percorre a árvore GATT, subscribe escreve no CCCD, e notified aguarda os pushes.

11.10.5. Desconexões e reconexão

Qualquer coisa que aconteça com o enlace de rádio aflora na corrotina que estava aguardando por ele. aioble.DeviceDisconnectedError é o sinal de que o peer se foi ou o timeout de supervisão disparou; a exceção encerra qualquer chamada read(), write() ou notified() em andamento, e qualquer bloco async with connection sai de forma limpa.

Um central que deve reconectar em caso de perda envolve o trabalho em seu próprio loop externo:

async def keep_streaming():
    while True:
        try:
            await stream_heart_rate()
        except aioble.DeviceDisconnectedError:
            print("disconnected, retrying...")
            await asyncio.sleep(2)

11.10.5.1. Delimitando uma sequência com timeout()

Quando várias operações GATT em sequência devem todas concluir dentro de um único orçamento – e não cada uma individualmente com seu próprio timeout_ms – use aioble.DeviceConnection.timeout() para envolvê-las. O gerenciador de contexto retornado cancela seu corpo se o orçamento se esgotar (lançando asyncio.TimeoutError) ou se o peer se desconectar (lançando aioble.DeviceDisconnectedError):

async with await device.connect() as connection:
    try:
        with connection.timeout(2000):                    # 2 s for the whole block
            service = await connection.service(HR_SERVICE)
            char = await service.characteristic(HR_MEASUREMENT)
            await char.subscribe(notify=True)
    except asyncio.TimeoutError:
        print("discovery + subscribe took too long")

Esta é a alternativa mais limpa a envolver cada chamada individualmente em asyncio.wait_for() e evita sucessos espúrios em que cada chamada cumpre seu próprio prazo, mas a sequência como um todo o ultrapassa. Passar timeout_ms=None para timeout() desativa o prazo e deixa ativa apenas a guarda de desconexão.