Coverage for openhcs/textual_tui/windows/dual_editor_window.py: 0.0%
244 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""DualEditor window - enhanced port of dual_editor_pane.py to Textual."""
3import logging
4from typing import Optional, Callable, Union, List
5# ModalScreen import removed - using BaseOpenHCSWindow instead
6from textual.app import ComposeResult
7from textual.message import Message
8from textual.widgets import Button, TabbedContent, TabPane, Input, Select
9from textual.reactive import reactive
10from textual.widgets._tabbed_content import ContentTabs, ContentSwitcher, ContentTab
11from textual.widgets._tabs import Tab
12from textual import events # NEW
13from textual.widgets import Tabs # NEW – for type hint below (optional)
14from itertools import zip_longest
16from openhcs.core.steps.function_step import FunctionStep
17from openhcs.textual_tui.services.pattern_data_manager import PatternDataManager
18from openhcs.processing.backends.lib_registry.registry_service import RegistryService
19from openhcs.textual_tui.windows.base_window import BaseOpenHCSWindow
20# Updated import to get the message class as well
21from openhcs.textual_tui.widgets.step_parameter_editor import (
22 StepParameterEditorWidget)
23from openhcs.textual_tui.widgets.function_list_editor import FunctionListEditorWidget
25logger = logging.getLogger(__name__)
28class ButtonTab(Tab):
29 """A fake tab that acts like a button."""
31 class ButtonClicked(Message):
32 """A button tab was clicked."""
33 def __init__(self, button_id: str) -> None:
34 self.button_id = button_id
35 super().__init__()
37 def __init__(self, label: str, button_id: str, disabled: bool = False):
38 # Use a unique ID for the button tab
39 super().__init__(label, id=f"button-{button_id}", disabled=disabled)
40 self.button_id = button_id
41 # Add button-like styling
42 self.add_class("button-tab")
44 def _on_click(self, event: events.Click) -> None:
45 """Send the ButtonClicked message and swallow the click."""
46 event.stop() # don't let real tab logic run
47 event.prevent_default() # prevent default tab behavior
48 self.post_message(self.ButtonClicked(self.button_id))
51class TabbedContentWithButtons(TabbedContent):
52 """Custom TabbedContent that adds Save/Close buttons to the tab bar."""
54 def __init__(self, *args, **kwargs):
55 super().__init__(*args, **kwargs)
56 self.save_callback = None
57 self.close_callback = None
59 def _get_content_switcher(self) -> Optional[ContentSwitcher]:
60 """Safely get the content switcher."""
61 try:
62 return self.query_one(ContentSwitcher)
63 except:
64 return None
66 def _safe_switch_content(self, content_id: str) -> bool:
67 """Safely switch content, returning True if successful."""
68 switcher = self._get_content_switcher()
69 if not switcher:
70 return False
72 try:
73 # Check if the content actually exists before switching
74 switcher.get_child_by_id(content_id)
75 switcher.current = content_id
76 return True
77 except:
78 # Content doesn't exist (probably a button tab), ignore silently
79 logger.debug(f"Ignoring switch to non-existent content: {content_id}")
80 return False
82 def set_callbacks(self, save_callback: Optional[Callable] = None, close_callback: Optional[Callable] = None):
83 """Set the callbacks for save and close buttons."""
84 self.save_callback = save_callback
85 self.close_callback = close_callback
87 def compose(self) -> ComposeResult:
88 """Compose the tabbed content with button tabs in the tab bar."""
89 # Wrap content in a `TabPane` if required (copied from parent)
90 pane_content = [
91 self._set_id(
92 (
93 content
94 if isinstance(content, TabPane)
95 else TabPane(title or self.render_str(f"Tab {index}"), content)
96 ),
97 self._generate_tab_id(),
98 )
99 for index, (title, content) in enumerate(
100 zip_longest(self.titles, self._tab_content), 1
101 )
102 ]
104 # Get a tab for each pane (copied from parent)
105 tabs = [
106 ContentTab(
107 content._title,
108 content.id or "",
109 disabled=content.disabled,
110 )
111 for content in pane_content
112 ]
114 # ── tab strip ───────────────────────────────────────────────────
115 # 1️⃣ regular content-selecting tabs
116 # 2️⃣ an elastic spacer tab (width: 1fr in CSS)
117 # 3️⃣ our two action "button-tabs"
119 spacer = Tab("", id="spacer_tab", disabled=True) # visual gap
120 spacer.add_class("spacer-tab")
122 # Create button tabs with explicit styling to distinguish them
123 save_button = ButtonTab("Save", "save", disabled=False) # Always enabled
124 close_button = ButtonTab("Close", "close")
126 # Add additional classes for safety
127 save_button.add_class("action-button")
128 close_button.add_class("action-button")
130 tabs.extend([
131 spacer,
132 save_button,
133 close_button,
134 ])
136 # yield the single ContentTabs row (must be an immediate child)
137 yield ContentTabs(*tabs,
138 active=self._initial or None,
139 tabbed_content=self)
141 # Yield the content switcher and panes (copied from parent)
142 with ContentSwitcher(initial=self._initial or None):
143 yield from pane_content
145 def _on_tabs_tab_activated(
146 self, event: Tabs.TabActivated
147 ) -> None:
148 """Override to safely handle tab activation, including button tabs."""
149 # Check if the activated tab is a button tab by ID
150 if hasattr(event, 'tab') and event.tab.id and event.tab.id.startswith('button-'):
151 # Don't activate button tabs as content tabs
152 event.stop()
153 return
155 # For regular tabs, use safe switching instead of the default behavior
156 if hasattr(event, 'tab') and event.tab.id:
157 content_id = ContentTab.sans_prefix(event.tab.id)
158 if self._safe_switch_content(content_id):
159 # Successfully switched, stop the event to prevent default handling
160 event.stop()
161 return
163 # If we get here, let the default behavior handle it
164 super()._on_tabs_tab_activated(event)
166 def on_button_tab_button_clicked(self, event: ButtonTab.ButtonClicked) -> None:
167 """Handle button tab clicks."""
168 if event.button_id == "save" and self.save_callback:
169 self.save_callback()
170 elif event.button_id == "close" and self.close_callback:
171 self.close_callback()
173 def on_resize(self, event) -> None:
174 """Handle window resize to readjust tab layout."""
175 # Refresh the tabs to recalculate their layout
176 try:
177 tabs = self.query_one(ContentTabs)
178 tabs.refresh()
179 except:
180 pass # Tabs not ready yet
185class DualEditorWindow(BaseOpenHCSWindow):
186 """
187 Enhanced modal screen for editing steps and functions.
189 Ports the complete functionality from dual_editor_pane.py with:
190 - Change tracking and validation
191 - Tab switching with proper state management
192 - Integration with all visual programming services
193 - Proper error handling and user feedback
194 """
196 DEFAULT_CSS = """
197 DualEditorWindow {
198 width: 80; height: 20;
199 min-width: 80; min-height: 20;
200 }
201 """
203 # Reactive state for change tracking
204 has_changes = reactive(False)
205 current_tab = reactive("step")
207 def __init__(self, step_data: Optional[FunctionStep] = None, is_new: bool = False,
208 on_save_callback: Optional[Callable] = None):
209 super().__init__(
210 window_id="dual_editor",
211 title="Dual Editor",
212 mode="temporary",
213 allow_maximize=True
214 )
215 self.step_data = step_data
216 self.is_new = is_new
217 self.on_save_callback = on_save_callback
219 # Initialize services (reuse existing business logic)
220 self.pattern_manager = PatternDataManager()
221 self.registry_service = RegistryService()
223 # Create working copy of step data
224 if self.step_data:
225 self.editing_step = self.pattern_manager.clone_pattern(self.step_data)
226 else:
227 # Create new step with empty function list (user adds functions manually)
228 self.editing_step = FunctionStep(
229 func=[], # Start with empty function list
230 name="New Step"
231 # Let variable_components and group_by use FunctionStep's defaults
232 )
234 # Store original for change detection
235 self.original_step = self.pattern_manager.clone_pattern(self.editing_step)
237 # Editor widgets (will be created in compose)
238 self.step_editor = None
239 self.func_editor = None
241 def compose(self) -> ComposeResult:
242 """Compose the enhanced dual editor modal."""
243 # Custom tabbed content with buttons - start directly with tabs
244 with TabbedContentWithButtons(id="editor_tabs") as tabbed_content:
245 tabbed_content.set_callbacks(
246 save_callback=self._handle_save,
247 close_callback=self._handle_cancel
248 )
250 with TabPane("Step Settings", id="step_tab"):
251 # Create step editor with correct constructor
252 self.step_editor = StepParameterEditorWidget(self.editing_step)
253 yield self.step_editor
255 with TabPane("Function Pattern", id="func_tab"):
256 # Create function editor with validated function data and step identifier
257 func_data = self._validate_function_data(self.editing_step.func)
258 step_id = getattr(self.editing_step, 'name', 'unknown_step')
259 self.func_editor = FunctionListEditorWidget(func_data, step_identifier=step_id)
261 # Initialize step configuration settings in function editor
262 self.func_editor.current_group_by = self.editing_step.group_by
263 self.func_editor.current_variable_components = self.editing_step.variable_components or []
265 yield self.func_editor
267 def on_mount(self) -> None:
268 """Called when the screen is mounted."""
269 # Update change tracking to set initial Save button state
270 self._update_change_tracking()
274 def on_input_changed(self, event: Input.Changed) -> None:
275 """Handle input changes from child widgets."""
276 logger.debug(f"Input changed: {event}")
278 # Update change tracking when any input changes
279 self._update_change_tracking()
280 self._update_status("Modified step parameters")
282 def on_select_changed(self, event: Select.Changed) -> None:
283 """Handle select changes from child widgets."""
284 logger.debug(f"Select changed: {event}")
286 # Update change tracking when any select changes
287 self._update_change_tracking()
288 self._update_status("Modified step parameters")
290 def on_button_pressed(self, event: Button.Pressed) -> None:
291 """Handle button presses from child widgets."""
292 # Note: Save/Close buttons are handled by TabbedContentWithButtons.on_button_tab_button_clicked
293 # This method only handles buttons from child widgets like StepParameterEditorWidget
294 logger.debug(f"Child widget button pressed: {event.button.id}")
295 self._update_change_tracking()
296 self._update_status("Modified step configuration")
298 def on_step_parameter_editor_widget_step_parameter_changed(
299 self, event: StepParameterEditorWidget.StepParameterChanged # Listen for the specific message
300 ) -> None:
301 """Handle parameter changes from the step editor widget."""
302 logger.debug("Received StepParameterChanged from child StepParameterEditorWidget")
304 # Sync step configuration settings to function editor for dynamic component selection
305 self.func_editor.current_group_by = self.editing_step.group_by
306 self.func_editor.current_variable_components = self.editing_step.variable_components or []
308 self._update_change_tracking()
309 self._update_status("Modified step parameters (via message)")
311 def on_function_list_editor_widget_function_pattern_changed(
312 self, event: FunctionListEditorWidget.FunctionPatternChanged
313 ) -> None:
314 """Handle function pattern changes from the function editor widget."""
315 logger.debug("Received FunctionPatternChanged from child FunctionListEditorWidget")
316 # Use current_pattern property to get List or Dict format
317 self.editing_step.func = self.func_editor.current_pattern
318 self._update_change_tracking()
319 self._update_status("Modified function pattern")
323 def _update_change_tracking(self) -> None:
324 """Update change tracking state."""
325 # Compare current editing step with original
326 has_changes = not self._steps_equal(self.editing_step, self.original_step)
327 self.has_changes = has_changes
329 # Always keep save button enabled - user requested this for better UX
330 # No need to disable save button based on changes
332 # Update save button state - always enabled
333 try:
334 # Find the save button tab by looking for button tabs
335 # Use try_query to avoid NoMatches exceptions during widget lifecycle
336 tabs = self.query(ButtonTab)
337 if tabs: # Check if any tabs were found
338 for tab in tabs:
339 if hasattr(tab, 'button_id') and tab.button_id == "save":
340 tab.disabled = False # Always enabled
341 logger.debug("Save button always enabled (user preference)")
342 break
343 else:
344 logger.debug("No ButtonTab widgets found yet (widget still mounting?)")
345 except Exception as e:
346 logger.debug(f"Error updating save button state: {e}")
348 def _validate_function_data(self, func_data) -> Union[List, callable, None]:
349 """Validate and normalize function data for FunctionListEditorWidget."""
350 if func_data is None:
351 return None
352 elif callable(func_data):
353 return func_data
354 elif isinstance(func_data, (list, dict)):
355 return func_data
356 else:
357 logger.warning(f"Invalid function data type: {type(func_data)}, using None")
358 return None
360 def _steps_equal(self, step1: FunctionStep, step2: FunctionStep) -> bool:
361 """Compare two FunctionSteps for equality."""
362 return (
363 step1.name == step2.name and
364 step1.func == step2.func and
365 step1.variable_components == step2.variable_components and
366 step1.group_by == step2.group_by
367 )
369 def _update_status(self, message: str) -> None:
370 """Update status message - now just logs since we removed the status bar."""
371 logger.debug(f"Status: {message}")
373 def watch_has_changes(self, has_changes: bool) -> None:
374 """React to changes in has_changes state."""
375 # Update window title to show unsaved changes
376 base_title = "Dual Editor - Edit Step" if not self.is_new else "Dual Editor - New Step"
377 if has_changes:
378 self.title = f"{base_title} *"
379 else:
380 self.title = base_title
384 def _handle_save(self) -> None:
385 """Handle save button with validation and type conversion."""
386 # Sync current UI values to editing_step before validation
387 self._sync_ui_to_editing_step()
389 # Debug logging to see what's happening with step name
390 logger.debug(f"Step name validation - editing_step.name: '{self.editing_step.name}', type: {type(self.editing_step.name)}")
392 # Validate step data
393 if not self.editing_step.name or not self.editing_step.name.strip():
394 self._update_status("Error: Step name cannot be empty")
395 return
397 if not self.editing_step.func:
398 self._update_status("Error: Function pattern cannot be empty")
399 return
401 # Validate and convert function parameter types
402 validation_errors = self._validate_and_convert_function_parameters()
403 if validation_errors:
404 # Show error dialog with specific validation errors
405 from openhcs.textual_tui.app import ErrorDialog
406 error_message = "Parameter validation failed. Please fix the following issues:"
407 error_details = "\n".join(f"• {error}" for error in validation_errors)
408 error_dialog = ErrorDialog(error_message, error_details)
409 self.app.push_screen(error_dialog)
410 self._update_status("Error: Invalid parameter values")
411 return
413 # Save successful
414 self._update_status("Saved successfully")
415 if self.on_save_callback:
416 self.on_save_callback(self.editing_step)
417 # Window remains open after save (user preference)
419 def _validate_and_convert_function_parameters(self) -> List[str]:
420 """
421 Validate and convert all function parameters using type hints.
423 Returns:
424 List of error messages. Empty list if all parameters are valid.
425 """
426 errors = []
428 try:
429 # Get current function pattern from the editor
430 current_pattern = self.func_editor.current_pattern
432 # Handle different pattern types (list or dict)
433 functions_to_validate = []
434 if isinstance(current_pattern, list):
435 functions_to_validate = current_pattern
436 elif isinstance(current_pattern, dict):
437 # Flatten all functions from all channels
438 for channel_functions in current_pattern.values():
439 if isinstance(channel_functions, list):
440 functions_to_validate.extend(channel_functions)
442 # Validate each function
443 for func_index, func_item in enumerate(functions_to_validate):
444 if isinstance(func_item, tuple) and len(func_item) == 2:
445 func, kwargs = func_item
447 # Get expected parameter types from function signature
448 from openhcs.introspection.signature_analyzer import SignatureAnalyzer
449 from openhcs.textual_tui.widgets.shared.parameter_form_manager import ParameterFormManager
451 param_info = SignatureAnalyzer.analyze(func)
453 # Validate each parameter
454 for param_name, info in param_info.items():
455 if param_name in kwargs:
456 current_value = kwargs[param_name]
457 expected_type = info.param_type
459 # Skip if value is already the correct type (not a string)
460 if not isinstance(current_value, str):
461 continue
463 # Try to convert using the enhanced type converter
464 try:
465 converted_value = ParameterFormManager.convert_string_to_type(
466 current_value, expected_type, strict=True
467 )
468 # Update the kwargs with the converted value
469 kwargs[param_name] = converted_value
471 except ValueError as e:
472 # Collect the error message
473 func_name = getattr(func, '__name__', str(func))
474 errors.append(f"Function '{func_name}', parameter '{param_name}': {str(e)}")
476 except Exception as e:
477 # Catch any unexpected errors during validation
478 errors.append(f"Validation error: {str(e)}")
480 return errors
482 def _sync_ui_to_editing_step(self) -> None:
483 """Sync current UI values to the editing_step object before validation."""
484 try:
485 # Sync step editor values (name, group_by, variable_components)
486 if self.step_editor:
487 # The step editor should have already updated editing_step via messages,
488 # but let's make sure by getting current values
489 pass # StepParameterEditorWidget updates editing_step directly
491 # Sync function editor values (func pattern)
492 if self.func_editor:
493 self.editing_step.func = self.func_editor.current_pattern
495 except Exception as e:
496 # Log but don't fail - validation will catch issues
497 logger.debug(f"Error syncing UI to editing_step: {e}")
499 def _handle_cancel(self) -> None:
500 """Handle cancel button with change confirmation."""
501 if self.has_changes:
502 # TODO: Add confirmation dialog for unsaved changes
503 self._update_status("Cancelled - changes discarded")
504 else:
505 self._update_status("Cancelled")
507 if self.on_save_callback:
508 self.on_save_callback(None)
509 self.close_window()