Coverage for openhcs/textual_tui/widgets/shared/parameter_form_manager.py: 0.0%

1# File: openhcs/textual_tui/widgets/shared/parameter_form_manager.py 

2 

3import dataclasses 

4import ast 

5import logging 

6from enum import Enum 

7from typing import Any, Dict, get_origin, get_args, Union, Optional, Type 

8 

9logger = logging.getLogger(__name__) 

10from textual.containers import Vertical, Horizontal 

11from textual.widgets import Static, Button, Collapsible 

12from textual.app import ComposeResult 

13 

14from .typed_widget_factory import TypedWidgetFactory 

15from .signature_analyzer import SignatureAnalyzer 

16from .clickable_help_label import ClickableParameterLabel, HelpIndicator 

17from ..different_values_input import DifferentValuesInput 

18 

19# Import simplified abstraction layer 

20from openhcs.ui.shared.parameter_form_abstraction import ( 

21 ParameterFormAbstraction, apply_lazy_default_placeholder 

22) 

23from openhcs.ui.shared.widget_creation_registry import create_textual_registry 

24from openhcs.ui.shared.textual_widget_strategies import create_different_values_widget 

25 

26class ParameterFormManager: 

27 """Mathematical: (parameters, types, field_id) → parameter form""" 

28 

29 def __init__(self, parameters: Dict[str, Any], parameter_types: Dict[str, type], field_id: str, parameter_info: Dict = None, is_global_config_editing: bool = False, global_config_type: Optional[Type] = None, placeholder_prefix: str = "Pipeline default"): 

30 # Initialize simplified abstraction layer 

31 self.form_abstraction = ParameterFormAbstraction( 

32 parameters, parameter_types, field_id, create_textual_registry(), parameter_info 

33 ) 

34 

35 # Maintain backward compatibility 

36 self.parameters = parameters.copy() 

37 self.parameter_types = parameter_types 

38 self.field_id = field_id 

39 self.parameter_info = parameter_info or {} 

40 self.is_global_config_editing = is_global_config_editing 

41 self.global_config_type = global_config_type 

42 self.placeholder_prefix = placeholder_prefix 

43 

44 def build_form(self) -> ComposeResult: 

45 """Build parameter form - pure function with recursive dataclass support.""" 

46 with Vertical() as form: 

47 form.styles.height = "auto" 

48 

49 for param_name, param_type in self.parameter_types.items(): 

50 current_value = self.parameters[param_name] 

51 

52 # Handle Optional[dataclass] types with checkbox wrapper 

53 if self._is_optional_dataclass(param_type): 

54 inner_dataclass_type = self._get_optional_inner_type(param_type) 

55 yield from self._build_optional_dataclass_form(param_name, inner_dataclass_type, current_value) 

56 # Handle nested dataclasses recursively 

57 elif dataclasses.is_dataclass(param_type): 

58 yield from self._build_nested_dataclass_form(param_name, param_type, current_value) 

59 else: 

60 yield from self._build_regular_parameter_form(param_name, param_type, current_value) 

61 

62 def _build_nested_dataclass_form(self, param_name: str, param_type: type, current_value: Any) -> ComposeResult: 

63 """Build form for nested dataclass parameter.""" 

64 # Create collapsible widget (no ID - just structure) 

65 collapsible = TypedWidgetFactory.create_widget(param_type, current_value, None) 

66 

67 # Analyze nested dataclass 

68 nested_param_info = SignatureAnalyzer.analyze(param_type) 

69 

70 # Get current values from nested dataclass instance 

71 nested_parameters = {} 

72 nested_parameter_types = {} 

73 

74 for nested_name, nested_info in nested_param_info.items(): 

75 if current_value: 

76 # For lazy dataclasses, preserve None values for placeholder behavior 

77 if hasattr(current_value, '_resolve_field_value'): 

78 nested_current_value = object.__getattribute__(current_value, nested_name) if hasattr(current_value, nested_name) else nested_info.default_value 

79 else: 

80 nested_current_value = getattr(current_value, nested_name, nested_info.default_value) 

81 else: 

82 nested_current_value = nested_info.default_value 

83 nested_parameters[nested_name] = nested_current_value 

84 nested_parameter_types[nested_name] = nested_info.param_type 

85 

86 # Create nested form manager with hierarchical underscore notation and parameter info 

87 nested_field_id = f"{self.field_id}_{param_name}" 

88 nested_form_manager = ParameterFormManager(nested_parameters, nested_parameter_types, nested_field_id, nested_param_info) 

89 

90 # Store the parent dataclass type for proper lazy resolution detection 

91 nested_form_manager._parent_dataclass_type = param_type 

92 

93 # Store reference to nested form manager for updates 

