11.10. Działanie jako central#

Drugą stroną rozmowy jest central – urządzenie, które skanuje w poszukiwaniu rozgłaszających urządzeń peryferyjnych, wybiera jedno do rozmowy, otwiera połączenie, przechodzi przez zdalną bazę danych GATT oraz odczytuje lub subskrybuje znajdujące się w niej charakterystyki. Kamera zbierająca odczyty z noszonego sensora, nasłuchująca beaconu lub rozmawiająca z towarzyszącym mikrokontrolerem jest urządzeniem central.

Wzorzec central w aioble przebiega przez cztery etapy: skanowanie, łączenie, odkrywanie, operowanie.

11.10.1. Skanowanie#

aioble.scan() zwraca asynchroniczny menedżer kontekstu, który pełni jednocześnie rolę asynchronicznego iteratora po wykrytych urządzeniach. Typowym zastosowaniem jest skanowanie do momentu pojawienia się interesującego urządzenia, a następnie wyjście z iteracji:

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 ogranicza czas trwania skanowania; duration_ms=0 skanuje bez końca (do momentu wyjścia z menedżera kontekstu). active=True żąda odpowiedzi skanowania, co podwaja rozmiar ładunku na urządzenie kosztem niewielkiej dodatkowej transmisji z obu stron. Pozostałe argumenty nazwane interval_us / window_us dostrajają cykl pracy radia samego skanera i rzadko są zmieniane względem wartości domyślnych.

Każdy aioble.ScanResult udostępnia adres urządzenia, ostatnie RSSI, surowe bajty rozgłaszania i odpowiedzi skanowania oraz pomocniki parsujące standardowe pola:

  • result.device – aioble.Device gotowe do wywołania na nim connect().

  • result.rssi – wskaźnik siły odbieranego sygnału w dBm, przydatny w logice „wybierz najbliższe”.

  • result.name() – ciąg z nazwą lokalną lub None, jeśli nie jest rozgłaszany.

  • result.services() – generator obiektów bluetooth.UUID dla każdej usługi, którą urządzenie rozgłasza.

  • result.manufacturer() – generator krotek (company_id, data) dla pól specyficznych dla producenta.

  • result.connectable – czy najnowsze rozgłoszenie było rozgłoszeniem dopuszczającym połączenie.

Ten sam ScanResult jest ponownie zwracany w miarę napływania nowych danych rozgłoszeniowych dla tego samego urządzenia, dzięki czemu pasywny nasłuchiwacz, który chce po prostu śledzić urządzenia bez końca, może uruchomić asynchroniczny iterator na zawsze i reagować na każde zdarzenie.

11.10.2. Łączenie#

Po zidentyfikowaniu docelowego urządzenia otwarcie połączenia to jedno await

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

aioble.Device.connect() przyjmuje timeout_ms (jak długo czekać na nawiązanie połączenia; domyślnie 10 s) oraz min_conn_interval_us / max_conn_interval_us (żądany zakres interwału połączenia z Połączenia).

11.10.2.1. Ponowne łączenie ze znanym partnerem bez skanowania#

Gdy istnieje już powiązanie z partnerem, adres jest już znany, a kolejna runda skanowania i wybierania to zmarnowany czas radia. Skonstruuj aioble.Device bezpośrednio z zapisanym adresem i przejdź od razu do 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

Pierwszy argument to jeden z aioble.ADDR_PUBLIC (fabryczny adres kontrolera) lub aioble.ADDR_RANDOM (wygenerowany statyczny albo rozwiązywalny adres prywatny); drugi to wartość bytes o długości sześciu bajtów lub ciąg szesnastkowy rozdzielony dwukropkami. Atrybuty addr_type i addr dowolnego Device (np. uzyskanego wcześniej z ScanResult) można utrwalić i podać tutaj z powrotem.

Zwracany aioble.DeviceConnection to obiekt, na którym opiera się reszta pracy urządzenia central. async with gwarantuje zamknięcie połączenia przy wyjściu z bloku – po sukcesie, po anulowaniu lub przy dowolnym wyjątku, w tym aioble.DeviceDisconnectedError z powodu zniknięcia partnera.

Jeśli central potrzebuje większej wartości charakterystyki, niż pozwala domyślny 23-bajtowy MTU, to jest miejsce, aby ją wynegocjować:

await connection.exchange_mtu(512)

(exchange_mtu() zwraca faktycznie wynegocjowany MTU, czyli minimum z żądanej wartości i tego, co obsługuje partner.)

11.10.3. Odkrywanie#

