Coverage for openhcs/pyqt_gui/widgets/plate_manager.py: 0.0%
928 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2Plate Manager Widget for PyQt6
4Manages plate selection, initialization, and execution with full feature parity
5to the Textual TUI version. Uses hybrid approach: extracted business logic + clean PyQt6 UI.
6"""
8import logging
9import asyncio
10import inspect
11import copy
12import sys
13import subprocess
14import tempfile
15from typing import List, Dict, Optional, Callable
16from pathlib import Path
18from PyQt6.QtWidgets import (
19 QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QListWidget,
20 QListWidgetItem, QLabel,
21 QSplitter, QApplication
22)
23from PyQt6.QtCore import Qt, pyqtSignal, pyqtSlot
24from PyQt6.QtGui import QFont
26from openhcs.core.config import GlobalPipelineConfig
27from openhcs.core.config import PipelineConfig
28from openhcs.io.filemanager import FileManager
29from openhcs.core.orchestrator.orchestrator import PipelineOrchestrator, OrchestratorState
30from openhcs.core.pipeline import Pipeline
31from openhcs.constants.constants import VariableComponents
32from openhcs.pyqt_gui.widgets.mixins import (
33 preserve_selection_during_update,
34 handle_selection_change_with_prevention
35)
36from openhcs.pyqt_gui.shared.style_generator import StyleSheetGenerator
37from openhcs.pyqt_gui.shared.color_scheme import PyQt6ColorScheme
39logger = logging.getLogger(__name__)
42class PlateManagerWidget(QWidget):
43 """
44 PyQt6 Plate Manager Widget.
46 Manages plate selection, initialization, compilation, and execution.
47 Preserves all business logic from Textual version with clean PyQt6 UI.
48 """
50 # Signals
51 plate_selected = pyqtSignal(str) # plate_path
52 status_message = pyqtSignal(str) # status message
53 orchestrator_state_changed = pyqtSignal(str, str) # plate_path, state
54 orchestrator_config_changed = pyqtSignal(str, object) # plate_path, effective_config
56 # Configuration change signals for tier 3 UI-code conversion
57 global_config_changed = pyqtSignal() # global config updated
58 pipeline_data_changed = pyqtSignal() # pipeline data updated
60 # Log viewer integration signals
61 subprocess_log_started = pyqtSignal(str) # base_log_path
62 subprocess_log_stopped = pyqtSignal()
63 clear_subprocess_logs = pyqtSignal()
65 # Progress update signals (thread-safe UI updates) - routed to status bar
66 progress_started = pyqtSignal(int) # max_value
67 progress_updated = pyqtSignal(int) # current_value
68 progress_finished = pyqtSignal()
70 # Error handling signals (thread-safe error reporting)
71 compilation_error = pyqtSignal(str, str) # plate_name, error_message
72 initialization_error = pyqtSignal(str, str) # plate_name, error_message
73 execution_error = pyqtSignal(str) # error_message
75 # Internal signals for thread-safe completion handling
76 _execution_complete_signal = pyqtSignal(dict, list) # result, ready_items
77 _execution_error_signal = pyqtSignal(str) # error_msg
79 def __init__(self, file_manager: FileManager, service_adapter,
80 color_scheme: Optional[PyQt6ColorScheme] = None, parent=None):
81 """
82 Initialize the plate manager widget.
84 Args:
85 file_manager: FileManager instance for file operations
86 service_adapter: PyQt service adapter for dialogs and operations
87 color_scheme: Color scheme for styling (optional, uses service adapter if None)
88 parent: Parent widget
89 """
90 super().__init__(parent)
92 # Core dependencies
93 self.file_manager = file_manager
94 self.service_adapter = service_adapter
95 self.global_config = service_adapter.get_global_config()
96 self.pipeline_editor = None # Will be set by main window
98 # Initialize color scheme and style generator
99 self.color_scheme = color_scheme or service_adapter.get_current_color_scheme()
100 self.style_generator = StyleSheetGenerator(self.color_scheme)
102 # Business logic state (extracted from Textual version)
103 self.plates: List[Dict] = [] # List of plate dictionaries
104 self.selected_plate_path: str = ""
105 self.orchestrators: Dict[str, PipelineOrchestrator] = {}
106 self.plate_configs: Dict[str, Dict] = {}
107 self.plate_compiled_data: Dict[str, tuple] = {} # Store compiled pipeline data
108 self.current_process = None
109 self.zmq_client = None # ZMQ execution client (when using ZMQ mode)
110 self.current_execution_id = None # Track current execution ID for cancellation
111 self.execution_state = "idle"
112 self.log_file_path: Optional[str] = None
113 self.log_file_position: int = 0
115 # UI components
116 self.plate_list: Optional[QListWidget] = None
117 self.buttons: Dict[str, QPushButton] = {}
118 self.status_label: Optional[QLabel] = None
120 # Setup UI
121 self.setup_ui()
122 self.setup_connections()
123 self.update_button_states()
125 # Connect internal signals for thread-safe completion handling
126 self._execution_complete_signal.connect(self._on_execution_complete)
127 self._execution_error_signal.connect(self._on_execution_error)
129 logger.debug("Plate manager widget initialized")
131 def cleanup(self):
132 """Cleanup resources before widget destruction."""
133 logger.info("🧹 Cleaning up PlateManagerWidget resources...")
135 # Disconnect and cleanup ZMQ client if it exists
136 if self.zmq_client is not None:
137 try:
138 logger.info("🧹 Disconnecting ZMQ client...")
139 self.zmq_client.disconnect()
140 except Exception as e:
141 logger.warning(f"Error disconnecting ZMQ client during cleanup: {e}")
142 finally:
143 self.zmq_client = None
145 # Terminate any running subprocess
146 if self.current_process is not None and self.current_process.poll() is None:
147 try:
148 logger.info("🧹 Terminating running subprocess...")
149 self.current_process.terminate()
150 self.current_process.wait(timeout=2)
151 except Exception as e:
152 logger.warning(f"Error terminating subprocess during cleanup: {e}")
153 try:
154 self.current_process.kill()
155 except:
156 pass
157 finally:
158 self.current_process = None
160 logger.info("✅ PlateManagerWidget cleanup completed")
162 # ========== UI Setup ==========
164 def setup_ui(self):
165 """Setup the user interface."""
166 layout = QVBoxLayout(self)
167 layout.setContentsMargins(2, 2, 2, 2)
168 layout.setSpacing(2)
170 # Header with title and status
171 header_widget = QWidget()
172 header_layout = QHBoxLayout(header_widget)
173 header_layout.setContentsMargins(5, 5, 5, 5)
175 title_label = QLabel("Plate Manager")
176 title_label.setFont(QFont("Arial", 12, QFont.Weight.Bold))
177 title_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.text_accent)};")
178 header_layout.addWidget(title_label)
180 header_layout.addStretch()
182 # Status label in header
183 self.status_label = QLabel("Ready")
184 self.status_label.setStyleSheet(f"color: {self.color_scheme.to_hex(self.color_scheme.status_success)}; font-weight: bold;")
185 header_layout.addWidget(self.status_label)
187 layout.addWidget(header_widget)
189 # Main content splitter
190 splitter = QSplitter(Qt.Orientation.Vertical)
191 layout.addWidget(splitter)
193 # Plate list
194 self.plate_list = QListWidget()
195 self.plate_list.setSelectionMode(QListWidget.SelectionMode.ExtendedSelection)
196 # Apply explicit styling to plate list for consistent background
197 self.plate_list.setStyleSheet(f"""
198 QListWidget {{
199 background-color: {self.color_scheme.to_hex(self.color_scheme.panel_bg)};
200 color: {self.color_scheme.to_hex(self.color_scheme.text_primary)};
201 border: none;
202 padding: 5px;
203 }}
204 QListWidget::item {{
205 padding: 8px;
206 border: none;
207 border-radius: 3px;
208 margin: 2px;
209 }}
210 QListWidget::item:selected {{
211 background-color: {self.color_scheme.to_hex(self.color_scheme.selection_bg)};
212 color: {self.color_scheme.to_hex(self.color_scheme.selection_text)};
213 }}
214 QListWidget::item:hover {{
215 background-color: {self.color_scheme.to_hex(self.color_scheme.hover_bg)};
216 }}
217 """)
218 # Apply centralized styling to main widget
219 self.setStyleSheet(self.style_generator.generate_plate_manager_style())
220 splitter.addWidget(self.plate_list)
222 # Button panel
223 button_panel = self.create_button_panel()
224 splitter.addWidget(button_panel)
226 # Set splitter proportions - make button panel much smaller
227 splitter.setSizes([400, 80])
229 def create_button_panel(self) -> QWidget:
230 """
231 Create the button panel with all plate management actions.
233 Returns:
234 Widget containing action buttons
235 """
236 panel = QWidget()
237 # Set consistent background
238 panel.setStyleSheet(f"""
239 QWidget {{
240 background-color: {self.color_scheme.to_hex(self.color_scheme.window_bg)};
241 border: none;
242 padding: 0px;
243 }}
244 """)
246 layout = QVBoxLayout(panel)
247 layout.setContentsMargins(0, 0, 0, 0)
248 layout.setSpacing(2)
250 # Button configurations (extracted from Textual version)
251 button_configs = [
252 ("Add", "add_plate", "Add new plate directory"),
253 ("Del", "del_plate", "Delete selected plates"),
254 ("Edit", "edit_config", "Edit plate configuration"),
255 ("Init", "init_plate", "Initialize selected plates"),
256 ("Compile", "compile_plate", "Compile plate pipelines"),
257 ("Run", "run_plate", "Run/Stop plate execution"),
258 ("Code", "code_plate", "Generate Python code"),
259 ("Viewer", "view_metadata", "View plate metadata"),
260 ]
262 # Create buttons in rows
263 for i in range(0, len(button_configs), 4):
264 row_layout = QHBoxLayout()
265 row_layout.setContentsMargins(2, 2, 2, 2)
266 row_layout.setSpacing(2)
268 for j in range(4):
269 if i + j < len(button_configs):
270 name, action, tooltip = button_configs[i + j]
272 button = QPushButton(name)
273 button.setToolTip(tooltip)
274 button.setMinimumHeight(30)
275 # Apply explicit button styling to ensure it works
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)
283 else:
284 row_layout.addStretch()
286 layout.addLayout(row_layout)
288 # Set maximum height to constrain the button panel (3 rows of buttons)
289 panel.setMaximumHeight(110)
291 return panel
295 def setup_connections(self):
296 """Setup signal/slot connections."""
297 # Plate list selection
298 self.plate_list.itemSelectionChanged.connect(self.on_selection_changed)
299 self.plate_list.itemDoubleClicked.connect(self.on_item_double_clicked)
301 # Internal signals
302 self.status_message.connect(self.update_status)
303 self.orchestrator_state_changed.connect(self.on_orchestrator_state_changed)
305 # Progress signals for thread-safe UI updates
306 self.progress_started.connect(self._on_progress_started)
307 self.progress_updated.connect(self._on_progress_updated)
308 self.progress_finished.connect(self._on_progress_finished)
310 # Error handling signals for thread-safe error reporting
311 self.compilation_error.connect(self._handle_compilation_error)
312 self.initialization_error.connect(self._handle_initialization_error)
313 self.execution_error.connect(self._handle_execution_error)
315 def handle_button_action(self, action: str):
316 """
317 Handle button actions (extracted from Textual version).
319 Args:
320 action: Action identifier
321 """
322 # Action mapping (preserved from Textual version)
323 action_map = {
324 "add_plate": self.action_add_plate,
325 "del_plate": self.action_delete_plate,
326 "edit_config": self.action_edit_config,
327 "init_plate": self.action_init_plate,
328 "compile_plate": self.action_compile_plate,
329 "code_plate": self.action_code_plate,
330 "view_metadata": self.action_view_metadata,
331 }
333 if action in action_map:
334 action_func = action_map[action]
336 # Handle async actions
337 if inspect.iscoroutinefunction(action_func):
338 self.run_async_action(action_func)
339 else:
340 action_func()
341 elif action == "run_plate":
342 if self.is_any_plate_running():
343 self.run_async_action(self.action_stop_execution)
344 else:
345 self.run_async_action(self.action_run_plate)
346 else:
347 logger.warning(f"Unknown action: {action}")
349 def run_async_action(self, async_func: Callable):
350 """
351 Run async action using service adapter.
353 Args:
354 async_func: Async function to execute
355 """
356 self.service_adapter.execute_async_operation(async_func)
358 def _update_orchestrator_global_config(self, orchestrator, new_global_config):
359 """Update orchestrator's global config reference and rebuild pipeline config if needed."""
360 from openhcs.config_framework.lazy_factory import rebuild_lazy_config_with_new_global_reference
361 from openhcs.core.config import GlobalPipelineConfig
363 # SIMPLIFIED: Update shared global context (dual-axis resolver handles context)
364 from openhcs.config_framework.lazy_factory import ensure_global_config_context
365 ensure_global_config_context(GlobalPipelineConfig, new_global_config)
367 # Rebuild orchestrator-specific config if it exists
368 if orchestrator.pipeline_config is not None:
369 orchestrator.pipeline_config = rebuild_lazy_config_with_new_global_reference(
370 orchestrator.pipeline_config,
371 new_global_config,
372 GlobalPipelineConfig
373 )
374 logger.info(f"Rebuilt orchestrator-specific config for plate: {orchestrator.plate_path}")
376 # Get effective config and emit signal for UI refresh
377 effective_config = orchestrator.get_effective_config()
378 self.orchestrator_config_changed.emit(str(orchestrator.plate_path), effective_config)
380 # ========== Business Logic Methods (Extracted from Textual) ==========
382 def action_add_plate(self):
383 """Handle Add Plate button (adapted from Textual version)."""
384 from openhcs.core.path_cache import PathCacheKey
386 # Use cached directory dialog (mirrors Textual TUI pattern)
387 directory_path = self.service_adapter.show_cached_directory_dialog(
388 cache_key=PathCacheKey.PLATE_IMPORT,
389 title="Select Plate Directory",
390 fallback_path=Path.home()
391 )
393 if directory_path:
394 self.add_plate_callback([directory_path])
396 def add_plate_callback(self, selected_paths: List[Path]):
397 """
398 Handle plate directory selection (extracted from Textual version).
400 Args:
401 selected_paths: List of selected directory paths
402 """
403 if not selected_paths:
404 self.status_message.emit("Plate selection cancelled")
405 return
407 added_plates = []
409 for selected_path in selected_paths:
410 # Check if plate already exists
411 if any(plate['path'] == str(selected_path) for plate in self.plates):
412 continue
414 # Add the plate to the list
415 plate_name = selected_path.name
416 plate_path = str(selected_path)
417 plate_entry = {
418 'name': plate_name,
419 'path': plate_path,
420 }
422 self.plates.append(plate_entry)
423 added_plates.append(plate_name)
425 if added_plates:
426 self.update_plate_list()
427 self.status_message.emit(f"Added {len(added_plates)} plate(s): {', '.join(added_plates)}")
428 else:
429 self.status_message.emit("No new plates added (duplicates skipped)")
431 def action_delete_plate(self):
432 """Handle Delete Plate button (extracted from Textual version)."""
433 selected_items = self.get_selected_plates()
434 if not selected_items:
435 self.service_adapter.show_error_dialog("No plate selected to delete.")
436 return
438 paths_to_delete = {p['path'] for p in selected_items}
439 self.plates = [p for p in self.plates if p['path'] not in paths_to_delete]
441 # Clean up orchestrators for deleted plates
442 for path in paths_to_delete:
443 if path in self.orchestrators:
444 del self.orchestrators[path]
446 if self.selected_plate_path in paths_to_delete:
447 self.selected_plate_path = ""
448 # Notify pipeline editor that no plate is selected (mirrors Textual TUI)
449 self.plate_selected.emit("")
451 self.update_plate_list()
452 self.status_message.emit(f"Deleted {len(paths_to_delete)} plate(s)")
454 def _validate_plates_for_operation(self, plates, operation_type):
455 """Unified functional validator for all plate operations."""
456 # Functional validation mapping
457 validators = {
458 'init': lambda p: True, # Init can work on any plates
459 'compile': lambda p: (
460 self.orchestrators.get(p['path']) and
461 self._get_current_pipeline_definition(p['path'])
462 ),
463 'run': lambda p: (
464 self.orchestrators.get(p['path']) and
465 self.orchestrators[p['path']].state in ['COMPILED', 'COMPLETED']
466 )
467 }
469 # Functional pattern: filter invalid plates in one pass
470 validator = validators.get(operation_type, lambda p: True)
471 return [p for p in plates if not validator(p)]
473 async def action_init_plate(self):
474 """Handle Initialize Plate button with unified validation."""
475 # CRITICAL: Set up global context in worker thread
476 # The service adapter runs this entire function in a worker thread,
477 # so we need to establish the global context here
478 from openhcs.config_framework.lazy_factory import ensure_global_config_context
479 from openhcs.core.config import GlobalPipelineConfig
480 ensure_global_config_context(GlobalPipelineConfig, self.global_config)
482 selected_items = self.get_selected_plates()
484 # Unified validation - let it fail if no plates
485 invalid_plates = self._validate_plates_for_operation(selected_items, 'init')
487 self.progress_started.emit(len(selected_items))
489 # Functional pattern: async map with enumerate
490 async def init_single_plate(i, plate):
491 plate_path = plate['path']
492 # Create orchestrator in main thread (has access to global context)
493 orchestrator = PipelineOrchestrator(
494 plate_path=plate_path,
495 storage_registry=self.file_manager.registry
496 )
497 # Only run heavy initialization in worker thread
498 # Need to set up context in worker thread too since initialize() runs there
499 def initialize_with_context():
500 from openhcs.config_framework.lazy_factory import ensure_global_config_context
501 from openhcs.core.config import GlobalPipelineConfig
502 ensure_global_config_context(GlobalPipelineConfig, self.global_config)
503 return orchestrator.initialize()
505 try:
506 await asyncio.get_event_loop().run_in_executor(
507 None,
508 initialize_with_context
509 )
511 self.orchestrators[plate_path] = orchestrator
512 self.orchestrator_state_changed.emit(plate_path, "READY")
514 if not self.selected_plate_path:
515 self.selected_plate_path = plate_path
516 self.plate_selected.emit(plate_path)
518 except Exception as e:
519 logger.error(f"Failed to initialize plate {plate_path}: {e}", exc_info=True)
520 # Create a failed orchestrator to track the error state
521 failed_orchestrator = PipelineOrchestrator(
522 plate_path=plate_path,
523 storage_registry=self.file_manager.registry
524 )
525 failed_orchestrator._state = OrchestratorState.INIT_FAILED
526 self.orchestrators[plate_path] = failed_orchestrator
527 # Emit signal to update UI with failed state
528 self.orchestrator_state_changed.emit(plate_path, OrchestratorState.INIT_FAILED.value)
529 # Show error dialog
530 self.initialization_error.emit(plate['name'], str(e))
532 self.progress_updated.emit(i + 1)
534 # Process all plates functionally
535 await asyncio.gather(*[
536 init_single_plate(i, plate)
537 for i, plate in enumerate(selected_items)
538 ])
540 self.progress_finished.emit()
542 # Count successes and failures
543 success_count = len([p for p in selected_items if self.orchestrators.get(p['path']) and self.orchestrators[p['path']].state == OrchestratorState.READY])
544 error_count = len([p for p in selected_items if self.orchestrators.get(p['path']) and self.orchestrators[p['path']].state == OrchestratorState.INIT_FAILED])
546 if error_count == 0:
547 self.status_message.emit(f"Successfully initialized {success_count} plate(s)")
548 else:
549 self.status_message.emit(f"Initialized {success_count} plate(s), {error_count} error(s)")
551 # Additional action methods would be implemented here following the same pattern...
552 # (compile_plate, run_plate, code_plate, view_metadata, edit_config)
554 def action_edit_config(self):
555 """
556 Handle Edit Config button - create per-orchestrator PipelineConfig instances.
558 This enables per-orchestrator configuration without affecting global configuration.
559 Shows resolved defaults from GlobalPipelineConfig with "Pipeline default: {value}" placeholders.
560 """
561 selected_items = self.get_selected_plates()
563 if not selected_items:
564 self.service_adapter.show_error_dialog("No plates selected for configuration.")
565 return
567 # Get selected orchestrators
568 selected_orchestrators = [
569 self.orchestrators[item['path']] for item in selected_items
570 if item['path'] in self.orchestrators
571 ]
573 if not selected_orchestrators:
574 self.service_adapter.show_error_dialog("No initialized orchestrators selected.")
575 return
577 # Load existing config or create new one for editing
578 representative_orchestrator = selected_orchestrators[0]
580 # CRITICAL FIX: Don't change thread-local context - preserve orchestrator context
581 # The config window should work with the current orchestrator context
582 # Reset behavior will be handled differently to avoid corrupting step editor context
584 # CRITICAL FIX: Create PipelineConfig that preserves user-set values but shows placeholders for inherited fields
585 # The orchestrator's pipeline_config has concrete values filled in from global config inheritance,
586 # but we need to distinguish between user-set values (keep concrete) and inherited values (show as placeholders)
587 from openhcs.config_framework.lazy_factory import create_dataclass_for_editing
588 from dataclasses import fields
590 # CRITICAL FIX: Create config for editing that preserves user values while showing placeholders for inherited fields
591 if representative_orchestrator.pipeline_config is not None:
592 # Orchestrator has existing config - preserve explicitly set fields, reset others to None for placeholders
593 existing_config = representative_orchestrator.pipeline_config
594 explicitly_set_fields = getattr(existing_config, '_explicitly_set_fields', set())
596 # Create field values: keep explicitly set values, use None for inherited fields
597 field_values = {}
598 for field in fields(PipelineConfig):
599 if field.name in explicitly_set_fields:
600 # User explicitly set this field - preserve the concrete value
601 field_values[field.name] = object.__getattribute__(existing_config, field.name)
602 else:
603 # Field was inherited from global config - use None to show placeholder
604 field_values[field.name] = None
606 # Create config with preserved user values and None for inherited fields
607 current_plate_config = PipelineConfig(**field_values)
608 # Preserve the explicitly set fields tracking (bypass frozen restriction)
609 object.__setattr__(current_plate_config, '_explicitly_set_fields', explicitly_set_fields.copy())
610 else:
611 # No existing config - create fresh config with all None values (all show as placeholders)
612 current_plate_config = create_dataclass_for_editing(PipelineConfig, self.global_config)
614 def handle_config_save(new_config: PipelineConfig) -> None:
615 """Apply per-orchestrator configuration without global side effects."""
616 # SIMPLIFIED: Debug logging without thread-local context
617 from dataclasses import fields
618 logger.debug(f"🔍 CONFIG SAVE - new_config type: {type(new_config)}")
619 for field in fields(new_config):
620 raw_value = object.__getattribute__(new_config, field.name)
621 logger.debug(f"🔍 CONFIG SAVE - new_config.{field.name} = {raw_value}")
623 for orchestrator in selected_orchestrators:
624 # Direct synchronous call - no async needed
625 orchestrator.apply_pipeline_config(new_config)
626 # Emit signal for UI components to refresh
627 effective_config = orchestrator.get_effective_config()
628 self.orchestrator_config_changed.emit(str(orchestrator.plate_path), effective_config)
630 # Auto-sync handles context restoration automatically when pipeline_config is accessed
631 if self.selected_plate_path and self.selected_plate_path in self.orchestrators:
632 logger.debug(f"Orchestrator context automatically maintained after config save: {self.selected_plate_path}")
634 count = len(selected_orchestrators)
635 # Success message dialog removed for test automation compatibility
637 # Open configuration window using PipelineConfig (not GlobalPipelineConfig)
638 # PipelineConfig already imported from openhcs.core.config
639 self._open_config_window(
640 config_class=PipelineConfig,
641 current_config=current_plate_config,
642 on_save_callback=handle_config_save,
643 orchestrator=representative_orchestrator # Pass orchestrator for context persistence
644 )
646 def _open_config_window(self, config_class, current_config, on_save_callback, orchestrator=None):
647 """
648 Open configuration window with specified config class and current config.
650 Args:
651 config_class: Configuration class type (PipelineConfig or GlobalPipelineConfig)
652 current_config: Current configuration instance
653 on_save_callback: Function to call when config is saved
654 orchestrator: Optional orchestrator reference for context persistence
655 """
656 from openhcs.pyqt_gui.windows.config_window import ConfigWindow
657 from openhcs.config_framework.context_manager import config_context
660 # SIMPLIFIED: ConfigWindow now uses the dataclass instance directly for context
661 # No need for external context management - the form manager handles it automatically
662 # CRITICAL: Pass orchestrator's plate_path as scope_id to limit cross-window updates to same orchestrator
663 scope_id = str(orchestrator.plate_path) if orchestrator else None
664 with config_context(orchestrator.pipeline_config):
665 config_window = ConfigWindow(
666 config_class, # config_class
667 current_config, # current_config
668 on_save_callback, # on_save_callback
669 self.color_scheme, # color_scheme
670 self, # parent
671 scope_id=scope_id # Scope to this orchestrator
672 )
674 # REMOVED: refresh_config signal connection - now obsolete with live placeholder context system
675 # Config windows automatically update their placeholders through cross-window signals
676 # when other windows save changes. No need to rebuild the entire form.
678 # Show as non-modal window (like main window configuration)
679 config_window.show()
680 config_window.raise_()
681 config_window.activateWindow()
683 def action_edit_global_config(self):
684 """
685 Handle global configuration editing - affects all orchestrators.
687 Uses concrete GlobalPipelineConfig for direct editing with static placeholder defaults.
688 """
689 from openhcs.core.config import GlobalPipelineConfig
691 # Get current global config from service adapter or use default
692 current_global_config = self.service_adapter.get_global_config() or GlobalPipelineConfig()
694 def handle_global_config_save(new_config: GlobalPipelineConfig) -> None:
695 """Apply global configuration to all orchestrators and save to cache."""
696 self.service_adapter.set_global_config(new_config) # Update app-level config
698 # Update thread-local storage for MaterializationPathConfig defaults
699 from openhcs.core.config import GlobalPipelineConfig
700 from openhcs.config_framework.global_config import set_global_config_for_editing
701 set_global_config_for_editing(GlobalPipelineConfig, new_config)
703 # Save to cache for persistence between sessions
704 self._save_global_config_to_cache(new_config)
706 for orchestrator in self.orchestrators.values():
707 self._update_orchestrator_global_config(orchestrator, new_config)
709 # SIMPLIFIED: Dual-axis resolver handles context discovery automatically
710 if self.selected_plate_path and self.selected_plate_path in self.orchestrators:
711 logger.debug(f"Global config applied to selected orchestrator: {self.selected_plate_path}")
713 self.service_adapter.show_info_dialog("Global configuration applied to all orchestrators")
715 # Open configuration window using concrete GlobalPipelineConfig
716 self._open_config_window(
717 config_class=GlobalPipelineConfig,
718 current_config=current_global_config,
719 on_save_callback=handle_global_config_save
720 )
722 def _save_global_config_to_cache(self, config: GlobalPipelineConfig):
723 """Save global config to cache for persistence between sessions."""
724 try:
725 # Use synchronous saving to ensure it completes
726 from openhcs.core.config_cache import _sync_save_config
727 from openhcs.core.xdg_paths import get_config_file_path
729 cache_file = get_config_file_path("global_config.config")
730 success = _sync_save_config(config, cache_file)
732 if success:
733 logger.info("Global config saved to cache for session persistence")
734 else:
735 logger.error("Failed to save global config to cache - sync save returned False")
736 except Exception as e:
737 logger.error(f"Failed to save global config to cache: {e}")
738 # Don't show error dialog as this is not critical for immediate functionality
740 async def action_compile_plate(self):
741 """Handle Compile Plate button - compile pipelines for selected plates."""
742 selected_items = self.get_selected_plates()
744 if not selected_items:
745 logger.warning("No plates available for compilation")
746 return
748 # Unified validation using functional validator
749 invalid_plates = self._validate_plates_for_operation(selected_items, 'compile')
751 # Let validation failures bubble up as status messages
752 if invalid_plates:
753 invalid_names = [p['name'] for p in invalid_plates]
754 self.status_message.emit(f"Cannot compile invalid plates: {', '.join(invalid_names)}")
755 return
757 # Start async compilation
758 await self._compile_plates_worker(selected_items)
760 async def _compile_plates_worker(self, selected_items: List[Dict]) -> None:
761 """Background worker for plate compilation."""
762 # CRITICAL: Set up global context in worker thread
763 # The service adapter runs this entire function in a worker thread,
764 # so we need to establish the global context here
765 from openhcs.config_framework.lazy_factory import ensure_global_config_context
766 from openhcs.core.config import GlobalPipelineConfig
767 ensure_global_config_context(GlobalPipelineConfig, self.global_config)
769 # Use signals for thread-safe UI updates
770 self.progress_started.emit(len(selected_items))
772 for i, plate_data in enumerate(selected_items):
773 plate_path = plate_data['path']
775 # Get definition pipeline - this is the ORIGINAL pipeline from the editor
776 # It should have func attributes intact
777 definition_pipeline = self._get_current_pipeline_definition(plate_path)
778 if not definition_pipeline:
779 logger.warning(f"No pipeline defined for {plate_data['name']}, using empty pipeline")
780 definition_pipeline = []
782 # Validate that steps have func attribute (required for ZMQ execution)
783 for i, step in enumerate(definition_pipeline):
784 if not hasattr(step, 'func'):
785 logger.error(f"Step {i} ({step.name}) missing 'func' attribute! Cannot execute via ZMQ.")
786 raise AttributeError(f"Step '{step.name}' is missing 'func' attribute. "
787 "This usually means the pipeline was loaded from a compiled state instead of the original definition.")
789 try:
790 # Get or create orchestrator for compilation
791 if plate_path in self.orchestrators:
792 orchestrator = self.orchestrators[plate_path]
793 if not orchestrator.is_initialized():
794 # Only run heavy initialization in worker thread
795 # Need to set up context in worker thread too since initialize() runs there
796 def initialize_with_context():
797 from openhcs.config_framework.lazy_factory import ensure_global_config_context
798 from openhcs.core.config import GlobalPipelineConfig
799 ensure_global_config_context(GlobalPipelineConfig, self.global_config)
800 return orchestrator.initialize()
802 import asyncio
803 loop = asyncio.get_event_loop()
804 await loop.run_in_executor(None, initialize_with_context)
805 else:
806 # Create orchestrator in main thread (has access to global context)
807 orchestrator = PipelineOrchestrator(
808 plate_path=plate_path,
809 storage_registry=self.file_manager.registry
810 )
811 # Only run heavy initialization in worker thread
812 # Need to set up context in worker thread too since initialize() runs there
813 def initialize_with_context():
814 from openhcs.config_framework.lazy_factory import ensure_global_config_context
815 from openhcs.core.config import GlobalPipelineConfig
816 ensure_global_config_context(GlobalPipelineConfig, self.global_config)
817 return orchestrator.initialize()
819 import asyncio
820 loop = asyncio.get_event_loop()
821 await loop.run_in_executor(None, initialize_with_context)
822 self.orchestrators[plate_path] = orchestrator
823 self.orchestrators[plate_path] = orchestrator
825 # Make fresh copy for compilation
826 execution_pipeline = copy.deepcopy(definition_pipeline)
828 # Fix step IDs after deep copy to match new object IDs
829 for step in execution_pipeline:
830 step.step_id = str(id(step))
831 # Ensure variable_components is never None - use FunctionStep default
832 if step.variable_components is None:
833 logger.warning(f"Step '{step.name}' has None variable_components, setting FunctionStep default")
834 step.variable_components = [VariableComponents.SITE]
835 # Also ensure it's not an empty list
836 elif not step.variable_components:
837 logger.warning(f"Step '{step.name}' has empty variable_components, setting FunctionStep default")
838 step.variable_components = [VariableComponents.SITE]
840 # Get wells and compile (async - run in executor to avoid blocking UI)
841 # Wrap in Pipeline object like test_main.py does
842 pipeline_obj = Pipeline(steps=execution_pipeline)
844 # Run heavy operations in executor to avoid blocking UI (works in Qt thread)
845 import asyncio
846 loop = asyncio.get_event_loop()
847 # Get wells using multiprocessing axis (WELL in default config)
848 from openhcs.constants import MULTIPROCESSING_AXIS
849 wells = await loop.run_in_executor(None, lambda: orchestrator.get_component_keys(MULTIPROCESSING_AXIS))
851 # Wrap compilation with context setup for worker thread
852 def compile_with_context():
853 from openhcs.config_framework.lazy_factory import ensure_global_config_context
854 from openhcs.core.config import GlobalPipelineConfig
855 ensure_global_config_context(GlobalPipelineConfig, self.global_config)
856 return orchestrator.compile_pipelines(pipeline_obj.steps, wells)
858 compilation_result = await loop.run_in_executor(None, compile_with_context)
860 # Extract compiled_contexts from the dict returned by compile_pipelines
861 # compile_pipelines now returns {'pipeline_definition': ..., 'compiled_contexts': ...}
862 compiled_contexts = compilation_result['compiled_contexts']
864 # Store compiled data AND original definition pipeline
865 # ZMQ mode needs the original definition, direct mode needs the compiled execution pipeline
866 self.plate_compiled_data[plate_path] = {
867 'definition_pipeline': definition_pipeline, # Original uncompiled pipeline for ZMQ
868 'execution_pipeline': execution_pipeline, # Compiled pipeline for direct mode
869 'compiled_contexts': compiled_contexts
870 }
871 logger.info(f"Successfully compiled {plate_path}")
873 # Update orchestrator state change signal
874 self.orchestrator_state_changed.emit(plate_path, "COMPILED")
876 except Exception as e:
877 logger.error(f"COMPILATION ERROR: Pipeline compilation failed for {plate_path}: {e}", exc_info=True)
878 plate_data['error'] = str(e)
879 # Don't store anything in plate_compiled_data on failure
880 self.orchestrator_state_changed.emit(plate_path, "COMPILE_FAILED")
881 # Use signal for thread-safe error reporting instead of direct dialog call
882 self.compilation_error.emit(plate_data['name'], str(e))
884 # Use signal for thread-safe progress update
885 self.progress_updated.emit(i + 1)
887 # Use signal for thread-safe progress completion
888 self.progress_finished.emit()
889 self.status_message.emit(f"Compilation completed for {len(selected_items)} plate(s)")
890 self.update_button_states()
892 async def action_run_plate(self):
893 """Handle Run Plate button - execute compiled plates using ZMQ."""
894 selected_items = self.get_selected_plates()
895 if not selected_items:
896 # Use signal for thread-safe error reporting from async context
897 self.execution_error.emit("No plates selected to run.")
898 return
900 ready_items = [item for item in selected_items if item.get('path') in self.plate_compiled_data]
901 if not ready_items:
902 # Use signal for thread-safe error reporting from async context
903 self.execution_error.emit("Selected plates are not compiled. Please compile first.")
904 return
906 await self._run_plates_zmq(ready_items)
908 async def _run_plates_zmq(self, ready_items):
909 """Run plates using ZMQ execution client (recommended)."""
910 try:
911 from openhcs.runtime.zmq_execution_client import ZMQExecutionClient
912 import asyncio
914 plate_paths_to_run = [item['path'] for item in ready_items]
915 logger.info(f"Starting ZMQ execution for {len(plate_paths_to_run)} plates")
917 # Clear subprocess logs before starting new execution
918 self.clear_subprocess_logs.emit()
920 # Get event loop (needed for all async operations)
921 loop = asyncio.get_event_loop()
923 # Always create a fresh client for each execution to avoid state conflicts
924 # Clean up old client if it exists
925 if self.zmq_client is not None:
926 logger.info("🧹 Disconnecting previous ZMQ client")
927 try:
928 def _disconnect_old():
929 self.zmq_client.disconnect()
930 await loop.run_in_executor(None, _disconnect_old)
931 except Exception as e:
932 logger.warning(f"Error disconnecting old client: {e}")
933 finally:
934 self.zmq_client = None
936 # Create new ZMQ client (persistent mode - server stays alive)
937 logger.info("🔌 Creating new ZMQ client")
938 self.zmq_client = ZMQExecutionClient(
939 port=7777,
940 persistent=True, # Server persists across executions
941 progress_callback=self._on_zmq_progress
942 )
944 # Connect to server (will spawn if needed)
945 def _connect():
946 return self.zmq_client.connect(timeout=15)
948 connected = await loop.run_in_executor(None, _connect)
950 if not connected:
951 raise RuntimeError("Failed to connect to ZMQ execution server")
953 logger.info("✅ Connected to ZMQ execution server")
955 # Update orchestrator states to show running state
956 for plate in ready_items:
957 plate_path = plate['path']
958 if plate_path in self.orchestrators:
959 self.orchestrators[plate_path]._state = OrchestratorState.EXECUTING
960 self.orchestrator_state_changed.emit(plate_path, OrchestratorState.EXECUTING.value)
962 self.execution_state = "running"
963 self.status_message.emit(f"Running {len(ready_items)} plate(s) via ZMQ...")
964 self.update_button_states()
966 # Execute each plate
967 for plate_path in plate_paths_to_run:
968 compiled_data = self.plate_compiled_data[plate_path]
970 # Use DEFINITION pipeline for ZMQ (server will compile)
971 # NOT the execution_pipeline (which is already compiled)
972 definition_pipeline = compiled_data['definition_pipeline']
974 # Get config for this plate
975 # CRITICAL: Send GlobalPipelineConfig (concrete) and PipelineConfig (lazy overrides) separately
976 # The server will merge them via the dual-axis resolver
977 if plate_path in self.orchestrators:
978 # Send the global config (concrete values) + pipeline config (lazy overrides)
979 global_config_to_send = self.global_config
980 pipeline_config = self.orchestrators[plate_path].pipeline_config
981 else:
982 # No orchestrator - send global config with empty pipeline config
983 global_config_to_send = self.global_config
984 from openhcs.core.config import PipelineConfig
985 pipeline_config = PipelineConfig()
987 logger.info(f"Executing plate: {plate_path}")
989 # Submit pipeline via ZMQ (non-blocking - returns immediately)
990 # Send original definition pipeline - server will compile it
991 def _submit():
992 return self.zmq_client.submit_pipeline(
993 plate_id=str(plate_path),
994 pipeline_steps=definition_pipeline,
995 global_config=global_config_to_send,
996 pipeline_config=pipeline_config
997 )
999 response = await loop.run_in_executor(None, _submit)
1001 # Track execution ID for cancellation
1002 if response.get('execution_id'):
1003 self.current_execution_id = response['execution_id']
1005 logger.info(f"Plate {plate_path} submission response: {response.get('status')}")
1007 # Handle submission response (not completion - that comes via progress callback)
1008 status = response.get('status')
1009 if status == 'accepted':
1010 # Execution submitted successfully - it's now running in background
1011 logger.info(f"Plate {plate_path} execution submitted successfully, ID={response.get('execution_id')}")
1012 self.status_message.emit(f"Executing {plate_path}... (check progress below)")
1014 # Start polling for completion in background (non-blocking)
1015 execution_id = response.get('execution_id')
1016 if execution_id:
1017 self._start_completion_poller(execution_id, plate_paths_to_run, ready_items)
1018 else:
1019 # Submission failed - handle error
1020 error_msg = response.get('message', 'Unknown error')
1021 logger.error(f"Plate {plate_path} submission failed: {error_msg}")
1022 self.execution_error.emit(f"Submission failed for {plate_path}: {error_msg}")
1024 # Reset state on submission failure
1025 self.execution_state = "idle"
1026 self.current_execution_id = None
1027 for plate in ready_items:
1028 plate_path = plate['path']
1029 if plate_path in self.orchestrators:
1030 self.orchestrators[plate_path]._state = OrchestratorState.READY
1031 self.orchestrator_state_changed.emit(plate_path, OrchestratorState.READY.value)
1032 self.update_button_states()
1034 except Exception as e:
1035 logger.error(f"Failed to execute plates via ZMQ: {e}", exc_info=True)
1036 # Use signal for thread-safe error reporting
1037 self.execution_error.emit(f"Failed to execute: {e}")
1038 self.execution_state = "idle"
1040 # Disconnect client on error
1041 if self.zmq_client is not None:
1042 try:
1043 def _disconnect():
1044 self.zmq_client.disconnect()
1045 await loop.run_in_executor(None, _disconnect)
1046 except Exception as disconnect_error:
1047 logger.warning(f"Failed to disconnect ZMQ client: {disconnect_error}")
1048 finally:
1049 self.zmq_client = None
1051 self.current_execution_id = None
1052 self.update_button_states()
1054 def _start_completion_poller(self, execution_id, plate_paths, ready_items):
1055 """
1056 Start background thread to poll for execution completion (non-blocking).
1058 Args:
1059 execution_id: Execution ID to poll
1060 plate_paths: List of plate paths being executed
1061 ready_items: List of plate items being executed
1062 """
1063 import threading
1065 def poll_completion():
1066 """Poll for completion in background thread."""
1067 try:
1068 # Wait for completion (blocking in this thread, but not UI thread)
1069 result = self.zmq_client.wait_for_completion(execution_id)
1071 # Emit completion signal (thread-safe via Qt signal)
1072 self._execution_complete_signal.emit(result, ready_items)
1074 except Exception as e:
1075 logger.error(f"Error polling for completion: {e}", exc_info=True)
1076 # Emit error signal (thread-safe via Qt signal)
1077 self._execution_error_signal.emit(str(e))
1079 # Start polling thread
1080 thread = threading.Thread(target=poll_completion, daemon=True)
1081 thread.start()
1083 def _on_execution_complete(self, result, ready_items):
1084 """Handle execution completion (called from main thread via signal)."""
1085 try:
1086 status = result.get('status')
1087 logger.info(f"Execution completed with status: {status}")
1089 if status == 'complete':
1090 self.status_message.emit(f"Completed {len(ready_items)} plate(s)")
1091 elif status == 'cancelled':
1092 self.status_message.emit(f"Execution cancelled")
1093 else:
1094 error_msg = result.get('message', 'Unknown error')
1095 self.execution_error.emit(f"Execution failed: {error_msg}")
1097 # Disconnect ZMQ client on completion
1098 if self.zmq_client is not None:
1099 try:
1100 logger.info("Disconnecting ZMQ client after execution completion")
1101 self.zmq_client.disconnect()
1102 except Exception as disconnect_error:
1103 logger.warning(f"Failed to disconnect ZMQ client: {disconnect_error}")
1104 finally:
1105 self.zmq_client = None
1107 # Update state
1108 self.execution_state = "idle"
1109 self.current_execution_id = None
1111 # Update orchestrator states
1112 # Note: orchestrator_state_changed signal triggers on_orchestrator_state_changed()
1113 # which calls update_plate_list(), so we don't need to call update_button_states() here
1114 # (calling it here causes recursive repaint and crashes)
1115 for plate in ready_items:
1116 plate_path = plate['path']
1117 if plate_path in self.orchestrators:
1118 if status == 'complete':
1119 self.orchestrators[plate_path]._state = OrchestratorState.COMPLETED
1120 self.orchestrator_state_changed.emit(plate_path, OrchestratorState.COMPLETED.value)
1121 else:
1122 self.orchestrators[plate_path]._state = OrchestratorState.READY
1123 self.orchestrator_state_changed.emit(plate_path, OrchestratorState.READY.value)
1125 except Exception as e:
1126 logger.error(f"Error handling execution completion: {e}", exc_info=True)
1128 def _on_execution_error(self, error_msg):
1129 """Handle execution error (called from main thread via signal)."""
1130 self.execution_error.emit(f"Execution error: {error_msg}")
1131 self.execution_state = "idle"
1132 self.current_execution_id = None
1133 self.update_button_states()
1135 def _on_zmq_progress(self, message):
1136 """
1137 Handle progress updates from ZMQ execution server.
1139 This is called from the progress listener thread (background thread),
1140 so we must use QMetaObject.invokeMethod to safely emit signals from the main thread.
1141 """
1142 try:
1143 well_id = message.get('well_id', 'unknown')
1144 step = message.get('step', 'unknown')
1145 status = message.get('status', 'unknown')
1147 # Emit progress message to UI (thread-safe)
1148 progress_text = f"[{well_id}] {step}: {status}"
1150 # Use QMetaObject.invokeMethod to emit signal from main thread
1151 from PyQt6.QtCore import QMetaObject, Qt
1152 QMetaObject.invokeMethod(
1153 self,
1154 "_emit_status_message",
1155 Qt.ConnectionType.QueuedConnection,
1156 progress_text
1157 )
1159 logger.debug(f"Progress: {progress_text}")
1161 except Exception as e:
1162 logger.warning(f"Failed to handle progress update: {e}")
1164 @pyqtSlot(str)
1165 def _emit_status_message(self, message: str):
1166 """Emit status message from main thread (called via QMetaObject.invokeMethod)."""
1167 self.status_message.emit(message)
1169 async def action_stop_execution(self):
1170 """Handle Stop Execution - cancel ZMQ execution or terminate subprocess.
1172 First click: Graceful shutdown, button changes to "Force Kill"
1173 Second click: Force shutdown
1175 Uses EXACT same code path as ZMQ browser quit button.
1176 """
1177 logger.info("🛑🛑🛑 action_stop_execution CALLED")
1178 logger.info(f"🛑 execution_state: {self.execution_state}")
1179 logger.info(f"🛑 zmq_client: {self.zmq_client}")
1180 logger.info(f"🛑 Button text: {self.buttons['run_plate'].text()}")
1182 # Check if this is a force kill (button text is "Force Kill")
1183 is_force_kill = self.buttons["run_plate"].text() == "Force Kill"
1185 # Check if using ZMQ execution
1186 if self.zmq_client:
1187 port = self.zmq_client.port
1189 # Change button to "Force Kill" IMMEDIATELY (before any async operations)
1190 if not is_force_kill:
1191 logger.info(f"🛑 Stop button pressed - changing to Force Kill")
1192 self.execution_state = "force_kill_ready"
1193 self.update_button_states()
1194 # Force immediate UI update
1195 QApplication.processEvents()
1197 # Use EXACT same code path as ZMQ browser quit button
1198 import threading
1200 def kill_server():
1201 from openhcs.runtime.zmq_base import ZMQClient
1202 try:
1203 graceful = not is_force_kill
1204 logger.info(f"🛑 {'Gracefully' if graceful else 'Force'} killing server on port {port}...")
1205 success = ZMQClient.kill_server_on_port(port, graceful=graceful)
1207 if success:
1208 logger.info(f"✅ Successfully {'quit' if graceful else 'force killed'} server on port {port}")
1209 # Emit signal to update UI on main thread
1210 self._execution_complete_signal.emit(
1211 {'status': 'cancelled'},
1212 [] # No ready_items needed for cancellation
1213 )
1214 else:
1215 logger.warning(f"❌ Failed to {'quit' if graceful else 'force kill'} server on port {port}")
1216 self._execution_error_signal.emit(f"Failed to stop execution on port {port}")
1218 except Exception as e:
1219 logger.error(f"❌ Error stopping server on port {port}: {e}")
1220 self._execution_error_signal.emit(f"Error stopping execution: {e}")
1222 # Run in background thread (same as ZMQ browser)
1223 thread = threading.Thread(target=kill_server, daemon=True)
1224 thread.start()
1226 return
1228 elif self.current_process and self.current_process.poll() is None: # Still running subprocess
1229 try:
1230 # Kill the entire process group, not just the parent process (matches TUI)
1231 # The subprocess creates its own process group, so we need to kill that group
1232 logger.info(f"🛑 Killing process group for PID {self.current_process.pid}...")
1234 # Get the process group ID (should be same as PID since subprocess calls os.setpgrp())
1235 process_group_id = self.current_process.pid
1237 # Kill entire process group (negative PID kills process group)
1238 import os
1239 import signal
1240 os.killpg(process_group_id, signal.SIGTERM)
1242 # Give processes time to exit gracefully
1243 import asyncio
1244 await asyncio.sleep(1)
1246 # Force kill if still alive
1247 try:
1248 os.killpg(process_group_id, signal.SIGKILL)
1249 logger.info(f"🛑 Force killed process group {process_group_id}")
1250 except ProcessLookupError:
1251 logger.info(f"🛑 Process group {process_group_id} already terminated")
1253 # Reset execution state
1254 self.execution_state = "idle"
1255 self.current_process = None
1257 # Update orchestrator states
1258 for orchestrator in self.orchestrators.values():
1259 if orchestrator.state == OrchestratorState.EXECUTING:
1260 orchestrator._state = OrchestratorState.COMPILED
1262 self.status_message.emit("Execution terminated by user")
1263 self.update_button_states()
1265 # Emit signal for log viewer
1266 self.subprocess_log_stopped.emit()
1268 except Exception as e:
1269 logger.warning(f"🛑 Error killing process group: {e}, falling back to single process kill")
1270 # Fallback to killing just the main process (original behavior)
1271 self.current_process.terminate()
1272 try:
1273 self.current_process.wait(timeout=5)
1274 except subprocess.TimeoutExpired:
1275 self.current_process.kill()
1276 self.current_process.wait()
1278 # Reset state even on fallback
1279 self.execution_state = "idle"
1280 self.current_process = None
1281 self.status_message.emit("Execution terminated by user")
1282 self.update_button_states()
1283 self.subprocess_log_stopped.emit()
1285 def action_code_plate(self):
1286 """Generate Python code for selected plates and their pipelines (Tier 3)."""
1287 logger.debug("Code button pressed - generating Python code for plates")
1289 selected_items = self.get_selected_plates()
1290 if not selected_items:
1291 self.service_adapter.show_error_dialog("No plates selected for code generation")
1292 return
1294 try:
1295 # Collect plate paths, pipeline data, and per-plate pipeline configs
1296 plate_paths = []
1297 pipeline_data = {}
1298 per_plate_configs = {} # Store pipeline config for each plate
1300 for plate_data in selected_items:
1301 plate_path = plate_data['path']
1302 plate_paths.append(plate_path)
1304 # Get pipeline definition for this plate
1305 definition_pipeline = self._get_current_pipeline_definition(plate_path)
1306 if not definition_pipeline:
1307 logger.warning(f"No pipeline defined for {plate_data['name']}, using empty pipeline")
1308 definition_pipeline = []
1310 pipeline_data[plate_path] = definition_pipeline
1312 # Get the actual pipeline config from this plate's orchestrator
1313 if plate_path in self.orchestrators:
1314 orchestrator = self.orchestrators[plate_path]
1315 if orchestrator.pipeline_config:
1316 per_plate_configs[plate_path] = orchestrator.pipeline_config
1318 # Generate complete orchestrator code using new per_plate_configs parameter
1319 from openhcs.debug.pickle_to_python import generate_complete_orchestrator_code
1321 python_code = generate_complete_orchestrator_code(
1322 plate_paths=plate_paths,
1323 pipeline_data=pipeline_data,
1324 global_config=self.global_config,
1325 per_plate_configs=per_plate_configs if per_plate_configs else None,
1326 clean_mode=True # Default to clean mode - only show non-default values
1327 )
1329 # Create simple code editor service (same pattern as tiers 1 & 2)
1330 from openhcs.pyqt_gui.services.simple_code_editor import SimpleCodeEditorService
1331 editor_service = SimpleCodeEditorService(self)
1333 # Check if user wants external editor (check environment variable)
1334 import os
1335 use_external = os.environ.get('OPENHCS_USE_EXTERNAL_EDITOR', '').lower() in ('1', 'true', 'yes')
1337 # Prepare code data for clean mode toggle
1338 code_data = {
1339 'clean_mode': True,
1340 'plate_paths': plate_paths,
1341 'pipeline_data': pipeline_data,
1342 'global_config': self.global_config,
1343 'per_plate_configs': per_plate_configs
1344 }
1346 # Launch editor with callback
1347 editor_service.edit_code(
1348 initial_content=python_code,
1349 title="Edit Orchestrator Configuration",
1350 callback=self._handle_edited_orchestrator_code,
1351 use_external=use_external,
1352 code_type='orchestrator',
1353 code_data=code_data
1354 )
1356 except Exception as e:
1357 logger.error(f"Failed to generate plate code: {e}")
1358 self.service_adapter.show_error_dialog(f"Failed to generate code: {str(e)}")
1360 def _patch_lazy_constructors(self):
1361 """Context manager that patches lazy dataclass constructors to preserve None vs concrete distinction."""
1362 from contextlib import contextmanager
1363 from openhcs.core.lazy_placeholder import LazyDefaultPlaceholderService
1364 import dataclasses
1366 @contextmanager
1367 def patch_context():
1368 # Store original constructors
1369 original_constructors = {}
1371 # Find all lazy dataclass types that need patching
1372 from openhcs.core.config import LazyZarrConfig, LazyStepMaterializationConfig, LazyWellFilterConfig
1373 lazy_types = [LazyZarrConfig, LazyStepMaterializationConfig, LazyWellFilterConfig]
1375 # Add any other lazy types that might be used
1376 for lazy_type in lazy_types:
1377 if LazyDefaultPlaceholderService.has_lazy_resolution(lazy_type):
1378 # Store original constructor
1379 original_constructors[lazy_type] = lazy_type.__init__
1381 # Create patched constructor that uses raw values
1382 def create_patched_init(original_init, dataclass_type):
1383 def patched_init(self, **kwargs):
1384 # Use raw value approach instead of calling original constructor
1385 # This prevents lazy resolution during code execution
1386 for field in dataclasses.fields(dataclass_type):
1387 value = kwargs.get(field.name, None)
1388 object.__setattr__(self, field.name, value)
1390 # Initialize any required lazy dataclass attributes
1391 if hasattr(dataclass_type, '_is_lazy_dataclass'):
1392 object.__setattr__(self, '_is_lazy_dataclass', True)
1394 return patched_init
1396 # Apply the patch
1397 lazy_type.__init__ = create_patched_init(original_constructors[lazy_type], lazy_type)
1399 try:
1400 yield
1401 finally:
1402 # Restore original constructors
1403 for lazy_type, original_init in original_constructors.items():
1404 lazy_type.__init__ = original_init
1406 return patch_context()
1408 def _handle_edited_orchestrator_code(self, edited_code: str):
1409 """Handle edited orchestrator code and update UI state (same logic as Textual TUI)."""
1410 logger.debug("Orchestrator code edited, processing changes...")
1411 try:
1412 # Ensure pipeline editor window is open before processing orchestrator code
1413 main_window = self._find_main_window()
1414 if main_window and hasattr(main_window, 'show_pipeline_editor'):
1415 main_window.show_pipeline_editor()
1417 # CRITICAL FIX: Execute code with lazy dataclass constructor patching to preserve None vs concrete distinction
1418 namespace = {}
1419 with self._patch_lazy_constructors():
1420 exec(edited_code, namespace)
1422 # Extract variables from executed code (same logic as Textual TUI)
1423 if 'plate_paths' in namespace and 'pipeline_data' in namespace:
1424 new_plate_paths = namespace['plate_paths']
1425 new_pipeline_data = namespace['pipeline_data']
1427 # Update global config if present
1428 if 'global_config' in namespace:
1429 new_global_config = namespace['global_config']
1430 # Update the global config (trigger UI refresh)
1431 self.global_config = new_global_config
1433 # CRITICAL: Apply new global config to all orchestrators (was missing!)
1434 # This ensures orchestrators use the updated global config from tier 3 edits
1435 for orchestrator in self.orchestrators.values():
1436 self._update_orchestrator_global_config(orchestrator, new_global_config)
1438 # SIMPLIFIED: Update service adapter (dual-axis resolver handles context)
1439 self.service_adapter.set_global_config(new_global_config)
1441 self.global_config_changed.emit()
1443 # Handle per-plate configs (preferred) or single pipeline_config (legacy)
1444 if 'per_plate_configs' in namespace:
1445 # New per-plate config system
1446 per_plate_configs = namespace['per_plate_configs']
1448 # CRITICAL FIX: Match string keys to actual plate path objects
1449 # The keys in per_plate_configs are strings, but orchestrators dict uses Path/str objects
1450 for plate_path_str, new_pipeline_config in per_plate_configs.items():
1451 # Find matching orchestrator by comparing string representations
1452 matched_orchestrator = None
1453 for orch_key, orchestrator in self.orchestrators.items():
1454 if str(orch_key) == str(plate_path_str):
1455 matched_orchestrator = orchestrator
1456 matched_key = orch_key
1457 break
1459 if matched_orchestrator:
1460 matched_orchestrator.apply_pipeline_config(new_pipeline_config)
1461 # Emit signal for UI components to refresh (including config windows)
1462 effective_config = matched_orchestrator.get_effective_config()
1463 self.orchestrator_config_changed.emit(str(matched_key), effective_config)
1464 logger.debug(f"Applied per-plate pipeline config to orchestrator: {matched_key}")
1465 else:
1466 logger.warning(f"No orchestrator found for plate path: {plate_path_str}")
1467 elif 'pipeline_config' in namespace:
1468 # Legacy single pipeline_config for all plates
1469 new_pipeline_config = namespace['pipeline_config']
1470 # Apply the new pipeline config to all affected orchestrators
1471 for plate_path in new_plate_paths:
1472 if plate_path in self.orchestrators:
1473 orchestrator = self.orchestrators[plate_path]
1474 orchestrator.apply_pipeline_config(new_pipeline_config)
1475 # Emit signal for UI components to refresh (including config windows)
1476 effective_config = orchestrator.get_effective_config()
1477 self.orchestrator_config_changed.emit(str(plate_path), effective_config)
1478 logger.debug(f"Applied tier 3 pipeline config to orchestrator: {plate_path}")
1480 # Update pipeline data for ALL affected plates with proper state invalidation
1481 if self.pipeline_editor and hasattr(self.pipeline_editor, 'plate_pipelines'):
1482 current_plate = getattr(self.pipeline_editor, 'current_plate', None)
1484 for plate_path, new_steps in new_pipeline_data.items():
1485 # Update pipeline data in the pipeline editor
1486 self.pipeline_editor.plate_pipelines[plate_path] = new_steps
1487 logger.debug(f"Updated pipeline for {plate_path} with {len(new_steps)} steps")
1489 # CRITICAL: Invalidate orchestrator state for ALL affected plates
1490 self._invalidate_orchestrator_compilation_state(plate_path)
1492 # If this is the currently displayed plate, trigger UI cascade
1493 if plate_path == current_plate:
1494 # Update the current pipeline steps to trigger cascade
1495 self.pipeline_editor.pipeline_steps = new_steps
1496 # Trigger UI refresh for the current plate
1497 self.pipeline_editor.update_step_list()
1498 # Emit pipeline changed signal to cascade to step editors
1499 self.pipeline_editor.pipeline_changed.emit(new_steps)
1500 logger.debug(f"Triggered UI cascade refresh for current plate: {plate_path}")
1501 else:
1502 logger.warning("No pipeline editor available to update pipeline data")
1504 # Trigger UI refresh
1505 self.pipeline_data_changed.emit()
1507 else:
1508 raise ValueError("No valid assignments found in edited code")
1510 except (SyntaxError, Exception) as e:
1511 import traceback
1512 full_traceback = traceback.format_exc()
1513 logger.error(f"Failed to parse edited orchestrator code: {e}\nFull traceback:\n{full_traceback}")
1514 # Re-raise so the code editor can handle it (keep dialog open, move cursor to error line)
1515 raise
1517 def _invalidate_orchestrator_compilation_state(self, plate_path: str):
1518 """Invalidate compilation state for an orchestrator when its pipeline changes.
1520 This ensures that tier 3 changes properly invalidate ALL affected orchestrators,
1521 not just the currently visible one.
1523 Args:
1524 plate_path: Path of the plate whose orchestrator state should be invalidated
1525 """
1526 # Clear compiled data from simple state
1527 if plate_path in self.plate_compiled_data:
1528 del self.plate_compiled_data[plate_path]
1529 logger.debug(f"Cleared compiled data for {plate_path}")
1531 # Reset orchestrator state to READY (initialized) if it was compiled
1532 orchestrator = self.orchestrators.get(plate_path)
1533 if orchestrator:
1534 from openhcs.constants.constants import OrchestratorState
1535 if orchestrator.state == OrchestratorState.COMPILED:
1536 orchestrator._state = OrchestratorState.READY
1537 logger.debug(f"Reset orchestrator state to READY for {plate_path}")
1539 # Emit state change signal for UI refresh
1540 self.orchestrator_state_changed.emit(plate_path, "READY")
1542 logger.debug(f"Invalidated compilation state for orchestrator: {plate_path}")
1544 def action_view_metadata(self):
1545 """View plate images and metadata in tabbed window. Opens one window per selected plate."""
1546 selected_items = self.get_selected_plates()
1548 if not selected_items:
1549 self.service_adapter.show_error_dialog("No plates selected.")
1550 return
1552 # Open plate viewer for each selected plate
1553 from openhcs.pyqt_gui.windows.plate_viewer_window import PlateViewerWindow
1555 for item in selected_items:
1556 plate_path = item['path']
1558 # Check if orchestrator is initialized
1559 if plate_path not in self.orchestrators:
1560 self.service_adapter.show_error_dialog(f"Plate must be initialized to view: {plate_path}")
1561 continue
1563 orchestrator = self.orchestrators[plate_path]
1565 try:
1566 # Create plate viewer window with tabs (Image Browser + Metadata)
1567 viewer = PlateViewerWindow(
1568 orchestrator=orchestrator,
1569 color_scheme=self.color_scheme,
1570 parent=self
1571 )
1572 viewer.show() # Use show() instead of exec() to allow multiple windows
1573 except Exception as e:
1574 logger.error(f"Failed to open plate viewer for {plate_path}: {e}", exc_info=True)
1575 self.service_adapter.show_error_dialog(f"Failed to open plate viewer: {str(e)}")
1577 # ========== UI Helper Methods ==========
1579 def update_plate_list(self):
1580 """Update the plate list widget using selection preservation mixin."""
1581 def format_plate_item(plate):
1582 """Format plate item for display."""
1583 display_text = f"{plate['name']} ({plate['path']})"
1585 # Add status indicators
1586 status_indicators = []
1587 if plate['path'] in self.orchestrators:
1588 orchestrator = self.orchestrators[plate['path']]
1589 if orchestrator.state == OrchestratorState.READY:
1590 status_indicators.append("✓ Init")
1591 elif orchestrator.state == OrchestratorState.COMPILED:
1592 status_indicators.append("✓ Compiled")
1593 elif orchestrator.state == OrchestratorState.EXECUTING:
1594 status_indicators.append("🔄 Running")
1595 elif orchestrator.state == OrchestratorState.COMPLETED:
1596 status_indicators.append("✅ Complete")
1597 elif orchestrator.state == OrchestratorState.INIT_FAILED:
1598 status_indicators.append("🚫 Init Failed")
1599 elif orchestrator.state == OrchestratorState.COMPILE_FAILED:
1600 status_indicators.append("❌ Compile Failed")
1601 elif orchestrator.state == OrchestratorState.EXEC_FAILED:
1602 status_indicators.append("❌ Exec Failed")
1604 if status_indicators:
1605 display_text = f"[{', '.join(status_indicators)}] {display_text}"
1607 return display_text, plate
1609 def update_func():
1610 """Update function that clears and rebuilds the list."""
1611 self.plate_list.clear()
1613 for plate in self.plates:
1614 display_text, plate_data = format_plate_item(plate)
1615 item = QListWidgetItem(display_text)
1616 item.setData(Qt.ItemDataRole.UserRole, plate_data)
1618 # Add tooltip
1619 if plate['path'] in self.orchestrators:
1620 orchestrator = self.orchestrators[plate['path']]
1621 item.setToolTip(f"Status: {orchestrator.state.value}")
1623 self.plate_list.addItem(item)
1625 # Auto-select first plate if no selection and plates exist
1626 if self.plates and not self.selected_plate_path:
1627 self.plate_list.setCurrentRow(0)
1629 # Use utility to preserve selection during update
1630 preserve_selection_during_update(
1631 self.plate_list,
1632 lambda item_data: item_data['path'] if isinstance(item_data, dict) and 'path' in item_data else str(item_data),
1633 lambda: bool(self.orchestrators),
1634 update_func
1635 )
1636 self.update_button_states()
1638 def get_selected_plates(self) -> List[Dict]:
1639 """
1640 Get currently selected plates.
1642 Returns:
1643 List of selected plate dictionaries
1644 """
1645 selected_items = []
1646 for item in self.plate_list.selectedItems():
1647 plate_data = item.data(Qt.ItemDataRole.UserRole)
1648 if plate_data:
1649 selected_items.append(plate_data)
1650 return selected_items
1652 def get_selected_orchestrator(self):
1653 """
1654 Get the orchestrator for the currently selected plate.
1656 Returns:
1657 PipelineOrchestrator or None if no plate selected or not initialized
1658 """
1659 if self.selected_plate_path and self.selected_plate_path in self.orchestrators:
1660 return self.orchestrators[self.selected_plate_path]
1661 return None
1663 def update_button_states(self):
1664 """Update button enabled/disabled states based on selection."""
1665 selected_plates = self.get_selected_plates()
1666 has_selection = len(selected_plates) > 0
1667 has_initialized = any(plate['path'] in self.orchestrators for plate in selected_plates)
1668 has_compiled = any(plate['path'] in self.plate_compiled_data for plate in selected_plates)
1669 is_running = self.is_any_plate_running()
1671 # Update button states (logic extracted from Textual version)
1672 self.buttons["del_plate"].setEnabled(has_selection and not is_running)
1673 self.buttons["edit_config"].setEnabled(has_initialized and not is_running)
1674 self.buttons["init_plate"].setEnabled(has_selection and not is_running)
1675 self.buttons["compile_plate"].setEnabled(has_initialized and not is_running)
1676 self.buttons["code_plate"].setEnabled(has_initialized and not is_running)
1677 self.buttons["view_metadata"].setEnabled(has_initialized and not is_running)
1679 # Run button - enabled if plates are compiled or if currently running (for stop)
1680 if self.execution_state == "stopping":
1681 # Stopping state - keep button as "Stop" but disable it
1682 self.buttons["run_plate"].setEnabled(False)
1683 self.buttons["run_plate"].setText("Stop")
1684 elif self.execution_state == "force_kill_ready":
1685 # Force kill ready state - button is "Force Kill" and enabled
1686 self.buttons["run_plate"].setEnabled(True)
1687 self.buttons["run_plate"].setText("Force Kill")
1688 elif is_running:
1689 # Running state - button is "Stop" and enabled
1690 self.buttons["run_plate"].setEnabled(True)
1691 self.buttons["run_plate"].setText("Stop")
1692 else:
1693 # Idle state - button is "Run" and enabled if plates are compiled
1694 self.buttons["run_plate"].setEnabled(has_compiled)
1695 self.buttons["run_plate"].setText("Run")
1697 def is_any_plate_running(self) -> bool:
1698 """
1699 Check if any plate is currently running.
1701 Returns:
1702 True if any plate is running, False otherwise
1703 """
1704 # Consider "running", "stopping", and "force_kill_ready" states as "busy"
1705 return self.execution_state in ("running", "stopping", "force_kill_ready")
1707 def update_status(self, message: str):
1708 """
1709 Update status label.
1711 Args:
1712 message: Status message to display
1713 """
1714 self.status_label.setText(message)
1716 def on_selection_changed(self):
1717 """Handle plate list selection changes using utility."""
1718 def on_selected(selected_plates):
1719 self.selected_plate_path = selected_plates[0]['path']
1720 self.plate_selected.emit(self.selected_plate_path)
1722 # SIMPLIFIED: Dual-axis resolver handles context discovery automatically
1723 if self.selected_plate_path in self.orchestrators:
1724 logger.debug(f"Selected orchestrator: {self.selected_plate_path}")
1726 def on_cleared():
1727 self.selected_plate_path = ""
1729 # Use utility to handle selection with prevention
1730 handle_selection_change_with_prevention(
1731 self.plate_list,
1732 self.get_selected_plates,
1733 lambda item_data: item_data['path'] if isinstance(item_data, dict) and 'path' in item_data else str(item_data),
1734 lambda: bool(self.orchestrators),
1735 lambda: self.selected_plate_path,
1736 on_selected,
1737 on_cleared
1738 )
1740 self.update_button_states()
1746 def on_item_double_clicked(self, item: QListWidgetItem):
1747 """Handle double-click on plate item."""
1748 plate_data = item.data(Qt.ItemDataRole.UserRole)
1749 if plate_data:
1750 # Double-click could trigger initialization or configuration
1751 if plate_data['path'] not in self.orchestrators:
1752 self.run_async_action(self.action_init_plate)
1754 def on_orchestrator_state_changed(self, plate_path: str, state: str):
1755 """
1756 Handle orchestrator state changes.
1758 Args:
1759 plate_path: Path of the plate
1760 state: New orchestrator state
1761 """
1762 self.update_plate_list()
1763 logger.debug(f"Orchestrator state changed: {plate_path} -> {state}")
1765 def on_config_changed(self, new_config: GlobalPipelineConfig):
1766 """
1767 Handle global configuration changes.
1769 Args:
1770 new_config: New global configuration
1771 """
1772 self.global_config = new_config
1774 # Apply new global config to all existing orchestrators
1775 # This rebuilds their pipeline configs preserving concrete values
1776 for orchestrator in self.orchestrators.values():
1777 self._update_orchestrator_global_config(orchestrator, new_config)
1779 # REMOVED: Thread-local modification - dual-axis resolver handles orchestrator context automatically
1781 logger.info(f"Applied new global config to {len(self.orchestrators)} orchestrators")
1783 # SIMPLIFIED: Dual-axis resolver handles placeholder updates automatically
1785 # REMOVED: _refresh_all_parameter_form_placeholders and _refresh_widget_parameter_forms
1786 # SIMPLIFIED: Dual-axis resolver handles placeholder updates automatically
1788 # ========== Helper Methods ==========
1790 def _get_current_pipeline_definition(self, plate_path: str) -> List:
1791 """
1792 Get the current pipeline definition for a plate.
1794 Args:
1795 plate_path: Path to the plate
1797 Returns:
1798 List of pipeline steps or empty list if no pipeline
1799 """
1800 if not self.pipeline_editor:
1801 logger.warning("No pipeline editor reference - using empty pipeline")
1802 return []
1804 # Get pipeline for specific plate (same logic as Textual TUI)
1805 if hasattr(self.pipeline_editor, 'plate_pipelines') and plate_path in self.pipeline_editor.plate_pipelines:
1806 pipeline_steps = self.pipeline_editor.plate_pipelines[plate_path]
1807 logger.debug(f"Found pipeline for plate {plate_path} with {len(pipeline_steps)} steps")
1808 return pipeline_steps
1809 else:
1810 logger.debug(f"No pipeline found for plate {plate_path}, using empty pipeline")
1811 return []
1813 def set_pipeline_editor(self, pipeline_editor):
1814 """
1815 Set the pipeline editor reference.
1817 Args:
1818 pipeline_editor: Pipeline editor widget instance
1819 """
1820 self.pipeline_editor = pipeline_editor
1821 logger.debug("Pipeline editor reference set in plate manager")
1823 def _find_main_window(self):
1824 """Find the main window by traversing parent hierarchy."""
1825 widget = self
1826 while widget:
1827 if hasattr(widget, 'floating_windows'):
1828 return widget
1829 widget = widget.parent()
1830 return None
1832 async def _start_monitoring(self):
1833 """Start monitoring subprocess execution."""
1834 if not self.current_process:
1835 return
1837 # Simple monitoring - check if process is still running
1838 def check_process():
1839 if self.current_process and self.current_process.poll() is not None:
1840 # Process has finished
1841 return_code = self.current_process.returncode
1842 logger.info(f"Subprocess finished with return code: {return_code}")
1844 # Reset execution state
1845 self.execution_state = "idle"
1846 self.current_process = None
1848 # Update orchestrator states based on return code
1849 for orchestrator in self.orchestrators.values():
1850 if orchestrator.state == OrchestratorState.EXECUTING:
1851 if return_code == 0:
1852 orchestrator._state = OrchestratorState.COMPLETED
1853 else:
1854 orchestrator._state = OrchestratorState.EXEC_FAILED
1856 if return_code == 0:
1857 self.status_message.emit("Execution completed successfully")
1858 else:
1859 self.status_message.emit(f"Execution failed with code {return_code}")
1861 self.update_button_states()
1863 # Emit signal for log viewer
1864 self.subprocess_log_stopped.emit()
1866 return False # Stop monitoring
1867 return True # Continue monitoring
1869 # Monitor process in background
1870 while check_process():
1871 await asyncio.sleep(1) # Check every second
1873 def _on_progress_started(self, max_value: int):
1874 """Handle progress started signal - route to status bar."""
1875 # Progress is now displayed in the status bar instead of a separate widget
1876 # This method is kept for signal compatibility but doesn't need to do anything
1877 pass
1879 def _on_progress_updated(self, value: int):
1880 """Handle progress updated signal - route to status bar."""
1881 # Progress is now displayed in the status bar instead of a separate widget
1882 # This method is kept for signal compatibility but doesn't need to do anything
1883 pass
1885 def _on_progress_finished(self):
1886 """Handle progress finished signal - route to status bar."""
1887 # Progress is now displayed in the status bar instead of a separate widget
1888 # This method is kept for signal compatibility but doesn't need to do anything
1889 pass
1891 def _handle_compilation_error(self, plate_name: str, error_message: str):
1892 """Handle compilation error on main thread (slot)."""
1893 self.service_adapter.show_error_dialog(f"Compilation failed for {plate_name}: {error_message}")
1895 def _handle_initialization_error(self, plate_name: str, error_message: str):
1896 """Handle initialization error on main thread (slot)."""
1897 self.service_adapter.show_error_dialog(f"Failed to initialize {plate_name}: {error_message}")
1899 def _handle_execution_error(self, error_message: str):
1900 """Handle execution error on main thread (slot)."""
1901 self.service_adapter.show_error_dialog(error_message)