94 if not hasattr(self, 'nested_managers'): 

95 self.nested_managers = {} 

96 self.nested_managers[param_name] = nested_form_manager 

97 

98 # Build nested form and add to collapsible 

99 with collapsible: 

100 yield from nested_form_manager.build_form() 

101 

102 yield collapsible 

103 

104 def _is_optional_dataclass(self, param_type: type) -> bool: 

105 """Check if parameter type is Optional[dataclass].""" 

106 from typing import get_origin, get_args, Union 

107 if get_origin(param_type) is Union: 

108 args = get_args(param_type) 

109 if len(args) == 2 and type(None) in args: 

110 inner_type = next(arg for arg in args if arg is not type(None)) 

111 return dataclasses.is_dataclass(inner_type) 

112 return False 

113 

114 def _get_optional_inner_type(self, param_type: type) -> type: 

115 """Extract the inner type from Optional[T].""" 

116 from typing import get_origin, get_args, Union 

117 if get_origin(param_type) is Union: 

118 args = get_args(param_type) 

119 if len(args) == 2 and type(None) in args: 

120 return next(arg for arg in args if arg is not type(None)) 

121 return param_type 

122 

123 def _build_optional_dataclass_form(self, param_name: str, dataclass_type: type, current_value: Any) -> ComposeResult: 

124 """Build form for Optional[dataclass] parameter with checkbox toggle.""" 

125 from textual.widgets import Checkbox 

126 

127 # Checkbox 

128 checkbox_id = f"{self.field_id}_{param_name}_enabled" 

129 checkbox = Checkbox( 

130 value=current_value is not None, 

131 label=f"Enable {param_name.replace('_', ' ').title()}", 

132 id=checkbox_id, 

133 compact=True 

134 ) 

135 yield checkbox 

136 

137 # Collapsible dataclass widget 

138 collapsible = TypedWidgetFactory.create_widget(dataclass_type, current_value, None) 

139 collapsible.collapsed = (current_value is None) 

140 

141 # Setup nested form 

142 nested_param_info = SignatureAnalyzer.analyze(dataclass_type) 

143 nested_parameters = {} 

144 for name, info in nested_param_info.items(): 

145 if current_value: 

146 # For lazy dataclasses, preserve None values for placeholder behavior 

147 if hasattr(current_value, '_resolve_field_value'): 

148 value = object.__getattribute__(current_value, name) if hasattr(current_value, name) else info.default_value 

149 else: 

150 value = getattr(current_value, name, info.default_value) 

151 else: 

152 value = info.default_value 

153 nested_parameters[name] = value 

154 nested_parameter_types = {name: info.param_type for name, info in nested_param_info.items()} 

155 

156 nested_form_manager = ParameterFormManager( 

157 nested_parameters, nested_parameter_types, f"{self.field_id}_{param_name}", nested_param_info, 

158 is_global_config_editing=self.is_global_config_editing 

159 ) 

160 

161 # Store the parent dataclass type for proper lazy resolution detection 

162 nested_form_manager._parent_dataclass_type = dataclass_type 

163 

164 # Store references 

165 if not hasattr(self, 'nested_managers'): 

166 self.nested_managers = {} 

167 if not hasattr(self, 'optional_checkboxes'): 

168 self.optional_checkboxes = {} 

169 self.nested_managers[param_name] = nested_form_manager 

170 self.optional_checkboxes[param_name] = checkbox 

171 

172 with collapsible: 

173 yield from nested_form_manager.build_form() 

174 yield collapsible 

175 

176 def _build_regular_parameter_form(self, param_name: str, param_type: type, current_value: Any) -> ComposeResult: 

177 """Build form for regular (non-dataclass) parameter.""" 

178 # Check if this field has different values across orchestrators 

179 config_analysis = getattr(self, 'config_analysis', {}) 

180 field_analysis = config_analysis.get(param_name, {}) 

181 

182 # Create widget using hierarchical underscore notation 

183 widget_id = f"{self.field_id}_{param_name}" 

184 

185 # Handle different values or create normal widget 

186 if field_analysis.get('type') == 'different': 

187 default_value = field_analysis.get('default') 

188 input_widget = create_different_values_widget(param_name, param_type, default_value, widget_id) 

189 else: 

190 # Use registry for widget creation and apply placeholder 

191 widget_value = current_value.value if hasattr(current_value, 'value') else current_value 

192 input_widget = self.form_abstraction.create_widget_for_parameter(param_name, param_type, widget_value) 

193 apply_lazy_default_placeholder(input_widget, param_name, current_value, self.parameter_types, 'textual', 

194 is_global_config_editing=self.is_global_config_editing, 

195 global_config_type=self.global_config_type, 

196 placeholder_prefix=self.placeholder_prefix) 

