Coverage for openhcs/pyqt_gui/main.py: 0.0%

1"""

2OpenHCS PyQt6 Main Window

4Main application window implementing QDockWidget system to replace

5textual-window floating windows with native Qt docking.

6"""

8import logging

9from typing import Optional, Dict, Any

10from pathlib import Path

12from PyQt6.QtWidgets import (

13 QMainWindow, QApplication, QDockWidget, QWidget, QVBoxLayout,

14 QHBoxLayout, QMenuBar, QStatusBar, QToolBar, QSplitter,

15 QMessageBox, QFileDialog, QDialog

16)

17from PyQt6.QtCore import Qt, QSettings, QTimer, pyqtSignal

18from PyQt6.QtGui import QAction, QIcon, QKeySequence

20from openhcs.core.config import GlobalPipelineConfig, get_default_global_config

21from openhcs.io.filemanager import FileManager

22from openhcs.io.base import storage_registry

24from openhcs.pyqt_gui.services.service_adapter import PyQtServiceAdapter

26logger = logging.getLogger(__name__)

29class OpenHCSMainWindow(QMainWindow):

30 """

31 Main OpenHCS PyQt6 application window.

33 Implements QDockWidget system to replace textual-window floating windows

34 with native Qt docking, providing better desktop integration.

35 """

37 # Signals for application events

38 config_changed = pyqtSignal(object) # GlobalPipelineConfig

39 status_message = pyqtSignal(str) # Status message

41 def __init__(self, global_config: Optional[GlobalPipelineConfig] = None):

42 """

43 Initialize the main OpenHCS window.

45 Args:

46 global_config: Global configuration (uses default if None)

47 """

48 super().__init__()

50 # Core configuration

51 self.global_config = global_config or get_default_global_config()

53 # Create shared components

54 self.storage_registry = storage_registry

55 self.file_manager = FileManager(self.storage_registry)

57 # Service adapter for Qt integration

58 self.service_adapter = PyQtServiceAdapter(self)

60 # Floating windows registry (replaces dock widgets)

61 self.floating_windows: Dict[str, QDialog] = {}

63 # Settings for window state persistence

64 self.settings = QSettings("OpenHCS", "PyQt6GUI")

66 # Initialize UI

67 self.setup_ui()

68 self.setup_dock_system()

69 self.create_floating_windows()

70 self.setup_menu_bar()

71 self.setup_status_bar()

72 self.setup_connections()

74 # Apply initial theme

75 self.apply_initial_theme()

77 # Restore window state

78 self.restore_window_state()

80 # Show default windows (plate manager and pipeline editor visible by default)

81 self.show_default_windows()

83 logger.info("OpenHCS PyQt6 main window initialized")

85 def setup_ui(self):

86 """Setup basic UI structure."""

87 self.setWindowTitle("OpenHCS - High-Content Screening Platform")

88 self.setMinimumSize(640, 480)

90 # Make main window floating (not tiled) like other OpenHCS components

91 from PyQt6.QtCore import Qt

92 self.setWindowFlags(Qt.WindowType.Dialog)

94 # Central widget with system monitor background

95 central_widget = QWidget()

96 central_layout = QVBoxLayout(central_widget)

97 central_layout.setContentsMargins(0, 0, 0, 0)

99 # System monitor widget (background)

100 from openhcs.pyqt_gui.widgets.system_monitor import SystemMonitorWidget

101 self.system_monitor = SystemMonitorWidget()

102 central_layout.addWidget(self.system_monitor)

103

104 self.setCentralWidget(central_widget)

105

106 def apply_initial_theme(self):

107 """Apply initial color scheme to the main window."""

108 # Get theme manager from service adapter

109 theme_manager = self.service_adapter.get_theme_manager()

110

111 # Note: ServiceAdapter already applied dark theme globally in its __init__

112 # Just register for theme change notifications, don't re-apply

113 theme_manager.register_theme_change_callback(self.on_theme_changed)

114

115 logger.debug("Registered for theme change notifications (theme already applied by ServiceAdapter)")

116

117 def on_theme_changed(self, color_scheme):

