Coverage for openhcs/pyqt_gui/widgets/pipeline_editor.py: 0.0%
551 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2Pipeline Editor Widget for PyQt6
4Pipeline step management with full feature parity to Textual TUI version.
5Uses hybrid approach: extracted business logic + clean PyQt6 UI.
6"""
8import logging
9import inspect
10from typing import List, Dict, Optional, Callable, Tuple
11from pathlib import Path
13from PyQt6.QtWidgets import (
14 QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QListWidget,
15 QListWidgetItem, QLabel, QSplitter, QStyledItemDelegate, QStyle,
16 QStyleOptionViewItem, QApplication
17)
18from PyQt6.QtCore import Qt, pyqtSignal
19from PyQt6.QtGui import QFont, QPainter, QColor, QPen, QFontMetrics
21from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator
22from openhcs.core.config import GlobalPipelineConfig
23from openhcs.io.filemanager import FileManager
24from openhcs.core.steps.function_step import FunctionStep
25from openhcs.pyqt_gui.widgets.mixins import (
26 preserve_selection_during_update,
27 handle_selection_change_with_prevention
28)
29from openhcs.pyqt_gui.shared.style_generator import StyleSheetGenerator
30from openhcs.pyqt_gui.shared.color_scheme import PyQt6ColorScheme
31from openhcs.pyqt_gui.config import PyQtGUIConfig, get_default_pyqt_gui_config
33logger = logging.getLogger(__name__)
36class StepListItemDelegate(QStyledItemDelegate):
37 """Custom delegate to render step name in white and preview text in grey without breaking hover/selection/borders."""
38 def __init__(self, name_color: QColor, preview_color: QColor, selected_text_color: QColor, parent=None):
39 super().__init__(parent)
40 self.name_color = name_color
41 self.preview_color = preview_color
42 self.selected_text_color = selected_text_color
44 def paint(self, painter: QPainter, option, index) -> None:
45 # Prepare a copy to let style draw backgrounds, hover, selection, borders, etc.
46 opt = QStyleOptionViewItem(option)
47 self.initStyleOption(opt, index)
49 # Capture text and prevent default text draw
50 text = opt.text or ""
51 opt.text = ""
53 painter.save()
54 # Let the current style paint the item (background, selection, hover, separators)
55 style = opt.widget.style() if opt.widget else QApplication.style()
56 style.drawControl(QStyle.ControlElement.CE_ItemViewItem, opt, painter, opt.widget)
58 # Check if item is selected
59 is_selected = opt.state & QStyle.StateFlag.State_Selected
61 # Check if step is disabled (stored in UserRole+1)
62 is_disabled = index.data(Qt.ItemDataRole.UserRole + 1) or False
64 # Now custom-draw the text with mixed colors
65 rect = opt.rect.adjusted(6, 0, -6, 0)
67 # Use strikethrough font for disabled steps
68 font = QFont(opt.font)
69 if is_disabled:
70 font.setStrikeOut(True)
71 painter.setFont(font)
73 fm = QFontMetrics(font)
74 baseline_y = rect.y() + (rect.height() + fm.ascent() - fm.descent()) // 2
76 sep_idx = text.find(" (")
77 if sep_idx != -1 and text.endswith(")"):
78 name_part = text[:sep_idx]
79 preview_part = text[sep_idx:]
81 # Use white for both parts when selected, otherwise use normal colors
82 if is_selected:
83 painter.setPen(QPen(self.selected_text_color))
84 painter.drawText(rect.x(), baseline_y, name_part)
85 name_width = fm.horizontalAdvance(name_part)
86 painter.drawText(rect.x() + name_width, baseline_y, preview_part)
87 else:
88 painter.setPen(QPen(self.name_color))
89 painter.drawText(rect.x(), baseline_y, name_part)
90 name_width = fm.horizontalAdvance(name_part)
92 painter.setPen(QPen(self.preview_color))
93 painter.drawText(rect.x() + name_width, baseline_y, preview_part)
94 else:
95 painter.setPen(QPen(self.selected_text_color if is_selected else self.name_color))
96 painter.drawText(rect.x(), baseline_y, text)
98 painter.restore()
100class ReorderableListWidget(QListWidget):
101 """
102 Custom QListWidget that properly handles drag and drop reordering.
103 Emits a signal when items are moved so the parent can update the data model.
104 """
106 items_reordered = pyqtSignal(int, int) # from_index, to_index
108 def __init__(self, parent=None):
109 super().__init__(parent)
110 self.setDragDropMode(QListWidget.DragDropMode.InternalMove)
112 def dropEvent(self, event):
113 """Handle drop events and emit reorder signal."""
114 # Get the item being dropped and its original position
115 source_item = self.currentItem()
116 if not source_item:
117 super().dropEvent(event)
118 return
120 source_index = self.row(source_item)
122 # Let the default drop behavior happen first
123 super().dropEvent(event)
125 # Find the new position of the item
126 target_index = self.row(source_item)
128 # Only emit signal if position actually changed
129 if source_index != target_index:
130 self.items_reordered.emit(source_index, target_index)
133class PipelineEditorWidget(QWidget):
134 """
135 PyQt6 Pipeline Editor Widget.
137 Manages pipeline steps with add, edit, delete, load, save functionality.
138 Preserves all business logic from Textual version with clean PyQt6 UI.
139 """
141 # Signals
142 pipeline_changed = pyqtSignal(list) # List[FunctionStep]
143 step_selected = pyqtSignal(object) # FunctionStep
144 status_message = pyqtSignal(str) # status message
146 def __init__(self, file_manager: FileManager, service_adapter,
147 color_scheme: Optional[PyQt6ColorScheme] = None, gui_config: Optional[PyQtGUIConfig] = None, parent=None):
148 """
149 Initialize the pipeline editor widget.
151 Args:
152 file_manager: FileManager instance for file operations
153 service_adapter: PyQt service adapter for dialogs and operations
154 color_scheme: Color scheme for styling (optional, uses service adapter if None)
155 gui_config: GUI configuration (optional, uses default if None)
156 parent: Parent widget
157 """
158 super().__init__(parent)
160 # Core dependencies
161 self.file_manager = file_manager
162 self.service_adapter = service_adapter
163 self.global_config = service_adapter.get_global_config()
164 self.gui_config = gui_config or get_default_pyqt_gui_config()
166 # Initialize color scheme and style generator
167 self.color_scheme = color_scheme or service_adapter.get_current_color_scheme()
168 self.style_generator = StyleSheetGenerator(self.color_scheme)
170 # Business logic state (extracted from Textual version)
171 self.pipeline_steps: List[FunctionStep] = []
172 self.current_plate: str = ""
173 self.selected_step: str = ""
174 self.plate_pipelines: Dict[str, List[FunctionStep]] = {} # Per-plate pipeline storage
176 # UI components
177 self.step_list: Optional[QListWidget] = None
178 self.buttons: Dict[str, QPushButton] = {}
179 self.status_label: Optional[QLabel] = None
181 # Reference to plate manager (set externally)
182 self.plate_manager = None
184 # Setup UI
185 self.setup_ui()
186 self.setup_connections()
187 self.update_button_states()
189 logger.debug("Pipeline editor widget initialized")
191 # ========== UI Setup ==========
193 def setup_ui(self):
194 """Setup the user interface."""
195 layout = QVBoxLayout(self)
196 layout.setContentsMargins(2, 2, 2, 2)
197 layout.setSpacing(2)
199 # Header with title and status
200 header_widget = QWidget()
201 header_layout = QHBoxLayout(header_widget)
202 header_layout.setContentsMargins(5, 5, 5, 5)
204 title_label = QLabel("Pipeline Editor")
205 title_label.setFont(QFont("Arial", 12, QFont.Weight.Bold))
206 title_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.text_accent)};")
207 header_layout.addWidget(title_label)
209 header_layout.addStretch()
211 # Status label in header
212 self.status_label = QLabel("Ready")
213 self.status_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.status_success)}; font-weight: bold;")
214 header_layout.addWidget(self.status_label)
216 layout.addWidget(header_widget)
218 # Main content splitter
219 splitter = QSplitter(Qt.Orientation.Vertical)
220 layout.addWidget(splitter)
222 # Pipeline steps list
223 self.step_list = ReorderableListWidget()
224 self.step_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
225 self.step_list.setStyleSheet(f"""
226 QListWidget {{
227 background-color: {self.color_scheme.to_hex(self.color_scheme.panel_bg)};
228 color: {self.color_scheme.to_hex(self.color_scheme.text_primary)};
229 border: none;
230 padding: 5px;
231 }}
232 QListWidget::item {{
233 padding: 8px;
234 border: none;
235 border-radius: 3px;
236 margin: 2px;
237 }}
238 QListWidget::item:selected {{
239 background-color: {self.color_scheme.to_hex(self.color_scheme.selection_bg)};
240 color: {self.color_scheme.to_hex(self.color_scheme.text_primary)};
241 }}
242 QListWidget::item:hover {{
243 background-color: {self.color_scheme.to_hex(self.color_scheme.hover_bg)};
244 }}
245 """)
246 # Set custom delegate to render white name and grey preview
247 try:
248 name_color = QColor(self.color_scheme.to_hex(self.color_scheme.text_primary))
249 preview_color = QColor(self.color_scheme.to_hex(self.color_scheme.text_disabled))
250 selected_text_color = QColor("#FFFFFF") # White text when selected
251 self.step_list.setItemDelegate(StepListItemDelegate(name_color, preview_color, selected_text_color, self.step_list))
252 except Exception:
253 # Fallback silently if color scheme isn't ready
254 pass
255 splitter.addWidget(self.step_list)
257 # Button panel
258 button_panel = self.create_button_panel()
259 splitter.addWidget(button_panel)
261 # Set splitter proportions
262 splitter.setSizes([400, 120])
264 def create_button_panel(self) -> QWidget:
265 """
266 Create the button panel with all pipeline actions.
268 Returns:
269 Widget containing action buttons
270 """
271 panel = QWidget()
272 panel.setStyleSheet(f"""
273 QWidget {{
274 background-color: {self.color_scheme.to_hex(self.color_scheme.window_bg)};
275 border: none;
276 padding: 0px;
277 }}
278 """)
280 layout = QVBoxLayout(panel)
281 layout.setContentsMargins(0, 0, 0, 0)
282 layout.setSpacing(0)
284 # Button configurations (extracted from Textual version)
285 button_configs = [
286 ("Add", "add_step", "Add new pipeline step"),
287 ("Del", "del_step", "Delete selected steps"),
288 ("Edit", "edit_step", "Edit selected step"),
289 ("Auto", "auto_load_pipeline", "Load basic_pipeline.py"),
290 ("Code", "code_pipeline", "Edit pipeline as Python code"),
291 ]
293 # Create buttons in a single row
294 row_layout = QHBoxLayout()
295 row_layout.setContentsMargins(2, 2, 2, 2)
296 row_layout.setSpacing(2)
298 for name, action, tooltip in button_configs:
299 button = QPushButton(name)
300 button.setToolTip(tooltip)
301 button.setMinimumHeight(30)
302 button.setStyleSheet(self.style_generator.generate_button_style())
304 # Connect button to action
305 button.clicked.connect(lambda checked, a=action: self.handle_button_action(a))
307 self.buttons[action] = button
308 row_layout.addWidget(button)
310 layout.addLayout(row_layout)
312 # Set maximum height to constrain the button panel
313 panel.setMaximumHeight(40)
315 return panel
319 def setup_connections(self):
320 """Setup signal/slot connections."""
321 # Step list selection
322 self.step_list.itemSelectionChanged.connect(self.on_selection_changed)
323 self.step_list.itemDoubleClicked.connect(self.on_item_double_clicked)
325 # Step list reordering
326 self.step_list.items_reordered.connect(self.on_steps_reordered)
328 # Internal signals
329 self.status_message.connect(self.update_status)
330 self.pipeline_changed.connect(self.on_pipeline_changed)
332 def handle_button_action(self, action: str):
333 """
334 Handle button actions (extracted from Textual version).
336 Args:
337 action: Action identifier
338 """
339 # Action mapping (preserved from Textual version)
340 action_map = {
341 "add_step": self.action_add_step,
342 "del_step": self.action_delete_step,
343 "edit_step": self.action_edit_step,
344 "auto_load_pipeline": self.action_auto_load_pipeline,
345 "code_pipeline": self.action_code_pipeline,
346 }
348 if action in action_map:
349 action_func = action_map[action]
351 # Handle async actions
352 if inspect.iscoroutinefunction(action_func):
353 # Run async action in thread
354 self.run_async_action(action_func)
355 else:
356 action_func()
358 def run_async_action(self, async_func: Callable):
359 """
360 Run async action using service adapter.
362 Args:
363 async_func: Async function to execute
364 """
365 self.service_adapter.execute_async_operation(async_func)
367 # ========== Business Logic Methods (Extracted from Textual) ==========
369 def format_item_for_display(self, step: FunctionStep) -> Tuple[str, str]:
370 """
371 Format step for display in the list with constructor value preview.
373 Args:
374 step: FunctionStep to format
376 Returns:
377 Tuple of (display_text, step_name)
378 """
379 step_name = getattr(step, 'name', 'Unknown Step')
381 # Build preview of key constructor values
382 preview_parts = []
384 # Function preview
385 func = getattr(step, 'func', None)
386 if func:
387 if isinstance(func, list) and func:
388 if len(func) == 1:
389 func_name = getattr(func[0], '__name__', str(func[0]))
390 preview_parts.append(f"func={func_name}")
391 else:
392 preview_parts.append(f"func=[{len(func)} functions]")
393 elif callable(func):
394 func_name = getattr(func, '__name__', str(func))
395 preview_parts.append(f"func={func_name}")
396 elif isinstance(func, dict):
397 preview_parts.append(f"func={{dict with {len(func)} keys}}")
399 # Variable components preview
400 var_components = getattr(step, 'variable_components', None)
401 if var_components:
402 if len(var_components) == 1:
403 comp_name = getattr(var_components[0], 'name', str(var_components[0]))
404 preview_parts.append(f"components=[{comp_name}]")
405 else:
406 comp_names = [getattr(c, 'name', str(c)) for c in var_components[:2]]
407 if len(var_components) > 2:
408 comp_names.append(f"+{len(var_components)-2} more")
409 preview_parts.append(f"components=[{', '.join(comp_names)}]")
411 # Group by preview
412 group_by = getattr(step, 'group_by', None)
413 if group_by and group_by.value is not None: # Check for GroupBy.NONE
414 group_name = getattr(group_by, 'name', str(group_by))
415 preview_parts.append(f"group_by={group_name}")
417 # Input source preview
418 input_source = getattr(step, 'input_source', None)
419 if input_source:
420 source_name = getattr(input_source, 'name', str(input_source))
421 if source_name != 'PREVIOUS_STEP': # Only show if not default
422 preview_parts.append(f"input={source_name}")
424 # Optional configurations preview
425 config_indicators = []
426 if hasattr(step, 'step_materialization_config') and step.step_materialization_config:
427 config_indicators.append("MAT")
428 if hasattr(step, 'napari_streaming_config') and step.napari_streaming_config:
429 config_indicators.append("NAP")
430 if hasattr(step, 'fiji_streaming_config') and step.fiji_streaming_config:
431 config_indicators.append("FIJI")
432 if hasattr(step, 'step_well_filter_config') and step.step_well_filter_config:
433 config_indicators.append("FILT")
435 if config_indicators:
436 preview_parts.append(f"configs=[{','.join(config_indicators)}]")
438 # Build display text
439 if preview_parts:
440 preview = " | ".join(preview_parts)
441 display_text = f"▶ {step_name} ({preview})"
442 else:
443 display_text = f"▶ {step_name}"
445 return display_text, step_name
447 def _create_step_tooltip(self, step: FunctionStep) -> str:
448 """Create detailed tooltip for a step showing all constructor values."""
449 step_name = getattr(step, 'name', 'Unknown Step')
450 tooltip_lines = [f"Step: {step_name}"]
452 # Function details
453 func = getattr(step, 'func', None)
454 if func:
455 if isinstance(func, list):
456 if len(func) == 1:
457 func_name = getattr(func[0], '__name__', str(func[0]))
458 tooltip_lines.append(f"Function: {func_name}")
459 else:
460 func_names = [getattr(f, '__name__', str(f)) for f in func[:3]]
461 if len(func) > 3:
462 func_names.append(f"... +{len(func)-3} more")
463 tooltip_lines.append(f"Functions: {', '.join(func_names)}")
464 elif callable(func):
465 func_name = getattr(func, '__name__', str(func))
466 tooltip_lines.append(f"Function: {func_name}")
467 elif isinstance(func, dict):
468 tooltip_lines.append(f"Function: Dictionary with {len(func)} routing keys")
469 else:
470 tooltip_lines.append("Function: None")
472 # Variable components
473 var_components = getattr(step, 'variable_components', None)
474 if var_components:
475 comp_names = [getattr(c, 'name', str(c)) for c in var_components]
476 tooltip_lines.append(f"Variable Components: [{', '.join(comp_names)}]")
477 else:
478 tooltip_lines.append("Variable Components: None")
480 # Group by
481 group_by = getattr(step, 'group_by', None)
482 if group_by and group_by.value is not None: # Check for GroupBy.NONE
483 group_name = getattr(group_by, 'name', str(group_by))
484 tooltip_lines.append(f"Group By: {group_name}")
485 else:
486 tooltip_lines.append("Group By: None")
488 # Input source
489 input_source = getattr(step, 'input_source', None)
490 if input_source:
491 source_name = getattr(input_source, 'name', str(input_source))
492 tooltip_lines.append(f"Input Source: {source_name}")
493 else:
494 tooltip_lines.append("Input Source: None")
496 # Additional configurations with details
497 config_details = []
498 if hasattr(step, 'step_materialization_config') and step.step_materialization_config:
499 config_details.append("• Materialization Config: Enabled")
500 if hasattr(step, 'napari_streaming_config') and step.napari_streaming_config:
501 napari_config = step.napari_streaming_config
502 port = getattr(napari_config, 'port', 'default')
503 config_details.append(f"• Napari Streaming: Port {port}")
504 if hasattr(step, 'fiji_streaming_config') and step.fiji_streaming_config:
505 config_details.append("• Fiji Streaming: Enabled")
506 if hasattr(step, 'step_well_filter_config') and step.step_well_filter_config:
507 well_config = step.step_well_filter_config
508 well_filter = getattr(well_config, 'well_filter', 'default')
509 config_details.append(f"• Well Filter: {well_filter}")
511 if config_details:
512 tooltip_lines.append("") # Empty line separator
513 tooltip_lines.extend(config_details)
515 return '\n'.join(tooltip_lines)
517 def action_add_step(self):
518 """Handle Add Step button (adapted from Textual version)."""
520 from openhcs.core.steps.function_step import FunctionStep
521 from openhcs.pyqt_gui.windows.dual_editor_window import DualEditorWindow
523 # Get orchestrator for step creation
524 orchestrator = self._get_current_orchestrator()
526 # Create new step
527 step_name = f"Step_{len(self.pipeline_steps) + 1}"
528 new_step = FunctionStep(
529 func=[], # Start with empty function list
530 name=step_name
531 )
535 def handle_save(edited_step):
536 """Handle step save from editor."""
537 self.pipeline_steps.append(edited_step)
538 self.update_step_list()
539 self.pipeline_changed.emit(self.pipeline_steps)
540 self.status_message.emit(f"Added new step: {edited_step.name}")
542 # Create and show editor dialog within the correct config context
543 orchestrator = self._get_current_orchestrator()
545 # SIMPLIFIED: Orchestrator context is automatically available through type-based registry
546 # No need for explicit context management - dual-axis resolver handles it automatically
547 if not orchestrator:
548 logger.info("No orchestrator found for step editor context, This should not happen.")
550 editor = DualEditorWindow(
551 step_data=new_step,
552 is_new=True,
553 on_save_callback=handle_save,
554 orchestrator=orchestrator,
555 gui_config=self.gui_config,
556 parent=self
557 )
558 # Set original step for change detection
559 editor.set_original_step_for_change_detection()
560 editor.show()
561 editor.raise_()
562 editor.activateWindow()
564 def action_delete_step(self):
565 """Handle Delete Step button (extracted from Textual version)."""
566 selected_items = self.get_selected_steps()
567 if not selected_items:
568 self.service_adapter.show_error_dialog("No steps selected to delete.")
569 return
571 # Remove selected steps
572 steps_to_remove = set(getattr(item, 'name', '') for item in selected_items)
573 new_steps = [step for step in self.pipeline_steps if getattr(step, 'name', '') not in steps_to_remove]
575 self.pipeline_steps = new_steps
576 self.update_step_list()
577 self.pipeline_changed.emit(self.pipeline_steps)
579 deleted_count = len(selected_items)
580 self.status_message.emit(f"Deleted {deleted_count} steps")
582 def action_edit_step(self):
583 """Handle Edit Step button (adapted from Textual version)."""
584 selected_items = self.get_selected_steps()
585 if not selected_items:
586 self.service_adapter.show_error_dialog("No step selected to edit.")
587 return
589 step_to_edit = selected_items[0]
591 # Open step editor dialog
592 from openhcs.pyqt_gui.windows.dual_editor_window import DualEditorWindow
594 def handle_save(edited_step):
595 """Handle step save from editor."""
596 # Find and replace the step in the pipeline
597 for i, step in enumerate(self.pipeline_steps):
598 if step is step_to_edit:
599 self.pipeline_steps[i] = edited_step
600 break
602 # Update the display
603 self.update_step_list()
604 self.pipeline_changed.emit(self.pipeline_steps)
605 self.status_message.emit(f"Updated step: {edited_step.name}")
607 # SIMPLIFIED: Orchestrator context is automatically available through type-based registry
608 # No need for explicit context management - dual-axis resolver handles it automatically
609 orchestrator = self._get_current_orchestrator()
611 editor = DualEditorWindow(
612 step_data=step_to_edit,
613 is_new=False,
614 on_save_callback=handle_save,
615 orchestrator=orchestrator,
616 gui_config=self.gui_config,
617 parent=self
618 )
619 # Set original step for change detection
620 editor.set_original_step_for_change_detection()
621 editor.show()
622 editor.raise_()
623 editor.activateWindow()
625 def action_auto_load_pipeline(self):
626 """Handle Auto button - load basic_pipeline.py automatically."""
627 if not self.current_plate:
628 self.service_adapter.show_error_dialog("No plate selected")
629 return
631 try:
632 from pathlib import Path
634 # Find basic_pipeline.py relative to openhcs package
635 import openhcs
636 openhcs_root = Path(openhcs.__file__).parent
637 pipeline_file = openhcs_root / "tests" / "basic_pipeline.py"
639 if not pipeline_file.exists():
640 self.service_adapter.show_error_dialog(f"Pipeline file not found: {pipeline_file}")
641 return
643 # Read the file content
644 python_code = pipeline_file.read_text()
646 # Execute the code to get pipeline_steps (same as _handle_edited_pipeline_code)
647 namespace = {}
648 with self._patch_lazy_constructors():
649 exec(python_code, namespace)
651 # Get the pipeline_steps from the namespace
652 if 'pipeline_steps' in namespace:
653 new_pipeline_steps = namespace['pipeline_steps']
654 # Update the pipeline with new steps
655 self.pipeline_steps = new_pipeline_steps
656 self.update_step_list()
657 self.pipeline_changed.emit(self.pipeline_steps)
658 self.status_message.emit(f"Auto-loaded {len(new_pipeline_steps)} steps from basic_pipeline.py")
659 else:
660 raise ValueError("No 'pipeline_steps = [...]' assignment found in basic_pipeline.py")
662 except Exception as e:
663 logger.error(f"Failed to auto-load basic_pipeline.py: {e}")
664 self.service_adapter.show_error_dialog(f"Failed to auto-load pipeline: {str(e)}")
666 def action_code_pipeline(self):
667 """Handle Code Pipeline button - edit pipeline as Python code."""
668 logger.debug("Code button pressed - opening code editor")
670 if not self.current_plate:
671 self.service_adapter.show_error_dialog("No plate selected")
672 return
674 try:
675 # Use complete pipeline steps code generation
676 from openhcs.debug.pickle_to_python import generate_complete_pipeline_steps_code
678 # Generate complete pipeline steps code with imports
679 python_code = generate_complete_pipeline_steps_code(
680 pipeline_steps=list(self.pipeline_steps),
681 clean_mode=True
682 )
684 # Create simple code editor service
685 from openhcs.pyqt_gui.services.simple_code_editor import SimpleCodeEditorService
686 editor_service = SimpleCodeEditorService(self)
688 # Check if user wants external editor (check environment variable)
689 import os
690 use_external = os.environ.get('OPENHCS_USE_EXTERNAL_EDITOR', '').lower() in ('1', 'true', 'yes')
692 # Launch editor with callback and code_type for clean mode toggle
693 editor_service.edit_code(
694 initial_content=python_code,
695 title="Edit Pipeline Steps",
696 callback=self._handle_edited_pipeline_code,
697 use_external=use_external,
698 code_type='pipeline',
699 code_data={'clean_mode': True}
700 )
702 except Exception as e:
703 logger.error(f"Failed to open pipeline code editor: {e}")
704 self.service_adapter.show_error_dialog(f"Failed to open code editor: {str(e)}")
706 def _handle_edited_pipeline_code(self, edited_code: str) -> None:
707 """Handle the edited pipeline code from code editor."""
708 logger.debug("Pipeline code edited, processing changes...")
709 try:
710 # Ensure we have a string
711 if not isinstance(edited_code, str):
712 logger.error(f"Expected string, got {type(edited_code)}: {edited_code}")
713 raise ValueError("Invalid code format received from editor")
715 # CRITICAL FIX: Execute code with lazy dataclass constructor patching to preserve None vs concrete distinction
716 namespace = {}
717 with self._patch_lazy_constructors():
718 exec(edited_code, namespace)
720 # Get the pipeline_steps from the namespace
721 if 'pipeline_steps' in namespace:
722 new_pipeline_steps = namespace['pipeline_steps']
723 # Update the pipeline with new steps
724 self.pipeline_steps = new_pipeline_steps
725 self.update_step_list()
726 self.pipeline_changed.emit(self.pipeline_steps)
727 self.status_message.emit(f"Pipeline updated with {len(new_pipeline_steps)} steps")
728 else:
729 raise ValueError("No 'pipeline_steps = [...]' assignment found in edited code")
731 except (SyntaxError, Exception) as e:
732 logger.error(f"Failed to parse edited pipeline code: {e}")
733 # Re-raise so the code editor can handle it (keep dialog open, move cursor to error line)
734 raise
736 def _patch_lazy_constructors(self):
737 """Context manager that patches lazy dataclass constructors to preserve None vs concrete distinction."""
738 from contextlib import contextmanager
739 from openhcs.core.lazy_placeholder import LazyDefaultPlaceholderService
740 import dataclasses
742 @contextmanager
743 def patch_context():
744 # Store original constructors
745 original_constructors = {}
747 # Find all lazy dataclass types that need patching
748 from openhcs.core.config import LazyZarrConfig, LazyStepMaterializationConfig, LazyWellFilterConfig
749 lazy_types = [LazyZarrConfig, LazyStepMaterializationConfig, LazyWellFilterConfig]
751 # Add any other lazy types that might be used
752 for lazy_type in lazy_types:
753 if LazyDefaultPlaceholderService.has_lazy_resolution(lazy_type):
754 # Store original constructor
755 original_constructors[lazy_type] = lazy_type.__init__
757 # Create patched constructor that uses raw values
758 def create_patched_init(original_init, dataclass_type):
759 def patched_init(self, **kwargs):
760 # Use raw value approach instead of calling original constructor
761 # This prevents lazy resolution during code execution
762 for field in dataclasses.fields(dataclass_type):
763 value = kwargs.get(field.name, None)
764 object.__setattr__(self, field.name, value)
766 # Initialize any required lazy dataclass attributes
767 if hasattr(dataclass_type, '_is_lazy_dataclass'):
768 object.__setattr__(self, '_is_lazy_dataclass', True)
770 return patched_init
772 # Apply the patch
773 lazy_type.__init__ = create_patched_init(original_constructors[lazy_type], lazy_type)
775 try:
776 yield
777 finally:
778 # Restore original constructors
779 for lazy_type, original_init in original_constructors.items():
780 lazy_type.__init__ = original_init
782 return patch_context()
784 def load_pipeline_from_file(self, file_path: Path):
785 """
786 Load pipeline from file with automatic migration for backward compatibility.
788 Args:
789 file_path: Path to pipeline file
790 """
791 try:
792 # Use migration utility to load with backward compatibility
793 from openhcs.io.pipeline_migration import load_pipeline_with_migration
795 steps = load_pipeline_with_migration(file_path)
797 if steps is not None:
798 self.pipeline_steps = steps
799 self.update_step_list()
800 self.pipeline_changed.emit(self.pipeline_steps)
801 self.status_message.emit(f"Loaded {len(steps)} steps from {file_path.name}")
802 else:
803 self.status_message.emit(f"Invalid pipeline format in {file_path.name}")
805 except Exception as e:
806 logger.error(f"Failed to load pipeline: {e}")
807 self.service_adapter.show_error_dialog(f"Failed to load pipeline: {e}")
809 def save_pipeline_to_file(self, file_path: Path):
810 """
811 Save pipeline to file (extracted from Textual version).
813 Args:
814 file_path: Path to save pipeline
815 """
816 try:
817 import dill as pickle
818 with open(file_path, 'wb') as f:
819 pickle.dump(list(self.pipeline_steps), f)
820 self.status_message.emit(f"Saved pipeline to {file_path.name}")
822 except Exception as e:
823 logger.error(f"Failed to save pipeline: {e}")
824 self.service_adapter.show_error_dialog(f"Failed to save pipeline: {e}")
826 def save_pipeline_for_plate(self, plate_path: str, pipeline: List[FunctionStep]):
827 """
828 Save pipeline for specific plate (extracted from Textual version).
830 Args:
831 plate_path: Path of the plate
832 pipeline: Pipeline steps to save
833 """
834 self.plate_pipelines[plate_path] = pipeline
835 logger.debug(f"Saved pipeline for plate: {plate_path}")
837 def set_current_plate(self, plate_path: str):
838 """
839 Set current plate and load its pipeline (extracted from Textual version).
841 Args:
842 plate_path: Path of the current plate
843 """
844 self.current_plate = plate_path
846 # Load pipeline for the new plate
847 if plate_path:
848 plate_pipeline = self.plate_pipelines.get(plate_path, [])
849 self.pipeline_steps = plate_pipeline
850 else:
851 self.pipeline_steps = []
853 self.update_step_list()
854 self.update_button_states()
855 logger.debug(f"Current plate changed: {plate_path}")
857 def on_orchestrator_config_changed(self, plate_path: str, effective_config):
858 """
859 Handle orchestrator configuration changes for placeholder refresh.
861 Args:
862 plate_path: Path of the plate whose orchestrator config changed
863 effective_config: The orchestrator's new effective configuration
864 """
865 # Only refresh if this is for the current plate
866 if plate_path == self.current_plate:
867 logger.debug(f"Refreshing placeholders for orchestrator config change: {plate_path}")
869 # SIMPLIFIED: Orchestrator context is automatically available through type-based registry
870 # No need for explicit context management - dual-axis resolver handles it automatically
871 orchestrator = self._get_current_orchestrator()
872 if orchestrator:
873 # Trigger refresh of any open configuration windows or step forms
874 # The type-based registry ensures they resolve against the updated orchestrator config
875 logger.debug(f"Step forms will now resolve against updated orchestrator config for: {plate_path}")
876 else:
877 logger.debug(f"No orchestrator found for config refresh: {plate_path}")
879 # ========== UI Helper Methods ==========
881 def update_step_list(self):
882 """Update the step list widget using selection preservation mixin."""
883 def format_step_item(step, step_index):
884 """Format step item for display."""
885 display_text, step_name = self.format_item_for_display(step)
886 return display_text, step_index # Store index instead of step object
888 def update_func():
889 """Update function that clears and rebuilds the list."""
890 self.step_list.clear()
892 for step_index, step in enumerate(self.pipeline_steps):
893 display_text, index_data = format_step_item(step, step_index)
894 item = QListWidgetItem(display_text)
895 item.setData(Qt.ItemDataRole.UserRole, index_data) # Store index, not step
896 item.setData(Qt.ItemDataRole.UserRole + 1, not step.enabled) # Store disabled status for strikethrough
897 item.setToolTip(self._create_step_tooltip(step))
898 self.step_list.addItem(item)
900 # Use utility to preserve selection during update
901 preserve_selection_during_update(
902 self.step_list,
903 lambda item_data: getattr(item_data, 'name', str(item_data)),
904 lambda: bool(self.pipeline_steps),
905 update_func
906 )
907 self.update_button_states()
909 def get_selected_steps(self) -> List[FunctionStep]:
910 """
911 Get currently selected steps.
913 Returns:
914 List of selected FunctionStep objects
915 """
916 selected_items = []
917 for item in self.step_list.selectedItems():
918 step_index = item.data(Qt.ItemDataRole.UserRole)
919 if step_index is not None and 0 <= step_index < len(self.pipeline_steps):
920 selected_items.append(self.pipeline_steps[step_index])
921 return selected_items
923 def update_button_states(self):
924 """Update button enabled/disabled states based on mathematical constraints (mirrors Textual TUI)."""
925 has_plate = bool(self.current_plate)
926 is_initialized = self._is_current_plate_initialized()
927 has_steps = len(self.pipeline_steps) > 0
928 has_selection = len(self.get_selected_steps()) > 0
930 # Mathematical constraints (mirrors Textual TUI logic):
931 # - Pipeline editing requires initialization
932 # - Step operations require steps to exist
933 # - Edit requires valid selection
934 self.buttons["add_step"].setEnabled(has_plate and is_initialized)
935 self.buttons["auto_load_pipeline"].setEnabled(has_plate and is_initialized)
936 self.buttons["del_step"].setEnabled(has_steps)
937 self.buttons["edit_step"].setEnabled(has_steps and has_selection)
938 self.buttons["code_pipeline"].setEnabled(has_plate and is_initialized) # Same as add button - orchestrator init is sufficient
940 def update_status(self, message: str):
941 """
942 Update status label.
944 Args:
945 message: Status message to display
946 """
947 self.status_label.setText(message)
949 def on_selection_changed(self):
950 """Handle step list selection changes using utility."""
951 def on_selected(selected_steps):
952 self.selected_step = getattr(selected_steps[0], 'name', '')
953 self.step_selected.emit(selected_steps[0])
955 def on_cleared():
956 self.selected_step = ""
958 # Use utility to handle selection with prevention
959 handle_selection_change_with_prevention(
960 self.step_list,
961 self.get_selected_steps,
962 lambda item_data: getattr(item_data, 'name', str(item_data)),
963 lambda: bool(self.pipeline_steps),
964 lambda: self.selected_step,
965 on_selected,
966 on_cleared
967 )
969 self.update_button_states()
971 def on_item_double_clicked(self, item: QListWidgetItem):
972 """Handle double-click on step item."""
973 step_index = item.data(Qt.ItemDataRole.UserRole)
974 if step_index is not None and 0 <= step_index < len(self.pipeline_steps):
975 # Double-click triggers edit
976 self.action_edit_step()
978 def on_steps_reordered(self, from_index: int, to_index: int):
979 """
980 Handle step reordering from drag and drop.
982 Args:
983 from_index: Original position of the moved step
984 to_index: New position of the moved step
985 """
986 # Update the underlying pipeline_steps list to match the visual order
987 current_steps = list(self.pipeline_steps)
989 # Move the step in the data model
990 step = current_steps.pop(from_index)
991 current_steps.insert(to_index, step)
993 # Update pipeline steps
994 self.pipeline_steps = current_steps
996 # Emit pipeline changed signal to notify other components
997 self.pipeline_changed.emit(self.pipeline_steps)
999 # Update status message
1000 step_name = getattr(step, 'name', 'Unknown Step')
1001 direction = "up" if to_index < from_index else "down"
1002 self.status_message.emit(f"Moved step '{step_name}' {direction}")
1004 logger.debug(f"Reordered step '{step_name}' from index {from_index} to {to_index}")
1006 def on_pipeline_changed(self, steps: List[FunctionStep]):
1007 """
1008 Handle pipeline changes.
1010 Args:
1011 steps: New pipeline steps
1012 """
1013 # Save pipeline to current plate if one is selected
1014 if self.current_plate:
1015 self.save_pipeline_for_plate(self.current_plate, steps)
1017 logger.debug(f"Pipeline changed: {len(steps)} steps")
1019 def _is_current_plate_initialized(self) -> bool:
1020 """Check if current plate has an initialized orchestrator (mirrors Textual TUI)."""
1021 if not self.current_plate:
1022 return False
1024 # Get plate manager from main window
1025 main_window = self._find_main_window()
1026 if not main_window:
1027 return False
1029 # Get plate manager widget from floating windows
1030 plate_manager_window = main_window.floating_windows.get("plate_manager")
1031 if not plate_manager_window:
1032 return False
1034 layout = plate_manager_window.layout()
1035 if not layout or layout.count() == 0:
1036 return False
1038 plate_manager_widget = layout.itemAt(0).widget()
1039 if not hasattr(plate_manager_widget, 'orchestrators'):
1040 return False
1042 orchestrator = plate_manager_widget.orchestrators.get(self.current_plate)
1043 if orchestrator is None:
1044 return False
1046 # Check if orchestrator is in an initialized state (mirrors Textual TUI logic)
1047 from openhcs.constants.constants import OrchestratorState
1048 return orchestrator.state in [OrchestratorState.READY, OrchestratorState.COMPILED,
1049 OrchestratorState.COMPLETED, OrchestratorState.COMPILE_FAILED,
1050 OrchestratorState.EXEC_FAILED]
1054 def _get_current_orchestrator(self) -> Optional[PipelineOrchestrator]:
1055 """Get the orchestrator for the currently selected plate."""
1056 if not self.current_plate:
1057 return None
1058 main_window = self._find_main_window()
1059 if not main_window:
1060 return None
1061 plate_manager_window = main_window.floating_windows.get("plate_manager")
1062 if not plate_manager_window:
1063 return None
1064 layout = plate_manager_window.layout()
1065 if not layout or layout.count() == 0:
1066 return None
1067 plate_manager_widget = layout.itemAt(0).widget()
1068 if not hasattr(plate_manager_widget, 'orchestrators'):
1069 return None
1070 return plate_manager_widget.orchestrators.get(self.current_plate)
1073 def _find_main_window(self):
1074 """Find the main window by traversing parent hierarchy."""
1075 widget = self
1076 while widget:
1077 if hasattr(widget, 'floating_windows'):
1078 return widget
1079 widget = widget.parent()
1080 return None
1082 def on_config_changed(self, new_config: GlobalPipelineConfig):
1083 """
1084 Handle global configuration changes.
1086 Args:
1087 new_config: New global configuration
1088 """
1089 self.global_config = new_config
1091 # CRITICAL FIX: Refresh all placeholders when global config changes
1092 # This ensures pipeline config editor shows updated inherited values
1093 if hasattr(self, 'form_manager') and self.form_manager:
1094 self.form_manager.refresh_placeholder_text()
1095 logger.info("Refreshed pipeline config placeholders after global config change")