Coverage for src / bluetooth_sig / registry / core / ad_types.py: 79%

62 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-01-11 20:14 +0000

1"""AD Types registry for Bluetooth SIG advertising data type definitions.""" 

2 

3from __future__ import annotations 

4 

5import logging 

6 

7import msgspec 

8 

9from bluetooth_sig.registry.base import BaseGenericRegistry 

10from bluetooth_sig.registry.utils import find_bluetooth_sig_path 

11from bluetooth_sig.types.registry.ad_types import AdTypeInfo 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class ADTypesRegistry(BaseGenericRegistry[AdTypeInfo]): 

17 """Registry for Bluetooth advertising data types with lazy loading. 

18 

19 This registry loads AD type definitions from the official Bluetooth SIG 

20 assigned_numbers YAML file, providing authoritative AD type information 

21 from the specification. 

22 """ 

23 

24 def __init__(self) -> None: 

25 """Initialize the AD types registry.""" 

26 super().__init__() 

27 self._ad_types: dict[int, AdTypeInfo] = {} 

28 self._ad_types_by_name: dict[str, AdTypeInfo] = {} 

29 

30 def _load(self) -> None: 

31 """Perform the actual loading of AD types data.""" 

32 base_path = find_bluetooth_sig_path() 

33 if not base_path: 

34 logger.warning("Bluetooth SIG path not found. AD types registry will be empty.") 

35 self._loaded = True 

36 return 

37 

38 yaml_path = base_path.parent / "core" / "ad_types.yaml" 

39 if not yaml_path.exists(): 

40 logger.warning( 

41 "AD types YAML file not found at %s. Registry will be empty.", 

42 yaml_path, 

43 ) 

44 self._loaded = True 

45 return 

46 

47 try: 

48 with yaml_path.open("r", encoding="utf-8") as f: 

49 data = msgspec.yaml.decode(f.read()) 

50 

51 if not data or "ad_types" not in data: 

52 logger.warning("Invalid AD types YAML format. Registry will be empty.") 

53 self._loaded = True 

54 return 

55 

56 for item in data["ad_types"]: 

57 value = item.get("value") 

58 name = item.get("name") 

59 reference = item.get("reference") 

60 

61 if value is None or not name: 

62 continue 

63 

64 # Handle hex values in YAML (e.g., 0x01) 

65 if isinstance(value, str): 

66 value = int(value, 16) 

67 

68 ad_type_info = AdTypeInfo( 

69 value=value, 

70 name=name, 

71 reference=reference, 

72 ) 

73 

74 self._ad_types[value] = ad_type_info 

75 self._ad_types_by_name[name.lower()] = ad_type_info 

76 

77 logger.info("Loaded %d AD types from specification", len(self._ad_types)) 

78 except (FileNotFoundError, OSError, msgspec.DecodeError, KeyError) as e: 

79 logger.warning( 

80 "Failed to load AD types from YAML: %s. Registry will be empty.", 

81 e, 

82 ) 

83 

84 self._loaded = True 

85 

86 def get_ad_type_info(self, ad_type: int) -> AdTypeInfo | None: 

87 """Get AD type info by value (lazy loads on first call). 

88 

89 Args: 

90 ad_type: The AD type value (e.g., 0x01 for Flags) 

91 

92 Returns: 

93 AdTypeInfo object, or None if not found 

94 """ 

95 self._ensure_loaded() 

96 with self._lock: 

97 return self._ad_types.get(ad_type) 

98 

99 def get_ad_type_by_name(self, name: str) -> AdTypeInfo | None: 

100 """Get AD type info by name (lazy loads on first call). 

101 

102 Args: 

103 name: AD type name (case-insensitive) 

104 

105 Returns: 

106 AdTypeInfo object, or None if not found 

107 """ 

108 self._ensure_loaded() 

109 with self._lock: 

110 return self._ad_types_by_name.get(name.lower()) 

111 

112 def is_known_ad_type(self, ad_type: int) -> bool: 

113 """Check if AD type is known (lazy loads on first call). 

114 

115 Args: 

116 ad_type: The AD type value to check 

117 

118 Returns: 

119 True if the AD type is registered, False otherwise 

120 """ 

121 self._ensure_loaded() 

122 with self._lock: 

123 return ad_type in self._ad_types 

124 

125 def get_all_ad_types(self) -> dict[int, AdTypeInfo]: 

126 """Get all registered AD types (lazy loads on first call). 

127 

128 Returns: 

129 Dictionary mapping AD type values to AdTypeInfo objects 

130 """ 

131 self._ensure_loaded() 

132 with self._lock: 

133 return self._ad_types.copy() 

134 

135 

136# Global singleton instance 

137ad_types_registry = ADTypesRegistry()