Coverage for openhcs/formats/func_arg_prep.py: 2.2%

1from typing import List, Dict 

2 

3 

4def _resolve_function_references(func_value): 

5 """ 

6 Recursively resolve FunctionReference objects to actual functions. 

7 

8 This handles all function pattern formats and resolves any FunctionReference 

9 objects back to their actual decorated functions from the registry. 

10 """ 

11 # Import here to avoid circular imports 

12 try: 

13 from openhcs.core.pipeline.compiler import FunctionReference 

14 except ImportError: 

15 # If FunctionReference doesn't exist, just return the original value 

16 return func_value 

17 

18 if isinstance(func_value, FunctionReference): 

19 # Resolve FunctionReference to actual function 

20 return func_value.resolve() 

21 elif isinstance(func_value, tuple) and len(func_value) == 2: 

22 # Tuple: (function_or_ref, kwargs) → (resolved_function, kwargs) 

23 func_or_ref, kwargs = func_value 

24 resolved_func = _resolve_function_references(func_or_ref) 

25 return (resolved_func, kwargs) 

26 elif isinstance(func_value, list): 

27 # List of functions/tuples → List of resolved functions/tuples 

28 return [_resolve_function_references(item) for item in func_value] 

29 elif isinstance(func_value, dict): 

30 # Dict of functions/tuples → Dict of resolved functions/tuples 

31 return {key: _resolve_function_references(value) for key, value in func_value.items()} 

32 else: 

33 # Not a function pattern or already a callable, return as-is 

34 return func_value 

35 

36 

37def prepare_patterns_and_functions(patterns, processing_funcs, component='default'): 

38 """ 

39 Prepare patterns, processing functions, and processing args for processing. 

40 

41 This function handles three main tasks: 

42 1. Ensuring patterns are in a component-keyed dictionary format 

43 2. Determining which processing functions to use for each component 

44 3. Determining which processing args to use for each component 

45 

46 Args: 

47 patterns (list or dict): Patterns to process, either as a flat list or grouped by component 

48 processing_funcs (callable, list, dict, tuple, optional): Processing functions to apply. 

49 Can be a single callable, a tuple of (callable, kwargs), a list of either, 

50 or a dictionary mapping component values to any of these. 

51 component (str): Component name for grouping (only used for clarity in the result) 

52 

53 Returns: 

54 tuple: (grouped_patterns, component_to_funcs, component_to_args) 

55 - grouped_patterns: Dictionary mapping component values to patterns 

56 - component_to_funcs: Dictionary mapping component values to processing functions 

57 - component_to_args: Dictionary mapping component values to processing args 

58 """ 

59 import logging 

60 logger = logging.getLogger(__name__) 

61 

62 # Debug: Log what we received 

63 logger.debug(f"🔍 PATTERN DEBUG: prepare_patterns_and_functions called") 

64 logger.debug(f"🔍 PATTERN DEBUG: patterns type: {type(patterns)}") 

65 logger.debug(f"🔍 PATTERN DEBUG: patterns keys/content: {list(patterns.keys()) if isinstance(patterns, dict) else f'List with {len(patterns)} items'}") 

66 logger.debug(f"🔍 PATTERN DEBUG: processing_funcs type: {type(processing_funcs)}") 

67 logger.debug(f"🔍 PATTERN DEBUG: processing_funcs keys: {list(processing_funcs.keys()) if isinstance(processing_funcs, dict) else 'Not a dict'}") 

68 logger.debug(f"🔍 PATTERN DEBUG: component: {component}") 

69 

70 # CRITICAL: Resolve any FunctionReference objects to actual functions 

71 # This ensures worker processes get properly decorated functions from their registry 

72 processing_funcs = _resolve_function_references(processing_funcs) 

73 logger.debug(f"🔧 FUNCTION RESOLUTION: Resolved FunctionReference objects in processing_funcs") 

74 

75 # Ensure patterns are in a dictionary format 

76 # If already a dict, use as is; otherwise wrap the list in a dictionary 

77 grouped_patterns = patterns if isinstance(patterns, dict) else {component: patterns} 

78 

79 logger.debug(f"🔍 PATTERN DEBUG: grouped_patterns keys: {list(grouped_patterns.keys())}") 

80 

81 # SMART FILTERING: If processing_funcs is a dict, only process components that have function definitions 

82 if isinstance(processing_funcs, dict) and isinstance(grouped_patterns, dict): 

83 original_components = set(grouped_patterns.keys()) 

84 function_components = set(processing_funcs.keys()) 

85 

86 # Handle type mismatches (string vs int keys) 

87 available_function_keys = set() 

88 for key in function_components: 

89 available_function_keys.add(key) 

90 available_function_keys.add(str(key)) # Add string version 

91 if isinstance(key, str) and key.isdigit(): 

92 available_function_keys.add(int(key)) # Add int version if string is numeric 

