Coverage for src / bluetooth_sig / registry / base.py: 85%
221 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 20:14 +0000
1"""Base registry class for Bluetooth SIG registries with UUID support."""
3from __future__ import annotations
5import threading
6from abc import ABC, abstractmethod
7from enum import Enum
8from pathlib import Path
9from typing import Any, Callable, Generic, TypeVar
11from bluetooth_sig.registry.utils import (
12 find_bluetooth_sig_path,
13 load_yaml_uuids,
14 normalize_uuid_string,
15)
16from bluetooth_sig.types.registry import BaseUuidInfo, generate_basic_aliases
17from bluetooth_sig.types.uuid import BluetoothUUID
19T = TypeVar("T")
20E = TypeVar("E", bound=Enum) # For enum-keyed registries
21C = TypeVar("C") # For class types
24class RegistryMixin:
25 """Mixin providing common registry patterns for singleton, thread safety, and lazy loading.
27 This mixin contains shared functionality used by both info-based and class-based registries.
28 """
30 _lock: threading.RLock
31 _loaded: bool
33 def _lazy_load(self, loaded_check: Callable[[], bool], loader: Callable[[], None]) -> bool:
34 """Thread-safe lazy loading helper using double-checked locking pattern.
36 Args:
37 loaded_check: Callable that returns True if data is already loaded
38 loader: Callable that performs the actual loading
40 Returns:
41 True if loading was performed, False if already loaded
42 """
43 if loaded_check():
44 return False
46 with self._lock:
47 # Double-check after acquiring lock for thread safety
48 if loaded_check():
49 return False
51 loader()
52 return True
54 def _ensure_loaded(self) -> None:
55 """Ensure the registry is loaded (thread-safe lazy loading).
57 This is a standard implementation that subclasses can use.
58 It calls _lazy_load with self._loaded check and self._load as the loader.
59 Subclasses that need custom behaviour can override this method.
60 """
61 self._lazy_load(lambda: self._loaded, self._load)
63 @abstractmethod
64 def _load(self) -> None:
65 """Perform the actual loading of registry data."""
68class BaseGenericRegistry(RegistryMixin, ABC, Generic[T]):
69 """Base class for generic Bluetooth SIG registries with singleton pattern and thread safety.
71 For registries that are not UUID-based.
72 """
74 _instance: BaseGenericRegistry[T] | None = None
75 _lock = threading.RLock()
77 def __init__(self) -> None:
78 """Initialize the registry."""
79 self._lock = threading.RLock()
80 self._loaded: bool = False
82 @classmethod
83 def get_instance(cls) -> BaseGenericRegistry[T]:
84 """Get the singleton instance of the registry."""
85 if cls._instance is None:
86 with cls._lock:
87 if cls._instance is None:
88 cls._instance = cls()
89 return cls._instance
92U = TypeVar("U", bound=BaseUuidInfo)
95class BaseUUIDRegistry(RegistryMixin, ABC, Generic[U]):
96 """Base class for Bluetooth SIG registries with singleton pattern, thread safety, and UUID support.
98 Provides canonical storage, alias indices, and extensible hooks for UUID-based registries.
100 Subclasses should:
101 1. Call super().__init__() in their __init__ (base class sets self._loaded = False)
102 2. Implement _load() to perform actual data loading (must set self._loaded = True when done)
103 3. Optionally override _load_yaml_path() to return the YAML file path relative to bluetooth_sig/
104 4. Optionally override _generate_aliases(info) for domain-specific alias heuristics
105 5. Optionally override _post_store(info) for enrichment (e.g., unit mappings)
106 6. Call _ensure_loaded() before accessing data (provided by base class)
107 """
109 _instance: BaseUUIDRegistry[U] | None = None
110 _lock = threading.RLock()
112 def __init__(self) -> None:
113 """Initialize the registry."""
114 self._lock = threading.RLock()
115 self._loaded: bool = False # Initialized in base class, accessed by subclasses
116 self._canonical_store: dict[str, U] = {} # normalized_uuid -> info
117 self._alias_index: dict[str, str] = {} # lowercased_alias -> normalized_uuid
118 self._runtime_overrides: dict[str, U] = {} # normalized_uuid -> original SIG info
120 @abstractmethod
121 def _load_yaml_path(self) -> str:
122 """Return the YAML file path relative to bluetooth_sig/ root."""
124 def _generate_aliases(self, info: U) -> set[str]:
125 """Generate alias keys for an info entry.
127 Default implementation uses conservative basic aliases.
128 Subclasses can override for domain-specific heuristics.
129 """
130 return generate_basic_aliases(info)
132 def _post_store(self, info: U) -> None:
133 """Perform post-store enrichment for an info entry.
135 Default implementation does nothing.
136 Subclasses can override for enrichment (e.g., cross-references).
137 """
139 def _store_info(self, info: U) -> None:
140 """Store info with canonical key and generate aliases."""
141 canonical_key = info.uuid.normalized
143 # Store in canonical location
144 self._canonical_store[canonical_key] = info
146 # Generate and store aliases
147 aliases = self._generate_aliases(info)
148 for alias in aliases:
149 self._alias_index[alias.lower()] = canonical_key
151 # Perform any post-store enrichment
152 self._post_store(info)
154 def _load_from_yaml(self, yaml_path: Path) -> None:
155 """Load UUIDs from YAML file and store them."""
156 for uuid_data in load_yaml_uuids(yaml_path):
157 uuid_str = normalize_uuid_string(uuid_data["uuid"])
158 bt_uuid = BluetoothUUID(uuid_str)
160 # Create info with available fields, defaults for missing
161 info = self._create_info_from_yaml(uuid_data, bt_uuid)
162 self._store_info(info)
164 def _load(self) -> None:
165 """Perform the actual loading of registry data from YAML.
167 Default implementation loads from the path returned by _load_yaml_path().
168 Subclasses can override for custom loading behaviour.
169 """
170 base_path = find_bluetooth_sig_path()
171 if base_path:
172 yaml_path = base_path / self._load_yaml_path()
173 if yaml_path.exists():
174 self._load_from_yaml(yaml_path)
175 self._loaded = True
177 @abstractmethod
178 def _create_info_from_yaml(self, uuid_data: dict[str, Any], uuid: BluetoothUUID) -> U:
179 """Create info instance from YAML data.
181 Subclasses must implement to create the appropriate info type.
182 """
184 def get_info(self, identifier: str | BluetoothUUID) -> U | None:
185 """Get info by UUID, name, ID, or alias.
187 Args:
188 identifier: UUID string/int/BluetoothUUID, or name/ID/alias
190 Returns:
191 Info if found, None otherwise
192 """
193 self._ensure_loaded()
194 with self._lock:
195 # Handle BluetoothUUID directly
196 if isinstance(identifier, BluetoothUUID):
197 canonical_key = identifier.full_form
198 return self._canonical_store.get(canonical_key)
200 # Normalize string identifier
201 search_key = str(identifier).strip()
203 # Try UUID normalization first
204 try:
205 bt_uuid = BluetoothUUID(search_key)
206 canonical_key = bt_uuid.full_form
207 if canonical_key in self._canonical_store:
208 return self._canonical_store[canonical_key]
209 except ValueError:
210 pass
212 # Check alias index (normalized to lowercase)
213 alias_key = self._alias_index.get(search_key.lower())
214 if alias_key and alias_key in self._canonical_store:
215 return self._canonical_store[alias_key]
217 return None
219 def register_runtime_entry(self, entry: object) -> None:
220 """Register a runtime UUID entry, preserving original SIG info if overridden.
222 Args:
223 entry: Custom entry with uuid, name, id, etc.
224 """
225 self._ensure_loaded()
226 with self._lock:
227 bt_uuid = getattr(entry, "uuid", None)
228 if bt_uuid is None:
229 raise ValueError("Entry must have a uuid attribute")
230 bt_uuid = bt_uuid if isinstance(bt_uuid, BluetoothUUID) else BluetoothUUID(bt_uuid)
231 canonical_key = bt_uuid.normalized
233 # Preserve original SIG info if we're overriding it
234 # Assume entries in canonical_store that aren't in runtime_overrides are SIG entries
235 if canonical_key in self._canonical_store and canonical_key not in self._runtime_overrides:
236 self._runtime_overrides[canonical_key] = self._canonical_store[canonical_key]
238 # Create runtime info
239 info = self._create_runtime_info(entry, bt_uuid)
240 self._store_info(info)
242 @abstractmethod
243 def _create_runtime_info(self, entry: object, uuid: BluetoothUUID) -> U:
244 """Create runtime info from entry."""
246 def remove_runtime_override(self, normalized_uuid: str) -> None:
247 """Remove runtime override, restoring original SIG info if available.
249 Args:
250 normalized_uuid: Normalized UUID string
251 """
252 self._ensure_loaded()
253 with self._lock:
254 # Restore original SIG info if we have it
255 if normalized_uuid in self._runtime_overrides:
256 original_info = self._runtime_overrides.pop(normalized_uuid)
257 self._store_info(original_info)
258 elif normalized_uuid in self._canonical_store:
259 # Remove runtime entry entirely
260 # NOTE: Runtime tracking is registry-specific (e.g., uuid_registry uses _runtime_uuids set)
261 del self._canonical_store[normalized_uuid]
262 # Remove associated aliases
263 aliases_to_remove = [alias for alias, key in self._alias_index.items() if key == normalized_uuid]
264 for alias in aliases_to_remove:
265 del self._alias_index[alias]
267 def list_registered(self) -> list[str]:
268 """List all registered normalized UUIDs."""
269 self._ensure_loaded()
270 with self._lock:
271 return list(self._canonical_store.keys())
273 def list_aliases(self, uuid: BluetoothUUID) -> list[str]:
274 """List all aliases for a normalized UUID."""
275 self._ensure_loaded()
276 with self._lock:
277 return [alias for alias, key in self._alias_index.items() if key == uuid.normalized]
279 @classmethod
280 def get_instance(cls) -> BaseUUIDRegistry[U]:
281 """Get the singleton instance of the registry."""
282 if cls._instance is None:
283 with cls._lock:
284 if cls._instance is None:
285 cls._instance = cls()
286 return cls._instance
289class BaseUUIDClassRegistry(RegistryMixin, ABC, Generic[E, C]):
290 """Base class for UUID-based registries that store classes with enum-keyed access.
292 This registry type is designed for GATT characteristics and services that need:
293 1. UUID → Class mapping (e.g., "2A19" → BatteryLevelCharacteristic class)
294 2. Enum → Class mapping (e.g., CharacteristicName.BATTERY_LEVEL → BatteryLevelCharacteristic class)
295 3. Runtime class registration with override protection
296 4. Thread-safe singleton pattern with lazy loading
298 Unlike BaseUUIDRegistry which stores info objects (metadata), this stores actual classes
299 that can be instantiated.
301 Subclasses should:
302 1. Call super().__init__() in their __init__
303 2. Implement _load() to perform class discovery (must set self._loaded = True)
304 3. Implement _build_enum_map() to create the enum → class mapping
305 4. Implement _discover_sig_classes() to find built-in SIG classes
306 5. Optionally override _allows_sig_override() for custom override rules
307 """
309 _instance: BaseUUIDClassRegistry[E, C] | None = None
310 _lock = threading.RLock()
312 def __init__(self) -> None:
313 """Initialize the class registry."""
314 self._lock = threading.RLock()
315 self._loaded: bool = False
316 self._custom_classes: dict[BluetoothUUID, type[C]] = {}
317 self._sig_class_cache: dict[BluetoothUUID, type[C]] | None = None
318 self._enum_map_cache: dict[E, type[C]] | None = None
320 @abstractmethod
321 def _get_base_class(self) -> type[C]:
322 """Return the base class that all registered classes must inherit from.
324 This is used for validation when registering custom classes.
326 Returns:
327 The base class type
328 """
330 @abstractmethod
331 def _build_enum_map(self) -> dict[E, type[C]]:
332 """Build the enum → class mapping.
334 This should discover all SIG-defined classes and map them to their enum values.
336 Returns:
337 Dictionary mapping enum members to class types
338 """
340 @abstractmethod
341 def _discover_sig_classes(self) -> list[type[C]]:
342 """Discover all SIG-defined classes in the package.
344 Returns:
345 List of discovered class types
346 """
348 def _allows_sig_override(self, custom_cls: type[C], sig_cls: type[C]) -> bool:
349 """Check if a custom class is allowed to override a SIG class.
351 Default implementation checks for _allows_sig_override attribute on the custom class.
352 Subclasses can override for custom logic.
354 Args:
355 custom_cls: The custom class attempting to override
356 sig_cls: The existing SIG class being overridden (unused in default implementation)
358 Returns:
359 True if override is allowed, False otherwise
360 """
361 del sig_cls # Unused in default implementation
362 return getattr(custom_cls, "_allows_sig_override", False)
364 def _get_enum_map(self) -> dict[E, type[C]]:
365 """Get the cached enum → class mapping, building if necessary."""
366 if self._enum_map_cache is None:
367 self._enum_map_cache = self._build_enum_map()
368 return self._enum_map_cache
370 def _get_sig_classes_map(self) -> dict[BluetoothUUID, type[C]]:
371 """Get the cached UUID → SIG class mapping, building if necessary."""
372 if self._sig_class_cache is None:
373 self._sig_class_cache = {}
374 for cls in self._discover_sig_classes():
375 try:
376 uuid_obj = cls.get_class_uuid() # type: ignore[attr-defined]
377 if uuid_obj:
378 self._sig_class_cache[uuid_obj] = cls
379 except (AttributeError, ValueError):
380 # Skip classes that can't resolve UUID
381 continue
382 return self._sig_class_cache
384 def register_class(self, uuid: str | BluetoothUUID | int, cls: type[C], override: bool = False) -> None:
385 """Register a custom class at runtime.
387 Args:
388 uuid: The UUID for this class (string, BluetoothUUID, or int)
389 cls: The class to register
390 override: Whether to override existing registrations
392 Raises:
393 TypeError: If cls is not the correct type
394 ValueError: If UUID conflicts with existing registration and override=False,
395 or if attempting to override SIG class without permission
396 """
397 self._ensure_loaded()
399 # Validate that cls is actually a subclass of the base class
400 base_class = self._get_base_class()
401 try:
402 if not issubclass(cls, base_class):
403 raise TypeError(f"Registered class must inherit from {base_class.__name__}, got {cls.__name__}")
404 except TypeError:
405 # issubclass raises TypeError if cls is not a class
406 raise TypeError(
407 f"Registered class must inherit from {base_class.__name__}, "
408 f"got non-class object of type {type(cls).__name__}"
409 ) from None
411 # Normalize to BluetoothUUID
412 if isinstance(uuid, BluetoothUUID):
413 bt_uuid = uuid
414 else:
415 bt_uuid = BluetoothUUID(uuid)
417 # Check for SIG class collision
418 sig_classes = self._get_sig_classes_map()
419 sig_cls = sig_classes.get(bt_uuid)
421 with self._lock:
422 # Check for custom class collision
423 if not override and bt_uuid in self._custom_classes:
424 raise ValueError(f"UUID {bt_uuid} already registered. Use override=True to replace.")
426 # If collides with SIG class, enforce override rules
427 if sig_cls is not None:
428 if not override:
429 raise ValueError(
430 f"UUID {bt_uuid} conflicts with existing SIG class {sig_cls.__name__}. "
431 "Use override=True to replace."
432 )
433 if not self._allows_sig_override(cls, sig_cls):
434 raise ValueError(
435 f"Override of SIG class {sig_cls.__name__} requires "
436 f"_allows_sig_override=True on {cls.__name__}."
437 )
439 self._custom_classes[bt_uuid] = cls
441 def unregister_class(self, uuid: str | BluetoothUUID | int) -> None:
442 """Unregister a custom class.
444 Args:
445 uuid: The UUID to unregister (string, BluetoothUUID, or int)
446 """
447 if isinstance(uuid, BluetoothUUID):
448 bt_uuid = uuid
449 else:
450 bt_uuid = BluetoothUUID(uuid)
451 with self._lock:
452 self._custom_classes.pop(bt_uuid, None)
454 def get_class_by_uuid(self, uuid: str | BluetoothUUID | int) -> type[C] | None:
455 """Get the class for a given UUID.
457 Checks custom classes first, then SIG classes.
459 Args:
460 uuid: The UUID to look up (string, BluetoothUUID, or int)
462 Returns:
463 The class if found, None otherwise
464 """
465 self._ensure_loaded()
467 # Normalize to BluetoothUUID
468 if isinstance(uuid, BluetoothUUID):
469 bt_uuid = uuid
470 else:
471 bt_uuid = BluetoothUUID(uuid)
473 # Check custom classes first
474 with self._lock:
475 if custom_cls := self._custom_classes.get(bt_uuid):
476 return custom_cls
478 # Check SIG classes
479 return self._get_sig_classes_map().get(bt_uuid)
481 def get_class_by_enum(self, enum_member: E) -> type[C] | None:
482 """Get the class for a given enum member.
484 Args:
485 enum_member: The enum member to look up
487 Returns:
488 The class if found, None otherwise
489 """
490 self._ensure_loaded()
491 return self._get_enum_map().get(enum_member)
493 def list_custom_uuids(self) -> list[BluetoothUUID]:
494 """List all custom registered UUIDs.
496 Returns:
497 List of UUIDs with custom class registrations
498 """
499 with self._lock:
500 return list(self._custom_classes.keys())
502 def clear_enum_map_cache(self) -> None:
503 """Clear the cached enum → class mapping.
505 Useful when classes are registered/unregistered at runtime.
506 """
507 self._enum_map_cache = None
508 self._sig_class_cache = None
510 @classmethod
511 def get_instance(cls) -> BaseUUIDClassRegistry[E, C]:
512 """Get the singleton instance of the registry."""
513 if cls._instance is None:
514 with cls._lock:
515 if cls._instance is None:
516 cls._instance = cls()
517 return cls._instance