Coverage for src/bluetooth_sig/device/peripheral.py: 76%

1"""Peripheral manager protocol for BLE GATT server adapters. 

2 

3Defines an async abstract base class that peripheral adapter implementations 

4(bless, bluez_peripheral, etc.) must inherit from to create BLE GATT servers 

5that broadcast services and characteristics. 

6 

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

8""" 

9 

10from __future__ import annotations 

11 

12from abc import ABC, abstractmethod 

13from collections.abc import Callable 

14from typing import ClassVar 

15 

16from typing_extensions import Self 

17 

18from bluetooth_sig.types.company import CompanyIdentifier, ManufacturerData 

19from bluetooth_sig.types.peripheral_types import CharacteristicDefinition, ServiceDefinition 

20from bluetooth_sig.types.uuid import BluetoothUUID 

21 

22 

23class PeripheralManagerProtocol(ABC): 

24 """Abstract base class for BLE peripheral/GATT server implementations. 

25 

26 .. deprecated:: 0.5.0 

27 Scheduled for removal; see ``docs/source/explanation/limitations.md``. 

28 

29 This protocol defines the interface for creating BLE peripherals that 

30 broadcast services and characteristics. Implementations wrap backend 

31 libraries like bless, bluez_peripheral, etc. 

32 

33 Uses a fluent builder pattern for advertisement configuration - call 

34 configuration methods before start() to customise advertising. 

35 

36 The workflow is: 

37 1. Create peripheral manager with a device name 

38 2. Configure advertising (optional): with_manufacturer_data(), with_tx_power(), etc. 

39 3. Add services and characteristics (using CharacteristicDefinition) 

40 4. Start advertising 

41 5. Update characteristic values as needed 

42 6. Stop when done 

43 

44 Example:: 

45 >>> from bluetooth_sig.gatt.characteristics import BatteryLevelCharacteristic 

46 >>> from bluetooth_sig.gatt.services import BatteryService 

47 >>> from bluetooth_sig.types.company import ManufacturerData 

48 >>> 

49 >>> # Create peripheral with fluent configuration 

50 >>> peripheral = SomePeripheralManager("My Sensor") 

51 >>> peripheral.with_tx_power(-10).with_connectable(True) 

52 >>> 

53 >>> # Define a service with battery level 

54 >>> char = BatteryLevelCharacteristic() 

55 >>> char_def = CharacteristicDefinition.from_characteristic(char, 85) 

56 >>> 

57 >>> service = ServiceDefinition( 

58 ... uuid=BatteryService.get_class_uuid(), 

59 ... characteristics=[char_def], 

60 ... ) 

61 >>> 

62 >>> await peripheral.add_service(service) 

63 >>> await peripheral.start() 

64 >>> 

65 >>> # Later, update the battery level 

66 >>> await peripheral.update_characteristic("2A19", char.build_value(75)) 

67 

68 """ 

69 

70 # Class-level flag indicating backend capabilities 

71 supports_advertising: ClassVar[bool] = True 

72 

73 def __init__(self, name: str) -> None: 

74 """Initialize the peripheral manager. 

75 

76 Args: 

77 name: The advertised device name visible to scanners 

78 

79 """ 

80 self._name = name 

81 

82 # Service and characteristic tracking 

83 self._services: list[ServiceDefinition] = [] 

84 self._char_definitions: dict[str, CharacteristicDefinition] = {} 

85 

86 # Advertisement configuration (set via fluent methods) 

87 self._manufacturer_data: ManufacturerData | None = None 

88 self._service_data: dict[BluetoothUUID, bytes] = {} 

89 self._tx_power: int | None = None 

90 self._is_connectable = True 

91 self._is_discoverable = True 

92 

93 # Callbacks for read/write handling 

94 self._read_callbacks: dict[str, Callable[[], bytearray]] = {} 

95 self._write_callbacks: dict[str, Callable[[bytearray], None]] = {} 

96 

97 @property 

98 def name(self) -> str: 

