[玩转树莓派CM0 之 BLE] Python 与 BLE 设备通信指南 -- P1

前言

树莓派 CM0 有带无线功能与不带无线功能两个版本, 具体可以参考如下型号说明:

或者查看你的板子中是否带有无线模块

如果你的 CM0 具备无线功能, 则可以继续按照本教程的后续步骤进行测试.

需要知道的基础知识

在BLE通信中, 设备工作于主从模式:

• 主设备可主动扫描并连接周围从设备的广播信号.
• 从设备则通过广播自身信号, 被动等待主设备发起连接.

用一个简单的比喻, 这就像美食街里有许多商贩在叫卖 (从设备广播) , 而顾客 (主设备) 听到感兴趣的吆喝后, 可以主动走过去交易. 商贩只能等待顾客上门, 而不能主动将商品塞给顾客.

在本次实验中, 树莓派CM0将会作为主设备, 对从设备进行操作.

实际上, 两片树莓派CM0之间也可设置为一主一从进行通信, 但受限于我只有一块CM0, 本次实验暂不演示该场景.

下一步安排: 为便于理解与实践, 下一章我们将以常见的小米温湿度计作为从设备, 完成从发现, 连接到数据读取的全过程演示.

环境准备与配置

启用蓝牙电源

输入bluetoothctl进入蓝牙管理交互界面

root@rpi-cm0:~# bluetoothctl hci0 new_settings: bondable ssp br/edr le secure-conn Agent registered [CHG] Controller AA:CC:DD:11:22:33 Pairable: yes [bluetoothctl]> 

开启蓝牙射频电源

在bluetoothctl交互界面中, 输入power on以开启蓝牙射频电源

[bluetoothctl]> power on [CHG] Controller AA:CC:DD:11:22:33 PowerState: off-enabling hci0 class of device changed: 0x400000 [CHG] Controller AA:CC:DD:11:22:33 Class: 0x00400000 (4194304) hci0 new_settings: powered bondable ssp br/edr le secure-conn Changing power on succeeded [CHG] Controller AA:CC:DD:11:22:33 PowerState: on [CHG] Controller AA:CC:DD:11:22:33 Powered: yes [bluetoothctl]> 

完成后, 输入 exit退出

[bluetoothctl]> exit root@rpi-cm0:~# 

检查并解除蓝牙射频锁定

查看当前状态

输入命令rfkill list查看状态, 观察Bluetooth项.

若Soft blocked为yes则需要解锁

root@rpi-cm0:~# rfkill list 0: hci0: Bluetooth Soft blocked: yes # 若为 yes, 表示被软锁定 Hard blocked: no 1: phy0: Wireless LAN Soft blocked: no Hard blocked: no 

解除射频锁定

命令rfkill unblock bluetooth, 无错误输出则为解锁成功.

root@rpi-cm0:~# rfkill unblock bluetooth 

确认解锁状态

再次运行rfkill list确认Soft blocked已变为no

root@rpi-cm0:~# rfkill list 0: hci0: Bluetooth Soft blocked: no # 若为 no, 表示已解锁 Hard blocked: no 1: phy0: Wireless LAN Soft blocked: no Hard blocked: no 

安装Python BLE相关库

本教程使用bleak库进行蓝牙开发. 为避免包冲突, 建议通过系统包管理器安装

root@rpi-cm0:~# apt update root@rpi-cm0:~# apt install python3-bleak 

BLE 设备发现与连接

扫描周围BLE设备

代码

以下脚本将扫描并列出附近所有 BLE 设备及其信号强度 (RSSI)

import asyncio from bleak import BleakScanner async def main(): print("正在扫描附近的 BLE 设备 (持续 5 秒)...") found_devices = await BleakScanner.discover(timeout=5.0, return_adv=True) print(f"{'MAC 地址':<20} {'设备名称':<20} {'RSSI'}") print("-" * 50) for device, advertisement_data in found_devices.values(): # 获取设备名称 name = device.name if device.name else "Unknown" # 获取RSSI rssi = advertisement_data.rssi print(f"{device.address:<20} {name:<20} {rssi} dBm") if __name__ == "__main__": asyncio.run(main()) 

部分要点解析

• BLE设备为节省电力会间歇性地发送广播信号. 设置足够的扫描时间 (如上面的5秒) 可以覆盖多个广播周期, 提高设备发现率.

列出指定BLE设备服务与特征列表

方式一: 通过MAC地址指定设备 (推荐, 大多数情况下MAC地址唯一)

import asyncio from bleak import BleakScanner, BleakClient # 目标设备的 MAC 地址 TARGET_ADDRESS = "AA:CC:22:66:55:88" async def main(): print("开始扫描设备...") device = await BleakScanner.find_device_by_filter( lambda d, ad: d.address and d.address == TARGET_ADDRESS ) if not device: print(f"未找到 MAC 地址为 {TARGET_ADDRESS} 的设备") return print(f"找到设备: {device.address}, 正在连接...") async with BleakClient(device) as client: print(f"已连接: {client.is_connected}") print("\n--- 服务发现开始 ---") for service in client.services: print(f"[服务] UUID: {service.uuid}") for char in service.characteristics: print(f" └── [特征] UUID: {char.uuid}") print(f" 属性: {char.properties}") print("--- 服务发现结束 ---\n") if __name__ == "__main__": asyncio.run(main()) 

