Coverage for src/bluetooth_sig/device/device.py: 78%

1"""Device class for grouping BLE device services, characteristics, encryption, and advertiser data.

3Provides a high-level Device abstraction that groups all services,

4characteristics, encryption requirements, and advertiser data for a BLE

5device.

6"""

8from __future__ import annotations

10from collections.abc import Callable

11from typing import Any, TypeVar, overload

13from ..advertising.registry import PayloadInterpreterRegistry

14from ..gatt.characteristics.base import BaseCharacteristic

15from ..gatt.characteristics.registry import CharacteristicName

16from ..gatt.context import DeviceInfo

17from ..gatt.descriptors.base import BaseDescriptor

18from ..gatt.descriptors.characteristic_user_description import (

19 CharacteristicUserDescriptionData,

20 CharacteristicUserDescriptionDescriptor,

21)

22from ..gatt.descriptors.registry import DescriptorRegistry

23from ..types import (

24 DescriptorData,

25 DescriptorInfo,

26)

27from ..types.advertising.result import AdvertisementData

28from ..types.device_types import ScannedDevice

29from ..types.gatt_enums import ServiceName

30from ..types.uuid import BluetoothUUID

31from .advertising import DeviceAdvertising

32from .characteristic_io import CharacteristicIO

33from .client import ClientManagerProtocol

34from .connected import DeviceConnected, DeviceEncryption, DeviceService

35from .dependency_resolver import DependencyResolutionMode, DependencyResolver

36from .protocols import SIGTranslatorProtocol

38# Type variable for generic characteristic return types

39T = TypeVar("T")

41__all__ = [

42 "DependencyResolutionMode",

43 "Device",

44 "DeviceAdvertising",

45 "DeviceConnected",

46 "SIGTranslatorProtocol",

47]

50class Device: # pylint: disable=too-many-instance-attributes,too-many-public-methods

51 r"""High-level BLE device abstraction using composition pattern.

53 Coordinates between connected GATT operations and advertising packet

54 interpretation through two subsystems:

56 - ``device.connected`` — GATT connection, services, characteristics

57 - ``device.advertising`` — Vendor-specific advertising interpretation

59 Convenience methods delegate to the appropriate subsystem, so callers

60 can use ``await device.read("battery_level")`` without knowing which

61 subsystem handles it.

62 """

64 def __init__(self, connection_manager: ClientManagerProtocol, translator: SIGTranslatorProtocol) -> None:

65 """Initialise Device instance with connection manager and translator.

67 Args:

68 connection_manager: Connection manager implementing ClientManagerProtocol

69 translator: SIGTranslatorProtocol instance

71 """

72 self.connection_manager = connection_manager

73 self.translator = translator

74 self._name: str = ""

76 # Connected subsystem (composition pattern)

77 self.connected = DeviceConnected(

78 mac_address=self.address,

79 connection_manager=connection_manager,

80 )

82 # Advertising subsystem (composition pattern)

83 self.advertising = DeviceAdvertising(self.address, connection_manager)

84 # Set up registry for auto-detection

85 self.advertising.set_registry(PayloadInterpreterRegistry())

87 # Dependency resolution delegate

88 self._dep_resolver = DependencyResolver(connection_manager, self.connected)

90 # Characteristic I/O delegate

91 self._char_io = CharacteristicIO(connection_manager, translator, self._dep_resolver, lambda: self.device_info)

93 # Cache for device_info property and last advertisement

94 self._device_info_cache: DeviceInfo | None = None

95 self._last_advertisement: AdvertisementData | None = None

97 def __str__(self) -> str:

98 """Return string representation of Device.

100 Returns:

101 str: String representation of Device.

102

103 """

104 service_count = len(self.connected.services)

105 char_count = sum(len(service.characteristics) for service in self.connected.services.values())

106 return f"Device({self.address}, name={self.name}, {service_count} services, {char_count} characteristics)"

107

108 @property

109 def services(self) -> dict[str, DeviceService]:

