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