`aioble` — BLE asynchrone¶

aioble est un encapsuleur de haut niveau compatible asyncio autour du module bluetooth. Il fournit des coroutines claires pour la recherche, la connexion, la diffusion (advertising), les services GATT et les canaux L2CAP.

Toutes les opérations distantes (connexion, déconnexion, lecture/écriture client, indication serveur, recv/send l2cap, appairage) sont awaitables et prennent en charge les délais d’expiration.

Rôles pris en charge :

Broadcaster (diffuseur) — génère les charges utiles de diffusion (advertising) et de réponse de scan pour les champs courants, répartit automatiquement la charge utile entre la diffusion et la réponse de scan, diffuse indéfiniment ou pendant une durée fixe.
Peripheral — attend une connexion provenant d’un central, attend l’échange de MTU.
Observer (scanner) — recherche passive et active, combine les charges utiles de diffusion et de réponse de scan pour le même appareil, analyse les champs courants des charges utiles de diffusion.
Central — se connecte à un périphérique, initie l’échange de MTU.
Client GATT — découvre les services / caractéristiques / descripteurs (éventuellement par UUID) ; lecture / écriture / écriture-avec-réponse sur les caractéristiques et les descripteurs ; s’abonne aux notifications et aux indications (via le CCCD) ; attend les notifications et les indications.
Serveur GATT — enregistre les services / caractéristiques / descripteurs ; attend les écritures sur les caractéristiques et les descripteurs ; intercepte les requêtes de lecture ; envoie les notifications et les indications (et attend la réponse).
L2CAP — accepte et établit des canaux L2CAP orientés connexion, gère le contrôle de flux des canaux.
Sécurité — gestion des clés/secrets appuyée sur du JSON, initiation de l’appairage, interrogation de l’état de chiffrement / d’authentification.

Exemples¶

Recherche les appareils BLE à proximité et affiche chacun d’eux dès qu’il est détecté

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

Se connecter en tant que central à un périphérique diffusant le service Heart Rate et s’abonner à ses notifications de mesure

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

Agir comme un peripheral : enregistrer un service GATT, le diffuser et envoyer des notifications à quiconque se connecte

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

Fonctions au niveau du module¶

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

Transmet vers bluetooth.BLE.config(), en s’assurant d’abord que la radio BLE est active.

args: Nom de paramètre unique optionnel à interroger.
kwargs: Arguments nommés pour définir les valeurs de configuration.

aioble.stop() → None¶: Désactive la radio BLE sous-jacente et exécute tous les gestionnaires d’arrêt de sous-module enregistrés. Après cet appel, les scanners, les diffuseurs, les connexions et les canaux L2CAP sont tous démantelés.

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

Renvoie un gestionnaire de contexte asynchrone / itérateur asynchrone scan qui produit des instances ScanResult pour chaque appareil unique découvert (ou pour chaque nouvelle donnée de diffusion provenant d’un appareil connu).

duration_ms: Durée de la recherche, en millisecondes. Passez 0 pour rechercher indéfiniment jusqu’à la sortie du gestionnaire de contexte.
interval_us: Intervalle de recherche en microsecondes. Par défaut 1 280 000.
window_us: Fenêtre de recherche en microsecondes (doit être inférieure ou égale à interval_us). Par défaut 11 250.
active: Si True, effectue une recherche active (demande les données de réponse de scan). Par défaut 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 asynchrone qui commence à diffuser et attend une connexion entrante d’un central. Renvoie un DeviceConnection représentant le central connecté, ou lève asyncio.TimeoutError en cas de délai dépassé.

interval_us: Intervalle de diffusion, en microsecondes.
adv_data: Charge utile de diffusion brute. Si non définie, adv_data est construite à partir des arguments nommés restants.
resp_data: Charge utile de réponse de scan brute. Renseignée automatiquement pour absorber le débordement d”adv_data si nécessaire.
connectable: Si True, il s’agit d’une diffusion connectable.
limited_disc: Utiliser l’indicateur de découverte limitée au lieu de générale.
br_edr: Définit l’indicateur de prise en charge BR/EDR.
name: Nom local complet optionnel à intégrer.
services: Itérable de bluetooth.UUID à diffuser.
appearance: Valeur d’apparence sur 16 bits (voir les numéros assignés Bluetooth).
manufacturer: Tuple (company_id, data_bytes) à diffuser comme données spécifiques au fabricant.
timeout_ms: Arrête la diffusion après ce nombre de millisecondes sans connexion. None signifie diffuser jusqu’à connexion.

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

Enregistre un ou plusieurs objets Service (ainsi que leurs caractéristiques et descripteurs) auprès du serveur GATT. Doit être appelé une fois avant de démarrer advertise. Les appels suivants remplacent l’enregistrement précédent.

