Coverage for openhcs/textual_tui/widgets/config_form.py: 0.0%

1"""Config form widget with reactive properties.""" 

2 

3from typing import Dict, Any 

4from textual.containers import ScrollableContainer 

5from textual.widgets import Static 

6from textual.app import ComposeResult 

7from textual.reactive import reactive 

8 

9 

10from .shared.parameter_form_manager import ParameterFormManager 

11from openhcs.introspection.signature_analyzer import SignatureAnalyzer 

12 

13 

14class ConfigFormWidget(ScrollableContainer): 

15 """Reactive form widget for config editing.""" 

16 

17 field_values = reactive(dict, recompose=False) # Prevent automatic recomposition during typing 

18 

19 

20 def __init__(self, dataclass_type: type, instance: Any = None, is_global_config_editing: bool = False, **kwargs): 

21 super().__init__(**kwargs) 

22 self.dataclass_type = dataclass_type 

23 self.instance = instance or dataclass_type() 

24 

25 # Analyze dataclass using unified parameter analysis 

26 param_info = SignatureAnalyzer.analyze(dataclass_type) 

27 

28 # Convert to form manager format 

29 parameters = {} 

30 parameter_types = {} 

31 param_defaults = {} 

32 

33 for name, info in param_info.items(): 

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

35 if hasattr(self.instance, '_resolve_field_value'): 

36 # This is a lazy dataclass - use object.__getattribute__ to get stored value 

37 current_value = object.__getattribute__(self.instance, name) if hasattr(self.instance, name) else info.default_value 

38 else: 

39 # Regular dataclass - use normal getattr 

40 current_value = getattr(self.instance, name, info.default_value) 

41 parameters[name] = current_value 

42 parameter_types[name] = info.param_type 

43 param_defaults[name] = info.default_value 

44 

45 # Create shared form manager with parameter info for help functionality 

46 # CRITICAL FIX: Use dataclass type name as field_id for root config (not artificial "config") 

47 root_field_id = dataclass_type.__name__ # e.g., "GlobalPipelineConfig" or "PipelineConfig" 

48 self.form_manager = ParameterFormManager(parameters, parameter_types, root_field_id, param_info, is_global_config_editing=is_global_config_editing) 

49 self.param_defaults = param_defaults 

50 

51 # Initialize field values for reactive updates 

52 self.field_values = parameters.copy() 

53 

54 @classmethod 

55 def from_dataclass(cls, dataclass_type: type, instance: Any = None, is_global_config_editing: bool = False, **kwargs): 

56 """Create ConfigFormWidget from dataclass type and instance.""" 

57 return cls(dataclass_type, instance, is_global_config_editing=is_global_config_editing, **kwargs) 

58 

59 def compose(self) -> ComposeResult: 

60 """Compose the config form using shared form manager.""" 

61 try: 

62 # Use shared form manager to build form 

63 yield from self.form_manager.build_form() 

64 except Exception as e: 

65 yield Static(f"[red]Error building config form: {e}[/red]") 

66 

67 def on_mount(self) -> None: 

68 """Called when the form is mounted - fix scroll position to top.""" 

69 # Force scroll to top after mounting to prevent automatic scrolling to bottom 

70 self.call_after_refresh(self._fix_scroll_position) 

71 

72 def _fix_scroll_position(self) -> None: 

73 """Fix scroll position to top of form.""" 

74 try: 

75 # Force scroll to top (0, 0) 

76 self.scroll_to(0, 0, animate=False) 

77 except Exception: 

78 # If anything goes wrong, just continue 

79 pass 

80 

81 

82 

83 def _on_field_change(self, field_name: str, value: Any) -> None: 

84 """Handle field value changes.""" 

85 if self.form_manager: 

86 self.form_manager.update_parameter(field_name, value) 

87 # Update internal field values without triggering reactive update 

88 # This prevents recomposition and focus loss during typing 

89 if not hasattr(self, '_internal_field_values'): 

90 self._internal_field_values = self.field_values.copy() 

91 

92 # For nested parameters like "path_planning_output_dir_suffix", 

93 # update the top-level "path_planning" parameter 

94 parts = field_name.split('_') 

95 if len(parts) >= 2: 

96 top_level_param = parts[0] 

97 if top_level_param in self.form_manager.parameters: 

98 self._internal_field_values[top_level_param] = self.form_manager.parameters[top_level_param] 

99 else: 

100 # Regular parameter 

101 if field_name in self.form_manager.parameters: 

102 self._internal_field_values[field_name] = self.form_manager.parameters[field_name] 

103 

104 def on_input_changed(self, event) -> None: 

105 """Handle input changes from shared components.""" 

106 if event.input.id.startswith("config_"): 

107 field_name = event.input.id.split("_", 1)[1] 

108 self._on_field_change(field_name, event.value) 

109 

110 def on_checkbox_changed(self, event) -> None: 

111 """Handle checkbox changes from shared components.""" 

112 if event.checkbox.id.startswith("config_"): 

113 field_name = event.checkbox.id.split("_", 1)[1] 

114 self._on_field_change(field_name, event.value) 

115 

116 def on_radio_set_changed(self, event) -> None: 

117 """Handle RadioSet changes from shared components.""" 

118 if event.radio_set.id.startswith("config_"): 

119 field_name = event.radio_set.id.split("_", 1)[1] 

120 if event.pressed and event.pressed.id: 

121 enum_value = event.pressed.id[5:] # Remove "enum_" prefix 

122 self._on_field_change(field_name, enum_value) 

123 

