Movesense传感器通过Python脚本无法接收数据，但移动端APP可正常工作

问题描述 #

我正在使用Movesense传感器，尝试用文档提供的Python脚本收集数据。路径和命令都已根据Movesense API文档验证正确，也和其他参考代码交叉核对过。Python客户端能成功连接传感器并发送启动数据流的命令，但完全接收不到数据。

我已经做了以下排查：

• 传感器功能正常，连接Movesense移动端APP时能正常流式传输数据。
• 使用的路径（如/Meas/ECG/200）正确，和文档一致。
• 写入命令发送成功（通过调试日志确认），通知已启用且无错误。
• 写入和通知特征的UUID都是正确的。

除此之外，我还尝试了：

• 在发送写入命令前后添加延迟。
• 多次重发写入命令。
• 测试期间确保传感器未连接其他设备。

Python脚本使用Bleak库进行BLE通信，日志显示无错误，传感器也确认收到了写入命令，但就是不发送数据。有没有人遇到过类似问题？或者我在配置或通信序列中可能漏掉了什么？非常感谢任何见解或建议。

附上我使用的代码：

# -*- coding: utf-8 -*-
"""
Python gatt_sensordata_app client example using the Bleak GATT client.

This example is based on the examples in the Bleak repo: https://github.com/hbldh/bleak
"""

import logging
import asyncio
import platform
import signal
from bleak import BleakClient
from bleak import _logger as logger
from bleak import discover
from functools import reduce
from typing import List
import struct
import sys

WRITE_CHARACTERISTIC_UUID = (
    <WRITE_CHARACTERISTIC_UUID>
)

NOTIFY_CHARACTERISTIC_UUID = (
    <NOTIFY_CHARACTERISTIC_UUID>
)


# https://stackoverflow.com/a/56243296
class DataView:
    def __init__(self, array, bytes_per_element=1):
        """
        bytes_per_element is the size of each element in bytes.
        By default we are assume the array is one byte per element.
        """
        self.array = array
        self.bytes_per_element = 1

    def __get_binary(self, start_index, byte_count, signed=False):
        integers = [self.array[start_index + x] for x in range(byte_count)]
        _bytes = [integer.to_bytes(
            self.bytes_per_element, byteorder='little', signed=signed) for integer in integers]
        return reduce(lambda a, b: a + b, _bytes)

    def get_uint_16(self, start_index):
        bytes_to_read = 2
        return int.from_bytes(self.__get_binary(start_index, bytes_to_read), byteorder='little')

    def get_uint_8(self, start_index):
        bytes_to_read = 1
        return int.from_bytes(self.__get_binary(start_index, bytes_to_read), byteorder='little')

    def get_uint_32(self, start_index):
        bytes_to_read = 4
        binary = self.__get_binary(start_index, bytes_to_read)
        return struct.unpack('<I', binary)[0]  # <f for little endian

    def get_int_32(self, start_index):
        bytes_to_read = 4
        binary = self.__get_binary(start_index, bytes_to_read)
        return struct.unpack('<i', binary)[0]  # < for little endian

    def get_float_32(self, start_index):
        bytes_to_read = 4
        binary = self.__get_binary(start_index, bytes_to_read)
        return struct.unpack('<f', binary)[0]  # <f for little endian



async def run_queue_consumer(queue: asyncio.Queue):
    while True:
        data = await queue.get()
        if data is None:
            logger.info(
                "Got message from client about disconnection. Exiting consumer loop..."
            )
            break
        else:
            # print to stdout
            print(data)

PACKET_TYPE_DATA = 2
PACKET_TYPE_DATA_PART2 = 3
ongoing_data_update = None

