ESP
DIYer
32
LAB

Moderní mikrokontrolér v rukou „bastlíře“

ESP32 jako BLE klient v MicroPythonu

Díl čtvrtý: ESP32 přes BT ovládá ESP32

V předchozích třech částech našeho miniseriálu jsme prozkoumali základy nízkoúrovňového BLE, naučili se efektivně programovat s moderní asynchronní knihovnou aioble a vyzkoušeli si ovládání mikrokontroléru přímo z prostředí webového prohlížeče. Dosud se však naše ESP32 nacházelo výhradně v roli serveru (Peripheral), který pasivně čekal, až jej klient – tedy telefon nebo počítač – vyhledá a připojí se k němu.


V dnešním čtvrtém díle tuto logiku obrátíme a zaměříme se na situaci, kdy modul ESP32 vystupuje jako aktivní BLE klient (Central). Čekají nás praktické ukázky kódů navržených tak, aby přímo spolupracovaly s našimi dřívějšími programy. Konkrétně si v prvním příkladu ukážeme, jak v MicroPythonu pro ESP32 vytvořit program, který se v roli klienta připojí k BLE serveru a bude prostřednictvím Bluetooth dálkově ovládat vestavěnou LED na straně serveru. V druhém příkladu si naopak ukážeme, jak realizovat BLE klienta, který zvládne načítání stavu tlačítka BOOT na straně serveru. Tímto způsobem si tedy předvedeme, jak mohou dva moduly ESP32 komunikovat přes Bluetooth napřímo mezi sebou, což otevírá dveře k pokročilejším projektům v oblasti domácí automatizace a bezdrátového měření.

Nalezení zařízení (serveru)

Než se vrhneme na samotné klientské aplikace, je nezbytné porozumět mechanismu, kterým se dvě BLE zařízení vůbec „uvidí“ a propojí. Aby mohl náš modul v roli klienta (Central) s kýmkoliv komunikovat, musí jej nejprve v rádiovém prostoru najít. Tento proces se nazývá skenování (scanning). Zatímco server (Peripheral) pravidelně vysílá svou přítomnost pomocí advertising paketů, klient aktivně naslouchá a snaží se tyto pakety zachytit a interpretovat.

Níže uvedený program představuje nízkoúrovňový BLE scanner postavený pouze na vestavěném modulu bluetooth. Neberme tento kód jen jako program, který nám zobrazí dostupná Bluetooth zařízení v okolí, ale i jako ukázku toho, jak MicroPython zpracovává příchozí data z okolních zařízení:

import time
import bluetooth
from micropython import const

# Definice nízkoúrovňových událostí skenování
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)

devices = {}

def decode_name(adv_data):
    """Pomocná funkce pro dekódování jména zařízení z binárních dat"""
    i = 0
    while i < len(adv_data):
        length = adv_data[i]
        if length == 0: break
        if i + length >= len(adv_data): break

        adv_type = adv_data[i + 1]
        # 0x09 = Complete Local Name, 0x08 = Shortened Local Name
        if adv_type == 0x09 or adv_type == 0x08:
            name = bytes(adv_data[i + 2:i + 1 + length])
            return name.decode("utf-8", "ignore")
        i += length + 1
    return None

def bt_irq(event, data):
    if event == _IRQ_SCAN_RESULT:
        # Přijetí výsledku skenu (MAC adresa, RSSI, data reklamy)
        addr_type, addr, connectable, rssi, adv_data = data
        mac = ':'.join('%02X' % b for b in addr)

        if mac not in devices:
            devices[mac] = {"name": None, "rssi": rssi}

        devices[mac]["rssi"] = rssi
        name = decode_name(adv_data)

        if name:
            devices[mac]["name"] = name

    elif event == _IRQ_SCAN_DONE:
        print("\nScan dokončen. Nalezená zařízení:")

        for mac in devices:
            name = devices[mac]["name"] or "bez_jména"
            print(f"{name} | {mac} | RSSI: {devices[mac]['rssi']} dBm")

# Inicializace BLE v modulu ESP32
bt = bluetooth.BLE()
bt.active(True)
bt.irq(bt_irq)

print("Aktivní BLE scan (10 sekund)...")
# Spuštění skenování: interval a okno nastaveny na 30ms, aktivní sken = True
bt.gap_scan(10000, 30000, 30000, True)

time.sleep_ms(10500)

Tento příklad nám ukazuje několik klíčových konceptů nízkoúrovňového BLE:

  • Zpracování událostí (IRQ): Stejně jako u serveru, i skenování běží přes přerušení. Událost _IRQ_SCAN_RESULT (číslo 5) nastane pokaždé, když ESP32 zachytí paket od nějakého zařízení v dosahu.
  • Binární data advertisingu: Jméno zařízení (např. ESP32-LED) není v paketu uloženo jako prostý text. Je „zabaleno“ v binární struktuře, kterou musí funkce decode_name ručně rozebrat podle specifikace BLE.
  • RSSI: Skenování nám vrací i sílu signálu (RSSI), což nám umožňuje odhadnout vzdálenost serveru.
  • Aktivní vs. pasivní sken: Poslední parametr True ve funkci gap_scan zapíná aktivní skenování, při kterém se ESP32 dotazuje zařízení na další podrobnosti (např. právě na ono jméno).

Aktivní vs. pasivní sken

V kontextu BLE komunikace je důležité rozlišit mezi pasivním a aktivním skenováním, které přímo ovlivňuje to, jaké informace z advertisingu (vysílání) získáme.

  • Pasivní skenování (Passive Scanning): V tomto režimu klient (Central) pouze naslouchá a zachytává reklamní pakety, které zařízení v okolí běžně vysílají. Výhodou je nižší spotřeba energie na straně klienta, protože pouze přijímá data a nic nevysílá. Nevýhodou je, že získá pouze základní data obsažená v hlavním advertising paketu, který má omezenou kapacitu (typicky 31 bajtů).
  • Aktivní skenování (Active Scanning): Pokud klient používá aktivní sken (v MicroPythonu parametr True ve funkci gap_scan), proces probíhá ve dvou krocích. Jakmile klient zachytí základní advertising paket, okamžitě pošle zařízení dotaz zvaný Scan Request. Zařízení (Advertiser) na něj odpoví doplňujícím paketem – Scan Response. Tento druhý paket může obsahovat další data, která se do prvního nevešla, například úplné jméno zařízení nebo seznam podporovaných UUID služeb.

V praxi to znamená, že pokud váš serverový program vysílá jméno zařízení, klient jej při pasivním skenu nemusí vidět, pokud je jméno uloženo až v paketu Scan Response. Aktivní skenování je tedy spolehlivější pro identifikaci zařízení, ale za cenu mírně vyšší energetické náročnosti, protože klient musí aktivně vysílat žádosti o data.

BLE scanner
Obrázek č. 1 – Ukázka výstupu BLE scanneru (výstup: jméno, BT MAC adresa a RSSI - síla signálu)

Způsob připojení a získání jména si ještě podrobněji ukážeme hned u prvního příkladu. Takže to zatím berme tak trochu ad-hoc.

Použití modulu aioble

V jednom z minulých dílů jsme se zmínili o knihovně aioble, která BLE komunikaci výrazně zjednodušuje. Podívejme se tedy, jak by stejný program BT scanner vypadal při použití této knihovny:

import asyncio
import aioble

def format_mac(addr):
    """
    Převede BLE adresu na formát:
    AA:BB:CC:DD:EE:FF
    """
    return ":".join("{:02X}".format(b) for b in addr)

async def scan_ble():
    devices = {}
    print("Aktivní BLE scan (10 sekund)...")

    async with aioble.scan(
        duration_ms=10000,
        interval_us=30000,
        window_us=30000,
        active=True
    ) as scanner:

        async for result in scanner:
            device = result.device

            # BLE adresa
            mac = format_mac(device.addr)

            # RSSI
            rssi = result.rssi

            # Název zařízení
            name = result.name()

            if mac not in devices:
                devices[mac] = {
                    "name": None,
                    "rssi": rssi
                }

            # Aktualizace RSSI při každém zachyceném packetu
            devices[mac]["rssi"] = rssi

            # Pokud jsme získali jméno, uložíme ho
            if name:
                devices[mac]["name"] = name

    print("\nScan dokončen. Nalezená zařízení:")

    for mac, data in devices.items():
        name = data["name"] or "bez_jména"

        print(
            "{} | {} | RSSI: {} dBm".format(
                name,
                mac,
                data["rssi"]
            )
        )

asyncio.run(scan_ble())

Zatímco u nízkoúrovňového přístupu musíme pracovat přímo s BLE událostmi, IRQ callbacky a binárními advertising daty, modul aioble tyto mechanismy zapouzdřuje do přehlednějšího a asynchronního rozhraní.