services: Une ou plusieurs instances Service.

Constantes au niveau du module¶

aioble.ADDR_PUBLIC¶: Type d’adresse d’appareil BLE publique (0).

aioble.ADDR_RANDOM¶: Type d’adresse d’appareil BLE aléatoire (1).

Exceptions¶

exception aioble.GattError¶: Levée lorsqu’une opération GATT distante (lecture / écriture / indication) se termine avec un statut non nul. Le code de statut est disponible dans l’attribut _status.

exception aioble.DeviceDisconnectedError¶: Levée à l’intérieur d’une opération asynchrone (par ex. read, write, notified) lorsque la connexion sous-jacente tombe pendant l’attente.

exception aioble.L2CAPDisconnectedError¶: Levée lorsqu’une opération send/recv/flush sur un canal L2CAP est tentée sur (ou interrompue par) un canal déconnecté.

exception aioble.L2CAPConnectionError¶: Levée par DeviceConnection.l2cap_connect lorsque l’établissement du canal échoue. Le code de statut Bluetooth est le premier argument.

Classes¶

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

Représente un appareil BLE distant par son adresse. Deux instances Device sont considérées égales si addr_type et addr correspondent toutes deux. Utilisé comme identifiant pour initier les connexions.

addr_type: Soit ADDR_PUBLIC, soit ADDR_RANDOM.
addr: Adresse de six octets sous forme de bytes, ou chaîne hexadécimale séparée par des deux-points (par ex. "aa:bb:cc:dd:ee:ff").

addr_type¶: Le type d’adresse avec lequel l’appareil a été construit.

addr¶: L’adresse brute de six octets de l’appareil.

addr_hex() → str¶: Renvoie l’adresse formatée sous forme de chaîne hexadécimale séparée par des deux-points.

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

Asynchrone. Initie une connexion GAP vers cet appareil et renvoie le DeviceConnection résultant. Annule toute recherche en cours.

timeout_ms: Durée d’attente pour que la connexion s’établisse.
scan_duration_ms: Durée de recherche initiale avant la connexion (spécifique au contrôleur).
min_conn_interval_us / max_conn_interval_us: Bornes optionnelles de l’intervalle de connexion, en microsecondes.

class aioble.DeviceConnection¶

Une connexion GAP active vers un Device. Renvoyée par Device.connect() ou advertise. Peut être utilisée comme gestionnaire de contexte async with qui se déconnecte automatiquement à la sortie.

Ne construisez pas directement.

device¶: Le Device sous-jacent.

encrypted¶: True une fois la liaison chiffrée (par ex. après l’appairage).

authenticated¶: True si la liaison a été authentifiée (appairage protégé contre les attaques MITM).

bonded¶: True si l’appairage a produit des clés de liaison (bonding).

key_size¶: Taille de la clé de chiffrement négociée en octets, ou False si non chiffrée.

mtu¶: MTU ATT négocié après exchange_mtu, ou None jusqu’à ce qu’il soit défini.

is_connected() → bool¶: Indique si la connexion est encore active.

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

Asynchrone. Déconnecte et attend l’IRQ de déconnexion.

timeout_ms: Temps maximal d’attente de la déconnexion.

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

Asynchrone. Attend que la connexion soit terminée par l’une ou l’autre des parties. Si disconnect vaut True, déconnecte activement au préalable.

timeout_ms: Temps maximal d’attente. None signifie attendre indéfiniment.
disconnect: Si True, initie la déconnexion.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Renvoie un gestionnaire de contexte qui annule son corps si le délai expire (en levant asyncio.TimeoutError) ou si l’appareil se déconnecte (en levant DeviceDisconnectedError).

timeout_ms: Délai d’expiration en millisecondes, ou None pour aucun délai.

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

Asynchrone. Initie un échange de MTU ATT et renvoie le MTU négocié.

mtu: MTU préféré optionnel à définir sur l’interface BLE sous-jacente avant l’échange.
timeout_ms: Délai d’expiration pour l’échange.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Asynchrone. Découvre un unique service distant correspondant à uuid, ou None si introuvable.

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

Renvoie un itérateur asynchrone d’objets ClientService distants. À utiliser avec async for en exécutant la boucle jusqu’à son terme.

uuid: Filtre UUID optionnel. None renvoie tous les services.
timeout_ms: Délai d’expiration par découverte.

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

Asynchrone. Initie l’appairage sur cette connexion. Met à jour les attributs encrypted / authenticated / bonded / key_size une fois terminé.

