11.10. Acționând ca central

Cealaltă parte a conversației este centralul – dispozitivul care scanează după periferice ce difuzează reclame, alege unul cu care să comunice, deschide o conexiune, parcurge baza de date GATT la distanță și citește sau se abonează la caracteristicile de pe aceasta. O cameră care colectează măsurători de la un senzor purtabil, ascultă un beacon sau comunică cu un microcontroler însoțitor este un central.

Tiparul de central în aioble parcurge patru etape: scanare, conectare, descoperire, operare.

11.10.1. Scanare

aioble.scan() returnează un manager de context asincron care funcționează și ca iterator asincron peste dispozitivele descoperite. Utilizarea tipică este de a scana până când apare un dispozitiv de interes, apoi de a ieși din iterație:

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 limitează durata scanării; duration_ms=0 scanează la nesfârșit (până când managerul de context iese). active=True solicită răspunsuri de scanare, ceea ce dublează dimensiunea sarcinii utile per dispozitiv, cu prețul unei mici transmisii suplimentare din ambele părți. Argumentele rămase interval_us / window_us reglează ciclul de funcționare al radioului scanerului și sunt rareori modificate față de valorile implicite.

Fiecare aioble.ScanResult expune adresa dispozitivului, ultimul RSSI, octeții bruți de reclamă și de răspuns de scanare, precum și utilitare care analizează câmpurile standard:

  • result.device – un aioble.Device pregătit pentru a apela connect().

  • result.rssi – indicatorul intensității semnalului recepționat în dBm, util pentru logica de tip „alege-l pe cel mai apropiat”.

  • result.name() – șirul cu numele local sau None dacă nu este difuzat.

  • result.services() – un generator de bluetooth.UUID pentru fiecare serviciu pe care îl difuzează dispozitivul.

  • result.manufacturer() – un generator de tupluri (company_id, data) pentru câmpurile specifice producătorului.

  • result.connectable – dacă cea mai recentă reclamă a fost una care permite conectarea.

Același ScanResult este produs din nou pe măsură ce sosesc date noi de reclamă pentru același dispozitiv, astfel încât un ascultător pasiv care dorește doar să urmărească dispozitivele la nesfârșit poate rula iteratorul asincron pe termen nelimitat și poate dispeceriza fiecare eveniment.

11.10.2. Conectare

Odată ce un dispozitiv țintă este identificat, deschiderea unei conexiuni se face cu un singur await

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

aioble.Device.connect() acceptă timeout_ms (cât timp să aștepte stabilirea conexiunii; implicit 10 s) și min_conn_interval_us / max_conn_interval_us (intervalul de conexiune solicitat de la Conexiuni).

11.10.2.1. Reconectarea la un partener cunoscut fără scanare

Odată ce există o legătură de tip bond cu un partener, adresa este deja cunoscută, iar o altă rundă de scanare-și-selectare este timp de radio irosit. Construiți un aioble.Device direct cu adresa salvată și treceți direct la 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

Primul argument este una dintre valorile aioble.ADDR_PUBLIC (adresa de fabrică a unui controler) sau aioble.ADDR_RANDOM (o adresă privată statică sau rezolvabilă generată); al doilea este fie o valoare bytes de șase octeți, fie un șir hexazecimal separat prin două puncte. Atributele addr_type și addr ale oricărui Device (de exemplu, unul obținut anterior dintr-un ScanResult) pot fi persistate și reintroduse aici.

Obiectul aioble.DeviceConnection returnat este cel de care depinde restul activității centralului. async with asigură închiderea conexiunii la ieșirea din bloc – la succes, la anulare sau la orice excepție, inclusiv aioble.DeviceDisconnectedError cauzată de dispariția partenerului.

Dacă centralul are nevoie de o valoare de caracteristică mai mare decât permite MTU-ul implicit de 23 de octeți, acesta este locul în care să o negocieze:

await connection.exchange_mtu(512)

(exchange_mtu() returnează MTU-ul efectiv negociat, care este minimul dintre valoarea solicitată și ceea ce suportă partenerul.)

11.10.3. Descoperire

Descoperirea parcurge baza de date GATT la distanță pentru a găsi serviciile și caracteristicile după UUID-urile lor. Există două variante: țintită (cunoașteți UUID-ul și doriți un anumit lucru specific) și exhaustivă (doriți totul).

