前提

1. 心率设备需支持心率广播功能
2. 心率广播功能需处于开启状态
3. 心率广播功能最多只能同时连接一台设备，如有其它设备正在连接需要先断开

使用方法

1. 安装蓝牙低功耗库<code>break

pip install bleak

2. 开启心率广播功能后设备界面会显示当前设备名称，将设备名称替换到变量<code>device_name
3. 运行程序，将自动扫描设备并连接到心率广播功能，异步获取心率数据

代码

import asyncio

from bleak import BleakClient, BleakScanner

from bleak.backends.characteristic import BleakGATTCharacteristic

# 设备名称

device_name: str = "HUAWEI WATCH HR-341"

# 心率

value = 0

# 最大连接超时时间

timeout = 10

# 通知处理器函数，当收到通知时调用

def notification_handler(characteristic: BleakGATTCharacteristic, data: bytearray):

global value

value = int(data.hex().split('06')[1], 16)

print(value)

# 异步函数：扫描设备并根据本地名称获取地址

async def scan_for_device(device_name: str):

devices = await BleakScanner.discover()

for device in devices:

if device.name == device_name:

return device

return None

# 异步函数：查找描述为“Heart Rate Measurement”的特征UUID

async def find_heart_rate_measurement_uuid(client: BleakClient):

for service in client.services:

for characteristic in service.characteristics:

if "Heart Rate Measurement" in characteristic.description:

return characteristic.uuid

return None

# 异步主函数

async def main():

print("Starting scan...")

# 根据本地名称扫描设备，也可以直接填入设备的物理地址

device = await scan_for_device(device_name)

if device is None:

print(f"Device with name {device_name} not found.")

return

print(f"Found device: {device.name} ({device.address})")

# 创建一个事件对象，用于处理断开连接的情况

disconnected_event = asyncio.Event()

# 断开连接时的回调函数

def disconnected_callback(client):

print("Disconnected callback called!")

disconnected_event.set()

print("Connecting to device...")

async with BleakClient(device, disconnected_callback=disconnected_callback, timeout=timeout) as client:

print("Connected")

# 查找“Heart Rate Measurement”特征UUID(心率广播)

hr_measurement_uuid = await find_heart_rate_measurement_uuid(client)

if hr_measurement_uuid:

# 开始通知，并设置通知处理器

await client.start_notify(hr_measurement_uuid, notification_handler)

# 等待断开连接事件

await disconnected_event.wait()

else:

print("Heart Rate Measurement characteristic not found.")

# 运行异步主函数

asyncio.run(main())