99 """Get the advertised device name. 

100 

101 Returns: 

102 The device name as it appears to BLE scanners 

103 

104 """ 

105 return self._name 

106 

107 @property 

108 def services(self) -> list[ServiceDefinition]: 

109 """Get the list of registered services. 

110 

111 Returns: 

112 List of ServiceDefinition objects added to this peripheral. 

113 

114 """ 

115 return self._services 

116 

117 @property 

118 def manufacturer_data(self) -> ManufacturerData | None: 

119 """Get the configured manufacturer data. 

120 

121 Returns: 

122 ManufacturerData if configured, None otherwise. 

123 

124 """ 

125 return self._manufacturer_data 

126 

127 @property 

128 def service_data(self) -> dict[BluetoothUUID, bytes]: 

129 """Get the configured service data. 

130 

131 Returns: 

132 Dictionary mapping service UUIDs to data bytes. 

133 

134 """ 

135 return self._service_data 

136 

137 @property 

138 def tx_power(self) -> int | None: 

139 """Get the configured TX power level. 

140 

141 Returns: 

142 TX power in dBm if configured, None otherwise. 

143 

144 """ 

145 return self._tx_power 

146 

147 @property 

148 def is_connectable_config(self) -> bool: 

149 """Get the connectable configuration. 

150 

151 Returns: 

152 True if peripheral is configured to accept connections. 

153 

154 """ 

155 return self._is_connectable 

156 

157 @property 

158 def is_discoverable_config(self) -> bool: 

159 """Get the discoverable configuration. 

160 

161 Returns: 

162 True if peripheral is configured to be discoverable. 

163 

164 """ 

165 return self._is_discoverable 

166 

167 def with_manufacturer_data(self, manufacturer_data: ManufacturerData) -> Self: 

168 r"""Set manufacturer-specific advertising data. 

169 

170 Args: 

171 manufacturer_data: ManufacturerData instance from the types module. 

172 

173 Returns: 

174 Self for method chaining. 

175 

176 Raises: 

177 RuntimeError: If called after start(). 

178 

179 Example:: 

180 >>> from bluetooth_sig.types.company import ManufacturerData 

181 >>> mfr = ManufacturerData.from_id_and_payload(0x004C, b"\x02\x15...") 

182 >>> peripheral.with_manufacturer_data(mfr) 

183 

184 """ 

185 if self.is_advertising: 

186 raise RuntimeError("Cannot configure after peripheral has started") 

187 self._manufacturer_data = manufacturer_data 

188 return self 

189 

190 def with_manufacturer_id( 

191 self, 

192 company_id: int | CompanyIdentifier, 

193 payload: bytes, 

194 ) -> Self: 

195 r"""Set manufacturer data from company ID and payload. 

196 

197 Args: 

198 company_id: Bluetooth SIG company identifier (e.g., 0x004C for Apple) 

199 or CompanyIdentifier instance. 

200 payload: Manufacturer-specific payload bytes. 

201 

202 Returns: 

203 Self for method chaining. 

204 

205 Raises: 

206 RuntimeError: If called after start(). 

207 

208 Example:: 

209 >>> peripheral.with_manufacturer_id(0x004C, b"\x02\x15...") 

210 

211 """ 

212 if self.is_advertising: 

213 raise RuntimeError("Cannot configure after peripheral has started") 

214 cid = company_id if isinstance(company_id, int) else company_id.id 

215 self._manufacturer_data = ManufacturerData.from_id_and_payload(cid, payload) 

216 return self 

217 

218 def with_service_data( 

219 self, 

220 service_uuid: BluetoothUUID, 

221 data: bytes, 

222 ) -> Self: 

