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

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 logging 

9from typing import Dict, List, Optional, Type 

10from .unified_registry import LibraryRegistryBase, FunctionMetadata, LIBRARY_REGISTRIES 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class RegistryService: 

16 """ 

17 Clean service for registry discovery and function metadata access. 

18  

19 Automatically discovers all registry implementations and provides 

20 unified access to their functions with caching. 

21 """ 

22 

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

24 

25 @classmethod 

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

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

28 if cls._metadata_cache is not None: 

29 logger.debug(f"🎯 REGISTRY SERVICE: Using cached metadata ({len(cls._metadata_cache)} functions)") 

30 return cls._metadata_cache 

31 

32 logger.debug("🎯 REGISTRY SERVICE: Discovering functions from all registries...") 

33 all_functions = {} 

34 

35 # Registries auto-discovered on first access to LIBRARY_REGISTRIES 

36 registry_classes = list(LIBRARY_REGISTRIES.values()) 

37 logger.debug(f"🎯 REGISTRY SERVICE: Found {len(registry_classes)} registered library registries") 

38 

39 # Load functions from each registry 

40 for registry_class in registry_classes: 

41 try: 

42 registry_instance = registry_class() 

43 

44 # Skip if library not available 

45 if not registry_instance.is_library_available(): 

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

47 continue 

48 

49 # Get functions from this registry (with caching) 

50 logger.debug(f"🎯 REGISTRY SERVICE: Calling _load_or_discover_functions for {registry_instance.library_name}") 

51 functions = registry_instance._load_or_discover_functions() 

52 logger.debug(f"🎯 REGISTRY SERVICE: 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 

69 

70 @classmethod 

71 def clear_metadata_cache(cls) -> None: 

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

73 cls._metadata_cache = None 

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

75 

76 

77# Backward compatibility aliases 

78FunctionRegistryService = RegistryService 

79get_all_functions_with_metadata = RegistryService.get_all_functions_with_metadata 

80clear_metadata_cache = RegistryService.clear_metadata_cache