Coverage for openhcs/textual_tui/services/function_registry_service.py: 0.0%
78 statements
« prev ^ index » next coverage.py v7.10.3, created at 2025-08-14 05:57 +0000
1"""
2Function Registry Service - Enhanced function discovery and metadata.
4This service extends existing func_registry functionality with enhanced metadata
5extraction and UI-specific formatting while avoiding import collisions.
6"""
8from typing import Dict, List, Tuple, Callable, Any, Optional
9import logging
11# Import from func_registry to avoid circular imports
12# The FUNC_REGISTRY is auto-initialized on import (following storage_registry pattern)
13from openhcs.processing.func_registry import FUNC_REGISTRY, get_function_info, get_functions_by_memory_type
15logger = logging.getLogger(__name__)
18class FunctionRegistryService:
19 """
20 Stateless service for function registry integration and metadata extraction.
22 Extends existing FUNC_REGISTRY functionality with enhanced metadata
23 and UI-specific formatting.
24 """
26 @staticmethod
27 def get_functions_by_backend() -> Dict[str, List[Tuple[Callable, str]]]:
28 """
29 Group FUNC_REGISTRY functions by backend with memory type display names.
31 EXACT backend grouping logic - preserves current behavior.
33 Returns:
34 Dict mapping backend names to lists of (function, display_name) tuples
35 """
36 functions_by_backend = {}
38 # CORRECT: Iterate over memory types, then functions within each type
39 for memory_type in FUNC_REGISTRY:
40 for func in FUNC_REGISTRY[memory_type]:
41 try:
42 # Get basic function info using existing registry
43 func_info = get_function_info(func)
45 backend = func_info.get('backend', 'unknown')
46 input_type = func_info.get('input_memory_type', 'unknown')
47 output_type = func_info.get('output_memory_type', 'unknown')
49 # Clean display format - show only function name, backend is shown separately
50 display_name = func.__name__
52 if backend not in functions_by_backend:
53 functions_by_backend[backend] = []
55 functions_by_backend[backend].append((func, display_name))
57 except Exception as e:
58 logger.warning(f"Failed to get info for function {func}: {e}")
59 # Add to unknown backend as fallback
60 if 'unknown' not in functions_by_backend:
61 functions_by_backend['unknown'] = []
62 functions_by_backend['unknown'].append((func, func.__name__))
64 return functions_by_backend
66 @staticmethod
67 def get_enhanced_function_metadata(func: Callable) -> Dict[str, Any]:
68 """
69 Enhanced version of get_function_info with validation and special inputs/outputs.
71 RENAMED to avoid import collision with existing get_function_info.
73 Args:
74 func: Function to analyze
76 Returns:
77 Enhanced metadata dict with same structure as get_function_info plus validation
78 """
79 try:
80 # Start with existing function info
81 base_info = get_function_info(func)
83 # Enhance with additional metadata
84 enhanced_info = {
85 'name': base_info.get('name', func.__name__),
86 'backend': base_info.get('backend', 'unknown'),
87 'input_memory_type': base_info.get('input_memory_type', 'unknown'),
88 'output_memory_type': base_info.get('output_memory_type', 'unknown'),
89 'special_inputs': base_info.get('special_inputs', []),
90 'special_outputs': base_info.get('special_outputs', []),
91 }
93 # Add validation status
94 enhanced_info['is_valid'] = True
95 enhanced_info['validation_errors'] = []
97 # Basic validation checks
98 if enhanced_info['backend'] == 'unknown':
99 enhanced_info['validation_errors'].append("Unknown backend")
101 if enhanced_info['input_memory_type'] == 'unknown':
102 enhanced_info['validation_errors'].append("Unknown input memory type")
104 if enhanced_info['output_memory_type'] == 'unknown':
105 enhanced_info['validation_errors'].append("Unknown output memory type")
107 if enhanced_info['validation_errors']:
108 enhanced_info['is_valid'] = False
110 return enhanced_info
112 except Exception as e:
113 logger.error(f"Failed to get enhanced metadata for {func}: {e}")
114 # Return minimal fallback info
115 return {
116 'name': getattr(func, '__name__', 'unknown'),
117 'backend': 'unknown',
118 'input_memory_type': 'unknown',
119 'output_memory_type': 'unknown',
120 'special_inputs': [],
121 'special_outputs': [],
122 'is_valid': False,
123 'validation_errors': [f"Metadata extraction failed: {e}"]
124 }
126 @staticmethod
127 def create_dropdown_options(functions_by_backend: Dict[str, List[Tuple[Callable, str]]]) -> List[Tuple[str, List[Tuple[Callable, str]]]]:
128 """
129 Format function groups for GroupedDropdown component.
131 Args:
132 functions_by_backend: Output from get_functions_by_backend()
134 Returns:
135 List of (group_name, options) tuples for GroupedDropdown
136 """
137 dropdown_options = []
139 # Sort backends for consistent ordering
140 sorted_backends = sorted(functions_by_backend.keys())
142 for backend in sorted_backends:
143 functions = functions_by_backend[backend]
144 # Sort functions by display name for consistent ordering
145 sorted_functions = sorted(functions, key=lambda x: x[1])
146 dropdown_options.append((backend, sorted_functions))
148 return dropdown_options
150 @staticmethod
151 def find_default_function() -> Optional[Callable]:
152 """
153 Get first available function for new items.
155 Returns:
156 First function from FUNC_REGISTRY, or None if empty
157 """
158 # CORRECT: Iterate through memory types to find first function
159 for memory_type in FUNC_REGISTRY:
160 if FUNC_REGISTRY[memory_type]: # Check if list is not empty
161 return FUNC_REGISTRY[memory_type][0]
162 return None
164 @staticmethod
165 def get_functions_by_memory_type_wrapper(memory_type: str) -> List[Callable]:
166 """
167 Wrapper around existing get_functions_by_memory_type for consistency.
169 Args:
170 memory_type: Memory type to filter by
172 Returns:
173 List of functions matching the memory type
174 """
175 try:
176 return get_functions_by_memory_type(memory_type)
177 except Exception as e:
178 logger.error(f"Failed to get functions by memory type {memory_type}: {e}")
179 return []
181 @staticmethod
182 def validate_function_in_registry(func: Callable) -> bool:
183 """
184 Check if function exists in FUNC_REGISTRY.
186 Args:
187 func: Function to check
189 Returns:
190 True if function is in registry, False otherwise
191 """
192 # CORRECT: Check if function exists in any memory type list
193 for memory_type in FUNC_REGISTRY:
194 if func in FUNC_REGISTRY[memory_type]:
195 return True
196 return False
198 @staticmethod
199 def get_function_display_name(func: Callable) -> str:
200 """
201 Get formatted display name for function.
203 Args:
204 func: Function to format
206 Returns:
207 Formatted display name with memory types
208 """
209 try:
210 # Clean display format - just return function name
211 # Backend information is displayed separately in the UI
212 return func.__name__
213 except Exception:
214 return func.__name__