Coverage for src / bluetooth_sig / types / device_types.py: 49%

74 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-11 20:14 +0000

1"""Device-related data types for BLE device management.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, Literal 

6 

7import msgspec 

8 

9from .advertising import AdvertisementData 

10from .uuid import BluetoothUUID 

11 

12# Type alias for scanning mode 

13ScanningMode = Literal["active", "passive"] 

14 

15# Circular dependency: gatt.characteristics.base imports from types, 

16# and this module needs to reference those classes for type hints. 

17# Type aliases using Callable with union syntax (|) must be in TYPE_CHECKING 

18# because GenericAlias doesn't support | operator at runtime in Python 3.9. 

19# PEP 604 (X | Y syntax) is Python 3.10+ only. 

20if TYPE_CHECKING: 

21 from collections.abc import Awaitable, Callable 

22 

23 from typing_extensions import TypeAlias 

24 

25 from ..gatt.characteristics.base import BaseCharacteristic 

26 from ..gatt.services.base import BaseGattService 

27 

28 # Type alias for scan detection callback: receives ScannedDevice as it's discovered 

29 ScanDetectionCallback: TypeAlias = Callable[["ScannedDevice"], Awaitable[None] | None] 

30 

31 # Type alias for scan filter function: returns True if device matches 

32 ScanFilterFunc: TypeAlias = Callable[["ScannedDevice"], bool] 

33 

34 

35class ScannedDevice(msgspec.Struct, kw_only=True): 

36 """Minimal wrapper for a device discovered during BLE scanning. 

37 

38 Field descriptions: 

39 address: Bluetooth MAC address or platform-specific identifier 

40 name: OS-provided device name (may be None) 

41 advertisement_data: Complete parsed advertising data (includes rssi, manufacturer_data, etc.) 

42 

43 """ 

44 

45 address: str 

46 name: str | None = None 

47 advertisement_data: AdvertisementData | None = None 

48 

49 

50class DeviceService(msgspec.Struct, kw_only=True): 

51 r"""Represents a service on a device with its characteristics. 

52 

53 The characteristics dictionary stores BaseCharacteristic instances. 

54 Access parsed data via characteristic.last_parsed property. 

55 

56 This provides a single source of truth: BaseCharacteristic[Any] instances 

57 maintain their own last_parsed CharacteristicData. 

58 

59 Example:: 

60 

61 # After discover_services() and read() 

62 service = device.services["0000180f..."] # Battery Service 

63 battery_char = service.characteristics["00002a19..."] # BatteryLevelCharacteristic instance 

64 

65 # Access last parsed result 

66 if battery_char.last_parsed: 

67 print(f"Battery: {battery_char.last_parsed.value}%") 

68 

69 # Or decode new data 

70 parsed_value = battery_char.parse_value(raw_data) 

71 """ 

72 

73 service: BaseGattService 

74 characteristics: dict[str, BaseCharacteristic[Any]] = msgspec.field(default_factory=dict) 

75 

76 

77class DeviceEncryption(msgspec.Struct, kw_only=True): 

78 """Encryption requirements and status for the device.""" 

79 

80 requires_authentication: bool = False 

81 requires_encryption: bool = False 

82 encryption_level: str = "" 

83 security_mode: int = 0 

84 key_size: int = 0 

85 

86 

87class ScanFilter(msgspec.Struct, kw_only=True): 

88 """Filter configuration for BLE scanning operations. 

89 

90 All filters are optional. When multiple filters are specified, a device must 

91 match ALL filters (AND logic) to be included in results. 

92 

93 For OR logic across different filter types, perform multiple scans or use 

94 a custom `filter_func`. 

95 

96 Field descriptions: 

97 service_uuids: Only include devices advertising these service UUIDs. 

98 On some platforms (macOS), this filtering happens at the OS level 

99 for better efficiency. UUIDs should be in standard string format 

100 (e.g., "0000180f-0000-1000-8000-00805f9b34fb" or short "180f"). 

101 addresses: Only include devices with these MAC addresses. 

102 Useful for reconnecting to known devices. Case-insensitive matching. 

103 names: Only include devices with names containing any of these substrings. 

104 Case-insensitive partial matching. Device must have a name to match. 

105 rssi_threshold: Only include devices with RSSI >= this value (in dBm). 

106 Typical values: -30 (very close), -60 (nearby), -90 (far). 

107 filter_func: Custom filter function for complex matching logic. 

108 Called with each ScannedDevice, return True to include. 

109 Runs after all other filters. 

110 

111 Example:: 

112 

113 # Find Heart Rate monitors nearby 

114 filters = ScanFilter( 

115 service_uuids=["180d"], # Heart Rate Service 

116 rssi_threshold=-70, 

117 ) 

118 

119 # Find specific known devices 

