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.