ESP32-C3使用aioble时Characteristic.written()无限等待问题求助

问题描述 #

使用ESP32-C3开发BLE GATT服务器，实现双向数据交互：

• 特征1（GATT_CHAR_UUID）负责ESP向nRF Connect发送数据，功能正常；
• 特征2（RECV_CHAR_UUID）负责ESP接收nRF Connect的数据，但调用connection, data = await recv_char.written(timeout_ms=_ADV_INTERVAL_MS)时无限等待，即使在nRF Connect执行写入操作也无响应。

已尝试nRF Connect多种发送方式、单特征收发拆分等方案，问题未解决。

原始代码 #

import sys

sys.path.append('')

import asyncio
import aioble
import bluetooth
import machine
import time

led = machine.Pin(8, machine.Pin.OUT)
GATT_UUID = bluetooth.UUID(0x1802)
GATT_CHAR_UUID = bluetooth.UUID("90D3D001-C950-4DD6-9410-2B7AEB1DD7D8")
RECV_CHAR_UUID = bluetooth.UUID("90D3D002-C950-4DD6-9410-2B7AEB1DD7D8")
_ADV_INTERVAL_MS = 250_000
# Register GATT server, the service and characteristics
ble_service = aioble.Service(GATT_UUID)
sensor_characteristic = aioble.Characteristic(ble_service, GATT_CHAR_UUID, read=True, notify=True, write=True,
                                              capture=True)
recv_char = aioble.Characteristic(ble_service, RECV_CHAR_UUID, write=True, notify=True, capture=True)

# Register service(s)
aioble.register_services(ble_service)


def _decode_data(data):
    try:
        if data is not None:
            # Decode the UTF-8 data
            number = int.from_bytes(data, 'big')
            return number
    except Exception as e:
        print("Error decoding temperature:", e)
        return None


async def peripheral_task():
    while True:
        try:
            async with await aioble.advertise(_ADV_INTERVAL_MS, name="MYNAME",
                                              services=[GATT_UUID], ) as connection:
                print("Connection from", connection.device)
                await connection.disconnected()
        except asyncio.CancelledError:
            # Catch the CancelledError
            print("Peripheral task cancelled")
        except Exception as e:
            print("Error in peripheral_task:", e)
        finally:
            # Ensure the loop continues to the next iteration
            await asyncio.sleep_ms(100)


async def wait_for_write():
    while True:
        try:
            connection, data = await recv_char.written(timeout_ms=_ADV_INTERVAL_MS)
            # Here it fails ^
            print('OLOLO')
            print(data)
            # print(type)
            data = _decode_data(data)
            print('Connection: ', connection)
            print('Data: ', data)
            if data == 1:
                print('Turning LED ON')
                led.value(1)
            elif data == 0:
                print('Turning LED OFF')
                led.value(0)
            else:
                print('Unknown command')
        except asyncio.CancelledError:
            # Catch the CancelledError
            print("Wait4Write task cancelled")
        except Exception as e:
            print("Error in Wait4write_task:", e)
        finally:
            # Ensure the loop continues to the next iteration
            await asyncio.sleep_ms(500)


def _encode_data(data):
    return str(data).encode('utf-8')


async def sensor_task():
    while True:
        value = b'Hello World!'
        sensor_characteristic.write(_encode_data(value), send_update=True)
        # print('New value written: ', value)
        await asyncio.sleep_ms(1000)


# Run tasks
async def ble_main():
    t1 = asyncio.create_task(peripheral_task())
    t2 = asyncio.create_task(sensor_task())
    t3 = asyncio.create_task(wait_for_write())
    await asyncio.gather(t1, t2, t3)

问题排查与解决方案 #

1. 特征属性冗余配置 #

接收特征recv_char不需要notify=True属性（notify用于ESP主动向客户端推送数据），冗余属性可能导致GATT服务属性异常，干扰写入事件捕获。

修改特征定义：

recv_char = aioble.Characteristic(ble_service, RECV_CHAR_UUID, write=True, capture=True)

2. 服务UUID冲突风险 #