Odkrywanie przechodzi przez zdalną bazę danych GATT, aby znaleźć usługi i charakterystyki po ich UUID. Istnieją dwa warianty: ukierunkowany (znasz UUID i chcesz jednej konkretnej rzeczy) oraz wyczerpujący (chcesz wszystkiego).

Ukierunkowany – najczęstszy przypadek:

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() przyjmują bluetooth.UUID i zwracają pasujący obiekt (lub None). Oba mają argument nazwany timeout_ms na pojedyncze odkrywanie, którego wartość domyślna wynosi 2 s.

Wyczerpujący:

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

Tak właśnie działają ogólne aplikacje typu Bluetooth-explorer – przydatne podczas rozwoju, mniej w kodzie produkcyjnym, który wie, jakich UUID oczekuje.

11.10.3.1. Sprawdzanie, co obsługuje charakterystyka#

Odkrywanie zwraca dla każdej charakterystyki maskę bitową właściwości GATT rozgłoszoną przez partnera jako properties. Bity to te zdefiniowane przez GATT – odczyt (0x02), zapis bez odpowiedzi (0x04), zapis (0x08), powiadomienie (0x10), wskazanie (0x20) i pokrewne. Sprawdzenie maski bitowej przed wydaniem operacji pozwala ogólnemu klientowi dostosować się do charakterystyk, których możliwości nie zna z góry:

_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

Kod produkcyjny, który zna już profil GATT partnera, zwykle tego nie potrzebuje – UUID zostały udokumentowane z góry. Opierają się na tym klienci ogólni / eksploracyjni (strona ustawień przechodząca przez nieznane urządzenie, host wtyczek).

11.10.4. Operowanie#

Gdy central dysponuje ClientCharacteristic, każda operacja GATT to jedno wywołanie korutyny:

  • Odczyt. Wydaj odczyt GATT i odbierz wartość z powrotem:

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

    Długie odczyty (wartości większe niż MTU) są obsługiwane w sposób przezroczysty.

  • Zapis. Wyślij nową wartość do serwera:

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

    response=True czeka na odpowiedź zapisu i zgłasza aioble.GattError, jeśli serwer odrzuci zapis. response=False to zapis bez odpowiedzi: wyślij i zapomnij. response=None (wartość domyślna) automatycznie dobiera wariant na podstawie tego, co rozgłosił partner.

  • Subskrypcja. Włącz powiadomienia lub wskazania, zapisując do CCCD charakterystyki:

    await char.subscribe(notify=True)

    Po jej powrocie central może czekać na przychodzące powiadomienia.

  • Powiadomienie / wskazanie. Czekaj na następne powiadomienie od serwera:

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

    timeout_ms=None (wartość domyślna) czeka bez końca; przekaż liczbę całkowitą w milisekundach, aby zrezygnować po pewnym czasie.

Połączenie tych czterech elementów daje kanoniczny program central typu „połącz, subskrybuj, strumieniuj”:

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

Całość liczy około tuzina linii i obejmuje przepływ od „żaden Bluetooth nie działa” do „strumieniowania danych na żywo”. Iterator skanowania pasuje do wzorca broadcaster/observer, connect otwiera połączenie GAP, service / characteristic przechodzi przez drzewo GATT, subscribe zapisuje CCCD, a notified czeka na powiadomienia.

11.10.5. Rozłączenia i ponowne łączenie#

Wszystko, co dzieje się z łączem radiowym, ujawnia się w korutynie, która na nie oczekiwała. aioble.DeviceDisconnectedError to sygnał, że partner zniknął lub że wyzwolił się limit czasu nadzoru; wyjątek przerywa dowolne będące w toku wywołanie read(), write() lub notified(), a każdy blok async with connection kończy się czysto.

Central, który ma ponownie łączyć się po utracie, opakowuje pracę we własną pętlę zewnętrzną:

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. Obejmowanie sekwencji za pomocą timeout()#

Gdy kilka następujących po sobie operacji GATT ma w całości zmieścić się w jednym budżecie – a nie każda osobno na własnym timeout_ms – użyj aioble.DeviceConnection.timeout(), aby je opakować. Zwracany menedżer kontekstu anuluje swoje ciało, jeśli budżet upłynie (zgłaszając asyncio.TimeoutError) lub jeśli partner się rozłączy (zgłaszając 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")

To czystsza alternatywa dla opakowywania każdego wywołania z osobna w asyncio.wait_for() i unika fałszywych sukcesów, w których każde wywołanie mieści się we własnym terminie, ale sekwencja jako całość go przekracza. Przekazanie timeout_ms=None do timeout() wyłącza termin i pozostawia aktywną jedynie ochronę przed rozłączeniem.