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

11import webbrowser

13from PyQt6.QtWidgets import (

14 QMainWindow, QApplication, QDockWidget, QWidget, QVBoxLayout,

15 QHBoxLayout, QMenuBar, QStatusBar, QToolBar, QSplitter,

16 QMessageBox, QFileDialog, QDialog

17)

18from PyQt6.QtCore import Qt, QSettings, QTimer, pyqtSignal, QUrl

19from PyQt6.QtGui import QAction, QIcon, QKeySequence, QDesktopServices

21from openhcs.core.config import GlobalPipelineConfig

22from openhcs.io.filemanager import FileManager

23from openhcs.io.base import storage_registry

25from openhcs.pyqt_gui.services.service_adapter import PyQtServiceAdapter

27logger = logging.getLogger(__name__)

30class OpenHCSMainWindow(QMainWindow):

31 """

32 Main OpenHCS PyQt6 application window.

34 Implements QDockWidget system to replace textual-window floating windows

35 with native Qt docking, providing better desktop integration.

36 """

38 # Signals for application events

39 config_changed = pyqtSignal(object) # GlobalPipelineConfig

40 status_message = pyqtSignal(str) # Status message

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

43 """

44 Initialize the main OpenHCS window.

46 Args:

47 global_config: Global configuration (uses default if None)

48 """

49 super().__init__()

51 # Core configuration

52 self.global_config = global_config or GlobalPipelineConfig()

54 # Create shared components

55 self.storage_registry = storage_registry

56 self.file_manager = FileManager(self.storage_registry)

58 # Service adapter for Qt integration

59 self.service_adapter = PyQtServiceAdapter(self)

61 # Floating windows registry (replaces dock widgets)

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

64 # Settings for window state persistence

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

67 # Initialize UI

68 self.setup_ui()

69 self.setup_dock_system()

70 self.create_floating_windows()

71 self.setup_menu_bar()

72 self.setup_status_bar()

73 self.setup_connections()

75 # Apply initial theme

76 self.apply_initial_theme()

78 # Restore window state

79 self.restore_window_state()

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

82 self.show_default_windows()

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

86 def setup_ui(self):

87 """Setup basic UI structure."""

88 self.setWindowTitle("OpenHCS")

89 self.setMinimumSize(640, 480)

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

92 from PyQt6.QtCore import Qt

93 self.setWindowFlags(Qt.WindowType.Dialog)

95 # Central widget with system monitor background

96 central_widget = QWidget()

97 central_layout = QVBoxLayout(central_widget)

98 central_layout.setContentsMargins(0, 0, 0, 0)

100 # System monitor widget (background)

101 from openhcs.pyqt_gui.widgets.system_monitor import SystemMonitorWidget

102 self.system_monitor = SystemMonitorWidget()

103 central_layout.addWidget(self.system_monitor)

104

105 self.setCentralWidget(central_widget)

106

107 def apply_initial_theme(self):

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

109 # Get theme manager from service adapter

110 theme_manager = self.service_adapter.get_theme_manager()

111

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

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

114 theme_manager.register_theme_change_callback(self.on_theme_changed)

115

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

117

118 def on_theme_changed(self, color_scheme):

119 """

120 Handle theme change notifications.

121

122 Args:

123 color_scheme: New color scheme that was applied

124 """

125 # Update any main window specific styling if needed

126 # Most styling is handled automatically by the theme manager

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

128

129 def setup_dock_system(self):

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

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

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

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

134 pass

135

136 def create_floating_windows(self):

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

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

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

140 self.floating_windows = {} # Track created windows

141

142 def show_default_windows(self):

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

144 # Show plate manager by default

145 self.show_plate_manager()

146

147 # Show pipeline editor by default

148 self.show_pipeline_editor()

149

150 def show_plate_manager(self):

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

152 if "plate_manager" not in self.floating_windows:

153 from openhcs.pyqt_gui.widgets.plate_manager import PlateManagerWidget

154

155 # Create floating window

156 window = QDialog(self)

157 window.setWindowTitle("Plate Manager")

158 window.setModal(False)

159 window.resize(600, 400)

160

161 # Add widget to window

162 layout = QVBoxLayout(window)

163 plate_widget = PlateManagerWidget(

164 self.file_manager,

165 self.service_adapter,

166 self.service_adapter.get_current_color_scheme()

167 )

168 layout.addWidget(plate_widget)

169

