Coverage for openhcs/pyqt_gui/widgets/pipeline_editor.py: 0.0%
343 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +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
11from typing import List, Dict, Optional, Callable, Tuple
12from pathlib import Path
14from PyQt6.QtWidgets import (
15 QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QListWidget,
16 QListWidgetItem, QLabel, QMessageBox, QFileDialog, QFrame,
17 QSplitter, QTextEdit, QScrollArea
18)
19from PyQt6.QtCore import Qt, pyqtSignal, QMimeData
20from PyQt6.QtGui import QFont, QDrag
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
32logger = logging.getLogger(__name__)
35class ReorderableListWidget(QListWidget):
36 """
37 Custom QListWidget that properly handles drag and drop reordering.
38 Emits a signal when items are moved so the parent can update the data model.
39 """
41 items_reordered = pyqtSignal(int, int) # from_index, to_index
43 def __init__(self, parent=None):
44 super().__init__(parent)
45 self.setDragDropMode(QListWidget.DragDropMode.InternalMove)
47 def dropEvent(self, event):
48 """Handle drop events and emit reorder signal."""
49 # Get the item being dropped and its original position
50 source_item = self.currentItem()
51 if not source_item:
52 super().dropEvent(event)
53 return
55 source_index = self.row(source_item)
57 # Let the default drop behavior happen first
58 super().dropEvent(event)
60 # Find the new position of the item
61 target_index = self.row(source_item)
63 # Only emit signal if position actually changed
64 if source_index != target_index:
65 self.items_reordered.emit(source_index, target_index)
68class PipelineEditorWidget(QWidget):
69 """
70 PyQt6 Pipeline Editor Widget.
72 Manages pipeline steps with add, edit, delete, load, save functionality.
73 Preserves all business logic from Textual version with clean PyQt6 UI.
74 """
76 # Signals
77 pipeline_changed = pyqtSignal(list) # List[FunctionStep]
78 step_selected = pyqtSignal(object) # FunctionStep
79 status_message = pyqtSignal(str) # status message
81 def __init__(self, file_manager: FileManager, service_adapter,
82 color_scheme: Optional[PyQt6ColorScheme] = None, parent=None):
83 """
84 Initialize the pipeline editor widget.
86 Args:
87 file_manager: FileManager instance for file operations
88 service_adapter: PyQt service adapter for dialogs and operations
89 color_scheme: Color scheme for styling (optional, uses service adapter if None)
90 parent: Parent widget
91 """
92 super().__init__(parent)
94 # Core dependencies
95 self.file_manager = file_manager
96 self.service_adapter = service_adapter
97 self.global_config = service_adapter.get_global_config()
99 # Initialize color scheme and style generator
100 self.color_scheme = color_scheme or service_adapter.get_current_color_scheme()
101 self.style_generator = StyleSheetGenerator(self.color_scheme)
103 # Business logic state (extracted from Textual version)
104 self.pipeline_steps: List[FunctionStep] = []
105 self.current_plate: str = ""
106 self.selected_step: str = ""
107 self.plate_pipelines: Dict[str, List[FunctionStep]] = {} # Per-plate pipeline storage
109 # UI components
110 self.step_list: Optional[QListWidget] = None
111 self.buttons: Dict[str, QPushButton] = {}
112 self.status_label: Optional[QLabel] = None
114 # Reference to plate manager (set externally)
115 self.plate_manager = None
117 # Setup UI
118 self.setup_ui()
119 self.setup_connections()
120 self.update_button_states()
122 logger.debug("Pipeline editor widget initialized")
124 # ========== UI Setup ==========
126 def setup_ui(self):
127 """Setup the user interface."""
128 layout = QVBoxLayout(self)
129 layout.setContentsMargins(5, 5, 5, 5)
130 layout.setSpacing(5)
132 # Title
133 title_label = QLabel("Pipeline Editor")
134 title_label.setFont(QFont("Arial", 12, QFont.Weight.Bold))
135 title_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.text_accent)}; padding: 5px;")
136 layout.addWidget(title_label)
138 # Main content splitter
139 splitter = QSplitter(Qt.Orientation.Vertical)
140 layout.addWidget(splitter)
142 # Pipeline steps list
143 self.step_list = ReorderableListWidget()
144 self.step_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
145 self.step_list.setStyleSheet(f"""
146 QListWidget {{
147 background-color: {self.color_scheme.to_hex(self.color_scheme.panel_bg)};
148 color: {self.color_scheme.to_hex(self.color_scheme.text_primary)};
149 border: 1px solid {self.color_scheme.to_hex(self.color_scheme.border_color)};
150 border-radius: 3px;
151 padding: 5px;
152 }}
153 QListWidget::item {{
154 padding: 8px;
155 border-bottom: 1px solid {self.color_scheme.to_hex(self.color_scheme.separator_color)};
156 border-radius: 3px;
157 margin: 2px;
158 }}
159 QListWidget::item:selected {{
160 background-color: {self.color_scheme.to_hex(self.color_scheme.selection_bg)};
161 }}
162 QListWidget::item:hover {{
163 background-color: {self.color_scheme.to_hex(self.color_scheme.hover_bg)};
164 }}
165 """)
166 splitter.addWidget(self.step_list)
168 # Button panel
169 button_panel = self.create_button_panel()
170 splitter.addWidget(button_panel)
172 # Status section
173 status_frame = self.create_status_section()
174 layout.addWidget(status_frame)
176 # Set splitter proportions
177 splitter.setSizes([400, 120])
179 def create_button_panel(self) -> QWidget:
180 """
181 Create the button panel with all pipeline actions.
183 Returns:
184 Widget containing action buttons
185 """
186 panel = QFrame()
187 panel.setFrameStyle(QFrame.Shape.Box)
188 panel.setStyleSheet(f"""
189 QFrame {{
190 background-color: {self.color_scheme.to_hex(self.color_scheme.frame_bg)};
191 border: 1px solid {self.color_scheme.to_hex(self.color_scheme.border_color)};
192 border-radius: 3px;
193 padding: 5px;
194 }}
195 """)
197 layout = QVBoxLayout(panel)
199 # Button configurations (extracted from Textual version)
200 button_configs = [
201 ("Add", "add_step", "Add new pipeline step"),
202 ("Del", "del_step", "Delete selected steps"),
203 ("Edit", "edit_step", "Edit selected step"),
204 ("Load", "load_pipeline", "Load pipeline from file"),
205 ("Save", "save_pipeline", "Save pipeline to file"),
206 ("Code", "code_pipeline", "Edit pipeline as Python code"),
207 ]
209 # Create buttons in rows
210 for i in range(0, len(button_configs), 3):
211 row_layout = QHBoxLayout()
213 for j in range(3):
214 if i + j < len(button_configs):
215 name, action, tooltip = button_configs[i + j]
217 button = QPushButton(name)
218 button.setToolTip(tooltip)
219 button.setMinimumHeight(30)
220 button.setStyleSheet(self.style_generator.generate_button_style())
222 # Connect button to action
223 button.clicked.connect(lambda checked, a=action: self.handle_button_action(a))
225 self.buttons[action] = button
226 row_layout.addWidget(button)
227 else:
228 row_layout.addStretch()
230 layout.addLayout(row_layout)
232 return panel
234 def create_status_section(self) -> QWidget:
235 """
236 Create the status section.
238 Returns:
239 Widget containing status information
240 """
241 frame = QFrame()
242 frame.setFrameStyle(QFrame.Shape.Box)
243 frame.setStyleSheet(f"""
244 QFrame {{
245 background-color: {self.color_scheme.to_hex(self.color_scheme.frame_bg)};
246 border: 1px solid {self.color_scheme.to_hex(self.color_scheme.border_color)};
247 border-radius: 3px;
248 padding: 5px;
249 }}
250 """)
252 layout = QVBoxLayout(frame)
254 # Status label
255 self.status_label = QLabel("Ready")
256 self.status_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.status_success)}; font-weight: bold;")
257 layout.addWidget(self.status_label)
259 return frame
261 def setup_connections(self):
262 """Setup signal/slot connections."""
263 # Step list selection
264 self.step_list.itemSelectionChanged.connect(self.on_selection_changed)
265 self.step_list.itemDoubleClicked.connect(self.on_item_double_clicked)
267 # Step list reordering
268 self.step_list.items_reordered.connect(self.on_steps_reordered)
270 # Internal signals
271 self.status_message.connect(self.update_status)
272 self.pipeline_changed.connect(self.on_pipeline_changed)
274 def handle_button_action(self, action: str):
275 """
276 Handle button actions (extracted from Textual version).
278 Args:
279 action: Action identifier
280 """
281 # Action mapping (preserved from Textual version)
282 action_map = {
283 "add_step": self.action_add_step,
284 "del_step": self.action_delete_step,
285 "edit_step": self.action_edit_step,
286 "load_pipeline": self.action_load_pipeline,
287 "save_pipeline": self.action_save_pipeline,
288 "code_pipeline": self.action_code_pipeline,
289 }
291 if action in action_map:
292 action_func = action_map[action]
294 # Handle async actions
295 if inspect.iscoroutinefunction(action_func):
296 # Run async action in thread
297 self.run_async_action(action_func)
298 else:
299 action_func()
301 def run_async_action(self, async_func: Callable):
302 """
303 Run async action using service adapter.
305 Args:
306 async_func: Async function to execute
307 """
308 self.service_adapter.execute_async_operation(async_func)
310 # ========== Business Logic Methods (Extracted from Textual) ==========
312 def format_item_for_display(self, step: FunctionStep) -> Tuple[str, str]:
313 """
314 Format step for display in the list (extracted from Textual version).
316 Args:
317 step: FunctionStep to format
319 Returns:
320 Tuple of (display_text, step_name)
321 """
322 step_name = getattr(step, 'name', 'Unknown Step')
323 display_text = f"📋 {step_name}"
324 return display_text, step_name
326 def action_add_step(self):
327 """Handle Add Step button (adapted from Textual version)."""
329 from openhcs.core.steps.function_step import FunctionStep
330 from openhcs.pyqt_gui.windows.dual_editor_window import DualEditorWindow
332 # Create new step
333 step_name = f"Step_{len(self.pipeline_steps) + 1}"
334 new_step = FunctionStep(
335 func=[], # Start with empty function list
336 name=step_name
337 )
339 def handle_save(edited_step):
340 """Handle step save from editor."""
341 self.pipeline_steps.append(edited_step)
342 self.update_step_list()
343 self.pipeline_changed.emit(self.pipeline_steps)
344 self.status_message.emit(f"Added new step: {edited_step.name}")
346 # Create and show editor dialog
347 editor = DualEditorWindow(
348 step_data=new_step,
349 is_new=True,
350 on_save_callback=handle_save,
351 parent=self
352 )
353 editor.show()
354 editor.raise_()
355 editor.activateWindow()
357 def action_delete_step(self):
358 """Handle Delete Step button (extracted from Textual version)."""
359 selected_items = self.get_selected_steps()
360 if not selected_items:
361 self.service_adapter.show_error_dialog("No steps selected to delete.")
362 return
364 # Remove selected steps
365 steps_to_remove = set(getattr(item, 'name', '') for item in selected_items)
366 new_steps = [step for step in self.pipeline_steps if getattr(step, 'name', '') not in steps_to_remove]
368 self.pipeline_steps = new_steps
369 self.update_step_list()
370 self.pipeline_changed.emit(self.pipeline_steps)
372 deleted_count = len(selected_items)
373 self.status_message.emit(f"Deleted {deleted_count} steps")
375 def action_edit_step(self):
376 """Handle Edit Step button (adapted from Textual version)."""
377 selected_items = self.get_selected_steps()
378 if not selected_items:
379 self.service_adapter.show_error_dialog("No step selected to edit.")
380 return
382 step_to_edit = selected_items[0]
384 # Open step editor dialog
385 from openhcs.pyqt_gui.windows.dual_editor_window import DualEditorWindow
387 def handle_save(edited_step):
388 """Handle step save from editor."""
389 # Find and replace the step in the pipeline
390 for i, step in enumerate(self.pipeline_steps):
391 if step is step_to_edit:
392 self.pipeline_steps[i] = edited_step
393 break
395 # Update the display
396 self.update_step_list()
397 self.pipeline_changed.emit(self.pipeline_steps)
398 self.status_message.emit(f"Updated step: {edited_step.name}")
400 # Create and show editor dialog
401 editor = DualEditorWindow(
402 step_data=step_to_edit,
403 is_new=False,
404 on_save_callback=handle_save,
405 parent=self
406 )
407 editor.show()
408 editor.raise_()
409 editor.activateWindow()
411 def action_load_pipeline(self):
412 """Handle Load Pipeline button (adapted from Textual version)."""
414 from openhcs.core.path_cache import PathCacheKey
416 # Use cached file dialog (mirrors Textual TUI pattern)
417 file_path = self.service_adapter.show_cached_file_dialog(
418 cache_key=PathCacheKey.PIPELINE_FILES,
419 title="Load Pipeline",
420 file_filter="Pipeline Files (*.pipeline);;All Files (*)",
421 mode="open"
422 )
424 if file_path:
425 self.load_pipeline_from_file(file_path)
427 def action_save_pipeline(self):
428 """Handle Save Pipeline button (adapted from Textual version)."""
429 if not self.pipeline_steps:
430 self.service_adapter.show_error_dialog("No pipeline steps to save.")
431 return
433 from openhcs.core.path_cache import PathCacheKey
435 # Use cached file dialog (mirrors Textual TUI pattern)
436 file_path = self.service_adapter.show_cached_file_dialog(
437 cache_key=PathCacheKey.PIPELINE_FILES,
438 title="Save Pipeline",
439 file_filter="Pipeline Files (*.pipeline);;All Files (*)",
440 mode="save"
441 )
443 if file_path:
444 self.save_pipeline_to_file(file_path)
446 def action_code_pipeline(self):
447 """Handle Code Pipeline button - edit pipeline as Python code."""
448 logger.debug("Code button pressed - opening code editor")
450 if not self.pipeline_steps:
451 self.service_adapter.show_error_dialog("No pipeline steps to edit")
452 return
454 if not self.current_plate:
455 self.service_adapter.show_error_dialog("No plate selected")
456 return
458 try:
459 # Use complete pipeline steps code generation
460 from openhcs.debug.pickle_to_python import generate_complete_pipeline_steps_code
462 # Generate complete pipeline steps code with imports
463 python_code = generate_complete_pipeline_steps_code(
464 pipeline_steps=list(self.pipeline_steps),
465 clean_mode=False
466 )
468 # Create simple code editor service
469 from openhcs.pyqt_gui.services.simple_code_editor import SimpleCodeEditorService
470 editor_service = SimpleCodeEditorService(self)
472 # Check if user wants external editor (check environment variable)
473 import os
474 use_external = os.environ.get('OPENHCS_USE_EXTERNAL_EDITOR', '').lower() in ('1', 'true', 'yes')
476 # Launch editor with callback
477 editor_service.edit_code(
478 initial_content=python_code,
479 title="Edit Pipeline Steps",
480 callback=self._handle_edited_pipeline_code,
481 use_external=use_external
482 )
484 except Exception as e:
485 logger.error(f"Failed to open pipeline code editor: {e}")
486 self.service_adapter.show_error_dialog(f"Failed to open code editor: {str(e)}")
488 def _handle_edited_pipeline_code(self, edited_code: str) -> None:
489 """Handle the edited pipeline code from code editor."""
490 logger.debug("Pipeline code edited, processing changes...")
491 try:
492 # Ensure we have a string
493 if not isinstance(edited_code, str):
494 logger.error(f"Expected string, got {type(edited_code)}: {edited_code}")
495 self.service_adapter.show_error_dialog("Invalid code format received from editor")
496 return
498 # Execute the code (it has all necessary imports)
499 namespace = {}
500 exec(edited_code, namespace)
502 # Get the pipeline_steps from the namespace
503 if 'pipeline_steps' in namespace:
504 new_pipeline_steps = namespace['pipeline_steps']
505 # Update the pipeline with new steps
506 self.pipeline_steps = new_pipeline_steps
507 self.update_step_list()
508 self.pipeline_changed.emit(self.pipeline_steps)
509 self.status_message.emit(f"Pipeline updated with {len(new_pipeline_steps)} steps")
510 else:
511 self.service_adapter.show_error_dialog("No 'pipeline_steps = [...]' assignment found in edited code")
513 except SyntaxError as e:
514 self.service_adapter.show_error_dialog(f"Invalid Python syntax: {e}")
515 except Exception as e:
516 logger.error(f"Failed to parse edited pipeline code: {e}")
517 self.service_adapter.show_error_dialog(f"Failed to parse pipeline code: {str(e)}")
519 def load_pipeline_from_file(self, file_path: Path):
520 """
521 Load pipeline from file (extracted from Textual version).
523 Args:
524 file_path: Path to pipeline file
525 """
526 try:
527 import dill as pickle
528 with open(file_path, 'rb') as f:
529 steps = pickle.load(f)
531 if isinstance(steps, list):
532 self.pipeline_steps = steps
533 self.update_step_list()
534 self.pipeline_changed.emit(self.pipeline_steps)
535 self.status_message.emit(f"Loaded {len(steps)} steps from {file_path.name}")
536 else:
537 self.status_message.emit(f"Invalid pipeline format in {file_path.name}")
539 except Exception as e:
540 logger.error(f"Failed to load pipeline: {e}")
541 self.service_adapter.show_error_dialog(f"Failed to load pipeline: {e}")
543 def save_pipeline_to_file(self, file_path: Path):
544 """
545 Save pipeline to file (extracted from Textual version).
547 Args:
548 file_path: Path to save pipeline
549 """
550 try:
551 import dill as pickle
552 with open(file_path, 'wb') as f:
553 pickle.dump(list(self.pipeline_steps), f)
554 self.status_message.emit(f"Saved pipeline to {file_path.name}")
556 except Exception as e:
557 logger.error(f"Failed to save pipeline: {e}")
558 self.service_adapter.show_error_dialog(f"Failed to save pipeline: {e}")
560 def save_pipeline_for_plate(self, plate_path: str, pipeline: List[FunctionStep]):
561 """
562 Save pipeline for specific plate (extracted from Textual version).
564 Args:
565 plate_path: Path of the plate
566 pipeline: Pipeline steps to save
567 """
568 self.plate_pipelines[plate_path] = pipeline
569 logger.debug(f"Saved pipeline for plate: {plate_path}")
571 def set_current_plate(self, plate_path: str):
572 """
573 Set current plate and load its pipeline (extracted from Textual version).
575 Args:
576 plate_path: Path of the current plate
577 """
578 self.current_plate = plate_path
580 # Load pipeline for the new plate
581 if plate_path:
582 plate_pipeline = self.plate_pipelines.get(plate_path, [])
583 self.pipeline_steps = plate_pipeline
584 else:
585 self.pipeline_steps = []
587 self.update_step_list()
588 self.update_button_states()
589 logger.debug(f"Current plate changed: {plate_path}")
591 # ========== UI Helper Methods ==========
593 def update_step_list(self):
594 """Update the step list widget using selection preservation mixin."""
595 def format_step_item(step):
596 """Format step item for display."""
597 display_text, step_name = self.format_item_for_display(step)
598 return display_text, step
600 def update_func():
601 """Update function that clears and rebuilds the list."""
602 self.step_list.clear()
604 for step in self.pipeline_steps:
605 display_text, step_data = format_step_item(step)
606 item = QListWidgetItem(display_text)
607 item.setData(Qt.ItemDataRole.UserRole, step_data)
608 item.setToolTip(f"Step: {getattr(step, 'name', 'Unknown')}")
609 self.step_list.addItem(item)
611 # Use utility to preserve selection during update
612 preserve_selection_during_update(
613 self.step_list,
614 lambda item_data: getattr(item_data, 'name', str(item_data)),
615 lambda: bool(self.pipeline_steps),
616 update_func
617 )
618 self.update_button_states()
620 def get_selected_steps(self) -> List[FunctionStep]:
621 """
622 Get currently selected steps.
624 Returns:
625 List of selected FunctionStep objects
626 """
627 selected_items = []
628 for item in self.step_list.selectedItems():
629 step_data = item.data(Qt.ItemDataRole.UserRole)
630 if step_data:
631 selected_items.append(step_data)
632 return selected_items
634 def update_button_states(self):
635 """Update button enabled/disabled states based on mathematical constraints (mirrors Textual TUI)."""
636 has_plate = bool(self.current_plate)
637 is_initialized = self._is_current_plate_initialized()
638 has_steps = len(self.pipeline_steps) > 0
639 has_selection = len(self.get_selected_steps()) > 0
641 # Mathematical constraints (mirrors Textual TUI logic):
642 # - Pipeline editing requires initialization
643 # - Step operations require steps to exist
644 # - Edit requires valid selection
645 self.buttons["add_step"].setEnabled(has_plate and is_initialized)
646 self.buttons["load_pipeline"].setEnabled(has_plate and is_initialized)
647 self.buttons["del_step"].setEnabled(has_steps)
648 self.buttons["edit_step"].setEnabled(has_steps and has_selection)
649 self.buttons["save_pipeline"].setEnabled(has_steps)
650 self.buttons["code_pipeline"].setEnabled(has_steps)
652 def update_status(self, message: str):
653 """
654 Update status label.
656 Args:
657 message: Status message to display
658 """
659 self.status_label.setText(message)
661 def on_selection_changed(self):
662 """Handle step list selection changes using utility."""
663 def on_selected(selected_steps):
664 self.selected_step = getattr(selected_steps[0], 'name', '')
665 self.step_selected.emit(selected_steps[0])
667 def on_cleared():
668 self.selected_step = ""
670 # Use utility to handle selection with prevention
671 handle_selection_change_with_prevention(
672 self.step_list,
673 self.get_selected_steps,
674 lambda item_data: getattr(item_data, 'name', str(item_data)),
675 lambda: bool(self.pipeline_steps),
676 lambda: self.selected_step,
677 on_selected,
678 on_cleared
679 )
681 self.update_button_states()
683 def on_item_double_clicked(self, item: QListWidgetItem):
684 """Handle double-click on step item."""
685 step_data = item.data(Qt.ItemDataRole.UserRole)
686 if step_data:
687 # Double-click triggers edit
688 self.action_edit_step()
690 def on_steps_reordered(self, from_index: int, to_index: int):
691 """
692 Handle step reordering from drag and drop.
694 Args:
695 from_index: Original position of the moved step
696 to_index: New position of the moved step
697 """
698 # Update the underlying pipeline_steps list to match the visual order
699 current_steps = list(self.pipeline_steps)
701 # Move the step in the data model
702 step = current_steps.pop(from_index)
703 current_steps.insert(to_index, step)
705 # Update pipeline steps
706 self.pipeline_steps = current_steps
708 # Emit pipeline changed signal to notify other components
709 self.pipeline_changed.emit(self.pipeline_steps)
711 # Update status message
712 step_name = getattr(step, 'name', 'Unknown Step')
713 direction = "up" if to_index < from_index else "down"
714 self.status_message.emit(f"Moved step '{step_name}' {direction}")
716 logger.debug(f"Reordered step '{step_name}' from index {from_index} to {to_index}")
718 def on_pipeline_changed(self, steps: List[FunctionStep]):
719 """
720 Handle pipeline changes.
722 Args:
723 steps: New pipeline steps
724 """
725 # Save pipeline to current plate if one is selected
726 if self.current_plate:
727 self.save_pipeline_for_plate(self.current_plate, steps)
729 logger.debug(f"Pipeline changed: {len(steps)} steps")
731 def _is_current_plate_initialized(self) -> bool:
732 """Check if current plate has an initialized orchestrator (mirrors Textual TUI)."""
733 if not self.current_plate:
734 return False
736 # Get plate manager from main window
737 main_window = self._find_main_window()
738 if not main_window:
739 return False
741 # Get plate manager widget from floating windows
742 plate_manager_window = main_window.floating_windows.get("plate_manager")
743 if not plate_manager_window:
744 return False
746 layout = plate_manager_window.layout()
747 if not layout or layout.count() == 0:
748 return False
750 plate_manager_widget = layout.itemAt(0).widget()
751 if not hasattr(plate_manager_widget, 'orchestrators'):
752 return False
754 orchestrator = plate_manager_widget.orchestrators.get(self.current_plate)
755 if orchestrator is None:
756 return False
758 # Check if orchestrator is in an initialized state (mirrors Textual TUI logic)
759 from openhcs.constants.constants import OrchestratorState
760 return orchestrator.state in [OrchestratorState.READY, OrchestratorState.COMPILED,
761 OrchestratorState.COMPLETED, OrchestratorState.COMPILE_FAILED,
762 OrchestratorState.EXEC_FAILED]
766 def _find_main_window(self):
767 """Find the main window by traversing parent hierarchy."""
768 widget = self
769 while widget:
770 if hasattr(widget, 'floating_windows'):
771 return widget
772 widget = widget.parent()
773 return None
775 def on_config_changed(self, new_config: GlobalPipelineConfig):
776 """
777 Handle global configuration changes.
779 Args:
780 new_config: New global configuration
781 """
782 self.global_config = new_config