`aioble` — Async BLE#

aioble is een asyncio-vriendelijke wrapper op hoog niveau rond de bluetooth-module. Het biedt overzichtelijke coroutines voor scannen, verbinden, adverteren, GATT-services en L2CAP-kanalen.

Alle bewerkingen op afstand (connect, disconnect, client read/write, server indicate, l2cap recv/send, pair) zijn awaitable en ondersteunen time-outs.

Ondersteunde rollen:

Broadcaster (advertiser) — genereert advertising- en scan-response-payloads voor veelvoorkomende velden, splitst de payload automatisch over advertising en scan response, adverteert onbeperkt of gedurende een vaste tijd.
Peripheral — wacht op een verbinding van een central, wacht op MTU-uitwisseling.
Observer (scanner) — passief en actief scannen, advertising- en scan-response-payloads voor hetzelfde apparaat combineren, veelvoorkomende velden uit advertising-payloads parsen.
Central — verbinden met een peripheral, MTU-uitwisseling starten.
GATT Client — services / characteristics / descriptors ontdekken (optioneel op UUID); read / write / write-with-response op characteristics en descriptors; abonneren op notifications en indications (via de CCCD); wachten op notifications en indications.
GATT Server — services / characteristics / descriptors registreren; wachten op writes op characteristics en descriptors; read-requests onderscheppen; notifications en indications versturen (en op antwoord wachten).
L2CAP — L2CAP connection-oriented kanalen accepteren en verbinden, flow control van het kanaal beheren.
Security — JSON-gebaseerd beheer van sleutels/geheimen, pairing starten, status van encryptie / authenticatie opvragen.

Voorbeelden#

Scan naar nabije BLE-apparaten en print elk apparaat zodra het wordt gezien:

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

Verbind als central met een peripheral die de Heart Rate-service adverteert en abonneer op de bijbehorende meet-notifications:

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

Treed op als peripheral: registreer een GATT-service, adverteer deze en stuur notifications naar iedereen die verbinding maakt:

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

Functies op moduleniveau#

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

Stuurt door naar bluetooth.BLE.config() en zorgt ervoor dat de BLE-radio eerst actief is.

args: Optionele naam van een enkele parameter om op te vragen.
kwargs: Keyword-argumenten om configuratiewaarden in te stellen.

aioble.stop() → None#: Deactiveer de onderliggende BLE-radio en voer eventuele geregistreerde shutdown-handlers van submodules uit. Na het aanroepen hiervan worden scanners, advertisers, verbindingen en L2CAP-kanalen allemaal afgebroken.

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

Retourneert een scan async context-manager / async-iterator die ScanResult-instanties oplevert voor elk uniek ontdekt apparaat (of voor elk nieuw stuk advertising-data van een bekend apparaat).

duration_ms: Hoe lang er gescand moet worden, in milliseconden. Geef 0 op om onbeperkt te scannen totdat de context manager wordt afgesloten.
interval_us: Scaninterval in microseconden. Standaard 1.280.000.
window_us: Scanvenster in microseconden (moet kleiner dan of gelijk aan interval_us zijn). Standaard 11.250.
active: Als True, voer een actieve scan uit (vraag scan-response-data aan). Standaard 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#

Async coroutine die begint met adverteren en wacht op een inkomende central-verbinding. Retourneert een DeviceConnection die de verbonden central voorstelt, of werpt asyncio.TimeoutError bij een time-out.

interval_us: Advertising-interval, in microseconden.
adv_data: Ruwe advertising-payload. Indien niet ingesteld, wordt adv_data opgebouwd uit de overige keyword-argumenten.
resp_data: Ruwe scan-response-payload. Wordt automatisch gevuld om overloop uit adv_data op te vangen indien nodig.
connectable: Als True, is dit een verbindbare advertentie.
limited_disc: Gebruik de limited-discoverable-vlag in plaats van general.
br_edr: Stel de BR/EDR-supported-vlag in.
name: Optionele volledige lokale naam om in te bedden.
services: Iterable van bluetooth.UUID om te adverteren.
appearance: 16-bits appearance-waarde (zie de Bluetooth assigned numbers).
manufacturer: Tuple van (company_id, data_bytes) om te adverteren als manufacturer-specifieke data.
timeout_ms: Stop met adverteren na zoveel milliseconden zonder verbinding. None betekent adverteren totdat er verbinding is.

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

Registreer een of meer Service-objecten (en hun characteristics en descriptors) bij de GATT-server. Moet één keer worden aangeroepen voordat advertise wordt gestart. Volgende aanroepen vervangen de vorige registratie.

services: Een of meer Service-instanties.

Constanten op moduleniveau#

aioble.ADDR_PUBLIC#: Publiek BLE-apparaatadrestype (0).