170 self.floating_windows["plate_manager"] = window

171

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

173 self._connect_plate_to_pipeline_manager(plate_widget)

174

175 # Show the window

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

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

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

179

180 def show_pipeline_editor(self):

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

182 if "pipeline_editor" not in self.floating_windows:

183 from openhcs.pyqt_gui.widgets.pipeline_editor import PipelineEditorWidget

184

185 # Create floating window

186 window = QDialog(self)

187 window.setWindowTitle("Pipeline Editor")

188 window.setModal(False)

189 window.resize(800, 600)

190

191 # Add widget to window

192 layout = QVBoxLayout(window)

193 pipeline_widget = PipelineEditorWidget(

194 self.file_manager,

195 self.service_adapter,

196 self.service_adapter.get_current_color_scheme()

197 )

198 layout.addWidget(pipeline_widget)

199

200 self.floating_windows["pipeline_editor"] = window

201

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

203 self._connect_pipeline_to_plate_manager(pipeline_widget)

204

205 # Show the window

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

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

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

209

210

211

212 def show_log_viewer(self):

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

214 if "log_viewer" not in self.floating_windows:

215 from openhcs.pyqt_gui.widgets.log_viewer import LogViewerWindow

216 from openhcs.pyqt_gui.widgets.plate_manager import PlateManagerWidget

217

218 # Create floating window

219 window = QDialog(self)

220 window.setWindowTitle("Log Viewer")

221 window.setModal(False)

222 window.resize(900, 700)

223

224 # Add widget to window

225 layout = QVBoxLayout(window)

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

227 layout.addWidget(log_viewer_widget)

228

229 self.floating_windows["log_viewer"] = window

230

231 # Connect to plate manager signals if it exists

232 if "plate_manager" in self.floating_windows:

233 plate_dialog = self.floating_windows["plate_manager"]

234 # Find the PlateManagerWidget inside the dialog

235 plate_widget = plate_dialog.findChild(PlateManagerWidget)

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

237 plate_widget.clear_subprocess_logs.connect(log_viewer_widget.clear_subprocess_logs)

238 plate_widget.subprocess_log_started.connect(log_viewer_widget.start_monitoring)

239 plate_widget.subprocess_log_stopped.connect(log_viewer_widget.stop_monitoring)

240

241 # Show the window

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

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

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

245

246 def setup_menu_bar(self):

247 """Setup application menu bar."""

248 menubar = self.menuBar()

249

250 # File menu

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

252

253 # Theme submenu

254 theme_menu = file_menu.addMenu("&Theme")

255

256 # Dark theme action

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

258 dark_theme_action.triggered.connect(self.switch_to_dark_theme)

259 theme_menu.addAction(dark_theme_action)

260

261 # Light theme action

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

263 light_theme_action.triggered.connect(self.switch_to_light_theme)

264 theme_menu.addAction(light_theme_action)

265

266 theme_menu.addSeparator()

267

268 # Load theme from file action

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

270 load_theme_action.triggered.connect(self.load_theme_from_file)

271 theme_menu.addAction(load_theme_action)

272

273 # Save theme to file action

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

275 save_theme_action.triggered.connect(self.save_theme_to_file)

276 theme_menu.addAction(save_theme_action)

277

278 file_menu.addSeparator()

279

280 # Exit action

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

282 exit_action.setShortcut(QKeySequence.StandardKey.Quit)

283 exit_action.triggered.connect(self.close)

284 file_menu.addAction(exit_action)

285

286 # View menu

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

288

289 # Plate Manager window

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

291 plate_action.setShortcut("Ctrl+P")

292 plate_action.triggered.connect(self.show_plate_manager)

293 view_menu.addAction(plate_action)

294

295 # Pipeline Editor window

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

297 pipeline_action.setShortcut("Ctrl+E")

298 pipeline_action.triggered.connect(self.show_pipeline_editor)

299 view_menu.addAction(pipeline_action)

300

301 # Log Viewer window

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

303 log_action.setShortcut("Ctrl+L")

304 log_action.triggered.connect(self.show_log_viewer)

305 view_menu.addAction(log_action)

306

307 # Configuration action

308 config_action = QAction("&Global Configuration", self)

309 config_action.setShortcut("Ctrl+G")

310 config_action.triggered.connect(self.show_configuration)

311 view_menu.addAction(config_action)

312

313 view_menu.addSeparator()

314

315

316 # Help menu

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

