Coverage for openhcs/utils/enum_factory.py: 27.7%

1""" 

2Dynamic enum creation utilities for OpenHCS. 

3 

4Provides functions for creating enums dynamically from introspection, 

5particularly for visualization colormaps and other runtime-discovered options. 

6 

7Caching: 

8- Colormap enums are cached to avoid expensive napari/matplotlib imports 

9- Cache invalidated on OpenHCS version change or after 30 days 

10- Provides ~20x speedup on subsequent runs 

11""" 

12from enum import Enum 

13from typing import List, Callable, Optional, Dict, Any 

14import logging 

15from openhcs.utils.environment import is_headless_mode 

16 

17logger = logging.getLogger(__name__) 

18 

19 

20# Lazy import cache manager to avoid circular dependencies 

21_cache_manager = None 

22 

23 

24def _get_colormap_cache_manager(): 

25 """Lazy import of cache manager for colormap enums.""" 

26 global _cache_manager 

27 if _cache_manager is None: 

28 try: 

29 from openhcs.core.registry_cache import RegistryCacheManager, CacheConfig 

30 

31 def get_version(): 

32 try: 

33 import openhcs 

34 return openhcs.__version__ 

35 except: 

36 return "unknown" 

37 

38 # Serializer for enum members (just store the dict) 

39 def serialize_enum_members(members: Dict[str, str]) -> Dict[str, Any]: 

40 return {'members': members} 

41 

42 # Deserializer for enum members 

43 def deserialize_enum_members(data: Dict[str, Any]) -> Dict[str, str]: 

44 return data.get('members', {}) 

45 

46 _cache_manager = RegistryCacheManager( 

47 cache_name="colormap_enum", 

48 version_getter=get_version, 

49 serializer=serialize_enum_members, 

50 deserializer=deserialize_enum_members, 

51 config=CacheConfig( 

52 max_age_days=30, # Longer cache for stable enums 

53 check_mtimes=False # No file tracking needed for external libs 

54 ) 

55 ) 

56 except Exception as e: 

57 logger.debug(f"Failed to initialize colormap cache manager: {e}") 

58 _cache_manager = False # Mark as failed to avoid retrying 

59 

60 return _cache_manager if _cache_manager is not False else None 

61 

62 

63def get_available_colormaps() -> List[str]: 

64 """ 

65 Get available colormaps using introspection - napari first, then matplotlib. 

66 

67 In headless/CI contexts, avoid importing viz libs; return minimal stable set. 

68 

69 Returns: 

70 List of available colormap names 

71 """ 

72 if is_headless_mode(): 

73 return ['gray', 'viridis'] 

74 

75 try: 

76 from napari.utils.colormaps import AVAILABLE_COLORMAPS 

77 return list(AVAILABLE_COLORMAPS.keys()) 

78 except ImportError: 

79 pass 

80 

81 try: 

82 import matplotlib.pyplot as plt 

83 return list(plt.colormaps()) 

84 except ImportError: 

85 pass 

86 

87 raise ImportError("Neither napari nor matplotlib colormaps are available. Install napari or matplotlib.") 

88 

89 

90def create_colormap_enum(lazy: bool = False, enable_cache: bool = True) -> Enum: 

91 """ 

92 Create a dynamic enum for available colormaps using pure introspection. 

93 

94 Caching is enabled by default to avoid expensive napari/matplotlib imports 

95 on subsequent runs (~20x speedup). 

96 

97 Args: 

98 lazy: If True, use minimal colormap set without importing napari/matplotlib. 

99 This avoids blocking imports (napari → dask → GPU libs). 

100 enable_cache: If True, use persistent cache for enum members (default: True) 

101 

102 Returns: 

103 Enum class with colormap names as members 

104 

105 Raises: 

106 ValueError: If no colormaps are available or no valid identifiers could be created 

107 """ 

108 # Try to load from cache first (if not lazy mode) 

109 cache_manager = _get_colormap_cache_manager() if enable_cache and not lazy else None 

110 

111 if cache_manager: 111 ↛ 112line 111 didn't jump to line 112 because the condition on line 111 was never true

