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 _sync_load_config, 

11 _sync_save_config 

12) 

13from openhcs.core.config import GlobalPipelineConfig 

14from pathlib import Path 

15 

16 

17class ConfigCacheWorker(QRunnable): 

18 """Qt worker for cache operations.""" 

19 

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

21 super().__init__() 

22 self.operation = operation 

23 self.cache_file = cache_file 

24 self.config = config 

25 self.callback = callback 

26 

27 def run(self): 

28 if self.operation == 'load': 

29 result = _sync_load_config(self.cache_file) 

30 if self.callback: 

31 self.callback(result) 

32 elif self.operation == 'save': 

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

34 if self.callback: 

35 self.callback(result) 

36 

37 

38class QtGlobalConfigCache(QObject): 

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

40 

41 config_loaded = pyqtSignal(object) # GlobalPipelineConfig or None 

42 config_saved = pyqtSignal(bool) # Success/failure 

43 

44 def __init__(self, cache_file=None): 

45 super().__init__() 

46 if cache_file is None: 

47 from openhcs.core.xdg_paths import get_config_file_path 

48 cache_file = get_config_file_path("global_config.config") 

49 self.cache_file = cache_file 

50 self.thread_pool = QThreadPool() 

51 

52 def load_cached_config_async(self): 

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

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

55 self.thread_pool.start(worker) 

56 

57 def save_config_to_cache_async(self, config: GlobalPipelineConfig): 

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

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

60 self.thread_pool.start(worker) 

61 

62 def _on_load_finished(self, result): 

63 self.config_loaded.emit(result) 

64 

65 def _on_save_finished(self, result): 

66 self.config_saved.emit(result) 

67 

68 

69# Global instance for singleton pattern 

70_global_qt_config_cache = None 

71 

72 

73# Backward compatibility aliases 

74GlobalConfigCache = QtGlobalConfigCache 

75 

76 

77def get_global_config_cache() -> QtGlobalConfigCache: 

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

79 global _global_qt_config_cache 

80 if _global_qt_config_cache is None: 

81 _global_qt_config_cache = QtGlobalConfigCache() 

82 return _global_qt_config_cache 

83 

84 

85# Re-export sync function for startup 

86load_cached_global_config_sync = load_cached_global_config_sync