110 """GATT services discovered on device.

111

112 Delegates to device.connected.services.

113

114 Returns:

115 Dictionary of service UUID → DeviceService

116

117 """

118 return self.connected.services

119

120 @property

121 def encryption(self) -> DeviceEncryption:

122 """Encryption state for connected device.

123

124 Delegates to device.connected.encryption.

125

126 Returns:

127 DeviceEncryption instance

128

129 """

130 return self.connected.encryption

131

132 @property

133 def address(self) -> str:

134 """Get the device address from the connection manager.

135

136 Returns:

137 BLE device address

138

139 """

140 return self.connection_manager.address

141

142 @staticmethod

143 async def scan(manager_class: type[ClientManagerProtocol], timeout: float = 5.0) -> list[ScannedDevice]:

144 """Scan for nearby BLE devices using a specific connection manager.

145

146 This is a static method that doesn't require a Device instance.

147 Use it to discover devices before creating Device instances.

148

149 Args:

150 manager_class: The connection manager class to use for scanning

151 (e.g., BleakRetryClientManager)

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

153

154 Returns:

155 List of discovered devices

156

157 Raises:

158 NotImplementedError: If the connection manager doesn't support scanning

159

160 Example::

161

162 from bluetooth_sig.device import Device

163 from connection_managers.bleak_retry import BleakRetryClientManager

164

165 # Scan for devices

166 devices = await Device.scan(BleakRetryClientManager, timeout=10.0)

167

168 # Create Device instance for first discovered device

169 if devices:

170 translator = BluetoothSIGTranslator()

171 device = Device(devices[0].address, translator)

172

173 """

174 return await manager_class.scan(timeout)

175

176 async def connect(self) -> None:

177 """Connect to the BLE device.

178

179 Convenience method that delegates to device.connected.connect().

180

181 Raises:

182 RuntimeError: If no connection manager is attached

183

184 """

185 await self.connected.connect()

186

187 async def disconnect(self) -> None:

188 """Disconnect from the BLE device.

189

190 Convenience method that delegates to device.connected.disconnect().

191

192 Raises:

193 RuntimeError: If no connection manager is attached

194

195 """

196 await self.connected.disconnect()

197

198 # ------------------------------------------------------------------

199 # Characteristic I/O (delegated to CharacteristicIO)

200 # ------------------------------------------------------------------

201

202 @overload

203 async def read(

204 self,

205 char: type[BaseCharacteristic[T]],

206 resolution_mode: DependencyResolutionMode = ...,

207 ) -> T | None: ...

208

209 @overload

210 async def read(

211 self,

212 char: str | CharacteristicName,

213 resolution_mode: DependencyResolutionMode = ...,

214 ) -> Any | None: ... # noqa: ANN401 # Runtime UUID dispatch cannot be type-safe

215

216 async def read(

217 self,

218 char: str | CharacteristicName | type[BaseCharacteristic[T]],

219 resolution_mode: DependencyResolutionMode = DependencyResolutionMode.NORMAL,

220 ) -> T | Any | None: # Runtime UUID dispatch cannot be type-safe

221 """Read a characteristic value from the device.

222

223 Delegates to :class:`CharacteristicIO`.

224

225 Args:

226 char: Name, enum, or characteristic class to read.

227 resolution_mode: How to handle automatic dependency resolution.

228

229 Returns:

230 Parsed characteristic value or None if read fails.

231

232 Raises:

233 RuntimeError: If no connection manager is attached

234 ValueError: If required dependencies cannot be resolved

235

236 """

237 return await self._char_io.read(char, resolution_mode)

238

239 @overload

240 async def write(

241 self,

242 char: type[BaseCharacteristic[T]],

243 data: T,

244 response: bool = ...,

245 ) -> None: ...

246

247 @overload

248 async def write(

249 self,

250 char: str | CharacteristicName,

251 data: bytes,

252 response: bool = ...,

253 ) -> None: ...

254

255 async def write(

256 self,

257 char: str | CharacteristicName | type[BaseCharacteristic[T]],

258 data: bytes | T,

259 response: bool = True,

260 ) -> None:

261 r"""Write data to a characteristic on the device.

262

263 Delegates to :class:`CharacteristicIO`.

264

265 Args:

266 char: Name, enum, or characteristic class to write to.

267 data: Raw bytes (for string/enum) or typed value (for characteristic class).

268 response: If True, use write-with-response. Default is True.

269

270 Raises:

271 RuntimeError: If no connection manager is attached

272 CharacteristicEncodeError: If encoding fails (when using characteristic class)

273

274 """

275 await self._char_io.write(char, data, response=response) # type: ignore[arg-type, misc] # Union narrowing handled by overloads; mypy can't infer across delegation

276

277 @overload

278 async def start_notify(

279 self,

280 char: type[BaseCharacteristic[T]],

281 callback: Callable[[T], None],

282 ) -> None: ...

283

284 @overload

285 async def start_notify(

286 self,

287 char: str | CharacteristicName,

288 callback: Callable[[Any], None],

289 ) -> None: ...

290

291 async def start_notify(

292 self,

293 char: str | CharacteristicName | type[BaseCharacteristic[T]],

294 callback: Callable[[T], None] | Callable[[Any], None],

295 ) -> None:

296 """Start notifications for a characteristic.

297

298 Delegates to :class:`CharacteristicIO`.

299

300 Args:

301 char: Name, enum, or characteristic class to monitor.

302 callback: Function to call when notifications are received.

303

304 Raises:

305 RuntimeError: If no connection manager is attached

306

307 """

308 await self._char_io.start_notify(char, callback)

309

310 async def stop_notify(self, char_name: str | CharacteristicName) -> None:

311 """Stop notifications for a characteristic.

312

313 Delegates to :class:`CharacteristicIO`.

314

315 Args:

316 char_name: Characteristic name or UUID

317

318 """

319 await self._char_io.stop_notify(char_name)

320

321 async def read_descriptor(

322 self,

323 desc_uuid: BluetoothUUID | BaseDescriptor,

324 *,

325 characteristic_uuid: BluetoothUUID | str | None = None,

326 ) -> DescriptorData:

327 """Read a descriptor value from the device.

328

329 Args:

330 desc_uuid: UUID of the descriptor to read or BaseDescriptor instance

331 characteristic_uuid: When reading a characteristic-scoped descriptor

332 (e.g. User Description 0x2901), pass the parent characteristic

333 UUID to store the parsed label on that characteristic instance.

334

335 Returns:

336 Parsed descriptor data with metadata

337

338 Raises:

339 RuntimeError: If no connection manager is attached

340

341 """

342 # Extract UUID from BaseDescriptor if needed

343 uuid = desc_uuid.uuid if isinstance(desc_uuid, BaseDescriptor) else desc_uuid

344

345 raw_data = await self.connected.read_descriptor(uuid)

346

347 # Try to create a descriptor instance and parse the data

348 descriptor = DescriptorRegistry.create_descriptor(str(uuid))

349 if descriptor:

350 parsed = descriptor.parse_value(raw_data)

351 if characteristic_uuid is not None:

352 self._attach_user_description_from_descriptor(characteristic_uuid, uuid, parsed)

353 return parsed

354

355 # If no registered descriptor found, return unparsed DescriptorData

356 unparsed = DescriptorData(

357 info=DescriptorInfo(uuid=uuid, name="Unknown Descriptor"),

358 value=raw_data,

359 raw_data=raw_data,

360 parse_success=False,

361 error_message="Unknown descriptor UUID - no parser available",

362 )

363 if characteristic_uuid is not None:

364 self._attach_user_description_from_descriptor(characteristic_uuid, uuid, unparsed)

365 return unparsed

366

367 def attach_user_description(self, characteristic_uuid: BluetoothUUID | str, raw_bytes: bytes) -> str | None:

368 """Parse and store User Description bytes on a cached characteristic.

369

370 Args:

371 characteristic_uuid: Parent characteristic UUID

372 raw_bytes: Raw User Description descriptor bytes (UTF-8)

373

374 Returns:

375 Parsed description string, or None if characteristic not cached

376 """

