保姆级教程:在Ubuntu 20.04上用Python和Bluez 5.58实现BLE心率数据模拟与推送
·
在Ubuntu 20.04上构建BLE心率模拟器的完整实践指南
当你第一次接触蓝牙低功耗(BLE)开发时,可能会被各种协议栈和工具链搞得晕头转向。本文将带你从零开始,在Ubuntu 20.04系统上,使用Python和Bluez 5.58构建一个完整的心率数据模拟器。这个项目不仅能让你理解BLE的核心工作机制,还能掌握如何将普通电脑变成智能穿戴设备的原型开发平台。
1. 环境准备与Bluez安装
在开始编码前,我们需要确保开发环境配置正确。Ubuntu 20.04默认的Bluez版本是5.53,而我们需要5.58版本来获得完整的GATT服务器支持。
首先安装编译依赖:
sudo apt update
sudo apt install -y libdbus-1-dev libudev-dev libical-dev libreadline-dev python3-dbus
下载并编译Bluez 5.58:
wget https://www.kernel.org/pub/linux/bluetooth/bluez-5.58.tar.xz
tar xvf bluez-5.58.tar.xz
cd bluez-5.58
./configure --prefix=/usr --mandir=/usr/share/man --sysconfdir=/etc --localstatedir=/var --enable-experimental
make -j$(nproc)
sudo make install
验证安装是否成功:
bluetoothd --version
提示:如果遇到权限问题,可能需要重启蓝牙服务:
sudo systemctl restart bluetooth
2. BLE心率服务原理剖析
在深入代码前,理解心率服务的标准规范至关重要。蓝牙SIG定义了心率服务的标准UUID和数据结构:
- 服务UUID:0x180D(心率服务)
- 特征值UUID:0x2A37(心率测量)
心率数据包的格式如下:
| 位偏移 | 长度 | 说明 |
|---|---|---|
| 0 | 1 | 标志位(bit0:心率格式;bit1:传感器接触状态) |
| 1 | 1-2 | 心率值(8位或16位,取决于标志位) |
| 2/3 | 可变 | 可选的能量消耗和RR间隔数据 |
典型的心率数据包示例:
# 16位心率值,传感器接触检测到
flags = 0x06 # 00000110
heart_rate = 72 # 假设心率为72bpm
value = [flags, heart_rate & 0xFF, (heart_rate >> 8) & 0xFF]
3. 构建GATT服务器
我们将基于Bluez提供的example-gatt-server.py进行修改。首先获取示例代码:
cp bluez-5.58/test/example-gatt-server.py ~/heartrate-server.py
chmod +x ~/heartrate-server.py
关键修改点在于HeartRateService类:
class HeartRateService(Service):
HR_UUID = '0000180d-0000-1000-8000-00805f9b34fb'
def __init__(self, bus, index):
Service.__init__(self, bus, index, self.HR_UUID, True)
self.add_characteristic(HeartRateMeasurementChrc(bus, 0, self))
self.add_characteristic(BodySensorLocationChrc(bus, 1, self))
心率测量特征值的实现:
class HeartRateMeasurementChrc(Characteristic):
HR_MSRMT_UUID = '00002a37-0000-1000-8000-00805f9b34fb'
def __init__(self, bus, index, service):
Characteristic.__init__(
self, bus, index, self.HR_MSRMT_UUID,
['notify'], service)
self.notifying = False
self.hr_value = 72
self.timer = None
def _update_hr_value(self):
# 模拟心率波动:60-100之间的随机漫步
change = random.choice([-2, -1, 0, 1, 2])
self.hr_value = max(60, min(100, self.hr_value + change))
flags = 0x06 # 16-bit value, sensor contact detected
value = [flags, self.hr_value & 0xFF, (self.hr_value >> 8) & 0xFF]
self.PropertiesChanged(GATT_CHRC_IFACE, {'Value': value}, [])
return self.notifying
def StartNotify(self):
if self.notifying:
return
self.notifying = True
self.timer = GLib.timeout_add(2000, self._update_hr_value) # 每2秒更新一次
def StopNotify(self):
if not self.notifying:
return
self.notifying = False
if self.timer:
GLib.source_remove(self.timer)
4. 运行与测试
启动GATT服务器:
python3 heartrate-server.py
在另一终端验证服务是否注册成功:
bluetoothctl
[bluetooth]# list-attributes
使用nRF Connect进行测试:
- 在手机上安装nRF Connect应用
- 扫描并连接你的电脑(显示为"Test Service")
- 在服务列表中找到心率服务(180D)
- 点击心率测量特征值(2A37)的"Enable notifications"按钮
你应该会看到每2秒更新一次的心率数据。为了验证数据真实性,可以修改代码增加更复杂的心率模拟算法:
def _update_hr_value(self):
# 更真实的心率模拟:休息时60-70,活动时80-100
if random.random() < 0.2: # 20%概率切换到活动状态
self.active = not self.active
if self.active:
self.hr_value = random.randint(80, 100)
else:
self.hr_value = random.randint(60, 70)
flags = 0x06
value = [flags, self.hr_value & 0xFF, (self.hr_value >> 8) & 0xFF]
self.PropertiesChanged(GATT_CHRC_IFACE, {'Value': value}, [])
return self.notifying
5. 高级功能扩展
5.1 添加RR间隔数据
真实的心率监测设备通常会包含RR间隔(心跳间期)数据。修改数据生成逻辑:
def _update_hr_value(self):
flags = 0x16 # 第4位表示包含RR间隔
hr_value = random.randint(60, 100)
rr_interval = int(60000 / hr_value) # 计算理论RR间隔(ms)
value = [flags, hr_value & 0xFF]
rr_bytes = rr_interval.to_bytes(2, byteorder='little')
value.extend(rr_bytes)
self.PropertiesChanged(GATT_CHRC_IFACE, {'Value': value}, [])
5.2 实现写操作支持
某些应用可能需要配置心率监测参数。添加写操作支持:
@dbus.service.method(GATT_CHRC_IFACE, in_signature='aya{sv}')
def WriteValue(self, value, options):
if len(value) < 1:
return
flags = value[0]
if flags & 0x01: # 8位心率值
self.hr_value = value[1]
else: # 16位心率值
self.hr_value = value[1] | (value[2] << 8)
print(f"Heart rate threshold set to: {self.hr_value}")
5.3 多客户端支持
默认实现只能通知最后一个连接的客户端。改进为支持多客户端:
def __init__(self, bus, index, service):
# ...原有代码...
self.clients = set()
def StartNotify(self):
# 获取客户端地址
sender = self._connection.get_unique_name()
self.clients.add(sender)
if not self.timer:
self.timer = GLib.timeout_add(2000, self._update_hr_value)
def StopNotify(self):
sender = self._connection.get_unique_name()
self.clients.discard(sender)
if not self.clients and self.timer:
GLib.source_remove(self.timer)
self.timer = None
6. 性能优化与调试技巧
当系统负载较高时,通知可能会延迟。我们可以通过以下方式优化:
- 调整DBus超时时间:
bus = dbus.SystemBus(private=True)
bus.set_timeout(5000) # 5秒超时
- 使用更高效的序列化方法:
def _update_hr_value(self):
# 使用memoryview减少拷贝
value = bytearray(5)
mv = memoryview(value)
mv[0] = 0x06
mv[1:3] = self.hr_value.to_bytes(2, 'little')
# ...其余代码...
- 监控系统蓝牙状态:
sudo btmon -w bluetooth.log
常见问题排查:
- 服务注册失败:检查是否以root权限运行,或确保用户属于bluetooth组
- 通知不工作:确认特征值属性包含'notify',且客户端已启用通知
- 连接不稳定:检查系统电源管理是否关闭(
sudo powertop)
在实际项目中,我发现最常出现的问题是DBus连接超时。一个可靠的解决方案是添加重试机制:
def safe_properties_changed(self, interface, changed, invalidated):
try:
self.PropertiesChanged(interface, changed, invalidated)
except dbus.exceptions.DBusException as e:
if 'Timeout' in str(e):
print("DBus timeout, retrying...")
GLib.timeout_add(100, lambda: self.safe_properties_changed(interface, changed, invalidated))
更多推荐



所有评论(0)