120 filters = ScanFilter( 

121 addresses=["AA:BB:CC:DD:EE:FF", "11:22:33:44:55:66"], 

122 ) 

123 

124 

125 # Custom filter for manufacturer-specific criteria 

126 def my_filter(device: ScannedDevice) -> bool: 

127 if device.advertisement_data is None: 

128 return False 

129 # Check for specific manufacturer ID 

130 mfr_data = device.advertisement_data.ad_structures.core.manufacturer_data 

131 return 0x004C in mfr_data # Apple devices 

132 

133 

134 filters = ScanFilter(filter_func=my_filter) 

135 

136 """ 

137 

138 service_uuids: list[str] | None = None 

139 addresses: list[str] | None = None 

140 names: list[str] | None = None 

141 rssi_threshold: int | None = None 

142 filter_func: ScanFilterFunc | None = None 

143 

144 def _passes_address_filter(self, device: ScannedDevice) -> bool: 

145 """Check if device passes the address filter (No filter means passes filter).""" 

146 addresses = self.addresses 

147 if addresses is None: 

148 return True 

149 return any(addr.upper() == device.address.upper() for addr in addresses) # pylint: disable=not-an-iterable 

150 

151 def _passes_name_filter(self, device: ScannedDevice) -> bool: 

152 """Check if device passes the name filter (No filter means passes filter).""" 

153 names = self.names 

154 if names is None: 

155 return True 

156 if device.name is None: 

157 return False 

158 device_name_lower = device.name.lower() 

159 return any(name.lower() in device_name_lower for name in names) # pylint: disable=not-an-iterable 

160 

161 def _passes_rssi_filter(self, device: ScannedDevice) -> bool: 

162 """Check if device passes the RSSI threshold (No filter means passes filter).""" 

163 if self.rssi_threshold is None: 

164 return True 

165 if device.advertisement_data is None or device.advertisement_data.rssi is None: 

166 return False 

167 return device.advertisement_data.rssi >= self.rssi_threshold 

168 

169 def _passes_service_uuid_filter(self, device: ScannedDevice) -> bool: 

170 """Check if device passes the service UUID filter (No filter means passes filter).""" 

171 service_uuids = self.service_uuids 

172 if service_uuids is None: 

173 return True 

174 if device.advertisement_data is None: 

175 return False 

176 advertised = device.advertisement_data.ad_structures.core.service_uuids 

177 # Normalise UUIDs for comparison using BluetoothUUID 

178 normalized_filters = {BluetoothUUID(uuid).normalized for uuid in service_uuids} # pylint: disable=not-an-iterable 

179 normalized_advertised = {uuid.normalized for uuid in advertised} 

180 return bool(normalized_filters & normalized_advertised) 

181 

182 def matches(self, device: ScannedDevice) -> bool: 

183 """Check if a device passes all specified filters. 

184 

185 Args: 

186 device: The scanned device to check 

187 

188 Returns: 

189 True if device passes all filters, False otherwise 

190 

191 """ 

192 # Check each filter criterion 

193 if not self._passes_address_filter(device): 

194 return False 

195 if not self._passes_name_filter(device): 

196 return False 

197 if not self._passes_rssi_filter(device): 

198 return False 

199 if not self._passes_service_uuid_filter(device): 

200 return False 

201 

202 # Check custom filter function 

203 filter_func = self.filter_func 

204 if filter_func is not None: 

205 if not filter_func(device): # pylint: disable=not-callable 

206 return False 

207 

208 return True 

209 

210 

211class ScanOptions(msgspec.Struct, kw_only=True): 

212 """Configuration options for scanning operations. 

213 

214 Combines filter criteria with scan behavior settings. 

215 

216 Field descriptions: 

217 timeout: Maximum scan duration in seconds. None for indefinite (use with callbacks). 

218 filters: Filter criteria for discovered devices. None for no filtering. 

219 scanning_mode: 'active' sends scan requests for additional data (default), 

220 'passive' only listens to advertisements (saves power, not supported on macOS). 

221 return_first_match: If True with filters, stop scanning on first matching device. 

222 Useful for targeted device discovery (faster than full scan + filter). 

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

224 None uses the default adapter. 

225 

226 Example:: 

227 

228 # Quick scan for a specific device type 

229 options = ScanOptions( 

230 timeout=10.0, 

231 filters=ScanFilter(service_uuids=["180d"]), 

232 return_first_match=True, 

233 ) 

234 

235 # Passive background scan 

236 options = ScanOptions( 

237 timeout=30.0, 

238 scanning_mode="passive", 

239 ) 

240 

241 """ 

242 

243 timeout: float | None = 5.0 

244 filters: ScanFilter | None = None 

245 scanning_mode: ScanningMode = "active" 

246 return_first_match: bool = False 

247 adapter: str | None = None