`aioble` — Asynkron BLE¶

aioble är ett högnivåomslag, anpassat för asyncio, runt modulen bluetooth. Det tillhandahåller rena coroutiner för skanning, anslutning, annonsering, GATT-tjänster och L2CAP-kanaler.

Alla fjärroperationer (connect, disconnect, klient-läs/skriv, server-indicate, l2cap recv/send, pair) är awaitable och stöder timeouter.

Roller som stöds:

Broadcaster (annonserare) — genererar nyttolaster för annonsering och skanningssvar för vanliga fält, delar automatiskt upp nyttolasten mellan annonsering och skanningssvar, annonserar obegränsat eller under en bestämd tid.
Peripheral — väntar på anslutning från en central, väntar på MTU-utbyte.
Observer (skanner) — passiv och aktiv skanning, kombinerar nyttolaster för annonsering och skanningssvar för samma enhet, tolkar vanliga fält från annonseringsnyttolaster.
Central — ansluter till en peripheral, initierar MTU-utbyte.
GATT Client — upptäcker tjänster / egenskaper / deskriptorer (valfritt via UUID); läser / skriver / skriv-med-svar på egenskaper och deskriptorer; prenumererar på notifieringar och indikationer (via CCCD); väntar på notifieringar och indikationer.
GATT Server — registrerar tjänster / egenskaper / deskriptorer; väntar på skrivningar till egenskaper och deskriptorer; fångar upp läsförfrågningar; skickar notifieringar och indikationer (och väntar på svar).
L2CAP — accepterar och ansluter anslutningsorienterade L2CAP-kanaler, hanterar flödeskontroll för kanalen.
Security — JSON-baserad hantering av nycklar/hemligheter, initierar parkoppling, frågar efter kryptering / autentiseringsstatus.

Exempel¶

Skanna efter närliggande BLE-enheter och skriv ut varje enhet när den upptäcks:

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

Anslut till en peripheral som annonserar Heart Rate-tjänsten som central och prenumerera på dess mätnotifieringar:

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

Agera som peripheral: registrera en GATT-tjänst, annonsera den och skicka notifieringar till den som ansluter:

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

Funktioner på modulnivå¶

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

Vidarebefordrar till bluetooth.BLE.config() och säkerställer först att BLE-radion är aktiv.

args: Valfritt enskilt parameternamn att fråga efter.
kwargs: Nyckelordsargument för att ange konfigurationsvärden.

aioble.stop() → None¶: Inaktivera den underliggande BLE-radion och kör alla registrerade avstängningshanterare för undermoduler. Efter detta anrop rivs alla skannrar, annonserare, anslutningar och L2CAP-kanaler ner.

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

Returnerar en scan-kontexthanterare/asynkron iterator som ger ScanResult-instanser för varje unik enhet som upptäcks (eller för varje ny bit annonseringsdata från en känd enhet).

duration_ms: Hur länge skanningen ska pågå, i millisekunder. Ange 0 för att skanna obegränsat tills kontexthanteraren avslutas.
interval_us: Skanningsintervall i mikrosekunder. Standardvärdet är 1 280 000.
window_us: Skanningsfönster i mikrosekunder (måste vara mindre än eller lika med interval_us). Standardvärdet är 11 250.
active: Om True utförs en aktiv skanning (begär skanningssvarsdata). Standardvärdet är 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¶

Asynkron coroutine som börjar annonsera och väntar på en inkommande central-anslutning. Returnerar en DeviceConnection som representerar den anslutna centralen, eller väcker asyncio.TimeoutError vid timeout.

interval_us: Annonseringsintervall, i mikrosekunder.
adv_data: Rå annonseringsnyttolast. Om den inte anges byggs adv_data upp från de återstående nyckelordsargumenten.
resp_data: Rå nyttolast för skanningssvar. Fylls automatiskt med överflöd från adv_data vid behov.
connectable: Om True är detta en anslutningsbar annons.
limited_disc: Använd flaggan för begränsad upptäckbarhet i stället för allmän.
br_edr: Sätt flaggan för BR/EDR-stöd.
name: Valfritt fullständigt lokalt namn att bädda in.
services: Itererbar samling av bluetooth.UUID att annonsera.
appearance: 16-bitars appearance-värde (se Bluetooths tilldelade nummer).
manufacturer: Tupel av (company_id, data_bytes) att annonsera som tillverkarspecifik data.
timeout_ms: Sluta annonsera efter så här många millisekunder utan anslutning. None innebär att annonseringen fortsätter tills anslutning sker.

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

Registrera ett eller flera Service-objekt (samt deras egenskaper och deskriptorer) hos GATT-servern. Måste anropas en gång innan advertise startas. Efterföljande anrop ersätter den tidigare registreringen.

