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

1from typing import List, Dict 

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

3 """ 

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

5 

6 This function handles three main tasks: 

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

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

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

10 

11 Args: 

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

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

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

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

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

17 

18 Returns: 

19 tuple: (grouped_patterns, component_to_funcs, component_to_args) 

20 - grouped_patterns: Dictionary mapping component values to patterns 

21 - component_to_funcs: Dictionary mapping component values to processing functions 

22 - component_to_args: Dictionary mapping component values to processing args 

23 """ 

24 import logging 

25 logger = logging.getLogger(__name__) 

26 

27 # Debug: Log what we received 

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

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

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

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

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

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

34 

35 # Ensure patterns are in a dictionary format 

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

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

38 

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

40 

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

42 if isinstance(processing_funcs, dict) and isinstance(grouped_patterns, dict): 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 original_components = set(grouped_patterns.keys()) 

44 function_components = set(processing_funcs.keys()) 

45 

46 # Handle type mismatches (string vs int keys) 

47 available_function_keys = set() 

48 for key in function_components: 

49 available_function_keys.add(key) 

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

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

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

53 

54 # Filter to only components that have function definitions 

55 filtered_grouped_patterns = { 

56 comp_value: patterns 

57 for comp_value, patterns in grouped_patterns.items() 

58 if comp_value in available_function_keys 

59 } 

60 

61 # Log what was filtered 

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

63 if filtered_out: 

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

65 

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

67 grouped_patterns = filtered_grouped_patterns 

68 

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

70 if not grouped_patterns: 

71 available_keys = list(processing_funcs.keys()) 

72 discovered_keys = list(original_components) 

73 raise ValueError( 

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

75 f"Discovered components: {discovered_keys}. " 

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

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

78 ) 

79 

80 # Initialize dictionaries for functions and args 

81 component_to_funcs = {} 

82 component_to_args = {} 

83 

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

85 def extract_func_and_args(func_item): 

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

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

88 return func_item[0], func_item[1] 

89 if callable(func_item): 89 ↛ 92line 89 didn't jump to line 92 because the condition on line 89 was always true

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

91 return func_item, {} 

92 if isinstance(func_item, dict): 

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

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

95 raise ValueError( 

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

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

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

99 ) 

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

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

102 

103 for comp_value in grouped_patterns.keys(): 

104 # Get functions and args for this component 

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

106 import logging 

107 logger = logging.getLogger(__name__) 

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

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

110 

111 if isinstance(processing_funcs, dict): 111 ↛ 114line 111 didn't jump to line 114 because the condition on line 111 was never true

112 # Direct lookup with type conversion fallback 

113 # Compile-time validation guarantees dict keys are valid 

114 if comp_value in processing_funcs: 

115 func_item = processing_funcs[comp_value] 

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

117 else: 

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

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

120 try: 

121 comp_value_int = int(comp_value) 

122 if comp_value_int in processing_funcs: 

123 func_item = processing_funcs[comp_value_int] 

124 else: 

125 # Try converting keys to int for comparison 

126 found = False 

127 for key in processing_funcs.keys(): 

128 try: 

129 if int(key) == comp_value_int: 

130 func_item = processing_funcs[key] 

131 found = True 

132 break 

133 except (ValueError, TypeError): 

134 continue 

135 if not found: 

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

137 func_item = processing_funcs[comp_value] 

138 except (ValueError, TypeError): 

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

140 func_item = processing_funcs[comp_value] 

141 else: 

142 # Use the same function for all components 

143 func_item = processing_funcs 

144 

145 # Extract function and args 

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

147 if isinstance(func_item, list): 

148 # List of functions or function tuples 

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

150 component_to_funcs[comp_value] = func_item 

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

152 component_to_args[comp_value] = {} 

153 else: 

154 # Single function or function tuple 

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

156 func, args = extract_func_and_args(func_item) 

157 component_to_funcs[comp_value] = func 

158 component_to_args[comp_value] = args 

159 

160 return grouped_patterns, component_to_funcs, component_to_args 

161 

162