0

我正在尝试制作一个非常简单的程序来通过 BLE 连接到NXP QN9080，接收一些数据，然后将其打印出来。但是，在连接过程中，我不断收到以下错误：

AttributeError：“NoneType”对象没有属性“ConnectionStatusChanged”

我能够使用以下代码块发现设备并查看其 MAC 地址（作为示例从黯淡的自述文件中提供）：

import asyncio
from bleak import discover

async def run():
    devices = await discover()
    for d in devices:
        print(d)

loop = asyncio.get_event_loop()
loop.run_until_complete(run())

但是当我尝试运行以下代码（也来自惨淡的示例）时，它会出错：

import asyncio
import platform

from bleak import BleakClient


async def print_services(mac_addr: str, loop: asyncio.AbstractEventLoop):
    async with BleakClient(mac_addr, loop=loop) as client:
        svcs = await client.get_services()
        print("Services:", svcs)


mac_addr = DEVICE_MAC_ADDR
loop = asyncio.get_event_loop()
loop.run_until_complete(print_services(mac_addr, loop))

BleakClient 在 init 上使用的连接函数如下：

    async def connect(self, **kwargs) -> bool:
        """Connect to the specified GATT server.

        Keyword Args:
            timeout (float): Timeout for required ``discover`` call. Defaults to 2.0.

        Returns:
            Boolean representing connection status.

        """
        # Try to find the desired device.
        devices = await discover(timeout=kwargs.get("timeout", 10.0), loop=self.loop)

        sought_device = list(
            filter(lambda x: x.address.upper() == self.address.upper(), devices)
        )

        if len(sought_device):
            self._device_info = sought_device[0].details
        else:
            raise BleakError(
                "Device with address {0} was " "not found.".format(self.address)
            )
        logger.debug("Connecting to BLE device @ {0}".format(self.address))

        args = [UInt64(self._device_info.BluetoothAddress)]
        if self._address_type is not None:
            args.append(
                BluetoothAddressType.Public
                if self._address_type == "public"
                else BluetoothAddressType.Random
            )
        self._requester = await wrap_IAsyncOperation(
            IAsyncOperation[BluetoothLEDevice](
                BluetoothLEDevice.FromBluetoothAddressAsync(*args)
            ),
            return_type=BluetoothLEDevice,
            loop=self.loop,
        )

        def _ConnectionStatusChanged_Handler(sender, args):
            logger.debug("_ConnectionStatusChanged_Handler: " + args.ToString())

        self._requester.ConnectionStatusChanged += _ConnectionStatusChanged_Handler

        # Obtain services, which also leads to connection being established.
        services = await self.get_services()
        connected = False
        if self._services_resolved:
            # If services has been resolved, then we assume that we are connected. This is due to
            # some issues with getting `is_connected` to give correct response here.
            connected = True
        else:
            for _ in range(5):
                await asyncio.sleep(0.2, loop=self.loop)
                connected = await self.is_connected()
                if connected:
                    break

        if connected:
            logger.debug("Connection successful.")
        else:
            raise BleakError(
                "Connection to {0} was not successful!".format(self.address)
            )

        return connected

所以这意味着 await wrap_IAsyncOperation(*) 出于某种原因返回 None ，这会导致错误。直到那条线的一切似乎都运行良好。知道为什么会这样吗？

4

1 回答 1

0 
import asyncio
from bleak import discover

devices = []
async def scan():
    dev = await discover()
    for i in range(0,len(dev)):
        print("["+str(i)+"]"+str(dev[i]))
        devices.append(dev[i])

from bleak import BleakClient

async def connect(address, loop):
    async with BleakClient(address, loop=loop) as client:
        services = await client.get_services()
        for ser in services:
            print(ser.uuid)

loop = asyncio.get_event_loop()
loop.run_until_complete(scan())
index = input('please select device from 0 to '+str(len(devices))+":")
index = int(index)
loop.run_until_complete(connect(devices[index].address, loop))
于 2019-09-03T09:25:18.177 回答