diff --git a/bless/backends/bluezdbus/dbus/application.py b/bless/backends/bluezdbus/dbus/application.py index cad34c1..bf41608 100644 --- a/bless/backends/bluezdbus/dbus/application.py +++ b/bless/backends/bluezdbus/dbus/application.py @@ -57,8 +57,12 @@ def __init__(self, name: str, destination: str, bus: MessageBus): self.Write: Optional[ Callable[[BlueZGattCharacteristic, bytes, Dict[str, Any]], None] ] = None - self.StartNotify: Optional[Callable[[None], None]] = None - self.StopNotify: Optional[Callable[[None], None]] = None + self.StartNotify: Optional[ + Callable[[BlueZGattCharacteristic, Dict[str, Any]], None] + ] = None + self.StopNotify: Optional[ + Callable[[BlueZGattCharacteristic, Dict[str, Any]], None] + ] = None self.subscribed_characteristics: List[str] = [] diff --git a/bless/backends/bluezdbus/dbus/characteristic.py b/bless/backends/bluezdbus/dbus/characteristic.py index 85bf6a4..ef59b47 100644 --- a/bless/backends/bluezdbus/dbus/characteristic.py +++ b/bless/backends/bluezdbus/dbus/characteristic.py @@ -12,7 +12,7 @@ if TYPE_CHECKING: from bless.backends.bluezdbus.dbus.service import ( # type: ignore # noqa: F401 - BlueZGattService + BlueZGattService, ) @@ -68,10 +68,7 @@ def __init__( self._service: "BlueZGattService" = service # noqa: F821 self._value: bytes = b"" - self._notifying: bool = ( - "notify" in self._flags - or "indicate" in self._flags - ) + self._notifying: bool = "notify" in self._flags or "indicate" in self._flags self.descriptors: List["BlueZGattDescriptor"] = [] # noqa: F821 super(BlueZGattCharacteristic, self).__init__(self.interface_name) @@ -91,9 +88,7 @@ def Value(self) -> "ay": # type: ignore # noqa: F821 N802 @Value.setter # type: ignore def Value(self, value: "ay"): # type: ignore # noqa: F821 N802 self._value = value - self.emit_properties_changed( - changed_properties={"Value": self._value} - ) + self.emit_properties_changed(changed_properties={"Value": self._value}) @dbus_property(access=PropertyAccess.READ) def Notifying(self) -> "b": # type: ignore # noqa: F821 N802 @@ -150,7 +145,7 @@ def StartNotify(self): # noqa: N802 f = self._service.app.StartNotify if f is None: raise NotImplementedError() - f(None) + f(self, {}) self._service.app.subscribed_characteristics.append(self._uuid) @method() @@ -161,7 +156,7 @@ def StopNotify(self): # noqa: N802 f = self._service.app.StopNotify if f is None: raise NotImplementedError() - f(None) + f(self, {}) self._service.app.subscribed_characteristics.remove(self._uuid) async def add_descriptor( @@ -180,9 +175,7 @@ async def add_descriptor( The descriptor's value """ index: int = len(self.descriptors) + 1 - descriptor: BlueZGattDescriptor = BlueZGattDescriptor( - uuid, flags, index, self - ) + descriptor: BlueZGattDescriptor = BlueZGattDescriptor(uuid, flags, index, self) descriptor._value = value # type: ignore self.descriptors.append(descriptor) await self._service.app._register_object(descriptor) @@ -198,6 +191,4 @@ async def get_obj(self) -> Dict: Dict The dictionary that describes the characteristic """ - return { - "UUID": Variant('s', self._uuid) - } + return {"UUID": Variant("s", self._uuid)} diff --git a/bless/backends/bluezdbus/server.py b/bless/backends/bluezdbus/server.py index 02498e1..d86aab6 100644 --- a/bless/backends/bluezdbus/server.py +++ b/bless/backends/bluezdbus/server.py @@ -68,10 +68,8 @@ async def setup(self: "BlessServerBlueZDBus"): self.app.Read = self.read self.app.Write = self.write - - # We don't need to define these - self.app.StartNotify = lambda x: None - self.app.StopNotify = lambda x: None + self.app.StartNotify = self.subscribe + self.app.StopNotify = self.unsubscribe potential_adapter: Optional[ProxyObject] = await get_adapter( self.bus, self._adapter @@ -248,8 +246,8 @@ async def add_new_descriptor( service.get_characteristic(std_char_uuid), ) std_desc_uuid = normalize_uuid_str(desc_uuid) - descriptor: BlessGATTDescriptorBlueZDBus = ( - BlessGATTDescriptorBlueZDBus(std_desc_uuid, properties, permissions, value) + descriptor: BlessGATTDescriptorBlueZDBus = BlessGATTDescriptorBlueZDBus( + std_desc_uuid, properties, permissions, value ) await descriptor.init(characteristic) @@ -333,3 +331,31 @@ def write( The value being requested to set """ return self.write_request(char.UUID, bytearray(value), options) + + def subscribe(self, char: BlueZGattCharacteristic, options: Dict[str, Any]): + """ + Subscribe request. + This function re-routes the subscribe request sent from the + BlueZGattApplication to the server function for re-route to the user + defined handler + + Parameters + ---------- + char : BlueZGattCharacteristic + The characteristic object involved in the request + """ + return self.subscribe_request(char.UUID, options) + + def unsubscribe(self, char: BlueZGattCharacteristic, options: Dict[str, Any]): + """ + Unsubscribe request. + This function re-routes the unsubscribe request sent from the + BlueZGattApplication to the server function for re-route to the user + defined handler + + Parameters + ---------- + char : BlueZGattCharacteristic + The characteristic object involved in the request + """ + return self.unsubscribe_request(char.UUID, options) diff --git a/bless/backends/corebluetooth/peripheral_manager_delegate.py b/bless/backends/corebluetooth/peripheral_manager_delegate.py index 0d4b342..adb871e 100644 --- a/bless/backends/corebluetooth/peripheral_manager_delegate.py +++ b/bless/backends/corebluetooth/peripheral_manager_delegate.py @@ -1,7 +1,10 @@ import os from typing import TYPE_CHECKING, Any, Callable, Dict, Optional +from warnings import warn + if TYPE_CHECKING: + class PeripheralManagerDelegate: # type: ignore # noqa: F811 event_loop: Optional["asyncio.AbstractEventLoop"] peripheral_manager: Any @@ -14,39 +17,43 @@ class PeripheralManagerDelegate: # type: ignore # noqa: F811 pyobjc_classMethods: Any @classmethod - def alloc(cls) -> "PeripheralManagerDelegate": - ... + def alloc(cls) -> "PeripheralManagerDelegate": ... - def init(self: "PeripheralManagerDelegate"): - ... + def init( + self: "PeripheralManagerDelegate", + server: Optional[Any] = None, + callbacks: Optional[Dict[str, Callable[..., Any]]] = None, + ) -> "PeripheralManagerDelegate": ... async def start_advertising( self, advertisement_data: Dict[str, Any], timeout: float = 2.0 - ) -> None: - ... + ) -> None: ... - async def stop_advertising(self) -> None: - ... + async def stop_advertising(self) -> None: ... - def is_connected(self) -> bool: - ... + def is_connected(self) -> bool: ... - def is_advertising(self) -> bool: - ... + def is_advertising(self) -> bool: ... - async def add_service(self, service: Any) -> None: - ... + async def add_service(self, service: Any) -> None: ... + + def compliant(self) -> bool: ... - def compliant(self) -> bool: - ... if os.environ.get("BLESS_DOCS_BUILD"): + class PeripheralManagerDelegate: # type: ignore # noqa: F811 """ Stub for documentation builds where CoreBluetooth is unavailable. """ - def init(self: "PeripheralManagerDelegate"): + def init( + self: "PeripheralManagerDelegate", + server: Optional[Any] = None, + callbacks: Optional[Dict[str, Callable[..., Any]]] = None, + ): + self.server = server + self._callbacks = callbacks or {} return self async def start_advertising(self, advertisement_data, timeout: float = 2.0): @@ -60,6 +67,7 @@ def is_connected(self) -> bool: def is_advertising(self) -> bool: return False + else: import objc # type: ignore import asyncio @@ -92,22 +100,25 @@ def is_advertising(self) -> bool: CBPeripheralManagerDelegate = objc.protocolNamed("CBPeripheralManagerDelegate") class PeripheralManagerDelegate( # type: ignore # noqa: F811 - NSObject, - protocols=[CBPeripheralManagerDelegate] - ): - def init(self: "PeripheralManagerDelegate"): + NSObject, protocols=[CBPeripheralManagerDelegate] + ): + def init( + self: "PeripheralManagerDelegate", + server: Optional[Any] = None, + callbacks: Optional[Dict[str, Callable]] = None, + ): self = objc.super(PeripheralManagerDelegate, self).init() self.event_loop: Optional[asyncio.AbstractEventLoop] = None - self.server: Optional[Any] = None + self.server: Optional[Any] = server - self.peripheral_manager: CBPeripheralManager = ( - CBPeripheralManager.alloc().initWithDelegate_queue_( - self, dispatch_queue_create(b"BLE", DISPATCH_QUEUE_SERIAL) - ) + self.peripheral_manager: ( + CBPeripheralManager + ) = CBPeripheralManager.alloc().initWithDelegate_queue_( + self, dispatch_queue_create(b"BLE", DISPATCH_QUEUE_SERIAL) ) - self._callbacks: Dict[str, Callable] = {} + self._callbacks: Dict[str, Callable] = callbacks or {} # Events self._powered_on_event: threading.Event = threading.Event() @@ -365,6 +376,10 @@ def peripheralManager_central_didSubscribeToCharacteristic_( # noqa: N802 ) else: self._central_subscriptions[central_uuid] = [char_uuid] + self._callbacks.get("subscribe", lambda x: None)( + characteristic.UUID().UUIDString() + ) + self.get_callback("subscribe")(characteristic.UUID().UUIDString()) def peripheralManager_central_didUnsubscribeFromCharacteristic_( # noqa: N802 E501 self, @@ -383,6 +398,8 @@ def peripheralManager_central_didUnsubscribeFromCharacteristic_( # noqa: N802 E if len(self._central_subscriptions[central_uuid]) < 1: del self._central_subscriptions[central_uuid] + self.get_callback("unsubscribe")(characteristic.UUID().UUIDString()) + def peripheralManagerIsReadyToUpdateSubscribers_( # noqa: N802 self, peripheral_manager: CBPeripheralManager ): @@ -400,7 +417,7 @@ def peripheralManager_didReceiveReadRequest_( # noqa: N802 ) ) request.setValue_( - self.read_request_func(request.characteristic().UUID().UUIDString()) + self.get_callback("read")(request.characteristic().UUID().UUIDString()) ) peripheral_manager.respondToRequest_withResult_(request, CBATTErrorSuccess) @@ -420,8 +437,17 @@ def peripheralManager_didReceiveWriteRequests_( # noqa: N802 value, ) ) - self.write_request_func(char.UUID().UUIDString(), value) + self.get_callback("write")(char.UUID().UUIDString(), value) peripheral_manager.respondToRequest_withResult_( requests[0], CBATTErrorSuccess ) + + def get_callback(self, callback_name: str) -> Callable: + if callback_name not in self._callbacks: + warn( + "Callback {} does not exist".format(callback_name), + UserWarning, + ) + return lambda x: None + return self._callbacks[callback_name] diff --git a/bless/backends/corebluetooth/server.py b/bless/backends/corebluetooth/server.py index 6d87dde..1c2aa0d 100644 --- a/bless/backends/corebluetooth/server.py +++ b/bless/backends/corebluetooth/server.py @@ -66,12 +66,17 @@ def __init__(self, name: str, loop: Optional[AbstractEventLoop] = None, **kwargs self.name: str = name - self.peripheral_manager_delegate: PeripheralManagerDelegate = ( - PeripheralManagerDelegate.alloc().init() + self.peripheral_manager_delegate: ( + PeripheralManagerDelegate + ) = PeripheralManagerDelegate.alloc().init( + self, + { + "read": self.read_request, + "write": self.write_request, + "subscribe": self.subscribe_request, + "unsubscribe": self.unsubscribe_request, + }, ) - self.peripheral_manager_delegate.server = self - self.peripheral_manager_delegate.read_request_func = self.read_request - self.peripheral_manager_delegate.write_request_func = self.write_request async def start( self, diff --git a/bless/backends/server.py b/bless/backends/server.py index 5559c40..ba2e9b3 100644 --- a/bless/backends/server.py +++ b/bless/backends/server.py @@ -290,9 +290,6 @@ def read_request(self, uuid: str, options: Optional[Dict] = None) -> bytearray: characteristics owned by our server. This function then hands off execution to the user-defiend callback functions - Note: read_request_func must be defined on the class that inherits this - base class - Parameters ---------- uuid : str @@ -314,14 +311,15 @@ def read_request(self, uuid: str, options: Optional[Dict] = None) -> bytearray: if not characteristic: raise BlessError("Invalid characteristic: {}".format(uuid)) - return self.read_request_func(characteristic) + return self.on_read(characteristic) - def write_request(self, uuid: str, value: Any, options: Optional[Dict] = None): + def write_request( + self, uuid: str, value: Any, options: Optional[Dict] = None + ) -> None: """ Obtain the characteristic to write and pass on to the user-defined - write_request_func + on_write - Note: write_request_func must be defined on the child class """ if options is not None: self._update_mtu_from_options(options) @@ -329,51 +327,33 @@ def write_request(self, uuid: str, value: Any, options: Optional[Dict] = None): uuid ) - self.write_request_func(characteristic, value) + self.on_write(characteristic, value) - @property - def read_request_func(self) -> Callable[[Any], Any]: - """ - Return an instance of the function to handle incoming read requests - - Note - ---- - This will be deprecated in version 0.4. Prefer using `on_read`. + def subscribe_request(self, uuid: str, options: Optional[Dict] = None) -> None: """ - return self.on_read - - @read_request_func.setter - def read_request_func(self, func: Callable): + Obtain the characteristic to subscribe to and pass on to the + user-defined on_subscribe """ - Set the function to handle incoming read requests - - Note - ---- - This will be deprecated in version 0.4. Prefer using `on_read`. - """ - self.on_read = func + if options is not None: + self._update_mtu_from_options(options) + characteristic: Optional[BlessGATTCharacteristic] = self.get_characteristic( + uuid + ) - @property - def write_request_func(self) -> Callable: - """ - Return an instance of the function to handle incoming write requests + self.on_subscribe(characteristic) - Note - ---- - This will be deprecated in version 0.4. Prefer using `on_write`. + def unsubscribe_request(self, uuid: str, options: Optional[Dict] = None) -> None: """ - return self.on_write - - @write_request_func.setter - def write_request_func(self, func: Callable): + Obtain the characteristic to unsubscribe from and pass on to the + user-defined on_unsubscribe """ - Set the function to handle incoming write requests + if options is not None: + self._update_mtu_from_options(options) + characteristic: Optional[BlessGATTCharacteristic] = self.get_characteristic( + uuid + ) - Note - ---- - This will be deprecated in version 0.4. Prefer using `on_write`. - """ - self.on_write = func + self.on_unsubscribe(characteristic) @property def on_read(self) -> Callable[[Any], Any]: @@ -411,6 +391,82 @@ def on_write(self, func: Callable): """ self._callbacks["write"] = func + @property + def on_subscribe(self) -> Callable: + """ + Alias for `subscribe_request_func`. + """ + func: Optional[Callable[[Any], Any]] = self._callbacks.get("subscribe") + if func is not None: + return func + else: + raise BlessError("Server: Subscribe Callback is undefined") + + @on_subscribe.setter + def on_subscribe(self, func: Callable): + """ """ + self._callbacks["subscribe"] = func + + @property + def on_unsubscribe(self) -> Callable: + """ + Alias for `unsubscribe_request_func`. + """ + func: Optional[Callable[[Any], Any]] = self._callbacks.get("unsubscribe") + if func is not None: + return func + else: + raise BlessError("Server: Unsubscribe Callback is undefined") + + @on_unsubscribe.setter + def on_unsubscribe(self, func: Callable): + """ """ + self._callbacks["unsubscribe"] = func + + @property + def read_request_func(self) -> Callable[[Any], Any]: + """ + Return an instance of the function to handle incoming read requests + + Note + ---- + This will be deprecated in version 0.4. Prefer using `on_read`. + """ + return self.on_read + + @read_request_func.setter + def read_request_func(self, func: Callable): + """ + Set the function to handle incoming read requests + + Note + ---- + This will be deprecated in version 0.4. Prefer using `on_read`. + """ + self.on_read = func + + @property + def write_request_func(self) -> Callable: + """ + Return an instance of the function to handle incoming write requests + + Note + ---- + This will be deprecated in version 0.4. Prefer using `on_write`. + """ + return self.on_write + + @write_request_func.setter + def write_request_func(self, func: Callable): + """ + Set the function to handle incoming write requests + + Note + ---- + This will be deprecated in version 0.4. Prefer using `on_write`. + """ + self.on_write = func + @property def mtu(self) -> Optional[int]: """ diff --git a/bless/backends/winrt/server.py b/bless/backends/winrt/server.py index 578dd31..c2a502c 100644 --- a/bless/backends/winrt/server.py +++ b/bless/backends/winrt/server.py @@ -443,7 +443,18 @@ def subscribe_characteristic(self, sender: GattLocalCharacteristic, args: Any): Additional arguments to use for the subscription """ clients = sender.subscribed_clients + + # Handle Callbacks + if clients is not None: + if len(list(clients)) > len(self._subscribed_clients): + self.subscribe_request(str(sender.uuid)) + elif len(list(clients)) < len(self._subscribed_clients): + self.unsubscribe_request(str(sender.uuid)) + + # Update Subscribed Clients self._subscribed_clients = list(clients) if clients is not None else [] + + # Process MTU if self._subscribed_clients: mtu_values = [ int(client.max_pdu_size) diff --git a/examples/gattserver.py b/examples/gattserver.py index ff4ddae..4cd73f3 100644 --- a/examples/gattserver.py +++ b/examples/gattserver.py @@ -2,6 +2,7 @@ Example for a BLE 4.0 Server using a GATT dictionary of services and characteristics """ + import sys import logging import asyncio @@ -26,12 +27,12 @@ trigger = asyncio.Event() -def read_request(characteristic: BlessGATTCharacteristic, **kwargs) -> bytearray: +def on_read(characteristic: BlessGATTCharacteristic, **kwargs) -> bytearray: logger.debug(f"Reading {characteristic.value}") return characteristic.value -def write_request(characteristic: BlessGATTCharacteristic, value: Any, **kwargs): +def on_write(characteristic: BlessGATTCharacteristic, value: Any, **kwargs): characteristic.value = value logger.debug(f"Char value set to {characteristic.value}") if characteristic.value == b"\x0f": @@ -39,6 +40,14 @@ def write_request(characteristic: BlessGATTCharacteristic, value: Any, **kwargs) trigger.set() +def on_subscribe(characteristic: BlessGATTCharacteristic, **kwargs): + logger.debug(f"Subscribed to {characteristic.uuid}") + + +def on_unsubscribe(characteristic: BlessGATTCharacteristic, **kwargs): + logger.debug(f"Unsubscribed from {characteristic.uuid}") + + async def run(loop): trigger.clear() @@ -68,8 +77,10 @@ async def run(loop): } my_service_name = "Test Service" server = BlessServer(name=my_service_name, loop=loop) - server.read_request_func = read_request - server.write_request_func = write_request + server.on_read = on_read + server.on_write = on_write + server.on_subscribe = on_subscribe + server.on_unsubscribe = on_unsubscribe await server.add_gatt(gatt) await server.start() diff --git a/test/backends/test_server.py b/test/backends/test_server.py index a95fa8c..f95ea81 100644 --- a/test/backends/test_server.py +++ b/test/backends/test_server.py @@ -9,27 +9,25 @@ from typing import Optional, List -from bless.backends.characteristic import ( # type: ignore - BlessGATTCharacteristic - ) +from bless.backends.characteristic import BlessGATTCharacteristic # type: ignore # Eventually should be removed when MacOS, Windows, and Linux are added -if sys.platform not in ['darwin', 'linux', 'win32']: +if sys.platform not in ["darwin", "linux", "win32"]: pytest.skip( - "Currently, testing only works on macOS, linux, and windows", - allow_module_level=True - ) + "Currently, testing only works on macOS, linux, and windows", + allow_module_level=True, + ) from bless import BlessServer # type: ignore # noqa: E402 from bless.backends.attribute import ( # noqa: E402 - GATTAttributePermissions, - ) + GATTAttributePermissions, +) from bless.backends.characteristic import ( # noqa: E402 - GATTCharacteristicProperties, - ) + GATTCharacteristicProperties, +) hardware_only = pytest.mark.skipif("os.environ.get('TEST_HARDWARE') is None") -use_encrypted = os.environ.get('TEST_ENCRYPTED') is not None +use_encrypted = os.environ.get("TEST_ENCRYPTED") is not None @hardware_only @@ -43,25 +41,28 @@ class TestBlessServer: def gen_hex_pairs(self) -> str: hex_words: List[str] = [ - 'DEAD', 'FACE', 'BABE', - 'CAFE', 'FADE', 'BAD', - 'DAD', 'ACE', 'BED' - ] - rng: np.random._generator.Generator = ( # type: ignore - np.random.default_rng() - ) - return ''.join(rng.choice(hex_words, 2, replace=False)) + "DEAD", + "FACE", + "BABE", + "CAFE", + "FADE", + "BAD", + "DAD", + "ACE", + "BED", + ] + rng: np.random._generator.Generator = np.random.default_rng() # type: ignore + return "".join(rng.choice(hex_words, 2, replace=False)) def hex_to_byte(self, hexstr: str) -> bytearray: return bytearray( - int(f"0x{hexstr}", 16).to_bytes( - length=int(np.ceil(len(hexstr)/2)), - byteorder="big" - ) - ) + int(f"0x{hexstr}", 16).to_bytes( + length=int(np.ceil(len(hexstr) / 2)), byteorder="big" + ) + ) def byte_to_hex(self, b: bytearray) -> str: - return ''.join([hex(x)[2:] for x in b]).upper() + return "".join([hex(x)[2:] for x in b]).upper() @pytest.mark.asyncio async def test_server(self): @@ -78,42 +79,45 @@ async def test_server(self): # setup a characteristic for the service char_uuid: str = str(uuid.uuid4()) char_flags: GATTCharacteristicProperties = ( - GATTCharacteristicProperties.read | - GATTCharacteristicProperties.write | - GATTCharacteristicProperties.notify - ) + GATTCharacteristicProperties.read + | GATTCharacteristicProperties.write + | GATTCharacteristicProperties.notify + ) value: Optional[bytearray] = None permissions: GATTAttributePermissions = ( - GATTAttributePermissions.readable | - GATTAttributePermissions.writeable - ) + GATTAttributePermissions.readable | GATTAttributePermissions.writeable + ) if use_encrypted: print("\nEncryption has been enabled, ensure that you are bonded") permissions = ( - GATTAttributePermissions.read_encryption_required | - GATTAttributePermissions.write_encryption_required - ) + GATTAttributePermissions.read_encryption_required + | GATTAttributePermissions.write_encryption_required + ) await server.add_new_characteristic( - service_uuid, - char_uuid, - char_flags, - value, - permissions - ) + service_uuid, char_uuid, char_flags, value, permissions + ) assert server.services[service_uuid].get_characteristic(char_uuid) - # Set up read and write callbacks + # Set up read, write, and subscribe callbacks def read(characteristic: BlessGATTCharacteristic) -> bytearray: return characteristic.value def write(characteristic: BlessGATTCharacteristic, value: bytearray): characteristic.value = value # type: ignore - server.read_request_func = read - server.write_request_func = write + def subscribe(characteristic: BlessGATTCharacteristic) -> None: + print("Subscribed") + + def unsubscribe(characteristic: BlessGATTCharacteristic) -> None: + print("Unsubscribed") + + server.on_read = read + server.on_write = write + server.on_subscribe = subscribe + server.on_unsubscribe = unsubscribe # Start advertising assert await server.is_advertising() is False @@ -126,9 +130,9 @@ def write(characteristic: BlessGATTCharacteristic, value: bytearray): assert await server.is_connected() is False print( - "\nPlease connect to the computer and " + - f"subscribe to characteristic {char_uuid}" - ) + "\nPlease connect to the computer and " + + f"subscribe to characteristic {char_uuid}" + ) await aioconsole.ainput("Press enter when ready...") assert await server.is_connected() is True @@ -137,9 +141,9 @@ def write(characteristic: BlessGATTCharacteristic, value: bytearray): hex_val: str = self.gen_hex_pairs() server.get_characteristic(char_uuid).value = self.hex_to_byte(hex_val) print( - "Trigger a read command and " + - "enter the capital letters you retrieve below" - ) + "Trigger a read command and " + + "enter the capital letters you retrieve below" + ) entered_value = await aioconsole.ainput("Value: ") assert entered_value == hex_val @@ -147,9 +151,7 @@ def write(characteristic: BlessGATTCharacteristic, value: bytearray): hex_val = self.gen_hex_pairs() print(f"Set the characteristic to this value: {hex_val}") await aioconsole.ainput("Press enter when ready...") - entered_value = self.byte_to_hex( - server.get_characteristic(char_uuid).value - ) + entered_value = self.byte_to_hex(server.get_characteristic(char_uuid).value) assert entered_value == hex_val # Notify Test