318

319 # General help action

320 help_action = QAction("&Documentation", self)

321 help_action.setShortcut("F1")

322 help_action.triggered.connect(self.show_help)

323 help_menu.addAction(help_action)

324

325

326 def setup_status_bar(self):

327 """Setup application status bar."""

328 self.status_bar = self.statusBar()

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

330

331 # Connect status message signal

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

333

334 def setup_connections(self):

335 """Setup signal/slot connections."""

336 # Connect config changes

337 self.config_changed.connect(self.on_config_changed)

338

339 # Connect service adapter to application

340 self.service_adapter.set_global_config(self.global_config)

341

342 # Setup auto-save timer for window state

343 self.auto_save_timer = QTimer()

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

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

346

347 def restore_window_state(self):

348 """Restore window state from settings."""

349 try:

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

351 if geometry:

352 self.restoreGeometry(geometry)

353

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

355 if window_state:

356 self.restoreState(window_state)

357

358 except Exception as e:

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

360

361 def save_window_state(self):

362 """Save window state to settings."""

363 # Skip settings save for now to prevent hanging

364 # TODO: Investigate QSettings hanging issue

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

366

367 # Menu action handlers

368 def new_pipeline(self):

369 """Create new pipeline."""

370 if "pipeline_editor" in self.dock_widgets:

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

372 if hasattr(pipeline_widget, 'new_pipeline'):

373 pipeline_widget.new_pipeline()

374

375 def open_pipeline(self):

376 """Open existing pipeline."""

377 file_path, _ = QFileDialog.getOpenFileName(

378 self,

379 "Open Pipeline",

380 "",

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

382 )

383

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

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

386 if hasattr(pipeline_widget, 'load_pipeline'):

387 pipeline_widget.load_pipeline(Path(file_path))

388

389 def save_pipeline(self):

390 """Save current pipeline."""

391 if "pipeline_editor" in self.dock_widgets:

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

393 if hasattr(pipeline_widget, 'save_pipeline'):

394 pipeline_widget.save_pipeline()

395

396 def show_configuration(self):

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

398 from openhcs.pyqt_gui.windows.config_window import ConfigWindow

399

400 def handle_config_save(new_config):

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

402 # new_config is already a GlobalPipelineConfig (concrete class)

403 self.global_config = new_config

404

405 # Update thread-local storage for MaterializationPathConfig defaults

406 from openhcs.core.config import GlobalPipelineConfig

407 from openhcs.config_framework.global_config import set_global_config_for_editing

408 set_global_config_for_editing(GlobalPipelineConfig, new_config)

409

410 # Emit signal for other components to update

411 self.config_changed.emit(new_config)

412

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

414 self._save_config_to_cache(new_config)

415

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

417 config_window = ConfigWindow(

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

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

420 handle_config_save, # on_save_callback

421 self.service_adapter.get_current_color_scheme(), # color_scheme

422 self # parent

423 )

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

425 config_window.show()

426 config_window.raise_()

427 config_window.activateWindow()

428

429 def _connect_pipeline_to_plate_manager(self, pipeline_widget):

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

431 # Get plate manager if it exists

432 if "plate_manager" in self.floating_windows:

433 plate_manager_window = self.floating_windows["plate_manager"]

434

435 # Find the actual plate manager widget

436 plate_manager_widget = None

437 for child in plate_manager_window.findChildren(QWidget):

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

439 plate_manager_widget = child

440 break

441

442 if plate_manager_widget:

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

444 plate_manager_widget.plate_selected.connect(pipeline_widget.set_current_plate)

445

446 # Connect orchestrator config changed signal for placeholder refresh

447 plate_manager_widget.orchestrator_config_changed.connect(pipeline_widget.on_orchestrator_config_changed)

448

449 # Set pipeline editor reference in plate manager

450 if hasattr(plate_manager_widget, 'set_pipeline_editor'):

451 plate_manager_widget.set_pipeline_editor(pipeline_widget)

452

453 # Set current plate if one is already selected

454 if plate_manager_widget.selected_plate_path:

455 pipeline_widget.set_current_plate(plate_manager_widget.selected_plate_path)

456

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

458 else:

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

460 else:

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

462

463 def _connect_plate_to_pipeline_manager(self, plate_manager_widget):

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

465 # Get pipeline editor if it exists

466 if "pipeline_editor" in self.floating_windows:

467 pipeline_editor_window = self.floating_windows["pipeline_editor"]

468

469 # Find the actual pipeline editor widget

470 pipeline_editor_widget = None

471 for child in pipeline_editor_window.findChildren(QWidget):

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

473 pipeline_editor_widget = child

474 break

475

476 if pipeline_editor_widget:

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

478 plate_manager_widget.plate_selected.connect(pipeline_editor_widget.set_current_plate)

479

480 # Connect orchestrator config changed signal for placeholder refresh

481 plate_manager_widget.orchestrator_config_changed.connect(pipeline_editor_widget.on_orchestrator_config_changed)

482

483 # Set pipeline editor reference in plate manager

484 if hasattr(plate_manager_widget, 'set_pipeline_editor'):

485 plate_manager_widget.set_pipeline_editor(pipeline_editor_widget)

486

487 # Set current plate if one is already selected

488 if plate_manager_widget.selected_plate_path:

489 pipeline_editor_widget.set_current_plate(plate_manager_widget.selected_plate_path)

490

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

492 else:

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

494 else:

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

496

497 def show_help(self):

498 """Opens documentation URL in default web browser."""

499 from openhcs.constants.constants import DOCUMENTATION_URL

500 import webbrowser

501

502 url = (DOCUMENTATION_URL)

503 if not QDesktopServices.openUrl(QUrl.fromUserInput(url)):

504 #fallback for wsl users because it wants to be special

505 webbrowser.open(url)

506

507

508 def on_config_changed(self, new_config: GlobalPipelineConfig):

509 """Handle global configuration changes."""

510 self.global_config = new_config

511 self.service_adapter.set_global_config(new_config)

512

513 # Notify all floating windows of config change

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

515 # Get the widget from the window's layout

516 layout = window.layout()

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

518 # Only call on_config_changed if the widget has this method

519 if hasattr(widget, 'on_config_changed'):

520 widget.on_config_changed(new_config)

521

522 def _save_config_to_cache(self, config):

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

524 try:

525 from openhcs.pyqt_gui.services.config_cache_adapter import get_global_config_cache

526 cache = get_global_config_cache()

527 cache.save_config_to_cache_async(config)

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

529 except Exception as e:

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

531

532 def closeEvent(self, event):

533 """Handle application close event."""

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

535

536 try:

537 # Stop system monitor first with timeout

538 if hasattr(self, 'system_monitor'):

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

540 self.system_monitor.stop_monitoring()

541

542 # Close floating windows and cleanup their resources

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

544 try:

545 layout = window.layout()

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

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

548 if hasattr(widget, 'cleanup'):

549 widget.cleanup()

550 window.close()

551 window.deleteLater()

552 except Exception as e:

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

554

555 # Clear floating windows dict

556 self.floating_windows.clear()

557

558 # Save window state

559 self.save_window_state()

560

561 # Force Qt to process pending events before shutdown

562 from PyQt6.QtWidgets import QApplication

563 QApplication.processEvents()

564

565 # Additional cleanup - force garbage collection

566 import gc

567 gc.collect()

568

569 except Exception as e:

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

571

572 # Accept close event

573 event.accept()

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

575

576 # Force application quit with a short delay

577 from PyQt6.QtCore import QTimer

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

579

580 # ========== THEME MANAGEMENT METHODS ==========

581

582 def switch_to_dark_theme(self):

583 """Switch to dark theme variant."""

584 self.service_adapter.switch_to_dark_theme()

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

586

587 def switch_to_light_theme(self):

588 """Switch to light theme variant."""

589 self.service_adapter.switch_to_light_theme()

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

591

592 def load_theme_from_file(self):

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

594 file_path, _ = QFileDialog.getOpenFileName(

595 self,

596 "Load Theme Configuration",

597 "",

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

599 )

600

601 if file_path:

602 success = self.service_adapter.load_theme_from_config(file_path)

603 if success:

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

605 else:

606 QMessageBox.warning(

607 self,

608 "Theme Load Error",

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

610 )

611

612 def save_theme_to_file(self):

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

614 file_path, _ = QFileDialog.getSaveFileName(

615 self,

616 "Save Theme Configuration",

617 "pyqt6_color_scheme.json",

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

619 )

620

621 if file_path:

622 success = self.service_adapter.save_current_theme(file_path)

623 if success:

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

625 else:

626 QMessageBox.warning(

627 self,

628 "Theme Save Error",

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

630 )