Při přímém použití modulu bluetooth je nutné registrovat vlastní IRQ handler:

bt.irq(bt_irq)

a následně rozlišovat jednotlivé události:

_IRQ_SCAN_RESULT
_IRQ_SCAN_DONE

Výsledek skenování se získává z callbacku a program musí sám zjistit, jaký typ události právě nastal.

aioble se tento přístup mění. Skenování probíhá asynchronně:

async with aioble.scan(...) as scanner:
    async for result in scanner:
       ...

Program tedy postupně dostává jednotlivé výsledky skenování přímo ve smyčce. Není potřeba vytvářet vlastní IRQ callback.

Jednou z největších změn je použití asynchronního programování. Skenování tedy může běžet jako asynchronní úloha a současně můžeme provádět další činnosti. To je velmi užitečné například u zařízení, které současně skenuje BLE, obsluhuje displej, komunikuje přes Wi-Fi nebo čte senzory.

V původním programu jsme museli procházet binární advertising data a hledat typy: 0x09 nebo 0x08 abychom zjistili název zařízení. Proto vznikla vlastní funkce:

def decode_name(adv_data):
    ...

aioble získáme název přímo z výsledku skenování:

name = result.name()

Odpadá tedy ruční procházení struktury advertising packetu. To ale neznamená, že aioble automaticky zpřístupní úplně všechna data bez další práce. Pokud potřebujeme například konkrétní manufacturer data nebo service data, můžeme sáhnout k podrobnějším údajům ze ScanResult.

U nízkoúrovňového API dostáváme z IRQ callbacku n-tici:

addr_type, addr, connectable, rssi, adv_data = data

Musíme tedy sami vědět, co která položka znamená. V aioble dostáváme objekt výsledku: result a z něj můžeme získat například:

result.device
result.rssi
result.name()

To je přehlednější a méně náchylné k chybám.

Dalším z rozdílů, na který můžeme narazit při přechodu na aioble, je práce s BLE adresou. U nízkoúrovňového API máme přímo:

addr_type, addr, connectable, rssi, adv_data = data

a addr jsou bajty adresy.

aioble získáme zařízení přes:

device = result.device

a jeho adresu přes:

device.addr

Podle verze MicroPythonu a aioble navíc může být device.addr reprezentována speciálním objektem Addr, takže při výpisu adresy je potřeba respektovat konkrétní API dané verze.

Další významný rozdíl je v rozsahu knihovny aioble není pouze náhrada za gap_scan(). Jak jsme již v předešlém článku viděli, je navržena jako vyšší BLE API pro různé operace, například:

  • skenování zařízení,
  • připojování k BLE periferiím,
  • práci se službami,
  • práci s charakteristikami,
  • čtení a zápis hodnot,
  • notifikace,
  • vytváření BLE serverů.

Díky tomu můžeme například po nalezení konkrétního zařízení pokračovat přímo připojením:

device = result.device
connection = await device.connect()

a následně pracovat s jeho GATT službami.

PŘÍKLAD č. 1: Kliente, rozsviť přes BLE na serveru LED

Tento program ukazuje typický způsob komunikace mezi dvěma zařízeními ESP32 pomocí Bluetooth Low Energy (BLE). Náš dnešní program zde funguje jako BLE klient, který sleduje stav tlačítka BOOT. Druhé ESP32 (naprogramováno v předešlých článcích) funguje jako BLE server, který má vytvořenou službu s charakteristikou pro příjem hodnot.

Princip celé ukázky je následující:

  1. Pokud klient zjistí, že bylo stisknuto tlačítko BOOT, odešle serveru hodnotu 1
  2. Čistě pro kontrolu rozsvítí vlastní LED.
  3. Server pozná změnu hodnoty charakteristiky (hodnota 1) a rozsvítí svou vestavěnou LED.
  4. Při uvolnění tlačítka odešle klient hodnotu 0 (a zhasne svou LED).
  5. Server opět na základě hodnoty charakteristiky (hodnota 0) a svou LED zhasne.

Princip našeho BLE klienta a jeho komunikace s BLE serverem znázorňuje animace na obrázku 2:

BLE klient - priklad 1
Obrázek č. 2 – Animace principu BLE klienta a jeho komunikace s  BLE serverem (příklad 1)

Server (viz předešlé články) podle přijaté hodnoty ovládá svou vestavěnou LED.

