Coverage for openhcs/introspection/unified_parameter_analyzer.py: 30.9%

105 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-11-04 02:09 +0000

1"""Unified parameter analysis interface for all parameter sources in OpenHCS TUI. 

2 

3This module provides a single, consistent interface for analyzing parameters from: 

4- Functions and methods 

5- Dataclasses and their fields 

6- Nested dataclass structures 

7- Any callable or type with parameters 

8 

9Replaces the fragmented approach of SignatureAnalyzer vs FieldIntrospector. 

10""" 

11 

12import inspect 

13import dataclasses 

14from typing import Dict, Union, Callable, Type, Any, Optional 

15from dataclasses import dataclass 

16 

17from openhcs.introspection.signature_analyzer import SignatureAnalyzer, ParameterInfo 

18 

19 

20@dataclass 

21class UnifiedParameterInfo: 

22 """Unified parameter information that works for all parameter sources.""" 

23 name: str 

24 param_type: Type 

25 default_value: Any 

26 is_required: bool 

27 description: Optional[str] = None 

28 source_type: str = "unknown" # "function", "dataclass", "nested" 

29 

30 @classmethod 

31 def from_parameter_info(cls, param_info: ParameterInfo, source_type: str = "function") -> "UnifiedParameterInfo": 

32 """Convert from existing ParameterInfo to unified format.""" 

33 return cls( 

34 name=param_info.name, 

35 param_type=param_info.param_type, 

36 default_value=param_info.default_value, 

37 is_required=param_info.is_required, 

38 description=param_info.description, 

39 source_type=source_type 

40 ) 

41 

42 

43class UnifiedParameterAnalyzer: 

44 """Single interface for analyzing parameters from any source. 

45  

46 This class provides a unified way to extract parameter information 

47 from functions, dataclasses, and other parameter sources, ensuring 

48 consistent behavior across the entire application. 

49 """ 

50 

51 @staticmethod 

52 def analyze(target: Union[Callable, Type, object], exclude_params: Optional[list] = None) -> Dict[str, UnifiedParameterInfo]: 

53 """Analyze parameters from any source. 

54 

55 Args: 

56 target: Function, method, dataclass type, or instance to analyze 

57 exclude_params: Optional list of parameter names to exclude from analysis 

58 

59 Returns: 

60 Dictionary mapping parameter names to UnifiedParameterInfo objects 

61 

62 Examples: 

63 # Function analysis 

64 param_info = UnifiedParameterAnalyzer.analyze(my_function) 

65 

66 # Dataclass analysis 

67 param_info = UnifiedParameterAnalyzer.analyze(MyDataclass) 

68 

69 # Instance analysis 

70 param_info = UnifiedParameterAnalyzer.analyze(my_instance) 

71 

72 # Instance analysis with exclusions (e.g., exclude 'func' from FunctionStep) 

73 param_info = UnifiedParameterAnalyzer.analyze(step_instance, exclude_params=['func']) 

74 """ 

75 if target is None: 75 ↛ 76line 75 didn't jump to line 76 because the condition on line 75 was never true

76 return {} 

77 

78 # Determine the type of target and route to appropriate analyzer 

79 if inspect.isfunction(target) or inspect.ismethod(target): 79 ↛ 81line 79 didn't jump to line 81 because the condition on line 79 was always true

80 result = UnifiedParameterAnalyzer._analyze_callable(target) 

81 elif inspect.isclass(target): 

82 if dataclasses.is_dataclass(target): 

83 result = UnifiedParameterAnalyzer._analyze_dataclass_type(target) 

84 else: 

85 # CRITICAL FIX: For classes, use _analyze_object_instance with use_signature_defaults=True 

86 # This traverses MRO to get all inherited parameters with signature defaults 

87 # Create a dummy instance just to get the class hierarchy analyzed 

88 try: 

89 dummy_instance = target.__new__(target) 

90 result = UnifiedParameterAnalyzer._analyze_object_instance(dummy_instance, use_signature_defaults=True) 

91 except: 

92 # If we can't create a dummy instance, fall back to just analyzing __init__ 

93 result = UnifiedParameterAnalyzer._analyze_callable(target.__init__) 

