Coverage for src / bluetooth_sig / gatt / services / registry.py: 85%

1"""Bluetooth SIG GATT service registry. 

2 

3This module contains the service registry implementation, including the 

4ServiceName enum, service class mappings, and the GattServiceRegistry 

5class. This was moved from __init__.py to follow Python best practices 

6of keeping __init__.py files lightweight. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing_extensions import TypeGuard 

12 

13from ...registry.base import BaseUUIDClassRegistry 

14from ...types.gatt_enums import ServiceName 

15from ...types.gatt_services import ServiceDiscoveryData 

16from ...types.uuid import BluetoothUUID 

17from ..exceptions import UUIDResolutionError 

18from ..registry_utils import ModuleDiscovery, TypeValidator 

19from ..uuid_registry import uuid_registry 

20from .base import BaseGattService 

21 

22__all__ = [ 

23 "ServiceName", 

24 "get_service_class_map", 

25 "GattServiceRegistry", 

26] 

27 

28 

29def _is_service_subclass(candidate: object) -> TypeGuard[type[BaseGattService]]: 

30 """Type guard to check if candidate is a BaseGattService subclass. 

31 

32 Args: 

33 candidate: Object to check 

34 

35 Returns: 

36 True if candidate is a subclass of BaseGattService 

37 """ 

38 return TypeValidator.is_subclass_of(candidate, BaseGattService) 

39 

40 

41def get_service_class_map() -> dict[ServiceName, type[BaseGattService]]: 

42 """Get the current service class map. 

43 

44 Returns: 

45 Dictionary mapping ServiceName enum to service classes 

46 """ 

47 return GattServiceRegistry.get_instance()._get_enum_map() # pylint: disable=protected-access 

48 

49 

50class GattServiceRegistry(BaseUUIDClassRegistry[ServiceName, BaseGattService]): 

51 """Registry for all supported GATT services.""" 

52 

53 _MODULE_EXCLUSIONS = {"__main__", "__init__", "base", "registry"} 

54 

55 def _get_base_class(self) -> type[BaseGattService]: 

56 """Return the base class for service validation.""" 

57 return BaseGattService 

58 

59 def _discover_sig_classes(self) -> list[type[BaseGattService]]: 

60 """Discover all SIG-defined service classes in the package.""" 

61 package_name = __package__ or "bluetooth_sig.gatt.services" 

62 module_names = ModuleDiscovery.iter_module_names(package_name, self._MODULE_EXCLUSIONS) 

63 

64 return ModuleDiscovery.discover_classes( 

65 module_names, 

66 BaseGattService, 

67 _is_service_subclass, 

68 ) 

69 

70 def _build_enum_map(self) -> dict[ServiceName, type[BaseGattService]]: 

71 """Build the enum → class mapping using runtime discovery.""" 

72 mapping: dict[ServiceName, type[BaseGattService]] = {} 

73 

74 for service_cls in self._discover_sig_classes(): 

75 try: 

76 uuid_obj = service_cls.get_class_uuid() 

77 except (AttributeError, ValueError, UUIDResolutionError): 

78 # Skip classes that can't resolve a UUID (e.g., abstract base classes) 

79 continue 

80 

81 # Find the corresponding enum member by UUID 

82 enum_member = None 

83 for candidate_enum in ServiceName: 

84 candidate_info = uuid_registry.get_service_info(candidate_enum.value) 

85 if candidate_info and candidate_info.uuid == uuid_obj: 

86 enum_member = candidate_enum 

87 break 

88 

89 if enum_member is None: 

90 continue 

91 

92 existing = mapping.get(enum_member) 

93 if existing is not None and existing is not service_cls: 

94 raise RuntimeError( 

95 f"Multiple service classes resolved for {enum_member.name}: " 

96 f"{existing.__name__} and {service_cls.__name__}" 

97 ) 

98 mapping[enum_member] = service_cls 

99 

100 return mapping 

101 

102 def _load(self) -> None: 

103 """Perform the actual loading of registry data.""" 

104 _ = self._get_enum_map() 

105 _ = self._get_sig_classes_map() 

106 self._loaded = True 

107 

108 # Backward compatibility aliases 

109 

110 @classmethod 

111 def register_service_class( 

112 cls, uuid: str | BluetoothUUID | int, service_cls: type[BaseGattService], override: bool = False 

113 ) -> None: 

114 """Register a custom service class at runtime. 

115 

116 Args: 