197 

198 # Get parameter info for help functionality 

199 param_info = self._get_parameter_info(param_name) 

200 param_description = param_info.description if param_info else None 

201 

202 # 3-column layout: label + input + reset 

203 with Horizontal() as row: 

204 row.styles.height = "auto" 

205 

206 # Parameter label with help (auto width - sizes to content) 

207 # Always use clickable label with help - provide default description if none exists 

208 description = param_description or f"Parameter: {param_name.replace('_', ' ')}" 

209 label = ClickableParameterLabel( 

210 param_name, 

211 description, 

212 param_type, 

213 classes="param-label clickable" 

214 ) 

215 

216 label.styles.width = "auto" 

217 label.styles.text_align = "left" 

218 label.styles.height = "1" 

219 yield label 

220 

221 # Input widget (flexible width, left aligned, with left margin for spacing) 

222 input_widget.styles.width = "1fr" 

223 input_widget.styles.text_align = "left" 

224 input_widget.styles.margin = (0, 0, 0, 1) # top, right, bottom, left margin 

225 yield input_widget 

226 

227 # Reset button (auto width) 

228 reset_btn = Button("Reset", id=f"reset_{widget_id}", compact=True) 

229 reset_btn.styles.width = "auto" 

230 yield reset_btn 

231 

232 def update_parameter(self, param_name: str, value: Any): 

233 """Update parameter value with centralized enum conversion and nested dataclass support.""" 

234 # Debug: Check if None values are being received and processed (path_planning only) 

235 if param_name == 'output_dir_suffix' or param_name == 'path_planning': 

236 logger.info(f"*** TEXTUAL UPDATE DEBUG *** {param_name} update_parameter called with: {value} (type: {type(value)})") 

237 if param_name == 'path_planning': 

238 import traceback 

239 logger.info(f"*** PATH_PLANNING SOURCE *** Call stack:") 

240 for line in traceback.format_stack()[-5:]: 

241 logger.info(f"*** PATH_PLANNING SOURCE *** {line.strip()}") 

242 # Parse hierarchical parameter name (e.g., "path_planning_global_output_folder") 

243 # Split and check if this is a nested parameter 

244 parts = param_name.split('_') 

245 if len(parts) >= 2: # nested_field format 

246 # Try to find nested manager by checking all possible prefixes 

247 for i in range(1, len(parts)): 

248 potential_nested = '_'.join(parts[:i]) 

249 if potential_nested in self.parameters and hasattr(self, 'nested_managers') and potential_nested in self.nested_managers: 

250 # Reconstruct the nested field name from remaining parts 

251 nested_field = '_'.join(parts[i:]) 

252 

253 # Update nested form manager 

254 if potential_nested == 'path_planning': 

255 logger.info(f"*** NESTED MANAGER UPDATE *** Updating {potential_nested}.{nested_field} = {value}") 

256 self.nested_managers[potential_nested].update_parameter(nested_field, value) 

257 

258 # Rebuild nested dataclass instance with lazy/concrete mixed behavior 

259 nested_values = self.nested_managers[potential_nested].get_current_values() 

260 

261 # Debug: Check what values the nested manager is returning 

262 if potential_nested == 'path_planning': 

263 logger.info(f"*** NESTED VALUES DEBUG *** nested_values from {potential_nested}: {nested_values}") 

264 if 'output_dir_suffix' in nested_values: 

265 logger.info(f"*** NESTED VALUES DEBUG *** output_dir_suffix in nested_values: {nested_values['output_dir_suffix']} (type: {type(nested_values['output_dir_suffix'])})") 

266 

267 # Also check what's in the nested manager's parameters directly 

268 nested_params = self.nested_managers[potential_nested].parameters 

269 logger.info(f"*** NESTED VALUES DEBUG *** nested_manager.parameters: {nested_params}") 

270 if 'output_dir_suffix' in nested_params: 

271 logger.info(f"*** NESTED VALUES DEBUG *** output_dir_suffix in nested_manager.parameters: {nested_params['output_dir_suffix']} (type: {type(nested_params['output_dir_suffix'])})") 

272 

273 nested_type = self.parameter_types[potential_nested] 

274 

275 # Resolve Union types (like Optional[DataClass]) to the actual dataclass type 

276 if self._is_optional_dataclass(nested_type): 

277 nested_type = self._get_optional_inner_type(nested_type) 

278 

279 # Create lazy dataclass instance with mixed concrete/lazy fields 

280 if self.is_global_config_editing: 

281 # Global config editing: use concrete dataclass 

282 self.parameters[potential_nested] = nested_type(**nested_values) 

283 else: 

