Skip to content

Commit

Permalink
winrt: allow unpair without being connected (#1012)
Browse files Browse the repository at this point in the history
  • Loading branch information
jochenjagers authored Sep 19, 2022
1 parent d343cdd commit 66f71c0
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 21 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.rst
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0
Changed
-------
* Relax ``async-timeout`` version to support different installations. Merged #1009.
* ``unpair`` function of ``BleakClient`` in WinRT backend can be called without being connected to remove stored device information

`0.17.0`_ (2022-09-12)
======================
Expand Down
63 changes: 42 additions & 21 deletions bleak/backends/winrt/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,22 @@
# protocol_error: typing.Optional[int]


def _address_to_int(address: str) -> 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
Expand Down Expand Up @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -452,27 +471,29 @@ 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,
):
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

Expand Down

0 comments on commit 66f71c0

Please sign in to comment.