Coverage for openhcs/core/pipeline_config.py: 38.6%

1""" 

2Pipeline-specific configuration classes and utilities. 

3 

4This module contains all pipeline-specific logic that was previously mixed 

5into the generic lazy configuration system. 

6""" 

7 

8from typing import Any, Type, Optional 

9from dataclasses import fields 

10from openhcs.core.config import ( 

11 GlobalPipelineConfig, StepMaterializationConfig, 

12 set_current_global_config, register_lazy_type_mapping 

13) 

14from openhcs.core.lazy_config import ( 

15 LazyDataclassFactory, create_config_for_editing, 

16 ensure_global_config_context, CONSTANTS 

17) 

18 

19 

20def set_current_pipeline_config(config: GlobalPipelineConfig) -> None: 

21 """Set the current pipeline config for MaterializationPathConfig defaults.""" 

22 set_current_global_config(GlobalPipelineConfig, config) 

23 

24 

25def ensure_pipeline_config_context(orchestrator_global_config: Any) -> None: 

26 """Ensure proper thread-local storage setup for pipeline configuration editing.""" 

27 ensure_global_config_context(GlobalPipelineConfig, orchestrator_global_config) 

28 

29 

30def create_pipeline_config_for_editing( 

31 source_config: Any, 

32 preserve_values: bool = False 

33) -> Any: 

34 """ 

35 Create PipelineConfig for editing - pipeline-specific wrapper. 

36 

37 Args: 

38 source_config: Instance to use for context and optionally field values 

39 preserve_values: 

40 - True: Preserve actual field values (direct editing) 

41 - False: Use None values for placeholders (hierarchical editing) 

42 

43 Returns: 

44 PipelineConfig instance with appropriate field initialization 

45 """ 

46 return create_config_for_editing( 

47 GlobalPipelineConfig, 

48 source_config, 

49 preserve_values=preserve_values, 

50 placeholder_prefix="Pipeline default" 

51 ) 

52 

53 

54def create_editing_config_from_existing_lazy_config( 

55 existing_lazy_config: Any, 

56 global_config: Any 

57) -> Any: 

58 """ 

59 Create an editing config from existing lazy config with user-set values preserved as actual field values. 

60 

61 This function is used when reopening orchestrator config editing to ensure that: 

62 - User-set values appear as actual field values (not placeholders) 

63 - Unset fields remain None for placeholder behavior 

64 - Thread-local context is properly set up 

65 

66 Args: 

67 existing_lazy_config: Existing lazy config with user customizations 

68 global_config: Global config for thread-local context setup 

69 

70 Returns: 

71 New lazy config suitable for editing with preserved user values 

72 """ 

73 if existing_lazy_config is None: 

74 return None 

75 

76 # Set up thread-local context with updated global config 

77 from openhcs.core.config import GlobalPipelineConfig 

78 from openhcs.core.lazy_config import ensure_global_config_context 

79 ensure_global_config_context(GlobalPipelineConfig, global_config) 

80 

81 # Extract field values, preserving user-set values as concrete values 

82 field_values = {} 

83 for field_obj in fields(existing_lazy_config): 

84 # Get raw stored value without triggering lazy resolution 

85 raw_value = object.__getattribute__(existing_lazy_config, field_obj.name) 

86 

87 if raw_value is not None: 

88 # User has explicitly set this field - preserve as concrete value 

89 # This includes nested dataclasses that have been modified 

90 field_values[field_obj.name] = raw_value 

91 else: 

92 # Field is None - keep as None for placeholder behavior 

93 field_values[field_obj.name] = None 

94 

95 return PipelineConfig(**field_values) 

96 

97 

98# Generate pipeline-specific lazy configuration classes 

99PipelineConfig = LazyDataclassFactory.make_lazy_thread_local( 

100 base_class=GlobalPipelineConfig, 

101 global_config_type=GlobalPipelineConfig, 

102 field_path=None, # Root instance 

103 lazy_class_name=CONSTANTS.PIPELINE_CONFIG_NAME, 

104 use_recursive_resolution=True 

105) 

106 

107LazyStepMaterializationConfig = LazyDataclassFactory.make_lazy_thread_local( 

108 base_class=StepMaterializationConfig, 

109 global_config_type=GlobalPipelineConfig, 

110 field_path=CONSTANTS.MATERIALIZATION_DEFAULTS_PATH, 

111 lazy_class_name=CONSTANTS.LAZY_STEP_MATERIALIZATION_CONFIG_NAME 

112) 

113 

114 

115def _add_to_base_config_method(lazy_class: Type, base_class: Type) -> None: 

116 """Add to_base_config method to lazy dataclass for orchestrator integration.""" 

117 def to_base_config(self): 

118 """Convert lazy config to base config, resolving None values to current defaults.""" 

119 # Get all field values, resolving None values through lazy loading 

120 resolved_values = {} 

121 for field in fields(self): 

122 value = getattr(self, field.name) # This triggers lazy resolution for None values 

123 resolved_values[field.name] = value 

124 

125 return base_class(**resolved_values) 

126 

127 # Bind the method to the lazy class 

128 lazy_class.to_base_config = to_base_config 

129 

130 

131# Add to_base_config method for orchestrator integration 

132_add_to_base_config_method(PipelineConfig, GlobalPipelineConfig) 

133 

134# Register type mappings for the placeholder service 

135register_lazy_type_mapping(PipelineConfig, GlobalPipelineConfig) 

136register_lazy_type_mapping(LazyStepMaterializationConfig, StepMaterializationConfig)