Coverage for openhcs/pyqt_gui/services/config_cache_adapter.py: 0.0%

1""" 

2PyQt-specific adapter for unified configuration cache. 

3 

4Provides Qt-compatible interface for existing PyQt GUI code. 

5""" 

6 

7from PyQt6.QtCore import QObject, pyqtSignal, QRunnable, QThreadPool 

8from openhcs.core.config_cache import ( 

9 load_cached_global_config_sync, 

10 get_global_config_cache as get_core_config_cache, 

11 QtExecutionStrategy, 

12 _sync_load_config, 

13 _sync_save_config 

14) 

15from openhcs.core.config import GlobalPipelineConfig 

16from pathlib import Path 

17 

18 

19class ConfigCacheWorker(QRunnable): 

20 """Qt worker for cache operations.""" 

21 

22 def __init__(self, operation: str, cache_file: Path, config=None, callback=None): 

23 super().__init__() 

24 self.operation = operation 

25 self.cache_file = cache_file 

26 self.config = config 

27 self.callback = callback 

28 

29 def run(self): 

30 if self.operation == 'load': 

31 result = _sync_load_config(self.cache_file) 

32 if self.callback: 

33 self.callback(result) 

34 elif self.operation == 'save': 

35 result = _sync_save_config(self.config, self.cache_file) 

36 if self.callback: 

37 self.callback(result) 

38 

39 

40class QtGlobalConfigCache(QObject): 

41 """Qt-specific wrapper for unified config cache.""" 

42 

43 config_loaded = pyqtSignal(object) # GlobalPipelineConfig or None 

44 config_saved = pyqtSignal(bool) # Success/failure 

45 

46 def __init__(self, cache_file=None): 

47 super().__init__() 

48 if cache_file is None: 

49 from openhcs.core.xdg_paths import get_config_file_path 

50 cache_file = get_config_file_path("global_config.config") 

51 self.cache_file = cache_file 

52 self.thread_pool = QThreadPool() 

53 

54 def load_cached_config_async(self): 

55 """Load cached config asynchronously with Qt threading.""" 

56 worker = ConfigCacheWorker('load', self.cache_file, callback=self._on_load_finished) 

57 self.thread_pool.start(worker) 

58 

59 def save_config_to_cache_async(self, config: GlobalPipelineConfig): 

60 """Save config asynchronously with Qt threading.""" 

61 worker = ConfigCacheWorker('save', self.cache_file, config, callback=self._on_save_finished) 

62 self.thread_pool.start(worker) 

63 

64 def _on_load_finished(self, result): 

65 self.config_loaded.emit(result) 

66 

67 def _on_save_finished(self, result): 

68 self.config_saved.emit(result) 

69 

70 

71# Global instance for singleton pattern 

72_global_qt_config_cache = None 

73 

74 

75# Backward compatibility aliases 

76GlobalConfigCache = QtGlobalConfigCache 

77 

78 

79def get_global_config_cache() -> QtGlobalConfigCache: 

80 """Get the global config cache instance (PyQt compatibility).""" 

81 global _global_qt_config_cache 

82 if _global_qt_config_cache is None: 

83 _global_qt_config_cache = QtGlobalConfigCache() 

84 return _global_qt_config_cache 

85 

86 

87# Re-export sync function for startup 

88load_cached_global_config_sync = load_cached_global_config_sync