223 r"""Add service data to advertisement. 

224 

225 Args: 

226 service_uuid: BluetoothUUID of the service. 

227 data: Service-specific data bytes. 

228 

229 Returns: 

230 Self for method chaining. 

231 

232 Raises: 

233 RuntimeError: If called after start(). 

234 

235 Example:: 

236 >>> from bluetooth_sig.gatt.services import BatteryService 

237 >>> peripheral.with_service_data( 

238 ... BatteryService.get_class_uuid(), 

239 ... b"\x50", # 80% battery 

240 ... ) 

241 

242 """ 

243 if self.is_advertising: 

244 raise RuntimeError("Cannot configure after peripheral has started") 

245 self._service_data[service_uuid] = data 

246 return self 

247 

248 def with_tx_power(self, power_dbm: int) -> Self: 

249 """Set TX power level for advertising. 

250 

251 Args: 

252 power_dbm: Transmission power in dBm (-127 to +127). 

253 

254 Returns: 

255 Self for method chaining. 

256 

257 Raises: 

258 RuntimeError: If called after start(). 

259 

260 """ 

261 if self.is_advertising: 

262 raise RuntimeError("Cannot configure after peripheral has started") 

263 self._tx_power = power_dbm 

264 return self 

265 

266 def with_connectable(self, connectable: bool) -> Self: 

267 """Set whether the peripheral accepts connections. 

268 

269 Args: 

270 connectable: True to accept connections (default), False for broadcast only. 

271 

272 Returns: 

273 Self for method chaining. 

274 

275 Raises: 

276 RuntimeError: If called after start(). 

277 

278 """ 

279 if self.is_advertising: 

280 raise RuntimeError("Cannot configure after peripheral has started") 

281 self._is_connectable = connectable 

282 return self 

283 

284 def with_discoverable(self, discoverable: bool) -> Self: 

285 """Set whether the peripheral is discoverable. 

286 

287 Args: 

288 discoverable: True to be discoverable (default), False otherwise. 

289 

290 Returns: 

291 Self for method chaining. 

292 

293 Raises: 

294 RuntimeError: If called after start(). 

295 

296 """ 

297 if self.is_advertising: 

298 raise RuntimeError("Cannot configure after peripheral has started") 

299 self._is_discoverable = discoverable 

300 return self 

301 

302 # ------------------------------------------------------------------------- 

303 # Service Management (Generic Implementation) 

304 # ------------------------------------------------------------------------- 

305 

306 async def add_service(self, service: ServiceDefinition) -> None: 

307 """Add a GATT service to the peripheral. 

308 

309 Services must be added before calling start(). Each service contains 

310 one or more characteristics that clients can interact with. 

311 

312 Args: 

313 service: The service definition to add 

314 

315 Raises: 

316 RuntimeError: If called after start() 

317 

318 """ 

319 if self.is_advertising: 

320 raise RuntimeError("Cannot add services after peripheral has started") 

321 

322 self._services.append(service) 

323 

324 # Track characteristic definitions for later lookup 

325 for char_def in service.characteristics: 

326 uuid_upper = str(char_def.uuid).upper() 

327 self._char_definitions[uuid_upper] = char_def 

328 

329 # Register any callbacks from the definition 

330 if char_def.on_read: 

331 self._read_callbacks[uuid_upper] = char_def.on_read 

332 if char_def.on_write: 

333 self._write_callbacks[uuid_upper] = char_def.on_write 

334 

335 def get_characteristic_definition( 

336 self, 

337 char_uuid: str | BluetoothUUID, 

338 ) -> CharacteristicDefinition | None: 

339 """Get the characteristic definition by UUID. 

340 

341 Args: 

342 char_uuid: UUID of the characteristic. 

343 

344 Returns: 

345 CharacteristicDefinition if found, None otherwise. 

346 

347 """ 

348 uuid_upper = str(char_uuid).upper() 

349 return self._char_definitions.get(uuid_upper) 

350 

351 def set_read_callback( 

352 self, 

353 char_uuid: str | BluetoothUUID, 

354 callback: Callable[[], bytearray], 

355 ) -> None: 

