* 采集电池数据流 以下示例展示了如何连接到设备并采集10秒的电池和RSSI数据 ``` import asyncio import sys from ble_sdk.client import BleClient from ble_sdk.constants import DeviceType from loguru import logger # 配置应用日志 logger.remove() logger.add( sys.stderr, level="INFO", filter=lambda record: not record["extra"].get("sdk_internal", False), ) DEVICE_ADDRESS = "D2:43:D5:88:4D:9A" # 替换为你的设备地址 def my_data_handler(data): """用户自定义的回调函数，用于处理接收到的数据""" logger.info(f"接收到数据: {data}") async def main(): client = BleClient(address=DEVICE_ADDRESS, device_type=DeviceType.BLE_8) try: async with client: logger.info("设备已连接。") await client.start_battery_stream(my_data_handler) logger.info("电池数据流已启动，监控10秒...") await asyncio.sleep(10) await client.stop_battery_stream() logger.info("电池数据流已停止。") except Exception as e: logger.error(f"发生错误: {e}") if __name__ == "__main__": asyncio.run(main()) ``` * 原始数据模式示例 SDK支持返回去除帧头帧尾的原始16进制数据，便于自定义处理 ``` import asyncio from ble_sdk.client import BleClient from ble_sdk.constants import DeviceType def raw_data_handler(data): """处理原始数据的回调函数""" if isinstance(data, bytearray): print(f"原始数据 (长度={len(data)}): {data.hex()}") else: print(f"解析数据: {data}") async def main(): client = BleClient(address="XX:XX:XX:XX:XX:XX", device_type=DeviceType.BLE_8) async with client: # 启动EEG原始数据流 await client.start_eeg_stream(raw_data_handler, raw_data_only=True) await asyncio.sleep(10) await client.stop_eeg_stream() asyncio.run(main()) ``` * 混合数据流示例混合模式可以同时采集主数据流（EEG或Impedance）、IMU和Lead数据 ``` import asyncio from ble_sdk.client import BleClient from ble_sdk.constants import DeviceType def mixed_data_handler(data): """处理混合数据的回调函数""" print(f"时间戳: {data['timestamp']}") print(f"主数据类型: {data['type']}") # 主数据（EEG或Impedance） if data['type'] == 'eeg': eeg_data = data['eeg_data'] if isinstance(eeg_data, bytearray): print(f"EEG原始数据: {eeg_data.hex()}") else: print(f"EEG通道数据: {len(eeg_data['channels'])} 通道") # IMU数据 imu_data = data.get('imu_data') if imu_data: if isinstance(imu_data, bytearray): print(f"IMU原始数据: {imu_data.hex()}") else: print(f"IMU数据: 加速度={imu_data['acceleration']}") # Lead数据 lead_data = data.get('lead_data') if lead_data: if isinstance(lead_data, bytearray): print(f"Lead原始数据: {lead_data.hex()}") else: print(f"Lead状态: P={lead_data['p_status']}") async def main(): client = BleClient(address="XX:XX:XX:XX:XX:XX", device_type=DeviceType.BLE_8) async with client: # 启动混合数据流，主数据为EEG，启用原始数据模式 await client.start_mixed_stream( mode="eeg", user_callback=mixed_data_handler, raw_main_data_only=True # 所有数据都返回原始格式 ) await asyncio.sleep(30) await client.stop_mixed_stream() asyncio.run(main()) ``` * 使用命令行工具 ( sdk\_cli\_monitor.py ) 我们提供了一个强大的命令行工具，用于快速测试所有数据流。 启动EEG数据流，持续20秒: python examples/sdk\_cli\_monitor.py --mode eeg --duration 20 启动EEG原始数据流: python examples/sdk\_cli\_monitor.py --mode eeg --raw\_data --duration 10 启动混合数据流（主数据为阻抗），持续30秒: python examples/sdk\_cli\_monitor.py --mode mixed --mixed\_primary impedance -- duration 30 启动混合数据流并输出原始数据: python examples/sdk\_cli\_monitor.py --mode mixed --mixed\_primary eeg --raw\_data -- duration 20 启动IMU数据流，并指定设备地址: python examples/sdk\_cli\_monitor.py --mode imu --address "XX:XX:XX:XX:XX:XX" -- raw\_data 查看所有可用参数: python examples/sdk\_cli\_monitor.py --help * 算法的使用 ``` from ble_sdk.algo.eeg import get_eeg_signal # input_list: 去掉header 和 tail 长度 192 的 eeg 数据 # channel_num: 1 2 3 4 通道数 nddata, psddd, ffttt = get_eeg_signal(input_list, channel_num=1) print(nddata.shape, psddd.shape, ffttt.shape) ```