Coverage for src / bluetooth_sig / types / device_types.py: 54%
76 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-18 11:17 +0000
1"""Device-related data types for BLE device management."""
3from __future__ import annotations
5from collections.abc import Awaitable, Callable
6from typing import Any, Literal, TypeAlias
8import msgspec
10from ..gatt.characteristics.base import BaseCharacteristic
11from ..gatt.services.base import BaseGattService
12from .advertising.result import AdvertisementData
13from .uuid import BluetoothUUID
15# Type alias for scanning mode
16ScanningMode = Literal["active", "passive"]
18# Type alias for scan detection callback: receives ScannedDevice as it's discovered
19ScanDetectionCallback: TypeAlias = Callable[["ScannedDevice"], Awaitable[None] | None]
21# Type alias for scan filter function: returns True if device matches
22ScanFilterFunc: TypeAlias = Callable[["ScannedDevice"], bool]
25class ScannedDevice(msgspec.Struct, kw_only=True):
26 """Minimal wrapper for a device discovered during BLE scanning.
28 Field descriptions:
29 address: Bluetooth MAC address or platform-specific identifier
30 name: OS-provided device name (may be None)
31 advertisement_data: Complete parsed advertising data (includes rssi, manufacturer_data, etc.)
33 """
35 address: str
36 name: str | None = None
37 advertisement_data: AdvertisementData | None = None
40class DeviceService(msgspec.Struct, kw_only=True):
41 r"""Represents a service on a device with its characteristics.
43 The characteristics dictionary stores BaseCharacteristic instances.
44 Access parsed data via characteristic.last_parsed property.
46 This provides a single source of truth: BaseCharacteristic[Any] instances
47 maintain their own last_parsed CharacteristicData.
49 Example::
51 # After discover_services() and read()
52 service = device.services["0000180f..."] # Battery Service
53 battery_char = service.characteristics["00002a19..."] # BatteryLevelCharacteristic instance
55 # Access last parsed result
56 if battery_char.last_parsed:
57 print(f"Battery: {battery_char.last_parsed.value}%")
59 # Or decode new data
60 parsed_value = battery_char.parse_value(raw_data)
61 """
63 service: BaseGattService
64 # Performance: str keys (UUID strings) for fast characteristic lookups by UUID
65 characteristics: dict[str, BaseCharacteristic[Any]] = msgspec.field(default_factory=dict)
68class DeviceEncryption(msgspec.Struct, kw_only=True):
69 """Encryption requirements and status for the device."""
71 requires_authentication: bool = False
72 requires_encryption: bool = False
73 encryption_level: str = ""
74 security_mode: int = 0
75 key_size: int = 0
78class ScanFilter(msgspec.Struct, kw_only=True):
79 """Filter configuration for BLE scanning operations.
81 All filters are optional. When multiple filters are specified, a device must
82 match ALL filters (AND logic) to be included in results.
84 For OR logic across different filter types, perform multiple scans or use
85 a custom `filter_func`.
87 Field descriptions:
88 service_uuids: Only include devices advertising these service UUIDs.
89 On some platforms (macOS), this filtering happens at the OS level
90 for better efficiency. UUIDs should be in standard string format
91 (e.g., "0000180f-0000-1000-8000-00805f9b34fb" or short "180f").
92 addresses: Only include devices with these MAC addresses.
93 Useful for reconnecting to known devices. Case-insensitive matching.
94 names: Only include devices with names containing any of these substrings.
95 Case-insensitive partial matching. Device must have a name to match.
96 rssi_threshold: Only include devices with RSSI >= this value (in dBm).
97 Typical values: -30 (very close), -60 (nearby), -90 (far).
98 filter_func: Custom filter function for complex matching logic.
99 Called with each ScannedDevice, return True to include.
100 Runs after all other filters.
102 Example::
104 # Find Heart Rate monitors nearby
105 filters = ScanFilter(
106 service_uuids=["180d"], # Heart Rate Service
107 rssi_threshold=-70,
108 )
110 # Find specific known devices
111 filters = ScanFilter(
112 addresses=["AA:BB:CC:DD:EE:FF", "11:22:33:44:55:66"],
113 )
116 # Custom filter for manufacturer-specific criteria
117 def my_filter(device: ScannedDevice) -> bool:
118 if device.advertisement_data is None:
119 return False
120 # Check for specific manufacturer ID
121 mfr_data = device.advertisement_data.ad_structures.core.manufacturer_data
122 return 0x004C in mfr_data # Apple devices
125 filters = ScanFilter(filter_func=my_filter)
127 """
129 service_uuids: list[str] | None = None
130 addresses: list[str] | None = None
131 names: list[str] | None = None
132 rssi_threshold: int | None = None
133 filter_func: ScanFilterFunc | None = None
135 def _passes_address_filter(self, device: ScannedDevice) -> bool:
136 """Check if device passes the address filter (No filter means passes filter)."""
137 addresses = self.addresses
138 if addresses is None:
139 return True
140 return any(addr.upper() == device.address.upper() for addr in addresses) # pylint: disable=not-an-iterable
142 def _passes_name_filter(self, device: ScannedDevice) -> bool:
143 """Check if device passes the name filter (No filter means passes filter)."""
144 names = self.names
145 if names is None:
146 return True
147 if device.name is None:
148 return False
149 device_name_lower = device.name.lower()
150 return any(name.lower() in device_name_lower for name in names) # pylint: disable=not-an-iterable
152 def _passes_rssi_filter(self, device: ScannedDevice) -> bool:
153 """Check if device passes the RSSI threshold (No filter means passes filter)."""
154 if self.rssi_threshold is None:
155 return True
156 if device.advertisement_data is None or device.advertisement_data.rssi is None:
157 return False
158 return device.advertisement_data.rssi >= self.rssi_threshold
160 def _passes_service_uuid_filter(self, device: ScannedDevice) -> bool:
161 """Check if device passes the service UUID filter (No filter means passes filter)."""
162 service_uuids = self.service_uuids
163 if service_uuids is None:
164 return True
165 if device.advertisement_data is None:
166 return False
167 advertised = device.advertisement_data.ad_structures.core.service_uuids
168 # Normalise UUIDs for comparison using BluetoothUUID
169 normalized_filters = {BluetoothUUID(uuid).normalized for uuid in service_uuids} # pylint: disable=not-an-iterable
170 normalized_advertised = {uuid.normalized for uuid in advertised}
171 return bool(normalized_filters & normalized_advertised)
173 def matches(self, device: ScannedDevice) -> bool:
174 """Check if a device passes all specified filters.
176 Args:
177 device: The scanned device to check
179 Returns:
180 True if device passes all filters, False otherwise
182 """
183 # Check each filter criterion
184 if not self._passes_address_filter(device):
185 return False
186 if not self._passes_name_filter(device):
187 return False
188 if not self._passes_rssi_filter(device):
189 return False
190 if not self._passes_service_uuid_filter(device):
191 return False
193 # Check custom filter function
194 filter_func = self.filter_func
195 return not (filter_func is not None and not filter_func(device)) # pylint: disable=not-callable
198class ScanOptions(msgspec.Struct, kw_only=True):
199 """Configuration options for scanning operations.
201 Combines filter criteria with scan behavior settings.
203 Field descriptions:
204 timeout: Maximum scan duration in seconds. None for indefinite (use with callbacks).
205 filters: Filter criteria for discovered devices. None for no filtering.
206 scanning_mode: 'active' sends scan requests for additional data (default),
207 'passive' only listens to advertisements (saves power, not supported on macOS).
208 return_first_match: If True with filters, stop scanning on first matching device.
209 Useful for targeted device discovery (faster than full scan + filter).
210 adapter: Backend-specific adapter identifier (e.g., "hci0" for BlueZ on Linux).
211 None uses the default adapter.
213 Example::
215 # Quick scan for a specific device type
216 options = ScanOptions(
217 timeout=10.0,
218 filters=ScanFilter(service_uuids=["180d"]),
219 return_first_match=True,
220 )
222 # Passive background scan
223 options = ScanOptions(
224 timeout=30.0,
225 scanning_mode="passive",
226 )
228 """
230 timeout: float | None = 5.0
231 filters: ScanFilter | None = None
232 scanning_mode: ScanningMode = "active"
233 return_first_match: bool = False
234 adapter: str | None = None