`aioble` — BLE asincrono¶

aioble è un wrapper di alto livello, compatibile con asyncio, attorno al modulo bluetooth. Fornisce coroutine pulite per la scansione, la connessione, l’advertising, i servizi GATT e i canali L2CAP.

Tutte le operazioni remote (connect, disconnect, lettura/scrittura del client, indicate del server, recv/send L2CAP, pair) sono awaitable e supportano i timeout.

Ruoli supportati:

Broadcaster (advertiser) — genera i payload di advertising e di scan-response per i campi più comuni, suddivide automaticamente il payload tra advertising e scan response, esegue l’advertising indefinitamente o per una durata fissa.
Peripheral — attende una connessione da un central, attende lo scambio MTU.
Observer (scanner) — scansione passiva e attiva, combina i payload di advertising e di scan-response dello stesso dispositivo, analizza i campi più comuni dai payload di advertising.
Central — si connette a un peripheral, avvia lo scambio MTU.
GATT Client — individua servizi / caratteristiche / descrittori (opzionalmente per UUID); legge / scrive / scrive-con-risposta su caratteristiche e descrittori; si iscrive a notifiche e indicazioni (tramite il CCCD); attende notifiche e indicazioni.
GATT Server — registra servizi / caratteristiche / descrittori; attende scritture su caratteristiche e descrittori; intercetta le richieste di lettura; invia notifiche e indicazioni (e attende la risposta).
L2CAP — accetta e connette canali L2CAP orientati alla connessione, gestisce il controllo di flusso del canale.
Security — gestione di chiavi/segreti basata su JSON, avvio del pairing, interrogazione dello stato di cifratura / autenticazione.

Esempi¶

Scansiona i dispositivi BLE nelle vicinanze e stampa ciascuno man mano che viene rilevato:

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

Connettiti come central a un peripheral che esegue l’advertising del servizio Heart Rate e iscriviti alle sue notifiche di misurazione:

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

Agisci come peripheral: registra un servizio GATT, eseguine l’advertising e invia notifiche a chiunque si connetta:

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

Funzioni a livello di modulo¶

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

Inoltra a bluetooth.BLE.config(), assicurandosi prima che la radio BLE sia attiva.

args: Nome di un singolo parametro opzionale da interrogare.
kwargs: Argomenti keyword per impostare i valori di configurazione.

aioble.stop() → None¶: Disattiva la radio BLE sottostante ed esegue gli eventuali handler di spegnimento dei sotto-moduli registrati. Dopo questa chiamata, scanner, advertiser, connessioni e canali L2CAP vengono tutti chiusi.

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

Restituisce un context-manager / iteratore asincrono scan che produce istanze di ScanResult per ogni dispositivo univoco rilevato (o per ogni nuovo dato di advertising proveniente da un dispositivo noto).

duration_ms: Durata della scansione, in millisecondi. Passa 0 per eseguire la scansione indefinitamente fino all’uscita dal context manager.
interval_us: Intervallo di scansione in microsecondi. Il valore predefinito è 1.280.000.
window_us: Finestra di scansione in microsecondi (deve essere minore o uguale a interval_us). Il valore predefinito è 11.250.
active: Se True, esegue una scansione attiva (richiede i dati di scan response). Il valore predefinito è 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 asincrona che avvia l’advertising e attende una connessione in arrivo da un central. Restituisce un DeviceConnection che rappresenta il central connesso, oppure solleva asyncio.TimeoutError in caso di timeout.

interval_us: Intervallo di advertising, in microsecondi.
adv_data: Payload di advertising grezzo. Se non impostato, adv_data viene costruito a partire dai restanti argomenti keyword.
resp_data: Payload di scan response grezzo. Popolato automaticamente con i dati eccedenti di adv_data, se necessario.
connectable: Se True, si tratta di un advertisement connettibile.
limited_disc: Usa il flag limited-discoverable invece di quello generale.
br_edr: Imposta il flag BR/EDR-supported.
name: Nome locale completo opzionale da incorporare.
services: Iterabile di bluetooth.UUID di cui eseguire l’advertising.
appearance: Valore appearance a 16 bit (vedi i Bluetooth assigned numbers).
manufacturer: Tupla (company_id, data_bytes) da pubblicizzare come dati specifici del produttore.
timeout_ms: Interrompe l’advertising dopo questo numero di millisecondi senza una connessione. None significa eseguire l’advertising finché non si è connessi.

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

Registra uno o più oggetti Service (e le relative caratteristiche e descrittori) presso il server GATT. Deve essere chiamato una volta prima di avviare advertise. Le chiamate successive sostituiscono la registrazione precedente.

