Coverage for src / bluetooth_sig / registry / core / uri_schemes.py: 80%

1"""URI Schemes registry for Bluetooth SIG URI beacon parsing. 

2 

3Used during advertising data parsing to decode Eddystone URI beacons 

4and other URI-based beacon formats. 

5""" 

6 

7from __future__ import annotations 

8 

9import logging 

10 

11import msgspec 

12 

13from bluetooth_sig.registry.base import BaseGenericRegistry 

14from bluetooth_sig.registry.utils import find_bluetooth_sig_path 

15from bluetooth_sig.types.registry.uri_schemes import UriSchemeInfo 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20class UriSchemesRegistry(BaseGenericRegistry[UriSchemeInfo]): 

21 """Registry for Bluetooth URI schemes with lazy loading. 

22 

23 This registry loads URI scheme definitions from the official Bluetooth SIG 

24 assigned_numbers YAML file, enabling URI beacon decoding for Eddystone 

25 and similar beacon formats. 

26 

27 The value field is used as a compact encoding for URI prefixes in 

28 advertising data, reducing packet size for common schemes like http://. 

29 

30 Examples: 

31 >>> from bluetooth_sig.registry.core.uri_schemes import uri_schemes_registry 

32 >>> info = uri_schemes_registry.get_uri_scheme_info(0x16) 

33 >>> info.name 

34 'http:' 

35 """ 

36 

37 def __init__(self) -> None: 

38 """Initialize the URI schemes registry.""" 

39 super().__init__() 

40 self._uri_schemes: dict[int, UriSchemeInfo] = {} 

41 self._uri_schemes_by_name: dict[str, UriSchemeInfo] = {} 

42 

43 def _load(self) -> None: 

44 """Perform the actual loading of URI schemes data.""" 

45 base_path = find_bluetooth_sig_path() 

46 if not base_path: 

47 logger.warning("Bluetooth SIG path not found. URI schemes registry will be empty.") 

48 self._loaded = True 

49 return 

50 

51 yaml_path = base_path.parent / "core" / "uri_schemes.yaml" 

52 if not yaml_path.exists(): 

53 logger.warning( 

54 "URI schemes YAML file not found at %s. Registry will be empty.", 

55 yaml_path, 

56 ) 

57 self._loaded = True 

58 return 

59 

60 try: 

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

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

63 

64 if not data or "uri_schemes" not in data: 

65 logger.warning("Invalid URI schemes YAML format. Registry will be empty.") 

66 self._loaded = True 

67 return 

68 

69 for item in data["uri_schemes"]: 

70 value = item.get("value") 

71 name = item.get("name") 

72 

73 if value is None or not name: 

74 continue 

75 

76 # Handle hex values in YAML (e.g., 0x16) 

77 if isinstance(value, str): 

78 value = int(value, 16) 

79 

80 uri_scheme_info = UriSchemeInfo( 

81 value=value, 

82 name=name, 

83 ) 

84 

85 self._uri_schemes[value] = uri_scheme_info 

86 self._uri_schemes_by_name[name.lower()] = uri_scheme_info 

87 

88 logger.info("Loaded %d URI schemes from specification", len(self._uri_schemes)) 

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

90 logger.warning( 

91 "Failed to load URI schemes from YAML: %s. Registry will be empty.", 

92 e, 

93 ) 

94 

95 self._loaded = True 

96 

97 def get_uri_scheme_info(self, value: int) -> UriSchemeInfo | None: 

98 """Get URI scheme info by value (lazy loads on first call). 

99 

100 Args: 

101 value: The URI scheme value (e.g., 0x16 for "http:") 

102 

103 Returns: 

104 UriSchemeInfo object, or None if not found 

105 """ 

106 self._ensure_loaded() 

107 with self._lock: 

108 return self._uri_schemes.get(value) 

109 

110 def get_uri_scheme_by_name(self, name: str) -> UriSchemeInfo | None: 

111 """Get URI scheme info by name (lazy loads on first call). 

112 

113 Args: 

114 name: URI scheme name (case-insensitive, e.g., "http:", "https:") 

115 

116 Returns: 

117 UriSchemeInfo object, or None if not found 

118 """ 

119 self._ensure_loaded() 

120 with self._lock: 

121 return self._uri_schemes_by_name.get(name.lower()) 

122 

123 def is_known_uri_scheme(self, value: int) -> bool: 

124 """Check if URI scheme is known (lazy loads on first call). 

125 

126 Args: 

127 value: The URI scheme value to check 

128 

129 Returns: 

130 True if the URI scheme is registered, False otherwise 

131 """ 

132 self._ensure_loaded() 

133 with self._lock: 

134 return value in self._uri_schemes 

135 

136 def get_all_uri_schemes(self) -> dict[int, UriSchemeInfo]: 

137 """Get all registered URI schemes (lazy loads on first call). 

138 

139 Returns: 

140 Dictionary mapping URI scheme values to UriSchemeInfo objects 

141 """ 

142 self._ensure_loaded() 

143 with self._lock: 

144 return self._uri_schemes.copy() 

145 

146 def decode_uri_prefix(self, value: int) -> str: 

147 """Decode a URI scheme value to its string prefix. 

148 

149 Convenience method for beacon parsing that returns the scheme 

150 string directly, or an empty string if unknown. 

151 

152 Args: 

153 value: The URI scheme value from beacon data 

154 

155 Returns: 

156 The URI scheme string (e.g., "http:"), or empty string if unknown 

157 """ 

158 info = self.get_uri_scheme_info(value) 

159 return info.name if info else "" 

160 

161 

162# Global singleton instance 

163uri_schemes_registry = UriSchemesRegistry()