Coverage for openhcs/pyqt_gui/services/service_adapter.py: 0.0%
184 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +0000
1"""
2PyQt6 Service Adapter
4Bridges OpenHCS services to PyQt6 context, replacing prompt_toolkit dependencies
5with Qt equivalents while preserving all business logic.
6"""
8import logging
9from typing import Any, Optional
10from pathlib import Path
12from PyQt6.QtWidgets import QMessageBox, QFileDialog, QApplication, QWidget
13from PyQt6.QtCore import QProcess, QThread, pyqtSignal
14from PyQt6.QtGui import QDesktopServices
15from PyQt6.QtCore import QUrl
17from openhcs.core.path_cache import PathCacheKey, get_cached_dialog_path, cache_dialog_path
18from openhcs.pyqt_gui.shared.palette_manager import ThemeManager
19from openhcs.pyqt_gui.shared.color_scheme import PyQt6ColorScheme
21logger = logging.getLogger(__name__)
24class PyQtServiceAdapter:
25 """
26 Adapter to bridge OpenHCS services to PyQt6 context.
28 Replaces prompt_toolkit dependencies (dialogs, system commands, etc.)
29 with PyQt6 equivalents while maintaining the same interface for services.
30 """
32 def __init__(self, main_window: QWidget):
33 """
34 Initialize the service adapter.
36 Args:
37 main_window: Main PyQt6 window for dialog parenting
38 """
39 self.main_window = main_window
40 self.app = QApplication.instance()
42 # Initialize theme manager for centralized color management
43 self.theme_manager = ThemeManager()
45 # Apply dark theme globally to ensure consistent dialog styling
46 self._apply_dark_theme()
48 logger.debug("PyQt6 service adapter initialized")
50 def _apply_dark_theme(self):
51 """Apply dark theme globally for consistent dialog styling."""
52 try:
53 # Create dark color scheme (same as enhanced path widget)
54 dark_scheme = PyQt6ColorScheme()
56 # Apply the dark theme globally so all dialogs use consistent styling
57 self.theme_manager.apply_color_scheme(dark_scheme)
59 logger.debug("Applied dark theme globally for consistent dialog styling")
60 except Exception as e:
61 logger.warning(f"Failed to apply dark theme: {e}")
63 def get_theme_manager(self):
64 """Get the theme manager instance."""
65 return self.theme_manager
67 def get_current_color_scheme(self):
68 """Get the current color scheme."""
69 return self.theme_manager.color_scheme
71 def execute_async_operation(self, async_func, *args, **kwargs):
72 """
73 Execute async operation using ThreadPoolExecutor (simpler and more reliable).
75 Args:
76 async_func: Async function to execute
77 *args: Function arguments
78 **kwargs: Function keyword arguments
79 """
80 import asyncio
81 from concurrent.futures import ThreadPoolExecutor
83 def run_async_in_thread():
84 """Run async function in thread with its own event loop."""
85 try:
86 # Create new event loop for this thread (like TUI executor)
87 loop = asyncio.new_event_loop()
88 asyncio.set_event_loop(loop)
90 # Run the async function
91 result = loop.run_until_complete(async_func(*args, **kwargs))
93 # Clean up
94 loop.close()
96 return result
98 except Exception as e:
99 logger.error(f"Async operation failed: {e}", exc_info=True)
100 raise
102 # Use ThreadPoolExecutor (simpler than Qt threading)
103 if not hasattr(self, '_thread_pool'):
104 self._thread_pool = ThreadPoolExecutor(max_workers=4)
106 # Submit to thread pool (non-blocking like TUI executor)
107 future = self._thread_pool.submit(run_async_in_thread)
109 def show_dialog(self, content: str, title: str = "OpenHCS") -> bool:
110 """
111 Replace prompt_toolkit dialogs with QMessageBox.
113 Args:
114 content: Dialog content text
115 title: Dialog title
117 Returns:
118 True if user clicked OK, False otherwise
119 """
120 msg = QMessageBox(self.main_window)
121 msg.setWindowTitle(title)
122 msg.setText(content)
123 msg.setStandardButtons(QMessageBox.StandardButton.Ok | QMessageBox.StandardButton.Cancel)
124 msg.setDefaultButton(QMessageBox.StandardButton.Ok)
126 result = msg.exec()
127 return result == QMessageBox.StandardButton.Ok
129 def show_error_dialog(self, error_message: str, title: str = "Error") -> None:
130 """
131 Show error dialog with error icon.
133 Args:
134 error_message: Error message to display
135 title: Dialog title
136 """
137 msg = QMessageBox(self.main_window)
138 msg.setWindowTitle(title)
139 msg.setText(error_message)
140 msg.setIcon(QMessageBox.Icon.Critical)
141 msg.setStandardButtons(QMessageBox.StandardButton.Ok)
142 msg.exec()
144 def show_info_dialog(self, info_message: str, title: str = "Information") -> None:
145 """
146 Show information dialog.
148 Args:
149 info_message: Information message to display
150 title: Dialog title
151 """
152 msg = QMessageBox(self.main_window)
153 msg.setWindowTitle(title)
154 msg.setText(info_message)
155 msg.setIcon(QMessageBox.Icon.Information)
156 msg.setStandardButtons(QMessageBox.StandardButton.Ok)
157 msg.exec()
159 def show_cached_file_dialog(
160 self,
161 cache_key: PathCacheKey,
162 title: str = "Select File",
163 file_filter: str = "All Files (*)",
164 mode: str = "open",
165 fallback_path: Optional[Path] = None
166 ) -> Optional[Path]:
167 """
168 Show file dialog with path caching (mirrors Textual TUI pattern).
170 Args:
171 cache_key: Cache key for remembering last used path
172 title: Dialog title
173 file_filter: File filter string (e.g., "Pipeline Files (*.pipeline)")
174 mode: "open" or "save"
175 fallback_path: Fallback path if no cached path exists
177 Returns:
178 Selected file path or None if cancelled
179 """
180 # Get cached initial directory
181 initial_dir = str(get_cached_dialog_path(cache_key, fallback_path))
183 try:
184 if mode == "save":
185 file_path, _ = QFileDialog.getSaveFileName(
186 self.main_window,
187 title,
188 initial_dir,
189 file_filter
190 )
191 else: # mode == "open"
192 file_path, _ = QFileDialog.getOpenFileName(
193 self.main_window,
194 title,
195 initial_dir,
196 file_filter
197 )
199 if file_path:
200 selected_path = Path(file_path)
201 # Cache the parent directory for future dialogs
202 cache_dialog_path(cache_key, selected_path.parent)
203 return selected_path
205 return None
207 except Exception as e:
208 logger.error(f"File dialog failed: {e}")
209 raise
211 def show_cached_directory_dialog(
212 self,
213 cache_key: PathCacheKey,
214 title: str = "Select Directory",
215 fallback_path: Optional[Path] = None
216 ) -> Optional[Path]:
217 """
218 Show directory dialog with path caching.
220 Args:
221 cache_key: Cache key for remembering last used path
222 title: Dialog title
223 fallback_path: Fallback path if no cached path exists
225 Returns:
226 Selected directory path or None if cancelled
227 """
228 # Get cached initial directory
229 initial_dir = str(get_cached_dialog_path(cache_key, fallback_path))
231 try:
232 dir_path = QFileDialog.getExistingDirectory(
233 self.main_window,
234 title,
235 initial_dir
236 )
238 if dir_path:
239 selected_path = Path(dir_path)
240 # Cache the selected directory
241 cache_dialog_path(cache_key, selected_path)
242 return selected_path
244 return None
246 except Exception as e:
247 logger.error(f"Directory dialog failed: {e}")
248 raise
250 def run_system_command(self, command: str, wait_for_finish: bool = True) -> bool:
251 """
252 Replace prompt_toolkit system command with QProcess.
254 Args:
255 command: System command to execute
256 wait_for_finish: Whether to wait for command completion
258 Returns:
259 True if command executed successfully, False otherwise
260 """
261 try:
262 process = QProcess(self.main_window)
264 if wait_for_finish:
265 process.start(command)
266 success = process.waitForFinished(30000) # 30 second timeout
267 return success and process.exitCode() == 0
268 else:
269 # Start detached process
270 return process.startDetached(command)
272 except Exception as e:
273 logger.error(f"System command failed: {command} - {e}")
274 self.show_error_dialog(f"Command failed: {e}")
275 return False
277 def open_external_editor(self, file_path: Path) -> bool:
278 """
279 Open file in external editor using system default.
281 Args:
282 file_path: Path to file to edit
284 Returns:
285 True if editor opened successfully, False otherwise
286 """
287 try:
288 url = QUrl.fromLocalFile(str(file_path))
289 return QDesktopServices.openUrl(url)
290 except Exception as e:
291 logger.error(f"Failed to open external editor: {e}")
292 self.show_error_dialog(f"Failed to open editor: {e}")
293 return False
295 def get_global_config(self):
296 """
297 Get global configuration from application.
299 Returns:
300 Global configuration object
301 """
302 # Access global config through application property
303 if hasattr(self.app, 'global_config'):
304 return self.app.global_config
305 else:
306 # Fallback to default config
307 from openhcs.core.config import get_default_global_config
308 return get_default_global_config()
310 def set_global_config(self, config):
311 """
312 Set global configuration on application.
314 Args:
315 config: Global configuration object
316 """
317 if hasattr(self.app, 'global_config'):
318 self.app.global_config = config
319 else:
320 # Set as application property
321 setattr(self.app, 'global_config', config)
323 # ========== THEME MANAGEMENT METHODS ==========
325 def get_theme_manager(self) -> ThemeManager:
326 """
327 Get the theme manager for color scheme management.
329 Returns:
330 ThemeManager: Current theme manager instance
331 """
332 return self.theme_manager
334 def get_current_color_scheme(self) -> PyQt6ColorScheme:
335 """
336 Get the current color scheme.
338 Returns:
339 PyQt6ColorScheme: Current color scheme
340 """
341 return self.theme_manager.color_scheme
343 def apply_color_scheme(self, color_scheme: PyQt6ColorScheme):
344 """
345 Apply a new color scheme to the entire application.
347 Args:
348 color_scheme: New color scheme to apply
349 """
350 self.theme_manager.apply_color_scheme(color_scheme)
352 def switch_to_dark_theme(self):
353 """Switch to dark theme variant."""
354 self.theme_manager.switch_to_dark_theme()
356 def switch_to_light_theme(self):
357 """Switch to light theme variant."""
358 self.theme_manager.switch_to_light_theme()
360 def load_theme_from_config(self, config_path: str) -> bool:
361 """
362 Load and apply theme from configuration file.
364 Args:
365 config_path: Path to JSON configuration file
367 Returns:
368 bool: True if successful, False otherwise
369 """
370 return self.theme_manager.load_theme_from_config(config_path)
372 def save_current_theme(self, config_path: str) -> bool:
373 """
374 Save current theme to configuration file.
376 Args:
377 config_path: Path to save JSON configuration file
379 Returns:
380 bool: True if successful, False otherwise
381 """
382 return self.theme_manager.save_current_theme(config_path)
384 def get_current_style_sheet(self) -> str:
385 """
386 Get the current complete application style sheet.
388 Returns:
389 str: Complete QStyleSheet for current theme
390 """
391 return self.theme_manager.get_current_style_sheet()
393 def register_theme_change_callback(self, callback):
394 """
395 Register a callback to be called when theme changes.
397 Args:
398 callback: Function to call with new color scheme
399 """
400 self.theme_manager.register_theme_change_callback(callback)
402 def get_file_manager(self):
403 """
404 Get FileManager instance from application.
406 Returns:
407 FileManager instance
408 """
409 if hasattr(self.app, 'file_manager'):
410 return self.app.file_manager
411 else:
412 # Create default FileManager
413 from openhcs.io.filemanager import FileManager
414 from openhcs.io.base import storage_registry
415 file_manager = FileManager(storage_registry)
416 setattr(self.app, 'file_manager', file_manager)
417 return file_manager
420class ExternalEditorProcess(QThread):
421 """
422 Thread for handling external editor processes.
424 Replaces prompt_toolkit's run_system_command for external editor integration.
425 """
427 finished = pyqtSignal(bool, str) # success, error_message
429 def __init__(self, command: str, file_path: Path):
430 super().__init__()
431 self.command = command
432 self.file_path = file_path
434 def run(self):
435 """Execute external editor command in thread."""
436 try:
437 process = QProcess()
438 process.start(self.command)
440 success = process.waitForFinished(300000) # 5 minute timeout
442 if success and process.exitCode() == 0:
443 self.finished.emit(True, "")
444 else:
445 error_msg = process.readAllStandardError().data().decode()
446 self.finished.emit(False, f"Editor failed: {error_msg}")
448 except Exception as e:
449 self.finished.emit(False, f"Editor process failed: {e}")
452class AsyncOperationThread(QThread):
453 """
454 Generic thread for async operations.
456 Converts async operations to Qt thread-based operations.
457 """
459 result_ready = pyqtSignal(object)
460 error_occurred = pyqtSignal(str)
462 def __init__(self, async_func, *args, **kwargs):
463 super().__init__()
464 self.async_func = async_func
465 self.args = args
466 self.kwargs = kwargs
468 def run(self):
469 """Execute async function in thread with event loop."""
470 try:
471 import asyncio
473 # Create new event loop for this thread
474 loop = asyncio.new_event_loop()
475 asyncio.set_event_loop(loop)
477 # Run async function
478 result = loop.run_until_complete(
479 self.async_func(*self.args, **self.kwargs)
480 )
482 self.result_ready.emit(result)
484 except Exception as e:
485 logger.error(f"Async operation failed: {e}")
486 self.error_occurred.emit(str(e))
487 finally:
488 # Clean up event loop
489 loop.close()