Coverage for src / bluetooth_sig / advertising / base.py: 94%

1"""Base classes for advertising data interpreters. 

2 

3Advertising data interpretation follows a two-layer architecture: 

4 

51. PDU Parsing (AdvertisingPDUParser): Raw bytes → AD structures 

6 - Extracts manufacturer_data, service_data, flags, local_name, etc. 

7 - Framework-agnostic, works with raw BLE PDU bytes 

8 

92. Data Interpretation (AdvertisingDataInterpreter): AD structures → typed results 

10 - Interprets vendor-specific protocols (BTHome, Xiaomi, RuuviTag, etc.) 

11 - Returns strongly-typed sensor data (temperature, humidity, etc.) 

12 - Maintains per-device state (encryption counters, packet IDs) 

13""" 

14 

15from __future__ import annotations 

16 

17from abc import ABC, abstractmethod 

18from enum import Enum 

19from typing import Any, Generic, TypeVar 

20 

21import msgspec 

22 

23from bluetooth_sig.types.uuid import BluetoothUUID 

24 

25# Generic type variable for interpreter result types 

26T = TypeVar("T") 

27 

28 

29class DataSource(Enum): 

30 """Primary data source for interpreter routing.""" 

31 

32 MANUFACTURER = "manufacturer" 

33 SERVICE = "service" 

34 LOCAL_NAME = "local_name" 

35 

36 

37class AdvertisingInterpreterInfo(msgspec.Struct, kw_only=True, frozen=True): 

38 """Interpreter metadata for routing and identification.""" 

39 

40 company_id: int | None = None 

41 service_uuid: BluetoothUUID | None = None 

42 name: str = "" 

43 data_source: DataSource = DataSource.MANUFACTURER 

44 

45 

46class AdvertisingDataInterpreter(ABC, Generic[T]): 

47 """Base class for vendor-specific advertising data interpretation. 

48 

49 Interprets manufacturer_data and service_data from BLE advertisements 

50 into strongly-typed domain objects (sensor readings, device state, etc.). 

51 

52 Workflow: 

53 

54 1. Registry routes advertisement to interpreter class via supports() 

55 2. Device creates/reuses interpreter instance per (mac_address, interpreter_type) 

56 3. interpreter.interpret() decodes payload, returns typed result 

57 4. Interpreter maintains internal state (counters, flags) across calls 

58 

59 Example: 

60 class BTHomeInterpreter(AdvertisingDataInterpreter[BTHomeData]): 

61 _info = AdvertisingInterpreterInfo( 

62 service_uuid=BluetoothUUID("0000fcd2-0000-1000-8000-00805f9b34fb"), 

63 name="BTHome", 

64 data_source=DataSource.SERVICE, 

65 ) 

66 

67 @classmethod 

68 def supports(cls, manufacturer_data, service_data, local_name): 

69 return "0000fcd2-0000-1000-8000-00805f9b34fb" in service_data 

70 

71 def interpret(self, manufacturer_data, service_data, local_name, rssi): 

72 # Parse BTHome service data and return BTHomeData 

73 ... 

74 """ 

75 

76 _info: AdvertisingInterpreterInfo 

77 _is_base_class: bool = False 

78 

79 def __init__(self, mac_address: str, bindkey: bytes | None = None) -> None: 

80 """Create interpreter instance for a specific device. 

81 

82 Args: 

83 mac_address: BLE device address (used for encryption nonce) 

84 bindkey: Optional encryption key for encrypted advertisements 

85 

86 """ 

87 self._mac_address = mac_address 

88 self._bindkey = bindkey 

89 self._state: dict[str, Any] = {} 

90 

91 @property 

92 def mac_address(self) -> str: 

93 """Device MAC address.""" 

94 return self._mac_address 

95 

96 @property 

97 def bindkey(self) -> bytes | None: 

98 """Encryption key for this device.""" 

99 return self._bindkey 

100 

101 @bindkey.setter 

102 def bindkey(self, value: bytes | None) -> None: 

103 """Set encryption key.""" 

104 self._bindkey = value 

105 

106 @property 

107 def state(self) -> dict[str, Any]: 

108 """Return the protocol-specific state dictionary.""" 

109 return self._state 

110 

111 @property 

112 def info(self) -> AdvertisingInterpreterInfo: 

113 """Interpreter metadata.""" 

114 return self._info 

115 

116 def __init_subclass__(cls, **kwargs: object) -> None: 

117 """Auto-register interpreter subclasses with the registry.""" 

118 super().__init_subclass__(**kwargs) 

119 

120 if getattr(cls, "_is_base_class", False): 

121 return 

122 if not hasattr(cls, "_info"): 

123 return 

124 

125 # pylint: disable-next=import-outside-toplevel,cyclic-import 

126 from bluetooth_sig.advertising.registry import advertising_interpreter_registry 

127 

128 advertising_interpreter_registry.register(cls) 

129 

130 @classmethod 

131 @abstractmethod 

132 def supports( 

133 cls, 

134 manufacturer_data: dict[int, bytes], 

135 service_data: dict[BluetoothUUID, bytes], 

136 local_name: str | None, 

137 ) -> bool: 

138 """Check if this interpreter handles the advertisement. 

139 

140 Called by registry for fast routing. Should be a quick check 

141 based on company_id, service_uuid, or local_name pattern. 

142 """ 

143 

144 @abstractmethod 

145 def interpret( 

146 self, 

147 manufacturer_data: dict[int, bytes], 

148 service_data: dict[BluetoothUUID, bytes], 

149 local_name: str | None, 

150 rssi: int, 

151 ) -> T: 

152 """Interpret advertisement data and return typed result. 

153 

154 Args: 

155 manufacturer_data: Company ID → payload bytes mapping 

156 service_data: Service UUID → payload bytes mapping 

157 local_name: Device local name (may contain protocol info) 

158 rssi: Signal strength in dBm 

159 

160 Returns: 

161 Typed result specific to this interpreter (e.g., SensorData) 

162 

163 """