services: Una o più istanze di Service.

Costanti a livello di modulo¶

aioble.ADDR_PUBLIC¶: Tipo di indirizzo del dispositivo BLE pubblico (0).

aioble.ADDR_RANDOM¶: Tipo di indirizzo del dispositivo BLE casuale (1).

Eccezioni¶

exception aioble.GattError¶: Sollevata quando un’operazione GATT remota (read / write / indicate) si completa con uno stato diverso da zero. Il codice di stato è disponibile nell’attributo _status.

exception aioble.DeviceDisconnectedError¶: Sollevata all’interno di un’operazione asincrona (ad esempio read, write, notified) quando la connessione sottostante cade durante l’attesa.

exception aioble.L2CAPDisconnectedError¶: Sollevata quando si tenta un’operazione di send/recv/flush su un canale L2CAP disconnesso (o quando viene interrotta da esso).

exception aioble.L2CAPConnectionError¶: Sollevata da DeviceConnection.l2cap_connect quando la creazione del canale fallisce. Il codice di stato Bluetooth è il primo argomento.

Classi¶

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

Rappresenta un dispositivo BLE remoto tramite indirizzo. Due istanze di Device risultano uguali se corrispondono sia addr_type sia addr. Usata come handle per avviare le connessioni.

addr_type: ADDR_PUBLIC oppure ADDR_RANDOM.
addr: Indirizzo a sei byte come bytes, oppure una stringa esadecimale separata da due punti (ad esempio "aa:bb:cc:dd:ee:ff").

addr_type¶: Il tipo di indirizzo con cui è stato costruito il dispositivo.

addr¶: L’indirizzo grezzo del dispositivo a sei byte.

addr_hex() → str¶: Restituisce l’indirizzo formattato come stringa esadecimale separata da due punti.

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

Asincrona. Avvia una connessione GAP verso questo dispositivo e restituisce il DeviceConnection risultante. Annulla qualsiasi scansione in corso.

timeout_ms: Quanto attendere il completamento della connessione.
scan_duration_ms: Durata della scansione iniziale prima della connessione (specifica del controller).
min_conn_interval_us / max_conn_interval_us: Limiti opzionali dell’intervallo di connessione, in microsecondi.

class aioble.DeviceConnection¶

Una connessione GAP attiva verso un Device. Restituita da Device.connect() o da advertise. Supporta l’uso come context manager async with che si disconnette automaticamente all’uscita.

Non costruire direttamente.

device¶: Il Device sottostante.

encrypted¶: True quando il collegamento è cifrato (ad esempio dopo il pairing).

authenticated¶: True se il collegamento è stato autenticato (pairing protetto da MITM).

bonded¶: True se il pairing ha prodotto chiavi di bonding.

key_size¶: Dimensione della chiave di cifratura negoziata in byte, oppure False se non cifrato.

mtu¶: MTU ATT negoziato dopo exchange_mtu, oppure None finché non è impostato.

is_connected() → bool¶: Restituisce se la connessione è ancora attiva.

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

Asincrona. Disconnette e attende l’IRQ di disconnessione.

timeout_ms: Tempo massimo di attesa per la disconnessione.

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

Asincrona. Attende che la connessione venga terminata da una delle due parti. Se disconnect è True, esegue prima attivamente la disconnessione.

timeout_ms: Tempo massimo di attesa. None significa attendere indefinitamente.
disconnect: Se True, avvia la disconnessione.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Restituisce un context manager che annulla il proprio corpo se trascorre il timeout (sollevando asyncio.TimeoutError) oppure se il dispositivo si disconnette (sollevando DeviceDisconnectedError).

timeout_ms: Timeout in millisecondi, oppure None per nessun timeout.

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

Asincrona. Avvia uno scambio MTU ATT e restituisce l’MTU negoziato.

mtu: MTU preferito opzionale da impostare sull’interfaccia BLE sottostante prima dello scambio.
timeout_ms: Timeout per lo scambio.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Asincrona. Individua un singolo servizio remoto corrispondente a uuid, oppure None se non trovato.

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

Restituisce un iteratore asincrono di oggetti ClientService remoti. Usalo con async for ed esegui il ciclo fino al completamento.

uuid: Filtro UUID opzionale. None restituisce ogni servizio.
timeout_ms: Timeout per ogni discovery.

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

Asincrona. Avvia il pairing su questa connessione. Aggiorna gli attributi encrypted / authenticated / bonded / key_size al completamento.

