diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 80c047a1..1e402d3f 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -13,6 +13,7 @@ and this project adheres to `Semantic Versioning int: + """Converts the Bluetooth device address string to its representing integer + + Args: + address (str): Bluetooth device address to convert + + Returns: + int: integer representation of the given Bluetooth device address + """ + _address_separators = [":", "-"] + for char in _address_separators: + address = address.replace(char, "") + + return int(address, base=16) + + def _ensure_success(result: Any, attr: Optional[str], fail_msg: str) -> Any: """ Ensures that *status* is ``GattCommunicationStatus.SUCCESS``, otherwise @@ -182,6 +198,18 @@ def __str__(self): # Connectivity methods + def _create_requester(self, bluetooth_address: int): + args = [ + bluetooth_address, + ] + if self._address_type is not None: + args.append( + BluetoothAddressType.PUBLIC + if self._address_type == "public" + else BluetoothAddressType.RANDOM + ) + return BluetoothLEDevice.from_bluetooth_address_async(*args) + async def connect(self, **kwargs) -> bool: """Connect to the specified GATT server. @@ -206,16 +234,7 @@ async def connect(self, **kwargs) -> bool: logger.debug("Connecting to BLE device @ %s", self.address) - args = [ - self._device_info, - ] - if self._address_type is not None: - args.append( - BluetoothAddressType.PUBLIC - if self._address_type == "public" - else BluetoothAddressType.RANDOM - ) - self._requester = await BluetoothLEDevice.from_bluetooth_address_async(*args) + self._requester = await self._create_requester(self._device_info) if self._requester is None: # https://github.com/microsoft/Windows-universal-samples/issues/1089#issuecomment-487586755 @@ -452,14 +471,17 @@ async def unpair(self) -> bool: Boolean on whether the unparing was successful. """ - - # New local device information object created since the object from the requester isn't updated - device_information = await DeviceInformation.create_from_id_async( - self._requester.device_information.id + device = await self._create_requester( + self._device_info + if self._device_info is not None + else _address_to_int(self.address) ) - if device_information.pairing.is_paired: - unpairing_result = await device_information.pairing.unpair_async() + if device is None: + raise BleakError(f"Device with address {self.address} was not found.") + + try: + unpairing_result = await device.device_information.pairing.unpair_async() if unpairing_result.status not in ( DeviceUnpairingResultStatus.UNPAIRED, DeviceUnpairingResultStatus.ALREADY_UNPAIRED, @@ -467,12 +489,11 @@ async def unpair(self) -> bool: raise BleakError( f"Could not unpair with device: {unpairing_result.status}" ) + logger.info("Unpaired with device.") + finally: + device.close() - else: - logger.info("Unpaired with device.") - return True - - return not device_information.pairing.is_paired + return True # GATT services methods