93 

94 # Filter to only components that have function definitions 

95 filtered_grouped_patterns = { 

96 comp_value: patterns 

97 for comp_value, patterns in grouped_patterns.items() 

98 if comp_value in available_function_keys 

99 } 

100 

101 # Log what was filtered 

102 filtered_out = original_components - set(filtered_grouped_patterns.keys()) 

103 if filtered_out: 

104 logger.debug(f"🔍 PATTERN DEBUG: Filtered out components without function definitions: {filtered_out}") 

105 

106 logger.debug(f"🔍 PATTERN DEBUG: Processing components: {list(filtered_grouped_patterns.keys())}") 

107 grouped_patterns = filtered_grouped_patterns 

108 

109 # Validate that we have at least one component to process 

110 if not grouped_patterns: 

111 available_keys = list(processing_funcs.keys()) 

112 discovered_keys = list(original_components) 

113 raise ValueError( 

114 f"No components match between discovered data and function pattern. " 

115 f"Discovered components: {discovered_keys}. " 

116 f"Function pattern keys: {available_keys}. " 

117 f"Function pattern keys must match discovered component values." 

118 ) 

119 

120 # Initialize dictionaries for functions and args 

121 component_to_funcs = {} 

122 component_to_args = {} 

123 

124 # Helper function to extract function and args from a function item 

125 def extract_func_and_args(func_item): 

126 if isinstance(func_item, tuple) and len(func_item) == 2 and callable(func_item[0]): 

127 # It's a (function, kwargs) tuple 

128 return func_item[0], func_item[1] 

129 if callable(func_item): 

130 # It's just a function, use default args 

131 return func_item, {} 

132 if isinstance(func_item, dict): 

133 # It's a dictionary pattern - this should be handled at a higher level 

134 # This indicates a logic error where the entire dict was passed instead of individual components 

135 raise ValueError( 

136 f"Dictionary pattern passed to extract_func_and_args: {func_item}. " 

137 f"This indicates a component lookup failure in prepare_patterns_and_functions. " 

138 f"Dictionary patterns should be resolved to individual function lists before reaching this point." 

139 ) 

140 # Fail loudly and early if the function item is invalid 

141 raise ValueError(f"Invalid function item for pattern processing: {func_item}") 

142 

143 for comp_value in grouped_patterns.keys(): 

144 # Get functions and args for this component 

145 # No special handling for 'channel' component (Clause 77: Rot Intolerance) 

146 import logging 

147 logger = logging.getLogger(__name__) 

148 logger.debug(f"Processing component value: '{comp_value}' (type: {type(comp_value)})") 

149 logger.debug(f"Function pattern keys: {list(processing_funcs.keys()) if isinstance(processing_funcs, dict) else 'Not a dict'}") 

150 

151 if isinstance(processing_funcs, dict): 

152 # Direct lookup with type conversion fallback 

153 # Compile-time validation guarantees dict keys are valid 

154 if comp_value in processing_funcs: 

155 func_item = processing_funcs[comp_value] 

156 logger.debug(f"Found direct match for '{comp_value}': {type(func_item)}") 

157 else: 

158 # Handle type mismatch: pattern detection returns strings, but function pattern might use integers 

159 logger.debug(f"No direct match for '{comp_value}', trying integer conversion") 

160 try: 

161 comp_value_int = int(comp_value) 

162 if comp_value_int in processing_funcs: 

163 func_item = processing_funcs[comp_value_int] 

164 else: 

165 # Try converting keys to int for comparison 

166 found = False 

167 for key in processing_funcs.keys(): 

168 try: 

169 if int(key) == comp_value_int: 

170 func_item = processing_funcs[key] 

171 found = True 

172 break 

173 except (ValueError, TypeError): 

174 continue 

175 if not found: 

176 # This should not happen due to compile-time validation 

177 func_item = processing_funcs[comp_value] 

178 except (ValueError, TypeError): 

179 # This should not happen due to compile-time validation 

180 func_item = processing_funcs[comp_value] 

181 else: 

182 # Use the same function for all components 

183 func_item = processing_funcs 

184 

185 # Extract function and args 

186 logger.debug(f"Processing func_item for '{comp_value}': {type(func_item)}") 

187 if isinstance(func_item, list): 

188 # List of functions or function tuples 

189 logger.debug(f"func_item is a list with {len(func_item)} items") 

190 component_to_funcs[comp_value] = func_item 

191 # For lists, we'll extract args during processing 

192 component_to_args[comp_value] = {} 

193 else: 

194 # Single function or function tuple 

195 logger.debug(f"Calling extract_func_and_args with: {type(func_item)}") 

196 func, args = extract_func_and_args(func_item) 

197 component_to_funcs[comp_value] = func 

198 component_to_args[comp_value] = args 

199 

200 return grouped_patterns, component_to_funcs, component_to_args 

201 

202