284 # Lazy context: always create lazy instance for thread-local resolution 

285 # Even if all values are None (especially after reset), we want lazy resolution 

286 from openhcs.core.lazy_config import LazyDataclassFactory 

287 

288 # Determine the correct field path using type inspection 

289 field_path = self._get_field_path_for_nested_type(nested_type) 

290 

291 lazy_nested_type = LazyDataclassFactory.make_lazy_thread_local( 

292 base_class=nested_type, 

293 field_path=field_path, 

294 lazy_class_name=f"Mixed{nested_type.__name__}" 

295 ) 

296 # Pass ALL fields: concrete values for edited fields, None for lazy resolution 

297 self.parameters[potential_nested] = lazy_nested_type(**nested_values) 

298 return 

299 

300 # Handle regular parameters (direct match) 

301 if param_name in self.parameters: 

302 # Handle literal "None" string - convert back to Python None 

303 if isinstance(value, str) and value == "None": 

304 value = None 

305 

306 # Convert string back to proper type (comprehensive conversion) 

307 # Skip type conversion for None values (preserve for lazy placeholder behavior) 

308 if param_name in self.parameter_types and value is not None: 

309 param_type = self.parameter_types[param_name] 

310 if hasattr(param_type, '__bases__') and Enum in param_type.__bases__: 

311 value = param_type(value) # Convert string → enum 

312 elif self._is_list_of_enums(param_type): 

313 # Handle List[Enum] types (like List[VariableComponents]) 

314 enum_type = self._get_enum_from_list(param_type) 

315 if enum_type: 

316 # Convert string value to enum, then wrap in list 

317 enum_value = enum_type(value) 

318 value = [enum_value] 

319 elif param_type == float: 

320 # Convert string → float, handle empty based on parameter requirements 

321 try: 

322 if value == "": 

323 # For empty values, we need to check if parameter is required 

324 # This requires access to parameter info, but we don't have it here 

325 # For now, convert empty to None (safer than 0.0) 

326 value = None 

327 else: 

328 value = float(value) 

329 except (ValueError, TypeError): 

330 value = None # Use None instead of 0.0 for failed conversions 

331 elif param_type == int: 

332 # Convert string → int, handle empty based on parameter requirements 

333 try: 

334 if value == "": 

335 # For empty values, convert to None (safer than 0) 

336 value = None 

337 else: 

338 value = int(value) 

339 except (ValueError, TypeError): 

340 value = None # Use None instead of 0 for failed conversions 

341 elif param_type == bool: 

342 # Convert string → bool 

343 if isinstance(value, str): 

344 value = value.lower() in ('true', '1', 'yes', 'on') 

345 # Add more type conversions as needed 

346 

347 self.parameters[param_name] = value 

348 

349 # FALLBACK: If this is a nested field that bypassed the nested logic, update the nested manager 

350 if param_name == 'output_dir_suffix': 

351 logger.info(f"*** FALLBACK DEBUG *** Checking fallback for {param_name}") 

352 logger.info(f"*** FALLBACK DEBUG *** hasattr nested_managers: {hasattr(self, 'nested_managers')}") 

353 if hasattr(self, 'nested_managers'): 

354 logger.info(f"*** FALLBACK DEBUG *** nested_managers keys: {list(self.nested_managers.keys())}") 

355 for nested_name, nested_manager in self.nested_managers.items(): 

356 logger.info(f"*** FALLBACK DEBUG *** Checking {nested_name}, parameter_types: {list(nested_manager.parameter_types.keys())}") 

357 if param_name in nested_manager.parameter_types: 

358 logger.info(f"*** FALLBACK UPDATE *** Updating nested manager {nested_name}.{param_name} = {value}") 

359 nested_manager.parameters[param_name] = value 

360 break 

361 else: 

362 logger.info(f"*** FALLBACK DEBUG *** {param_name} not found in {nested_name}") 

363 else: 

364 logger.info(f"*** FALLBACK DEBUG *** No nested_managers attribute") 

365 elif hasattr(self, 'nested_managers'): 

366 for nested_name, nested_manager in self.nested_managers.items(): 

367 if param_name in nested_manager.parameter_types: 

368 nested_manager.parameters[param_name] = value 

369 break 

370 

371 # Debug: Check what was actually stored (path_planning only) 

372 if param_name == 'output_dir_suffix' or param_name == 'path_planning': 

373 stored_value = self.parameters.get(param_name) 

374 logger.info(f"*** TEXTUAL UPDATE DEBUG *** {param_name} stored as: {stored_value} (type: {type(stored_value)})") 

375 

376 def reset_parameter(self, param_name: str, default_value: Any = None): 