bond: Conserver les clés d’appairage.
le_secure: Utiliser LE Secure Connections.
mitm: Exiger une protection contre les attaques de l’homme du milieu.
io: Constante de capacité d’E/S (par ex. 3 pour aucune entrée/sortie).
timeout_ms: Délai d’expiration de l’appairage.

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

Asynchrone. Écoute sur le PSM donné et renvoie un L2CAPChannel dès que le distant l’ouvre.

psm: Multiplexeur de protocole/service sur lequel écouter.
mtu: Taille de réception maximale, en octets.
timeout_ms: Temps maximal d’attente de la connexion du distant.

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

Asynchrone. Ouvre un canal L2CAP vers le distant sur le PSM donné.

psm: Multiplexeur de protocole/service auquel se connecter.
mtu: Taille de réception maximale, en octets.
timeout_ms: Délai d’expiration de la connexion.

class aioble.ScanResult¶

Un appareil unique découvert pendant scan. La même instance est de nouveau produite à mesure que de nouvelles données de diffusion arrivent.

Ne construisez pas directement.

device¶: Le Device sous-jacent.

rssi¶: Dernier RSSI rapporté, en dBm.

adv_data¶: Charge utile de diffusion brute (bytes ou None).

resp_data¶: Charge utile de réponse de scan brute (bytes ou None), si la recherche active est activée.

connectable¶: True si la diffusion la plus récente était connectable.

name() → str | None¶: Décode le nom local diffusé complet (ou abrégé) depuis la charge utile, ou None s’il est absent.

services() → Iterator[bluetooth.UUID]¶: Générateur produisant chaque bluetooth.UUID diffusé dans les champs de liste de services 16/32/128 bits.

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

Générateur produisant des tuples (company_id, data) à partir des champs de diffusion spécifiques au fabricant.

filter: Si fourni, ne produit que les entrées dont l’identifiant de société correspond.

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

Un service GATT local. Construisez un service avec une ou plusieurs instances Characteristic, puis passez-le à register_services.

uuid: L’UUID du service.

uuid¶: L’UUID du service.

characteristics¶: Liste des objets Characteristic liés à ce service.

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

Une caractéristique GATT locale. En construire une l’ajoute automatiquement à service.

service: Le Service propriétaire.
uuid: L’UUID de la caractéristique.
read, write, write_no_response, notify, indicate: Booléens sélectionnant les opérations GATT prises en charge.
initial: Valeur initiale optionnelle (bytes).
capture: Si True, les valeurs écrites sont mises en file (jusqu’à 10 de profondeur) afin que les écritures rapides successives ne soient pas perdues. Chaque appel à written renvoie alors un tuple (connection, data).

uuid¶: L’UUID de la caractéristique.

flags¶: Masque de bits des indicateurs de propriété GATT construit à partir du constructeur.

descriptors¶: Liste des objets Descriptor liés à cette caractéristique.

read() → bytes¶: Lit la valeur courante depuis la base de données GATT locale.

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

Met à jour la valeur dans la base de données GATT locale.

data: Octets de la nouvelle valeur.
send_update: Si True, notifie/indique également chaque connexion abonnée.

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

Envoie un GATT Notify vers connection.

connection: La connexion client cible.
data: Charge utile à envoyer. Si None, la valeur locale courante est envoyée.

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

Asynchrone. Envoie un GATT Indicate vers connection et attend la confirmation du client. Lève GattError en cas de statut non nul.

connection: La connexion client cible.
data: Charge utile à indiquer, ou None pour envoyer la valeur locale.
timeout_ms: Temps maximal d’attente de la confirmation.

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

Asynchrone. Attend une écriture distante. Renvoie le DeviceConnection ayant écrit, ou (connection, data) si la caractéristique a été créée avec capture=True.

timeout_ms: Temps maximal d’attente. None attend indéfiniment.

on_read(connection: DeviceConnection) → int¶: Point d’extension (hook) à surcharger, invoqué de manière synchrone à la réception d’une lecture distante. Renvoyez 0 pour autoriser la lecture ou un code d’erreur ATT non nul pour la rejeter. L’implémentation par défaut renvoie 0.

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

Une Characteristic dont le tampon GATT sous-jacent peut être configuré. Utile pour recevoir des valeurs plus grandes que la taille d’attribut par défaut, ou pour mettre en file des écritures successives.

max_len: Taille du tampon, en octets.
append: Si True, les écritures séquentielles s’ajoutent dans le tampon au lieu de l’écraser.

Les autres arguments sont transmis à Characteristic.

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

Un descripteur GATT local. En construire un l’ajoute automatiquement à characteristic. Hérite de read, write et written de Characteristic.

characteristic: La Characteristic propriétaire.
uuid: L’UUID du descripteur.
read, write: Booléens sélectionnant les opérations GATT prises en charge.
initial: Valeur initiale optionnelle (bytes).

