Coverage for openhcs/processing/backends/experimental_analysis/format_registry_service.py: 25.9%

1""" 

2Format registry service for automatic discovery and management. 

3 

4This module provides automatic discovery of microscope format registries 

5following OpenHCS generic solution principles. 

6""" 

7 

8from typing import Dict, List, Optional, Type 

9from pathlib import Path 

10 

11from .format_registry import ( 

12 MicroscopeFormatRegistryBase, 

13 FormatDetectionError, 

14 MICROSCOPE_FORMAT_REGISTRIES 

15) 

16 

17 

18class FormatRegistryService: 

19 """ 

20 Service for automatic discovery and access to microscope format registries. 

21  

22 Following OpenHCS generic solution principles, this service automatically 

23 discovers all format registry implementations without hardcoded imports. 

24 """ 

25 

26 _registry_cache: Optional[Dict[str, Type[MicroscopeFormatRegistryBase]]] = None 

27 _instance_cache: Optional[Dict[str, MicroscopeFormatRegistryBase]] = None 

28 

29 @classmethod 

30 def _discover_registries(cls) -> Dict[str, Type[MicroscopeFormatRegistryBase]]: 

31 """ 

32 Get all format registry classes from auto-registered dict. 

33 

34 Returns: 

35 Dictionary mapping format names to registry classes 

36 """ 

37 if cls._registry_cache is not None: 

38 return cls._registry_cache 

39 

40 # Registries auto-discovered on first access to MICROSCOPE_FORMAT_REGISTRIES 

41 cls._registry_cache = MICROSCOPE_FORMAT_REGISTRIES.copy() 

42 return cls._registry_cache 

43 

44 

45 

46 @classmethod 

47 def get_all_format_registries(cls) -> Dict[str, Type[MicroscopeFormatRegistryBase]]: 

48 """ 

49 Get all discovered format registry classes. 

50  

51 Returns: 

52 Dictionary mapping format names to registry classes 

53 """ 

54 return cls._discover_registries() 

55 

56 @classmethod 

57 def get_registry_class_for_format(cls, format_name: str) -> Type[MicroscopeFormatRegistryBase]: 

58 """ 

59 Get registry class for specific format. 

60  

61 Args: 

62 format_name: Name of the microscope format 

63  

64 Returns: 

65 Registry class for the format 

66  

67 Raises: 

68 FormatDetectionError: If format is not supported 

69 """ 

70 registries = cls.get_all_format_registries() 

71 

72 if format_name not in registries: 

73 available_formats = list(registries.keys()) 

74 raise FormatDetectionError( 

75 f"Unsupported format '{format_name}'. Available formats: {available_formats}" 

76 ) 

77 

78 return registries[format_name] 

79 

80 @classmethod 

81 def get_registry_instance_for_format(cls, format_name: str) -> MicroscopeFormatRegistryBase: 

82 """ 

83 Get registry instance for specific format. 

84  

85 Args: 

86 format_name: Name of the microscope format 

87  

88 Returns: 

89 Registry instance for the format 

90  

91 Raises: 

92 FormatDetectionError: If format is not supported 

93 """ 

94 if cls._instance_cache is None: 

95 cls._instance_cache = {} 

96 

97 if format_name not in cls._instance_cache: 

98 registry_class = cls.get_registry_class_for_format(format_name) 

99 cls._instance_cache[format_name] = registry_class() 

100 

101 return cls._instance_cache[format_name] 

102 

103 @classmethod 

104 def detect_format_from_file(cls, file_path: str) -> str: 

105 """ 

106 Automatically detect microscope format from file. 

107  

108 Args: 

109 file_path: Path to the results file 

110  

111 Returns: 

112 Detected format name 

113  

114 Raises: 

115 FormatDetectionError: If format cannot be detected 

116 """ 

117 file_path_obj = Path(file_path) 

118 

119 if not file_path_obj.exists(): 

120 raise FormatDetectionError(f"File not found: {file_path}") 

121 

122 registries = cls.get_all_format_registries() 

123 

124 # Try each registry to see which one can handle the file 

125 for format_name, registry_class in registries.items(): 

126 try: 

127 registry_instance = cls.get_registry_instance_for_format(format_name) 

128 

129 # Check if file extension is supported 

130 if file_path_obj.suffix in registry_instance.SUPPORTED_EXTENSIONS: 

131 # Try to read and process a small sample 

132 try: 

133 raw_df = registry_instance.read_results(file_path) 

134 features = registry_instance.extract_features(raw_df) 

135 

136 # If we can extract features, this format works 

137 if features: 

138 return format_name 

139 

140 except Exception: 

141 # This format doesn't work, try next one 

142 continue 

143 

144 except Exception: 

145 # Skip this registry if it fails 

146 continue 

147 

148 # If no format worked, raise error 

149 available_formats = list(registries.keys()) 

150 raise FormatDetectionError( 

151 f"Could not detect format for file {file_path}. " 

152 f"Available formats: {available_formats}" 

153 ) 

154 

155 @classmethod 

156 def get_supported_formats(cls) -> List[str]: 

157 """ 

158 Get list of all supported format names. 

159  

160 Returns: 

161 List of supported format names 

162 """ 

163 registries = cls.get_all_format_registries() 

164 return list(registries.keys()) 

165 

166 @classmethod 

167 def get_supported_extensions(cls) -> Dict[str, List[str]]: 

168 """ 

169 Get mapping of formats to their supported file extensions. 

170  

171 Returns: 

172 Dictionary mapping format names to supported extensions 

173 """ 

174 registries = cls.get_all_format_registries() 

175 extensions_map = {} 

176 

177 for format_name, registry_class in registries.items(): 

178 registry_instance = cls.get_registry_instance_for_format(format_name) 

179 extensions_map[format_name] = list(registry_instance.SUPPORTED_EXTENSIONS) 

180 

181 return extensions_map 

182 

183 @classmethod 

184 def clear_cache(cls): 

185 """Clear registry and instance caches (useful for testing).""" 

186 cls._registry_cache = None 

187 cls._instance_cache = None