Coverage for openhcs/core/components/validation.py: 64.6%

1""" 

2Generic validation system for component-agnostic validation. 

3 

4This module provides a generic replacement for the component-specific validation 

5logic, supporting any component configuration and validation patterns. 

6""" 

7 

8import logging 

9from typing import Generic, TypeVar, List, Optional, Dict, Any, Union, Type 

10from enum import Enum 

11from dataclasses import dataclass 

12 

13from .framework import ComponentConfiguration 

14 

15logger = logging.getLogger(__name__) 

16 

17T = TypeVar('T', bound=Enum) 

18U = TypeVar('U', bound=Enum) 

19 

20 

21def convert_enum_by_value(source_enum: T, target_enum_class: Type[U]) -> Optional[U]: 

22 """ 

23 Generic utility to convert between enum types with matching .value attributes. 

24 

25 This function enables conversion between any two enum classes that have 

26 overlapping values, without requiring hardcoded mappings. 

27 

28 Args: 

29 source_enum: Source enum instance to convert from 

30 target_enum_class: Target enum class to convert to 

31 

32 Returns: 

33 Target enum instance with matching value, or None if no match found 

34 

35 Example: 

36 >>> convert_enum_by_value(VariableComponents.CHANNEL, GroupBy) 

37 <GroupBy.CHANNEL: 'channel'> 

38 """ 

39 source_value = source_enum.value 

40 

41 for target_enum in target_enum_class: 41 ↛ 45line 41 didn't jump to line 45 because the loop on line 41 didn't complete

42 if target_enum.value == source_value: 

43 return target_enum 

44 

45 return None 

46 

47 

48@dataclass 

49class ValidationResult: 

50 """Result of a validation operation.""" 

51 is_valid: bool 

52 error_message: Optional[str] = None 

53 warnings: Optional[List[str]] = None 

54 

55 

56class GenericValidator(Generic[T]): 

57 """ 

58 Generic validator for component-agnostic validation. 

59  

60 This class replaces the hardcoded component-specific validation logic 

61 with a configurable system that works with any component configuration. 

62 """ 

63 

64 def __init__(self, config: ComponentConfiguration[T]): 

65 """ 

66 Initialize the validator with a component configuration. 

67  

68 Args: 

69 config: ComponentConfiguration for validation rules 

70 """ 

71 self.config = config 

72 logger.debug(f"GenericValidator initialized for components: {[c.value for c in config.all_components]}") 

73 

74 def validate_step( 

75 self, 

76 variable_components: List[T], 

77 group_by: Optional[Union[T, 'GroupBy']], 

78 func_pattern: Any, 

79 step_name: str 

80 ) -> ValidationResult: 

81 """ 

82 Validate a step configuration using generic rules. 

83  

84 Args: 

85 variable_components: List of variable components 

86 group_by: Optional group_by component 

87 func_pattern: Function pattern (callable, dict, or list) 

88 step_name: Name of the step for error reporting 

89  

90 Returns: 

91 ValidationResult indicating success or failure 

92 """ 

93 try: 

94 # 1. Validate component combination 

95 self.config.validate_combination(variable_components, group_by) 

96 

97 # 2. Validate dict pattern requirements 

98 if isinstance(func_pattern, dict) and not group_by: 98 ↛ 99line 98 didn't jump to line 99 because the condition on line 98 was never true

99 return ValidationResult( 

100 is_valid=False, 

101 error_message=f"Dict pattern requires group_by in step '{step_name}'" 

102 ) 

103 

104 # 3. Validate components are in remaining components (not multiprocessing axis) 

105 remaining_components = self.config.get_remaining_components() 

106 remaining_values = {comp.value for comp in remaining_components} 

107 

108 for component in variable_components: 

109 if component.value not in remaining_values: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 return ValidationResult( 

111 is_valid=False, 

112 error_message=f"Variable component {component.value} not available (multiprocessing axis: {self.config.multiprocessing_axis.value})" 

113 ) 

114 

115 if group_by and group_by.value not in remaining_values: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 return ValidationResult( 

117 is_valid=False, 

118 error_message=f"Group_by component {group_by.value} not available (multiprocessing axis: {self.config.multiprocessing_axis.value})" 

119 ) 

120 

121 return ValidationResult(is_valid=True) 

122 

123 except ValueError as e: 

124 return ValidationResult( 

125 is_valid=False, 

126 error_message=str(e) 

127 ) 

128 

129 def validate_dict_pattern_keys( 

130 self, 

131 func_pattern: Dict[str, Any], 

132 group_by: T, 

133 step_name: str, 

134 orchestrator 

135 ) -> ValidationResult: 

136 """ 

137 Validate that dict function pattern keys match available component keys. 

138  

139 This validation ensures compile-time guarantee that dict patterns will work 

140 at runtime by checking that all dict keys exist in the actual component data. 

141  

142 Args: 

143 func_pattern: Dict function pattern to validate 

144 group_by: GroupBy component specifying component type 

145 step_name: Name of the step containing the function 

146 orchestrator: Orchestrator for component key access 

147  

148 Returns: 

149 ValidationResult indicating success or failure 

150 """ 

151 try: 

152 # Use enum objects directly - orchestrator now accepts VariableComponents 

153 available_keys = orchestrator.get_component_keys(group_by) 

154 available_keys_set = set(str(key) for key in available_keys) 

155 

156 # Check each dict key against available keys 

157 pattern_keys = list(func_pattern.keys()) 

158 pattern_keys_set = set(str(key) for key in pattern_keys) 

159 

160 # Try direct string match first 

161 missing_keys = pattern_keys_set - available_keys_set 

162 

163 if missing_keys: 163 ↛ 165line 163 didn't jump to line 165 because the condition on line 163 was never true

164 # Try numeric conversion for better error reporting 

165 try: 

166 available_numeric = {str(int(float(k))) for k in available_keys if str(k).replace('.', '').isdigit()} 

167 pattern_numeric = {str(int(float(k))) for k in pattern_keys if str(k).replace('.', '').isdigit()} 

168 missing_numeric = pattern_numeric - available_numeric 

169 

170 if missing_numeric: 

171 return ValidationResult( 

172 is_valid=False, 

173 error_message=( 

174 f"Function pattern keys {sorted(missing_numeric)} not found in available " 

175 f"{group_by.value} components {sorted(available_numeric)} for step '{step_name}'" 

176 ) 

177 ) 

178 except (ValueError, TypeError): 

179 # Fall back to string comparison 

180 return ValidationResult( 

181 is_valid=False, 

182 error_message=( 

183 f"Function pattern keys {sorted(missing_keys)} not found in available " 

184 f"{group_by.value} components {sorted(available_keys_set)} for step '{step_name}'" 

185 ) 

186 ) 

187 

188 return ValidationResult(is_valid=True) 

189 

190 except Exception as e: 

191 return ValidationResult( 

192 is_valid=False, 

193 error_message=f"Failed to validate dict pattern keys for {group_by.value}: {e}" 

194 ) 

195 

196 def validate_component_combination_constraint( 

197 self, 

198 variable_components: List[T], 

199 group_by: Optional[T] 

200 ) -> ValidationResult: 

201 """ 

202 Validate the core constraint: group_by ∉ variable_components. 

203  

204 Args: 

205 variable_components: List of variable components 

206 group_by: Optional group_by component 

207  

208 Returns: 

209 ValidationResult indicating success or failure 

210 """ 

211 try: 

212 self.config.validate_combination(variable_components, group_by) 

213 return ValidationResult(is_valid=True) 

214 except ValueError as e: 

215 return ValidationResult( 

216 is_valid=False, 

217 error_message=str(e) 

218 )