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

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

2 

3Defines an async abstract base class that adapter implementations (Bleak, 

4SimplePyBLE, etc.) must inherit from so the `Device` class can operate 

5independently of the underlying BLE library. 

6 

7Adapters must provide async implementations of all abstract methods below. 

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

9expose an async interface. 

10""" 

11# pylint: disable=duplicate-code # Pattern repetition is expected for protocol definitions 

12# pylint: disable=too-many-public-methods # BLE connection protocol requires complete interface 

13 

14from __future__ import annotations 

15 

16from abc import ABC, abstractmethod 

17from collections.abc import AsyncIterator 

18from typing import TYPE_CHECKING, Callable, ClassVar 

19 

20from bluetooth_sig.types.advertising import AdvertisementData 

21from bluetooth_sig.types.device_types import ( 

22 DeviceService, 

23 ScanFilter, 

24 ScannedDevice, 

25 ScanningMode, 

26) 

27from bluetooth_sig.types.uuid import BluetoothUUID 

28 

29if TYPE_CHECKING: 

30 from bluetooth_sig.types.device_types import ScanDetectionCallback 

31 

32 

33class ConnectionManagerProtocol(ABC): 

34 """Abstract base class describing the transport operations Device expects. 

35 

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

37 libraries like Bleak. Synchronous libraries must be wrapped by adapters 

38 to provide async interfaces. 

39 

40 Subclasses MUST implement all abstract methods and properties. 

41 """ 

42 

43 # Class-level flag to indicate if this backend supports scanning 

44 supports_scanning: ClassVar[bool] = False 

45 

46 def __init__(self, address: str) -> None: 

47 """Initialize the connection manager. 

48 

49 Args: 

50 address: The Bluetooth device address (MAC address) 

51 

52 """ 

53 self._address = address 

54 

55 @property 

56 def address(self) -> str: 

57 """Get the device address. 

58 

59 Returns: 

60 Bluetooth device address (MAC address) 

61 

62 Note: 

63 Subclasses may override this to provide address from underlying library. 

64 

65 """ 

66 return self._address 

67 

68 @abstractmethod 

69 async def connect(self) -> None: 

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

71 

72 @abstractmethod 

73 async def disconnect(self) -> None: 

74 """Close the connection to the device.""" 

75 

76 @abstractmethod 

77 async def read_gatt_char(self, char_uuid: BluetoothUUID) -> bytes: 

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

79 

80 @abstractmethod 

81 async def write_gatt_char(self, char_uuid: BluetoothUUID, data: bytes, response: bool = True) -> None: 

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

83 

84 Args: 

85 char_uuid: The UUID of the characteristic to write to 

86 data: The raw bytes to write 

87 response: If True, use write-with-response (wait for acknowledgment). 

88 If False, use write-without-response (faster but no confirmation). 

89 Default is True for reliability. 

90 

91 """ 

92 

93 @abstractmethod 

94 async def read_gatt_descriptor(self, desc_uuid: BluetoothUUID) -> bytes: 

95 """Read the raw bytes of a descriptor identified by `desc_uuid`. 

96 

97 Args: 

98 desc_uuid: The UUID of the descriptor to read 

99 

100 Returns: 

101 The raw descriptor data as bytes 

102 

103 """ 

104 

105 @abstractmethod 

106 async def write_gatt_descriptor(self, desc_uuid: BluetoothUUID, data: bytes) -> None: 

107 """Write raw bytes to a descriptor identified by `desc_uuid`. 

108 

109 Args: 

110 desc_uuid: The UUID of the descriptor to write to 

111 data: The raw bytes to write 

112 

113 """ 

114 

115 @abstractmethod 

116 async def get_services(self) -> list[DeviceService]: 

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

118 

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

120 this only for enumeration in examples. Adapters should provide 

121 iterable objects with `.characteristics` elements that have 

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

123 a mapping. 

124 """ 

125 

126 @abstractmethod 

127 async def start_notify(self, char_uuid: BluetoothUUID, callback: Callable[[str, bytes], None]) -> None: 

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

129 

130 @abstractmethod 

131 async def stop_notify(self, char_uuid: BluetoothUUID) -> None: 

132 """Stop notifications for `char_uuid`.""" 

133 

134 @abstractmethod 

135 async def pair(self) -> None: 

136 """Pair with the device. 

137 

138 Raises an exception if pairing fails. 

139 

140 Note: 

141 On macOS, pairing is automatic when accessing authenticated characteristics. 

142 This method may not be needed on that platform. 

143 