377 """Reset parameter to appropriate default value based on lazy vs concrete dataclass context.""" 

378 # Determine the correct reset value if not provided 

379 if default_value is None: 

380 default_value = self._get_reset_value_for_parameter(param_name) 

381 

382 # Parse hierarchical parameter name for nested parameters 

383 parts = param_name.split('_') 

384 if len(parts) >= 2: # nested_field format 

385 # Try to find nested manager by checking all possible prefixes 

386 for i in range(1, len(parts)): 

387 potential_nested = '_'.join(parts[:i]) 

388 if potential_nested in self.parameters and hasattr(self, 'nested_managers') and potential_nested in self.nested_managers: 

389 # Reconstruct the nested field name 

390 nested_field = '_'.join(parts[i:]) 

391 

392 # Get appropriate reset value for nested field 

393 nested_reset_value = self._get_reset_value_for_nested_parameter(potential_nested, nested_field) 

394 

395 # Reset in nested form manager 

396 self.nested_managers[potential_nested].reset_parameter(nested_field, nested_reset_value) 

397 

398 # Rebuild nested dataclass instance 

399 nested_values = self.nested_managers[potential_nested].get_current_values() 

400 

401 # Resolve Union types (like Optional[DataClass]) to the actual dataclass type 

402 if self._is_optional_dataclass(self.parameter_types[potential_nested]): 

403 nested_type = self._get_optional_inner_type(self.parameter_types[potential_nested]) 

404 else: 

405 nested_type = self.parameter_types[potential_nested] 

406 

407 # Create lazy dataclass instance with mixed concrete/lazy fields 

408 if self.is_global_config_editing: 

409 # Global config editing: use concrete dataclass 

410 self.parameters[potential_nested] = nested_type(**nested_values) 

411 else: 

412 # Lazy context: always create lazy instance for thread-local resolution 

413 # Even if all values are None (especially after reset), we want lazy resolution 

414 from openhcs.core.lazy_config import LazyDataclassFactory 

415 

416 # Determine the correct field path using type inspection 

417 field_path = self._get_field_path_for_nested_type(nested_type) 

418 

419 lazy_nested_type = LazyDataclassFactory.make_lazy_thread_local( 

420 base_class=nested_type, 

421 field_path=field_path, 

422 lazy_class_name=f"Mixed{nested_type.__name__}" 

423 ) 

424 # Pass ALL fields: concrete values for edited fields, None for lazy resolution 

425 self.parameters[potential_nested] = lazy_nested_type(**nested_values) 

426 return 

427 

428 # Handle regular parameters 

429 if param_name in self.parameters: 

430 self.parameters[param_name] = default_value 

431 

432 # Handle special reset behavior for DifferentValuesInput widgets 

433 self._handle_different_values_reset(param_name) 

434 

435 # Re-apply placeholder styling if value is None (for reset functionality) 

436 if default_value is None: 

437 self._reapply_placeholder_if_needed(param_name) 

438 

439 def _reapply_placeholder_if_needed(self, param_name: str): 

440 """Re-apply placeholder styling to a widget when its value is set to None.""" 

441 # For Textual, we need to find the widget and re-apply placeholder 

442 # This is more complex than PyQt since Textual widgets are reactive 

443 # For now, we'll rely on the reactive nature of Textual widgets 

444 # The placeholder should be re-applied automatically when the value changes to None 

445 pass 

446 

447 def _get_reset_value_for_parameter(self, param_name: str) -> Any: 

448 """ 

449 Get the appropriate reset value for a parameter based on lazy vs concrete dataclass context. 

450 

451 For concrete dataclasses (like GlobalPipelineConfig): 

452 - Reset to static class defaults 

453 

454 For lazy dataclasses (like PipelineConfig for orchestrator configs): 

455 - Reset to None to preserve placeholder behavior and inheritance hierarchy 

456 """ 

457 if param_name not in self.parameter_info: 

458 return None 

459 

460 param_info = self.parameter_info[param_name] 

461 param_type = self.parameter_types[param_name] 

462 

463 # For global config editing, always use static defaults 

464 if self.is_global_config_editing: 

465 return param_info.default_value 

466 

467 # For nested dataclass fields, check if we should use concrete values 

468 if hasattr(param_type, '__dataclass_fields__'): 

469 # This is a dataclass field - determine if it should be concrete or None 

470 current_value = self.parameters.get(param_name) 

471 if self._should_use_concrete_nested_values(current_value): 

472 # Use static default for concrete nested dataclass 

473 return param_info.default_value 

474 else: 

475 # Use None for lazy nested dataclass to preserve placeholder behavior 

476 return None 

477 

478 # For non-dataclass fields in lazy context, use None to preserve placeholder behavior 