377 char = self.connected.get_cached_characteristic(BluetoothUUID(characteristic_uuid))

378 if char is None:

379 return None

380 descriptor = CharacteristicUserDescriptionDescriptor()

381 parsed = descriptor.parse_value(raw_bytes)

382 if isinstance(parsed.value, CharacteristicUserDescriptionData):

383 char.user_description = parsed.value.description

384 return char.user_description

385 return None

386

387 def get_user_description_for_characteristic(self, characteristic_uuid: BluetoothUUID | str) -> str | None:

388 """Return cached User Description label for a characteristic, if set.

389

390 Args:

391 characteristic_uuid: Characteristic UUID from discovered services

392

393 Returns:

394 User Description string when previously read or attached, else None

395 """

396 char = self.connected.get_cached_characteristic(BluetoothUUID(characteristic_uuid))

397 return char.user_description if char is not None else None

398

399 def _attach_user_description_from_descriptor(

400 self,

401 characteristic_uuid: BluetoothUUID | str,

402 descriptor_uuid: BluetoothUUID,

403 descriptor_data: DescriptorData,

404 ) -> None:

405 """Store User Description text on the parent characteristic when applicable."""

406 user_desc_uuid = CharacteristicUserDescriptionDescriptor().uuid

407 if BluetoothUUID(descriptor_uuid).normalized != user_desc_uuid.normalized:

408 return

409 char = self.connected.get_cached_characteristic(BluetoothUUID(characteristic_uuid))

410 if char is None:

411 return

412 value = descriptor_data.value

413 if isinstance(value, CharacteristicUserDescriptionData):

414 char.user_description = value.description

415 elif isinstance(value, str):

416 char.user_description = value

417

418 async def write_descriptor(self, desc_uuid: BluetoothUUID | BaseDescriptor, data: bytes | DescriptorData) -> None:

419 """Write data to a descriptor on the device.

420

421 Args:

422 desc_uuid: UUID of the descriptor to write to or BaseDescriptor instance

423 data: Either raw bytes to write, or a DescriptorData object.

424 If DescriptorData is provided, its raw_data will be written.

425

426 Raises:

427 RuntimeError: If no connection manager is attached

428

429 """

430 # Extract UUID from BaseDescriptor if needed

431 uuid = desc_uuid.uuid if isinstance(desc_uuid, BaseDescriptor) else desc_uuid

432

433 # Extract raw bytes from DescriptorData if needed

434 raw_data: bytes

435 raw_data = data.raw_data if isinstance(data, DescriptorData) else data

436

437 await self.connected.write_descriptor(uuid, raw_data)

438

439 async def pair(self) -> None:

440 """Pair with the device.

441

442 Convenience method that delegates to device.connected.pair().

443

444 Raises:

445 RuntimeError: If no connection manager is attached

446

447 """

448 await self.connected.pair()

449

450 async def unpair(self) -> None:

451 """Unpair from the device.

452

453 Convenience method that delegates to device.connected.unpair().

454

455 Raises:

456 RuntimeError: If no connection manager is attached

457

458 """

459 await self.connected.unpair()

460

461 async def read_rssi(self) -> int:

462 """Read the RSSI (signal strength) of the connection.

463

464 Convenience method that delegates to device.connected.read_rssi().

465

466 Returns:

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

468

469 Raises:

470 RuntimeError: If no connection manager is attached

471

472 """

473 return await self.connected.read_rssi()

474

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

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

477

478 Convenience method that delegates to device.connected.set_disconnected_callback().

479

480 Args:

481 callback: Function to call when disconnection occurs

482

483 Raises:

484 RuntimeError: If no connection manager is attached

485

486 """

487 self.connected.set_disconnected_callback(callback)

488

489 @property

490 def mtu_size(self) -> int:

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

492

493 Delegates to device.connected.mtu_size.

494

495 Returns:

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

497

498 Raises:

499 RuntimeError: If no connection manager is attached

500

501 """

502 return self.connected.mtu_size