356 """Set a callback for dynamic read value generation. 

357 

358 When a client reads the characteristic, this callback will be invoked 

359 to generate the current value instead of returning the stored value. 

360 

361 Args: 

362 char_uuid: UUID of the characteristic 

363 callback: Function that returns the encoded value to serve 

364 

365 Raises: 

366 KeyError: If characteristic UUID not found 

367 

368 """ 

369 uuid_str = str(char_uuid).upper() 

370 if uuid_str not in self._char_definitions: 

371 raise KeyError(f"Characteristic {uuid_str} not found") 

372 self._read_callbacks[uuid_str] = callback 

373 

374 def set_write_callback( 

375 self, 

376 char_uuid: str | BluetoothUUID, 

377 callback: Callable[[bytearray], None], 

378 ) -> None: 

379 """Set a callback for handling client writes. 

380 

381 When a client writes to the characteristic, this callback will be 

382 invoked with the written data. 

383 

384 Args: 

385 char_uuid: UUID of the characteristic 

386 callback: Function called with the written data 

387 

388 Raises: 

389 KeyError: If characteristic UUID not found 

390 

391 """ 

392 uuid_str = str(char_uuid).upper() 

393 if uuid_str not in self._char_definitions: 

394 raise KeyError(f"Characteristic {uuid_str} not found") 

395 self._write_callbacks[uuid_str] = callback 

396 

397 # ------------------------------------------------------------------------- 

398 # Abstract Methods (Backend-Specific Implementation Required) 

399 # ------------------------------------------------------------------------- 

400 

401 @abstractmethod 

402 async def start(self) -> None: 

403 """Start advertising and accepting connections. 

404 

405 Backend implementations must: 

406 1. Create the platform-specific GATT server 

407 2. Register all services and characteristics from self._services 

408 3. Configure advertisement data from self._manufacturer_data, etc. 

409 4. Begin advertising 

410 

411 Raises: 

412 RuntimeError: If no services have been added 

413 

414 """ 

415 

416 @abstractmethod 

417 async def stop(self) -> None: 

418 """Stop advertising and disconnect all clients.""" 

419 

420 @property 

421 @abstractmethod 

422 def is_advertising(self) -> bool: 

423 """Check if the peripheral is currently advertising. 

424 

425 Returns: 

426 True if advertising, False otherwise 

427 

428 """ 

429 

430 @abstractmethod 

431 async def update_characteristic( 

432 self, 

433 char_uuid: str | BluetoothUUID, 

434 value: bytearray, 

435 *, 

436 notify: bool = True, 

437 ) -> None: 

438 """Update a characteristic's value. 

439 

440 This sets the new value that will be returned when clients read the 

441 characteristic. If notify=True and the characteristic supports 

442 notifications, subscribed clients will be notified of the change. 

443 

444 Args: 

445 char_uuid: UUID of the characteristic to update 

446 value: New encoded value (use characteristic.build_value() to encode) 

447 notify: If True, notify subscribed clients of the change 

448 

449 Raises: 

450 KeyError: If characteristic UUID not found 

451 RuntimeError: If peripheral not started 

452 

453 """ 

454 

455 @abstractmethod 

456 async def get_characteristic_value(self, char_uuid: str | BluetoothUUID) -> bytearray: 

457 """Get the current value of a characteristic. 

458 

459 Args: 

460 char_uuid: UUID of the characteristic 

461 

462 Returns: 

463 The current encoded value 

464 

465 Raises: 

466 KeyError: If characteristic UUID not found 

467 

468 """ 

469 

470 @property 

471 def connected_clients(self) -> int: 

472 """Get the number of currently connected clients. 

473 

474 Returns: 

475 Number of connected BLE centrals 

476 

477 Raises: 

478 NotImplementedError: If backend doesn't track connections 

479 

480 """ 

481 raise NotImplementedError(f"{self.__class__.__name__} does not track connected clients") 

482 

483 

484__all__ = [ 

485 "CharacteristicDefinition", 

486 "PeripheralManagerProtocol", 

487 "ServiceDefinition", 

488]