118 """

119 Handle theme change notifications.

120

121 Args:

122 color_scheme: New color scheme that was applied

123 """

124 # Update any main window specific styling if needed

125 # Most styling is handled automatically by the theme manager

126 logger.debug("Main window received theme change notification")

127

128 def setup_dock_system(self):

129 """Setup window system mirroring Textual TUI floating windows."""

130 # In Textual TUI, widgets are floating windows, not docked

131 # We'll create windows on-demand when menu items are clicked

132 # Only the system monitor stays as the central background widget

133 pass

134

135 def create_floating_windows(self):

136 """Create floating windows mirroring Textual TUI window system."""

137 # Windows are created on-demand when menu items are clicked

138 # This mirrors the Textual TUI pattern where windows are mounted dynamically

139 self.floating_windows = {} # Track created windows

140

141 def show_default_windows(self):

142 """Show plate manager and pipeline editor by default (like Textual TUI)."""

143 # Show plate manager by default

144 self.show_plate_manager()

145

146 # Show pipeline editor by default

147 self.show_pipeline_editor()

148

149 def show_plate_manager(self):

150 """Show plate manager window (mirrors Textual TUI pattern)."""

151 if "plate_manager" not in self.floating_windows:

152 from openhcs.pyqt_gui.widgets.plate_manager import PlateManagerWidget

153

154 # Create floating window

155 window = QDialog(self)

156 window.setWindowTitle("Plate Manager")

157 window.setModal(False)

158 window.resize(600, 400)

159

160 # Add widget to window

161 layout = QVBoxLayout(window)

162 plate_widget = PlateManagerWidget(

163 self.file_manager,

164 self.service_adapter,

165 self.service_adapter.get_current_color_scheme()

166 )

167 layout.addWidget(plate_widget)

168

169 self.floating_windows["plate_manager"] = window

170

171 # Connect to pipeline editor if it exists (mirrors Textual TUI)

172 self._connect_plate_to_pipeline_manager(plate_widget)

173

174 # Show the window

175 self.floating_windows["plate_manager"].show()

176 self.floating_windows["plate_manager"].raise_()

177 self.floating_windows["plate_manager"].activateWindow()

178

179 def show_pipeline_editor(self):

180 """Show pipeline editor window (mirrors Textual TUI pattern)."""

181 if "pipeline_editor" not in self.floating_windows:

182 from openhcs.pyqt_gui.widgets.pipeline_editor import PipelineEditorWidget

183

184 # Create floating window

185 window = QDialog(self)

186 window.setWindowTitle("Pipeline Editor")

187 window.setModal(False)

188 window.resize(800, 600)

189

190 # Add widget to window

191 layout = QVBoxLayout(window)

192 pipeline_widget = PipelineEditorWidget(

193 self.file_manager,

194 self.service_adapter,

195 self.service_adapter.get_current_color_scheme()

196 )

197 layout.addWidget(pipeline_widget)

198

199 self.floating_windows["pipeline_editor"] = window

200

201 # Connect to plate manager for current plate selection (mirrors Textual TUI)

202 self._connect_pipeline_to_plate_manager(pipeline_widget)

203

204 # Show the window

205 self.floating_windows["pipeline_editor"].show()

206 self.floating_windows["pipeline_editor"].raise_()

207 self.floating_windows["pipeline_editor"].activateWindow()

208

209

210

211 def show_log_viewer(self):

212 """Show log viewer window (mirrors Textual TUI pattern)."""

213 if "log_viewer" not in self.floating_windows:

214 from openhcs.pyqt_gui.widgets.log_viewer import LogViewerWindow

215 from openhcs.pyqt_gui.widgets.plate_manager import PlateManagerWidget

216

217 # Create floating window

218 window = QDialog(self)

219 window.setWindowTitle("Log Viewer")

220 window.setModal(False)

221 window.resize(900, 700)

222

223 # Add widget to window

224 layout = QVBoxLayout(window)

225 log_viewer_widget = LogViewerWindow(self.file_manager, self.service_adapter)

226 layout.addWidget(log_viewer_widget)

227

228 self.floating_windows["log_viewer"] = window

229

230 # Connect to plate manager signals if it exists

