Micropython aioble BLE服务器数据接收异常问题求助

基于aioble外设传感器示例扩展代码实现数据接收功能，但无法接收前端通过navigator.bluetooth API发送的数据。尝试使用writeValueWithoutResponse和writeValue两种方式发送，设备端recv_actor函数读取的数据始终为None。

设备端Python代码 #

from micropython import const
import uasyncio as asyncio
import bluetooth
import aioble
import struct
import random
import json
import time


_ENV_SENSE_UUID = bluetooth.UUID("90D3D000-C950-4DD6-9410-2B7AEB1DD7D8")
_ENV_SENSE_TEMP_UUID = bluetooth.UUID("90D3D001-C950-4DD6-9410-2B7AEB1DD7D8")
_ENV_SENSE_RECV_UUID = bluetooth.UUID("90D3D002-C950-4DD6-9410-2B7AEB1DD7D8")
_ADV_APPEARANCE_GENERIC_THERMOMETER = const(768)
_ADV_INTERVAL_US = 250_000

# 注册GATT服务
device_service = aioble.Service(_ENV_SENSE_UUID)
temp_characteristic = aioble.Characteristic(device_service, _ENV_SENSE_TEMP_UUID, read=True, notify=True)
recv_characteristic = aioble.Characteristic(device_service, _ENV_SENSE_RECV_UUID, write=True, read=True, notify=True)

aioble.register_services(device_service)

async def temp_sensor():
    # 模拟周期性读取硬件传感器数据
    t = 24.5
    while True:
        data = {'t': t}
        data_encoded = bytes(json.dumps(data), 'utf8')
        temp_characteristic.write(data_encoded, send_update=True)
        t += random.uniform(-0.5, 0.5)
        await asyncio.sleep_ms(1000)

# 问题代码：始终读取到None
async def recv_actor():
    print('READING', recv_characteristic)
    while True:
        data = await recv_characteristic.read()
        await asyncio.sleep(1)

# 处理连接逻辑：连接时停止广播，断开后重新广播
async def peripheral_task():
    while True:
        async with await aioble.advertise(
            _ADV_INTERVAL_US,
            name="MicroIO_BLE",
            services=[_ENV_SENSE_UUID],
            appearance=_ADV_APPEARANCE_GENERIC_THERMOMETER,
        ) as connection:
            print("连接来自:", connection.device)
            global _connected_timer_start
            _connected_timer_start = time.ticks_ms()
            # 替换原disconnected()逻辑，避免超时崩溃
            while connection.is_connected() == True:
                await asyncio.sleep_ms(1000)
            print('连接断开，重新进入广播模式')


async def main():
    print('启动蓝牙传感器示例')
    await asyncio.gather(
        asyncio.create_task(temp_sensor()),
        asyncio.create_task(recv_actor()),
        asyncio.create_task(peripheral_task()),
    )
    print('示例结束')

asyncio.run(main())

前端JavaScript发送代码 #

characteristic.writeValueWithoutResponse(value);

characteristic.writeValue(value);

问题原因与解决方法 #

1. 错误的写入监听逻辑 #

你使用的await recv_characteristic.read()是主动读取特征当前值的方法，不会监听前端的写入事件。当没有新数据写入时，它自然返回None。正确的做法是使用aioble提供的wait_for_write()方法，该方法会阻塞直到有数据写入特征：

修改后的recv_actor函数 #

async def recv_actor():
    print('等待接收数据...')
    while True:
        # 阻塞等待前端写入数据
        await recv_characteristic.wait_for_write()
        # 读取写入的内容
        data = recv_characteristic.read()
        if data:
            print('收到数据:', data.decode('utf-8'))
        # 短休眠避免循环占用过多资源
        await asyncio.sleep_ms(100)

2. 特征属性匹配确认（可选） #

虽然你已设置write=True，若需同时支持带确认和无确认的写入，可显式指定属性（默认已支持，可省略）：

recv_characteristic = aioble.Characteristic(
    device_service, 
    _ENV_SENSE_RECV_UUID, 
    write=True, 
    read=True, 
    notify=True,
    write_no_response=True  # 支持无响应写入
)

3. 前端额外检查点 #

• 确保前端正确获取了_ENV_SENSE_RECV_UUID对应的特征，而非温度特征
• 发送的数据需为ArrayBuffer或Uint8Array类型，示例：

  const value = new TextEncoder().encode("测试数据");
  characteristic.writeValue(value);
  

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