services: En eller flera Service-instanser.

Konstanter på modulnivå¶

aioble.ADDR_PUBLIC¶: Publik BLE-enhetsadresstyp (0).

aioble.ADDR_RANDOM¶: Slumpmässig BLE-enhetsadresstyp (1).

Undantag¶

exception aioble.GattError¶: Väcks när en GATT-fjärroperation (läs / skriv / indicate) slutförs med en status som inte är noll. Statuskoden är tillgänglig via attributet _status.

exception aioble.DeviceDisconnectedError¶: Väcks inuti en asynkron operation (t.ex. läs, skriv, notified) när den underliggande anslutningen bryts under väntan.

exception aioble.L2CAPDisconnectedError¶: Väcks när en send/recv/flush-operation på en L2CAP-kanal försöks utföras på (eller avbryts av) en frånkopplad kanal.

exception aioble.L2CAPConnectionError¶: Väcks av DeviceConnection.l2cap_connect när upprättandet av kanalen misslyckas. Bluetooth-statuskoden är det första argumentet.

Klasser¶

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

Representerar en BLE-fjärrenhet via adress. Två Device-instanser jämförs som lika om både addr_type och addr stämmer överens. Används som handtag för att initiera anslutningar.

addr_type: Antingen ADDR_PUBLIC eller ADDR_RANDOM.
addr: Sex byte lång adress som bytes, eller en kolonseparerad hexsträng (t.ex. "aa:bb:cc:dd:ee:ff").

addr_type¶: Adresstypen som enheten konstruerades med.

addr¶: Den råa, sex byte långa enhetsadressen.

addr_hex() → str¶: Returnera adressen formaterad som en kolonseparerad hexsträng.

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

Asynkron. Initiera en GAP-anslutning till denna enhet och returnera den resulterande DeviceConnection. Avbryter all pågående skanning.

timeout_ms: Hur länge man ska vänta på att anslutningen ska slutföras.
scan_duration_ms: Inledande skanningstid före anslutning (kontrollerspecifikt).
min_conn_interval_us / max_conn_interval_us: Valfria gränser för anslutningsintervall, i mikrosekunder.

class aioble.DeviceConnection¶

En aktiv GAP-anslutning till en Device. Returneras av Device.connect() eller advertise. Stöder användning som en async with-kontexthanterare som automatiskt kopplar från vid avslut.

Konstruera inte direkt.

device¶: Den underliggande Device.

encrypted¶: True när länken är krypterad (t.ex. efter parkoppling).

authenticated¶: True om länken autentiserades (MITM-skyddad parkoppling).

bonded¶: True om parkopplingen gav bindningsnycklar.

key_size¶: Förhandlad krypteringsnyckelstorlek i byte, eller False om den inte är krypterad.

mtu¶: Förhandlad ATT MTU efter exchange_mtu, eller None tills den är satt.

is_connected() → bool¶: Returnera huruvida anslutningen fortfarande är aktiv.

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

Asynkron. Koppla från och vänta på frånkopplings-IRQ:n.

timeout_ms: Maximal tid att vänta på frånkopplingen.

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

Asynkron. Vänta tills anslutningen avslutas av endera sidan. Om disconnect är True kopplas den aktivt från först.

timeout_ms: Maximal tid att vänta. None innebär att vänta för evigt.
disconnect: Om True initieras frånkoppling.

timeout(timeout_ms: int | None) → DeviceTimeout¶

Returnera en kontexthanterare som avbryter sin kropp om antingen timeouten löper ut (väcker asyncio.TimeoutError) eller enheten kopplas från (väcker DeviceDisconnectedError).

timeout_ms: Timeout i millisekunder, eller None för ingen timeout.

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

Asynkron. Initiera ett ATT MTU-utbyte och returnera den förhandlade MTU:n.

mtu: Valfri önskad MTU att sätta på det underliggande BLE-gränssnittet före utbytet.
timeout_ms: Timeout för utbytet.

service(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientService | None]¶: Asynkron. Upptäck en enskild fjärrtjänst som matchar uuid, eller None om ingen hittas.

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

Returnera en asynkron iterator av ClientService-fjärrobjekt. Använd med async for och kör slingan tills den är klar.

uuid: Valfritt UUID-filter. None returnerar varje tjänst.
timeout_ms: Timeout per upptäckt.

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

Asynkron. Initiera parkoppling på denna anslutning. Uppdaterar attributen encrypted / authenticated / bonded / key_size när det är klart.

bond: Spara parkopplingsnycklar permanent.
le_secure: Använd LE Secure Connections.
mitm: Kräv man-in-the-middle-skydd.
io: Konstant för IO-förmåga (t.ex. 3 för ingen in-/utmatning).
timeout_ms: Timeout för parkoppling.

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

