基于Python蓝牙低功耗库bleak获及时获取心率广播心率 - ToB企服应用市场:ToB评测及商务社交产业平台

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

复制代码