From d1a1d43bd8f80f20b412e97ab1d5e4722325cf0e Mon Sep 17 00:00:00 2001 From: linyimin <18316471919@139.com> Date: Thu, 22 Oct 2026 22:52:23 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E8=93=9D=E7=89=99=E4=B8=BB=E4=BB=8E?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E8=84=9A=E6=9C=AC(bleak=E8=BF=9E=E6=8E=A5/bl?= =?UTF-8?q?ess=20GATT=E4=BB=8E=E6=9C=BA+bluetoothctl=E5=B9=BF=E6=92=AD)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- test/test_bluetooth.py | 135 +++++++++++++++++ test/test_bluetooth_peripheral.py | 239 ++++++++++++++++++++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 test/test_bluetooth.py create mode 100644 test/test_bluetooth_peripheral.py diff --git a/test/test_bluetooth.py b/test/test_bluetooth.py new file mode 100644 index 0000000..d62ba8c --- /dev/null +++ b/test/test_bluetooth.py @@ -0,0 +1,135 @@ +#!/usr/bin/env python3 +# test_bluetooth.py +# 参考: https://wiki.sipeed.com/maixpy/doc/zh/modules/bluetooth.html +# +# 依赖: +# 1. 使能蓝牙: bluetoothctl power on +# (可选开机自启: echo "bluetoothctl power on" >> /etc/rc.local) +# 2. 安装依赖: pip install bleak + +import asyncio +import sys +from collections import namedtuple + +try: + from bleak import BleakScanner, BleakClient +except ImportError: + print("ModuleNotFoundError: No module named 'bleak'") + print("请先在设备上安装依赖:") + print(" pip install bleak") + print("并使能蓝牙:") + print(" bluetoothctl power on") + sys.exit(1) + +MODEL_NBR_UUID = "1A2A" + +BLEDev = namedtuple("BLEDev", ["address", "name"]) + + +async def scan_devices(timeout=5.0): + """扫描周围的蓝牙设备""" + print("=" * 60) + print("Scanning BLE devices...") + print("=" * 60) + + found = {} + + def _cb(device, adv): + found[device.address] = adv.local_name or device.name + + try: + await BleakScanner.discover(timeout=timeout, detection_callback=_cb) + except Exception as e: + print(f"Scan failed: {e}") + print("Possible causes:") + print(" 1. Bluetooth not powered on: bluetoothctl power on") + print(" 2. bleak not installed: pip install bleak") + print(" 3. BLE hardware/driver not available") + return [] + + if not found: + print("No devices found.") + return [] + + devices = [] + for address, name in found.items(): + devices.append(BLEDev(address, name)) + print(f" {address} {name}") + return devices + + +async def connect_device(address, timeout=20.0, retries=3): + """连接蓝牙设备并尝试读取特征值""" + print("=" * 60) + print(f"Connecting to {address} ...") + print("=" * 60) + + for attempt in range(1, retries + 1): + try: + async with BleakClient(address, timeout=timeout) as client: + print(f"Connected: {client.is_connected}") + + # 遍历服务和特征值 + for service in client.services: + print(f" Service: {service.uuid}") + for char in service.characteristics: + print(f" Char: {char.uuid} props={list(char.properties)}") + + # 尝试读取 Model Number 特征 + try: + value = await client.read_gatt_char(MODEL_NBR_UUID) + print(f"Model Number: {value.decode(errors='replace')}") + except Exception as e: + print(f"Read {MODEL_NBR_UUID} failed (device may not have it): {e}") + + return True + except Exception as e: + print(f"[attempt {attempt}/{retries}] Connect failed: {e}") + if attempt < retries: + await asyncio.sleep(2) + + print("\nAll attempts failed. 请检查:") + print(" 1. forrily 是否已连到手机/其它主机（BLE 外设通常只允许一连接），") + print(" 先在手机上断开它") + print(" 2. 清除 BlueZ 缓存后重试:") + print(f" bluetoothctl remove {address}") + print(" systemctl restart bluetooth") + print(" bluetoothctl power on") + print(" 3. 升级依赖:") + print(" pip install -U bleak dbus-fast") + return False + + +async def main(): + target_name = "forrily" + + devices = await scan_devices(timeout=5.0) + if not devices: + return + + if len(sys.argv) > 1: + address = sys.argv[1] + else: + # 按名字查找目标设备，不连第一个 + target = None + for d in devices: + if d.name and d.name.strip().lower() == target_name.lower(): + target = d + break + if target is None: + for d in devices: + if d.name and target_name.lower() in d.name.strip().lower(): + target = d + break + if target is None: + print(f"\nDevice named '{target_name}' not found!") + print("Scanned devices are listed above.") + return + address = target.address + print(f"\nFound '{target_name}': {address}") + + await connect_device(address) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/test/test_bluetooth_peripheral.py b/test/test_bluetooth_peripheral.py new file mode 100644 index 0000000..de5c36e --- /dev/null +++ b/test/test_bluetooth_peripheral.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +# test_bluetooth_peripheral.py +# 设备(MaixCAM)作为 BLE 从机(Peripheral/GATT Server)广播给手机连接。 +# 参考: https://wiki.sipeed.com/maixpy/doc/zh/modules/bluetooth.html +# +# 用法: +# python test_bluetooth_peripheral.py [广播名字] # 默认 shoot +# +# 依赖: +# 1. 使能蓝牙: bluetoothctl power on +# 2. 安装依赖: pip install bless +# 3. 部分 BlueZ 需要 experimental 模式才支持 GATT Server: +# 编辑 /etc/bluetooth/main.conf 保证有 Experimental = true +# 然后 systemctl restart bluetooth && bluetoothctl power on +# +# 手机端: 用 nRF Connect / 任意 BLE 调试 App 搜索广播名连接， +# 读写 characteristic 并订阅 notify，每 2 秒会收到计数通知。 +# +# 实现: +# - GATT 服务由 bless 注册到 BlueZ +# - 广播由常驻 bluetoothctl 子进程控制 (menu advertise / name / advertise on)， +# 这是已验证手机能搜到的路径; 若 bluetoothctl 不可用则回退到 btmgmt adv on + +import asyncio +import shutil +import subprocess +import sys +import time + +try: + from bless import ( + BlessServer, + GATTCharacteristicProperties, + GATTAttributePermissions, + ) +except ImportError: + print("ModuleNotFoundError: No module named 'bless'") + print("请先在设备上安装依赖:") + print(" pip install bless") + print("并使能蓝牙:") + print(" bluetoothctl power on") + raise SystemExit(1) + +DEVICE_NAME = sys.argv[1] if len(sys.argv) > 1 else "shoot" +SERVICE_UUID = "A07498CA-AD5B-474E-940D-16F1FBE7E8CD" +CHAR_UUID = "51FF12BB-3ED8-46E5-B4F9-D64E2FEC021B" +BTCTL_LOG = "/tmp/btctl_adv.log" + +# 不同 bless 版本枚举名不同: writeable / writable +try: + WRITE_PERMISSION = GATTAttributePermissions.writeable +except AttributeError: + WRITE_PERMISSION = GATTAttributePermissions.writable + +_btctl = None + + +def on_read(characteristic, request): + print(f"[GATT] read {characteristic.uuid}") + return characteristic.value + + +def on_write(characteristic, value, request): + characteristic.value = value + print(f"[GATT] write {characteristic.uuid} = {bytes(value)}") + + +def on_subscribe(characteristic, session): + print(f"[GATT] client subscribed: {characteristic.uuid}") + + +def on_unsubscribe(characteristic, session): + print(f"[GATT] client unsubscribed: {characteristic.uuid}") + + +def _run(cmd): + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=10) + out = ((r.stdout or "") + (r.stderr or "")).strip() + if out: + print(f"[adv] {' '.join(cmd)} -> {out}") + return r.returncode == 0 + except FileNotFoundError: + print(f"[adv] {cmd[0]} not found") + return False + except Exception as e: + print(f"[adv] {' '.join(cmd)} failed: {e}") + return False + + +def adv_start(name): + """启动广播: 优先 bluetoothctl (常驻子进程), 失败则回退 btmgmt""" + global _btctl + + if shutil.which("bluetoothctl") is not None: + try: + log = open(BTCTL_LOG, "w") + _btctl = subprocess.Popen( + ["bluetoothctl"], + stdin=subprocess.PIPE, + stdout=log, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + except Exception as e: + print(f"[adv] start bluetoothctl failed: {e}") + _btctl = None + + if _btctl is not None: + for cmd in ( + "menu advertise", + f"name {name}", + "discoverable on", + "menu main", + "advertise on", + ): + try: + _btctl.stdin.write(cmd + "\n") + _btctl.stdin.flush() + except Exception as e: + print(f"[adv] write '{cmd}' failed: {e}") + break + time.sleep(0.3) + + time.sleep(1.0) + if _btctl.poll() is None: + print(f"[adv] bluetoothctl advertising on (log: {BTCTL_LOG})") + return True + print("[adv] bluetoothctl exited early, fallback to btmgmt") + + # 回退: mgmt 传统广播 + if shutil.which("btmgmt") is None: + print("[adv] bluetoothctl/btmgmt 均不可用") + return False + _run(["btmgmt", "le", "on"]) + _run(["btmgmt", "connectable", "on"]) + return _run(["btmgmt", "adv", "on"]) + + +def adv_stop(): + global _btctl + if _btctl is not None: + try: + _btctl.stdin.write("advertise off\n") + _btctl.stdin.flush() + time.sleep(0.3) + _btctl.terminate() + except Exception: + pass + _btctl = None + elif shutil.which("btmgmt") is not None: + _run(["btmgmt", "adv", "off"]) + + +async def run(): + gatt = { + SERVICE_UUID: { + CHAR_UUID: { + "Properties": ( + GATTCharacteristicProperties.read + | GATTCharacteristicProperties.write + | GATTCharacteristicProperties.notify + ), + "Permissions": ( + GATTAttributePermissions.readable | WRITE_PERMISSION + ), + "Value": bytearray(b"hello from maixcam"), + } + } + } + + server = BlessServer( + name=DEVICE_NAME, + on_read=on_read, + on_write=on_write, + on_subscribe=on_subscribe, + on_unsubscribe=on_unsubscribe, + ) + + print("Adding GATT ...") + await server.add_gatt(gatt) + + print(f"Starting BLE peripheral, name = '{DEVICE_NAME}' ...") + try: + await server.start() + except Exception as e: + print(f"start() failed: {e}") + print("排查:") + print(" 1. 蓝牙是否上电: bluetoothctl power on (bluetoothctl show 看 Powered)") + print(" 2. /etc/bluetooth/main.conf 加 Experimental = true 后重启:") + print(" systemctl restart bluetooth && bluetoothctl power on") + print(" 3. 依赖: pip install -U bless dbus-next bleak") + return + + adv = await server.is_advertising() + print("=" * 60) + print(f"'{DEVICE_NAME}' advertising status: {adv}") + + print("Starting advertising ...") + if not adv_start(DEVICE_NAME): + print("广播启动失败，请手动执行: bluetoothctl -> advertise on") + + print("手机用 nRF Connect 等 App 搜索并连接，") + print(f"服务 {SERVICE_UUID}") + print(f"特征 {CHAR_UUID} (read/write/notify)") + print("Ctrl+C 退出") + print("=" * 60) + + try: + # 每 2 秒向订阅者推送一次计数 + counter = 0 + while True: + counter += 1 + try: + char = server.get_characteristic(CHAR_UUID) + if char is not None: + char.value = bytearray(str(counter).encode()) + server.update_value(SERVICE_UUID, CHAR_UUID) + except Exception as e: + # 没有客户端订阅时可能失败，忽略 + print(f"notify failed: {e}") + + if counter % 10 == 0: + try: + print(f"[status] advertising={await server.is_advertising()} " + f"connected={await server.is_connected()}") + except Exception: + pass + await asyncio.sleep(2) + finally: + adv_stop() + + +if __name__ == "__main__": + try: + asyncio.run(run()) + except KeyboardInterrupt: + print("\nStopped.")