240 lines
7.4 KiB
Python
240 lines
7.4 KiB
Python
#!/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.")
|