Coverage for openhcs/core/metadata_cache.py: 73.1%

1"""Simple metadata cache service for OpenHCS.""" 

2 

3import threading 

4from pathlib import Path 

5from typing import Dict, Optional 

6 

7from openhcs.core.components.validation import convert_enum_by_value 

8from openhcs.constants.constants import AllComponents 

9 

10 

11class MetadataCache: 

12 """Stores component key→name mappings with basic invalidation and thread safety.""" 

13 

14 def __init__(self): 

15 self._cache: Dict['AllComponents', Dict[str, Optional[str]]] = {} 

16 self._metadata_file_mtimes: Dict[Path, float] = {} 

17 self._lock = threading.Lock() 

18 

19 def cache_metadata(self, microscope_handler, plate_path: Path, component_keys_cache: Dict) -> None: 

20 """Cache all metadata from metadata handler.""" 

21 with self._lock: 

22 # Parse all metadata once 

23 metadata = microscope_handler.metadata_handler.parse_metadata(plate_path) 

24 

25 # Initialize all components with keys mapped to None 

26 for component in AllComponents: 

27 component_keys = component_keys_cache.get(component, []) 

28 self._cache[component] = {key: None for key in component_keys} 

29 

30 # Update with actual metadata where available 

31 for component_name, mapping in metadata.items(): 

32 component = AllComponents(component_name) 

33 if component in self._cache: 33 ↛ 41line 33 didn't jump to line 41 because the condition on line 33 was always true

34 combined_cache = self._cache[component].copy() 

35 for metadata_key in mapping.keys(): 

36 if metadata_key not in combined_cache: 36 ↛ 37line 36 didn't jump to line 37 because the condition on line 36 was never true

37 combined_cache[metadata_key] = None 

38 combined_cache.update(mapping) 

39 self._cache[component] = combined_cache 

40 else: 

41 self._cache[component] = mapping 

42 

43 # Store metadata file mtime for invalidation 

44 metadata_file = microscope_handler.metadata_handler.find_metadata_file(plate_path) 

45 if metadata_file and metadata_file.exists(): 45 ↛ exitline 45 didn't jump to the function exit

46 self._metadata_file_mtimes[metadata_file] = metadata_file.stat().st_mtime 

47 

48 def get_component_metadata(self, component, key: str) -> Optional[str]: 

49 """Get metadata display name for a component key. Accepts GroupBy or VariableComponents.""" 

50 with self._lock: 

51 if not self._is_cache_valid(): 

52 self._cache.clear() 

53 return None 

54 

55 # Convert GroupBy to AllComponents using OpenHCS generic utility 

56 component = convert_enum_by_value(component, AllComponents) or component 

57 

58 return self._cache.get(component, {}).get(key) 

59 

60 def get_cached_metadata(self, component: 'AllComponents') -> Optional[Dict[str, Optional[str]]]: 

61 """Get all cached metadata for a component.""" 

62 with self._lock: 

63 if not self._is_cache_valid(): 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true

64 self._cache.clear() 

65 return None 

66 return self._cache.get(component) 

67 

68 def clear_cache(self) -> None: 

69 """Clear cached metadata.""" 

70 with self._lock: 

71 self._cache.clear() 

72 self._metadata_file_mtimes.clear() 

73 

74 def _is_cache_valid(self) -> bool: 

75 """Check if cache is valid by comparing file mtimes.""" 

76 for metadata_file, cached_mtime in self._metadata_file_mtimes.items(): 

77 if not metadata_file.exists() or metadata_file.stat().st_mtime != cached_mtime: 77 ↛ 78line 77 didn't jump to line 78 because the condition on line 77 was never true

78 return False 

79 return True 

80 

81 

82# Global cache instance 

83_global_metadata_cache: Optional[MetadataCache] = None 

84 

85 

86def get_metadata_cache() -> MetadataCache: 

87 """Get global metadata cache instance.""" 

88 global _global_metadata_cache 

89 if _global_metadata_cache is None: 

90 _global_metadata_cache = MetadataCache() 

91 return _global_metadata_cache