class aioble.ClientService¶

Un service GATT distant découvert sur un pair. Renvoyé par DeviceConnection.service() ou parcouru depuis DeviceConnection.services().

Ne construisez pas directement.

connection¶: Le DeviceConnection propriétaire.

uuid¶: L’UUID du service distant.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Asynchrone. Découvre une unique caractéristique par UUID, ou None si introuvable.

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

Renvoie un itérateur asynchrone d’objets ClientCharacteristic. À utiliser avec async for en exécutant la boucle jusqu’à son terme.

uuid: Filtre UUID optionnel.
timeout_ms: Délai d’expiration par découverte.

class aioble.ClientCharacteristic¶

Une caractéristique GATT distante découverte sur un pair. Renvoyée par ClientService.characteristic() ou parcourue depuis ClientService.characteristics().

Ne construisez pas directement.

service¶: Le ClientService propriétaire.

uuid¶: L’UUID de la caractéristique.

properties¶: Masque de bits des opérations GATT prises en charge tel que rapporté par le pair.

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

Asynchrone. Émet un GATT Read et renvoie la valeur. Lève GattError en cas de statut non nul.

timeout_ms: Délai d’expiration de lecture.

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

Asynchrone. Émet un GATT Write.

data: Valeur à écrire.
response: True pour exiger une réponse d’écriture (et lever GattError en cas d’échec). False pour une écriture-sans-réponse. None (par défaut) sélectionne automatiquement selon ce que le pair annonce.
timeout_ms: Délai d’expiration d’écriture (pertinent uniquement si response vaut True).

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

Asynchrone. Attend la prochaine notification sur cette caractéristique et renvoie sa charge utile. Renvoie immédiatement si une notification est déjà en file.

timeout_ms: Temps maximal d’attente. None attend indéfiniment.

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

Asynchrone. Attend la prochaine indication sur cette caractéristique et renvoie sa charge utile.

timeout_ms: Temps maximal d’attente.

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

Asynchrone. Écrit le Client Characteristic Configuration Descriptor (CCCD) pour s’abonner (ou se désabonner) aux notifications et/ou aux indications.

notify: Activer les notifications.
indicate: Activer les indications.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Asynchrone. Découvre un unique descripteur par UUID, ou None si introuvable.

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: Renvoie un itérateur asynchrone d’objets ClientDescriptor. À utiliser avec async for en exécutant la boucle jusqu’à son terme.

class aioble.ClientDescriptor¶

Un descripteur GATT distant découvert sur un pair. Hérite de read et write de ClientCharacteristic.

Ne construisez pas directement.

characteristic¶: La ClientCharacteristic propriétaire.

uuid¶: L’UUID du descripteur.

class aioble.L2CAPChannel¶

Un canal L2CAP orienté connexion actif. Renvoyé par DeviceConnection.l2cap_accept() ou DeviceConnection.l2cap_connect(). Peut être utilisé comme gestionnaire de contexte async with qui se déconnecte automatiquement à la sortie.

Ne construisez pas directement.

our_mtu¶: Taille maximale, en octets, que le pair peut nous envoyer dans une seule SDU.

peer_mtu¶: Taille maximale, en octets, que nous pouvons envoyer au pair dans une seule SDU.

available() → bool¶: Renvoie de manière synchrone True si des données de réception sont prêtes en tampon (c.-à-d. que recvinto ne bloquera pas).

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

Asynchrone. Reçoit dans buf et renvoie le nombre d’octets lus. Attend de nouvelles données si le canal est vide.

buf: Tampon préalloué à remplir.
timeout_ms: Temps maximal d’attente. None attend indéfiniment.

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

Asynchrone. Envoie buf sur le canal, en fragmentant les charges utiles plus grandes en morceaux de la taille du MTU. Attend les crédits de contrôle de flux selon les besoins.

buf: Objet de type bytes à envoyer.
timeout_ms: Temps maximal d’attente par morceau.
chunk_size: Surcharge optionnelle de la taille de morceau par appel. Plafonnée à min(our_mtu * 2, peer_mtu).

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

Asynchrone. Attend que tout send bloqué ait été vidé par le contrôleur.

timeout_ms: Temps maximal d’attente.

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

Asynchrone. Déconnecte le canal et attend l’IRQ de déconnexion.

timeout_ms: Temps maximal d’attente.

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

Asynchrone. Attend que le canal soit déconnecté par l’une ou l’autre des parties.

timeout_ms: Temps maximal d’attente.

aioble — BLE asynchrone¶

Exemples¶

Fonctions au niveau du module¶

Constantes au niveau du module¶

Exceptions¶

Classes¶

`aioble` — BLE asynchrone¶