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

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. Payload Interpretation (PayloadInterpreter): AD structures → typed results 

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

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

12 - State managed externally by caller; interpreter updates state directly 

13 - Errors are raised as exceptions (consistent with GATT characteristic parsing) 

14 

15Error Handling: 

16 Interpreters raise exceptions for error conditions instead of returning 

17 status codes. This is consistent with GATT characteristic parsing. 

18 

19 Exceptions: 

20 EncryptionRequiredError: Payload encrypted, no bindkey available 

21 DecryptionFailedError: Decryption failed (wrong key or corrupt data) 

22 ReplayDetectedError: Counter not increasing (potential replay attack) 

23 DuplicatePacketError: Same packet_id as previous 

24 AdvertisingParseError: General parse failure 

25 UnsupportedVersionError: Unknown protocol version 

26""" 

27 

28from __future__ import annotations 

29 

30from abc import ABC, abstractmethod 

31from enum import Enum 

32from typing import Generic, TypeVar 

33 

34import msgspec 

35 

36from bluetooth_sig.advertising.state import DeviceAdvertisingState 

37from bluetooth_sig.types.company import ManufacturerData 

38from bluetooth_sig.types.uuid import BluetoothUUID 

39 

40 

41class AdvertisingData(msgspec.Struct, kw_only=True, frozen=True): 

42 """Complete advertising data from a BLE advertisement packet. 

43 

44 Encapsulates all extracted AD structures from a BLE PDU. 

45 Interpreters access only the fields they need. 

46 

47 Attributes: 

48 manufacturer_data: Company ID → ManufacturerData mapping. 

49 Each entry contains resolved company info and payload bytes. 

50 service_data: Service UUID → payload bytes mapping. 

51 local_name: Device local name (may contain protocol info). 

52 rssi: Signal strength in dBm. 

53 timestamp: Advertisement timestamp (Unix epoch seconds). 

54 

55 """ 

56 

57 manufacturer_data: dict[int, ManufacturerData] = msgspec.field(default_factory=dict) 

58 service_data: dict[BluetoothUUID, bytes] = msgspec.field(default_factory=dict) 

59 local_name: str | None = None 

60 rssi: int | None = None 

61 timestamp: float | None = None 

62 

63 

64# Generic type variable for interpreter result types 

65T = TypeVar("T") 

66 

67 

68class DataSource(Enum): 

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

70 

71 MANUFACTURER = "manufacturer" 

72 SERVICE = "service" 

73 LOCAL_NAME = "local_name" 

74 

75 

76class InterpreterInfo(msgspec.Struct, kw_only=True, frozen=True): 

77 """Interpreter metadata for routing and identification. 

78 

79 Attributes: 

80 company_id: Bluetooth SIG company ID for manufacturer data routing. 

81 service_uuid: Service UUID for service data routing. 

82 name: Human-readable interpreter name. 

83 data_source: Primary data source for fast routing. 

84 

85 """ 

86 

87 company_id: int | None = None 

88 service_uuid: BluetoothUUID | None = None 

89 name: str = "" 

90 data_source: DataSource = DataSource.MANUFACTURER 

91 

92 

93class PayloadInterpreter(ABC, Generic[T]): 

94 """Base class for payload interpretation (service data + manufacturer data). 

95 

96 Interprets raw bytes from BLE advertisements into typed domain objects. 

97 State is managed externally by the caller - interpreter receives state 

98 and updates it directly. Errors are raised as exceptions. 

99 

100 Encryption Flow (following BTHome/Xiaomi patterns): 

101 1. Check if payload is encrypted (flag byte in payload header) 

102 2. If encrypted, check state.encryption.bindkey 

103 3. If no bindkey, raise EncryptionRequiredError 

104 4. Extract counter from payload, compare to state.encryption.encryption_counter 

105 5. If counter <= old counter, raise ReplayDetectedError 

106 6. Attempt decryption with AES-CCM 

107 7. If decryption fails, raise DecryptionFailedError 

108 8. Parse decrypted payload 

109 9. Update state.encryption.encryption_counter directly 

110 10. Return parsed data 

111 

112 Example:: 

113 

114 class BTHomeInterpreter(PayloadInterpreter[BTHomeData]): 

115 _info = InterpreterInfo( 

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

117 name="BTHome", 

118 data_source=DataSource.SERVICE, 

119 ) 

120 

121 @classmethod 

122 def supports(cls, advertising_data): 

123 return "0000fcd2-0000-1000-8000-00805f9b34fb" in advertising_data.service_data 

124 

125 def interpret(self, advertising_data, state): 

126 # Parse BTHome service data 

127 # Update state.encryption.encryption_counter if encrypted 

128 # Raise exceptions on error 

129 # Return BTHomeData on success 

130 ... 

131 

132 """ 

133 

134 _info: InterpreterInfo 

135 _is_base_class: bool = False 

136 

137 def __init__(self, mac_address: str) -> None: 

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

139 

140 Args: 

141 mac_address: BLE device address (used for encryption nonce construction). 

142 

143 """ 

144 self._mac_address = mac_address 

145 

146 @property 

147 def mac_address(self) -> str: 

148 """Device MAC address.""" 

149 return self._mac_address 

150 

151 @property 

152 def info(self) -> InterpreterInfo: 

153 """Interpreter metadata.""" 

154 return self._info 

155 

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

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

158 super().__init_subclass__(**kwargs) 

159 

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

161 return 

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

163 return 

164 

165 # Lazy import to avoid circular dependency at module load time 

166 from bluetooth_sig.advertising.registry import payload_interpreter_registry # noqa: PLC0415 

167 

168 payload_interpreter_registry.register(cls) 

169 

170 @classmethod 

171 @abstractmethod 

172 def supports(cls, advertising_data: AdvertisingData) -> bool: 

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

174 

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

176 based on company_id, service_uuid, or local_name pattern. 

177 

178 Args: 

179 advertising_data: Complete advertising data from BLE packet. 

180 

181 Returns: 

182 True if this interpreter can handle the advertisement. 

183 

184 """ 

185 

186 @abstractmethod 

187 def interpret( 

188 self, 

189 advertising_data: AdvertisingData, 

190 state: DeviceAdvertisingState, 

191 ) -> T: 

192 """Interpret payload bytes and return typed result. 

193 

194 Updates state directly (state is mutable). Raises exceptions for errors. 

195 

196 Args: 

197 advertising_data: Complete advertising data from BLE packet. 

198 state: Current device advertising state (caller-managed, mutable). 

199 

200 Returns: 

201 Parsed data of type T. 

202 

203 Raises: 

204 EncryptionRequiredError: Payload encrypted, no bindkey available. 

205 DecryptionFailedError: Decryption failed. 

206 ReplayDetectedError: Encryption counter not increasing. 

207 DuplicatePacketError: Same packet_id as previous. 

208 AdvertisingParseError: General parse failure. 

209 UnsupportedVersionError: Unknown protocol version. 

210 

211 """