Coverage for src / bluetooth_sig / registry / base.py: 84%
216 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-18 11:17 +0000
1"""Base registry class for Bluetooth SIG registries with UUID support."""
3from __future__ import annotations
5import threading
6from abc import ABC, abstractmethod
7from collections.abc import Callable
8from enum import Enum
9from pathlib import Path
10from typing import Any, Generic, TypeVar
12from bluetooth_sig.registry.utils import (
13 find_bluetooth_sig_path,
14 load_yaml_uuids,
15 normalize_uuid_string,
16)
17from bluetooth_sig.types.registry import BaseUuidInfo, generate_basic_aliases
18from bluetooth_sig.types.uuid import BluetoothUUID
20T = TypeVar("T")
21E = TypeVar("E", bound=Enum) # For enum-keyed registries
22C = TypeVar("C") # For class types
25class RegistryMixin:
26 """Mixin providing common registry patterns for singleton, thread safety, and lazy loading.
28 This mixin contains shared functionality used by both info-based and class-based registries.
29 """
31 _lock: threading.RLock
32 _loaded: bool
34 def _lazy_load(self, loaded_check: Callable[[], bool], loader: Callable[[], None]) -> bool:
35 """Thread-safe lazy loading helper using double-checked locking pattern.
37 Args:
38 loaded_check: Callable that returns True if data is already loaded
39 loader: Callable that performs the actual loading
41 Returns:
42 True if loading was performed, False if already loaded
43 """
44 if loaded_check():
45 return False
47 with self._lock:
48 # Double-check after acquiring lock for thread safety
49 if loaded_check():
50 return False
52 loader()
53 return True
55 def _ensure_loaded(self) -> None:
56 """Ensure the registry is loaded (thread-safe lazy loading).
58 This is a standard implementation that subclasses can use.
59 It calls _lazy_load with self._loaded check and self._load as the loader.
60 Subclasses that need custom behaviour can override this method.
61 """
62 self._lazy_load(lambda: self._loaded, self._load)
64 def ensure_loaded(self) -> None:
65 """Public API to eagerly load the registry.
67 Equivalent to :meth:`_ensure_loaded` but intended for consumers
68 that need to pre-warm registries (e.g. during application startup).
69 """
70 self._ensure_loaded()
72 @abstractmethod
73 def _load(self) -> None:
74 """Perform the actual loading of registry data."""
77class BaseGenericRegistry(RegistryMixin, ABC, Generic[T]):
78 """Base class for generic Bluetooth SIG registries with singleton pattern and thread safety.
80 For registries that are not UUID-based.
81 """
83 _instance: BaseGenericRegistry[T] | None = None
84 _lock = threading.RLock()
86 def __init__(self) -> None:
87 """Initialize the registry."""
88 self._lock = threading.RLock()
89 self._loaded: bool = False
91 @classmethod
92 def get_instance(cls) -> BaseGenericRegistry[T]:
93 """Get the singleton instance of the registry."""
94 if cls._instance is None:
95 with cls._lock:
96 if cls._instance is None:
97 cls._instance = cls()
98 return cls._instance
101U = TypeVar("U", bound=BaseUuidInfo)
104class BaseUUIDRegistry(RegistryMixin, ABC, Generic[U]):
105 """Base class for Bluetooth SIG registries with singleton pattern, thread safety, and UUID support.
107 Provides canonical storage, alias indices, and extensible hooks for UUID-based registries.
109 Subclasses should:
110 1. Call super().__init__() in their __init__ (base class sets self._loaded = False)
111 2. Implement _load() to perform actual data loading (must set self._loaded = True when done)
112 3. Optionally override _load_yaml_path() to return the YAML file path relative to bluetooth_sig/
113 4. Optionally override _generate_aliases(info) for domain-specific alias heuristics
114 5. Optionally override _post_store(info) for enrichment (e.g., unit mappings)
115 6. Call _ensure_loaded() before accessing data (provided by base class)
116 """
118 _instance: BaseUUIDRegistry[U] | None = None
119 _lock = threading.RLock()
121 def __init__(self) -> None:
122 """Initialize the registry."""
123 self._lock = threading.RLock()
124 self._loaded: bool = False # Initialized in base class, accessed by subclasses
125 # Performance: Use str keys instead of BluetoothUUID for O(1) dict lookups
126 # String hashing is faster and these dicts are accessed frequently
127 self._canonical_store: dict[str, U] = {} # normalized_uuid -> info
128 self._alias_index: dict[str, str] = {} # lowercased_alias -> normalized_uuid
129 self._runtime_overrides: dict[str, U] = {} # normalized_uuid -> original SIG info
131 @abstractmethod
132 def _load_yaml_path(self) -> str:
133 """Return the YAML file path relative to bluetooth_sig/ root."""
135 def _generate_aliases(self, info: U) -> set[str]:
136 """Generate alias keys for an info entry.
138 Default implementation uses conservative basic aliases.
139 Subclasses can override for domain-specific heuristics.
140 """
141 return generate_basic_aliases(info)
143 def _post_store(self, info: U) -> None:
144 """Perform post-store enrichment for an info entry.
146 Default implementation does nothing.
147 Subclasses can override for enrichment (e.g., cross-references).
148 """
150 def _store_info(self, info: U) -> None:
151 """Store info with canonical key and generate aliases."""
152 canonical_key = info.uuid.normalized
154 # Store in canonical location
155 self._canonical_store[canonical_key] = info
157 # Generate and store aliases
158 aliases = self._generate_aliases(info)
159 for alias in aliases:
160 self._alias_index[alias.lower()] = canonical_key
162 # Perform any post-store enrichment
163 self._post_store(info)
165 def _load_from_yaml(self, yaml_path: Path) -> None:
166 """Load UUIDs from YAML file and store them."""
167 for uuid_data in load_yaml_uuids(yaml_path):
168 uuid_str = normalize_uuid_string(uuid_data["uuid"])
169 bt_uuid = BluetoothUUID(uuid_str)
171 # Create info with available fields, defaults for missing
172 info = self._create_info_from_yaml(uuid_data, bt_uuid)
173 self._store_info(info)
175 def _load(self) -> None:
176 """Perform the actual loading of registry data from YAML.
178 Default implementation loads from the path returned by _load_yaml_path().
179 Subclasses can override for custom loading behaviour.
180 """
181 base_path = find_bluetooth_sig_path()
182 if base_path:
183 yaml_path = base_path / self._load_yaml_path()
184 if yaml_path.exists():
185 self._load_from_yaml(yaml_path)
186 self._loaded = True
188 @abstractmethod
189 def _create_info_from_yaml(self, uuid_data: dict[str, Any], uuid: BluetoothUUID) -> U:
190 """Create info instance from YAML data.
192 Subclasses must implement to create the appropriate info type.
193 """
195 def get_info(self, identifier: str | BluetoothUUID) -> U | None:
196 """Get info by UUID, name, ID, or alias.
198 Args:
199 identifier: UUID string/int/BluetoothUUID, or name/ID/alias
201 Returns:
202 Info if found, None otherwise
203 """
204 self._ensure_loaded()
205 with self._lock:
206 # Handle BluetoothUUID directly
207 if isinstance(identifier, BluetoothUUID):
208 canonical_key = identifier.full_form
209 return self._canonical_store.get(canonical_key)
211 # Normalize string identifier
212 search_key = str(identifier).strip()
214 # Try UUID normalization first
215 try:
216 bt_uuid = BluetoothUUID(search_key)
217 canonical_key = bt_uuid.full_form
218 if canonical_key in self._canonical_store:
219 return self._canonical_store[canonical_key]
220 except ValueError:
221 pass # UUID normalization failed, continue to alias lookup
222 # Check alias index (normalized to lowercase)
223 alias_key = self._alias_index.get(search_key.lower())
224 if alias_key and alias_key in self._canonical_store:
225 return self._canonical_store[alias_key]
227 return None
229 def register_runtime_entry(self, entry: object) -> None:
230 """Register a runtime UUID entry, preserving original SIG info if overridden.
232 Args:
233 entry: Custom entry with uuid, name, id, etc.
234 """
235 self._ensure_loaded()
236 with self._lock:
237 bt_uuid = getattr(entry, "uuid", None)
238 if bt_uuid is None:
239 raise ValueError("Entry must have a uuid attribute")
240 bt_uuid = bt_uuid if isinstance(bt_uuid, BluetoothUUID) else BluetoothUUID(bt_uuid)
241 canonical_key = bt_uuid.normalized
243 # Preserve original SIG info if we're overriding it
244 # Assume entries in canonical_store that aren't in runtime_overrides are SIG entries
245 if canonical_key in self._canonical_store and canonical_key not in self._runtime_overrides:
246 self._runtime_overrides[canonical_key] = self._canonical_store[canonical_key]
248 # Create runtime info
249 info = self._create_runtime_info(entry, bt_uuid)
250 self._store_info(info)
252 @abstractmethod
253 def _create_runtime_info(self, entry: object, uuid: BluetoothUUID) -> U:
254 """Create runtime info from entry."""
256 def remove_runtime_override(self, normalized_uuid: BluetoothUUID) -> None:
257 """Remove runtime override, restoring original SIG info if available.
259 Args:
260 normalized_uuid: UUID to remove override for
261 """
262 self._ensure_loaded()
263 with self._lock:
264 # Restore original SIG info if we have it
265 uuid_key = normalized_uuid.full_form
266 if uuid_key in self._runtime_overrides:
267 original_info = self._runtime_overrides.pop(uuid_key)
268 self._store_info(original_info)
269 elif uuid_key in self._canonical_store:
270 # Remove runtime entry entirely
271 # NOTE: Runtime tracking is registry-specific (e.g., uuid_registry uses _runtime_uuids set)
272 del self._canonical_store[uuid_key]
273 # Remove associated aliases
274 aliases_to_remove = [alias for alias, key in self._alias_index.items() if key == uuid_key]
275 for alias in aliases_to_remove:
276 del self._alias_index[alias]
278 def list_registered(self) -> list[str]:
279 """List all registered normalized UUIDs."""
280 self._ensure_loaded()
281 with self._lock:
282 return list(self._canonical_store.keys())
284 def list_aliases(self, uuid: BluetoothUUID) -> list[str]:
285 """List all aliases for a normalized UUID."""
286 self._ensure_loaded()
287 with self._lock:
288 return [alias for alias, key in self._alias_index.items() if key == uuid.normalized]
290 @classmethod
291 def get_instance(cls) -> BaseUUIDRegistry[U]:
292 """Get the singleton instance of the registry."""
293 if cls._instance is None:
294 with cls._lock:
295 if cls._instance is None:
296 cls._instance = cls()
297 return cls._instance
300class BaseUUIDClassRegistry(RegistryMixin, ABC, Generic[E, C]):
301 """Base class for UUID-based registries that store classes with enum-keyed access.
303 This registry type is designed for GATT characteristics and services that need:
304 1. UUID → Class mapping (e.g., "2A19" → BatteryLevelCharacteristic class)
305 2. Enum → Class mapping (e.g., CharacteristicName.BATTERY_LEVEL → BatteryLevelCharacteristic class)
306 3. Runtime class registration with override protection
307 4. Thread-safe singleton pattern with lazy loading
309 Unlike BaseUUIDRegistry which stores info objects (metadata), this stores actual classes
310 that can be instantiated.
312 Subclasses should:
313 1. Call super().__init__() in their __init__
314 2. Implement _load() to perform class discovery (must set self._loaded = True)
315 3. Implement _build_enum_map() to create the enum → class mapping
316 4. Implement _discover_sig_classes() to find built-in SIG classes
317 5. Optionally override _allows_sig_override() for custom override rules
318 """
320 _instance: BaseUUIDClassRegistry[E, C] | None = None
321 _lock = threading.RLock()
323 def __init__(self) -> None:
324 """Initialize the class registry."""
325 self._lock = threading.RLock()
326 self._loaded: bool = False
327 self._custom_classes: dict[BluetoothUUID, type[C]] = {}
328 self._sig_class_cache: dict[BluetoothUUID, type[C]] | None = None
329 self._enum_map_cache: dict[E, type[C]] | None = None
331 @abstractmethod
332 def _get_base_class(self) -> type[C]:
333 """Return the base class that all registered classes must inherit from.
335 This is used for validation when registering custom classes.
337 Returns:
338 The base class type
339 """
341 @abstractmethod
342 def _build_enum_map(self) -> dict[E, type[C]]:
343 """Build the enum → class mapping.
345 This should discover all SIG-defined classes and map them to their enum values.
347 Returns:
348 Dictionary mapping enum members to class types
349 """
351 @abstractmethod
352 def _discover_sig_classes(self) -> list[type[C]]:
353 """Discover all SIG-defined classes in the package.
355 Returns:
356 List of discovered class types
357 """
359 def _allows_sig_override(self, custom_cls: type[C], sig_cls: type[C]) -> bool:
360 """Check if a custom class is allowed to override a SIG class.
362 Default implementation checks for _allows_sig_override attribute on the custom class.
363 Subclasses can override for custom logic.
365 Args:
366 custom_cls: The custom class attempting to override
367 sig_cls: The existing SIG class being overridden (unused in default implementation)
369 Returns:
370 True if override is allowed, False otherwise
371 """
372 del sig_cls # Unused in default implementation
373 return getattr(custom_cls, "_allows_sig_override", False)
375 def _get_enum_map(self) -> dict[E, type[C]]:
376 """Get the cached enum → class mapping, building if necessary."""
377 if self._enum_map_cache is None:
378 self._enum_map_cache = self._build_enum_map()
379 return self._enum_map_cache
381 def _get_sig_classes_map(self) -> dict[BluetoothUUID, type[C]]:
382 """Get the cached UUID → SIG class mapping, building if necessary."""
383 if self._sig_class_cache is None:
384 self._sig_class_cache = {}
385 for cls in self._discover_sig_classes():
386 try:
387 uuid_obj = cls.get_class_uuid() # type: ignore[attr-defined] # Generic C bound lacks this method; runtime dispatch is correct
388 if uuid_obj:
389 self._sig_class_cache[uuid_obj] = cls
390 except (AttributeError, ValueError):
391 # Skip classes that can't resolve UUID
392 continue
393 return self._sig_class_cache
395 def register_class(self, uuid: str | BluetoothUUID | int, cls: type[C], override: bool = False) -> None:
396 """Register a custom class at runtime.
398 Args:
399 uuid: The UUID for this class (string, BluetoothUUID, or int)
400 cls: The class to register
401 override: Whether to override existing registrations
403 Raises:
404 TypeError: If cls is not the correct type
405 ValueError: If UUID conflicts with existing registration and override=False,
406 or if attempting to override SIG class without permission
407 """
408 self._ensure_loaded()
410 # Validate that cls is actually a subclass of the base class
411 base_class = self._get_base_class()
412 if not issubclass(cls, base_class):
413 raise TypeError(f"Registered class must inherit from {base_class.__name__}, got {cls.__name__}")
415 # Normalize to BluetoothUUID
416 bt_uuid = uuid if isinstance(uuid, BluetoothUUID) else BluetoothUUID(uuid)
418 # Check for SIG class collision
419 sig_classes = self._get_sig_classes_map()
420 sig_cls = sig_classes.get(bt_uuid)
422 with self._lock:
423 # Check for custom class collision
424 if not override and bt_uuid in self._custom_classes:
425 raise ValueError(f"UUID {bt_uuid} already registered. Use override=True to replace.")
427 # If collides with SIG class, enforce override rules
428 if sig_cls is not None:
429 if not override:
430 raise ValueError(
431 f"UUID {bt_uuid} conflicts with existing SIG class {sig_cls.__name__}. "
432 "Use override=True to replace."
433 )
434 if not self._allows_sig_override(cls, sig_cls):
435 raise ValueError(
436 f"Override of SIG class {sig_cls.__name__} requires "
437 f"_allows_sig_override=True on {cls.__name__}."
438 )
440 self._custom_classes[bt_uuid] = cls
442 def unregister_class(self, uuid: str | BluetoothUUID | int) -> None:
443 """Unregister a custom class.
445 Args:
446 uuid: The UUID to unregister (string, BluetoothUUID, or int)
447 """
448 bt_uuid = uuid if isinstance(uuid, BluetoothUUID) else BluetoothUUID(uuid)
449 with self._lock:
450 self._custom_classes.pop(bt_uuid, None)
452 def get_class_by_uuid(self, uuid: str | BluetoothUUID | int) -> type[C] | None:
453 """Get the class for a given UUID.
455 Checks custom classes first, then SIG classes.
457 Args:
458 uuid: The UUID to look up (string, BluetoothUUID, or int)
460 Returns:
461 The class if found, None otherwise
462 """
463 self._ensure_loaded()
465 # Normalize to BluetoothUUID
466 bt_uuid = uuid if isinstance(uuid, BluetoothUUID) else BluetoothUUID(uuid)
468 # Check custom classes first
469 with self._lock:
470 if custom_cls := self._custom_classes.get(bt_uuid):
471 return custom_cls
473 # Check SIG classes
474 return self._get_sig_classes_map().get(bt_uuid)
476 def get_class_by_enum(self, enum_member: E) -> type[C] | None:
477 """Get the class for a given enum member.
479 Args:
480 enum_member: The enum member to look up
482 Returns:
483 The class if found, None otherwise
484 """
485 self._ensure_loaded()
486 return self._get_enum_map().get(enum_member)
488 def list_custom_uuids(self) -> list[BluetoothUUID]:
489 """List all custom registered UUIDs.
491 Returns:
492 List of UUIDs with custom class registrations
493 """
494 with self._lock:
495 return list(self._custom_classes.keys())
497 def clear_enum_map_cache(self) -> None:
498 """Clear the cached enum → class mapping.
500 Useful when classes are registered/unregistered at runtime.
501 """
502 self._enum_map_cache = None
503 self._sig_class_cache = None
505 @classmethod
506 def get_instance(cls) -> BaseUUIDClassRegistry[E, C]:
507 """Get the singleton instance of the registry."""
508 if cls._instance is None:
509 with cls._lock:
510 if cls._instance is None:
511 cls._instance = cls()
512 return cls._instance