ble-sdk 1.0.0

Python SDK for BLE IoT devices

BLE SDK

一个用于与BLE（低功耗蓝牙）物联网设备通信的Python SDK。

功能特性

• 简单易用的API：通过高级方法 (start_eeg_stream, start_battery_stream 等) 轻松启动和停止各种数据流。
• 多种数据支持：内置对多种数据类型的解析和处理，包括：
  • 电池电量与信号强度 (Battery/RSSI)
  • 脑电 (EEG)
  • 阻抗 (Impedance)
  • 惯性测量单元 (IMU)
  • 电极导联状态 (Lead Status)
• 原始数据模式：所有数据流都支持 raw_data_only 参数，可返回去除帧头帧尾的原始16进制数据，便于自定义数据处理。
• 混合数据模式：支持将 EEG/Impedance 主数据流与 IMU、Lead 数据流同步采集，并通过单一回调函数返回整合后的数据。混合模式支持统一的原始数据输出控制。
• 异步设计：完全基于 asyncio，支持高并发操作和异步上下文管理器 (async with)，确保资源被妥善管理。
• 灵活的回调机制：为每个数据流提供简单的回调函数接口，支持解析数据和原始数据两种模式。
• 完善的异常处理：定义了清晰的异常类，便于开发者进行错误处理。
• 隔离的日志系统：SDK内部使用独立的日志系统，不会干扰上层应用的日志配置。
• 动态通道配置：支持不同设备的通道数量配置（1、2、4、8通道），通过设备配置自动传递。
• 智能缓存机制：使用装饰器实现配置缓存，提高数据处理性能。
• 命令行监控工具：提供一个开箱即用的命令行工具，支持所有数据流模式和原始数据输出。

安装

通过 pip 从 PyPI 安装 (推荐):

# 此处为示例，当发布到PyPI后可使用
pip install ble-sdk

或者从源码安装以进行开发：

git clone https://github.com/your-repo/ble_sdk.git
cd ble_sdk
# 安装为可编辑模式，方便开发
pip install -e .

日志系统

SDK采用了隔离的日志系统设计，确保SDK内部日志和上层应用日志互不干扰：

SDK内部日志

• 使用loguru库，但通过过滤器实现逻辑隔离
• 格式：时间 | 级别 | SDK | 模块:函数:行号 | 消息
• 在__init__.py中全局初始化，默认INFO级别

上层应用日志

• 可以自由配置loguru或其他日志库
• 通过过滤器排除SDK内部日志，避免重复输出
• 建议格式：时间 | 级别 | APP | 模块:函数:行号 | 消息

示例配置

import sys
from loguru import logger

# 配置上层应用日志（在导入SDK之前）
logger.remove()  # 移除默认handler
logger.add(
    sys.stderr, 
    level="INFO",
    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level: <8} | APP | {name}:{function}:{line} | {message}",
    filter=lambda record: not record["extra"].get("sdk_internal", False),  # 排除SDK日志
)

# 然后导入SDK
from ble_sdk.client import BleClient

使用示例

1. 采集电池数据流

以下示例展示了如何连接到设备并采集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())

2. 原始数据模式示例

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())

3. 混合数据流示例

混合模式可以同时采集主数据流（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())

4. 使用命令行工具 (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

5. 算法的使用

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) 

项目结构

ble_sdk/
├── docs/                 # 文档
├── examples/             # 示例脚本
│   ├── ble_scan.py       # BLE设备扫描示例
│   ├── mixed_data_monitor.py  # 混合数据流监控示例
│   ├── sdk_cli_monitor.py     # 命令行监控工具
│   └── sdk_metrics_example.py # 性能指标示例
├── src/                  # 源码
│   └── ble_sdk/
│       ├── __init__.py   # SDK包初始化，导出公共API
│       ├── algo/         # 数据处理算法
│       │   ├── __init__.py
│       │   ├── eeg.py    # EEG数据处理算法[psd、fft]
│       │   └── impedance.py  # 阻抗转换算法
│       ├── client.py     # 核心 BleClient 类
│       ├── constants.py  # 常量定义 (UUID, 命令, 设备配置)
│       ├── exceptions.py # 自定义异常
│       ├── log.py        # SDK日志系统
│       ├── scanner.py    # BLE设备扫描功能
│       ├── utils/        # 工具模块
│       │   ├── __init__.py
│       │   └── cache.py  # 缓存装饰器和配置构建器
│       └── handles/      # 各数据流的回调处理器和解析器
│           ├── __init__.py
│           ├── battery_rssi_handler.py  # 电池和RSSI处理
│           ├── eeg_handler.py           # EEG数据处理
│           ├── impedance_handler.py     # 阻抗数据处理
│           ├── imu_handler.py           # IMU数据处理
│           └── lead_handler.py          # 导联状态处理
├── tests/                # 测试套件
├── .gitignore
├── LICENSE
├── pyproject.toml        # 项目配置文件 (PEP 621)
├── README.md
└── uv.lock               # 依赖锁定文件

API 参考

BleClient 类

BleClient 是 SDK 的核心类，提供了与 BLE 设备交互的所有功能。

主要方法

• start_eeg_stream(user_callback, raw_data_only=False) - 启动EEG数据流
• start_imu_stream(user_callback, raw_data_only=False) - 启动IMU数据流
• start_impedance_stream(user_callback, raw_data_only=False) - 启动阻抗数据流
• start_lead_stream(user_callback, raw_data_only=False) - 启动导联状态流
• start_battery_stream(user_callback) - 启动电池数据流
• start_mixed_stream(mode, user_callback, raw_main_data_only=False) - 启动混合数据流

原始数据模式

所有数据流方法都支持 raw_data_only 参数：

• raw_data_only=False（默认）：回调函数接收解析后的字典数据
• raw_data_only=True：回调函数接收去除帧头帧尾的原始 bytearray 数据

混合数据流使用 raw_main_data_only 参数控制所有数据类型的输出格式。

异常类

所有异常继承自基础的 BleError 类：

• ConnectionError - 连接失败时抛出。
• DisconnectionError - 断开连接失败时抛出。
• NotificationError - 启用或停止通知失败时抛出。
• DeviceNotFoundError - 未找到指定地址的设备时抛出。
• WriteError - 写入特性失败时抛出。

最佳实践

1. 使用异步上下文管理器：始终使用 async with 来确保正确的资源释放。这可以保证在操作完成或发生错误时，BleClient 实例能够正确地断开连接并清理资源。

2. 实现高效的回调函数：确保回调函数是异步的，并能高效地处理数据，避免阻塞主事件循环。对于原始数据模式，注意数据类型判断。

3. 适当处理异常：通过使用 try-except 块，可以优雅地处理连接失败、通知错误或设备未找到等情况，从而提高应用程序的健壮性。

4. 使用日志记录：集成一个成熟的日志系统（如 Python loguru），记录关键操作（连接尝试、通知启用/停止、数据接收、错误）的详细信息，这对于调试和监控至关重要。

5. 选择合适的数据模式：

   • 使用解析模式（默认）进行常规数据分析
   • 使用原始数据模式进行自定义数据处理或性能优化
   • 混合模式适用于需要多种数据类型同步采集的场景

6. 日志内容和函数docstring用英文：保持代码注释和文档的语言一致性，有助于国际化合作和提高可读性。

开发者指南

环境设置

# 创建并激活虚拟环境
python -m venv .venv
# On Windows: .venv\Scripts\activate
# On macOS/Linux: source .venv/bin/activate

# 安装开发依赖
pip install -e ".[dev]"

运行测试

pytest

许可证

MIT