Coverage for src/bluetooth_sig/device/connection.py: 100%

1"""Connection manager protocol for BLE transport adapters. 

2 

3Defines an async protocol that adapter implementations (Bleak, 

4SimplePyBLE, etc.) should follow so the `Device` class can operate 

5independently of the underlying BLE library. 

6 

7Adapters should provide async implementations of the methods below. For 

8sync-only libraries an adapter can run sync calls in a thread and expose 

9an async interface. 

10""" 

11 

12from __future__ import annotations 

13 

14from typing import Any, Callable, Protocol 

15 

16from bluetooth_sig.types.uuid import BluetoothUUID 

17 

18 

19class ConnectionManagerProtocol(Protocol): 

20 """Protocol describing the transport operations Device expects. 

21 

22 All methods are async so adapters can integrate naturally with async 

23 libraries like Bleak. Synchronous libraries can be wrapped by 

24 adapters. 

25 """ 

26 

27 address: str 

28 

29 async def connect(self) -> None: # pragma: no cover - implemented by adapter 

30 """Open a connection to the device.""" 

31 raise NotImplementedError() 

32 

33 async def disconnect(self) -> None: # pragma: no cover - implemented by adapter 

34 """Close the connection to the device.""" 

35 raise NotImplementedError() 

36 

37 async def read_gatt_char(self, char_uuid: BluetoothUUID) -> bytes: # pragma: no cover 

38 """Read the raw bytes of a characteristic identified by `char_uuid`.""" 

39 raise NotImplementedError() 

40 

41 async def write_gatt_char(self, char_uuid: BluetoothUUID, data: bytes) -> None: # pragma: no cover 

42 """Write raw bytes to a characteristic identified by `char_uuid`.""" 

43 raise NotImplementedError() 

44 

45 async def get_services(self) -> Any: # noqa: ANN401 # pragma: no cover # Adapter-specific service collection type 

46 """Return a structure describing services/characteristics from the adapter. 

47 

48 The concrete return type depends on the adapter; `Device` uses 

49 this only for enumeration in examples. Adapters should provide 

50 iterable objects with `.characteristics` elements that have 

51 `.uuid` and `.properties` attributes, or the adapter can return 

52 a mapping. 

53 """ 

54 raise NotImplementedError() 

55 

56 async def start_notify( 

57 self, char_uuid: BluetoothUUID, callback: Callable[[str, bytes], None] 

58 ) -> None: # pragma: no cover 

59 """Start notifications for `char_uuid` and invoke `callback(uuid, data)` on updates.""" 

60 raise NotImplementedError() 

61 

62 async def stop_notify(self, char_uuid: BluetoothUUID) -> None: # pragma: no cover 

63 """Stop notifications for `char_uuid`.""" 

64 raise NotImplementedError() 

65 

66 @property 

67 def is_connected(self) -> bool: # pragma: no cover 

68 """Check if the connection is currently active. 

69 

70 Returns: 

71 True if connected to the device, False otherwise 

72 

73 """ 

74 raise NotImplementedError() 

75 

76 

77__all__ = ["ConnectionManagerProtocol"]