bond: Mantiene in modo persistente le chiavi di pairing.
le_secure: Usa LE Secure Connections.
mitm: Richiede la protezione man-in-the-middle.
io: Costante della capacità di IO (ad esempio 3 per nessun input/output).
timeout_ms: Timeout del pairing.

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

Asincrona. Si mette in ascolto sul PSM specificato e restituisce un L2CAPChannel non appena il remoto lo apre.

psm: Protocol/Service Multiplexer su cui ascoltare.
mtu: Dimensione massima di ricezione, in byte.
timeout_ms: Tempo massimo di attesa per la connessione del remoto.

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

Asincrona. Apre un canale L2CAP verso il remoto sul PSM specificato.

psm: Protocol/Service Multiplexer a cui connettersi.
mtu: Dimensione massima di ricezione, in byte.
timeout_ms: Timeout di connessione.

class aioble.ScanResult¶

Un singolo dispositivo rilevato durante scan. La stessa istanza viene riprodotta man mano che arrivano nuovi dati di advertising.

Non costruire direttamente.

device¶: Il Device sottostante.

rssi¶: Ultimo RSSI riportato, in dBm.

adv_data¶: Payload di advertising grezzo (bytes oppure None).

resp_data¶: Payload di scan response grezzo (bytes oppure None), se la scansione attiva è abilitata.

connectable¶: True se l’advertisement più recente era connettibile.

name() → str | None¶: Decodifica il nome locale pubblicizzato completo (o abbreviato) dal payload, oppure None se assente.

services() → Iterator[bluetooth.UUID]¶: Generatore che produce ogni bluetooth.UUID pubblicizzato nei campi service-list a 16/32/128 bit.

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

Generatore che produce tuple (company_id, data) dai campi di advertising specifici del produttore.

filter: Se fornito, produce solo le voci il cui company ID corrisponde.

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

Un servizio GATT locale. Costruisci un servizio con una o più istanze di Characteristic, quindi passalo a register_services.

uuid: L’UUID del servizio.

uuid¶: L’UUID del servizio.

characteristics¶: Elenco di oggetti Characteristic associati a questo servizio.

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

Una caratteristica GATT locale. La sua costruzione la aggiunge automaticamente a service.

service: Il Service proprietario.
uuid: L’UUID della caratteristica.
read, write, write_no_response, notify, indicate: Booleani che selezionano le operazioni GATT supportate.
initial: Valore iniziale opzionale (bytes).
capture: Se True, i valori scritti vengono accodati (fino a 10 in profondità) in modo che scritture rapide consecutive non vadano perse. Ogni chiamata a written restituisce quindi una tupla (connection, data).

uuid¶: L’UUID della caratteristica.

flags¶: Bitmask dei flag di proprietà GATT costruita a partire dal costruttore.

descriptors¶: Elenco di oggetti Descriptor associati a questa caratteristica.

read() → bytes¶: Legge il valore corrente dal database GATT locale.

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

Aggiorna il valore nel database GATT locale.

data: Nuovi byte di valore.
send_update: Se True, notifica/indica anche a ogni connessione iscritta.

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

Invia un GATT Notify a connection.

connection: La connessione client di destinazione.
data: Payload da inviare. Se None, viene inviato il valore locale corrente.

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

Asincrona. Invia un GATT Indicate a connection e attende la conferma del client. Solleva GattError in caso di stato diverso da zero.

connection: La connessione client di destinazione.
data: Payload da indicare, oppure None per inviare il valore locale.
timeout_ms: Tempo massimo di attesa per la conferma.

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

Asincrona. Attende una scrittura remota. Restituisce il DeviceConnection che scrive, oppure (connection, data) se la caratteristica è stata creata con capture=True.

timeout_ms: Tempo massimo di attesa. None attende indefinitamente.

on_read(connection: DeviceConnection) → int¶: Hook di override invocato in modo sincrono quando viene ricevuta una lettura remota. Restituisce 0 per consentire la lettura oppure un codice di errore ATT diverso da zero per rifiutarla. L’implementazione predefinita restituisce 0.

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

Una Characteristic il cui buffer GATT di supporto può essere configurato. Utile per ricevere valori più grandi della dimensione predefinita dell’attributo, o per accodare scritture consecutive.

max_len: Dimensione del buffer, in byte.
append: Se True, le scritture sequenziali vengono accodate nel buffer invece di sovrascriverlo.

Gli altri argomenti vengono inoltrati a Characteristic.

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

Un descrittore GATT locale. La sua costruzione lo aggiunge automaticamente a characteristic. Eredita read, write e written da Characteristic.