94 elif dataclasses.is_dataclass(target): 

95 # Instance of dataclass 

96 result = UnifiedParameterAnalyzer._analyze_dataclass_instance(target) 

97 else: 

98 # Try to analyze as callable 

99 if callable(target): 

100 result = UnifiedParameterAnalyzer._analyze_callable(target) 

101 else: 

102 # For regular object instances (like step instances), analyze their class constructor 

103 result = UnifiedParameterAnalyzer._analyze_object_instance(target) 

104 

105 # Apply exclusions if specified 

106 if exclude_params: 106 ↛ 107line 106 didn't jump to line 107 because the condition on line 106 was never true

107 result = {name: info for name, info in result.items() if name not in exclude_params} 

108 

109 return result 

110 

111 @staticmethod 

112 def _analyze_callable(callable_obj: Callable) -> Dict[str, UnifiedParameterInfo]: 

113 """Analyze a callable (function, method, etc.).""" 

114 # Use existing SignatureAnalyzer for callables 

115 param_info_dict = SignatureAnalyzer.analyze(callable_obj) 

116 

117 # Convert to unified format 

118 unified_params = {} 

119 for name, param_info in param_info_dict.items(): 

120 unified_params[name] = UnifiedParameterInfo.from_parameter_info( 

121 param_info, 

122 source_type="function" 

123 ) 

124 

125 return unified_params 

126 

127 @staticmethod 

128 def _analyze_dataclass_type(dataclass_type: Type) -> Dict[str, UnifiedParameterInfo]: 

129 """Analyze a dataclass type using existing SignatureAnalyzer infrastructure.""" 

130 # CRITICAL FIX: Use existing SignatureAnalyzer._analyze_dataclass method 

131 # which already handles all the docstring extraction properly 

132 param_info_dict = SignatureAnalyzer._analyze_dataclass(dataclass_type) 

133 

134 # Convert to unified format 

135 unified_params = {} 

136 for name, param_info in param_info_dict.items(): 

137 unified_params[name] = UnifiedParameterInfo.from_parameter_info( 

138 param_info, 

139 source_type="dataclass" 

140 ) 

141 

142 return unified_params 

143 

144 @staticmethod 

145 def _analyze_object_instance(instance: object, use_signature_defaults: bool = False) -> Dict[str, UnifiedParameterInfo]: 

146 """Analyze a regular object instance by examining its full inheritance hierarchy. 

147 

148 Args: 

149 instance: Object instance to analyze 

150 use_signature_defaults: If True, use signature defaults instead of instance values 

151 """ 

152 # Use MRO to get all constructor parameters from the inheritance chain 

153 instance_class = type(instance) 

154 all_params = {} 

155 

156 # Traverse MRO from most specific to most general (like dual-axis resolver) 

157 for cls in instance_class.__mro__: 

158 if cls == object: 

159 continue 

160 

161 # Skip classes without custom __init__ 

162 if not hasattr(cls, '__init__') or cls.__init__ == object.__init__: 

163 continue 

164 

165 try: 

166 # Analyze this class's constructor 

167 class_params = UnifiedParameterAnalyzer._analyze_callable(cls.__init__) 

168 

169 # Remove 'self' parameter 

170 if 'self' in class_params: 

171 del class_params['self'] 

172 

173 # Special handling for **kwargs - if we see 'kwargs', skip this class 

174 # and let parent classes provide the actual parameters 

175 if 'kwargs' in class_params and len(class_params) <= 2: 

176 # This class uses **kwargs, skip it and let parent classes define parameters 

177 continue 

178 

179 # Add parameters that haven't been seen yet (most specific wins) 

180 for param_name, param_info in class_params.items(): 

181 if param_name not in all_params and param_name != 'kwargs': 

182 # CRITICAL FIX: For reset functionality, use signature defaults instead of instance values 

183 if use_signature_defaults: 

184 default_value = param_info.default_value 

185 else: 

186 # Get current value from instance if it exists 

187 default_value = getattr(instance, param_name, param_info.default_value) 

188 

189 # Create parameter info with appropriate default value 

