BLE+Python3：服务器响应客户端但无法返回MAC地址的问题

我有Python BLE客户端（运行在PC-1）和服务器（运行在PC-2），客户端尝试读取PC-2的MAC地址时，服务器能触发ReadValue方法（终端打印MAC-Adress-Function getting cast by client...），但客户端始终收不到返回数据。

服务器端相关代码与输出 #

MAC地址特征实现代码 #

# GetMacAdress   
class MACaddressCharacteristic(Characteristic):
    macUUID = "10001802-0000-1000-8000-00805f9b34fb"

    def __init__(self, bus, index, service):
        Characteristic.__init__(
            self, bus, index, 
            self.macUUID, 
            ["read"], 
            service)

    def ReadValue(self, options):
        print("MAC-Adress-Function getting cast by client...")
        return [ 'Test' ]

服务器端终端输出 #

root@emblinux:/tmp# python3 Agent_creator.py 
Enter the name under which you want to find the device:GATEway
2023-11-14 12:12:04,381 - __main__ - INFO - Registering GATT application...
GetManagedObjects
org.bluez.GattService1 interface implementation
UUID of properties: 10001801-0000-1000-8000-00805f9b34fb
Primary of properties: True
get back path: /org/bluez/example/service1
get back path: /org/bluez/example/service1
2023-11-14 12:12:04,417 - __main__ - INFO - GetAll
2023-11-14 12:12:04,419 - __main__ - INFO - returning props
2023-11-14 12:12:04,426 - __main__ - INFO - GATT application registered
2023-11-14 12:12:04,429 - __main__ - INFO - Advertisement registered
MAC-Adress-Function getting cast by client...

客户端相关代码与输出 #

请求MAC地址的代码片段 #

async def talk_to_device(deviceAddress):

    async with BleakClient(deviceAddress) as client:
        await client.is_connected()

        # Set up the main loop.
        DBusGMainLoop(set_as_default=True)
        global bus
        bus = dbus.SystemBus()
        global mainloop
        mainloop = GLib.MainLoop()

        om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'), DBUS_OM_IFACE)
        objects = om.GetManagedObjects()
        chrcs = []

        # List characteristics found
        for path, interfaces in objects.items():
            if GATT_CHRC_IFACE not in interfaces.keys():
                continue
            chrcs.append(path)
    
        # List sevices found
        for path, interfaces in objects.items():
            if GATT_SERVICE_IFACE not in interfaces.keys():
                continue

            chrc_paths = [d for d in chrcs if d.startswith(path + "/")]

            if process_di_service(path, chrc_paths):
                break

        if not svcUUID:
            print('No Device Service found')
            sys.exit(1)

        while True:
            print("______________________________________________________________________")
            print("Which information do you want?:")
            print("1: Device MAC-Address")
            print("2: Device name")
            print("3: serial number")
            print("4: ip-address")
            print("else:")
            print("----------------------------------------------------------------------")
            print("5: newConnection")
            print("6: close")
            print("______________________________________________________________________")
            command = input("type 1,2,3,4,5 or 6: ")
            print("______________________________________________________________________")
            
            if command == "1":
                print("Asking for Mac-Address...")
                macUUID[0].ReadValue({}, reply_handler=mac_cb,
                                    error_handler=generic_error_cb,
                                dbus_interface=GATT_CHRC_IFACE)
            elif command == "2":
                print("Asking for Device-Name...")
                print("Code")
            elif command == "3":
                print("Asking for Serial-Number...")
                print("Code")
            elif command == "4":
                print("Asking for IP-Adress...")
                print("Code")
            elif command == "5":
                return True
            elif command == "6":
                closingProgram()
            else:
                print("Invalid command")
                continue

客户端终端输出 #

ssv@jsz:~/Documents/Praxisphase/ProjektEmbededSystemConfigurationApp/BluetoothConnectionViaBluez$ sudo python3 client.py 
Searching for devices....
______________________________________________________________________
1: name:60-21-CA-FC-64-A8 | address: 60:21:CA:FC:64:A8
2: name:55-8B-98-49-5F-F0 | address: 55:8B:98:49:5F-F0
3: name:47-CC-0E-74-C0-36 | address: 47:CC:0E:74:C0:36
4: name:GATEway | address: 94:54:93:56:57:8C
5: name:D1-8F-34-77-62-6F | address: D1:8F:34:77:62:6F
6: name:EE-6B-EB-56-F4-29 | address: EE:6B:EB:56:F4:29
______________________________________________________________________
7: rescan
8: close
Name the device you want to connect with, (type number): 4
/home/ssv/Documents/Praxisphase/ProjektEmbededSystemConfigurationApp/BluetoothConnectionViaBluez/client.py:106: FutureWarning: is_connected has been changed to a property. Calling it as an async method will be removed in a future version
  await client.is_connected()
Device Information Service found: /org/bluez/hci0/dev_94_54_93_56_57_8C/service008d
______________________________________________________________________
Which information do you want?:
1: Device MAC-Address
2: Device name
3: serial number
4: ip-address
else:
----------------------------------------------------------------------
5: newConnection
6: close
______________________________________________________________________
type 1,2,3,4,5 or 6: 1
______________________________________________________________________
Asking for Mac-Address...

解决思路 #

• 修正服务器返回格式：Bluez要求GATT特征的ReadValue方法返回字节数组格式，当前返回的字符串列表不符合要求。修改为：

  def ReadValue(self, options):
      print("MAC-Adress-Function getting cast by client...")
      # 转为字节数组，示例返回模拟MAC地址
      return bytearray("AA:BB:CC:DD:EE:FF", 'utf-8')
      # 或逐个字符转ASCII码列表
      # return [ord(c) for c in "AA:BB:CC:DD:EE:FF"]
  

• 启动客户端DBus主循环：客户端创建了GLib.MainLoop()但未启动，DBus的异步回调（如mac_cb）需要主循环运行才能处理返回数据。可将输入逻辑放在线程中避免阻塞，再启动主循环：

  import threading
  
  def input_loop():
      while True:
          # 原输入交互逻辑代码
          pass
  
  # 在talk_to_device函数中添加
  threading.Thread(target=input_loop, daemon=True).start()
  mainloop.run()
  

• 统一客户端API使用方式：当前同时混用Bleak和直接DBus调用，易导致上下文冲突。建议二选一：
  • 用Bleak异步API读取特征：

    if command == "1":
        print("Asking for Mac-Address...")
        value = await client.read_gatt_char("10001802-0000-1000-8000-00805f9b34fb")
        print(f"Received MAC: {value.decode('utf-8')}")
    

  • 全用DBus接口，确保上下文一致。
• 验证回调函数实现：确认mac_cb已正确定义，能处理返回数据：

  def mac_cb(value):
      mac_str = bytes(value).decode('utf-8')
      print(f"Received MAC Address: {mac_str}")
  
  def generic_error_cb(error):
      print(f"Error occurred: {error}")
  

• 检查UUID一致性：确认客户端中macUUID[0]对应的UUID与服务器端的macUUID完全匹配，注意大小写和格式细节。

内容的提问来源于stack exchange，提问作者CodyOnStack