aioble.ADDR_RANDOM#: Willekeurig BLE-apparaatadrestype (1).

Uitzonderingen#

exception aioble.GattError#: Wordt geworpen wanneer een externe GATT-bewerking (read / write / indicate) wordt voltooid met een status ongelijk aan nul. De statuscode is beschikbaar via het _status-attribuut.

exception aioble.DeviceDisconnectedError#: Wordt geworpen binnen een async-bewerking (bijv. read, write, notified) wanneer de onderliggende verbinding wegvalt tijdens het wachten.

exception aioble.L2CAPDisconnectedError#: Wordt geworpen wanneer een send/recv/flush-bewerking op een L2CAP-kanaal wordt geprobeerd op (of onderbroken door) een verbroken kanaal.

exception aioble.L2CAPConnectionError#: Wordt geworpen door DeviceConnection.l2cap_connect wanneer het tot stand brengen van het kanaal mislukt. De Bluetooth-statuscode is het eerste argument.

Klassen#

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

Stelt een extern BLE-apparaat voor op basis van adres. Twee Device-instanties zijn gelijk als zowel addr_type als addr overeenkomen. Wordt gebruikt als handle voor het starten van verbindingen.

addr_type: Ofwel ADDR_PUBLIC of ADDR_RANDOM.
addr: Adres van zes bytes als bytes, of een door dubbele punten gescheiden hex-string (bijv. "aa:bb:cc:dd:ee:ff").

addr_type#: Het adrestype waarmee het apparaat is geconstrueerd.

addr#: Het ruwe apparaatadres van zes bytes.

addr_hex() → str#: Retourneert het adres opgemaakt als een door dubbele punten gescheiden hex-string.

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

Async. Start een GAP-verbinding met dit apparaat en retourneert de resulterende DeviceConnection. Annuleert een eventueel lopende scan.

timeout_ms: Hoe lang er gewacht moet worden tot de verbinding voltooid is.
scan_duration_ms: Initiële scanduur voordat er verbinding wordt gemaakt (controllerspecifiek).
min_conn_interval_us / max_conn_interval_us: Optionele grenzen voor het verbindingsinterval, in microseconden.

class aioble.DeviceConnection#

Een actieve GAP-verbinding met een Device. Geretourneerd door Device.connect() of advertise. Ondersteunt gebruik als async with context manager die automatisch ontkoppelt bij het afsluiten.

Niet rechtstreeks construeren.

device#: De onderliggende Device.

encrypted#: True zodra de link versleuteld is (bijv. na pairing).

authenticated#: True als de link geauthenticeerd was (MITM-beschermde pair).

bonded#: True als pairing bonding-sleutels heeft opgeleverd.

key_size#: Onderhandelde encryptiesleutelgrootte in bytes, of False indien niet versleuteld.

mtu#: Onderhandelde ATT-MTU na exchange_mtu, of None totdat deze is ingesteld.

is_connected() → bool#: Retourneert of de verbinding nog actief is.

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

Async. Verbreek de verbinding en wacht op de disconnection-IRQ.

timeout_ms: Maximale tijd om te wachten op de verbreking.

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

Async. Wacht tot de verbinding door een van beide partijen wordt beëindigd. Als disconnect True is, wordt de verbinding eerst actief verbroken.

timeout_ms: Maximale wachttijd. None betekent oneindig wachten.
disconnect: Als True, initieer het verbreken van de verbinding.

timeout(timeout_ms: int | None) → DeviceTimeout#

Retourneert een context manager die zijn body annuleert als ofwel de time-out verstrijkt (waarbij asyncio.TimeoutError wordt geworpen) of het apparaat de verbinding verbreekt (waarbij DeviceDisconnectedError wordt geworpen).

timeout_ms: Time-out in milliseconden, of None voor geen time-out.

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

Async. Start een ATT-MTU-uitwisseling en retourneert de onderhandelde MTU.

mtu: Optionele voorkeurs-MTU om in te stellen op de onderliggende BLE-interface vóór de uitwisseling.
timeout_ms: Time-out voor de uitwisseling.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]#: Async. Ontdek een enkele externe service die overeenkomt met uuid, of None indien niet gevonden.

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

Retourneert een async iterator van externe ClientService-objecten. Gebruik met async for en laat de lus tot het einde doorlopen.

uuid: Optioneel UUID-filter. None retourneert elke service.
timeout_ms: Time-out per ontdekking.

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

Async. Start pairing op deze verbinding. Werkt de attributen encrypted / authenticated / bonded / key_size bij wanneer voltooid.

bond: Bewaar pairing-sleutels.
le_secure: Gebruik LE Secure Connections.
mitm: Vereis man-in-the-middle-bescherming.
io: IO-capability-constante (bijv. 3 voor geen input/output).
timeout_ms: Pairing-time-out.

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