124 def on_button_pressed(self, event) -> None: 

125 """Handle reset button presses from shared components.""" 

126 if event.button.id.startswith("reset_config_"): 

127 field_name = event.button.id.split("_", 2)[2] 

128 self._reset_field(field_name) 

129 

130 def _reset_field(self, field_name: str) -> None: 

131 """Reset a field to its default value.""" 

132 if not self.form_manager: 

133 return 

134 

135 # Handle both top-level and nested parameters 

136 if field_name in self.param_defaults: 

137 # Top-level parameter 

138 default_value = self.param_defaults[field_name] 

139 self.form_manager.reset_parameter(field_name, default_value) 

140 self._on_field_change(field_name, default_value) 

141 # Refresh the UI widget to show the reset value 

142 self._refresh_field_widget(field_name, default_value) 

143 else: 

144 # Check if it's a nested parameter (e.g., "path_planning_output_dir_suffix" or "nested_nested_bool") 

145 parts = field_name.split('_') 

146 if len(parts) >= 2: 

147 # Try to find the nested parameter by checking prefixes from longest to shortest 

148 # This handles cases like "nested_nested_bool" where "nested" is the parent 

149 for i in range(len(parts) - 1, 0, -1): # Start from longest prefix 

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

151 if potential_nested in self.param_defaults: 

152 # Found the nested parent, get the nested field name 

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

154 

155 # Get the nested default value from the default dataclass instance 

156 nested_parent_default = self.param_defaults[potential_nested] 

157 if hasattr(nested_parent_default, nested_field): 

158 nested_default = getattr(nested_parent_default, nested_field) 

159 

160 # The form manager's reset_parameter method handles nested parameters automatically 

161 # Just pass the full hierarchical name and it will find the right nested manager 

162 self.form_manager.reset_parameter(field_name, nested_default) 

163 self._on_field_change(field_name, nested_default) 

164 # Refresh the UI widget to show the reset value 

165 self._refresh_field_widget(field_name, nested_default) 

166 return 

167 

168 # If we get here, the parameter wasn't found 

169 import logging 

170 logger = logging.getLogger(__name__) 

171 logger.warning(f"Could not reset field {field_name}: not found in defaults") 

172 

173 def _sync_field_values(self) -> None: 

174 """Sync internal field values to reactive property when safe to do so.""" 

175 if hasattr(self, '_internal_field_values'): 

176 self.field_values = self._internal_field_values.copy() 

177 

178 def _refresh_field_widget(self, field_name: str, value: Any) -> None: 

179 """Refresh a specific field widget to show the new value.""" 

180 try: 

181 widget_id = f"config_{field_name}" 

182 

183 # Try to find the widget 

184 try: 

185 widget = self.query_one(f"#{widget_id}") 

186 except Exception: 

187 # Widget not found with exact ID, try searching more broadly 

188 widgets = self.query(f"[id$='{field_name}']") # Find widgets ending with field_name 

189 if widgets: 

190 widget = widgets[0] 

191 else: 

192 return # Widget not found 

193 

194 # Update widget based on type 

195 from textual.widgets import Input, Checkbox, RadioSet, Collapsible 

196 from .shared.enum_radio_set import EnumRadioSet 

197 

198 if isinstance(widget, Input): 

199 # Input widget (int, float, str) - set value as string 

200 display_value = value.value if hasattr(value, 'value') else value 

201 widget.value = str(display_value) if display_value is not None else "" 

202 

203 elif isinstance(widget, Checkbox): 

204 # Checkbox widget (bool) - set boolean value 

205 widget.value = bool(value) 

206 

207 elif isinstance(widget, (RadioSet, EnumRadioSet)): 

208 # RadioSet/EnumRadioSet widget (Enum, List[Enum]) - find and press the correct radio button 

209 # Handle both enum values and string values 

210 if hasattr(value, 'value'): 

211 # Enum value - use the .value attribute 

212 target_value = value.value 

213 elif isinstance(value, list) and len(value) > 0: 

214 # List[Enum] - get first item's value 

215 first_item = value[0] 

216 target_value = first_item.value if hasattr(first_item, 'value') else str(first_item) 

217 else: 

218 # String value or other 

219 target_value = str(value) 

220 

221 # Find and press the correct radio button 

222 target_id = f"enum_{target_value}" 

223 for radio in widget.query("RadioButton"): 

224 if radio.id == target_id: 

225 radio.value = True 

226 break 

227 else: 

228 # Unpress other radio buttons 

229 radio.value = False 

230 

231 elif isinstance(widget, Collapsible): 

232 # Collapsible widget (nested dataclass) - cannot be reset directly 

233 # The nested parameters are handled by their own reset buttons 

234 pass 

235 

236 elif hasattr(widget, 'value'): 

237 # Generic widget with value attribute - fallback 

238 display_value = value.value if hasattr(value, 'value') else value 

239 widget.value = str(display_value) if display_value is not None else "" 

240 

241 except Exception as e: 

242 # Widget not found or update failed - this is expected for some field types 

243 import logging 

244 logger = logging.getLogger(__name__) 

245 logger.debug(f"Could not refresh widget for field {field_name}: {e}") 

246 

247 def get_config_values(self) -> Dict[str, Any]: 

248 """Get current config values from form manager.""" 

249 if self.form_manager: 

250 return self.form_manager.get_current_values() 

251 else: 

252 # Fallback to internal field values if available, otherwise reactive field_values 

253 if hasattr(self, '_internal_field_values'): 

254 return self._internal_field_values.copy() 

255 return self.field_values.copy()