# -*- coding: utf-8 -*-
"""
Base class for backend clients.
Created on 2018-04-23 by hbldh <henrik.blidh@nedomkull.com>
"""
import abc
import asyncio
import os
import platform
import sys
import uuid
from typing import Callable, Optional, Type, Union
from warnings import warn
if sys.version_info < (3, 12):
from typing_extensions import Buffer
else:
from collections.abc import Buffer
from ..exc import BleakError
from .service import BleakGATTServiceCollection
from .characteristic import BleakGATTCharacteristic
from .device import BLEDevice
NotifyCallback = Callable[[bytearray], None]
[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: Optional[BleakGATTServiceCollection] = None
self._timeout = kwargs.get("timeout", 10.0)
self._disconnected_callback: Optional[Callable[[], None]] = kwargs.get(
"disconnected_callback"
)
@property
@abc.abstractmethod
def mtu_size(self) -> int:
"""Gets the negotiated MTU."""
raise NotImplementedError
# Connectivity methods
[docs] def set_disconnected_callback(
self, callback: Optional[Callable[[], None]], **kwargs
) -> None:
"""Set the disconnect callback.
The callback will only be called on unsolicited disconnect event.
Set the callback to ``None`` to remove any existing callback.
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,
characteristic: BleakGATTCharacteristic,
data: Buffer,
response: bool,
) -> None:
"""
Perform a write operation on the specified GATT characteristic.
Args:
characteristic: The characteristic to write to.
data: The data to send.
response: If write-with-response operation should be done.
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def write_gatt_descriptor(self, handle: int, data: Buffer) -> None:
"""Perform a write operation on the specified GATT descriptor.
Args:
handle: The handle of the descriptor to read from.
data: The data to send (any bytes-like object).
"""
raise NotImplementedError()
[docs] @abc.abstractmethod
async def start_notify(
self,
characteristic: BleakGATTCharacteristic,
callback: NotifyCallback,
**kwargs,
) -> None:
"""
Activate notifications/indications on a characteristic.
Implementers should call the OS function to enable notifications or
indications on the characteristic.
To keep things the same cross-platform, notifications should be preferred
over indications if possible when a characteristic supports both.
"""
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()