Țintită – cazul comun:

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() și aioble.ClientService.characteristic() acceptă fiecare un bluetooth.UUID și returnează obiectul corespunzător (sau None). Ambele au un argument timeout_ms per descoperire, cu valoarea implicită de 2 s.

Exhaustivă:

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

Acesta este ceea ce fac aplicațiile generice de explorare Bluetooth – util pentru dezvoltare, mai puțin util pentru codul de producție care știe ce UUID-uri așteaptă.

11.10.3.1. Inspectarea a ceea ce suportă o caracteristică

Descoperirea returnează masca de biți a proprietăților GATT pe care partenerul a difuzat-o pentru fiecare caracteristică, sub forma properties. Biții sunt cei definiți de GATT – citire (0x02), scriere-fără-răspuns (0x04), scriere (0x08), notificare (0x10), indicare (0x20) și altele. Inspectarea măștii de biți înainte de a emite o operație permite unui client generic să se adapteze la caracteristici ale căror capabilități nu le cunoaște dinainte:

_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

Codul de producție care cunoaște deja profilul GATT al partenerului de obicei nu are nevoie de acest lucru – UUID-urile au fost documentate din start. Clienții generici / exploratori (o pagină de setări care parcurge un dispozitiv necunoscut, o gazdă de plugin-uri) se bazează pe el.

11.10.4. Operare

Odată ce centralul deține o ClientCharacteristic, fiecare operație GATT este un singur apel de corutină:

  • Citire. Emiteți o citire GATT și obțineți valoarea înapoi:

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

    Citirile lungi (valori mai mari decât MTU-ul) sunt gestionate transparent.

  • Scriere. Trimiteți o valoare nouă către server:

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

    response=True așteaptă un răspuns de scriere și generează aioble.GattError dacă serverul respinge scrierea. response=False este scriere-fără-răspuns: trimite-și-uită. response=None (valoarea implicită) alege automat în funcție de ceea ce a difuzat partenerul.

  • Abonare. Activați notificările sau indicațiile scriind în CCCD-ul caracteristicii:

    await char.subscribe(notify=True)

    După ce acest apel returnează, centralul poate aștepta sosirea unor transmisii.

  • Notificat / indicat. Așteptați următoarea transmisie de la server:

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

    timeout_ms=None (valoarea implicită) așteaptă la nesfârșit; transmiteți un întreg în milisecunde pentru a renunța după un timp.

Punând cele patru etape laolaltă obținem programul canonic de central „conectare, abonare, transmisie continuă”:

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

Întregul lucru are aproximativ o duzină de linii și acoperă fluxul de la „niciun Bluetooth în execuție” la „date transmise în direct”. Iteratorul de scanare corespunde tiparului emițător/observator, connect deschide conexiunea GAP, service / characteristic parcurge arborele GATT, subscribe scrie CCCD-ul, iar notified așteaptă transmisiile.

11.10.5. Deconectări și reconectare

Orice se întâmplă cu legătura radio apare în corutina care aștepta pe ea. aioble.DeviceDisconnectedError este semnalul că partenerul a dispărut sau că s-a declanșat timeout-ul de supraveghere; excepția întrerupe orice apel read(), write() sau notified() aflat în desfășurare, iar orice bloc async with connection iese curat.

Un central care ar trebui să se reconecteze în caz de pierdere încadrează activitatea în propria sa buclă exterioară:

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. Încadrarea unei secvențe cu timeout()

Când mai multe operații GATT consecutive ar trebui să se încheie toate în cadrul unui singur buget – nu fiecare individual cu propriul timeout_ms – folosiți aioble.DeviceConnection.timeout() pentru a le încadra. Managerul de context returnat anulează corpul său dacă bugetul expiră (generând asyncio.TimeoutError) sau dacă partenerul se deconectează (generând 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")

Aceasta este alternativa mai curată față de încadrarea fiecărui apel individual în asyncio.wait_for() și evită succesele false în care fiecare apel își respectă propriul termen-limită, dar secvența în ansamblu îl depășește. Transmiterea timeout_ms=None către timeout() dezactivează termenul-limită și lasă activă doar protecția împotriva deconectării.