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(číslo5) 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í funkcedecode_ručně rozebrat podle specifikace BLE.name - 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
Trueve funkcigap_zapíná aktivní skenování, při kterém se ESP32 dotazuje zařízení na další podrobnosti (např. právě na ono jméno).scan
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
Trueve funkcigap_), proces probíhá ve dvou krocích. Jakmile klient zachytí základní advertising paket, okamžitě pošle zařízení dotaz zvanýscan 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.
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.
V 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):
...
V 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.
V 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. 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_. 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í:
- Pokud klient zjistí, že bylo stisknuto tlačítko
BOOT, odešle serveru hodnotu1 - Čistě pro kontrolu rozsvítí vlastní LED.
- Server pozná změnu hodnoty charakteristiky (hodnota
1) a rozsvítí svou vestavěnou LED. - Při uvolnění tlačítka odešle klient hodnotu
0(a zhasne svou LED). - 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:
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_ 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., 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í. (hodnota0by znamenala nekonečné skenování aNoneskenová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).
- Při pasivním skenování (
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. 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_ 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: 6E40
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: 6E40
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_ nepředává UUID charakteristiky, ale char_.
Jakmile je známe conn_ a char_ 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_ určuje spojení se zařízením, char_ 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
asyncioprogram v hlavní funkcimain()vytvoří dvě nezávislé úlohy pomocíasyncio.. Úlohacreate_ task() ble_se stará výhradně o vyhledání serveru a udržování spojení, zatímcotask button_běží paralelně a monitoruje fyzický stav tlačítka.task - Efektivní skenování a identifikace: Skenování probíhá v kontextovém manažeru
async with aioble., který automaticky zajistí správné spuštění a ukončení skeneru. Program v asynchronní smyčce prochází výsledky a pomocí metodyscan result.okamžitě zjišťuje jméno zařízení, aniž by musel ručně dekódovat binární strukturu advertising paketů.name() - Abstrakce GATT operací: Program se již nestará o číselné handly spojení nebo charakteristik; místo toho pracuje s objekty
connection,serviceawrite_.char - Zápis hodnot: Samotné odeslání dat (např.
"1"při stisku) probíhá asynchronně metodouwrite_. Parametrchar. write() response=zajistí, že program počká na potvrzení o úspěšném přijetí dat druhou stranou.True
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í:
- Pokud stiskneme na serveru tlačítko
BOOT(dojde ke změně stavu), nastaví server jako notifikaci hodnotu charakteristiky na1. - Na změnu hodnoty charakteristiky (notify) zareaguje klient a podle hodnoty
1rozsvítí svou LED. - Po uvolnění tlačítka na straně serveru, dojde opět k odeslání notifikace – hodnota se změní na
0. - 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:
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:
- aktivace BLE,
- scan,
- nalezení serveru,
- připojení přes GAP,
- nalezení GATT služby,
- 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. CCCD – Client 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_ a _IRQ_. 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_ 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 a1b2? 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_ pro GATT komunikaci a SERVICE_ 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_ 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_ 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_ 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 charakteristikycccd_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_, 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_ se zde používá pouze pro povolení NOTIFY.
Potvrzení konfigurace
Po zápisu do CCCD přijde událost: _IRQ_
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., pokud přijde 0, LED zhasne: led..
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_ a _DESCRIPTOR_.
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_, protože po novém připojení může mít CCCD jiný handle.
Program poté zhasne LED led. a znovu zahájí scan.
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_prohledává advertising data (service result.) a zjišťuje, zda zařízení inzeruje konkrétní 128bitové UUID. Pokud ano, klient se k němu pokusí připojit.adv_ data - 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 hodnotu0x0100. Knihovnaaiobletento proces zcela automatizuje – stačí zavolatawait notify_, což zajistí veškerou potřebnou konfiguraci na pozadí.char. subscribe( notify= True) - Reaktivní příjem dat: Program nečeká v přerušení (IRQ), ale používá metodu
await notify_. Tato metoda pozastaví vykonávání smyčkychar. notified() while connection.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.is_ connected() - 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
asynciomůž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!