async def run_ble_client( end_of_serial: str, sensor_types: List[str], queue: asyncio.Queue):

    # Check the device is available
    devices = await discover()
    
    found = False
    address = None
    for d in devices:
        logger.debug("device:", d)
        if d.name and d.name.endswith(end_of_serial):
            logger.info("device found")
            address = d.address
            found = True
            break

    # This event is set if device disconnects or ctrl+c is pressed
    disconnected_event = asyncio.Event()

    def raise_graceful_exit(*args):
        disconnected_event.set()

    def disconnect_callback(client):
        logger.info("Disconnected callback called!")
        disconnected_event.set()

    async def notification_handler(sender, data):
        """Simple notification handler which prints the data received."""
        logger.info(f"Notification received from {sender}: {data}")
        print(f"Raw data: {data}")
        d = DataView(data)

        packet_type= d.get_uint_8(0)
        reference = d.get_uint_8(1)

        global ongoing_data_update
        # print("packet ", packet_type, ", ongoing:",ongoing_data_update)
        if packet_type == PACKET_TYPE_DATA:
            # ECG (reference 100) fits in one packet
            if reference == 100:
                timestamp = d.get_uint_32(2)
            
                for i in range(0,16):
                    # Interpolate timestamp within the data notification
                    row_timestamp = timestamp + int(i*1000/200)
                    ## ECG package starts with timestamp and then array of 16 samples
                    # Sample scaling is 0.38 uV/sample
                    sample_mV = d.get_int_32(6+i*4) * 0.38 *0.001
                    msg_row = "ECG,{},{:.3f}".format(row_timestamp, sample_mV)
                
                    # queue message for later consumption (output)
                    await queue.put(msg_row)

            else:
            # print("PACKET_TYPE_DATA")
            # Store 1st part of the incoming data
                ongoing_data_update = d
    
        elif packet_type == PACKET_TYPE_DATA_PART2:
            # print("PACKET_TYPE_DATA_PART2. len:",len(data))
                
            # Create combined DataView that contains the whole data packet
            # (skip type_id + ref num of the data_part2)
            d = DataView( ongoing_data_update.array + data[2:])
            ongoing_data_update = None

            # Dig data from the binary
            # msg = "Data: offset: {}, len: {}".format(d.get_uint_32(2), 
            #     len(d.array))
            timestamp = d.get_uint_32(2)
            for i in range(0,8):
                # Interpolate timestamp within the data notification
                row_timestamp = timestamp + int(i*1000/104)
                ## IMU9 package starts with timestamp and then three arrays (len 8*4 bytes) of xyz's 
                ## Each "row" therefore starts (3*4 bytes after each other interleaving to acc, gyro and magn)
                offset = 6 + i * 3* 4
                skip = 3*8*4
                msg_row = "IMU9,{},{:.2f},{:.2f},{:.2f},{:.2f},{:.2f},{:.2f},{:.2f},{:.2f},{:.2f}".format(row_timestamp
                                                            ,  d.get_float_32(offset)
                                                            ,  d.get_float_32(offset+4)
                                                            ,  d.get_float_32(offset+8)
                                                            ,  d.get_float_32(offset+skip+0)
                                                            ,  d.get_float_32(offset+skip+4)
                                                            ,  d.get_float_32(offset+skip+8)
                                                            ,  d.get_float_32(offset+2*skip+0)
                                                            ,  d.get_float_32(offset+2*skip+4)
                                                            ,  d.get_float_32(offset+2*skip+8))
                # queue message for later consumption (output)
                await queue.put(msg_row)

    if found:
        async with BleakClient(address, disconnected_callback=disconnect_callback) as client:

            # Add signal handler for ctrl+c
            signal.signal(signal.SIGINT, raise_graceful_exit)
            signal.signal(signal.SIGTERM, raise_graceful_exit)
           
            # Start notifications and subscribe to acceleration @ 13Hz
            logger.info("Attempting to enable notifications")
            await client.start_notify(NOTIFY_CHARACTERISTIC_UUID, notification_handler)
            logger.info("Notifications successfully enabled")
            if "ECG" in sensor_types:
                logger.info("Sending write command for ECG")
                await client.write_gatt_char(WRITE_CHARACTERISTIC_UUID, bytearray([1, 2])+bytearray("/Meas/ECG/200", "utf-8"), response=True)
                logger.info("Write command for ECG sent successfully")
            if "IMU9" in sensor_types:
                logger.info("Sending write command for IMU9")
                await client.write_gatt_char(WRITE_CHARACTERISTIC_UUID, bytearray([1, 99])+bytearray("/Meas/IMU9/104", "utf-8"), response=True)
                logger.info("Write command for IMU9 sent successfully")

            # Run until disconnect event is set
            await disconnected_event.wait()

我的解答 #

我之前帮几个开发者解决过类似的Movesense+Bleak问题，结合你的排查步骤和代码，发现了几个关键问题，按优先级整理如下：

1. 订阅参考号不匹配（最可能的原因）

你的代码里有一个明显的矛盾：

• 发送ECG订阅命令时，用的参考号是2（写入命令的第二个字节：bytearray([1, 2])）
• 但在notification_handler里，处理ECG数据的条件却是if reference == 100

Movesense的通知数据包里的reference值，就是你订阅时指定的那个第二个字节，这两个值必须完全对应才能触发数据解析逻辑！你要么把发送命令的参考号改成100，要么把处理逻辑里的reference == 100改成reference == 2，二选一就行。

同样检查IMU9的部分：你发送用的参考号是99，但处理IMU9数据的逻辑里没有做reference判断，不过因为IMU9用的是分包（PACKET_TYPE_DATA_PART2），目前的逻辑是靠ongoing_data_update来衔接，但最好也加上reference ==99的判断，避免和其他订阅的数据包混淆。

2. 检查通知是否真的触发了

先在notification_handler的最开头加一行打印，比如：

async def notification_handler(sender, data):
    d = DataView(data)
    print(f"收到通知！类型：{d.get_uint_8(0)}，参考号：{d.get_uint_8(1)}")
    # 原有的代码...

如果运行脚本后完全看不到这行打印，说明BLE通知根本没触发，那可能是：

• Bleak版本兼容性问题：试试升级到最新稳定版（pip install --upgrade bleak），旧版本对某些BLE栈的通知支持有bug
• 平台BLE栈的问题：比如Windows的BLE栈有时候需要手动配对，或者Linux下需要给BLE设备权限（用bluetoothctl配对后再运行脚本）

3. 订阅命令的格式细节

Movesense的订阅命令格式是正确的，但要确保：

• 写入命令的response=True参数是对的，你已经加了，这个没问题
• 发送完订阅命令后，给传感器一点启动时间，比如在发送命令后加await asyncio.sleep(1)，虽然你试过加延迟，但可能之前的延迟时机不对（比如加在连接后而不是发送命令后）

4. 日志排查

把Bleak的日志级别调到DEBUG，看看有没有隐藏的错误：

logging.basicConfig(level=logging.DEBUG)

这样能看到BLE通信的每一步细节，比如传感器是否真的回复了订阅确认，有没有通知数据包被遗漏。

备注：内容来源于stack exchange，提问作者Shreyas Ramachandran