树莓派 CM0 Python BLE 设备通信指南

完成后，输入 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())

效果