Блог

Я успешно установил соединение с двумя BLE-серверами одновременно через pyDbus. Однако, когда я хочу прочитать / записать на первый сервер BLE, я не могу этого сделать. Я проверил эту проблему и обнаружил, что второе подключение к серверу BLE превосходит первое. Прилагается фрагмент кода:

 def device_Init(macAdd):

     adapter = bus.get('org.bluez','/org/bluez/hci0')
     dev = bus.get('org.bluez','/dev_(the macAdd)')
     mngr = bus.get('org.bluez','/')
    
     dev.Connect()
  

тогда, конечно, когда нужно отключить, тогда проблема dev.Disconnect()

Как разделить части подключения, чтения / записи и отключения между двумя подключенными устройствами? Я пытался создать другую функцию, но безуспешно.

Проблема с вашим кодом заключается в том, что dev свойство перезаписывается, когда вы создаете его во второй раз с другим адресом устройства. У вас могла бы быть вторая переменная для второго устройства.

Однако более питоническим способом сделать это, вероятно, было бы создать класс, который вы создаете дважды.

Вот пример того, как это могло бы выглядеть:

 from time import sleep
import pydbus

bus = pydbus.SystemBus()

mngr = bus.get('org.bluez', '/')

def get_characteristic_path(dev_path, uuid):
    mng_objs = mngr.GetManagedObjects()
    for path in mng_objs:
        chr_uuid = mng_objs[path].get('org.bluez.GattCharacteristic1', {}).get('UUID')
        if path.startswith(dev_path) and chr_uuid == uuid:
           return path

class MyRemoteDevice:
    CHAR_UUID = 'e95dda90-251d-470a-a062-fa1922dfa9a8'

    def __init__(self, mac_addr):
        device_path = f"/org/bluez/hci0/dev_{mac_addr.replace(':', '_')}"
        self.device = bus.get('org.bluez', device_path)

        # Placeholder for characteristic details
        self.characteristic = None

    def _get_gatt_details(self):
        char_path = get_characteristic_path(self.device._path,
                                            MyRemoteDevice.CHAR_UUID)
        self.characteristic = bus.get('org.bluez', char_path)


    def connect(self):
        self.device.Connect()
        # Check GATT services have been resolved before continuing
        while not self.device.ServicesResolved:
            sleep(0.25)
        self._get_gatt_details()

    def disconnect(self):
        self.device.Disconnect()

    def read(self):
        return self.characteristic.ReadValue({})

    def write(self, new_value):
        self.characteristic.WriteValue(new_value, {})


my_first_dev = MyRemoteDevice('11:22:33:44:55:66')
my_second_dev = MyRemoteDevice('22:33:44:55:66:77')

my_first_dev.connect()
my_second_dev.connect()

print(my_first_dev.read())
print(my_second_dev.read())

my_first_dev.disconnect()
my_second_dev.disconnect()