112 try: 

113 cached_data = cache_manager.load_cache() 

114 if cached_data is not None: 

115 # Cache hit - reconstruct enum from cached members 

116 members = cached_data 

117 logger.debug(f"✅ Loaded {len(members)} colormap enum members from cache") 

118 

119 NapariColormap = Enum('NapariColormap', members) 

120 NapariColormap.__module__ = 'openhcs.core.config' 

121 NapariColormap.__qualname__ = 'NapariColormap' 

122 return NapariColormap 

123 except Exception as e: 

124 logger.debug(f"Cache load failed for colormap enum: {e}") 

125 

126 # Cache miss or disabled - discover colormaps 

127 if lazy: 127 ↛ 131line 127 didn't jump to line 131 because the condition on line 127 was always true

128 # Use minimal set without importing visualization libraries 

129 available_cmaps = ['gray', 'viridis', 'magma', 'inferno', 'plasma', 'cividis'] 

130 else: 

131 available_cmaps = get_available_colormaps() 

132 

133 if not available_cmaps: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 raise ValueError("No colormaps available for enum creation") 

135 

136 members = {} 

137 for cmap_name in available_cmaps: 

138 enum_name = cmap_name.replace(' ', '_').replace('-', '_').replace('.', '_').upper() 

139 if enum_name and enum_name[0].isdigit(): 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true

140 enum_name = f"CMAP_{enum_name}" 

141 if enum_name and enum_name.replace('_', '').replace('CMAP', '').isalnum(): 141 ↛ 137line 141 didn't jump to line 137 because the condition on line 141 was always true

142 members[enum_name] = cmap_name 

143 

144 if not members: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 raise ValueError("No valid colormap identifiers could be created") 

146 

147 # Save to cache if enabled 

148 if cache_manager: 148 ↛ 149line 148 didn't jump to line 149 because the condition on line 148 was never true

149 try: 

150 cache_manager.save_cache(members) 

151 logger.debug(f"💾 Saved {len(members)} colormap enum members to cache") 

152 except Exception as e: 

153 logger.debug(f"Failed to save colormap enum cache: {e}") 

154 

155 NapariColormap = Enum('NapariColormap', members) 

156 

157 # Set proper module and qualname for pickling support 

158 NapariColormap.__module__ = 'openhcs.core.config' 

159 NapariColormap.__qualname__ = 'NapariColormap' 

160 

161 return NapariColormap 

162 

163 

164def create_enum_from_source( 

165 enum_name: str, 

166 source_func: Callable[[], List[str]], 

167 name_transform: Optional[Callable[[str], str]] = None 

168) -> Enum: 

169 """ 

170 Generic factory for creating enums from introspection source functions. 

171 

172 Args: 

173 enum_name: Name for the created enum class 

174 source_func: Function that returns list of string values for enum members 

175 name_transform: Optional function to transform value strings to enum member names 

176 

177 Returns: 

178 Dynamically created Enum class 

179 

180 Example: 

181 >>> def get_luts(): 

182 ... return ['Grays', 'Fire', 'Ice'] 

183 >>> FijiLUT = create_enum_from_source('FijiLUT', get_luts) 

184 """ 

185 values = source_func() 

186 if not values: 

187 raise ValueError(f"No values available for {enum_name} creation") 

188 

189 members = {} 

190 for value in values: 

191 if name_transform: 

192 member_name = name_transform(value) 

193 else: 

194 member_name = value.replace(' ', '_').replace('-', '_').replace('.', '_').upper() 

195 if member_name and member_name[0].isdigit(): 

196 member_name = f"VAL_{member_name}" 

197 

198 if member_name and member_name.replace('_', '').replace('VAL', '').isalnum(): 

199 members[member_name] = value 

200 

201 if not members: 

202 raise ValueError(f"No valid identifiers could be created for {enum_name}") 

203 

204 EnumClass = Enum(enum_name, members) 

205 

206 # Set proper module and qualname for pickling support 

207 EnumClass.__module__ = 'openhcs.core.config' 

208 EnumClass.__qualname__ = enum_name 

209 

210 return EnumClass 

211