190 all_params[param_name] = UnifiedParameterInfo( 

191 name=param_name, 

192 param_type=param_info.param_type, 

193 default_value=default_value, 

194 is_required=param_info.is_required, 

195 description=param_info.description, # CRITICAL FIX: Include description 

196 source_type="object_instance" 

197 ) 

198 

199 except Exception: 

200 # Skip classes that can't be analyzed - this is legitimate since some classes 

201 # in MRO might not have analyzable constructors (e.g., ABC, object) 

202 continue 

203 

204 return all_params 

205 

206 @staticmethod 

207 def _analyze_dataclass_instance(instance: object) -> Dict[str, UnifiedParameterInfo]: 

208 """Analyze a dataclass instance.""" 

209 from openhcs.utils.performance_monitor import timer 

210 

211 # Get the type and analyze it 

212 with timer(f" Analyze dataclass type {type(instance).__name__}", threshold_ms=5.0): 

213 dataclass_type = type(instance) 

214 unified_params = UnifiedParameterAnalyzer._analyze_dataclass_type(dataclass_type) 

215 

216 # Check if this specific instance is a lazy config - if so, use raw field values 

217 with timer(" Check lazy config", threshold_ms=1.0): 

218 from openhcs.config_framework.lazy_factory import get_base_type_for_lazy 

219 # CRITICAL FIX: Don't check class name - PipelineConfig is lazy but doesn't start with "Lazy" 

220 # get_base_type_for_lazy() is the authoritative check for lazy dataclasses 

221 is_lazy_config = get_base_type_for_lazy(dataclass_type) is not None 

222 

223 # Update default values with current instance values 

224 with timer(f" Extract {len(unified_params)} field values from instance", threshold_ms=5.0): 

225 for name, param_info in unified_params.items(): 

226 if hasattr(instance, name): 

227 if is_lazy_config: 

228 # For lazy configs, get raw field value to avoid triggering resolution 

229 # Use object.__getattribute__() to bypass lazy property getters 

230 current_value = object.__getattribute__(instance, name) 

231 else: 

232 # For regular dataclasses, use normal getattr 

233 current_value = getattr(instance, name) 

234 

235 # Create new UnifiedParameterInfo with current value as default 

236 unified_params[name] = UnifiedParameterInfo( 

237 name=param_info.name, 

238 param_type=param_info.param_type, 

239 default_value=current_value, 

240 is_required=param_info.is_required, 

241 description=param_info.description, 

242 source_type="dataclass_instance" 

243 ) 

244 

245 return unified_params 

246 

247 @staticmethod 

248 def analyze_nested(target: Union[Callable, Type, object], parent_info: Dict[str, UnifiedParameterInfo] = None) -> Dict[str, UnifiedParameterInfo]: 

249 """Analyze parameters with nested dataclass support. 

250  

251 This method provides enhanced analysis that can handle nested dataclasses 

252 and maintain parent context information. 

253  

254 Args: 

255 target: The target to analyze 

256 parent_info: Optional parent parameter information for context 

257  

258 Returns: 

259 Dictionary of unified parameter information with nested support 

260 """ 

261 base_params = UnifiedParameterAnalyzer.analyze(target) 

262 

263 # For each parameter, check if it's a nested dataclass 

264 enhanced_params = {} 

265 for name, param_info in base_params.items(): 

266 enhanced_params[name] = param_info 

267 

268 # If this parameter is a dataclass, mark it as having nested structure 

269 if dataclasses.is_dataclass(param_info.param_type): 

270 # Update source type to indicate nesting capability 

271 enhanced_params[name] = UnifiedParameterInfo( 

272 name=param_info.name, 

273 param_type=param_info.param_type, 

274 default_value=param_info.default_value, 

275 is_required=param_info.is_required, 

276 description=param_info.description, 

277 source_type=f"{param_info.source_type}_nested" 

278 ) 

279 

280 return enhanced_params 

281 

282 

283# Backward compatibility aliases 

284# These allow existing code to continue working while migration happens 

285ParameterAnalyzer = UnifiedParameterAnalyzer 

286analyze_parameters = UnifiedParameterAnalyzer.analyze