Coverage for src / bluetooth_sig / device / dependency_resolver.py: 45%

1"""Dependency resolution for characteristic reads. 

2 

3Resolves required and optional dependencies before reading a characteristic, 

4building a ``CharacteristicContext`` will all resolved dependency values. 

5""" 

6 

7from __future__ import annotations 

8 

9import logging 

10from enum import Enum 

11from typing import Any 

12 

13from ..gatt.characteristics.base import BaseCharacteristic 

14from ..gatt.characteristics.registry import CharacteristicRegistry 

15from ..gatt.characteristics.unknown import UnknownCharacteristic 

16from ..gatt.context import CharacteristicContext, DeviceInfo 

17from ..types import CharacteristicInfo 

18from ..types.uuid import BluetoothUUID 

19from .client import ClientManagerProtocol 

20from .connected import DeviceConnected 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25class DependencyResolver: 

26 """Resolves characteristic dependencies by reading them from the device. 

27 

28 Encapsulates the logic for: 

29 - Discovering which dependencies a characteristic declares 

30 - Reading dependency values from the device (with caching) 

31 - Building a ``CharacteristicContext`` for the target characteristic 

32 

33 Uses ``DeviceConnected`` for characteristic instance caching and 

34 ``ClientManagerProtocol`` for BLE reads. 

35 """ 

36 

37 def __init__( 

38 self, 

39 connection_manager: ClientManagerProtocol, 

40 connected: DeviceConnected, 

41 ) -> None: 

42 """Initialise with connection manager and connected subsystem. 

43 

44 Args: 

45 connection_manager: Connection manager for BLE reads 

46 connected: Connected subsystem for characteristic cache 

47 

48 """ 

49 self._connection_manager = connection_manager 

50 self._connected = connected 

51 

52 async def resolve( 

53 self, 

54 char_class: type[BaseCharacteristic[Any]], 

55 resolution_mode: DependencyResolutionMode, 

56 device_info: DeviceInfo, 

57 ) -> CharacteristicContext: 

58 """Ensure all dependencies for a characteristic are resolved. 

59 

60 Automatically reads feature characteristics needed for validation 

61 of measurement characteristics. Feature characteristics are cached 

62 after first read. 

63 

64 Args: 

65 char_class: The characteristic class to resolve dependencies for 

66 resolution_mode: How to handle dependency resolution 

67 device_info: Current device info for context construction 

68 

69 Returns: 

70 CharacteristicContext with resolved dependencies 

71 

72 Raises: 

73 RuntimeError: If no connection manager is attached 

74 

75 """ 

76 optional_deps = getattr(char_class, "_optional_dependencies", []) 

77 required_deps = getattr(char_class, "_required_dependencies", []) 

78 

79 context_chars: dict[str, Any] = {} 

80 

81 for dep_class in required_deps + optional_deps: 

82 is_required = dep_class in required_deps 

83 

84 dep_uuid = dep_class.get_class_uuid() 

85 if not dep_uuid: 

86 if is_required: 

87 raise ValueError(f"Required dependency {dep_class.__name__} has no UUID") 

88 continue 

89 

90 dep_uuid_str = str(dep_uuid) 

91 

92 if resolution_mode == DependencyResolutionMode.SKIP_DEPENDENCIES: 

93 continue 

94 

95 # Check cache (unless force refresh) 

96 if resolution_mode != DependencyResolutionMode.FORCE_REFRESH: 

97 cached_char = self._connected.get_cached_characteristic(dep_uuid) 

98 if cached_char is not None and cached_char.last_parsed is not None: 

99 context_chars[dep_uuid_str] = cached_char.last_parsed 

100 continue 

101 

102 parsed_data = await self._resolve_single(dep_uuid, is_required, dep_class) 

103 if parsed_data is not None: 

104 context_chars[dep_uuid_str] = parsed_data 

105 

106 return CharacteristicContext( 

107 device_info=device_info, 

108 other_characteristics=context_chars, 

109 ) 

110 

111 async def _resolve_single( 

112 self, 

113 dep_uuid: BluetoothUUID, 

114 is_required: bool, 

115 dep_class: type[BaseCharacteristic[Any]], 

116 ) -> Any | None: # noqa: ANN401 # Dependency can be any characteristic type 

117 """Read and parse a single dependency characteristic. 

118 

119 Args: 

120 dep_uuid: UUID of the dependency characteristic 

121 is_required: Whether this is a required dependency 

122 dep_class: The dependency characteristic class 

123 

124 Returns: 

125 Parsed characteristic data, or None if optional and failed 

126 

127 Raises: 

128 ValueError: If required dependency fails to read 

129 

130 """ 

131 dep_uuid_str = str(dep_uuid) 

132 

133 try: 

134 raw_data = await self._connection_manager.read_gatt_char(dep_uuid) 

135 

136 char_instance = self._connected.get_cached_characteristic(dep_uuid) 

137 if char_instance is None: 

138 char_class_or_none = CharacteristicRegistry.get_characteristic_class_by_uuid(dep_uuid) 

139 if char_class_or_none: 

140 char_instance = char_class_or_none() 

141 else: 

142 char_info = CharacteristicInfo(uuid=dep_uuid, name=dep_uuid_str) 

143 char_instance = UnknownCharacteristic(info=char_info) 

144 

145 self._connected.cache_characteristic(dep_uuid, char_instance) 

146 

147 return char_instance.parse_value(raw_data) 

148 

149 except Exception as e: # pylint: disable=broad-exception-caught 

150 if is_required: 

151 raise ValueError( 

152 f"Failed to read required dependency {dep_class.__name__} ({dep_uuid_str}): {e}" 

153 ) from e 

154 logger.warning("Failed to read optional dependency %s: %s", dep_class.__name__, e) 

155 return None 

156 

157 

158class DependencyResolutionMode(Enum): 

159 """Mode for automatic dependency resolution during characteristic reads. 

160 

161 Attributes: 

162 NORMAL: Auto-resolve dependencies, use cache when available 

163 SKIP_DEPENDENCIES: Skip dependency resolution and validation 

164 FORCE_REFRESH: Re-read dependencies from device, ignoring cache 

165 

166 """ 

167 

168 NORMAL = "normal" 

169 SKIP_DEPENDENCIES = "skip_dependencies" 

170 FORCE_REFRESH = "force_refresh"