ESP32 Bluetooth "[Errno 12] ENOMEM" after a few seconds

All ESP32 boards running MicroPython.
Target audience: MicroPython users with an ESP32 board.
Post Reply
Peacekeep3r
Posts: 1
Joined: Mon Jul 25, 2022 11:38 pm

ESP32 Bluetooth "[Errno 12] ENOMEM" after a few seconds

Post by Peacekeep3r » Tue Jul 26, 2022 12:20 am

I would like to get several data values with approx. 100 Hz from a mobile sensor (ESP32-DevKitC V4) and send them to my Win10 PC via bluetooth.
I have worked with several code examples from the Git and they all work - until after a few seconds when they crash with a memory error. However, the memory of the ESP32 does not seem to be full. As already described in this topic viewtopic.php?f=20&t=8844 it is probably related to the bluetooth stack. Unfortunately there was no solution for me there.

edit: It looks like it does not crash with time.sleep(2) but even with time.sleep(1) it crashes after 8-12s. Still, I manage to send 10-20 notifications with "high" frequencies like 20 ms.


Code: Select all

Traceback (most recent call last):
  File "<stdin>", line 136, in <module>
  File "<stdin>", line 131, in run_ESP32
  File "<stdin>", line 102, in send
OSError: [Errno 12] ENOMEM
my code:

Code: Select all

import io, os
import machine
import bluetooth
import random
import time
from ble_advertising import advertising_payload
from micropython import const

# bluetooth magic settings
_IRQ_CENTRAL_CONNECT = const(1)
_IRQ_CENTRAL_DISCONNECT = const(2)
_IRQ_GATTS_WRITE = const(3)

_FLAG_READ = const(0x0002)
_FLAG_WRITE_NO_RESPONSE = const(0x0004)
_FLAG_WRITE = const(0x0008)
_FLAG_NOTIFY = const(0x0010)

_UART_UUID = bluetooth.UUID("00000000-0000-1000-8000-00805F9B34FB")
_UART_TX = (
    bluetooth.UUID("00000002-0000-1000-8000-00805F9B34FB"),
    _FLAG_READ | _FLAG_NOTIFY,
)
_UART_RX = (
    bluetooth.UUID("00000001-0000-1000-8000-00805F9B34FB"),
    _FLAG_WRITE | _FLAG_WRITE_NO_RESPONSE,
)
_UART_SERVICE = (
    _UART_UUID,
    (_UART_TX, _UART_RX),
)

class BLESimplePeripheral:
    def __init__(self, ble, name="esp32"):
        self._ble = ble
        self._ble.active(True)
        self._ble.irq(self._irq)
        ((self._handle_tx, self._handle_rx),) = self._ble.gatts_register_services((_UART_SERVICE,))
        self._connections = set()
        self._write_callback = None
        self._payload = advertising_payload(name=name, services=[_UART_UUID])
        self._advertise()

    def _irq(self, event, data):
        # Track connections so we can send notifications.
        if event == _IRQ_CENTRAL_CONNECT:
            conn_handle, _, _ = data
            print("New connection", conn_handle)
            self._connections.add(conn_handle)
        elif event == _IRQ_CENTRAL_DISCONNECT:
            conn_handle, _, _ = data
            print("Disconnected", conn_handle)
            self._connections.remove(conn_handle)
            # Start advertising again to allow a new connection.
            self._advertise()
        elif event == _IRQ_GATTS_WRITE:
            conn_handle, value_handle = data
            value = self._ble.gatts_read(value_handle)
            if value_handle == self._handle_rx and self._write_callback:
                self._write_callback(value)

    def send(self, data):
        self._ble.gatts_set_buffer(self._handle_tx, 1000, False) # ??
        self._ble.gatts_write(self._handle_tx, str.encode(data))
        
        for conn_handle in self._connections:
            self._ble.gatts_notify(conn_handle, self._handle_tx)

 
    def is_connected(self):
        return len(self._connections) > 0

    def _advertise(self, interval_us=500000):
        print("Starting advertising")
        self._ble.gap_advertise(interval_us, adv_data=self._payload)

    def on_write(self, callback):
        self._write_callback = callback


