Coverage for openhcs/config_framework/dual_axis_resolver.py: 22.1%
200 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +0000
1"""
2Generic dual-axis resolver for lazy configuration inheritance.
4This module provides the core inheritance resolution logic as a pure function,
5supporting both context hierarchy (X-axis) and sibling inheritance (Y-axis).
7The resolver is completely generic and has no application-specific dependencies.
8"""
10import logging
11from typing import Any, Dict, Type, Optional, List
12from dataclasses import is_dataclass
14logger = logging.getLogger(__name__)
17def _has_concrete_field_override(source_class, field_name: str) -> bool:
18 """
19 Check if a class has a concrete field override (not None).
21 This determines inheritance design based on static class definition:
22 - Concrete default (not None) = never inherits
23 - None default = always inherits (inherit_as_none design)
24 """
25 # CRITICAL FIX: Check class attribute directly, not dataclass field default
26 # The @global_pipeline_config decorator modifies field defaults to None
27 # but the class attribute retains the original concrete value
28 if hasattr(source_class, field_name):
29 class_attr_value = getattr(source_class, field_name)
30 has_override = class_attr_value is not None
31 return has_override
32 return False
35# Priority functions removed - MRO-based resolution is sufficient
38def resolve_field_inheritance_old(
39 obj,
40 field_name: str,
41 available_configs: Dict[str, Any]
42) -> Any:
43 """
44 Pure function for cross-dataclass inheritance resolution.
46 This replaces the complex RecursiveContextualResolver with explicit parameter passing.
48 Args:
49 obj: The object requesting field resolution
50 field_name: Name of the field to resolve
51 available_configs: Dict mapping config type names to config instances
52 e.g., {'GlobalPipelineConfig': global_config, 'StepConfig': step_config}
54 Returns:
55 Resolved field value or None if not found
57 Algorithm:
58 1. Check if obj has concrete value for field_name
59 2. Check Y-axis inheritance within obj's MRO for concrete values
60 3. Check related config types in available_configs for cross-dataclass inheritance
61 4. Return class defaults as final fallback
62 """
63 obj_type = type(obj)
65 # Step 1: Check concrete value in merged context for obj's type (HIGHEST PRIORITY)
66 # CRITICAL: Context values take absolute precedence over inheritance blocking
67 # The config_context() manager merges concrete values into available_configs
68 for config_name, config_instance in available_configs.items():
69 if type(config_instance) == obj_type:
70 try:
71 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
72 value = object.__getattribute__(config_instance, field_name)
73 if value is not None:
74 if field_name == 'well_filter':
75 logger.debug(f"🔍 CONTEXT: Found concrete value in merged context {obj_type.__name__}.{field_name}: {value}")
76 return value
77 except AttributeError:
78 # Field doesn't exist on this config type
79 continue
81 # Step 1b: Check concrete value on obj instance itself (fallback)
82 # Use object.__getattribute__ to avoid recursion with lazy __getattribute__
83 try:
84 value = object.__getattribute__(obj, field_name)
85 if value is not None:
86 if field_name == 'well_filter':
87 logger.debug(f"🔍 INSTANCE: Found concrete value on instance {obj_type.__name__}.{field_name}: {value}")
88 return value
89 except AttributeError:
90 # Field doesn't exist on the object
91 pass
93 # Step 2: FIELD-SPECIFIC INHERITANCE BLOCKING
94 # Check if this specific field has a concrete value in the exact same type
95 # Only block inheritance if the EXACT same type has a non-None value
96 for config_name, config_instance in available_configs.items():
97 if type(config_instance) == obj_type:
98 try:
99 field_value = object.__getattribute__(config_instance, field_name)
100 if field_value is not None:
101 # This exact type has a concrete value - use it, don't inherit
102 if field_name == 'well_filter':
103 logger.debug(f"🔍 FIELD-SPECIFIC BLOCKING: {obj_type.__name__}.{field_name} = {field_value} (concrete) - blocking inheritance")
104 return field_value
105 except AttributeError:
106 continue
108 # DEBUG: Log what we're trying to resolve
109 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
110 logger.debug(f"🔍 RESOLVING {obj_type.__name__}.{field_name} - checking context and inheritance")
111 logger.debug(f"🔍 AVAILABLE CONFIGS: {list(available_configs.keys())}")
113 # Step 3: Y-axis inheritance within obj's MRO
114 blocking_class = _find_blocking_class_in_mro(obj_type, field_name)
116 for parent_type in obj_type.__mro__[1:]:
117 if not is_dataclass(parent_type):
118 continue
120 # Check blocking logic
121 if blocking_class and parent_type != blocking_class:
122 continue
124 if blocking_class and parent_type == blocking_class:
125 # Check if blocking class has concrete value in available configs
126 for config_name, config_instance in available_configs.items():
127 if type(config_instance) == parent_type:
128 try:
129 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
130 value = object.__getattribute__(config_instance, field_name)
131 if value is None:
132 # Blocking class has None - inheritance blocked
133 break
134 else:
135 logger.debug(f"Inherited from blocking class {parent_type.__name__}: {value}")
136 return value
137 except AttributeError:
138 # Field doesn't exist on this config type
139 continue
140 break
142 # Normal inheritance - check for concrete values
143 for config_name, config_instance in available_configs.items():
144 if type(config_instance) == parent_type:
145 try:
146 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
147 value = object.__getattribute__(config_instance, field_name)
148 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
149 logger.debug(f"🔍 Y-AXIS INHERITANCE: {parent_type.__name__}.{field_name} = {value}")
150 if value is not None:
151 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
152 logger.debug(f"🔍 Y-AXIS INHERITANCE: FOUND {parent_type.__name__}.{field_name}: {value} (returning)")
153 logger.debug(f"Inherited from {parent_type.__name__}: {value}")
154 return value
155 except AttributeError:
156 # Field doesn't exist on this config type
157 continue
159 # Step 4: Cross-dataclass inheritance from related config types (PRIORITY-ORDERED)
160 # NOTE: Inheritance blocking was already applied in Step 2, so this only runs for types without concrete overrides
161 # CRITICAL FIX: Process configs in priority order to ensure proper inheritance hierarchy
162 sorted_configs = _sort_configs_by_priority(available_configs)
164 for config_name, config_instance in sorted_configs:
165 config_type = type(config_instance)
166 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
167 priority = _get_config_priority(config_type)
168 logger.debug(f"🔍 CROSS-DATACLASS: Checking {config_type.__name__} (priority {priority}) for {field_name}")
170 if _is_related_config_type(obj_type, config_type):
171 # Skip if this is the same type as the requesting object (avoid self-inheritance)
172 if config_type == obj_type:
173 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
174 logger.debug(f"🔍 CROSS-DATACLASS: Skipping self-inheritance from {config_type.__name__}")
175 continue
177 # CRITICAL FIX: Prevent lower-priority configs from inheriting from higher-priority configs
178 # Base classes (higher priority numbers) should NOT inherit from derived classes (lower priority numbers)
179 obj_priority = _get_config_priority(obj_type)
180 config_priority = _get_config_priority(config_type)
182 if obj_priority > config_priority:
183 # Requesting object has LOWER priority (higher number) than the config - skip inheritance
184 # Example: WellFilterConfig (priority 11) should NOT inherit from StepWellFilterConfig (priority 2)
185 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
186 logger.debug(f"🔍 CROSS-DATACLASS: Skipping inheritance from higher-priority {config_type.__name__} (priority {config_priority}) to lower-priority {obj_type.__name__} (priority {obj_priority})")
187 continue
189 try:
190 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
191 value = object.__getattribute__(config_instance, field_name)
192 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
193 logger.debug(f"🔍 CROSS-DATACLASS: {config_type.__name__}.{field_name} = {value} (related config)")
194 if value is not None:
195 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
196 logger.debug(f"🔍 CROSS-DATACLASS: FOUND {config_type.__name__}.{field_name}: {value} (priority {priority})")
197 logger.debug(f"Cross-dataclass inheritance from {config_type.__name__}: {value}")
198 return value
199 except AttributeError:
200 # Field doesn't exist on this config type
201 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
202 logger.debug(f"🔍 CROSS-DATACLASS: {config_type.__name__} has no field {field_name}")
203 continue
204 else:
205 if field_name in ['output_dir_suffix', 'sub_dir']:
206 logger.debug(f"🔍 CROSS-DATACLASS: {config_type.__name__} not related to {obj_type.__name__}")
208 # Step 4: Class defaults as final fallback
209 if blocking_class:
210 try:
211 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
212 class_default = object.__getattribute__(blocking_class, field_name)
213 if class_default is not None:
214 logger.debug(f"Using class default from blocking class {blocking_class.__name__}: {class_default}")
215 return class_default
216 except AttributeError:
217 # Field doesn't exist on blocking class
218 pass
220 logger.debug(f"No resolution found for {obj_type.__name__}.{field_name}")
221 return None
224def _is_related_config_type(obj_type: Type, config_type: Type) -> bool:
225 """
226 Check if config_type is related to obj_type for cross-dataclass inheritance.
228 CRITICAL FIX: Only allow inheritance from parent classes or sibling classes at the same level,
229 NOT from child classes. This prevents WellFilterConfig from inheriting from StepWellFilterConfig.
231 Args:
232 obj_type: The type requesting field resolution
233 config_type: The type being checked for relationship
235 Returns:
236 True if config_type should be considered for cross-dataclass inheritance
237 """
238 # CRITICAL: Only allow inheritance from parent classes (obj_type inherits from config_type)
239 # This prevents base classes from inheriting from their derived classes
240 if issubclass(obj_type, config_type):
241 return True
243 # Allow sibling inheritance only if they share a common parent but neither inherits from the other
244 # This allows StepMaterializationConfig to inherit from both StepWellFilterConfig and PathPlanningConfig
245 if not issubclass(config_type, obj_type): # config_type is NOT a child of obj_type
246 # Check if they share a common dataclass ancestor (excluding themselves)
247 obj_ancestors = set(cls for cls in obj_type.__mro__[1:] if is_dataclass(cls)) # Skip obj_type itself
248 config_ancestors = set(cls for cls in config_type.__mro__[1:] if is_dataclass(cls)) # Skip config_type itself
250 shared_ancestors = obj_ancestors & config_ancestors
251 if shared_ancestors:
252 return True
254 return False
257def resolve_field_inheritance(
258 obj,
259 field_name: str,
260 available_configs: Dict[str, Any]
261) -> Any:
262 """
263 Simplified MRO-based inheritance resolution.
265 ALGORITHM:
266 1. Check if obj has concrete value for field_name in context
267 2. Traverse obj's MRO from most to least specific
268 3. For each MRO class, check if there's a config instance in context with concrete (non-None) value
269 4. Return first concrete value found
271 Args:
272 obj: The object requesting field resolution
273 field_name: Name of the field to resolve
274 available_configs: Dict mapping config type names to config instances
276 Returns:
277 Resolved field value or None if not found
278 """
279 obj_type = type(obj)
281 # Step 1: Check if exact same type has concrete value in context
282 for config_name, config_instance in available_configs.items():
283 if type(config_instance) == obj_type: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 try:
285 field_value = object.__getattribute__(config_instance, field_name)
286 if field_value is not None:
287 if field_name == 'well_filter':
288 logger.debug(f"🔍 CONCRETE VALUE: {obj_type.__name__}.{field_name} = {field_value}")
289 return field_value
290 except AttributeError:
291 continue
293 # Step 2: MRO-based inheritance - traverse MRO from most to least specific
294 # For each class in the MRO, check if there's a config instance in context with concrete value
295 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
296 logger.debug(f"🔍 MRO-INHERITANCE: Resolving {obj_type.__name__}.{field_name}")
297 logger.debug(f"🔍 MRO-INHERITANCE: MRO = {[cls.__name__ for cls in obj_type.__mro__]}")
299 for mro_class in obj_type.__mro__:
300 if not is_dataclass(mro_class):
301 continue
303 # Look for a config instance of this MRO class type in the available configs
304 for config_name, config_instance in available_configs.items():
305 if type(config_instance) == mro_class:
306 try:
307 value = object.__getattribute__(config_instance, field_name)
308 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
309 logger.debug(f"🔍 MRO-INHERITANCE: {mro_class.__name__}.{field_name} = {value}")
310 if value is not None:
311 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
312 logger.debug(f"🔍 MRO-INHERITANCE: FOUND {mro_class.__name__}.{field_name}: {value} (returning)")
313 return value
314 except AttributeError:
315 continue
317 # Step 3: Class defaults as final fallback
318 try:
319 class_default = object.__getattribute__(obj_type, field_name)
320 if class_default is not None: 320 ↛ 321line 320 didn't jump to line 321 because the condition on line 320 was never true
321 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
322 logger.debug(f"🔍 CLASS-DEFAULT: {obj_type.__name__}.{field_name} = {class_default}")
323 return class_default
324 except AttributeError:
325 pass
327 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']: 327 ↛ 328line 327 didn't jump to line 328 because the condition on line 327 was never true
328 logger.debug(f"🔍 NO-RESOLUTION: {obj_type.__name__}.{field_name} = None")
329 return None
332# Utility functions for inheritance detection (kept from original resolver)
334def _has_concrete_field_override(config_class: Type, field_name: str) -> bool:
335 """
336 Check if a class has a concrete field override (not None).
338 This determines class-level inheritance blocking behavior based on static class definition.
339 Now checks the entire MRO chain to handle inherited fields properly.
340 """
341 try:
342 # Check the entire MRO chain for concrete field values
343 for cls in config_class.__mro__:
344 if hasattr(cls, field_name):
345 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
346 class_attr_value = object.__getattribute__(cls, field_name)
347 if class_attr_value is not None:
348 has_override = True
349 logger.debug(f"Class override check {config_class.__name__}.{field_name}: found concrete value {class_attr_value} in {cls.__name__}, has_override={has_override}")
350 return has_override
352 # No concrete value found in any class in the MRO
353 logger.debug(f"Class override check {config_class.__name__}.{field_name}: no concrete value in MRO, has_override=False")
354 return False
355 except AttributeError:
356 # Field doesn't exist on class
357 return False
360def _find_blocking_class_in_mro(base_type: Type, field_name: str) -> Optional[Type]:
361 """
362 Find the first class in MRO that has a concrete field override AND blocks inheritance from parent classes.
364 A class blocks inheritance only if:
365 1. It has a concrete field override
366 2. There are parent classes in the MRO that also have the same field
368 This prevents legitimate inheritance sources (like GlobalPipelineConfig) from being treated as blockers.
370 Returns:
371 The first class in MRO order that blocks inheritance, or None if no blocking class found.
372 """
373 for i, cls in enumerate(base_type.__mro__):
374 if not is_dataclass(cls):
375 continue
376 if _has_concrete_field_override(cls, field_name):
377 # Check if there are parent classes that also have this field
378 has_parent_with_field = False
379 for parent_cls in base_type.__mro__[i + 1:]:
380 if not is_dataclass(parent_cls):
381 continue
382 try:
383 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
384 object.__getattribute__(parent_cls, field_name)
385 has_parent_with_field = True
386 break
387 except AttributeError:
388 # Field doesn't exist on this parent class
389 continue
391 if has_parent_with_field:
392 logger.debug(f"Found blocking class {cls.__name__} for {base_type.__name__}.{field_name} (blocks parent inheritance)")
393 return cls
394 else:
395 logger.debug(f"Class {cls.__name__} has concrete override but no parents with field - not blocking")
396 return None
399# All legacy functions removed - use resolve_field_inheritance() instead