479 # This allows the field to inherit from the parent config hierarchy 

480 if not self.is_global_config_editing: 

481 return None 

482 

483 # Fallback to static default 

484 return param_info.default_value 

485 

486 def _get_reset_value_for_nested_parameter(self, nested_param_name: str, nested_field_name: str) -> Any: 

487 """Get appropriate reset value for a nested parameter field.""" 

488 nested_type = self.parameter_types[nested_param_name] 

489 nested_param_info = SignatureAnalyzer.analyze(nested_type) 

490 

491 if nested_field_name not in nested_param_info: 

492 return None 

493 

494 nested_field_info = nested_param_info[nested_field_name] 

495 

496 # For global config editing, always use static defaults 

497 if self.is_global_config_editing: 

498 return nested_field_info.default_value 

499 

500 # For lazy context, check if nested dataclass should use concrete values 

501 current_nested_value = self.parameters.get(nested_param_name) 

502 if self._should_use_concrete_nested_values(current_nested_value): 

503 return nested_field_info.default_value 

504 else: 

505 return None 

506 

507 def _get_field_path_for_nested_type(self, nested_type: Type) -> Optional[str]: 

508 """ 

509 Automatically determine the field path for a nested dataclass type using type inspection. 

510 

511 This method examines the GlobalPipelineConfig fields and their type annotations 

512 to find which field corresponds to the given nested_type. This eliminates the need 

513 for hardcoded string mappings and automatically works with new nested dataclass fields. 

514 

515 Args: 

516 nested_type: The dataclass type to find the field path for 

517 

518 Returns: 

519 The field path string (e.g., 'path_planning', 'vfs') or None if not found 

520 """ 

521 try: 

522 from openhcs.core.config import GlobalPipelineConfig 

523 from dataclasses import fields 

524 import typing 

525 

526 # Get all fields from GlobalPipelineConfig 

527 global_config_fields = fields(GlobalPipelineConfig) 

528 

529 for field in global_config_fields: 

530 field_type = field.type 

531 

532 # Handle Optional types (Union[Type, None]) 

533 if hasattr(typing, 'get_origin') and typing.get_origin(field_type) is typing.Union: 

534 # Get the non-None type from Optional[Type] 

535 args = typing.get_args(field_type) 

536 if len(args) == 2 and type(None) in args: 

537 field_type = args[0] if args[1] is type(None) else args[1] 

538 

539 # Check if the field type matches our nested type 

540 if field_type == nested_type: 

541 return field.name 

542 

543 

544 

545 return None 

546 

547 except Exception as e: 

548 # Fallback to None if type inspection fails 

549 import logging 

550 logger = logging.getLogger(__name__) 

551 logger.debug(f"Failed to determine field path for {nested_type.__name__}: {e}") 

552 return None 

553 

554 def _should_use_concrete_nested_values(self, current_value: Any) -> bool: 

555 """ 

556 Determine if nested dataclass fields should use concrete values or None for placeholders. 

557 This mirrors the logic from the PyQt form manager. 

558 

559 Returns True if: 

560 1. Global config editing (always concrete) 

561 2. Regular concrete dataclass (always concrete) 

562 

563 Returns False if: 

564 1. Lazy dataclass (supports mixed lazy/concrete states per field) 

565 2. None values (show placeholders) 

566 

567 Note: This method now supports mixed states within nested dataclasses. 

568 Individual fields can be lazy (None) or concrete within the same dataclass. 

569 """ 

570 # Global config editing always uses concrete values 

571 if self.is_global_config_editing: 

572 return True 

573 

574 # If current_value is None, use placeholders 

575 if current_value is None: 

576 return False 

577 

578 # If current_value is a concrete dataclass instance, use its values 

579 if hasattr(current_value, '__dataclass_fields__') and not hasattr(current_value, '_resolve_field_value'): 

580 return True 

581 

582 # For lazy dataclasses, always return False to enable mixed lazy/concrete behavior 

583 # Individual field values will be checked separately in the nested form creation 

584 if hasattr(current_value, '_resolve_field_value'): 

585 return False 

586 

587 # Default to placeholder behavior for lazy contexts 

588 return False 

589 

590 def handle_optional_checkbox_change(self, param_name: str, enabled: bool): 

591 """Handle checkbox change for Optional[dataclass] parameters.""" 

592 if param_name in self.parameter_types and self._is_optional_dataclass(self.parameter_types[param_name]): 

593 dataclass_type = self._get_optional_inner_type(self.parameter_types[param_name]) 

594 nested_managers = getattr(self, 'nested_managers', {}) 

595 self.parameters[param_name] = ( 

596 dataclass_type(**nested_managers[param_name].get_current_values()) 

597 if enabled and param_name in nested_managers 

598 else dataclass_type() if enabled 

599 else None 

600 ) 

