136 lines
4.2 KiB
Python
136 lines
4.2 KiB
Python
#!/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())
|