characteristic: La Characteristic proprietaria.
uuid: L’UUID del descrittore.
read, write: Booleani che selezionano le operazioni GATT supportate.
initial: Valore iniziale opzionale (bytes).

class aioble.ClientService¶

Un servizio GATT remoto individuato su un peer. Restituito da DeviceConnection.service() o iterato da DeviceConnection.services().

Non costruire direttamente.

connection¶: Il DeviceConnection proprietario.

uuid¶: L’UUID del servizio remoto.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Asincrona. Individua una singola caratteristica per UUID, oppure None se non trovata.

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

Restituisce un iteratore asincrono di oggetti ClientCharacteristic. Usalo con async for ed esegui il ciclo fino al completamento.

uuid: Filtro UUID opzionale.
timeout_ms: Timeout per ogni discovery.

class aioble.ClientCharacteristic¶

Una caratteristica GATT remota individuata su un peer. Restituita da ClientService.characteristic() o iterata da ClientService.characteristics().

Non costruire direttamente.

service¶: Il ClientService proprietario.

uuid¶: L’UUID della caratteristica.

properties¶: Bitmask delle operazioni GATT supportate come riportato dal peer.

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

Asincrona. Emette un GATT Read e restituisce il valore. Solleva GattError in caso di stato diverso da zero.

timeout_ms: Timeout di lettura.

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

Asincrona. Emette un GATT Write.

data: Valore da scrivere.
response: True per richiedere una write-response (e sollevare GattError in caso di errore). False per la write-without-response. None (predefinito) seleziona automaticamente in base a ciò che il peer pubblicizza.
timeout_ms: Timeout di scrittura (rilevante solo se response è True).

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

Asincrona. Attende la prossima notifica su questa caratteristica e ne restituisce il payload. Restituisce immediatamente se una notifica è già in coda.

timeout_ms: Tempo massimo di attesa. None attende indefinitamente.

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

Asincrona. Attende la prossima indicazione su questa caratteristica e ne restituisce il payload.

timeout_ms: Tempo massimo di attesa.

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

Asincrona. Scrive il Client Characteristic Configuration Descriptor (CCCD) per iscriversi (o disiscriversi) a notifiche e/o indicazioni.

notify: Abilita le notifiche.
indicate: Abilita le indicazioni.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Asincrona. Individua un singolo descrittore per UUID, oppure None se non trovato.

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: Restituisce un iteratore asincrono di oggetti ClientDescriptor. Usalo con async for ed esegui il ciclo fino al completamento.

class aioble.ClientDescriptor¶

Un descrittore GATT remoto individuato su un peer. Eredita read e write da ClientCharacteristic.

Non costruire direttamente.

characteristic¶: La ClientCharacteristic proprietaria.

uuid¶: L’UUID del descrittore.

class aioble.L2CAPChannel¶

Un canale L2CAP attivo orientato alla connessione. Restituito da DeviceConnection.l2cap_accept() o DeviceConnection.l2cap_connect(). Supporta l’uso come context manager async with che si disconnette automaticamente all’uscita.

Non costruire direttamente.

our_mtu¶: Dimensione massima, in byte, che il peer può inviarci in una singola SDU.

peer_mtu¶: Dimensione massima, in byte, che possiamo inviare al peer in una singola SDU.

available() → bool¶: Restituisce in modo sincrono True se i dati di ricezione bufferizzati sono pronti (cioè recvinto non si bloccherà).

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

Asincrona. Riceve in buf, restituendo il numero di byte letti. Attende nuovi dati se il canale è vuoto.

buf: Buffer pre-allocato da riempire.
timeout_ms: Tempo massimo di attesa. None attende indefinitamente.

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

Asincrona. Invia buf sul canale, frammentando i payload più grandi in blocchi di dimensione MTU. Attende i credit di controllo di flusso secondo necessità.

buf: Oggetto bytes-like da inviare.
timeout_ms: Tempo massimo di attesa per ogni blocco.
chunk_size: Override opzionale per la dimensione del blocco per ogni chiamata. Limitato a min(our_mtu * 2, peer_mtu).

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

Asincrona. Attende finché un eventuale send in stallo non è stato svuotato dal controller.

timeout_ms: Tempo massimo di attesa.

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

Asincrona. Disconnette il canale e attende l’IRQ di disconnessione.

timeout_ms: Tempo massimo di attesa.

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

Asincrona. Attende finché il canale non viene disconnesso da una delle due parti.

timeout_ms: Tempo massimo di attesa.

aioble — BLE asincrono¶

Esempi¶

Funzioni a livello di modulo¶

Costanti a livello di modulo¶

Eccezioni¶

Classi¶

`aioble` — BLE asincrono¶