ESP32 BLE Example
Introduction
Call To Action
This tutorial is part of a bigger course called "Intelligent Product Design", which I am building out at https://backpack.danaukes.com/course/intelligent_product_design/. Please pardon the dust as I build it out.
If you just want the tutorial, you're in the right spot. But, if you need a little more, such as live lectures, a discussion board, or a community of other students, consider enrolling once the course opens. For now, please feel free to explore https://backpack.danaukes.com.
Prep
Create new project folders
mkdir -p project01-ble/esp/lib project01-ble/pc
download micropython lib and extract to the lib directory:
curl -L -o micropython-lib.zip https://github.com/micropython/micropython-lib/archive/refs/heads/master.zip
unzip micropython-lib.zip "micropython-lib-master/micropython/bluetooth/aioble/aioble/*" -d "project01-ble/esp/lib/aioble"
rm micropython-lib.zip
invoke-webrequest "https://github.com/micropython/micropython-lib/archive/refs/heads/master.zip" -outfile micropython-lib.zip
expand-archive micropython-lib.zip
mv micropython-lib/micropython-lib-master/micropython/bluetooth/aioble/aioble project01-ble/lib
rm -r micropython-lib micropython-lib.zip
Micropython Code
create project01-ble/esp/main.py and paste in the following:
from micropython import const
import asyncio
import aioble
import bluetooth
import ubinascii
UUID_SERVICE = bluetooth.UUID(0x1234)
UUID_CHARACTERISTIC = bluetooth.UUID(0x2345)
ADVERTISING_INTERVAL = 250000
NAME = 'ESP32'
service = aioble.Service(UUID_SERVICE)
characteristic = aioble.Characteristic(service, UUID_CHARACTERISTIC, read=True, notify=True)
aioble.register_services(service)
ble = bluetooth.BLE()
dummy, mac = ble.config('mac')
print("MAC:",':'.join(['{:02x}'.format(x) for x in mac]))
async def write_characteristic():
while True:
characteristic.write(b'hello', send_update=True)
await asyncio.sleep_ms(1000)
async def advertise():
while True:
async with await aioble.advertise(ADVERTISING_INTERVAL,name=NAME,services=[UUID_SERVICE],) as connection:
print("Connection from", connection.device)
await connection.disconnected(timeout_ms=None)
async def main():
write_task = asyncio.create_task(write_characteristic())
advertise_task = asyncio.create_task(advertise())
await asyncio.gather(write_task, advertise_task)
asyncio.run(main())
create a new pymakr project in the esp folder
open the terminal, make sure the ESP32 is stopped (ctrl+c)
use the pymakr interface to upload all project files to the ESP32
run the esp32 by either pressing the "en" button on the device or by typing ctrl+d to soft boot it from the terminal
note the MAC address printed in the terminal once the code starts
PC Code
create project01-ble/pc/main.py and paste in the following making sure to replace the "
import asyncio
from bleak import BleakClient, uuids
import bleak
connected = False
MAC = "<ENTER MAC HERE>"
UUID_SERVICE = uuids.normalize_uuid_16(0x1234)
UUID_CHARACTERISTIC = uuids.normalize_uuid_16(0x2345)
def callback(sender, data):
print(f"Callback: {sender}: {data.decode()}")
def scan_callback(device, advertising_data):
print("device: ", device, ", advertising_data: ", advertising_data)
stop_event = asyncio.Event()
async def scanner_loop():
async with bleak.BleakScanner(scan_callback,[UUID_SERVICE],scanning_mode="active",) as scanner:
await stop_event.wait()
while True:
await asyncio.sleep(2)
print("scanner sleeping")
async def connection_main_loop():
global connected
while True:
client = None
try:
client = BleakClient(MAC)
await client.connect()
connected = True
print("Connected")
services = client.services
print("Services: ", services)
await client.start_notify(UUID_CHARACTERISTIC, callback)
while True:
await asyncio.sleep(1)
print("sleeping")
if not client.is_connected:
break
except asyncio.CancelledError:
raise
except Exception as e:
print(f"Connection failed", e)
finally:
connected = False
if client:
try:
if client.is_connected:
await asyncio.wait_for(client.disconnect(), timeout=2.0)
except:
pass
print("Reconnecting in 3 Seconds...")
await asyncio.sleep(3)
async def general_loop():
await asyncio.sleep(10)
stop_event.set()
async def main():
try:
scan_loop = asyncio.create_task(scanner_loop())
gen_loop = asyncio.create_task(general_loop())
connection_loop = asyncio.create_task(connection_main_loop())
await asyncio.gather(scan_loop, gen_loop, connection_loop)
except asyncio.CancelledError as e:
print(e)
except KeyboardInterrupt:
print(e)
except Exception as e:
print(e)
finally:
connected = False
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