503

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

505 """Get advertisement data from the connection manager.

506

507 Args:

508 refresh: If ``True``, perform an active scan for fresh data.

509

510 Returns:

511 Interpreted :class:`AdvertisementData`, or ``None`` if unavailable.

512

513 Raises:

514 RuntimeError: If no connection manager is attached.

515

516 """

517 if not self.connection_manager:

518 raise RuntimeError("No connection manager attached to Device")

519

520 advertisement = await self.connection_manager.get_latest_advertisement(refresh=refresh)

521 if advertisement is None:

522 return None

523

524 # Process and cache the advertisement

525 processed_ad, _result = self.advertising.process_from_connection_manager(advertisement)

526 self._last_advertisement = processed_ad

527

528 # Update device name if not set

529 if advertisement.ad_structures.core.local_name and not self.name:

530 self.name = advertisement.ad_structures.core.local_name

531

532 return processed_ad

533

534 def parse_raw_advertisement(self, raw_data: bytes, rssi: int = 0) -> AdvertisementData:

535 """Parse raw advertising PDU bytes directly.

536

537 Args:

538 raw_data: Raw BLE advertising PDU bytes.

539 rssi: Received signal strength in dBm.

540

541 Returns:

542 AdvertisementData with parsed AD structures and vendor interpretation.

543

544 """

545 # Delegate to advertising subsystem

546 advertisement, _result = self.advertising.parse_raw_pdu(raw_data, rssi)

547 self._last_advertisement = advertisement

548

549 # Update device name if present and not already set

550 if advertisement.ad_structures.core.local_name and not self.name:

551 self.name = advertisement.ad_structures.core.local_name

552

553 return advertisement

554

555 def get_characteristic_data(self, char_uuid: BluetoothUUID) -> Any | None: # noqa: ANN401 # Heterogeneous cache

556 """Get parsed characteristic data via ``characteristic.last_parsed``.

557

558 Args:

559 char_uuid: UUID of the characteristic.

560

561 Returns:

562 Parsed characteristic value if found, ``None`` otherwise.

563

564 """

565 char_instance = self.connected.get_cached_characteristic(char_uuid)

566 if char_instance is not None:

567 return char_instance.last_parsed

568 return None

569

570 async def discover_services(self) -> dict[str, Any]:

571 """Discover services and characteristics from the connected BLE device.

572

573 Performs BLE service discovery via the connection manager. The

574 discovered :class:`DeviceService` objects (with characteristic

575 instances and runtime properties) are stored in ``self.services``.

576

577 Returns:

578 Dictionary mapping service UUIDs to DeviceService objects.

579

580 Raises:

581 RuntimeError: If no connection manager is attached.

582

583 """

584 # Delegate to connected subsystem

585 services_list = await self.connected.discover_services()

586

587 # Invalidate device_info cache since services changed

588 self._device_info_cache = None

589

590 # Return as dict for backward compatibility

591 return {str(svc.uuid): svc for svc in services_list}

592

593 async def get_characteristic_info(self, char_uuid: str) -> Any | None: # noqa: ANN401 # Adapter-specific characteristic metadata

594 """Get information about a characteristic from the connection manager.

595

596 Args:

597 char_uuid: UUID of the characteristic

598

599 Returns:

600 Characteristic information or None if not found

601

602 Raises:

603 RuntimeError: If no connection manager is attached

604

605 """

606 if not self.connection_manager:

607 raise RuntimeError("No connection manager attached to Device")

608

609 services_data = await self.connection_manager.get_services()

610 for service_info in services_data:

611 for char_uuid_key, char_info in service_info.characteristics.items():

612 if char_uuid_key == char_uuid:

613 return char_info

614 return None

615

616 async def read_multiple(self, char_names: list[str | CharacteristicName]) -> dict[str, Any | None]:

617 """Read multiple characteristics in batch.

618

619 Delegates to :class:`CharacteristicIO`.

620

621 Args:

622 char_names: List of characteristic names or enums to read

623

624 Returns:

625 Dictionary mapping characteristic UUIDs to parsed values

626

627 """