Asynkron. Lyssna på den angivna PSM:en och returnera en L2CAPChannel när fjärrsidan öppnar den.

psm: Protocol/Service Multiplexer att lyssna på.
mtu: Maximal mottagningsstorlek, i byte.
timeout_ms: Maximal tid att vänta på att fjärrsidan ska ansluta.

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

Asynkron. Öppna en L2CAP-kanal till fjärrsidan på den angivna PSM:en.

psm: Protocol/Service Multiplexer att ansluta till.
mtu: Maximal mottagningsstorlek, i byte.
timeout_ms: Timeout för anslutning.

class aioble.ScanResult¶

En enskild enhet som upptäcks under scan. Samma instans ges på nytt allt eftersom ny annonseringsdata anländer.

Konstruera inte direkt.

device¶: Den underliggande Device.

rssi¶: Senast rapporterade RSSI, i dBm.

adv_data¶: Rå annonseringsnyttolast (bytes eller None).

resp_data¶: Rå nyttolast för skanningssvar (bytes eller None), om aktiv skanning är aktiverad.

connectable¶: True om den senaste annonsen var anslutningsbar.

name() → str | None¶: Avkoda det fullständiga (eller förkortade) annonserade lokala namnet från nyttolasten, eller None om det inte finns.

services() → Iterator[bluetooth.UUID]¶: Generator som ger varje bluetooth.UUID som annonseras i 16/32/128-bitars tjänstelistefälten.

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

Generator som ger (company_id, data)-tupler från tillverkarspecifika annonseringsfält.

filter: Om angivet ges endast poster vars company ID matchar.

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

En lokal GATT-tjänst. Bygg en tjänst med en eller flera Characteristic-instanser och skicka den sedan till register_services.

uuid: Tjänstens UUID.

uuid¶: Tjänstens UUID.

characteristics¶: Lista över Characteristic-objekt knutna till denna tjänst.

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

En lokal GATT-egenskap. Att konstruera en lägger automatiskt till den i service.

service: Den ägande Service.
uuid: Egenskapens UUID.
read, write, write_no_response, notify, indicate: Booleska värden som väljer vilka GATT-operationer som stöds.
initial: Valfritt initialvärde (bytes).
capture: Om True köas skrivna värden (upp till 10 djupt) så att snabba skrivningar i följd inte går förlorade. Varje written-anrop returnerar då en (connection, data)-tupel.

uuid¶: Egenskapens UUID.

flags¶: Bitmask av GATT-egenskapsflaggorna som byggts upp från konstruktorn.

descriptors¶: Lista över Descriptor-objekt knutna till denna egenskap.

read() → bytes¶: Läs det aktuella värdet från den lokala GATT-databasen.

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

Uppdatera värdet i den lokala GATT-databasen.

data: Nya värdebyte.
send_update: Om True notifieras/indikeras också varje prenumererad anslutning.

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

Skicka en GATT Notify till connection.

connection: Målets klientanslutning.
data: Nyttolast att skicka. Om None skickas det aktuella lokala värdet.

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

Asynkron. Skicka en GATT Indicate till connection och vänta på klientbekräftelsen. Väcker GattError vid en status som inte är noll.

connection: Målets klientanslutning.
data: Nyttolast att indikera, eller None för att skicka det lokala värdet.
timeout_ms: Maximal tid att vänta på bekräftelse.

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

Asynkron. Vänta på en fjärrskrivning. Returnerar den skrivande DeviceConnection, eller (connection, data) om egenskapen skapades med capture=True.

timeout_ms: Maximal tid att vänta. None väntar för evigt.

on_read(connection: DeviceConnection) → int¶: Åsidosättningskrok som anropas synkront när en fjärrläsning tas emot. Returnera 0 för att tillåta läsningen eller en ATT-felkod som inte är noll för att avvisa den. Standardimplementationen returnerar 0.

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

En Characteristic vars underliggande GATT-buffert kan konfigureras. Användbar för att ta emot värden större än standardattributstorleken, eller för att köa skrivningar i följd.

max_len: Buffertstorlek, i byte.
append: Om True läggs sekventiella skrivningar till i bufferten i stället för att skriva över.

Övriga argument vidarebefordras till Characteristic.

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

En lokal GATT-deskriptor. Att konstruera en lägger automatiskt till den i characteristic. Ärver read, write och written från Characteristic.

characteristic: Den ägande Characteristic.
uuid: Deskriptorns UUID.
read, write: Booleska värden som väljer vilka GATT-operationer som stöds.
initial: Valfritt initialvärde (bytes).

class aioble.ClientService¶

