test: 蓝牙主从测试脚本(bleak连接/bless GATT从机+bluetoothctl广播)

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

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.")