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

1"""Step parameter editor widget - port of step_parameter_editor.py to Textual.""" 

2 

3import logging 

4from typing import Any 

5from pathlib import Path 

6from textual.containers import Horizontal, ScrollableContainer 

7from textual.widgets import Button, Static 

8from textual.app import ComposeResult 

9from textual.message import Message 

10 

11from openhcs.core.steps.function_step import FunctionStep 

12from openhcs.core.steps.abstract import AbstractStep 

13from .shared.parameter_form_manager import ParameterFormManager 

14from openhcs.introspection.signature_analyzer import SignatureAnalyzer 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class StepParameterEditorWidget(ScrollableContainer): 

20 """ 

21 Step parameter editor using dynamic form generation. 

22 Builds forms based on FunctionStep constructor signature. 

23 """ 

24 

25 class StepParameterChanged(Message): 

26 """Message to indicate a parameter has changed.""" 

27 pass 

28 

29 def __init__(self, step: FunctionStep): 

30 super().__init__() 

31 self.step = step 

32 

33 # Create parameter form manager using shared components 

34 # Analyze AbstractStep to get all inherited parameters including materialization_config 

35 # Auto-detection correctly identifies constructors and includes all parameters 

36 param_info = SignatureAnalyzer.analyze(AbstractStep.__init__) 

37 

38 # Get current parameter values from step instance 

39 parameters = {} 

40 parameter_types = {} 

41 param_defaults = {} 

42 

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

44 # All AbstractStep parameters are relevant for editing 

45 current_value = getattr(self.step, name, info.default_value) 

46 parameters[name] = current_value 

47 parameter_types[name] = info.param_type 

48 param_defaults[name] = info.default_value 

49 

50 # Configure form manager for step editing with pipeline context 

51 from openhcs.core.config import GlobalPipelineConfig 

52 self.form_manager = ParameterFormManager( 

53 parameters, parameter_types, "step", param_info, 

54 global_config_type=GlobalPipelineConfig, 

55 placeholder_prefix="Pipeline default" 

56 ) 

57 self.param_defaults = param_defaults 

58 

59 def compose(self) -> ComposeResult: 

60 """Compose the step parameter form dynamically.""" 

61 yield Static("[bold]Step Parameters[/bold]") 

62 

63 # Build form using shared form manager 

64 yield from self._build_step_parameter_form() 

65 

66 # Action buttons 

67 with Horizontal(): 

68 yield Button("Load .step", id="load_step_btn", compact=True) 

69 yield Button("Save .step As", id="save_as_btn", compact=True) 

70 

71 def _build_step_parameter_form(self) -> ComposeResult: 

72 """Generate form widgets using shared ParameterFormManager.""" 

73 try: 

74 # Use shared form manager to build form 

75 yield from self.form_manager.build_form() 

76 except Exception as e: 

77 yield Static(f"[red]Error building step parameter form: {e}[/red]") 

78 

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

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

81 if event.input.id.startswith("step_"): 

82 param_name = event.input.id.split("_", 1)[1] 

83 if self.form_manager: 

84 self.form_manager.update_parameter(param_name, event.value) 

85 final_value = self.form_manager.parameters[param_name] 

86 self._handle_parameter_change(param_name, final_value) 

87 

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

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

90 if not event.checkbox.id.startswith("step_") or not self.form_manager: 

91 return 

92 

93 checkbox_id = event.checkbox.id 

94 if checkbox_id.endswith("_enabled"): 

95 # Optional dataclass checkbox 

96 param_name = checkbox_id.replace("step_", "").replace("_enabled", "") 

97 self.form_manager.handle_optional_checkbox_change(param_name, event.value) 

98 else: 

99 # Regular checkbox 

100 param_name = checkbox_id.split("_", 1)[1] 

101 self.form_manager.update_parameter(param_name, event.value) 

102 

103 final_value = self.form_manager.parameters[param_name] 

104 self._handle_parameter_change(param_name, final_value) 

105 

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

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

108 if event.radio_set.id.startswith("step_"): 

109 param_name = event.radio_set.id.split("_", 1)[1] 

110 if event.pressed and event.pressed.id: 

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

112 if self.form_manager: 

113 self.form_manager.update_parameter(param_name, enum_value) 

114 final_value = self.form_manager.parameters[param_name] 

115 self._handle_parameter_change(param_name, final_value) 

116 

117 def on_button_pressed(self, event: Button.Pressed) -> None: 

118 """Handle button presses.""" 

119 if event.button.id == "load_step_btn": 

120 self.run_worker(self._load_step()) 

121 elif event.button.id == "save_as_btn": 

122 self.run_worker(self._save_step_as()) 

123 elif event.button.id.startswith("reset_step_"): 

124 # Individual parameter reset 

125 param_name = event.button.id.split("_", 2)[2] 

126 self._reset_parameter(param_name) 

127 

128 def _handle_parameter_change(self, param_name: str, value: Any) -> None: 

129 """Update step parameter and emit change message.""" 

130 try: 

131 # Convert value to appropriate type 

132 if param_name == 'force_disk_output': 

133 value = bool(value) 

134 elif param_name in ('input_dir', 'output_dir') and value: 

135 value = Path(value) 

136 

137 # Update step attribute 

138 setattr(self.step, param_name, value) 

139 logger.debug(f"Updated step parameter {param_name}={value}") 

140 self.post_message(self.StepParameterChanged()) 

141 except Exception as e: 

142 logger.error(f"Error updating step parameter {param_name}: {e}") 

143 

144 def _reset_parameter(self, param_name: str) -> None: 

145 """Reset a specific parameter to its default value.""" 

146 if not self.form_manager or param_name not in self.param_defaults: 

147 return 

148 

149 try: 

150 # Use form manager to reset parameter 

151 default_value = self.param_defaults[param_name] 

152 self.form_manager.reset_parameter(param_name, default_value) 

153 

154 # Update step instance and notify parent 

155 self._handle_parameter_change(param_name, default_value) 

156 

157 # Refresh the UI to show the reset value 

158 self._refresh_form_widgets() 

159 

160 logger.debug(f"Reset step parameter {param_name} to default: {default_value}") 

161 except Exception as e: 

162 logger.error(f"Error resetting step parameter {param_name}: {e}") 

163 

164 def _refresh_form_widgets(self) -> None: 

165 """Refresh form widgets to show current parameter values.""" 

166 try: 

167 current_values = self.form_manager.get_current_values() 

168 

169 # Update each widget with current value 

170 for param_name, value in current_values.items(): 

171 widget_id = f"step_{param_name}" 

172 try: 

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

174 

175 # Update widget based on type 

176 if hasattr(widget, 'value'): 

177 # Convert enum to string for display 

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

179 widget.value = display_value 

180 elif hasattr(widget, 'pressed'): 

181 # RadioSet - find and press the correct radio button 

182 if hasattr(value, 'value'): 

183 target_id = f"enum_{value.value}" 

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

185 if radio.id == target_id: 

186 radio.value = True 

187 break 

188 except Exception as widget_error: 

189 logger.debug(f"Could not update widget {widget_id}: {widget_error}") 

190 

191 except Exception as e: 

192 logger.warning(f"Failed to refresh form widgets: {e}") 

193 

194 async def _load_step(self) -> None: 

195 """Load step configuration from file.""" 

196 from openhcs.textual_tui.windows import open_file_browser_window, BrowserMode 

197 from openhcs.textual_tui.services.file_browser_service import SelectionMode 

198 from openhcs.constants.constants import Backend 

199 

200 def handle_result(result): 

201 if result and isinstance(result, Path): 

202 self._load_step_from_file(result) 

203 

204 # Launch file browser window for .step files 

205 from openhcs.core.path_cache import get_cached_browser_path, PathCacheKey 

206 

207 await open_file_browser_window( 

208 app=self.app, 

209 file_manager=self.app.filemanager, 

210 initial_path=get_cached_browser_path(PathCacheKey.STEP_SETTINGS), 

211 backend=Backend.DISK, 

212 title="Load Step Settings (.step)", 

213 mode=BrowserMode.LOAD, 

214 selection_mode=SelectionMode.FILES_ONLY, 

215 filter_extensions=['.step'], 

216 cache_key=PathCacheKey.STEP_SETTINGS, 

217 on_result_callback=handle_result, 

218 caller_id="step_parameter_editor" 

219 ) 

220 

221 async def _save_step_as(self) -> None: 

222 """Save step configuration to file.""" 

223 from openhcs.textual_tui.windows import open_file_browser_window, BrowserMode 

224 from openhcs.textual_tui.services.file_browser_service import SelectionMode 

225 from openhcs.constants.constants import Backend 

226 

227 def handle_result(result): 

228 if result and isinstance(result, Path): 

229 self._save_step_to_file(result) 

230 

231 # Launch file browser window for saving .step files 

232 from openhcs.core.path_cache import get_cached_browser_path, PathCacheKey 

233 

234 await open_file_browser_window( 

235 app=self.app, 

236 file_manager=self.app.filemanager, 

237 initial_path=get_cached_browser_path(PathCacheKey.STEP_SETTINGS), 

238 backend=Backend.DISK, 

239 title="Save Step Settings (.step)", 

240 mode=BrowserMode.SAVE, 

241 selection_mode=SelectionMode.FILES_ONLY, 

242 filter_extensions=['.step'], 

243 default_filename="step_settings.step", 

244 cache_key=PathCacheKey.STEP_SETTINGS, 

245 on_result_callback=handle_result, 

246 caller_id="step_parameter_editor" 

247 ) 

248 

249 def _load_step_from_file(self, file_path: Path) -> None: 

250 """Load step parameters from .step file.""" 

251 import dill as pickle 

252 try: 

253 with open(file_path, 'rb') as f: 

254 step_data = pickle.load(f) 

255 

256 # Update both the step object and form manager 

257 for param_name, value in step_data.items(): 

258 if param_name in self.form_manager.parameters: 

259 # Update form manager first 

260 self.form_manager.update_parameter(param_name, value) 

261 # Then update step object and emit change message 

262 self._handle_parameter_change(param_name, value) 

263 

264 # Refresh the UI to show loaded values 

265 self._refresh_form_widgets() 

266 

267 logger.debug(f"Loaded {len(step_data)} parameters from {file_path.name}") 

268 except Exception as e: 

269 logger.error(f"Failed to load step: {e}") 

270 

271 def _save_step_to_file(self, file_path: Path) -> None: 

272 """Save step parameters to .step file.""" 

273 import pickle 

274 try: 

275 step_data = self.form_manager.get_current_values() 

276 with open(file_path, 'wb') as f: 

277 pickle.dump(step_data, f) 

278 except Exception as e: 

279 logger.error(f"Failed to save step: {e}")