628 return await self._char_io.read_multiple(char_names)

629

630 async def write_multiple(

631 self, data_map: dict[str | CharacteristicName, bytes], response: bool = True

632 ) -> dict[str, bool]:

633 """Write to multiple characteristics in batch.

634

635 Delegates to :class:`CharacteristicIO`.

636

637 Args:

638 data_map: Dictionary mapping characteristic names/enums to data bytes

639 response: If True, use write-with-response for all writes.

640

641 Returns:

642 Dictionary mapping characteristic UUIDs to success status

643

644 """

645 return await self._char_io.write_multiple(data_map, response=response)

646

647 @property

648 def device_info(self) -> DeviceInfo:

649 """Get cached device info object.

650

651 Returns:

652 DeviceInfo with current device metadata

653

654 """

655 if self._device_info_cache is None:

656 self._device_info_cache = DeviceInfo(

657 address=self.address,

658 name=self.name,

659 manufacturer_data=self._last_advertisement.ad_structures.core.manufacturer_data

660 if self._last_advertisement

661 else {},

662 service_uuids=self._last_advertisement.ad_structures.core.service_uuids

663 if self._last_advertisement

664 else [],

665 )

666 else:

667 # Update existing cache object with current data

668 self._device_info_cache.name = self.name

669 if self._last_advertisement:

670 self._device_info_cache.manufacturer_data = (

671 self._last_advertisement.ad_structures.core.manufacturer_data

672 )

673 self._device_info_cache.service_uuids = self._last_advertisement.ad_structures.core.service_uuids

674 return self._device_info_cache

675

676 @property

677 def name(self) -> str:

678 """Get the device name."""

679 return self._name

680

681 @name.setter

682 def name(self, value: str) -> None:

683 """Set the device name and update cached device_info."""

684 self._name = value

685 # Update existing cache object if it exists

686 if self._device_info_cache is not None:

687 self._device_info_cache.name = value

688

689 @property

690 def is_connected(self) -> bool:

691 """Check if the device is currently connected.

692

693 Delegates to device.connected.is_connected.

694

695 Returns:

696 True if connected, False otherwise

697

698 """

699 return self.connected.is_connected

700

701 @property

702 def last_advertisement(self) -> AdvertisementData | None:

703 """Get the last received advertisement data.

704

705 This is automatically updated when advertisements are processed via

706 refresh_advertisement() or parse_raw_advertisement().

707

708 Returns:

709 AdvertisementData with AD structures and interpreted_data if available,

710 None if no advertisement has been received yet.

711

712 """

713 return self._last_advertisement

714

715 def get_service_by_uuid(self, service_uuid: str) -> DeviceService | None:

716 """Get a service by its UUID.

717

718 Args:

719 service_uuid: UUID of the service

720

721 Returns:

722 DeviceService instance or None if not found

723

724 """

725 return self.services.get(service_uuid)

726

727 def get_services_by_name(self, service_name: str | ServiceName) -> list[DeviceService]:

728 """Get services by name.

729

730 Args:

731 service_name: Name or enum of the service

732

733 Returns:

734 List of matching DeviceService instances

735

736 """

737 service_uuid = self.translator.get_service_uuid_by_name(

738 service_name if isinstance(service_name, str) else service_name.value

739 )

740 if service_uuid and str(service_uuid) in self.services:

741 return [self.services[str(service_uuid)]]

742 return []

743

744 def list_characteristics(self, service_uuid: str | None = None) -> dict[str, list[str]]:

745 """List all characteristics, optionally filtered by service.

746

747 Args:

748 service_uuid: Optional service UUID to filter by

749

750 Returns:

751 Dictionary mapping service UUIDs to lists of characteristic UUIDs

752

753 """

754 if service_uuid:

755 service = self.services.get(service_uuid)

756 if service:

757 return {service_uuid: list(service.characteristics.keys())}

758 return {}

759

760 return {svc_uuid: list(service.characteristics.keys()) for svc_uuid, service in self.services.items()}