Výsledkem by tedy mělo být takové „dálkové BLE ovládání“. Po stisku tlačítka BOOT na desce klienta (ESP32 #1) se přes Bluetooth informace přenese do serveru (ESP32 #2), kde se na základě přijaté hodnoty rozsvítí vestavěná LED. Samotný modul klient pro kontrolu také rozsvěcí svou LED. V této dané konfiguraci by tedy výsledkem mělo být současné rozsvěcení vestavěných LED na obou modulech.

Nejdříve si uvedeme celý kód programu, pak jej postupně rozebereme:

import time
import bluetooth
from micropython import const
from machine import Pin

# BLE události
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_SERVICE_RESULT = const(9)
_IRQ_GATTC_SERVICE_DONE = const(10)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_WRITE_DONE = const(17)

# UUID služby a charakteristiky serveru
SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")

# Jméno BLE serveru
SERVER_NAME = "ESP32-LED"

# BOOT tlačítko
button = Pin(0, Pin.IN, Pin.PULL_UP)

# Vestavěná LED klienta
led = Pin(2, Pin.OUT)
led.value(0)

# BLE
ble = bluetooth.BLE()

conn_handle = None
service_start = None
service_end = None
char_value_handle = None

connecting = False

# Získá jméno z advertising nebo Scan Response paketu
def decode_name(data):
    i = 0
    while i < len(data):
        length = data[i]

        if length == 0:
            break

        if i + length >= len(data):
            break

        ad_type = data[i + 1]

        # 0x09 = úplný název
        # 0x08 = zkrácený název

        if ad_type == 0x09 or ad_type == 0x08:
            return bytes(
                data[i + 2:i + 1 + length]
            ).decode("utf-8", "ignore")

        i += length + 1
    return ""

# BLE callback
def irq(event, data):
    global conn_handle
    global service_start
    global service_end
    global char_value_handle
    global connecting

    # Nalezeno BLE zařízení
    if event == _IRQ_SCAN_RESULT:
        addr_type, addr, adv_type, rssi, adv_data = data

        name = decode_name(adv_data)

        # Výpis nalezených zařízení se jménem
        if name:
            mac = ':'.join('%02X' % b for b in addr)
            print("Nalezeno:", name, "|", mac, "| adv_type:", adv_type, "| RSSI:", rssi)

        # Hledáme náš server podle jeho jména.
        # Jméno může být v advertisingu nebo ve Scan Response,
        # proto zde nekontrolujeme adv_type.

        if name == SERVER_NAME and not connecting:
            mac = ':'.join('%02X' % b for b in addr)
            print()
            print("--------------------------------")
            print("Nalezen BLE server")
            print("Jmeno:", name)
            print("MAC  :", mac)
            print("RSSI :", rssi, "dBm")
            print("adv_type:", adv_type)
            print("--------------------------------")

            connecting = True

            # Ukončíme skenování před připojením.
            ble.gap_scan(None)

            print("Pripojuji se...")

            # Adresu z IRQ kopírujeme pomocí bytes().
            ble.gap_connect(
                addr_type,
                bytes(addr)
            )

    # Scan dokončen
    elif event == _IRQ_SCAN_DONE:
        if not connecting:
            print("Scan dokoncen, server nenalezen")
            ble.gap_scan(
                10000,
                30000,
                30000,
                True
            )

    # Připojeno k serveru
    elif event == _IRQ_PERIPHERAL_CONNECT:
        conn_handle, addr_type, addr = data
        connecting = False

        print()
        print("================================")
        print("BLE server PRIPOJEN")
        print("================================")

        print("Hledam BLE sluzbu...")

        # Po připojení začneme hledat konkrétní službu
        # podle jejího UUID.

        ble.gattc_discover_services(
            conn_handle,
            SERVICE_UUID
        )

    # Nalezena naše služba
    elif event == _IRQ_GATTC_SERVICE_RESULT:
        conn, start, end, uuid = data

        if conn == conn_handle and uuid == SERVICE_UUID:
            service_start = start
            service_end = end

            print("BLE sluzba nalezena")

    # Hledání služby je dokončeno.
    elif event == _IRQ_GATTC_SERVICE_DONE:
        if service_start is not None:
            print("Hledam WRITE charakteristiku...")

            ble.gattc_discover_characteristics(
                conn_handle,
                service_start,
                service_end,
                CHAR_UUID
            )

    # Nalezena naše charakteristika
    elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
        conn, def_handle, value_handle, properties, uuid = data

        if conn == conn_handle and uuid == CHAR_UUID:
            char_value_handle = value_handle

            print("WRITE charakteristika nalezena")
            print()
            print("Cekam na tlacitko BOOT...")

    # Potvrzení zápisu do charakteristiky
    elif event == _IRQ_GATTC_WRITE_DONE:
        conn, handle, status = data

        if status == 0:
            print("Hodnota uspesne odeslana")
        else:
            print("Chyba pri zapisu:", status)

    # Odpojeno od serveru
    elif event == _IRQ_PERIPHERAL_DISCONNECT:
        print()
        print("BLE server ODPOJEN")

        conn_handle = None
        service_start = None
        service_end = None
        char_value_handle = None

        connecting = False

        led.value(0)

        print("Znovu hledam BLE server...")

        ble.gap_scan(10000, 30000, 30000, True)

# START BLE
ble.active(True)
ble.irq(irq)

print()
print("================================")
print("SPOUSTIM BLE KLIENTA")
print("================================")
print("Hledam:", SERVER_NAME)
print()

# Aktivní scan = kromě advertising paketů přijímá
# také Scan Response.
# To je důležité, protože server může mít jméno
# právě v Scan Response.

ble.gap_scan(10000, 30000, 30000, True)

# Hlavní smyčka
last_state = button.value()

while True:
    state = button.value()

    if state != last_state:
        last_state = state

        # BOOT STISKNUTO
        if state == 0:
            print()
            print("TLAČÍTKO BOOT: STISKNUTO")

            # Rozsvítíme LED přímo na klientovi.
            led.value(1)

            # Pokud jsme připojeni a známe handle
            # charakteristiky, pošleme serveru "1".
            if (
                conn_handle is not None
                and char_value_handle is not None
            ):
                ble.gattc_write(
                    conn_handle,
                    char_value_handle,
                    b"1",
                    1
                )

                print("Odeslano BLE: 1")

        # BOOT UVOLNĚNO
        else:
            print()
            print("TLAČÍTKO BOOT: UVOLNĚNO")

            # Zhasneme LED na klientovi.
            led.value(0)

            # Serveru pošleme "0".
            if (
                conn_handle is not None
                and char_value_handle is not None
            ):
                ble.gattc_write(
                    conn_handle,
                    char_value_handle,
                    b"0",
                    1
                )

                print("Odeslano BLE: 0")

    time.sleep_ms(20)

Z pohledu BLE komunikace je celý program možné shrnout do následujícího řetězce:

Aktivuj BLE
        ⇓
Spusť BLE scan
        ⇓
Přijímej advertising / Scan Response
        ⇓
Najdi zařízení "ESP32-LED"
        ⇓
Zastav scan
        ⇓
GAP connect
        ⇓
BLE spojení navázáno
        ⇓
GATT service discovery
        ⇓
Najdi SERVICE_UUID
        ⇓
GATT characteristic discovery
        ⇓
Najdi CHAR_UUID
        ⇓
Ulož value_handle
        ⇓
Čekej na změnu tlačítka
        ⇓
    stisk → GATT WRITE "1"
    uvolnění → GATT WRITE "0"

GAP versus GATT

V programu se používají dvě skupiny funkcí, které je dobré rozlišovat.

GAP

Funkce jako:

ble.gap_scan(...)
ble.gap_connect(...)

se starají o samotné hledání zařízení a navazování BLE spojení.

Zjednodušeně: GAP řeší: „S jakým zařízením se spojím?“

GATT

Funkce jako:

ble.gattc_discover_services(...)
ble.gattc_discover_characteristics(...)
ble.gattc_write(...)

řeší práci s daty, která jsou na připojeném zařízení vystavena.

Zjednodušeně: GATT řeší: „Kde jsou data a jak s nimi budu pracovat?“

V tomto programu tedy nejprve pomocí GAP najdeme a připojíme server a následně pomocí GATT najdeme požadovanou službu a charakteristiku.

Inicializace BLE

Na začátku program načte modul bluetooth a vytvoří BLE rozhraní:

ble = bluetooth.BLE()

Samotné vytvoření objektu ještě BLE nezapne. Aktivace přijde později:

ble.active(True)

Tím ESP32 získá aktivní Bluetooth Low Energy rozhraní. Důležitá je také registrace funkce irq():

ble.irq(irq)

MicroPython používá událostní model. BLE komunikace tedy neprobíhá tak, že bychom neustále ručně kontrolovali, zda se něco stalo. Bluetooth stack (zásobník) při jednotlivých událostech zavolá naši funkci irq().

Například:

  • nalezení zařízení při skenování,
  • dokončení skenování,
  • připojení k serveru,
  • nalezení služby,
  • nalezení charakteristiky,
  • dokončení zápisu,
  • odpojení.

Jednotlivé události jsou reprezentovány číselnými kódy, které si převedeme do pojmenovaných konstant, aby pak byl kód přehlednější – například:

_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_SERVICE_RESULT = const(9)
_IRQ_GATTC_SERVICE_DONE = const(10)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_WRITE_DONE = const(17)

Konstanta _IRQ_SCAN_RESULT zkrátka znamená, že skener právě přijal informace o nějakém nalezeném BLE zařízení.

Identifikace serveru

Může se stát, že v okolí našeho klienta se může nalézat více BT zařízení – modul ESP32 s naprogramovaným serverem, fitness náramek na našem zápěstí… apod. Jak poznáme, se kterým zařízením má náš klient komunikovat? Těch možností je hned několik. Je tu možnost třeba zkontrolovat MAC adresu serveru – to je skvělé, jednoznačné a fungující pro konkrétní čip ESP32. V případě výměny serveru, je třeba přeprogramovat všechny klienty. Dále je tu možnost pomocí jeho jména – to si zachovává svou určitou „jedinečnost“, ale není to vázáno na konkrétní hardware. Proč to tedy nezkusit? Trochu napovíme dopředu, že držet se pouze jména serveru není úplně ideální řešení, protože až nám klient detekuje dvě a více zařízení se jménem MojeESP, pochopíme, kde je asi problém této metody.

V tomto ukázkovém příkladu se ale tohoto způsobu budeme držet. Takže program zná jméno serveru:

SERVER_NAME = "ESP32-LED"

Podle tohoto jména klient při skenování pozná, ke kterému zařízení se má připojit. Jak jsme ale naznačili, pro samotnou komunikaci nestačí znát jméno zařízení. Po připojení musí klient najít konkrétní GATT službu a v ní konkrétní charakteristiku.

K tomu nám poslouží UUID:

SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")

Po aktivaci BLE začne klient hledat okolní zařízení:

ble.gap_scan(10000, 30000, 30000, True)

Příkaz ble.gap_scan(10000, 30000, 30000, True), který jsme již potkali v ukázce BLE scanner, v MicroPythonu slouží ke spuštění skenování okolních Bluetooth Low Energy (BLE) zařízení. Zatímco předtím jsme jej v tichosti přešli, nyní se na něj podíváme podrobněji – konkrétně na jeho parametry.

Jednotlivé parametry zleva doprava znamenají:

  • 10000 – Doba trvání skenování (duration_ms) – celková doba, po kterou bude skenování probíhat, v milisekundách (!), pak se automaticky zastaví. (hodnota 0 by znamenala nekonečné skenování a None skenování okamžitě ukončí)
  • 30000 – Interval skenování (interval_us) – jak často se skenování opakuje, v mikrosekundách (!).
  • 30000 – Skenovací okno (window_us) – jak dlouho uvnitř každého intervalu rádio fyzicky poslouchá, v mikrosekundách (pozn.: v našem kódu zařízení skenuje nepřetržitě 100 % času)
  • True – Aktivní skenování (active) – určuje typ skenování (True = aktivní, False = pasivní).
    • Při pasivním skenování (False) pouze odposloucháváš data, která zařízení sama vysílají.
    • aktivním skenování (True) zařízení po zachycení vysílání navíc pošle požadavek (SCAN_REQ) a reklamující zařízení mu obratem pošle doplňující data (SCAN_RSP), jako je např. úplný název zařízení (Bluetooth name).

Důležitý je zde především poslední parametr True. Ten znamená, že se má pracovat také se Scan Response pakety. To je praktické proto, že BLE zařízení nemusí mít své jméno přímo v hlavním advertising paketu. Může ho poslat až v navazující Scan Response.

A proč by to dělalo? No, to je lehká odpověď – protože se prostě do advertising paketu nevejde! Když jsme se na začátku zmiňovali o jednoznačné identifikaci BT zařízení, asi jsme už trochu cítili, že pouhé jméno pro tu jednoznačnou identifikaci nestačí. Z tohoto důvodu se do advertising paketu někdy přidává i UUID poskytované služby. A jak víme z předchozích článků, UUID může mít krátkou, ale i dlouhou variantou. A právě u té dlouhé varianty může být problém – dlouhé UUID může zabrat většinu advertising paketu, takže na celý název zařízení moc místa už nezbyde. Nezbývá tedy nic jiného, než název zařízení poslat až v navazující Scan Response. To však již vyžaduje aktivní skenování.

Podíváme se na advertising trochu podrobněji.

Už asi víme, že BLE zařízení, která jsou viditelná okolí, pravidelně vysílají takzvané advertising pakety. Klient je při skenování zachytává. Při každém nalezeném zařízení MicroPython zavolá:

irq(event, data)

kde:

event == _IRQ_SCAN_RESULT

Data mají podobu:

addr_type, addr, adv_type, rssi, adv_data = data

Program získá například:

  • typ BLE adresy,
  • Bluetooth MAC adresu,
  • typ advertisingu,
  • sílu signálu RSSI,
  • vlastní advertising data.

Právě adv_data obsahuje jednotlivé informační položky advertisingu. Advertising data nejsou obyčejný text. Jedná se o posloupnost položek, z nichž každá má strukturu přibližně:

[length] [type] [data...]

Proto program obsahuje funkci:

def decode_name(data):

Ta postupně prochází jednotlivé položky a hledá typ: 0x09 nebo 0x08. Tyto hodnoty označují:

  • 0x09 – úplný název zařízení,
  • 0x08 – zkrácený název zařízení.

Jakmile funkce takovou položku najde, převede její data na text:

return bytes(
    data[i + 2:i + 1 + length]
).decode("utf-8", "ignore")

Výsledkem může být například: ESP32-LED. Klient potom porovnává nalezený název s SERVER_NAME. Pokud klient při skenování najde zařízení se jménem: ESP32-LED, víme, že se máme připojit.

if name == SERVER_NAME and not connecting:

Před samotným připojením však ještě nejdříve zastaví další skenování:

ble.gap_scan(None)

To je důležité. Zařízení už bylo nalezeno a klient se nyní chce připojit. Až potom se zavolá:

ble.gap_connect(
    addr_type,
    bytes(addr)
)

Tím se zahájí BLE spojení s nalezeným zařízením.

Proměnná: connecting = True zároveň zabraňuje tomu, aby program během navazování spojení opakovaně reagoval na stejné zařízení.

Po ble.gap_connect(...) je zařízení sice fyzicky připojeno přes BLE, ale klient ještě neví, kam má zapisovat data.

BLE používá nad samotným spojením protokol GATT. Server má určitou strukturu. Klient proto musí po připojení zjistit, kde se požadovaná služba a charakteristika nachází.

Když dojde k připojení, přijde událost:

_IRQ_PERIPHERAL_CONNECT

a program uloží identifikátor spojení:

conn_handle, addr_type, addr = data

conn_handle je velmi důležitý. Jedná se o handle aktuálního BLE spojení, pomocí kterého MicroPython později ví, ke kterému připojenému zařízení má GATT operaci provést.

Po připojení klient zahájí hledání služby:

ble.gattc_discover_services(
    conn_handle,
    SERVICE_UUID
)

Zde už se nepoužívá jméno ESP32-LED. Jméno sloužilo pouze k nalezení správného BLE zařízení během scanu. Nyní se používá jednoznačné UUID služby: 6E400001-B5A3-F393-E0A9-E50E24DCCA9E

Klient tedy říká přibližně: „Jsem připojen k tomuto zařízení. Najdi mi v něm službu s tímto UUID.“

Pokud ji BLE stack najde, vyvolá:

_IRQ_GATTC_SERVICE_RESULT

Program z události získá:

conn, start, end, uuid = data

a uloží si:

service_start = start
service_end = end

Tyto hodnoty představují rozsah handleů patřících nalezené službě.

Jakmile je dokončeno hledání služby, přijde:

_IRQ_GATTC_SERVICE_DONE

Pokud byla služba nalezena, program pokračuje:

ble.gattc_discover_characteristics(
    conn_handle,
    service_start,
    service_end,
    CHAR_UUID
)

Tentokrát už hledá konkrétní charakteristiku: 6E400002-B5A3-F393-E0A9-E50E24DCCA9E

Její nalezení vyvolá:

_IRQ_GATTC_CHARACTERISTIC_RESULT

Z události získáme:

conn, def_handle, value_handle, properties, uuid = data

Nejdůležitější je hodnota:

value_handle

Tím program získá handle na konkrétní paměťové místo dané charakteristiky, proto si tuto hodnotu program uloží:

char_value_handle = value_handle

A právě tento handle bude později použit při zápisu hodnoty.

UUID je lidsky i programátorsky dobře použitelný identifikátor charakteristiky, ale při vlastní GATT komunikaci se pracuje také s takzvanými handle. Můžeme si to zjednodušeně představit takto: Program nejprve pomocí UUID zjistí, kde charakteristika je, a potom používá její handle pro vlastní operace. To je důvod, proč se do gattc_write() nepředává UUID charakteristiky, ale char_value_handle.

Jakmile je známe conn_handle a char_value_handle může klient konečně zapisovat hodnotu do charakteristiky.

Například při stisknutí tlačítka:

ble.gattc_write(
    conn_handle,
    char_value_handle,
    b"1",
    1
)

conn_handle určuje spojení se zařízením, char_value_handle konkrétní místo charakteristiky. Server tak do své charakteristiky GATT dostane hodnotu: 1

Při uvolnění tlačítka se obdobně odešle hodnota 0:

ble.gattc_write(
    conn_handle,
    char_value_handle,
    b"0",
    1
)

V tomto případě se posílají bajty: b"1" a b"0". Nejde tedy o Pythonovské integer hodnoty 1 a 0, ale o jednobajtová data obsahující ASCII znaky "1" a "0". To souvisí s tím, jak máme tyto hodnoty nastavené na straně serveru – server musí samozřejmě vědět, jak tato data interpretovat!

U předešlého zápisu jsme se nezmínili o poslední parametru, který určuje způsob zápisu. Hodnota 1 znamená write without response. Klient tedy provede zápis bez požadavku na potvrzení na úrovni GATT.

Program ale zároveň obsluhuje událost:

_IRQ_GATTC_WRITE_DONE

a podle status zjistí, zda MicroPython BLE zásobník zápis dokončil úspěšně:

if status == 0:
    print("Hodnota uspesne odeslana")

Je dobré si uvědomit, že jde o dvě různé věci: vlastní režim GATT zápisu a následnou událost, kterou BLE zásobník oznámí dokončení operace.

Tím máme skoro hotovo!

Pochopitelně musíme ještě dořešit samotné sledování tlačítka. Tlačítko je připojeno na GPIO0:

button = Pin(0, Pin.IN, Pin.PULL_UP)

Používá se interní pull-up rezistor. Proto je logika tlačítka obrácená: uvolněno → 1 a stisknuto → 0.

Program si nejprve uloží aktuální stav:

last_state = button.value()

a potom v nekonečné smyčce čte nový stav:

state = button.value()

Změnu pozná pomocí:

if state != last_state:

Tím vzniká jednoduchá detekce změny stavu.

Pokud je state == 0 tlačítko bylo stisknuto. Klient okamžitě rozsvítí svou vlastní LED:

led.value(1)

a pokud je současně navázáno BLE spojení a byla nalezena charakteristika:

if (
    conn_handle is not None
    and char_value_handle is not None
):

Odešle se hodnota: b"1"

Server potom může na základě této hodnoty rozsvítit svou LED.

Při návratu vstupu do stavu 1 program LED klienta zhasne:

led.value(0)

a serveru odešle hodnota: b"0"

Co se stane při odpojení

BLE spojení nemusí být trvalé. Zařízení může být vypnuto, vzdálit se mimo dosah nebo může dojít k jiné chybě spojení. V takovém případě přijde: _IRQ_PERIPHERAL_DISCONNECT

Program vynuluje všechny důležité handly:

conn_handle = None
service_start = None
service_end = None
char_value_handle = None

To je důležité, protože staré hodnoty už po odpojení nemají být používány.

Program také zhasne vlastní LED:

led.value(0)

a znovu spustí skenování:

ble.gap_scan(
    10000,
    30000,
    30000,
    True
)

Klient se tak dokáže k serveru znovu automaticky připojit.

A nyní s aioble:

Předchozí nízkoúrovňové řešení nám dalo, jak AI ráda ve svých textech říká, nahlédnout „pod kapotu“ BT klienta. Ale tak jako dnes na složitější výpočty vezmeme kalkulačku, asi i zde by to chtělo nějaký nástroj, který by nám určitou „námahu“ ušetřil. Tímto naším pomocníkem může být knihovna aioble, které jsme ohledně jejího použití věnovali celý jeden článek. Asi i výše uvedený příklad BLE scanneru ukázal, její přínos.

Předchozí program řešený pomocí modulu aioble by tedy mohl vypadat kupříkladu takto:

import asyncio
import aioble
import bluetooth
from machine import Pin

# UUID služby a charakteristiky
SERVICE_UUID = bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")
CHAR_UUID = bluetooth.UUID("6E400002-B5A3-F393-E0A9-E50E24DCCA9E")

# Jméno BLE serveru
SERVER_NAME = "ESP32-LED"

# BOOT tlačítko
button = Pin(0, Pin.IN, Pin.PULL_UP)

# Vestavěná LED klienta
led = Pin(2, Pin.OUT)
led.value(0)

# Stav BLE
connection = None
write_char = None

# Formátování BLE adresy
def format_mac(addr):
    return ":".join(
        "{:02X}".format(b)
        for b in addr
    )

# Hlavní BLE úloha
async def ble_task():
    global connection
    global write_char

    while True:
        connection = None
        write_char = None

        print()
        print("================================")
        print("HLEDAM BLE SERVER")
        print("================================")
        print("Hledam:", SERVER_NAME)
        print()

        # SCAN
        device = None
        async with aioble.scan(
            duration_ms=10000,
            interval_us=30000,
            window_us=30000,
            active=True
        ) as scanner:

            async for result in scanner:
                name = result.name()
                # Výpis nalezených zařízení se jménem
                if name:
                    mac = format_mac(result.device.addr)
                    print(
                        "Nalezeno:",
                        name,
                        "|",
                        mac,
                        "| RSSI:",
                        result.rssi
                    )

                # Hledáme náš server
                if name == SERVER_NAME:
                    device = result.device
                    mac = format_mac(
                        device.addr
                    )

                    print()
                    print("--------------------------------")
                    print("Nalezen BLE server")
                    print("Jmeno:", name)
                    print("MAC  :", mac)
                    print("RSSI :", result.rssi, "dBm")
                    print("--------------------------------")

                    # Server máme.
                    # Scan ukončíme opuštěním async with.
                    break

        # SERVER NENALEZEN
        if device is None:
            print()
            print("Scan dokoncen, server nenalezen")

            # Malá pauza před dalším skenem
            await asyncio.sleep_ms(100)

            continue

        # PŘIPOJENÍ
        try:
            print()
            print("Pripojuji se...")

            connection = await device.connect()

            print()
            print("================================")
            print("BLE server PRIPOJEN")
            print("================================")

            # HLEDÁNÍ SLUŽBY
            print("Hledam BLE sluzbu...")
            service = await connection.service(
                SERVICE_UUID
            )

            if service is None:
                print("BLE sluzba nenalezena")
                await connection.disconnect()

                connection = None

                continue

            print("BLE sluzba nalezena")

            # HLEDÁNÍ WRITE CHARAKTERISTIKY
            print("Hledam WRITE charakteristiku...")
            write_char = await service.characteristic(
                CHAR_UUID
            )

            if write_char is None:
                print("WRITE charakteristika nenalezena")

                await connection.disconnect()

                connection = None

                continue

            print("WRITE charakteristika nalezena")
            print()
            print("Cekam na tlacitko BOOT...")

            # ČEKÁNÍ NA ODPOJENÍ
            # BLE připojení držíme aktivní.
            # Obsluha tlačítka běží v jiné asyncio úloze.
            while connection.is_connected():
                await asyncio.sleep_ms(100)

        except Exception as e:
            print()
            print("BLE chyba:", e)

        # ODPOJENÍ / CHYBA
        print()
        print("BLE server ODPOJEN")

        connection = None
        write_char = None

        led.value(0)

        print("Znovu hledam BLE server...")

        await asyncio.sleep_ms(100)

# Odeslání hodnoty do BLE charakteristiky
async def send_value(value):
    global connection
    global write_char

    if connection is None:
        return

    if write_char is None:
        return

    if not connection.is_connected():
        return

    try:
        await write_char.write(
            value,
            response=True
        )

        print("Odeslano BLE:", value.decode())

    except Exception as e:
        print("Chyba pri zapisu:", e)

# Obsluha tlačítka
async def button_task():
    last_state = button.value()

    while True:
        state = button.value()

        if state != last_state:
            last_state = state

            # BOOT STISKNUTO
            if state == 0:
                print()
                print("TLACITKO BOOT: STISKNUTO")

                # LED na klientovi
                led.value(1)

                # Serveru pošleme "1"
                await send_value(b"1")

            # BOOT UVOLNENO
            else:
                print()
                print("TLACITKO BOOT: UVOLNENO")

                # LED na klientovi
                led.value(0)

                # Serveru pošleme "0"
                await send_value(b"0")

        await asyncio.sleep_ms(20)

# HLAVNÍ PROGRAM
async def main():
    print()
    print("================================")
    print("SPOUSTIM BLE KLIENTA")
    print("================================")
    print("Hledam:", SERVER_NAME)
    print()

    # BLE scanner / klient
    asyncio.create_task(ble_task())

    # Tlačítko
    asyncio.create_task(button_task())

    # Hlavní úloha musí zůstat aktivní
    while True:
        await asyncio.sleep_ms(1000)

# START
asyncio.run(main())

Program postavený na knihovně aioble zásadně mění architekturu aplikace z událostmi řízeného modelu (callbacky) na lineární asynchronní tok, který je mnohem čitelnější a snazší na údržbu.

  • Paralelní zpracování úloh: Díky modulu asyncio program v hlavní funkci main() vytvoří dvě nezávislé úlohy pomocí asyncio.create_task(). Úloha ble_task se stará výhradně o vyhledání serveru a udržování spojení, zatímco button_task běží paralelně a monitoruje fyzický stav tlačítka.
  • Efektivní skenování a identifikace: Skenování probíhá v kontextovém manažeru async with aioble.scan, který automaticky zajistí správné spuštění a ukončení skeneru. Program v asynchronní smyčce prochází výsledky a pomocí metody result.name() okamžitě zjišťuje jméno zařízení, aniž by musel ručně dekódovat binární strukturu advertising paketů.
  • Abstrakce GATT operací: Program se již nestará o číselné handly spojení nebo charakteristik; místo toho pracuje s objekty connection, service a write_char.
  • Zápis hodnot: Samotné odeslání dat (např. "1" při stisku) probíhá asynchronně metodou write_char.write(). Parametr response=True zajistí, že program počká na potvrzení o úspěšném přijetí dat druhou stranou.
BLE klient - LED
Obrázek č. 3 – Ukázka výstupu na straně BLE klienta (zápis do charakteristiky)

PŘÍKLAD č. 2: Kliente, detekuj stav tlačítka na serveru

Pokud jsme v předešlém programu simulovali jakési dálkové ovládání, které zasílá na server data. Server tedy čeká na data, podle kterých může vykonávat nějakou činnost, nyní bychom chtěli získávat data obráceně. Server tedy bude své stavy (data) aktivně „vnucovat“ připojenému klientovi.

V následujícím příkladu tedy budeme načítat stav GPIO0, na kterém je připojeno vestavěné tlačítko BOOT. Právě jeho stiskáváním budeme simulovat nějaké události na straně serveru a budeme chtít tento stav znát i na straně klienta.

Princip naší ukázky je následující:

  1. Pokud stiskneme na serveru tlačítko BOOT (dojde ke změně stavu), nastaví server jako notifikaci hodnotu charakteristiky na 1.
  2. Na změnu hodnoty charakteristiky (notify) zareaguje klient a podle hodnoty 1 rozsvítí svou LED.
  3. Po uvolnění tlačítka na straně serveru, dojde opět k odeslání notifikace – hodnota se změní na 0.
  4. Klient dle nové hodnoty charakteristiky zhasne svou LED.

Princip našeho BLE klienta a jeho komunikace s BLE serverem znázorňuje animace na obrázku 4:

BLE klient - priklad 2
Obrázek č. 4 – Animace principu BLE klienta a jeho komunikace s  BLE serverem (příklad 2)

Klient tedy bude fungovat jako GATT klient přijímající notifikace, zatímco server bude změny aktivně oznamovat.

Určitou novou částí programu je postup:

najdi charakteristiku
        ⇓
najdi CCCD descriptor
        ⇓
zapiš 0x0100 do CCCD
        ⇓
NOTIFY je povoleno
        ⇓
server odešle NOTIFY
        ⇓
_IRQ_GATTC_NOTIFY
        ⇓
změň LED

Opět nejdříve uvedeme celý kód programu:

import time
import bluetooth
from micropython import const
from machine import Pin

# BLE události
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
_IRQ_PERIPHERAL_CONNECT = const(7)
_IRQ_PERIPHERAL_DISCONNECT = const(8)
_IRQ_GATTC_SERVICE_RESULT = const(9)
_IRQ_GATTC_SERVICE_DONE = const(10)
_IRQ_GATTC_CHARACTERISTIC_RESULT = const(11)
_IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
_IRQ_GATTC_DESCRIPTOR_RESULT = const(13)
_IRQ_GATTC_DESCRIPTOR_DONE = const(14)
_IRQ_GATTC_WRITE_DONE = const(17)
_IRQ_GATTC_NOTIFY = const(18)

# UUID zařízení
SERVICE_UUID = bluetooth.UUID("a1b2c300-1234-5678-9abc-def012345678")
CHAR_UUID = bluetooth.UUID("a1b2c301-1234-5678-9abc-def012345678")

# UUID služby v advertising paketu.
# BLE ukládá 128bit UUID v paketu v little-endian pořadí.
SERVICE_UUID_ADV = bytes([
    0x78, 0x56, 0x34, 0x12,
    0xf0, 0xde,
    0xbc, 0x9a,
    0x78, 0x56,
    0x34, 0x12,
    0x00, 0xc3, 0xb2, 0xa1
])

# LED
led = Pin(2, Pin.OUT)
led.value(0)

ble = bluetooth.BLE()

conn_handle = None
service_start = None
service_end = None
char_value_handle = None
cccd_handle = None

# Zjistí, zda advertising obsahuje naši službu
def has_service(data):
    i = 0
    while i < len(data):
        length = data[i]
        if length == 0:
            break

        ad_type = data[i + 1]
        # 0x06 / 0x07 = seznam 128bitových UUID
        if ad_type == 0x06 or ad_type == 0x07:
            if SERVICE_UUID_ADV in bytes(
                data[i + 2:i + 1 + length]
            ):
                return True

        i += length + 1

    return False
# BLE callback
def irq(event, data):
    global conn_handle
    global service_start, service_end
    global char_value_handle
    global cccd_handle

    # Nalezeno zařízení
    if event == _IRQ_SCAN_RESULT:
        addr_type, addr, adv_type, rssi, adv_data = data
        if has_service(adv_data):
            # 0 = ADV_IND = connectable + scannable
            if adv_type != 0:
                return

            mac = ':'.join(
                '%02X' % b for b in addr
            )

            print()
            print("--------------------------------")
            print("Nalezen BLE server")
            print("MAC :", mac)
            print("RSSI:", rssi, "dBm")
            print("--------------------------------")

            ble.gap_scan(None)

            print("Pripojuji se...")

            # Adresu z IRQ je vhodné zkopírovat.
            ble.gap_connect(addr_type, bytes(addr))

    # Připojeno
    elif event == _IRQ_PERIPHERAL_CONNECT:
        conn_handle, addr_type, addr = data
        print()
        print("================================")
        print("BLE server PRIPOJEN")
        print("================================")
        print("Hledam BLE sluzbu...")

        ble.gattc_discover_services(conn_handle, SERVICE_UUID)

    # Nalezena služba
    elif event == _IRQ_GATTC_SERVICE_RESULT:
        conn, start, end, uuid = data
        if conn == conn_handle and uuid == SERVICE_UUID:
            service_start = start
            service_end = end

            print("BLE sluzba nalezena")

    elif event == _IRQ_GATTC_SERVICE_DONE:
        if service_start is not None:
            print("Hledam charakteristiku...")

            ble.gattc_discover_characteristics(
                conn_handle,
                service_start,
                service_end,
                CHAR_UUID
            )

    # Nalezena charakteristika
    elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
        conn, def_handle, value_handle, properties, uuid = data
        if conn == conn_handle and uuid == CHAR_UUID:
            char_value_handle = value_handle
            print("Notify charakteristika nalezena")
            # Najdeme CCCD descriptor (UUID 0x2902).
            ble.gattc_discover_descriptors(
                conn_handle,
                char_value_handle,
                service_end
            )

    elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
        pass

    # Nalezen CCCD
    elif event == _IRQ_GATTC_DESCRIPTOR_RESULT:
        conn, handle, uuid = data
        if conn == conn_handle and uuid == bluetooth.UUID(0x2902):
            cccd_handle = handle
            # =================================================
            # DŮLEŽITÉ:
            # PROPERTY_NOTIFY samo o sobě nestačí.
            # Klient musí do CCCD zapsat 0x0100.
            # 01 00 = Enable Notification
            # Bez tohoto zápisu server sice může zavolat
            # notify(), ale klient Notify nepřijme.
            # =================================================
            ble.gattc_write(
                conn_handle,
                cccd_handle,
                b"\x01\x00",
                1
            )

    elif event == _IRQ_GATTC_DESCRIPTOR_DONE:
        pass

    # Potvrzení zápisu CCCD
    elif event == _IRQ_GATTC_WRITE_DONE:
        conn, handle, status = data

        if status == 0:
            print()
            print(">>> NOTIFY POVOLENO <<<")
            print("Cekam na stisk tlacitka...")

    # Přijatý Notify
    elif event == _IRQ_GATTC_NOTIFY:
        conn, handle, value = data
        if conn == conn_handle and handle == char_value_handle:
            if len(value):
                # 0 = tlačítko uvolněno
                # 1 = tlačítko stisknuto
                state = value[0]
                print()

                if state == 1:
                    print("TLAČÍTKO: STISKNUTO")
                    print("Přijato Notify: 1")

                    led.value(1)

                else:
                    print("TLAČÍTKO: UVOLNĚNO")
                    print("Přijato Notify: 0")

                    led.value(0)

    # Odpojeno
    elif event == _IRQ_PERIPHERAL_DISCONNECT:
        print()
        print("BLE server ODPOJEN")

        conn_handle = None
        service_start = None
        service_end = None
        char_value_handle = None
        cccd_handle = None

        led.value(0)

        print("Znovu hledam BLE server...")

        ble.gap_scan(
            10000,
            30000,
            30000,
            True
        )

# START
ble.active(True)
ble.irq(irq)

print()
print("================================")
print("SPOUSTIM BLE KLIENTA")
print("================================")
print("Hledam BLE server...")

# Aktivní scan = přijímá také Scan Response.
ble.gap_scan(10000, 30000, 30000, True)

# Hlavní smyčka
while True:
    time.sleep(1)

Co je na tomto programu nové

Základní část BLE komunikace zůstává stejná jako v předchozím programu:

  1. aktivace BLE,
  2. scan,
  3. nalezení serveru,
  4. připojení přes GAP,
  5. nalezení GATT služby,
  6. nalezení charakteristiky.

Nově ale po nalezení charakteristiky nestačí říct: „Charakteristika podporuje NOTIFY, takže už budu dostávat data.“ To totiž nestačí. Klient musí serveru prostřednictvím speciálního descriptoru říct: „Chci odebírat notifikace z této charakteristiky.“ K tomu slouží tzv. CCCDClient Characteristic Configuration Descriptor.

Oproti předchozímu programu zde přibyly především čtyři důležité události:

_IRQ_GATTC_CHARACTERISTIC_DONE = const(12)
_IRQ_GATTC_DESCRIPTOR_RESULT = const(13)
_IRQ_GATTC_DESCRIPTOR_DONE = const(14)
_IRQ_GATTC_NOTIFY = const(18)

Pro pochopení NOTIFY jsou nejdůležitější: _IRQ_GATTC_DESCRIPTOR_RESULT a _IRQ_GATTC_NOTIFY. První oznámí, že byl nalezen požadovaný descriptor. Druhá oznámí, že server právě poslal notifikaci.

Identifikace serveru pomocí UUID služby

V předchozím programu klient hledal server podle jeho názvu: SERVER_NAME = "ESP32-LED". Už tam jsme naznačovali, že to není úplně ideální. Zkusíme tedy v tomto programu tento proces pojmout trochu „profesionálněji“. Zkusíme spárovat naše zařízení podle jedinečného identifikátoru UUID použité služby. Pokud se se serverem dokážeme shodnout na jednoznačném UUID je asi jasné, že tenhle páreček server-klient k sobě zkrátka patří!

Funkce has_service(data) tedy kontroluje přímo obsah advertising paketu a hledá v něm UUID služby. Pochopitelně v tom případě musí být tato služba advertising paketu nabízena! Takže pokud se podíváme na náš nízkoúrovňový příklad z prvního článku Nízkoúrovňové BLE v MicroPython, kde jsme advertising trochu „odflákli“ a funkci neregistrovali, máme smůlu! Následující program tedy bude komunikovat jen s programy serverů, ze článku o knihovně aioble: Moderní BLE s modulem aioble nebo o BLE v Arduino IDE – Začínáme s ESP32 Bluetooth Low Energy (BLE).

Následující program má pro advertising připravenou speciální podobu UUID:

SERVICE_UUID_ADV = bytes([
    0x78, 0x56, 0x34, 0x12,
    0xf0, 0xde,
    0xbc, 0x9a,
    0x78, 0x56,
    0x34, 0x12,
    0x00, 0xc3, 0xb2, 0xa1
])

Máte pocit, že tam je uložena úplně jiná hodnota UUID, než je a1b2c300-1234-5678-9abc-def012345678? Důvodem je little-endian uložení 128bitového UUID v BLE advertising datech. Data jsou uložena pozpátku – to je důležitý detail!

Také si všimněte, že UUID není v advertising paketu zapsáno jako textová posloupnost bajtů. Proto je v programu zvlášť proměnná SERVICE_UUID pro GATT komunikaci a SERVICE_UUID_ADV pro hledání služby v advertising datech.

Již zmíněná funkce:

def has_service(data):

prochází advertising data podobně jako předchozí funkce decode_name(). Tentokrát ale nehledá typ 0x08/0x09 pro název zařízení. Hledá 0x06 nebo 0x07. Tyto hodnoty označují seznam 128bitových UUID služeb.

Program potom provede:

if SERVICE_UUID_ADV in bytes(
    data[i + 2:i + 1 + length]
):
    return True

Tím se zjišťuje: „Obsahuje advertising paket UUID služby, kterou hledám?“ Pokud ano, zařízení je považováno za hledaný server.

Po nalezení služby program ještě kontroluje:

if adv_type != 0:
   return

Komentář v programu říká: 0 = ADV_IND = connectable + scannable. Klient tedy nechce reagovat na libovolný paket obsahující správné UUID. Potřebuje zařízení, ke kterému se může připojit. Tím se zamezí například situaci, kdy klient zachytí relevantní advertising data ze zařízení, které není vhodné pro následné připojení.

Od připojení až k charakteristice je princip stejný…

Po nalezení zařízení následuje:

ble.gap_scan(None)

a:

ble.gap_connect(
    addr_type,
    bytes(addr)
)

Po připojení: _IRQ_PERIPHERAL_CONNECT se provede nalezení služby:

ble.gattc_discover_services(
    conn_handle,
    SERVICE_UUID
)

Po jejím nalezení se vyhledá charakteristika:

ble.gattc_discover_characteristics(
    conn_handle,
    service_start,
    service_end,
    CHAR_UUID
)

Tyto části fungují stejně jako v předchozím programu, takže podstatná novinka začíná až po nalezení charakteristiky.

Po nalezení charakteristiky následuje důležitý krok.

Při: _IRQ_GATTC_CHARACTERISTIC_RESULT program získá:

conn, def_handle, value_handle, properties, uuid = data

a uloží:

char_value_handle = value_handle

Tady je důležitá proměnná properties. Ta obsahuje informace o vlastnostech charakteristiky.

Mezi možné vlastnosti patří například:

READ
WRITE
NOTIFY
INDICATE

V tomto programu očekáváme, že charakteristika podporuje NOTIFY. Ale samotná vlastnost NOTIFY ještě neznamená, že klient začne automaticky dostávat data. A právě zde přichází CCCD.

Co je CCCD

CCCD znamená Client Characteristic Configuration Descriptor. Jedná se o speciální GATT descriptor, který umožňuje klientovi nastavit, zda chce od dané charakteristiky dostávat:

  • notifikace,
  • indikace,
  • případně je vypnout.

Jeho UUID je: 0x2902. Program proto po nalezení charakteristiky začne hledat deskriptory:

ble.gattc_discover_descriptors(
    conn_handle,
    char_value_handle,
    service_end
)

Tím říká přibližně: „Prohledej deskriptory za touto charakteristikou a najdi ty, které patří k této části GATT databáze.“

Při: _IRQ_GATTC_DESCRIPTOR_RESULT program dostane:

conn, handle, uuid = data

a kontroluje:

if conn == conn_handle and uuid == bluetooth.UUID(0x2902):

Pokud UUID odpovídá 0x2902, našel právě CCCD. Jeho handle uloží:

cccd_handle = handle

Takže máme nyní dva důležité handly:

  • char_value_handle – samotná hodnota charakteristiky
  • cccd_handle – nastavení odběru NOTIFY

NOTIFY se musí nejprve povolit

Po nalezení CCCD program provede:

ble.gattc_write(
    conn_handle,
    cccd_handle,
    b"\x01\x00",
    1
)

Tento zápis je klíčovou částí celého programu. Do CCCD se zapisuje: 01 00, což znamená: Enable Notification. Klient tím serveru nastaví: „Chci dostávat NOTIFY z této charakteristiky.“ Toto je asi nejdůležitější koncept celého příkladu. Na straně serveru může charakteristika deklarovat: NOTIFY. To znamená: „Tato charakteristika podporuje mechanismus notifikací.“ Ale klient musí ještě odběr aktivovat!

CCCD je dvoubajtová hodnota. Pro NOTIFY se používá: 01 00. Druhý bit se používá pro indikace: 0x02 0x00.

Zjednodušeně tedy:

01 00 → Notification ON
02 00 → Indication ON
00 00 → vypnuto

Teprve potom může server začít posílat NOTIFY. Respektive, bez zápisu: 01 00 do CCCD může server zavolat vlastní notify(), ale klient nebude notifikace standardním způsobem přijímat.

Co přesně znamená 01 00?

CCCD je dvoubajtová hodnota. Pro NOTIFY se používá: 01 00. Druhý bit se používá pro indikace: 0x02 0x00.

Je zajímavé, že program zde používá gattc_write(), přestože hlavním účelem programu je přijímat NOTIFY. Důvodem je, že samotné NOTIFY není datový zápis klienta. Zápis do CCCD je pouze konfigurační operace. Takže gattc_write() se zde používá pouze pro povolení NOTIFY.

Potvrzení konfigurace

Po zápisu do CCCD přijde událost: _IRQ_GATTC_WRITE_DONE

Program kontroluje:

if status == 0:

a vypíše:

>>> NOTIFY POVOLENO <<<
Cekam na stisk tlacitka...

Od této chvíle je klient připraven přijímat notifikace. Je to tedy prakticky konec inicializační fáze BLE.

Jakmile server zjistí například změnu svého tlačítka, může poslat klientovi notifikaci. Na straně klienta se tato událost projeví:

_IRQ_GATTC_NOTIFY

To je nejdůležitější nová BLE událost tohoto programu. Callback dostane:

conn, handle, value = data

Program ověří:

if conn == conn_handle and handle == char_value_handle:

Tím se ujistí, že:

  • notifikace přišla z našeho BLE spojení,
  • pochází z naší charakteristiky.

Přijatá data jsou ve value. Program nejprve kontroluje:

if len(value):

tedy zda vůbec nějaká data dorazila.

Potom vezme první bajt:

state = value[0]

Server posílá jednoduchou hodnotu: 0 (tlačítko uvolněno), 1 (tlačítko stisknuto).

Pokud přijde: 1, klient rozsvítí LED led.value(1), pokud přijde 0, LED zhasne: led.value(0).

V programu jsou zachyceny i tyto události:

elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
    pass

a

elif event == _IRQ_GATTC_DESCRIPTOR_DONE:
    pass

Zde se jejich událost pouze zachytává, ale program na ně nemusí reagovat. Důvod je, že pro tento konkrétní scénář je rozhodující informace získaná už během: _CHARACTERISTIC_RESULT a _DESCRIPTOR_RESULT.

Události _DONE pouze oznamují dokončení daného discovery procesu. Jejich obsluha by byla užitečná například v robustnější implementaci, kde bychom chtěli explicitně kontrolovat, zda hledání skutečně skončilo a zda byl požadovaný objekt nalezen.

Stejně jako v předchozím programu je potřeba při odpojení zneplatnit staré handly:

conn_handle = None
service_start = None
service_end = None
char_value_handle = None
cccd_handle = None

Tentokrát je navíc důležité vynulovat cccd_handle, protože po novém připojení může mít CCCD jiný handle.

Program poté zhasne LED led.value(0) a znovu zahájí scan.

BLE klient - TLACITKO
Obrázek č. 5 – Ukázka výstupu na straně BLE klienta (příjem notifikace)

Jak to bude s aioble?

I tento kód si zkusíme vytvořit pomocí modulu aioble:

import asyncio
import aioble
import bluetooth
from machine import Pin

# UUID zařízení
SERVICE_UUID = bluetooth.UUID("a1b2c300-1234-5678-9abc-def012345678")
CHAR_UUID = bluetooth.UUID("a1b2c301-1234-5678-9abc-def012345678")

# UUID služby v advertising paketu
# BLE ukládá 128bit UUID v paketu v little-endian pořadí.
SERVICE_UUID_ADV = bytes([
    0x78, 0x56, 0x34, 0x12,
    0xf0, 0xde,
    0xbc, 0x9a,
    0x78, 0x56,
    0x34, 0x12,
    0x00, 0xc3, 0xb2, 0xa1
])

# LED
led = Pin(2, Pin.OUT)
led.value(0)

# Stav BLE
connection = None
notify_char = None

# Formátování BLE adresy
def format_mac(addr):
    return ":".join(
        "{:02X}".format(b)
        for b in addr
    )

# Kontrola advertising dat
# Hledáme 128bit UUID naší služby.
# 0x06 = Incomplete List of 128-bit Service UUIDs
# 0x07 = Complete List of 128-bit Service UUIDs
def has_service(data):
    # Některé ScanResult mohou mít adv_data = None.
    if data is None:
        return False

    i = 0
    while i < len(data):
        length = data[i]

        # Konec advertising dat
        if length == 0:
            break

        # Ochrana před poškozeným packetem
        if i + length >= len(data):
            break

        ad_type = data[i + 1]

        if ad_type == 0x06 or ad_type == 0x07:
            uuid_data = bytes(
                data[
                    i + 2:
                    i + 1 + length
                ]
            )

            if SERVICE_UUID_ADV in uuid_data:
                return True

        # Přesun na další AD strukturu
        i += length + 1

    return False

# BLE úloha
async def ble_task():
    global connection
    global notify_char

    while True:
        connection = None
        notify_char = None

        led.value(0)

        # SCAN
        print()
        print("================================")
        print("HLEDAM BLE SERVER")
        print("================================")
        print("Hledam BLE server...")
        print()

        device = None

        async with aioble.scan(
            duration_ms=10000,
            interval_us=30000,
            window_us=30000,
            active=True
        ) as scanner:

            async for result in scanner:
                # Některé verze aioble mohou mít adv_data = None.
                if not has_service(
                    result.adv_data
                ):
                    continue

                # Našli jsme zařízení, které v advertisingu
                # obsahuje naši službu.
                # Už nekontrolujeme adv_type, protože naše verze
                # aioble tento atribut v ScanResult nemá.
                device = result.device

                mac = format_mac(
                    device.addr
                )

                print()
                print("--------------------------------")
                print("Nalezen BLE server")
                print("MAC :", mac)
                print("RSSI:", result.rssi, "dBm")
                print("--------------------------------")

                # Server nalezen.
                # Opustíme async with a tím ukončíme scan.

                break

        # SERVER NENALEZEN
        if device is None:
            print()
            print("Scan dokoncen, server nenalezen")

            await asyncio.sleep_ms(100)

            continue

        # PŘIPOJENÍ
        try:
            print("Pripojuji se...")

            connection = await device.connect()

            print()
            print("================================")
            print("BLE server PRIPOJEN")
            print("================================")

            # HLEDÁNÍ SLUŽBY
            print("Hledam BLE sluzbu...")
            service = await connection.service(SERVICE_UUID)

            if service is None:
                print("BLE sluzba nenalezena")

                await connection.disconnect()

                connection = None

                continue

            print("BLE sluzba nalezena")

            # HLEDÁNÍ CHARAKTERISTIKY
            print("Hledam charakteristiku...")

            notify_char = await service.characteristic(CHAR_UUID)

            if notify_char is None:
                print("Notify charakteristika nenalezena")

                await connection.disconnect()

                connection = None

                continue

            print("Notify charakteristika nalezena")

            # POVOLENÍ NOTIFIKACÍ
            print("Povoluji NOTIFY...")

            await notify_char.subscribe(
                notify=True
            )

            print()
            print(">>> NOTIFY POVOLENO <<<")
            print("Cekam na stisk tlacitka...")

            # ČEKÁNÍ NA NOTIFY
            while connection.is_connected():
                try:
                    value = await notify_char.notified()

                    if len(value) == 0:
                        continue

                    # Hodnota 0 = uvolněno
                    # Hodnota 1 = stisknuto
                    state = value[0]

                    print()

                    if state == 1:
                        print("TLACITKO: STISKNUTO")
                        print("Prijato Notify: 1")

                        led.value(1)
                    else:
                        print("TLACITKO: UVOLNENO")
                        print("Prijato Notify: 0")

                        led.value(0)

                except Exception as e:
                    print("Chyba pri prijmu Notify:", e)
                    break

        except Exception as e:
            print()
            print("BLE chyba:", e)

        # ODPOJENÍ
        print()
        print("BLE server ODPOJEN")

        connection = None
        notify_char = None

        led.value(0)

        print("Znovu hledam BLE server...")

        await asyncio.sleep_ms(100)

# HLAVNÍ PROGRAM
async def main():
    print()
    print("================================")
    print("SPOUSTIM BLE KLIENTA")
    print("================================")
    print("Hledam BLE server...")
    print()

    # Spustíme BLE úlohu
    asyncio.create_task(ble_task())

    # Hlavní úloha
    while True:
        await asyncio.sleep_ms(1000)

# START
asyncio.run(main())

V tomto příkladu aioble demonstruje svou sílu v pokročilé práci s GATT databází a automatizaci konfigurace pro příjem dat (notifikací).

  • Profesionální párování přes UUID: Na rozdíl od prvního příkladu tento program nehledá server podle jména, ale podle UUID služby, kterou server nabízí. Pomocná funkce has_service prohledává advertising data (result.adv_data) a zjišťuje, zda zařízení inzeruje konkrétní 128bitové UUID. Pokud ano, klient se k němu pokusí připojit.
  • Automatizace CCCD deskriptoru: Pro příjem notifikací musí klient v nízkoúrovňovém BLE ručně najít deskriptor CCCD (UUID 0x2902) a zapsat do něj hodnotu 0x0100. Knihovna aioble tento proces zcela automatizuje – stačí zavolat await notify_char.subscribe(notify=True), což zajistí veškerou potřebnou konfiguraci na pozadí.
  • Reaktivní příjem dat: Program nečeká v přerušení (IRQ), ale používá metodu await notify_char.notified(). Tato metoda pozastaví vykonávání smyčky while connection.is_connected() a obnoví ji až v okamžiku, kdy ze serveru dorazí nová hodnota (notifikace). Tím je zajištěno, že se CPU nezatěžuje zbytečným dotazováním (pollingem) a reaguje pouze na skutečné změny stavu.
  • Robustní správa chyb: Celá komunikace je uzavřena v blocích try-except, což umožňuje programu elegantně reagovat na výpadky spojení nebo chyby při zápisu a automaticky se vrátit do stavu skenování bez pádu aplikace.

Závěr:

V tomto díle našeho seriálu jsme prošli cestu od úplných základů skenování v rádiovém prostoru až po komunikaci mezi dvěma moduly ESP32.

Nejprve jsme si ukázali, že porozumění nízkoúrovňovým principům, jako jsou callbacky (IRQ), dekódování binárních dat v advertising paketech a práce s identifikátory handle, je klíčové pro pochopení toho, jak Bluetooth Low Energy skutečně funguje. Viděli jsme, že proces navázání spojení přes GAP a následná manipulace s daty přes GATT vyžaduje precizní sekvenci kroků – od vyhledání služby přes charakteristiku až po případnou konfiguraci CCCD deskriptoru pro příjem notifikací.

Následné srovnání s knihovnou aioble nám však jasně demonstrovalo, jakým směrem se ubírá moderní vývoj v MicroPythonu. Asynchronní přístup nám umožnil:

  • Zpřehlednit kód: Namísto složitého přepínače v IRQ handleru píšeme lineární, snadno čitelný kód s využitím await.
  • Zvýšit efektivitu: Díky úlohám v asyncio může náš BLE klient paralelně obsluhovat tlačítka, senzory nebo Wi-Fi konektivitu, aniž by jedno blokovalo druhé.
  • Abstrahovat složitost: Operace jako aktivace notifikací (zápis do CCCD) se smrskly na jediné volání metody subscribe(), což eliminuje časté chyby při ruční manipulaci s bajty.

Schopnost vytvořit BLE klienta a tím pádem propojit dvě ESP32 napřímo otevírá obrovské možnosti v oblasti domácí automatizace a průmyslového měření. Nyní již nejste odkázáni pouze na telefon jako na prostředníka, ale můžete vytvářet sítě autonomních senzorů a akčních členů, které spolu komunikují úsporně a spolehlivě.

Tím skončil náš miniseriál o komunikaci Bluetooth v MicroPythonu na modulu ESP32.

Tímto textem uzavíráme kapitolu na zaměřenou na Bluetooth v MicroPythonu na modulu ESP32. Máte nyní v rukou veškeré nástroje k tomu, abyste své MicroPython projekty zbavili drátů a posunuli je na další možnou úroveň bezdrátové komunikace.

Možná právě Váš příští projekt bude ten, který budete pohodlně ovládat pomocí Bluetooth přímo z displeje svého mobilu!

Pošli signál autorovi: