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