方式二: 通过设备名称指定设备

import asyncio from bleak import BleakScanner, BleakClient # 目标设备的名称 TARGET_NAME = "LYWSD03MMC" async def main(): print("开始扫描设备...") device = await BleakScanner.find_device_by_filter( lambda d, ad: d.name and d.name == TARGET_NAME ) if not device: print(f"未找到名为 {TARGET_NAME} 的设备") return print(f"找到设备: {device.address}, 正在连接...") async with BleakClient(device) as client: print(f"已连接: {client.is_connected}") print("\n--- 服务发现开始 ---") for service in client.services: print(f"[服务] UUID: {service.uuid}") for char in service.characteristics: print(f" └── [特征] UUID: {char.uuid}") print(f" 属性: {char.properties}") print("--- 服务发现结束 ---\n") if __name__ == "__main__": asyncio.run(main()) 

部分要点解析

• 不管是通过MAC地址还是设备名称指定设备, 都可以使用BleakScanner.find_device_by_filter方法来扫描和连接BLE设备, 区别在于筛选条件不同.
• BLE的设备名称不唯一, 因此通过名称指定设备时可能会出现多个设备同时被找到的情况. 在实际使用中建议通过MAC地址指定设备.

数据读写操作

读取指定BLE设备指定特征数据

连接设备后, 可通过主动读取 (Read)或订阅通知 (Notify)两种方式获取特征值中的数据.

若特征仅支持读取, 请注释掉通知订阅部分代码.

代码

import asyncio import struct from bleak import BleakClient # ================= 配置区域 ================= # 1. 设备 MAC 地址 DEVICE_ADDRESS = "AA:CC:22:66:55:88" # 2. 待读取的数据特征 UUID SENSOR_CHAR_UUID = "8edfffef-3d1b-9c37-4623-ad7265f14076" # =========================================== # 处理实时通知的回调函数 def notification_handler(sender, data: bytearray): """ 当传感器主动推送数据时, 会触发这个函数 sender: 发送者句柄 data: 接收到的原始字节数据 """ print(f"[通知] 收到数据 (Hex): {data.hex()}") async def main(): print(f"正在连接设备 {DEVICE_ADDRESS} ...") try: async with BleakClient(DEVICE_ADDRESS) as client: if client.is_connected: print(f"成功连接到: {DEVICE_ADDRESS}") # --- 方式 A: 主动读取一次 (Read) --- try: print("正在尝试主动读取...") val = await client.read_gatt_char(SENSOR_CHAR_UUID) print(f"[读取] 原始数据: {val.hex()}") except Exception as e: print(f"[读取] 失败 (可能该特征不支持读取): {e}") # --- 方式 B: 开启实时通知 (Notify) --- print("正在开启实时通知 (监听 10 秒)...") # 开启通知, 绑定回调函数 await client.start_notify(SENSOR_CHAR_UUID, notification_handler) # 让程序保持运行 10 秒, 给传感器发送数据的时间 await asyncio.sleep(10) # 关闭通知 await client.stop_notify(SENSOR_CHAR_UUID) print("通知已关闭") except Exception as e: print(f"连接或操作发生错误: {e}") if __name__ == "__main__": asyncio.run(main()) 

部分要点解析

• BLE设备的特征值有读取(Read), 写入(Write), 通知(Notify)等属性, 一个特征值可以有多个属性.
• 特征值的属性可以用上一节列出指定BLE设备服务与特征列表提供的代码来查看.

写入指定BLE设备指定特征数据

代码

尝试写入数据Hello Raspberry Pi CM0!到指定设备.

import asyncio from bleak import BleakScanner, BleakClient # ================= 配置区域 ================= # 1. 设备 MAC 地址 TARGET_ADDRESS = "AA:CC:22:66:55:88" # 2. 待写入的数据特征 UUID CHARACTERISTIC_UUID = "0000fff2-0000-1000-8000-00805f9b34fb" # =========================================== async def main(): print("开始扫描设备...") device = await BleakScanner.find_device_by_filter( lambda d, ad: d.address and d.address == TARGET_ADDRESS ) if not device: print(f"未找到 MAC 地址为 {TARGET_ADDRESS} 的设备") return print(f"找到设备: {device.address}, 正在连接...") async with BleakClient(device) as client: print(f"已连接: {client.is_connected}") # 尝试写入数据 try: data_to_send = bytes("Hello Raspberry Pi CM0!", 'utf-8') print(f"尝试写入 UUID: {CHARACTERISTIC_UUID}") await client.write_gatt_char(CHARACTERISTIC_UUID, data_to_send) print("写入成功！") except Exception as e: print(f"写入失败: {e}") if __name__ == "__main__": asyncio.run(main()) 

效果