En GATT-fjärrtjänst som upptäckts på en motpart. Returneras av DeviceConnection.service() eller itereras från DeviceConnection.services().

Konstruera inte direkt.

connection¶: Den ägande DeviceConnection.

uuid¶: Fjärrtjänstens UUID.

characteristic(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientCharacteristic | None]¶: Asynkron. Upptäck en enskild egenskap via UUID, eller None om ingen hittas.

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

Returnera en asynkron iterator av ClientCharacteristic-objekt. Använd med async for och kör slingan tills den är klar.

uuid: Valfritt UUID-filter.
timeout_ms: Timeout per upptäckt.

class aioble.ClientCharacteristic¶

En GATT-fjärregenskap som upptäckts på en motpart. Returneras av ClientService.characteristic() eller itereras från ClientService.characteristics().

Konstruera inte direkt.

service¶: Den ägande ClientService.

uuid¶: Egenskapens UUID.

properties¶: Bitmask av GATT-operationer som stöds enligt vad motparten rapporterar.

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

Asynkron. Utfärda en GATT Read och returnera värdet. Väcker GattError vid en status som inte är noll.

timeout_ms: Timeout för läsning.

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

Asynkron. Utfärda en GATT Write.

data: Värde att skriva.
response: True för att kräva ett skrivsvar (och väcka GattError vid fel). False för skrivning-utan-svar. None (standard) väljer automatiskt baserat på vad motparten annonserar.
timeout_ms: Timeout för skrivning (endast relevant om response är True).

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

Asynkron. Vänta på nästa notifiering på denna egenskap och returnera dess nyttolast. Returnerar omedelbart om en notifiering redan är köad.

timeout_ms: Maximal tid att vänta. None väntar för evigt.

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

Asynkron. Vänta på nästa indikation på denna egenskap och returnera dess nyttolast.

timeout_ms: Maximal tid att vänta.

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

Asynkron. Skriv Client Characteristic Configuration Descriptor (CCCD) för att prenumerera (eller avsluta prenumeration) på notifieringar och/eller indikationer.

notify: Aktivera notifieringar.
indicate: Aktivera indikationer.

descriptor(uuid: bluetooth.UUID, timeout_ms: int = 2000) → Awaitable[ClientDescriptor | None]¶: Asynkron. Upptäck en enskild deskriptor via UUID, eller None om ingen hittas.

descriptors(timeout_ms: int = 2000) → ClientDiscover¶: Returnera en asynkron iterator av ClientDescriptor-objekt. Använd med async for och kör slingan tills den är klar.

class aioble.ClientDescriptor¶

En GATT-fjärrdeskriptor som upptäckts på en motpart. Ärver read och write från ClientCharacteristic.

Konstruera inte direkt.

characteristic¶: Den ägande ClientCharacteristic.

uuid¶: Deskriptorns UUID.

class aioble.L2CAPChannel¶

En aktiv anslutningsorienterad L2CAP-kanal. Returneras av DeviceConnection.l2cap_accept() eller DeviceConnection.l2cap_connect(). Stöder användning som en async with-kontexthanterare som automatiskt kopplar från vid avslut.

Konstruera inte direkt.

our_mtu¶: Maximal storlek, i byte, som motparten får skicka till oss i en enda SDU.

peer_mtu¶: Maximal storlek, i byte, som vi får skicka till motparten i en enda SDU.

available() → bool¶: Returnerar synkront True om buffrad mottagningsdata är redo (dvs. recvinto kommer inte att blockera).

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

Asynkron. Ta emot data till buf och returnera antalet lästa byte. Väntar på ny data om kanalen är tom.

buf: Förallokerad buffert att fylla.
timeout_ms: Maximal tid att vänta. None väntar för evigt.

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

Asynkron. Skicka buf på kanalen och fragmentera större nyttolaster i MTU-stora delar. Väntar på flödeskontrollkrediter vid behov.

buf: Byteliknande objekt att skicka.
timeout_ms: Maximal tid att vänta per del.
chunk_size: Valfri åsidosättning av delstorleken per anrop. Begränsas till min(our_mtu * 2, peer_mtu).

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

Asynkron. Vänta tills en eventuellt stoppad send har tömts av kontrollern.

timeout_ms: Maximal tid att vänta.

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

Asynkron. Koppla från kanalen och vänta på frånkopplings-IRQ:n.

timeout_ms: Maximal tid att vänta.

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

Asynkron. Vänta tills kanalen kopplas från av endera sidan.

timeout_ms: Maximal tid att vänta.

aioble — Asynkron BLE¶

Exempel¶

Funktioner på modulnivå¶

Konstanter på modulnivå¶

Undantag¶

Klasser¶

`aioble` — Asynkron BLE¶