使用的GATT_UUID = bluetooth.UUID(0x1802)是蓝牙官方定义的Immediate Alert Service UUID，可能与系统默认服务或客户端预期的服务行为冲突。建议替换为自定义服务UUID：

GATT_UUID = bluetooth.UUID("90D3D000-C950-4DD6-9410-2B7AEB1DD7D8")

3. 写入事件捕获逻辑优化 #

• recv_char.written()的超时时间设置为250秒（_ADV_INTERVAL_MS=250_000）过长，可缩短至5秒（5000ms），便于快速排查超时问题；
• 确保nRF Connect中使用Write Request（而非Write Command）写入特征，因为written()方法仅捕获带响应的写入操作（Write Request会等待设备响应，Write Command无响应）。

修改wait_for_write中的超时参数：

connection, data = await recv_char.written(timeout_ms=5000)

4. 连接上下文与任务调度验证 #

peripheral_task中的await connection.disconnected()会阻塞直到连接断开，这本身不影响wait_for_write任务的事件监听，但需确保：

• 设备成功与nRF Connect建立连接（控制台输出Connection from [设备地址]）；
• 写入操作针对的是RECV_CHAR_UUID对应的特征，而非其他特征。

修改后完整代码 #

import sys

sys.path.append('')

import asyncio
import aioble
import bluetooth
import machine
import time

led = machine.Pin(8, machine.Pin.OUT)
# 替换为自定义服务UUID
GATT_UUID = bluetooth.UUID("90D3D000-C950-4DD6-9410-2B7AEB1DD7D8")
GATT_CHAR_UUID = bluetooth.UUID("90D3D001-C950-4DD6-9410-2B7AEB1DD7D8")
RECV_CHAR_UUID = bluetooth.UUID("90D3D002-C950-4DD6-9410-2B7AEB1DD7D8")
_ADV_INTERVAL_MS = 250_000

# 注册GATT服务与特征
ble_service = aioble.Service(GATT_UUID)
sensor_characteristic = aioble.Characteristic(ble_service, GATT_CHAR_UUID, read=True, notify=True, write=True, capture=True)
# 移除冗余的notify属性
recv_char = aioble.Characteristic(ble_service, RECV_CHAR_UUID, write=True, capture=True)

aioble.register_services(ble_service)


def _decode_data(data):
    try:
        if data is not None:
            number = int.from_bytes(data, 'big')
            return number
    except Exception as e:
        print("Error decoding data:", e)
        return None


async def peripheral_task():
    while True:
        try:
            async with await aioble.advertise(_ADV_INTERVAL_MS, name="MYNAME", services=[GATT_UUID]) as connection:
                print("Connection from", connection.device)
                await connection.disconnected()
        except asyncio.CancelledError:
            print("Peripheral task cancelled")
        except Exception as e:
            print("Error in peripheral_task:", e)
        finally:
            await asyncio.sleep_ms(100)


async def wait_for_write():
    while True:
        try:
            # 缩短超时时间至5秒
            connection, data = await recv_char.written(timeout_ms=5000)
            print('Received raw data:', data)
            decoded_data = _decode_data(data)
            print('Connection: ', connection)
            print('Decoded data: ', decoded_data)
            if decoded_data == 1:
                print('Turning LED ON')
                led.value(1)
            elif decoded_data == 0:
                print('Turning LED OFF')
                led.value(0)
            else:
                print('Unknown command')
        except asyncio.TimeoutError:
            # 捕获超时异常，避免任务终止
            print("Wait for write timeout, continue listening...")
        except asyncio.CancelledError:
            print("Wait4Write task cancelled")
        except Exception as e:
            print("Error in wait_for_write:", e)
        finally:
            await asyncio.sleep_ms(500)


def _encode_data(data):
    return str(data).encode('utf-8')


async def sensor_task():
    while True:
        value = b'Hello World!'
        sensor_characteristic.write(_encode_data(value), send_update=True)
        await asyncio.sleep_ms(1000)


async def ble_main():
    t1 = asyncio.create_task(peripheral_task())
    t2 = asyncio.create_task(sensor_task())
    t3 = asyncio.create_task(wait_for_write())
    await asyncio.gather(t1, t2, t3)

内容的提问来源于stack exchange，提问作者Enq Enq