601 

602 def _handle_different_values_reset(self, param_name: str): 

603 """Handle reset behavior for DifferentValuesInput widgets.""" 

604 # Check if this field has different values across orchestrators 

605 config_analysis = getattr(self, 'config_analysis', {}) 

606 field_analysis = config_analysis.get(param_name, {}) 

607 

608 if field_analysis.get('type') == 'different': 

609 # For different values fields, reset means go back to "DIFFERENT VALUES" state 

610 # We need to find the widget and call its reset method 

611 widget_id = f"{self.field_id}_{param_name}" 

612 

613 # This will be handled by the screen/container that manages the widgets 

614 # The widget itself will handle the reset via its reset_to_different() method 

615 # We just need to ensure the parameter value reflects the "different" state 

616 pass # Widget-level reset will be handled by the containing screen 

617 

618 def reset_all_parameters(self, defaults: Dict[str, Any] = None): 

619 """Reset all parameters to appropriate defaults based on lazy vs concrete dataclass context.""" 

620 # If no defaults provided, generate them based on context 

621 if defaults is None: 

622 defaults = {} 

623 for param_name in self.parameters.keys(): 

624 defaults[param_name] = self._get_reset_value_for_parameter(param_name) 

625 

626 for param_name, default_value in defaults.items(): 

627 if param_name in self.parameters: 

628 # Handle nested dataclasses 

629 if dataclasses.is_dataclass(self.parameter_types.get(param_name)): 

630 if hasattr(self, 'nested_managers') and param_name in self.nested_managers: 

631 # Generate appropriate reset values for nested parameters 

632 nested_type = self.parameter_types[param_name] 

633 nested_param_info = SignatureAnalyzer.analyze(nested_type) 

634 

635 # Use lazy-aware reset logic for nested parameters with mixed state support 

636 nested_defaults = {} 

637 for nested_field_name in nested_param_info.keys(): 

638 # For nested fields in lazy contexts, always reset to None to preserve lazy behavior 

639 # This ensures individual fields can maintain placeholder behavior regardless of other field states 

640 if not self.is_global_config_editing: 

641 nested_defaults[nested_field_name] = None 

642 else: 

643 nested_defaults[nested_field_name] = self._get_reset_value_for_nested_parameter(param_name, nested_field_name) 

644 

645 self.nested_managers[param_name].reset_all_parameters(nested_defaults) 

646 

647 # Rebuild nested dataclass instance 

648 nested_values = self.nested_managers[param_name].get_current_values() 

649 

650 # Resolve Union types (like Optional[DataClass]) to the actual dataclass type 

651 if self._is_optional_dataclass(nested_type): 

652 nested_type = self._get_optional_inner_type(nested_type) 

653 

654 # Create lazy dataclass instance with mixed concrete/lazy fields 

655 if self.is_global_config_editing: 

656 # Global config editing: use concrete dataclass 

657 self.parameters[param_name] = nested_type(**nested_values) 

658 else: 

659 # Lazy context: always create lazy instance for thread-local resolution 

660 # Even if all values are None (especially after reset), we want lazy resolution 

661 from openhcs.core.lazy_config import LazyDataclassFactory 

662 

663 # Determine the correct field path using type inspection 

664 field_path = self._get_field_path_for_nested_type(nested_type) 

665 

666 lazy_nested_type = LazyDataclassFactory.make_lazy_thread_local( 

667 base_class=nested_type, 

668 field_path=field_path, 

669 lazy_class_name=f"Mixed{nested_type.__name__}" 

670 ) 

671 # Pass ALL fields: concrete values for edited fields, None for lazy resolution 

672 self.parameters[param_name] = lazy_nested_type(**nested_values) 

673 else: 

674 self.parameters[param_name] = default_value 

675 else: 

676 self.parameters[param_name] = default_value 

677 

678 def reset_parameter_by_path(self, parameter_path: str): 

679 """Reset a parameter by its full path (supports nested parameters). 

680 

681 Args: 

682 parameter_path: Either a simple parameter name (e.g., 'num_workers') 

683 or a nested path (e.g., 'path_planning.output_dir_suffix') 

684 """ 

685 if '.' in parameter_path: 

686 # Handle nested parameter 

687 parts = parameter_path.split('.', 1) 

688 nested_name = parts[0] 

689 nested_param = parts[1] 

690 

691 if hasattr(self, 'nested_managers') and nested_name in self.nested_managers: 

692 nested_manager = self.nested_managers[nested_name] 

693 if '.' in nested_param: 

694 # Further nesting 

695 nested_manager.reset_parameter_by_path(nested_param) 