231 if "plate_manager" in self.floating_windows:

232 plate_dialog = self.floating_windows["plate_manager"]

233 # Find the PlateManagerWidget inside the dialog

234 plate_widget = plate_dialog.findChild(PlateManagerWidget)

235 if plate_widget and hasattr(plate_widget, 'clear_subprocess_logs'):

236 plate_widget.clear_subprocess_logs.connect(log_viewer_widget.clear_subprocess_logs)

237 plate_widget.subprocess_log_started.connect(log_viewer_widget.start_monitoring)

238 plate_widget.subprocess_log_stopped.connect(log_viewer_widget.stop_monitoring)

239

240 # Show the window

241 self.floating_windows["log_viewer"].show()

242 self.floating_windows["log_viewer"].raise_()

243 self.floating_windows["log_viewer"].activateWindow()

244

245 def setup_menu_bar(self):

246 """Setup application menu bar."""

247 menubar = self.menuBar()

248

249 # File menu

250 file_menu = menubar.addMenu("&File")

251

252 # New pipeline action

253 new_action = QAction("&New Pipeline", self)

254 new_action.setShortcut(QKeySequence.StandardKey.New)

255 new_action.triggered.connect(self.new_pipeline)

256 file_menu.addAction(new_action)

257

258 # Open pipeline action

259 open_action = QAction("&Open Pipeline", self)

260 open_action.setShortcut(QKeySequence.StandardKey.Open)

261 open_action.triggered.connect(self.open_pipeline)

262 file_menu.addAction(open_action)

263

264 # Save pipeline action

265 save_action = QAction("&Save Pipeline", self)

266 save_action.setShortcut(QKeySequence.StandardKey.Save)

267 save_action.triggered.connect(self.save_pipeline)

268 file_menu.addAction(save_action)

269

270 file_menu.addSeparator()

271

272 # Exit action

273 exit_action = QAction("E&xit", self)

274 exit_action.setShortcut(QKeySequence.StandardKey.Quit)

275 exit_action.triggered.connect(self.close)

276 file_menu.addAction(exit_action)

277

278 # View menu

279 view_menu = menubar.addMenu("&View")

280

281 # Plate Manager window

282 plate_action = QAction("&Plate Manager", self)

283 plate_action.setShortcut("Ctrl+P")

284 plate_action.triggered.connect(self.show_plate_manager)

285 view_menu.addAction(plate_action)

286

287 # Pipeline Editor window

288 pipeline_action = QAction("Pipeline &Editor", self)

289 pipeline_action.setShortcut("Ctrl+E")

290 pipeline_action.triggered.connect(self.show_pipeline_editor)

291 view_menu.addAction(pipeline_action)

292

293

294

295 # Log Viewer window

296 log_action = QAction("&Log Viewer", self)

297 log_action.setShortcut("Ctrl+L")

298 log_action.triggered.connect(self.show_log_viewer)

299 view_menu.addAction(log_action)

300

301 # Tools menu

302 tools_menu = menubar.addMenu("&Tools")

303

304 # Configuration action

305 config_action = QAction("&Configuration", self)

306 config_action.triggered.connect(self.show_configuration)

307 tools_menu.addAction(config_action)

308

309 tools_menu.addSeparator()

310

311 # Theme submenu

312 theme_menu = tools_menu.addMenu("&Theme")

313

314 # Dark theme action

315 dark_theme_action = QAction("&Dark Theme", self)

316 dark_theme_action.triggered.connect(self.switch_to_dark_theme)

317 theme_menu.addAction(dark_theme_action)

318

319 # Light theme action

320 light_theme_action = QAction("&Light Theme", self)

321 light_theme_action.triggered.connect(self.switch_to_light_theme)

322 theme_menu.addAction(light_theme_action)

323

324 theme_menu.addSeparator()

325

326 # Load theme from file action

327 load_theme_action = QAction("&Load Theme from File...", self)

328 load_theme_action.triggered.connect(self.load_theme_from_file)

329 theme_menu.addAction(load_theme_action)

330

331 # Save theme to file action

332 save_theme_action = QAction("&Save Theme to File...", self)