144 """ 

145 

146 @abstractmethod 

147 async def unpair(self) -> None: 

148 """Unpair from the device. 

149 

150 Raises an exception if unpairing fails. 

151 

152 """ 

153 

154 @abstractmethod 

155 async def read_rssi(self) -> int: 

156 """Read the RSSI (signal strength) during an active connection. 

157 

158 This reads RSSI from the active BLE connection. Not all backends 

159 support this - some only provide RSSI from advertisement data. 

160 

161 Returns: 

162 RSSI value in dBm (typically negative, e.g., -60) 

163 

164 Raises: 

165 NotImplementedError: If this backend doesn't support connection RSSI 

166 RuntimeError: If not connected 

167 

168 """ 

169 

170 @abstractmethod 

171 async def get_advertisement_rssi(self, refresh: bool = False) -> int | None: 

172 """Get the RSSI from advertisement data. 

173 

174 This returns the RSSI from advertisement data. Does not require 

175 an active connection. 

176 

177 Args: 

178 refresh: If True, perform an active scan to get fresh RSSI. 

179 If False, return the cached RSSI from last advertisement. 

180 

181 Returns: 

182 RSSI value in dBm, or None if no advertisement has been received 

183 

184 """ 

185 

186 @abstractmethod 

187 def set_disconnected_callback(self, callback: Callable[[], None]) -> None: 

188 """Set a callback to be invoked when the device disconnects. 

189 

190 Args: 

191 callback: Function to call when disconnection occurs 

192 

193 """ 

194 

195 @property 

196 @abstractmethod 

197 def is_connected(self) -> bool: 

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

199 

200 Returns: 

201 True if connected to the device, False otherwise 

202 

203 """ 

204 

205 @property 

206 @abstractmethod 

207 def mtu_size(self) -> int: 

208 """Get the negotiated MTU size in bytes. 

209 

210 Returns: 

211 The MTU size negotiated for this connection (typically 23-512 bytes) 

212 

213 """ 

214 

215 @property 

216 @abstractmethod 

217 def name(self) -> str: 

218 """Get the name of the device. 

219 

220 Returns: 

221 The name of the device as a string 

222 

223 """ 

224 

225 @classmethod 

226 async def scan( # pylint: disable=too-many-arguments 

227 cls, 

228 timeout: float = 5.0, 

229 *, 

230 filters: ScanFilter | None = None, 

231 scanning_mode: ScanningMode = "active", 

232 adapter: str | None = None, 

233 callback: ScanDetectionCallback | None = None, 

234 ) -> list[ScannedDevice]: 

235 """Scan for nearby BLE devices. 

236 

237 This is a class method that doesn't require an instance. Not all backends 

238 support scanning - check the `supports_scanning` class attribute. 

239 

240 Args: 

241 timeout: Scan duration in seconds (default: 5.0) 

242 filters: Optional filter criteria. Devices not matching filters are excluded. 

243 Filtering may happen at OS level (more efficient) or post-scan depending 

244 on backend capabilities. 

245 scanning_mode: 'active' (default) sends scan requests for scan response data, 

246 'passive' only listens to advertisements (saves power, faster). 

247 Note: Passive scanning is NOT supported on macOS. 

248 adapter: Backend-specific adapter identifier (e.g., "hci0" for BlueZ). 

249 None uses the default adapter. 

250 callback: Optional async or sync function called with each ScannedDevice 

251 as it's discovered. Enables real-time UI updates and early processing. 

252 If async, it's awaited before continuing. 

253 

254 Returns: 

255 List of discovered devices matching the filters 

256 

257 Raises: 

258 NotImplementedError: If this backend doesn't support scanning 

259 

260 Example:: 

261 

262 # Basic scan 

263 devices = await MyConnectionManager.scan(timeout=5.0) 

264 

265 # Filtered scan for Heart Rate monitors 

266 from bluetooth_sig.types.device_types import ScanFilter 

267 

268 filters = ScanFilter(service_uuids=["180d"], rssi_threshold=-70) 

269 devices = await MyConnectionManager.scan(timeout=10.0, filters=filters) 

270 

271 

272 # Scan with real-time callback 

273 async def on_device(device: ScannedDevice) -> None: 

274 print(f"Found: {device.name or device.address}") 

275 

276 

277 devices = await MyConnectionManager.scan(timeout=10.0, callback=on_device) 

278 

279 """ 

280 raise NotImplementedError(f"{cls.__name__} does not support scanning") 

281 

282 @classmethod 