696 else: 

697 # Direct nested parameter 

698 nested_manager.reset_parameter(nested_param) 

699 else: 

700 logger.warning(f"Nested manager '{nested_name}' not found for parameter path '{parameter_path}'") 

701 else: 

702 # Handle top-level parameter 

703 self.reset_parameter(parameter_path) 

704 

705 def _is_list_of_enums(self, param_type) -> bool: 

706 """Check if parameter type is List[Enum].""" 

707 try: 

708 # Check if it's a generic type (like List[Something]) 

709 origin = get_origin(param_type) 

710 if origin is list: 

711 # Get the type arguments (e.g., VariableComponents from List[VariableComponents]) 

712 args = get_args(param_type) 

713 if args and len(args) > 0: 

714 inner_type = args[0] 

715 # Check if the inner type is an enum 

716 return hasattr(inner_type, '__bases__') and Enum in inner_type.__bases__ 

717 return False 

718 except Exception: 

719 return False 

720 

721 def _get_enum_from_list(self, param_type): 

722 """Extract enum type from List[Enum] type.""" 

723 return self._get_enum_from_list_static(param_type) 

724 

725 @staticmethod 

726 def _is_list_of_enums_static(param_type) -> bool: 

727 """Static version of _is_list_of_enums for use in convert_string_to_type.""" 

728 try: 

729 # Check if it's a generic type (like List[Something]) 

730 origin = get_origin(param_type) 

731 if origin is list: 

732 # Get the type arguments (e.g., VariableComponents from List[VariableComponents]) 

733 args = get_args(param_type) 

734 if args and len(args) > 0: 

735 inner_type = args[0] 

736 # Check if the inner type is an enum 

737 return hasattr(inner_type, '__bases__') and Enum in inner_type.__bases__ 

738 return False 

739 except Exception: 

740 return False 

741 

742 @staticmethod 

743 def _get_enum_from_list_static(param_type): 

744 """Static version of _get_enum_from_list for use in convert_string_to_type.""" 

745 try: 

746 args = get_args(param_type) 

747 if args and len(args) > 0: 

748 return args[0] # Return the enum type (e.g., VariableComponents) 

749 return None 

750 except Exception: 

751 return None 

752 

753 def get_current_values(self) -> Dict[str, Any]: 

754 """Get current parameter values.""" 

755 return self.parameters.copy() 

756 

757 def _get_parameter_info(self, param_name: str): 

758 """Get parameter info for help functionality.""" 

759 return self.parameter_info.get(param_name) 

760 

761 # Old placeholder methods removed - now using centralized abstraction layer 

762 

763 @staticmethod 

764 def convert_string_to_type(string_value: str, param_type: type, strict: bool = False) -> Any: 

765 """ 

766 Convert string input to expected type using existing type conversion logic. 

767 

768 Args: 

769 string_value: The string value from user input 

770 param_type: The expected type from function signature 

771 strict: If True, raise errors on conversion failure. If False, return None. 

772 

773 Returns: 

774 Converted value of the expected type 

775 

776 Raises: 

777 ValueError: If strict=True and conversion fails with specific error message 

778 """ 

779 # Handle empty/None values - let compiler validate if required 

780 if string_value == "" or string_value is None: 

781 return None 

782 

783 try: 

784 # Handle Union types (like Optional[List[float]] which is Union[List[float], None]) 

785 origin = get_origin(param_type) 

786 if origin is Union: 

787 # Try each type in the Union until one works 

788 union_args = get_args(param_type) 

789 last_error = None 

790 

791 for union_type in union_args: 

792 # Skip NoneType - we handle None separately 

793 if union_type is type(None): 

794 continue 

795 

796 try: 

797 # Recursively try to convert to this union member type 

798 return ParameterFormManager.convert_string_to_type(string_value, union_type, strict=True) 

799 except (ValueError, TypeError, SyntaxError) as e: 

800 last_error = e 

801 continue 

802 

803 # If no union type worked, raise the last error 

804 if last_error: 

805 raise last_error 

806 else: 

807 raise ValueError(f"No valid conversion found for Union type {param_type}") 

808 

809 # Use existing type conversion logic from update_parameter 

810 elif hasattr(param_type, '__bases__') and Enum in param_type.__bases__: 

811 return param_type(string_value) # Convert string → enum 

812 elif ParameterFormManager._is_list_of_enums_static(param_type): 

813 # Handle List[Enum] types (like List[VariableComponents]) 

814 enum_type = ParameterFormManager._get_enum_from_list_static(param_type) 

815 if enum_type: 

816 # Convert string value to enum, then wrap in list 

817 enum_value = enum_type(string_value) 

818 return [enum_value]