333 save_theme_action.triggered.connect(self.save_theme_to_file)

334 theme_menu.addAction(save_theme_action)

335

336 # Help menu

337 help_menu = menubar.addMenu("&Help")

338

339 # General help action

340 help_action = QAction("&OpenHCS Help", self)

341 help_action.setShortcut("F1")

342 help_action.triggered.connect(self.show_help)

343 help_menu.addAction(help_action)

344

345 help_menu.addSeparator()

346

347 # About action

348 about_action = QAction("&About OpenHCS", self)

349 about_action.triggered.connect(self.show_about)

350 help_menu.addAction(about_action)

351

352 def setup_status_bar(self):

353 """Setup application status bar."""

354 self.status_bar = self.statusBar()

355 self.status_bar.showMessage("OpenHCS PyQt6 GUI Ready")

356

357 # Connect status message signal

358 self.status_message.connect(self.status_bar.showMessage)

359

360 def setup_connections(self):

361 """Setup signal/slot connections."""

362 # Connect config changes

363 self.config_changed.connect(self.on_config_changed)

364

365 # Connect service adapter to application

366 self.service_adapter.set_global_config(self.global_config)

367

368 # Setup auto-save timer for window state

369 self.auto_save_timer = QTimer()

370 self.auto_save_timer.timeout.connect(self.save_window_state)

371 self.auto_save_timer.start(30000) # Save every 30 seconds

372

373 def restore_window_state(self):

374 """Restore window state from settings."""

375 try:

376 geometry = self.settings.value("geometry")

377 if geometry:

378 self.restoreGeometry(geometry)

379

380 window_state = self.settings.value("windowState")

381 if window_state:

382 self.restoreState(window_state)

383

384 except Exception as e:

385 logger.warning(f"Failed to restore window state: {e}")

386

387 def save_window_state(self):

388 """Save window state to settings."""

389 # Skip settings save for now to prevent hanging

390 # TODO: Investigate QSettings hanging issue

391 logger.debug("Skipping window state save to prevent hanging")

392

393 # Menu action handlers

394 def new_pipeline(self):

395 """Create new pipeline."""

396 if "pipeline_editor" in self.dock_widgets:

397 pipeline_widget = self.dock_widgets["pipeline_editor"].widget()

398 if hasattr(pipeline_widget, 'new_pipeline'):

399 pipeline_widget.new_pipeline()

400

401 def open_pipeline(self):

402 """Open existing pipeline."""

403 file_path, _ = QFileDialog.getOpenFileName(

404 self,

405 "Open Pipeline",

406 "",

407 "Function Files (*.func);;All Files (*)"

408 )

409

410 if file_path and "pipeline_editor" in self.dock_widgets:

411 pipeline_widget = self.dock_widgets["pipeline_editor"].widget()

412 if hasattr(pipeline_widget, 'load_pipeline'):

413 pipeline_widget.load_pipeline(Path(file_path))

414

415 def save_pipeline(self):

416 """Save current pipeline."""

417 if "pipeline_editor" in self.dock_widgets:

418 pipeline_widget = self.dock_widgets["pipeline_editor"].widget()

419 if hasattr(pipeline_widget, 'save_pipeline'):

420 pipeline_widget.save_pipeline()

421

422 def show_configuration(self):

423 """Show configuration dialog for global config editing."""

424 from openhcs.pyqt_gui.windows.config_window import ConfigWindow

425

426 def handle_config_save(new_config):

427 """Handle configuration save (mirrors Textual TUI pattern)."""

428 # new_config is already a GlobalPipelineConfig (concrete class)

429 self.global_config = new_config

430

431 # Update thread-local storage for MaterializationPathConfig defaults

432 from openhcs.core.config import set_current_global_config, GlobalPipelineConfig

433 set_current_global_config(GlobalPipelineConfig, new_config)

434

435 # Emit signal for other components to update

436 self.config_changed.emit(new_config)

437

438 # Save config to cache for future sessions (matches TUI)

439 self._save_config_to_cache(new_config)

440

441 # Use concrete GlobalPipelineConfig for global config editing (static context)