Async. Luister op de opgegeven PSM en retourneert een L2CAPChannel zodra de remote deze opent.

psm: Protocol/Service Multiplexer om op te luisteren.
mtu: Maximale ontvangstgrootte, in bytes.
timeout_ms: Maximale tijd om te wachten tot de remote verbinding maakt.

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

Async. Open een L2CAP-kanaal naar de remote op de opgegeven PSM.

psm: Protocol/Service Multiplexer om mee te verbinden.
mtu: Maximale ontvangstgrootte, in bytes.
timeout_ms: Verbindings-time-out.

class aioble.ScanResult#

Een enkel apparaat dat tijdens scan is ontdekt. Dezelfde instantie wordt opnieuw opgeleverd wanneer er nieuwe advertising-data binnenkomt.

Niet rechtstreeks construeren.

device#: De onderliggende Device.

rssi#: Laatst gerapporteerde RSSI, in dBm.

adv_data#: Ruwe advertising-payload (bytes of None).

resp_data#: Ruwe scan-response-payload (bytes of None), als actief scannen is ingeschakeld.

connectable#: True als de meest recente advertentie verbindbaar was.

name() → str | None#: Decodeer de volledige (of verkorte) geadverteerde lokale naam uit de payload, of None indien niet aanwezig.

services() → Iterator[bluetooth.UUID]#: Generator die elke bluetooth.UUID oplevert die geadverteerd wordt in de 16/32/128-bits service-list-velden.

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

Generator die (company_id, data)-tuples oplevert uit manufacturer-specifieke advertising-velden.

filter: Indien opgegeven, lever alleen entries op waarvan de company-ID overeenkomt.

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

Een lokale GATT-service. Bouw een service met een of meer Characteristic-instanties en geef deze vervolgens door aan register_services.

uuid: De service-UUID.

uuid#: De service-UUID.

characteristics#: Lijst van Characteristic-objecten die aan deze service zijn gekoppeld.

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

Een lokale GATT-characteristic. Bij het construeren ervan wordt deze automatisch aan service toegevoegd.

service: De bezittende Service.
uuid: De characteristic-UUID.
read, write, write_no_response, notify, indicate: Booleans die de ondersteunde GATT-bewerkingen selecteren.
initial: Optionele initiële waarde (bytes).
capture: Als True worden geschreven waarden in een wachtrij geplaatst (tot 10 diep) zodat snel opeenvolgende writes niet verloren gaan. Elke written-aanroep retourneert dan een (connection, data)-tuple.

uuid#: De characteristic-UUID.

flags#: Bitmasker van de GATT-property-vlaggen, opgebouwd vanuit de constructor.

descriptors#: Lijst van Descriptor-objecten die aan deze characteristic zijn gekoppeld.

read() → bytes#: Lees de huidige waarde uit de lokale GATT-database.

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

Werk de waarde in de lokale GATT-database bij.

data: Nieuwe waardebytes.
send_update: Als True, stuur ook een notify/indicate naar elke geabonneerde verbinding.

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

Stuur een GATT Notify naar connection.

connection: De doel-clientverbinding.
data: Payload om te versturen. Als None, wordt de huidige lokale waarde verstuurd.

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

Async. Stuur een GATT Indicate naar connection en wacht op de bevestiging van de client. Werpt GattError bij een status ongelijk aan nul.

connection: De doel-clientverbinding.
data: Payload om te indicaten, of None om de lokale waarde te versturen.
timeout_ms: Maximale tijd om te wachten op bevestiging.

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

Async. Wacht op een externe write. Retourneert de schrijvende DeviceConnection, of (connection, data) als de characteristic met capture=True is aangemaakt.

timeout_ms: Maximale wachttijd. None wacht oneindig.

on_read(connection: DeviceConnection) → int#: Override-hook die synchroon wordt aangeroepen wanneer een externe read wordt ontvangen. Retourneer 0 om de read toe te staan of een ATT-foutcode ongelijk aan nul om deze te weigeren. De standaardimplementatie retourneert 0.

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

Een Characteristic waarvan de onderliggende GATT-buffer kan worden geconfigureerd. Nuttig voor het ontvangen van waarden die groter zijn dan de standaard attribuutgrootte, of voor het in de wachtrij plaatsen van opeenvolgende writes.

max_len: Buffergrootte, in bytes.
append: Als True worden opeenvolgende writes aan de buffer toegevoegd in plaats van overschreven.

Overige argumenten worden doorgegeven aan Characteristic.

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

Een lokale GATT-descriptor. Bij het construeren ervan wordt deze automatisch aan characteristic toegevoegd. Erft read, write en written van Characteristic.

