Coverage for openhcs/pyqt_gui/widgets/pipeline_editor.py: 0.0%
536 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +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 asyncio
10import inspect
11import contextlib
12from typing import List, Dict, Optional, Callable, Tuple
13from pathlib import Path
15from PyQt6.QtWidgets import (
16 QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QListWidget,
17 QListWidgetItem, QLabel, QMessageBox, QFileDialog, QFrame,
18 QSplitter, QTextEdit, QScrollArea, QStyledItemDelegate, QStyle,
19 QStyleOptionViewItem, QApplication
20)
21from PyQt6.QtCore import Qt, pyqtSignal, QMimeData
22from PyQt6.QtGui import QFont, QDrag, QPainter, QColor, QPen, QFontMetrics
24from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator
25from openhcs.core.config import GlobalPipelineConfig
26from openhcs.config_framework.global_config import set_current_global_config, get_current_global_config
27from openhcs.io.filemanager import FileManager
28from openhcs.core.steps.function_step import FunctionStep
29from openhcs.pyqt_gui.widgets.mixins import (
30 preserve_selection_during_update,
31 handle_selection_change_with_prevention
32)
33from openhcs.pyqt_gui.shared.style_generator import StyleSheetGenerator
34from openhcs.pyqt_gui.shared.color_scheme import PyQt6ColorScheme
35from openhcs.pyqt_gui.config import PyQtGUIConfig, get_default_pyqt_gui_config
37logger = logging.getLogger(__name__)
40class StepListItemDelegate(QStyledItemDelegate):
41 """Custom delegate to render step name in white and preview text in grey without breaking hover/selection/borders."""
42 def __init__(self, name_color: QColor, preview_color: QColor, parent=None):
43 super().__init__(parent)
44 self.name_color = name_color
45 self.preview_color = preview_color
47 def paint(self, painter: QPainter, option, index) -> None:
48 # Prepare a copy to let style draw backgrounds, hover, selection, borders, etc.
49 opt = QStyleOptionViewItem(option)
50 self.initStyleOption(opt, index)
52 # Capture text and prevent default text draw
53 text = opt.text or ""
54 opt.text = ""
56 painter.save()
57 # Let the current style paint the item (background, selection, hover, separators)
58 style = opt.widget.style() if opt.widget else QApplication.style()
59 style.drawControl(QStyle.ControlElement.CE_ItemViewItem, opt, painter, opt.widget)
61 # Now custom-draw the text with mixed colors
62 rect = opt.rect.adjusted(6, 0, -6, 0)
63 painter.setFont(opt.font)
64 fm = QFontMetrics(opt.font)
65 baseline_y = rect.y() + (rect.height() + fm.ascent() - fm.descent()) // 2
67 sep_idx = text.find(" (")
68 if sep_idx != -1 and text.endswith(")"):
69 name_part = text[:sep_idx]
70 preview_part = text[sep_idx:]
72 painter.setPen(QPen(self.name_color))
73 painter.drawText(rect.x(), baseline_y, name_part)
74 name_width = fm.horizontalAdvance(name_part)
76 painter.setPen(QPen(self.preview_color))
77 painter.drawText(rect.x() + name_width, baseline_y, preview_part)
78 else:
79 painter.setPen(QPen(self.name_color))
80 painter.drawText(rect.x(), baseline_y, text)
82 painter.restore()
84class ReorderableListWidget(QListWidget):
85 """
86 Custom QListWidget that properly handles drag and drop reordering.
87 Emits a signal when items are moved so the parent can update the data model.
88 """
90 items_reordered = pyqtSignal(int, int) # from_index, to_index
92 def __init__(self, parent=None):
93 super().__init__(parent)
94 self.setDragDropMode(QListWidget.DragDropMode.InternalMove)
96 def dropEvent(self, event):
97 """Handle drop events and emit reorder signal."""
98 # Get the item being dropped and its original position
99 source_item = self.currentItem()
100 if not source_item:
101 super().dropEvent(event)
102 return
104 source_index = self.row(source_item)
106 # Let the default drop behavior happen first
107 super().dropEvent(event)
109 # Find the new position of the item
110 target_index = self.row(source_item)
112 # Only emit signal if position actually changed
113 if source_index != target_index:
114 self.items_reordered.emit(source_index, target_index)
117class PipelineEditorWidget(QWidget):
118 """
119 PyQt6 Pipeline Editor Widget.
121 Manages pipeline steps with add, edit, delete, load, save functionality.
122 Preserves all business logic from Textual version with clean PyQt6 UI.
123 """
125 # Signals
126 pipeline_changed = pyqtSignal(list) # List[FunctionStep]
127 step_selected = pyqtSignal(object) # FunctionStep
128 status_message = pyqtSignal(str) # status message
130 def __init__(self, file_manager: FileManager, service_adapter,
131 color_scheme: Optional[PyQt6ColorScheme] = None, gui_config: Optional[PyQtGUIConfig] = None, parent=None):
132 """
133 Initialize the pipeline editor widget.
135 Args:
136 file_manager: FileManager instance for file operations
137 service_adapter: PyQt service adapter for dialogs and operations
138 color_scheme: Color scheme for styling (optional, uses service adapter if None)
139 gui_config: GUI configuration (optional, uses default if None)
140 parent: Parent widget
141 """
142 super().__init__(parent)
144 # Core dependencies
145 self.file_manager = file_manager
146 self.service_adapter = service_adapter
147 self.global_config = service_adapter.get_global_config()
148 self.gui_config = gui_config or get_default_pyqt_gui_config()
150 # Initialize color scheme and style generator
151 self.color_scheme = color_scheme or service_adapter.get_current_color_scheme()
152 self.style_generator = StyleSheetGenerator(self.color_scheme)
154 # Business logic state (extracted from Textual version)
155 self.pipeline_steps: List[FunctionStep] = []
156 self.current_plate: str = ""
157 self.selected_step: str = ""
158 self.plate_pipelines: Dict[str, List[FunctionStep]] = {} # Per-plate pipeline storage
160 # UI components
161 self.step_list: Optional[QListWidget] = None
162 self.buttons: Dict[str, QPushButton] = {}
163 self.status_label: Optional[QLabel] = None
165 # Reference to plate manager (set externally)
166 self.plate_manager = None
168 # Setup UI
169 self.setup_ui()
170 self.setup_connections()
171 self.update_button_states()
173 logger.debug("Pipeline editor widget initialized")
175 # ========== UI Setup ==========
177 def setup_ui(self):
178 """Setup the user interface."""
179 layout = QVBoxLayout(self)
180 layout.setContentsMargins(2, 2, 2, 2)
181 layout.setSpacing(2)
183 # Title
184 title_label = QLabel("Pipeline Editor")
185 title_label.setFont(QFont("Arial", 12, QFont.Weight.Bold))
186 title_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.text_accent)}; padding: 5px;")
187 layout.addWidget(title_label)
189 # Main content splitter
190 splitter = QSplitter(Qt.Orientation.Vertical)
191 layout.addWidget(splitter)
193 # Pipeline steps list
194 self.step_list = ReorderableListWidget()
195 self.step_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
196 self.step_list.setStyleSheet(f"""
197 QListWidget {{
198 background-color: {self.color_scheme.to_hex(self.color_scheme.panel_bg)};
199 color: {self.color_scheme.to_hex(self.color_scheme.text_primary)};
200 border: none;
201 padding: 5px;
202 }}
203 QListWidget::item {{
204 padding: 8px;
205 border: none;
206 border-radius: 3px;
207 margin: 2px;
208 }}
209 QListWidget::item:selected {{
210 background-color: {self.color_scheme.to_hex(self.color_scheme.selection_bg)};
211 }}
212 QListWidget::item:hover {{
213 background-color: {self.color_scheme.to_hex(self.color_scheme.hover_bg)};
214 }}
215 """)
216 # Set custom delegate to render white name and grey preview
217 try:
218 name_color = QColor(self.color_scheme.to_hex(self.color_scheme.text_primary))
219 preview_color = QColor(self.color_scheme.to_hex(self.color_scheme.text_disabled))
220 self.step_list.setItemDelegate(StepListItemDelegate(name_color, preview_color, self.step_list))
221 except Exception:
222 # Fallback silently if color scheme isn't ready
223 pass
224 splitter.addWidget(self.step_list)
226 # Button panel
227 button_panel = self.create_button_panel()
228 splitter.addWidget(button_panel)
230 # Status section
231 status_frame = self.create_status_section()
232 layout.addWidget(status_frame)
234 # Set splitter proportions
235 splitter.setSizes([400, 120])
237 def create_button_panel(self) -> QWidget:
238 """
239 Create the button panel with all pipeline actions.
241 Returns:
242 Widget containing action buttons
243 """
244 panel = QWidget()
245 panel.setStyleSheet(f"""
246 QWidget {{
247 background-color: {self.color_scheme.to_hex(self.color_scheme.window_bg)};
248 border: none;
249 padding: 0px;
250 }}
251 """)
253 layout = QVBoxLayout(panel)
254 layout.setContentsMargins(0, 0, 0, 0)
255 layout.setSpacing(0)
257 # Button configurations (extracted from Textual version)
258 button_configs = [
259 ("Add", "add_step", "Add new pipeline step"),
260 ("Del", "del_step", "Delete selected steps"),
261 ("Edit", "edit_step", "Edit selected step"),
262 ("Load", "load_pipeline", "Load pipeline from file"),
263 ("Save", "save_pipeline", "Save pipeline to file"),
264 ("Code", "code_pipeline", "Edit pipeline as Python code"),
265 ]
267 # Create buttons in a single row
268 row_layout = QHBoxLayout()
269 row_layout.setContentsMargins(2, 2, 2, 2)
270 row_layout.setSpacing(2)
272 for name, action, tooltip in button_configs:
273 button = QPushButton(name)
274 button.setToolTip(tooltip)
275 button.setMinimumHeight(30)
276 button.setStyleSheet(self.style_generator.generate_button_style())
278 # Connect button to action
279 button.clicked.connect(lambda checked, a=action: self.handle_button_action(a))
281 self.buttons[action] = button
282 row_layout.addWidget(button)
284 layout.addLayout(row_layout)
286 # Set maximum height to constrain the button panel
287 panel.setMaximumHeight(40)
289 return panel
291 def create_status_section(self) -> QWidget:
292 """
293 Create the status section.
295 Returns:
296 Widget containing status information
297 """
298 frame = QWidget()
299 frame.setStyleSheet(f"""
300 QWidget {{
301 background-color: {self.color_scheme.to_hex(self.color_scheme.window_bg)};
302 border: none;
303 padding: 2px;
304 }}
305 """)
307 layout = QVBoxLayout(frame)
308 layout.setContentsMargins(2, 2, 2, 2)
309 layout.setSpacing(2)
311 # Status label
312 self.status_label = QLabel("Ready")
313 self.status_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.status_success)}; font-weight: bold;")
314 layout.addWidget(self.status_label)
316 return frame
318 def setup_connections(self):
319 """Setup signal/slot connections."""
320 # Step list selection
321 self.step_list.itemSelectionChanged.connect(self.on_selection_changed)
322 self.step_list.itemDoubleClicked.connect(self.on_item_double_clicked)
324 # Step list reordering
325 self.step_list.items_reordered.connect(self.on_steps_reordered)
327 # Internal signals
328 self.status_message.connect(self.update_status)
329 self.pipeline_changed.connect(self.on_pipeline_changed)
331 def handle_button_action(self, action: str):
332 """
333 Handle button actions (extracted from Textual version).
335 Args:
336 action: Action identifier
337 """
338 # Action mapping (preserved from Textual version)
339 action_map = {
340 "add_step": self.action_add_step,
341 "del_step": self.action_delete_step,
342 "edit_step": self.action_edit_step,
343 "load_pipeline": self.action_load_pipeline,
344 "save_pipeline": self.action_save_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, 'napari_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_load_pipeline(self):
626 """Handle Load Pipeline button (adapted from Textual version)."""
628 from openhcs.core.path_cache import PathCacheKey
630 # Use cached file dialog (mirrors Textual TUI pattern)
631 file_path = self.service_adapter.show_cached_file_dialog(
632 cache_key=PathCacheKey.PIPELINE_FILES,
633 title="Load Pipeline",
634 file_filter="Pipeline Files (*.pipeline);;All Files (*)",
635 mode="open"
636 )
638 if file_path:
639 self.load_pipeline_from_file(file_path)
641 def action_save_pipeline(self):
642 """Handle Save Pipeline button (adapted from Textual version)."""
643 if not self.pipeline_steps:
644 self.service_adapter.show_error_dialog("No pipeline steps to save.")
645 return
647 from openhcs.core.path_cache import PathCacheKey
649 # Use cached file dialog (mirrors Textual TUI pattern)
650 file_path = self.service_adapter.show_cached_file_dialog(
651 cache_key=PathCacheKey.PIPELINE_FILES,
652 title="Save Pipeline",
653 file_filter="Pipeline Files (*.pipeline);;All Files (*)",
654 mode="save"
655 )
657 if file_path:
658 self.save_pipeline_to_file(file_path)
660 def action_code_pipeline(self):
661 """Handle Code Pipeline button - edit pipeline as Python code."""
662 logger.debug("Code button pressed - opening code editor")
664 if not self.current_plate:
665 self.service_adapter.show_error_dialog("No plate selected")
666 return
668 try:
669 # Use complete pipeline steps code generation
670 from openhcs.debug.pickle_to_python import generate_complete_pipeline_steps_code
672 # Generate complete pipeline steps code with imports
673 python_code = generate_complete_pipeline_steps_code(
674 pipeline_steps=list(self.pipeline_steps),
675 clean_mode=True
676 )
678 # Create simple code editor service
679 from openhcs.pyqt_gui.services.simple_code_editor import SimpleCodeEditorService
680 editor_service = SimpleCodeEditorService(self)
682 # Check if user wants external editor (check environment variable)
683 import os
684 use_external = os.environ.get('OPENHCS_USE_EXTERNAL_EDITOR', '').lower() in ('1', 'true', 'yes')
686 # Launch editor with callback
687 editor_service.edit_code(
688 initial_content=python_code,
689 title="Edit Pipeline Steps",
690 callback=self._handle_edited_pipeline_code,
691 use_external=use_external
692 )
694 except Exception as e:
695 logger.error(f"Failed to open pipeline code editor: {e}")
696 self.service_adapter.show_error_dialog(f"Failed to open code editor: {str(e)}")
698 def _handle_edited_pipeline_code(self, edited_code: str) -> None:
699 """Handle the edited pipeline code from code editor."""
700 logger.debug("Pipeline code edited, processing changes...")
701 try:
702 # Ensure we have a string
703 if not isinstance(edited_code, str):
704 logger.error(f"Expected string, got {type(edited_code)}: {edited_code}")
705 self.service_adapter.show_error_dialog("Invalid code format received from editor")
706 return
708 # CRITICAL FIX: Execute code with lazy dataclass constructor patching to preserve None vs concrete distinction
709 namespace = {}
710 with self._patch_lazy_constructors():
711 exec(edited_code, namespace)
713 # Get the pipeline_steps from the namespace
714 if 'pipeline_steps' in namespace:
715 new_pipeline_steps = namespace['pipeline_steps']
716 # Update the pipeline with new steps
717 self.pipeline_steps = new_pipeline_steps
718 self.update_step_list()
719 self.pipeline_changed.emit(self.pipeline_steps)
720 self.status_message.emit(f"Pipeline updated with {len(new_pipeline_steps)} steps")
721 else:
722 self.service_adapter.show_error_dialog("No 'pipeline_steps = [...]' assignment found in edited code")
724 except SyntaxError as e:
725 self.service_adapter.show_error_dialog(f"Invalid Python syntax: {e}")
726 except Exception as e:
727 logger.error(f"Failed to parse edited pipeline code: {e}")
728 self.service_adapter.show_error_dialog(f"Failed to parse pipeline code: {str(e)}")
730 def _patch_lazy_constructors(self):
731 """Context manager that patches lazy dataclass constructors to preserve None vs concrete distinction."""
732 from contextlib import contextmanager
733 from openhcs.core.lazy_placeholder import LazyDefaultPlaceholderService
734 import dataclasses
736 @contextmanager
737 def patch_context():
738 # Store original constructors
739 original_constructors = {}
741 # Find all lazy dataclass types that need patching
742 from openhcs.core.config import LazyZarrConfig, LazyStepMaterializationConfig, LazyWellFilterConfig
743 lazy_types = [LazyZarrConfig, LazyStepMaterializationConfig, LazyWellFilterConfig]
745 # Add any other lazy types that might be used
746 for lazy_type in lazy_types:
747 if LazyDefaultPlaceholderService.has_lazy_resolution(lazy_type):
748 # Store original constructor
749 original_constructors[lazy_type] = lazy_type.__init__
751 # Create patched constructor that uses raw values
752 def create_patched_init(original_init, dataclass_type):
753 def patched_init(self, **kwargs):
754 # Use raw value approach instead of calling original constructor
755 # This prevents lazy resolution during code execution
756 for field in dataclasses.fields(dataclass_type):
757 value = kwargs.get(field.name, None)
758 object.__setattr__(self, field.name, value)
760 # Initialize any required lazy dataclass attributes
761 if hasattr(dataclass_type, '_is_lazy_dataclass'):
762 object.__setattr__(self, '_is_lazy_dataclass', True)
764 return patched_init
766 # Apply the patch
767 lazy_type.__init__ = create_patched_init(original_constructors[lazy_type], lazy_type)
769 try:
770 yield
771 finally:
772 # Restore original constructors
773 for lazy_type, original_init in original_constructors.items():
774 lazy_type.__init__ = original_init
776 return patch_context()
778 def load_pipeline_from_file(self, file_path: Path):
779 """
780 Load pipeline from file with automatic migration for backward compatibility.
782 Args:
783 file_path: Path to pipeline file
784 """
785 try:
786 # Use migration utility to load with backward compatibility
787 from openhcs.io.pipeline_migration import load_pipeline_with_migration
789 steps = load_pipeline_with_migration(file_path)
791 if steps is not None:
792 self.pipeline_steps = steps
793 self.update_step_list()
794 self.pipeline_changed.emit(self.pipeline_steps)
795 self.status_message.emit(f"Loaded {len(steps)} steps from {file_path.name}")
796 else:
797 self.status_message.emit(f"Invalid pipeline format in {file_path.name}")
799 except Exception as e:
800 logger.error(f"Failed to load pipeline: {e}")
801 self.service_adapter.show_error_dialog(f"Failed to load pipeline: {e}")
803 def save_pipeline_to_file(self, file_path: Path):
804 """
805 Save pipeline to file (extracted from Textual version).
807 Args:
808 file_path: Path to save pipeline
809 """
810 try:
811 import dill as pickle
812 with open(file_path, 'wb') as f:
813 pickle.dump(list(self.pipeline_steps), f)
814 self.status_message.emit(f"Saved pipeline to {file_path.name}")
816 except Exception as e:
817 logger.error(f"Failed to save pipeline: {e}")
818 self.service_adapter.show_error_dialog(f"Failed to save pipeline: {e}")
820 def save_pipeline_for_plate(self, plate_path: str, pipeline: List[FunctionStep]):
821 """
822 Save pipeline for specific plate (extracted from Textual version).
824 Args:
825 plate_path: Path of the plate
826 pipeline: Pipeline steps to save
827 """
828 self.plate_pipelines[plate_path] = pipeline
829 logger.debug(f"Saved pipeline for plate: {plate_path}")
831 def set_current_plate(self, plate_path: str):
832 """
833 Set current plate and load its pipeline (extracted from Textual version).
835 Args:
836 plate_path: Path of the current plate
837 """
838 self.current_plate = plate_path
840 # Load pipeline for the new plate
841 if plate_path:
842 plate_pipeline = self.plate_pipelines.get(plate_path, [])
843 self.pipeline_steps = plate_pipeline
844 else:
845 self.pipeline_steps = []
847 self.update_step_list()
848 self.update_button_states()
849 logger.debug(f"Current plate changed: {plate_path}")
851 def on_orchestrator_config_changed(self, plate_path: str, effective_config):
852 """
853 Handle orchestrator configuration changes for placeholder refresh.
855 Args:
856 plate_path: Path of the plate whose orchestrator config changed
857 effective_config: The orchestrator's new effective configuration
858 """
859 # Only refresh if this is for the current plate
860 if plate_path == self.current_plate:
861 logger.debug(f"Refreshing placeholders for orchestrator config change: {plate_path}")
863 # SIMPLIFIED: Orchestrator context is automatically available through type-based registry
864 # No need for explicit context management - dual-axis resolver handles it automatically
865 orchestrator = self._get_current_orchestrator()
866 if orchestrator:
867 # Trigger refresh of any open configuration windows or step forms
868 # The type-based registry ensures they resolve against the updated orchestrator config
869 logger.debug(f"Step forms will now resolve against updated orchestrator config for: {plate_path}")
870 else:
871 logger.debug(f"No orchestrator found for config refresh: {plate_path}")
873 # ========== UI Helper Methods ==========
875 def update_step_list(self):
876 """Update the step list widget using selection preservation mixin."""
877 def format_step_item(step):
878 """Format step item for display."""
879 display_text, step_name = self.format_item_for_display(step)
880 return display_text, step
882 def update_func():
883 """Update function that clears and rebuilds the list."""
884 self.step_list.clear()
886 for step in self.pipeline_steps:
887 display_text, step_data = format_step_item(step)
888 item = QListWidgetItem(display_text)
889 item.setData(Qt.ItemDataRole.UserRole, step_data)
890 item.setToolTip(self._create_step_tooltip(step))
891 self.step_list.addItem(item)
893 # Use utility to preserve selection during update
894 preserve_selection_during_update(
895 self.step_list,
896 lambda item_data: getattr(item_data, 'name', str(item_data)),
897 lambda: bool(self.pipeline_steps),
898 update_func
899 )
900 self.update_button_states()
902 def get_selected_steps(self) -> List[FunctionStep]:
903 """
904 Get currently selected steps.
906 Returns:
907 List of selected FunctionStep objects
908 """
909 selected_items = []
910 for item in self.step_list.selectedItems():
911 step_data = item.data(Qt.ItemDataRole.UserRole)
912 if step_data:
913 selected_items.append(step_data)
914 return selected_items
916 def update_button_states(self):
917 """Update button enabled/disabled states based on mathematical constraints (mirrors Textual TUI)."""
918 has_plate = bool(self.current_plate)
919 is_initialized = self._is_current_plate_initialized()
920 has_steps = len(self.pipeline_steps) > 0
921 has_selection = len(self.get_selected_steps()) > 0
923 # Mathematical constraints (mirrors Textual TUI logic):
924 # - Pipeline editing requires initialization
925 # - Step operations require steps to exist
926 # - Edit requires valid selection
927 self.buttons["add_step"].setEnabled(has_plate and is_initialized)
928 self.buttons["load_pipeline"].setEnabled(has_plate and is_initialized)
929 self.buttons["del_step"].setEnabled(has_steps)
930 self.buttons["edit_step"].setEnabled(has_steps and has_selection)
931 self.buttons["save_pipeline"].setEnabled(has_steps)
932 self.buttons["code_pipeline"].setEnabled(has_plate and is_initialized) # Same as add button - orchestrator init is sufficient
934 def update_status(self, message: str):
935 """
936 Update status label.
938 Args:
939 message: Status message to display
940 """
941 self.status_label.setText(message)
943 def on_selection_changed(self):
944 """Handle step list selection changes using utility."""
945 def on_selected(selected_steps):
946 self.selected_step = getattr(selected_steps[0], 'name', '')
947 self.step_selected.emit(selected_steps[0])
949 def on_cleared():
950 self.selected_step = ""
952 # Use utility to handle selection with prevention
953 handle_selection_change_with_prevention(
954 self.step_list,
955 self.get_selected_steps,
956 lambda item_data: getattr(item_data, 'name', str(item_data)),
957 lambda: bool(self.pipeline_steps),
958 lambda: self.selected_step,
959 on_selected,
960 on_cleared
961 )
963 self.update_button_states()
965 def on_item_double_clicked(self, item: QListWidgetItem):
966 """Handle double-click on step item."""
967 step_data = item.data(Qt.ItemDataRole.UserRole)
968 if step_data:
969 # Double-click triggers edit
970 self.action_edit_step()
972 def on_steps_reordered(self, from_index: int, to_index: int):
973 """
974 Handle step reordering from drag and drop.
976 Args:
977 from_index: Original position of the moved step
978 to_index: New position of the moved step
979 """
980 # Update the underlying pipeline_steps list to match the visual order
981 current_steps = list(self.pipeline_steps)
983 # Move the step in the data model
984 step = current_steps.pop(from_index)
985 current_steps.insert(to_index, step)
987 # Update pipeline steps
988 self.pipeline_steps = current_steps
990 # Emit pipeline changed signal to notify other components
991 self.pipeline_changed.emit(self.pipeline_steps)
993 # Update status message
994 step_name = getattr(step, 'name', 'Unknown Step')
995 direction = "up" if to_index < from_index else "down"
996 self.status_message.emit(f"Moved step '{step_name}' {direction}")
998 logger.debug(f"Reordered step '{step_name}' from index {from_index} to {to_index}")
1000 def on_pipeline_changed(self, steps: List[FunctionStep]):
1001 """
1002 Handle pipeline changes.
1004 Args:
1005 steps: New pipeline steps
1006 """
1007 # Save pipeline to current plate if one is selected
1008 if self.current_plate:
1009 self.save_pipeline_for_plate(self.current_plate, steps)
1011 logger.debug(f"Pipeline changed: {len(steps)} steps")
1013 def _is_current_plate_initialized(self) -> bool:
1014 """Check if current plate has an initialized orchestrator (mirrors Textual TUI)."""
1015 if not self.current_plate:
1016 return False
1018 # Get plate manager from main window
1019 main_window = self._find_main_window()
1020 if not main_window:
1021 return False
1023 # Get plate manager widget from floating windows
1024 plate_manager_window = main_window.floating_windows.get("plate_manager")
1025 if not plate_manager_window:
1026 return False
1028 layout = plate_manager_window.layout()
1029 if not layout or layout.count() == 0:
1030 return False
1032 plate_manager_widget = layout.itemAt(0).widget()
1033 if not hasattr(plate_manager_widget, 'orchestrators'):
1034 return False
1036 orchestrator = plate_manager_widget.orchestrators.get(self.current_plate)
1037 if orchestrator is None:
1038 return False
1040 # Check if orchestrator is in an initialized state (mirrors Textual TUI logic)
1041 from openhcs.constants.constants import OrchestratorState
1042 return orchestrator.state in [OrchestratorState.READY, OrchestratorState.COMPILED,
1043 OrchestratorState.COMPLETED, OrchestratorState.COMPILE_FAILED,
1044 OrchestratorState.EXEC_FAILED]
1048 def _get_current_orchestrator(self) -> Optional[PipelineOrchestrator]:
1049 """Get the orchestrator for the currently selected plate."""
1050 if not self.current_plate:
1051 return None
1052 main_window = self._find_main_window()
1053 if not main_window:
1054 return None
1055 plate_manager_window = main_window.floating_windows.get("plate_manager")
1056 if not plate_manager_window:
1057 return None
1058 layout = plate_manager_window.layout()
1059 if not layout or layout.count() == 0:
1060 return None
1061 plate_manager_widget = layout.itemAt(0).widget()
1062 if not hasattr(plate_manager_widget, 'orchestrators'):
1063 return None
1064 return plate_manager_widget.orchestrators.get(self.current_plate)
1067 def _find_main_window(self):
1068 """Find the main window by traversing parent hierarchy."""
1069 widget = self
1070 while widget:
1071 if hasattr(widget, 'floating_windows'):
1072 return widget
1073 widget = widget.parent()
1074 return None
1076 def on_config_changed(self, new_config: GlobalPipelineConfig):
1077 """
1078 Handle global configuration changes.
1080 Args:
1081 new_config: New global configuration
1082 """
1083 self.global_config = new_config
1085 # CRITICAL FIX: Refresh all placeholders when global config changes
1086 # This ensures pipeline config editor shows updated inherited values
1087 if hasattr(self, 'form_manager') and self.form_manager:
1088 self.form_manager.refresh_placeholder_text()
1089 logger.info("Refreshed pipeline config placeholders after global config change")