442 config_window = ConfigWindow(

443 GlobalPipelineConfig, # config_class (concrete class for static context)

444 self.service_adapter.get_global_config(), # current_config (concrete instance)

445 handle_config_save, # on_save_callback

446 self.service_adapter.get_current_color_scheme(), # color_scheme

447 self, # parent

448 is_global_config_editing=True # This is global config editing

449 )

450 # Show as non-modal window (like plate manager and pipeline editor)

451 config_window.show()

452 config_window.raise_()

453 config_window.activateWindow()

454

455 def _connect_pipeline_to_plate_manager(self, pipeline_widget):

456 """Connect pipeline editor to plate manager (mirrors Textual TUI pattern)."""

457 # Get plate manager if it exists

458 if "plate_manager" in self.floating_windows:

459 plate_manager_window = self.floating_windows["plate_manager"]

460

461 # Find the actual plate manager widget

462 plate_manager_widget = None

463 for child in plate_manager_window.findChildren(QWidget):

464 if hasattr(child, 'selected_plate_path') and hasattr(child, 'orchestrators'):

465 plate_manager_widget = child

466 break

467

468 if plate_manager_widget:

469 # Connect plate selection signal to pipeline editor (mirrors Textual TUI)

470 plate_manager_widget.plate_selected.connect(pipeline_widget.set_current_plate)

471

472 # Set pipeline editor reference in plate manager

473 if hasattr(plate_manager_widget, 'set_pipeline_editor'):

474 plate_manager_widget.set_pipeline_editor(pipeline_widget)

475

476 # Set current plate if one is already selected

477 if plate_manager_widget.selected_plate_path:

478 pipeline_widget.set_current_plate(plate_manager_widget.selected_plate_path)

479

480 logger.debug("Connected pipeline editor to plate manager")

481 else:

482 logger.warning("Could not find plate manager widget to connect")

483 else:

484 logger.debug("Plate manager not yet created - connection will be made when both exist")

485

486 def _connect_plate_to_pipeline_manager(self, plate_manager_widget):

487 """Connect plate manager to pipeline editor (reverse direction)."""

488 # Get pipeline editor if it exists

489 if "pipeline_editor" in self.floating_windows:

490 pipeline_editor_window = self.floating_windows["pipeline_editor"]

491

492 # Find the actual pipeline editor widget

493 pipeline_editor_widget = None

494 for child in pipeline_editor_window.findChildren(QWidget):

495 if hasattr(child, 'set_current_plate') and hasattr(child, 'pipeline_steps'):

496 pipeline_editor_widget = child

497 break

498

499 if pipeline_editor_widget:

500 # Connect plate selection signal to pipeline editor (mirrors Textual TUI)

501 plate_manager_widget.plate_selected.connect(pipeline_editor_widget.set_current_plate)

502

503 # Set pipeline editor reference in plate manager

504 if hasattr(plate_manager_widget, 'set_pipeline_editor'):

505 plate_manager_widget.set_pipeline_editor(pipeline_editor_widget)

506

507 # Set current plate if one is already selected

508 if plate_manager_widget.selected_plate_path:

509 pipeline_editor_widget.set_current_plate(plate_manager_widget.selected_plate_path)

510

511 logger.debug("Connected plate manager to pipeline editor")

512 else:

513 logger.warning("Could not find pipeline editor widget to connect")

514 else:

515 logger.debug("Pipeline editor not yet created - connection will be made when both exist")

516

517 def show_help(self):

518 """Show general OpenHCS help - reuses Textual TUI help system."""

519 from openhcs.pyqt_gui.windows.help_window import HelpWindow

520

521 # Create and show help window (reuses existing help content)

522 help_window = HelpWindow(parent=self)

523 help_window.show()

524

525 def show_about(self):

526 """Show about dialog."""

527 QMessageBox.about(

528 self,

529 "About OpenHCS",

530 "OpenHCS - High-Content Screening Platform\n\n"

531 "A comprehensive platform for microscopy image processing\n"

532 "and high-content screening analysis.\n\n"

533 "PyQt6 GUI Version 1.0.0"

534 )

535

536 def on_config_changed(self, new_config: GlobalPipelineConfig):