characteristic: De bezittende Characteristic.
uuid: De descriptor-UUID.
read, write: Booleans die de ondersteunde GATT-bewerkingen selecteren.
initial: Optionele initiële waarde (bytes).

class aioble.ClientService#

Een externe GATT-service die op een peer is ontdekt. Geretourneerd door DeviceConnection.service() of geïtereerd vanuit DeviceConnection.services().

Niet rechtstreeks construeren.

connection#: De bezittende DeviceConnection.

uuid#: De externe service-UUID.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]#: Async. Ontdek een enkele characteristic op UUID, of None indien niet gevonden.

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

Retourneert een async iterator van ClientCharacteristic-objecten. Gebruik met async for en laat de lus tot het einde doorlopen.

uuid: Optioneel UUID-filter.
timeout_ms: Time-out per ontdekking.

class aioble.ClientCharacteristic#

Een externe GATT-characteristic die op een peer is ontdekt. Geretourneerd door ClientService.characteristic() of geïtereerd vanuit ClientService.characteristics().

Niet rechtstreeks construeren.

service#: De bezittende ClientService.

uuid#: De characteristic-UUID.

properties#: Bitmasker van ondersteunde GATT-bewerkingen zoals gerapporteerd door de peer.

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

Async. Voer een GATT Read uit en retourneer de waarde. Werpt GattError bij een status ongelijk aan nul.

timeout_ms: Read-time-out.

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

Async. Voer een GATT Write uit.

data: Waarde om te schrijven.
response: True om een write-response te vereisen (en GattError te werpen bij mislukking). False voor write-without-response. None (standaard) selecteert automatisch op basis van wat de peer adverteert.
timeout_ms: Write-time-out (alleen relevant als response True is).

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

Async. Wacht op de volgende notification op deze characteristic en retourneer de payload ervan. Retourneert onmiddellijk als er al een notification in de wachtrij staat.

timeout_ms: Maximale wachttijd. None wacht oneindig.

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

Async. Wacht op de volgende indication op deze characteristic en retourneer de payload ervan.

timeout_ms: Maximale wachttijd.

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

Async. Schrijf de Client Characteristic Configuration Descriptor (CCCD) om te abonneren (of afmelden) voor notifications en/of indications.

notify: Schakel notifications in.
indicate: Schakel indications in.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]#: Async. Ontdek een enkele descriptor op UUID, of None indien niet gevonden.

descriptors(timeout_ms: int = 2000) → ClientDiscover#: Retourneert een async iterator van ClientDescriptor-objecten. Gebruik met async for en laat de lus tot het einde doorlopen.

class aioble.ClientDescriptor#

Een externe GATT-descriptor die op een peer is ontdekt. Erft read en write van ClientCharacteristic.

Niet rechtstreeks construeren.

characteristic#: De bezittende ClientCharacteristic.

uuid#: De descriptor-UUID.

class aioble.L2CAPChannel#

Een actief L2CAP connection-oriented kanaal. Geretourneerd door DeviceConnection.l2cap_accept() of DeviceConnection.l2cap_connect(). Ondersteunt gebruik als async with context manager die automatisch ontkoppelt bij het afsluiten.

Niet rechtstreeks construeren.

our_mtu#: Maximale grootte, in bytes, die de peer in een enkele SDU naar ons mag sturen.

peer_mtu#: Maximale grootte, in bytes, die wij in een enkele SDU naar de peer mogen sturen.

available() → bool#: Retourneert synchroon True als er gebufferde ontvangstdata klaarstaat (d.w.z. recvinto zal niet blokkeren).

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

Async. Ontvang in buf en retourneer het aantal gelezen bytes. Wacht op nieuwe data als het kanaal leeg is.

buf: Vooraf toegewezen buffer om te vullen.
timeout_ms: Maximale wachttijd. None wacht oneindig.

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

Async. Stuur buf over het kanaal en fragmenteer grotere payloads in chunks ter grootte van de MTU. Wacht op flow-control-credits indien nodig.

buf: Bytes-achtig object om te versturen.
timeout_ms: Maximale tijd om per chunk te wachten.
chunk_size: Optionele override voor de chunkgrootte per aanroep. Begrensd tot min(our_mtu * 2, peer_mtu).

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

Async. Wacht totdat een eventueel gestokte send door de controller is afgevoerd.

timeout_ms: Maximale wachttijd.

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

Async. Verbreek het kanaal en wacht op de disconnection-IRQ.

timeout_ms: Maximale wachttijd.

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

Async. Wacht totdat het kanaal door een van beide partijen wordt verbroken.

timeout_ms: Maximale wachttijd.

aioble — Async BLE#

Voorbeelden#

Functies op moduleniveau#

Constanten op moduleniveau#

Uitzonderingen#

Klassen#

`aioble` — Async BLE#