Bluetooth LE heartrate sensor communication for ULX3S ESP32

heartrate.py

# Bluetooth LE heartrate sensor communication
#
# Connect to the first available Bluetooth LE heartrate sensor, subscribes to
# notifications, then sends the values to the FPGA through the UART in single-byte packets.
#
# Mara "vmedea" 2020
# SPDX-License-Identifier: MIT
from binascii import hexlify
from micropython import const
from machine import UART
import ubluetooth

# BT "IRQ" event type constants
_IRQ_SCAN_RESULT = const(5)
_IRQ_SCAN_DONE = const(6)
_IRQ_PERIPHERAL_CONNECT = const(7)
_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_NOTIFY = const(18)

# UUIDs
HEARTRATE_SERVICE_UUID = ubluetooth.UUID(0x180d)
HEARTRATE_CHARACTERISTIC_UUID = ubluetooth.UUID(0x2a37)

# global state
verbose = True
sensor_addr = None
sensor_handle = None
sensor_value_handle = None
sensor_notify_handle = None

# find adv_type item in advertising data, return it if found else None
def adv_decode(adv_type, data):
    i = 0
    while i + 1 < len(data):
        if data[i + 1] == adv_type:
            return data[i + 2: i + data[i] + 1]
        i += 1 + data[i]
    return None

# find a service class in a list of 16-bit services, return True if found False otherwise
def find_service16(data, uuid):
    if data is None:
        return False
    i = 0
    while i < len(data):
        if ubluetooth.UUID(data[i] | data[i+1]<<8) == uuid:
            return True
        i += 2
    return False

# search for service class in complete or incomplete list of services
def have_service(adv_data, uuid):
    return find_service16(adv_decode(0x02, adv_data), uuid) or find_service16(adv_decode(0x03, adv_data), uuid)

# bluetooth "IRQ" handler
def bt_irq(event, data):
    if event == _IRQ_SCAN_RESULT:
        global sensor_addr
        addr_type, addr, adv_type, rssi, adv_data = data
        if sensor_addr is None and have_service(adv_data, HEARTRATE_SERVICE_UUID):
            print('[found]', addr_type, hexlify(addr), adv_type, rssi)
            sensor_addr = (addr_type, addr)

            ble.gap_connect(addr_type, addr, 10000)
    elif event == _IRQ_SCAN_DONE:
        print('[scan complete]')
    elif event == _IRQ_PERIPHERAL_CONNECT:
        global sensor_handle
        conn_handle, addr_type, addr = data
        print('[connected]', conn_handle)
        sensor_handle = conn_handle
        ble.gattc_discover_services(conn_handle, HEARTRATE_SERVICE_UUID)
    elif event == _IRQ_GATTC_SERVICE_RESULT:
        conn_handle, start_handle, end_handle, uuid = data
        print('[service result]', conn_handle, start_handle, end_handle, uuid)
        ble.gattc_discover_characteristics(conn_handle, start_handle, end_handle, HEARTRATE_CHARACTERISTIC_UUID)
    elif event == _IRQ_GATTC_SERVICE_DONE:
        conn_handle, status = data
        print('[service done]', conn_handle, status)
    elif event == _IRQ_GATTC_CHARACTERISTIC_RESULT:
        conn_handle, def_handle, value_handle, properties, uuid = data
        print('[characteristic result]', conn_handle, def_handle, value_handle, properties, uuid)
        # properties should include 0x10 (Notify)
        if properties & 0x10:
            global sensor_value_handle, sensor_notify_handle
            sensor_value_handle = value_handle
            sensor_notify_handle = value_handle + 1 # TODO: though it should always follow directly, safer to scan for 0x2902 descriptor UUID?
            # start listen to notifications
            ble.gattc_write(sensor_handle, sensor_notify_handle, b'\x01\x00')
        else:
            print('ERROR: HEARTRATE characteristic misses NOTIFY flag')
    elif event == _IRQ_GATTC_CHARACTERISTIC_DONE:
        conn_handle, status = data
        print('[characteristic done]', conn_handle, status)
    #elif event == _IRQ_GATTC_DESCRIPTOR_RESULT:
    #    conn_handle, dsc_handle, uuid = data
    #    print('[descriptor result]', conn_handle, dsc_handle, uuid)
    #elif event == _IRQ_GATTC_DESCRIPTOR_DONE:
    #    conn_handle, status = data
    #    print('[descriptor done]', conn_handle, status)
    elif event == _IRQ_GATTC_NOTIFY:
        conn_handle, value_handle, notify_data = data
        if conn_handle == sensor_handle and value_handle == sensor_value_handle:
            if verbose:
                print('[data flags={:08b} rate={}]'.format(notify_data[0], notify_data[1]))
            uart.write(notify_data[1:2])

ble = ubluetooth.BLE()
ble.irq(bt_irq)
ble.active(True)
ble.gap_scan(10000, 30000, 30000)

uart = UART(2) # 16:RX 17:TX
uart.init(baudrate=115200, bits=8, parity=None, stop=1)