537 """Handle global configuration changes."""

538 self.global_config = new_config

539 self.service_adapter.set_global_config(new_config)

540

541 # Notify all floating windows of config change

542 for window in self.floating_windows.values():

543 # Get the widget from the window's layout

544 layout = window.layout()

545 widget = layout.itemAt(0).widget()

546 # Only call on_config_changed if the widget has this method

547 if hasattr(widget, 'on_config_changed'):

548 widget.on_config_changed(new_config)

549

550 def _save_config_to_cache(self, config):

551 """Save config to cache asynchronously (matches TUI pattern)."""

552 try:

553 from openhcs.pyqt_gui.services.config_cache_adapter import get_global_config_cache

554 cache = get_global_config_cache()

555 cache.save_config_to_cache_async(config)

556 logger.info("Global config save to cache initiated")

557 except Exception as e:

558 logger.error(f"Error saving global config to cache: {e}")

559

560 def closeEvent(self, event):

561 """Handle application close event."""

562 logger.info("Starting application shutdown...")

563

564 try:

565 # Stop system monitor first with timeout

566 if hasattr(self, 'system_monitor'):

567 logger.info("Stopping system monitor...")

568 self.system_monitor.stop_monitoring()

569

570 # Close floating windows and cleanup their resources

571 for window_name, window in list(self.floating_windows.items()):

572 try:

573 layout = window.layout()

574 if layout and layout.count() > 0:

575 widget = layout.itemAt(0).widget()

576 if hasattr(widget, 'cleanup'):

577 widget.cleanup()

578 window.close()

579 window.deleteLater()

580 except Exception as e:

581 logger.warning(f"Error cleaning up window {window_name}: {e}")

582

583 # Clear floating windows dict

584 self.floating_windows.clear()

585

586 # Save window state

587 self.save_window_state()

588

589 # Force Qt to process pending events before shutdown

590 from PyQt6.QtWidgets import QApplication

591 QApplication.processEvents()

592

593 # Additional cleanup - force garbage collection

594 import gc

595 gc.collect()

596

597 except Exception as e:

598 logger.error(f"Error during shutdown: {e}")

599

600 # Accept close event

601 event.accept()

602 logger.info("OpenHCS PyQt6 application closed")

603

604 # Force application quit with a short delay

605 from PyQt6.QtCore import QTimer

606 QTimer.singleShot(100, lambda: QApplication.instance().quit())

607

608 # ========== THEME MANAGEMENT METHODS ==========

609

610 def switch_to_dark_theme(self):

611 """Switch to dark theme variant."""

612 self.service_adapter.switch_to_dark_theme()

613 self.status_message.emit("Switched to dark theme")

614

615 def switch_to_light_theme(self):

616 """Switch to light theme variant."""

617 self.service_adapter.switch_to_light_theme()

618 self.status_message.emit("Switched to light theme")

619

620 def load_theme_from_file(self):

621 """Load theme from JSON configuration file."""

622 file_path, _ = QFileDialog.getOpenFileName(

623 self,

624 "Load Theme Configuration",

625 "",

626 "JSON Files (*.json);;All Files (*)"

627 )

628

629 if file_path:

630 success = self.service_adapter.load_theme_from_config(file_path)

631 if success:

632 self.status_message.emit(f"Loaded theme from {Path(file_path).name}")

633 else:

634 QMessageBox.warning(

635 self,

636 "Theme Load Error",

637 f"Failed to load theme from {Path(file_path).name}"

638 )

639

640 def save_theme_to_file(self):

641 """Save current theme to JSON configuration file."""

642 file_path, _ = QFileDialog.getSaveFileName(

643 self,

644 "Save Theme Configuration",

645 "pyqt6_color_scheme.json",

646 "JSON Files (*.json);;All Files (*)"

647 )

648

649 if file_path:

650 success = self.service_adapter.save_current_theme(file_path)

651 if success:

652 self.status_message.emit(f"Saved theme to {Path(file_path).name}")

653 else:

654 QMessageBox.warning(

655 self,

656 "Theme Save Error",

657 f"Failed to save theme to {Path(file_path).name}"

658 )