def run_ESP32():
    # to allow code cancellation in Thonny
    time.sleep_ms(3000)

    # initialize bluetooth hardware
    ble = bluetooth.BLE()
    p = BLESimplePeripheral(ble)
   
    while True:   
        # synthetic measurement data
        data = "hello world" # this also fails after some time
        time.sleep(0.128)
      
        # send data if connection is available  
        if p.is_connected():
            p.send(data)
            print(data)

if __name__ == "__main__":
    while True:
        run_ESP32() 

        
and the needed ble_advertising:

Code: Select all

# Helpers for generating BLE advertising payloads.

from micropython import const
import struct
import bluetooth

# Advertising payloads are repeated packets of the following form:
#   1 byte data length (N + 1)
#   1 byte type (see constants below)
#   N bytes type-specific data

_ADV_TYPE_FLAGS = const(0x01)
_ADV_TYPE_NAME = const(0x09)
_ADV_TYPE_UUID16_COMPLETE = const(0x3)
_ADV_TYPE_UUID32_COMPLETE = const(0x5)
_ADV_TYPE_UUID128_COMPLETE = const(0x7)
_ADV_TYPE_UUID16_MORE = const(0x2)
_ADV_TYPE_UUID32_MORE = const(0x4)
_ADV_TYPE_UUID128_MORE = const(0x6)
_ADV_TYPE_APPEARANCE = const(0x19)


# Generate a payload to be passed to gap_advertise(adv_data=...).
def advertising_payload(limited_disc=False, br_edr=False, name=None, services=None, appearance=0):
    payload = bytearray()

    def _append(adv_type, value):
        nonlocal payload
        payload += struct.pack("BB", len(value) + 1, adv_type) + value

    _append(
        _ADV_TYPE_FLAGS,
        struct.pack("B", (0x01 if limited_disc else 0x02) + (0x18 if br_edr else 0x04)),
    )

    if name:
        _append(_ADV_TYPE_NAME, name)

    if services:
        for uuid in services:
            b = bytes(uuid)
            if len(b) == 2:
                _append(_ADV_TYPE_UUID16_COMPLETE, b)
            elif len(b) == 4:
                _append(_ADV_TYPE_UUID32_COMPLETE, b)
            elif len(b) == 16:
                _append(_ADV_TYPE_UUID128_COMPLETE, b)

    # See org.bluetooth.characteristic.gap.appearance.xml
    if appearance:
        _append(_ADV_TYPE_APPEARANCE, struct.pack("<h", appearance))

    return payload


def decode_field(payload, adv_type):
    i = 0
    result = []
    while i + 1 < len(payload):
        if payload[i + 1] == adv_type:
            result.append(payload[i + 2 : i + payload[i] + 1])
        i += 1 + payload[i]
    return result


def decode_name(payload):
    n = decode_field(payload, _ADV_TYPE_NAME)
    return str(n[0], "utf-8") if n else ""


def decode_services(payload):
    services = []
    for u in decode_field(payload, _ADV_TYPE_UUID16_COMPLETE):
        services.append(bluetooth.UUID(struct.unpack("<h", u)[0]))
    for u in decode_field(payload, _ADV_TYPE_UUID32_COMPLETE):
        services.append(bluetooth.UUID(struct.unpack("<d", u)[0]))
    for u in decode_field(payload, _ADV_TYPE_UUID128_COMPLETE):
        services.append(bluetooth.UUID(u))
    return services


def demo():
    payload = advertising_payload(
        name="micropython",
        services=[bluetooth.UUID(0x181A), bluetooth.UUID("6E400001-B5A3-F393-E0A9-E50E24DCCA9E")],
    )
    print(payload)
    print(decode_name(payload))
    print(decode_services(payload))


if __name__ == "__main__":
    demo()

Do you guys mangage to send more then a few slow messages every now and then? Is there any solution yet?

User avatar
jimmo
Posts: 2754
Joined: Tue Aug 08, 2017 1:57 am
Location: Sydney, Australia
Contact:

Re: ESP32 Bluetooth "[Errno 12] ENOMEM" after a few seconds

Post by jimmo » Wed Aug 17, 2022 12:14 pm

Peacekeep3r wrote:
Tue Jul 26, 2022 12:20 am
self._ble.gatts_set_buffer(self._handle_tx, 1000, False) # ??
Try removing this line -- it's causing a 1kib allocation every time you send a value.

Post Reply