本文旨在技术性地探讨如何利用 Python 脚本与支持蓝牙低功耗(Bluetooth Low Energy, BLE)心率广播功能的设备进行交互,以实时获取心率数据。我们将重点分析涉及的 BLE 协议、GATT 服务模型以及使用 bleak
库的具体 Python 实现,并说明为何像部分华为手表(HUAWEI WATCH)等设备能通过此标准方式被监测。
技术背景:BLE 与 GATT 心率服务
1. BLE 角色与通信模式:
在本次应用中,Python 脚本运行的计算机作为 Central(中心设备),而心率监测设备(如手表、心率带)作为 Peripheral(外设)。我们利用的是 BLE 的 广播(Advertising) 和 连接(Connection) 模式。支持心率广播的外设会主动发出广播包(Advertising Packets),其中通常包含设备信息和其支持的关键服务 UUID。
2. GATT (Generic Attribute Profile):
GATT 是 BLE 协议栈上层定义数据交换格式和结构的标准。数据被组织在服务(Services)和特征(Characteristics)中:
- Heart Rate Service (HRS): 由蓝牙技术联盟(Bluetooth SIG)定义的标准服务,UUID 为
0x180D
。外设通过在广播包中包含此 UUID 或在连接后暴露此服务来表明其支持心率测量功能。 - Heart Rate Measurement Characteristic: HRS 内的核心特征,UUID 为
0x2A37
。该特征的值包含了实际的心率测量数据。它的关键属性(Property)是 Notify。 - Notify 属性: 对于像心率这样实时变化的数据,Notify 机制最为高效。Central 设备向 Peripheral 设备订阅(Subscribe)此特征后,每当 Peripheral 的心率数据更新时,它会主动将新数据通过通知(Notification)发送给 Central,而无需 Central 不断轮询(Read)。
3. Heart Rate Measurement 数据结构 (UUID 0x2A37
):
根据 Bluetooth SIG 规范,此特征的数据包通常结构如下:
- Byte 0: Flags:
- Bit 0: 心率值格式 (0 = UINT8, 1 = UINT16)。
- Bits 1-2: 传感器接触状态。
- Bit 3: 能耗指示是否存在。
- Bit 4: RR-Interval 数据是否存在。
- Byte 1 onwards:
- 心率值 (根据 Flags Bit 0 决定是 1 字节还是 2 字节,小端序 Little-Endian for UINT16)。
- (可选) 能耗值 (UINT16)。
- (可选) RR-Interval 值序列 (每项 UINT16,单位 1/1024 秒)。
Python 实现 (使用 bleak
库)
bleak
是一个基于 asyncio
的现代 Python 库,提供了跨平台(Windows, macOS, Linux)的 BLE Central 功能接口。
核心代码解析:
import asyncio
from bleak import BleakScanner, BleakClient, BleakError
# 标准蓝牙心率服务和特征 UUID
HEART_RATE_SERVICE_UUID = "0000180d-0000-1000-8000-00805f9b34fb"
HEART_RATE_MEASUREMENT_CHAR_UUID = "00002a37-0000-1000-8000-00805f9b34fb"
# 数据处理回调
def handle_heart_rate_notification(data: bytearray):
"""
回调函数,解析来自 0x2A37 特征的通知数据.
sender: 在 Windows 上是特征句柄 (handle), 其他平台可能是 0.
data: 原始字节数据 (bytearray).
"""
try:
flags = data[0]
hr_format_is_uint16 = (flags & 0x01)
# 根据 Flag bit 0 解析心率值
if hr_format_is_uint16:
# UINT16 format, bytes 1 and 2, Little-Endian
heart_rate_value = int.from_bytes(data[1:3], byteorder='little')
else:
# UINT8 format, byte 1
heart_rate_value = data[1]
print(f"Heart Rate: {heart_rate_value} bpm")
# 可选: 解析 RR-Intervals (如果 Flag bit 4 被设置)
# rr_interval_present = (flags >> 4) & 0x01
# if rr_interval_present:
# offset = 3 if hr_format_is_uint16 else 2
# rr_intervals_ms = []
# while offset < len(data):
# rr_raw = int.from_bytes(data[offset:offset+2], byteorder='little')
# rr_intervals_ms.append(round((rr_raw / 1024.0) * 1000.0)) # Convert to ms
# offset += 2
# if rr_intervals_ms:
# print(f" RR Intervals (ms): {rr_intervals_ms}")
except IndexError:
print(f"Error: Received incomplete data: {data.hex()}")
except Exception as e:
print(f"Error parsing HR data: {e}, Raw: {data.hex()}")
# 主异步任务
async def run_hr_monitor():
print("Scanning for devices advertising Heart Rate Service...")
# 1. 扫描: 使用 BleakScanner 查找广播了 HRS UUID 的设备
try:
device = await BleakScanner.find_device_by_filter(
lambda d, ad: HEART_RATE_SERVICE_UUID.lower() in ad.service_uuids,
timeout=10.0
)
except BleakError as e:
print(f"Bluetooth scanning error: {e}")
device = None # Fallback or handle specific errors
# 也可以使用 discover 获取列表供选择
# devices = await BleakScanner.discover(service_uuids=[HEART_RATE_SERVICE_UUID], timeout=10.0)
# ... (device selection logic) ...
if device is None:
print("No device advertising the Heart Rate Service found.")
print("Ensure the device's HR broadcasting is enabled in its settings.")
return
print(f"Found device: {device.name} ({device.address})")
print("Connecting...")
# 2. 连接与交互: 使用 BleakClient
async with BleakClient(device.address, timeout=20.0) as client:
if not client.is_connected:
print("Failed to connect.")
return
print("Connected successfully!")
try:
# 3. 订阅通知: 关键步骤
print(f"Subscribing to Heart Rate Measurement notifications ({HEART_RATE_MEASUREMENT_CHAR_UUID})...")
await client.start_notify(HEART_RATE_MEASUREMENT_CHAR_UUID, handle_heart_rate_notification)
print("Subscription successful. Waiting for data... (Press Ctrl+C to stop)")
# 保持运行以接收通知
while client.is_connected:
await asyncio.sleep(1.0) # Keep event loop alive
except BleakError as e:
print(f"Bluetooth operation error: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
finally:
# 4. 清理: 停止通知 (通常在 async with 退出时自动处理部分断连)
if client.is_connected:
try:
await client.stop_notify(HEART_RATE_MEASUREMENT_CHAR_UUID)
print("Notifications stopped.")
except BleakError as e:
print(f"Error stopping notifications: {e}")
# 运行入口
if __name__ == "__main__":
# 运行 asyncio 事件循环
try:
asyncio.run(run_hr_monitor())
except KeyboardInterrupt:
print("\nMonitoring stopped by user.")
except Exception as e:
# Catch potential top-level errors (e.g., asyncio issues)
print(f"\nTop-level error: {e}")
关键实现点:
BleakScanner.find_device_by_filter
/discover
: 用于基于广播的服务 UUID (0x180D
) 识别潜在的心率设备。这是利用 BLE 广播进行服务发现的核心。BleakClient(address)
: 建立到目标设备的 GATT 连接。client.start_notify(UUID, callback)
: 向外设请求开启指定特征 (0x2A37
) 的通知,并将接收到的数据包传递给handle_heart_rate_notification
回调函数处理。这是获取实时数据的关键。- Data Parsing (
handle_heart_rate_notification
): 严格按照 Bluetooth SIG 对0x2A37
特征数据格式的定义来解析bytearray
,提取心率值和其他可选信息。 asyncio
:bleak
依赖asyncio
进行异步操作管理,允许程序在等待蓝牙 I/O 时保持响应。
关于华为手表(及类似设备)的兼容性
华为手表或其他品牌设备能否被此脚本监测,取决于该设备是否遵循了 Bluetooth SIG 定义的标准 Heart Rate Service (UUID 0x180D
) 规范,并且是否提供了“心率广播”或类似的模式。
这种模式意味着:
- 设备在其 BLE 广播包中宣告 HRS 服务,或者在连接后提供该服务。
- 允许中心设备在无需强制配对的情况下连接并订阅 Heart Rate Measurement 特征 (
0x2A37
) 的通知。
用户必须在设备端(通常是手表设置 如华为手表的设置-心率广播)手动开启此“心率广播”功能。 这不是协议层面的特殊处理,而是设备固件提供的一个标准功能开关。如果设备遵循标准且此功能已开启,上述 Python 脚本就能与之正常通信。但如小米手环等设备,因其进行了数据加密,无法直接通过蓝牙协议进行通信。
技术考量
- 异步编程模型: 理解
asyncio
的事件循环、async/await
语法对于调试和扩展bleak
应用至关重要。 - 权限: 运行 BLE 扫描和连接通常需要相应的系统权限。
- 连接稳定性与错误处理: 实际应用中需处理连接丢失 (
client.is_connected
检查)、设备无响应、数据解析错误等多种异常情况。
总结
通过利用 Python 的 bleak
库和 asyncio
,开发者可以有效地与实现了标准 BLE Heart Rate Service 并支持广播模式的设备进行通信。核心在于理解 GATT 服务发现、特征订阅(特别是 Notify 属性)以及按照官方规范解析特征数据。设备(如兼容的华为手表)的可用性依赖于其对蓝牙标准的遵循和用户侧相应功能的启用。