Coverage for openhcs/formats/func_arg_prep.py: 67.6%
96 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
3def _resolve_function_references(func_value):
4 """
5 Recursively resolve FunctionReference objects to actual functions.
7 This handles all function pattern formats and resolves any FunctionReference
8 objects back to their actual decorated functions from the registry.
9 """
10 # Import here to avoid circular imports
11 try:
12 from openhcs.core.pipeline.compiler import FunctionReference
13 except ImportError:
14 # If FunctionReference doesn't exist, just return the original value
15 return func_value
17 if isinstance(func_value, FunctionReference):
18 # Resolve FunctionReference to actual function
19 return func_value.resolve()
20 elif isinstance(func_value, tuple) and len(func_value) == 2:
21 # Tuple: (function_or_ref, kwargs) → (resolved_function, kwargs)
22 func_or_ref, kwargs = func_value
23 resolved_func = _resolve_function_references(func_or_ref)
24 return (resolved_func, kwargs)
25 elif isinstance(func_value, list):
26 # List of functions/tuples → List of resolved functions/tuples
27 return [_resolve_function_references(item) for item in func_value]
28 elif isinstance(func_value, dict): 28 ↛ 33line 28 didn't jump to line 33 because the condition on line 28 was always true
29 # Dict of functions/tuples → Dict of resolved functions/tuples
30 return {key: _resolve_function_references(value) for key, value in func_value.items()}
31 else:
32 # Not a function pattern or already a callable, return as-is
33 return func_value
36def prepare_patterns_and_functions(patterns, processing_funcs, component='default'):
37 """
38 Prepare patterns, processing functions, and processing args for processing.
40 This function handles three main tasks:
41 1. Ensuring patterns are in a component-keyed dictionary format
42 2. Determining which processing functions to use for each component
43 3. Determining which processing args to use for each component
45 Args:
46 patterns (list or dict): Patterns to process, either as a flat list or grouped by component
47 processing_funcs (callable, list, dict, tuple, optional): Processing functions to apply.
48 Can be a single callable, a tuple of (callable, kwargs), a list of either,
49 or a dictionary mapping component values to any of these.
50 component (str): Component name for grouping (only used for clarity in the result)
52 Returns:
53 tuple: (grouped_patterns, component_to_funcs, component_to_args)
54 - grouped_patterns: Dictionary mapping component values to patterns
55 - component_to_funcs: Dictionary mapping component values to processing functions
56 - component_to_args: Dictionary mapping component values to processing args
57 """
58 import logging
59 logger = logging.getLogger(__name__)
61 # Debug: Log what we received
62 logger.debug("🔍 PATTERN DEBUG: prepare_patterns_and_functions called")
63 logger.debug(f"🔍 PATTERN DEBUG: patterns type: {type(patterns)}")
64 logger.debug(f"🔍 PATTERN DEBUG: patterns keys/content: {list(patterns.keys()) if isinstance(patterns, dict) else f'List with {len(patterns)} items'}")
65 logger.debug(f"🔍 PATTERN DEBUG: processing_funcs type: {type(processing_funcs)}")
66 logger.debug(f"🔍 PATTERN DEBUG: processing_funcs keys: {list(processing_funcs.keys()) if isinstance(processing_funcs, dict) else 'Not a dict'}")
67 logger.debug(f"🔍 PATTERN DEBUG: component: {component}")
69 # CRITICAL: Resolve any FunctionReference objects to actual functions
70 # This ensures worker processes get properly decorated functions from their registry
71 processing_funcs = _resolve_function_references(processing_funcs)
72 logger.debug("🔧 FUNCTION RESOLUTION: Resolved FunctionReference objects in processing_funcs")
74 # Ensure patterns are in a dictionary format
75 # If already a dict, use as is; otherwise wrap the list in a dictionary
76 grouped_patterns = patterns if isinstance(patterns, dict) else {component: patterns}
78 logger.debug(f"🔍 PATTERN DEBUG: grouped_patterns keys: {list(grouped_patterns.keys())}")
80 # SMART FILTERING: If processing_funcs is a dict, only process components that have function definitions
81 if isinstance(processing_funcs, dict) and isinstance(grouped_patterns, dict):
82 original_components = set(grouped_patterns.keys())
83 function_components = set(processing_funcs.keys())
85 # Handle type mismatches (string vs int keys)
86 available_function_keys = set()
87 for key in function_components:
88 available_function_keys.add(key)
89 available_function_keys.add(str(key)) # Add string version
90 if isinstance(key, str) and key.isdigit(): 90 ↛ 87line 90 didn't jump to line 87 because the condition on line 90 was always true
91 available_function_keys.add(int(key)) # Add int version if string is numeric
93 # Filter to only components that have function definitions
94 filtered_grouped_patterns = {
95 comp_value: patterns
96 for comp_value, patterns in grouped_patterns.items()
97 if comp_value in available_function_keys
98 }
100 # Log what was filtered
101 filtered_out = original_components - set(filtered_grouped_patterns.keys())
102 if filtered_out: 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true
103 logger.debug(f"🔍 PATTERN DEBUG: Filtered out components without function definitions: {filtered_out}")
105 logger.debug(f"🔍 PATTERN DEBUG: Processing components: {list(filtered_grouped_patterns.keys())}")
106 grouped_patterns = filtered_grouped_patterns
108 # Validate that we have at least one component to process
109 if not grouped_patterns: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true
110 available_keys = list(processing_funcs.keys())
111 discovered_keys = list(original_components)
112 raise ValueError(
113 f"No components match between discovered data and function pattern. "
114 f"Discovered components: {discovered_keys}. "
115 f"Function pattern keys: {available_keys}. "
116 f"Function pattern keys must match discovered component values."
117 )
119 # Initialize dictionaries for functions and args
120 component_to_funcs = {}
121 component_to_args = {}
123 # Helper function to extract function and args from a function item
124 def extract_func_and_args(func_item):
125 if isinstance(func_item, tuple) and len(func_item) == 2 and callable(func_item[0]):
126 # It's a (function, kwargs) tuple
127 return func_item[0], func_item[1]
128 if callable(func_item): 128 ↛ 131line 128 didn't jump to line 131 because the condition on line 128 was always true
129 # It's just a function, use default args
130 return func_item, {}
131 if isinstance(func_item, dict):
132 # It's a dictionary pattern - this should be handled at a higher level
133 # This indicates a logic error where the entire dict was passed instead of individual components
134 raise ValueError(
135 f"Dictionary pattern passed to extract_func_and_args: {func_item}. "
136 f"This indicates a component lookup failure in prepare_patterns_and_functions. "
137 f"Dictionary patterns should be resolved to individual function lists before reaching this point."
138 )
139 # Fail loudly and early if the function item is invalid
140 raise ValueError(f"Invalid function item for pattern processing: {func_item}")
142 for comp_value in grouped_patterns.keys():
143 # Get functions and args for this component
144 # No special handling for 'channel' component (Clause 77: Rot Intolerance)
145 import logging
146 logger = logging.getLogger(__name__)
147 logger.debug(f"Processing component value: '{comp_value}' (type: {type(comp_value)})")
148 logger.debug(f"Function pattern keys: {list(processing_funcs.keys()) if isinstance(processing_funcs, dict) else 'Not a dict'}")
150 if isinstance(processing_funcs, dict):
151 # Direct lookup with type conversion fallback
152 # Compile-time validation guarantees dict keys are valid
153 if comp_value in processing_funcs: 153 ↛ 158line 153 didn't jump to line 158 because the condition on line 153 was always true
154 func_item = processing_funcs[comp_value]
155 logger.debug(f"Found direct match for '{comp_value}': {type(func_item)}")
156 else:
157 # Handle type mismatch: pattern detection returns strings, but function pattern might use integers
158 logger.debug(f"No direct match for '{comp_value}', trying integer conversion")
159 try:
160 comp_value_int = int(comp_value)
161 if comp_value_int in processing_funcs:
162 func_item = processing_funcs[comp_value_int]
163 else:
164 # Try converting keys to int for comparison
165 found = False
166 for key in processing_funcs.keys():
167 try:
168 if int(key) == comp_value_int:
169 func_item = processing_funcs[key]
170 found = True
171 break
172 except (ValueError, TypeError):
173 continue
174 if not found:
175 # This should not happen due to compile-time validation
176 func_item = processing_funcs[comp_value]
177 except (ValueError, TypeError):
178 # This should not happen due to compile-time validation
179 func_item = processing_funcs[comp_value]
180 else:
181 # Use the same function for all components
182 func_item = processing_funcs
184 # Extract function and args
185 logger.debug(f"Processing func_item for '{comp_value}': {type(func_item)}")
186 if isinstance(func_item, list):
187 # List of functions or function tuples
188 logger.debug(f"func_item is a list with {len(func_item)} items")
189 component_to_funcs[comp_value] = func_item
190 # For lists, we'll extract args during processing
191 component_to_args[comp_value] = {}
192 else:
193 # Single function or function tuple
194 logger.debug(f"Calling extract_func_and_args with: {type(func_item)}")
195 func, args = extract_func_and_args(func_item)
196 component_to_funcs[comp_value] = func
197 component_to_args[comp_value] = args
199 return grouped_patterns, component_to_funcs, component_to_args