BLE UART Service - Host

Contents

BLE UART Service - Host#

Verify Bluetooth is available and working. This has been tested on a Raspberry Pi 4. If you get an error, your host may not have Bluetooth or it is not configured.

!bluetoothctl scan on

[CHG] Controller 00:1A:7D:DA:71:03 UUIDs: 00001801-0000-1000-8000-00805f9b34fb
[CHG] Controller 00:1A:7D:DA:71:03 UUIDs: 00001800-0000-1000-8000-00805f9b34fb
[CHG] Controller 00:1A:7D:DA:71:03 UUIDs: 0000110c-0000-1000-8000-00805f9b34fb
[CHG] Controller 00:1A:7D:DA:71:03 UUIDs: 0000110e-0000-1000-8000-00805f9b34fb
AdvertisementMonitor path registered

Python bluetooth scanner:

import asyncio
from bleak import BleakScanner

async def main():
    devices = await BleakScanner.discover()
    for d in devices:
        print(d)

await main()

UART Service - CPython#

import asyncio
import sys

from bleak import BleakScanner, BleakClient
from bleak.backends.scanner import AdvertisementData
from bleak.backends.device import BLEDevice

# https://github.com/hbldh/bleak/blob/develop/examples/uart_service.py

class BLE_UART:
    
    UART_SERVICE_UUID = "6E400001-B5A3-F393-E0A9-E50E24DCCA9E"
    UART_RX_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
    UART_TX_CHAR_UUID = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"

    # All BLE devices have MTU of at least 23. Subtracting 3 bytes overhead, we can
    # safely send 20 bytes at a time to any device supporting this service.
    UART_SAFE_SIZE = 20
    
    def __init__(self, peripheral_name='mpy-uart'):
        self._peripheral_name = peripheral_name
        self._rx_queue = asyncio.Queue()
        
    async def read(self):
        msg = await self._rx_queue.get()
        return msg
    
    async def write(self, msg):
        if isinstance(msg, str):
            msg = msg.encode()
        await self._client.write_gatt_char(self.UART_RX_CHAR_UUID, msg)
        
    async def connect(self):
        self._discovery_queue = asyncio.Queue()
        device = None
        print(f"scanning for {self._peripheral_name}")
        async with BleakScanner(detection_callback=self._find_uart_device):
            device: BLEDevice = await self._discovery_queue.get()
        print(f"connecting to {self._peripheral_name} ...", end="")
        client = self._client = BleakClient(device, disconnected_callback=self._handle_disconnect)
        await client.connect()
        await client.start_notify(self.UART_TX_CHAR_UUID, self._rx_handler)    
        print(f" connected")
        
    async def disconnect(self):
        await self._client.disconnect()
    
    async def __aenter__(self):
        return self
    
    async def __aexit__(self, *args):
        await self.disconnect()
        
    def _rx_handler(self, _: int, data: bytearray):
        self._rx_queue.put_nowait(data)
    
    def _find_uart_device(self, device: BLEDevice, adv: AdvertisementData):
        # called whenever a device is detected during discovery
        # ignore all but target device
        if device.name == self._peripheral_name:
            self._discovery_queue.put_nowait(device)
        
    def _handle_disconnect(self, _: BleakClient):
        self._rx_queue.put_nowait(None)

Receive messages from mpy-uart:

async with BLE_UART() as uart:
    await uart.connect()

    for i in range(3):
        msg = await uart.read()
        msg = msg.decode()
        print("received", msg)
        await uart.write(f"echo {msg}")

scanning for mpy-uart
connecting to mpy-uart ... connected
received hello # 1 from esp32
received hello # 2 from esp32
received hello # 3 from esp32