# -*- coding: utf-8 -*-
"""
Base class for backend clients.
Created on 2018-04-23 by hbldh <henrik.blidh@nedomkull.com>
"""
import abc
import asyncio
import uuid
from typing import Callable, Optional, Union
from warnings import warn
from bleak.backends.service import BleakGATTServiceCollection
from bleak.backends.characteristic import BleakGATTCharacteristic
from bleak.backends.device import BLEDevice
[docs]class BaseBleakClient(abc.ABC):
"""The Client Interface for Bleak Backend implementations to implement.
The documentation of this interface should thus be safe to use as a reference for your implementation.
Args:
address_or_ble_device (`BLEDevice` or str): The Bluetooth address of the BLE peripheral to connect to or the `BLEDevice` object representing it.
Keyword Args:
timeout (float): Timeout for required ``discover`` call. Defaults to 10.0.
disconnected_callback (callable): Callback that will be scheduled in the
event loop when the client is disconnected. The callable must take one
argument, which will be this client object.
"""
def __init__(self, address_or_ble_device: Union[BLEDevice, str], **kwargs):
if isinstance(address_or_ble_device, BLEDevice):
self.address = address_or_ble_device.address
else:
self.address = address_or_ble_device
self.services = BleakGATTServiceCollection()
self._services_resolved = False
self._notification_callbacks = {}
self._timeout = kwargs.get("timeout", 10.0)
self._disconnected_callback = kwargs.get("disconnected_callback")
def __str__(self):
return "{0}, {1}".format(self.__class__.__name__, self.address)
def __repr__(self):
return "<{0}, {1}, {2}>".format(
self.__class__.__name__,
self.address,
super(BaseBleakClient, self).__repr__(),
)
# Async Context managers
async def __aenter__(self):
await self.connect()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self.disconnect()
# Connectivity methods
[docs] def set_disconnected_callback(
self, callback: Optional[Callable[["BaseBleakClient"], None]], **kwargs
) -> None:
"""Set the disconnect callback.
The callback will only be called on unsolicited disconnect event.
Callbacks must accept one input which is the client object itself.
Set the callback to ``None`` to remove any existing callback.
.. code-block:: python
def callback(client):
print("Client with address {} got disconnected!".format(client.address))
client.set_disconnected_callback(callback)
client.connect()
Args:
callback: callback to be called on disconnection.
"""
self._disconnected_callback = callback
[docs] @abc.abstractmethod
async def connect(self, **kwargs) -> bool:
"""Connect to the specified GATT server.
Returns:
Boolean representing connection status.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def disconnect(self) -> bool:
"""Disconnect from the specified GATT server.
Returns:
Boolean representing connection status.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def pair(self, *args, **kwargs) -> bool:
"""Pair with the peripheral."""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def unpair(self) -> bool:
"""Unpair with the peripheral."""
raise NotImplementedError()
@property
@abc.abstractmethod
def is_connected(self) -> bool:
"""Check connection status between this client and the server.
Returns:
Boolean representing connection status.
"""
raise NotImplementedError()
class _DeprecatedIsConnectedReturn:
"""Wrapper for ``is_connected`` return value to provide deprecation warning."""
def __init__(self, value: bool):
self._value = value
def __bool__(self):
return self._value
def __call__(self) -> bool:
warn(
"is_connected has been changed to a property. Calling it as an async method will be removed in a future version",
FutureWarning,
stacklevel=2,
)
f = asyncio.Future()
f.set_result(self._value)
return f
def __repr__(self) -> str:
return repr(self._value)
# GATT services methods
[docs] @abc.abstractmethod
async def get_services(self, **kwargs) -> BleakGATTServiceCollection:
"""Get all services registered for this GATT server.
Returns:
A :py:class:`bleak.backends.service.BleakGATTServiceCollection` with this device's services tree.
"""
raise NotImplementedError()
# I/O methods
[docs] @abc.abstractmethod
async def read_gatt_char(
self,
char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
**kwargs
) -> bytearray:
"""Perform read operation on the specified GATT characteristic.
Args:
char_specifier (BleakGATTCharacteristic, int, str or UUID): The characteristic to read from,
specified by either integer handle, UUID or directly by the
BleakGATTCharacteristic object representing it.
Returns:
(bytearray) The read data.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def read_gatt_descriptor(self, handle: int, **kwargs) -> bytearray:
"""Perform read operation on the specified GATT descriptor.
Args:
handle (int): The handle of the descriptor to read from.
Returns:
(bytearray) The read data.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def write_gatt_char(
self,
char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
data: Union[bytes, bytearray, memoryview],
response: bool = False,
) -> None:
"""Perform a write operation on the specified GATT characteristic.
Args:
char_specifier (BleakGATTCharacteristic, int, str or UUID): The characteristic to write
to, specified by either integer handle, UUID or directly by the
BleakGATTCharacteristic object representing it.
data (bytes or bytearray): The data to send.
response (bool): If write-with-response operation should be done. Defaults to `False`.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def write_gatt_descriptor(
self, handle: int, data: Union[bytes, bytearray, memoryview]
) -> None:
"""Perform a write operation on the specified GATT descriptor.
Args:
handle (int): The handle of the descriptor to read from.
data (bytes or bytearray): The data to send.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def start_notify(
self,
char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID],
callback: Callable[[int, bytearray], None],
**kwargs
) -> None:
"""Activate notifications/indications on a characteristic.
Callbacks must accept two inputs. The first will be a integer handle of the characteristic generating the
data and the second will be a ``bytearray``.
.. code-block:: python
def callback(sender: int, data: bytearray):
print(f"{sender}: {data}")
client.start_notify(char_uuid, callback)
Args:
char_specifier (BleakGATTCharacteristic, int, str or UUID): The characteristic to activate
notifications/indications on a characteristic, specified by either integer handle,
UUID or directly by the BleakGATTCharacteristic object representing it.
callback (function): The function to be called on notification.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def stop_notify(
self, char_specifier: Union[BleakGATTCharacteristic, int, str, uuid.UUID]
) -> None:
"""Deactivate notification/indication on a specified characteristic.
Args:
char_specifier (BleakGATTCharacteristic, int, str or UUID): The characteristic to deactivate
notification/indication on, specified by either integer handle, UUID or
directly by the BleakGATTCharacteristic object representing it.
"""
raise NotImplementedError()