283 async def find_device( 

284 cls, 

285 filters: ScanFilter, 

286 timeout: float = 10.0, 

287 *, 

288 scanning_mode: ScanningMode = "active", 

289 adapter: str | None = None, 

290 ) -> ScannedDevice | None: 

291 """Find the first device matching the filter criteria. 

292 

293 This is more efficient than a full scan when looking for a specific device. 

294 Use ScanFilter to match by address, name, service UUIDs, or custom function. 

295 

296 Args: 

297 filters: Filter criteria. Use ScanFilter(addresses=[...]) for address, 

298 ScanFilter(names=[...]) for name, or ScanFilter(filter_func=...) for 

299 custom matching logic. 

300 timeout: Maximum time to scan in seconds (default: 10.0) 

301 scanning_mode: 'active' or 'passive' scanning mode. 

302 adapter: Backend-specific adapter identifier. None uses default. 

303 

304 Returns: 

305 The first matching device, or None if not found within timeout 

306 

307 Raises: 

308 NotImplementedError: If this backend doesn't support scanning 

309 

310 Example:: 

311 

312 # Find by address 

313 device = await MyConnectionManager.find_device( 

314 ScanFilter(addresses=["AA:BB:CC:DD:EE:FF"]), 

315 timeout=15.0, 

316 ) 

317 

318 # Find by name 

319 device = await MyConnectionManager.find_device( 

320 ScanFilter(names=["Polar H10"]), 

321 timeout=15.0, 

322 ) 

323 

324 

325 # Find with custom filter 

326 def has_apple_data(device: ScannedDevice) -> bool: 

327 if device.advertisement_data is None: 

328 return False 

329 mfr = device.advertisement_data.ad_structures.core.manufacturer_data 

330 return 0x004C in mfr 

331 

332 

333 device = await MyConnectionManager.find_device( 

334 ScanFilter(filter_func=has_apple_data), 

335 timeout=10.0, 

336 ) 

337 

338 """ 

339 raise NotImplementedError(f"{cls.__name__} does not support find_device") 

340 

341 @classmethod 

342 def scan_stream( 

343 cls, 

344 timeout: float | None = 5.0, 

345 *, 

346 filters: ScanFilter | None = None, 

347 scanning_mode: ScanningMode = "active", 

348 adapter: str | None = None, 

349 ) -> AsyncIterator[ScannedDevice]: 

350 """Stream discovered devices as an async iterator. 

351 

352 This provides the most Pythonic way to process devices as they're 

353 discovered, with full async/await support and easy early termination. 

354 

355 Args: 

356 timeout: Scan duration in seconds. None for indefinite. 

357 filters: Optional filter criteria. 

358 scanning_mode: 'active' or 'passive'. 

359 adapter: Backend-specific adapter identifier. 

360 

361 Yields: 

362 ScannedDevice objects as they are discovered 

363 

364 Raises: 

365 NotImplementedError: If this backend doesn't support streaming 

366 

367 Example:: 

368 

369 async for device in MyConnectionManager.scan_stream(timeout=10.0): 

370 print(f"Found: {device.name}") 

371 if device.name == "My Target Device": 

372 break # Stop scanning early 

373 

374 """ 

375 raise NotImplementedError(f"{cls.__name__} does not support scan_stream") 

376 

377 @abstractmethod 

378 async def get_latest_advertisement(self, refresh: bool = False) -> AdvertisementData | None: 

379 """Return the most recently received advertisement data. 

380 

381 Args: 

382 refresh: If True, perform an active scan to get fresh data. 

383 If False, return the last cached advertisement. 

384 

385 Returns: 

386 Latest AdvertisementData, or None if none received yet 

387 

388 """ 

389 

390 @classmethod 

391 @abstractmethod 

392 def convert_advertisement(cls, advertisement: object) -> AdvertisementData: 

393 """Convert framework-specific advertisement data to AdvertisementData. 

394 

395 This method bridges the gap between BLE framework-specific advertisement 

396 representations (Bleak's AdvertisementData, SimpleBLE's Peripheral, etc.) 

397 and our unified AdvertisementData type. 

398 

399 Each connection manager implementation knows how to extract manufacturer_data, 

400 service_data, local_name, RSSI, etc. from its framework's format. 

401 

402 Args: 

403 advertisement: Framework-specific advertisement object 

404 (e.g., bleak.backends.scanner.AdvertisementData) 

405 

406 Returns: 

407 Unified AdvertisementData with ad_structures populated 

408 

409 """ 

410 

411 

412__all__ = ["ConnectionManagerProtocol"]