117 uuid: The service UUID 

118 service_cls: The service class to register 

119 override: Whether to override existing registrations 

120 

121 Raises: 

122 TypeError: If service_cls does not inherit from BaseGattService 

123 ValueError: If UUID conflicts with existing registration and override=False 

124 """ 

125 instance = cls.get_instance() 

126 instance.register_class(uuid, service_cls, override) 

127 

128 @classmethod 

129 def unregister_service_class(cls, uuid: str | BluetoothUUID | int) -> None: 

130 """Unregister a custom service class. 

131 

132 Args: 

133 uuid: The service UUID to unregister 

134 """ 

135 instance = cls.get_instance() 

136 instance.unregister_class(uuid) 

137 

138 @classmethod 

139 def _get_services(cls) -> list[type[BaseGattService]]: 

140 """Get the list of service classes.""" 

141 instance = cls.get_instance() 

142 return list(instance._get_enum_map().values()) # pylint: disable=protected-access 

143 

144 @classmethod 

145 def get_service_class(cls, uuid: str | BluetoothUUID | int) -> type[BaseGattService] | None: 

146 """Get the service class for a given UUID. 

147 

148 Args: 

149 uuid: The service UUID 

150 

151 Returns: 

152 Service class if found, None otherwise 

153 

154 Raises: 

155 ValueError: If uuid format is invalid 

156 """ 

157 # Normalize to BluetoothUUID (let ValueError propagate) 

158 if isinstance(uuid, BluetoothUUID): 

159 bt_uuid = uuid 

160 else: 

161 bt_uuid = BluetoothUUID(uuid) 

162 

163 instance = cls.get_instance() 

164 service_cls = instance.get_class_by_uuid(bt_uuid) 

165 if service_cls: 

166 return service_cls 

167 

168 # Fallback: check if any service matches this UUID via matches_uuid() 

169 for service_class in cls._get_services(): 

170 if service_class.matches_uuid(bt_uuid): 

171 return service_class 

172 return None 

173 

174 @classmethod 

175 def get_service_class_by_name(cls, name: str | ServiceName) -> type[BaseGattService] | None: 

176 """Get the service class for a given name or enum.""" 

177 if isinstance(name, str): 

178 # For string names, find the matching enum member 

179 for enum_member in ServiceName: 

180 if enum_member.value == name: 

181 name = enum_member 

182 break 

183 else: 

184 return None # No matching enum found 

185 

186 instance = cls.get_instance() 

187 return instance.get_class_by_enum(name) 

188 

189 @classmethod 

190 def get_service_class_by_uuid(cls, uuid: str | BluetoothUUID | int) -> type[BaseGattService] | None: 

191 """Get the service class for a given UUID (alias for get_service_class).""" 

192 return cls.get_service_class(uuid) 

193 

194 @classmethod 

195 def create_service( 

196 cls, uuid: str | BluetoothUUID | int, characteristics: ServiceDiscoveryData 

197 ) -> BaseGattService | None: 

198 """Create a service instance for the given UUID and characteristics. 

199 

200 Args: 

201 uuid: Service UUID 

202 characteristics: Dict mapping characteristic UUIDs to CharacteristicInfo 

203 

204 Returns: 

205 Service instance if found, None otherwise 

206 

207 Raises: 

208 ValueError: If uuid format is invalid 

209 """ 

210 service_class = cls.get_service_class(uuid) 

211 if not service_class: 

212 return None 

213 

214 service = service_class() 

215 service.process_characteristics(characteristics) 

216 return service 

217 

218 @classmethod 

219 def get_all_services(cls) -> list[type[BaseGattService]]: 

220 """Get all registered service classes. 

221 

222 Returns: 

223 List of all registered service classes 

224 

225 """ 

226 return cls._get_services() 

227 

228 @classmethod 

229 def supported_services(cls) -> list[str]: 

230 """Get a list of supported service UUIDs.""" 

231 return [str(service().uuid) for service in cls._get_services()] 

232 

233 @classmethod 

234 def supported_service_names(cls) -> list[str]: 

235 """Get a list of supported service names.""" 

236 return [e.value for e in ServiceName] 

237 

238 @classmethod 

239 def clear_custom_registrations(cls) -> None: 

240 """Clear all custom service registrations (for testing).""" 

241 instance = cls.get_instance() 

242 for uuid in list(instance.list_custom_uuids()): 

243 instance.unregister_class(uuid)