Coverage for openhcs/processing/backends/lib_registry/registry_service.py: 89.3%

1""" 

2Registry Service - Clean function discovery and metadata access. 

3 

4Provides unified access to all registry implementations with automatic discovery. 

5Follows OpenHCS generic solution principle - automatically adapts to new registries. 

6""" 

7 

8import pkgutil 

9import importlib 

10import inspect 

11import logging 

12from typing import Dict, List, Optional 

13from .unified_registry import LibraryRegistryBase, FunctionMetadata 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18class RegistryService: 

19 """ 

20 Clean service for registry discovery and function metadata access. 

21  

22 Automatically discovers all registry implementations and provides 

23 unified access to their functions with caching. 

24 """ 

25 

26 _metadata_cache: Optional[Dict[str, FunctionMetadata]] = None 

27 

28 @classmethod 

29 def get_all_functions_with_metadata(cls) -> Dict[str, FunctionMetadata]: 

30 """Get unified metadata for all functions from all registries.""" 

31 if cls._metadata_cache is not None: 

32 return cls._metadata_cache 

33 

34 logger.info("Discovering functions from all registries...") 

35 all_functions = {} 

36 

37 # Discover all registry classes automatically 

38 registry_classes = cls._discover_registries() 

39 

40 # Load functions from each registry 

41 for registry_class in registry_classes: 

42 try: 

43 registry_instance = registry_class() 

44 

45 # Skip if library not available 

46 if not registry_instance.is_library_available(): 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 logger.warning(f"Library {registry_instance.library_name} not available, skipping") 

48 continue 

49 

50 # Get functions from this registry (with caching) 

51 functions = registry_instance._load_or_discover_functions() 

52 logger.info(f"Retrieved {len(functions)} {registry_instance.library_name} functions") 

53 

54 # Use composite keys to prevent function name collisions between backends 

55 # Format: "backend:function_name" (e.g., "torch:stack_percentile_normalize") 

56 for func_name, metadata in functions.items(): 

57 composite_key = f"{registry_instance.library_name}:{func_name}" 

58 all_functions[composite_key] = metadata 

59 

60 except Exception as e: 

61 logger.warning(f"Failed to load registry {registry_class.__name__}: {e}") 

62 continue 

63 

64 logger.info(f"Total functions discovered: {len(all_functions)}") 

65 cls._metadata_cache = all_functions 

66 return all_functions 

67 

68 @classmethod 

69 def _discover_registries(cls) -> List[type]: 

70 """Automatically discover all registry implementations.""" 

71 registry_classes = [] 

72 

73 # Scan lib_registry package for modules 

74 import openhcs.processing.backends.lib_registry as registry_package 

75 registry_path = registry_package.__path__ 

76 registry_prefix = registry_package.__name__ + "." 

77 

78 for importer, module_name, ispkg in pkgutil.iter_modules(registry_path, registry_prefix): 

79 # Skip unified_registry (contains base classes) 

80 if module_name.endswith('.unified_registry'): 

81 continue 

82 

83 try: 

84 module = importlib.import_module(module_name) 

85 

86 # Find registry classes in module 

87 for name, obj in inspect.getmembers(module, inspect.isclass): 

88 if (issubclass(obj, LibraryRegistryBase) and 

89 obj is not LibraryRegistryBase and 

90 obj.__module__ == module_name): 

91 

92 registry_classes.append(obj) 

93 

94 except Exception as e: 

95 logger.warning(f"Failed to load registry module {module_name}: {e}") 

96 continue 

97 

98 logger.info(f"Discovered {len(registry_classes)} registry implementations") 

99 return registry_classes 

100 

101 @classmethod 

102 def clear_metadata_cache(cls) -> None: 

103 """Clear cached metadata to force re-discovery.""" 

104 cls._metadata_cache = None 

105 logger.info("Registry metadata cache cleared") 

106 

107 

108# Backward compatibility aliases 

109FunctionRegistryService = RegistryService 

110get_all_functions_with_metadata = RegistryService.get_all_functions_with_metadata 

111clear_metadata_cache = RegistryService.clear_metadata_cache