Coverage for src/bluetooth_sig/gatt/descriptors/registry.py: 92%

1"""Descriptor registry and resolution.""" 

2 

3from __future__ import annotations 

4 

5import inspect 

6import logging 

7import threading 

8from importlib import import_module 

9from typing import ClassVar 

10 

11from ...types.uuid import BluetoothUUID 

12from ..registry_utils import ModuleDiscovery 

13from .base import BaseDescriptor 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18class DescriptorRegistry: 

19 """Registry for descriptor classes.""" 

20 

21 _registry: ClassVar[dict[str, type[BaseDescriptor]]] = {} 

22 _loaded: ClassVar[bool] = False 

23 _lock: ClassVar[threading.RLock] = threading.RLock() 

24 _MODULE_EXCLUSIONS: ClassVar[set[str]] = {"__init__", "__main__", "_export_map", "base", "registry"} 

25 

26 @classmethod 

27 def _ensure_loaded(cls) -> None: 

28 with cls._lock: 

29 if cls._loaded: 

30 return 

31 cls._register_builtins() 

32 cls._loaded = True 

33 

34 @classmethod 

35 def _register_builtins(cls) -> None: 

36 package_name = __package__ or "bluetooth_sig.gatt.descriptors" 

37 module_names = ModuleDiscovery.iter_module_names(package_name, cls._MODULE_EXCLUSIONS) 

38 for module_name in module_names: 

39 module = import_module(module_name) 

40 for _, obj in inspect.getmembers(module, inspect.isclass): 

41 if not isinstance(obj, type) or not issubclass(obj, BaseDescriptor): 

42 continue 

43 if obj is BaseDescriptor or getattr(obj, "_is_template", False): 

44 continue 

45 if obj.__module__ != module.__name__: 

46 continue 

47 cls.register(obj) 

48 

49 @classmethod 

50 def register(cls, descriptor_class: type[BaseDescriptor]) -> None: 

51 """Register a descriptor class.""" 

52 try: 

53 instance = descriptor_class() 

54 uuid_str = str(instance.uuid) 

55 cls._registry[uuid_str] = descriptor_class 

56 except (ValueError, TypeError, AttributeError): 

57 logger.warning("Failed to register descriptor class %s", descriptor_class.__name__) 

58 

59 @classmethod 

60 def get_descriptor_class(cls, uuid: str | BluetoothUUID | int) -> type[BaseDescriptor] | None: 

61 """Get descriptor class for UUID. 

62 

63 Args: 

64 uuid: The descriptor UUID 

65 

66 Returns: 

67 Descriptor class if found, None otherwise 

68 

69 Raises: 

70 ValueError: If uuid format is invalid 

71 """ 

72 cls._ensure_loaded() 

73 uuid_obj = BluetoothUUID(uuid) 

74 full_uuid_str = uuid_obj.dashed_form 

75 return cls._registry.get(full_uuid_str) 

76 

77 @classmethod 

78 def create_descriptor(cls, uuid: str | BluetoothUUID | int) -> BaseDescriptor | None: 

79 """Create descriptor instance for UUID. 

80 

81 Args: 

82 uuid: The descriptor UUID 

83 

84 Returns: 

85 Descriptor instance if found, None otherwise 

86 

87 Raises: 

88 ValueError: If uuid format is invalid 

89 """ 

90 descriptor_class = cls.get_descriptor_class(uuid) 

91 if descriptor_class: 

92 try: 

93 return descriptor_class() 

94 except (ValueError, TypeError, AttributeError): 

95 return None 

96 return None 

97 

98 @classmethod 

99 def list_registered_descriptors(cls) -> list[str]: 

100 """List all registered descriptor UUIDs.""" 

101 cls._ensure_loaded() 

102 return list(cls._registry.keys())