Coverage for openhcs/core/components/metaprogramming.py: 0.0%
100 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""
2Metaprogramming system for dynamic interface generation based on component enums.
4This module provides a metaprogramming framework that dynamically generates interface
5classes based on the contents of component enums, enabling truly generic component
6processing without hardcoded method names or component assumptions.
7"""
9import logging
10from abc import abstractmethod
11from typing import Any, Dict, Type, TypeVar, Optional
12from enum import Enum
14logger = logging.getLogger(__name__)
16T = TypeVar('T', bound=Enum)
19class MethodSignature:
20 """Represents a dynamically generated method signature."""
22 def __init__(self, name: str, return_type: Type = Any, **kwargs):
23 self.name = name
24 self.return_type = return_type
25 self.parameters = kwargs
27 def __repr__(self):
28 params = ", ".join(f"{k}: {v.__name__ if hasattr(v, '__name__') else v}"
29 for k, v in self.parameters.items())
30 return f"{self.name}({params}) -> {self.return_type.__name__ if hasattr(self.return_type, '__name__') else self.return_type}"
33class ComponentMethodRegistry:
34 """Registry for tracking dynamically generated methods."""
36 def __init__(self):
37 self._methods: Dict[str, Dict[str, MethodSignature]] = {}
39 def register_method(self, interface_name: str, method: MethodSignature):
40 """Register a method for an interface."""
41 if interface_name not in self._methods:
42 self._methods[interface_name] = {}
43 self._methods[interface_name][method.name] = method
44 logger.debug(f"Registered method {method.name} for interface {interface_name}")
46 def get_methods(self, interface_name: str) -> Dict[str, MethodSignature]:
47 """Get all methods for an interface."""
48 return self._methods.get(interface_name, {})
50 def has_method(self, interface_name: str, method_name: str) -> bool:
51 """Check if an interface has a specific method."""
52 return (interface_name in self._methods and
53 method_name in self._methods[interface_name])
56# Global method registry
57_method_registry = ComponentMethodRegistry()
60class DynamicInterfaceMeta(type):
61 """
62 Metaclass that dynamically generates interface methods based on component enums.
64 This metaclass inspects the component enum during class creation and generates
65 abstract methods for each component, enabling truly generic component processing.
66 """
68 def __new__(mcs, name, bases, namespace, component_enum=None, method_patterns=None, **kwargs):
69 """
70 Create a new interface class with dynamically generated methods.
72 Args:
73 name: Class name
74 bases: Base classes
75 namespace: Class namespace
76 component_enum: Enum class to generate methods from
77 method_patterns: List of method patterns to generate
78 **kwargs: Additional arguments
79 """
80 # Default method patterns if not specified
81 if method_patterns is None:
82 method_patterns = ['process', 'validate', 'get_keys']
84 # Generate methods if component_enum is provided
85 if component_enum is not None:
86 logger.info(f"Generating dynamic interface {name} for enum {component_enum.__name__}")
87 mcs._generate_methods(namespace, component_enum, method_patterns, name)
89 # Create the class
90 cls = super().__new__(mcs, name, bases, namespace)
92 # Register the interface
93 if component_enum is not None:
94 cls._component_enum = component_enum
95 cls._method_patterns = method_patterns
96 logger.info(f"Created dynamic interface {name} with {len(_method_registry.get_methods(name))} methods")
98 return cls
100 @staticmethod
101 def _generate_methods(namespace: Dict[str, Any], component_enum: Type[Enum],
102 method_patterns: list, interface_name: str):
103 """Generate abstract methods for each component and pattern combination."""
104 for component in component_enum:
105 component_name = component.value
107 for pattern in method_patterns:
108 method_name = f"{pattern}_{component_name}"
110 # Create method signature
111 signature = MethodSignature(
112 name=method_name,
113 return_type=Any,
114 context=Any,
115 **{f"{component_name}_value": str}
116 )
118 # Register the method
119 _method_registry.register_method(interface_name, signature)
121 # Create abstract method
122 def create_abstract_method(method_name=method_name):
123 @abstractmethod
124 def abstract_method(self, context: Any, **kwargs) -> Any:
125 """Dynamically generated abstract method."""
126 raise NotImplementedError(f"Method {method_name} must be implemented")
128 abstract_method.__name__ = method_name
129 abstract_method.__qualname__ = f"{interface_name}.{method_name}"
130 return abstract_method
132 # Add method to namespace
133 namespace[method_name] = create_abstract_method()
134 logger.debug(f"Generated abstract method {method_name} for {interface_name}")
137class ComponentProcessorInterface(metaclass=DynamicInterfaceMeta):
138 """
139 Base interface for component processors with dynamically generated methods.
141 This class uses the DynamicInterfaceMeta metaclass to automatically generate
142 abstract methods based on the component enum, providing a truly generic
143 interface that adapts to any component configuration.
144 """
146 def __init__(self, component_enum: Type[T]):
147 """
148 Initialize the processor interface.
150 Args:
151 component_enum: The component enum to process
152 """
153 self.component_enum = component_enum
154 self._validate_implementation()
156 def _validate_implementation(self):
157 """Validate that all required methods are implemented."""
158 interface_name = self.__class__.__name__
159 required_methods = _method_registry.get_methods(interface_name)
161 for method_name, signature in required_methods.items():
162 if not hasattr(self, method_name):
163 raise NotImplementedError(
164 f"Class {self.__class__.__name__} must implement method {method_name}"
165 )
167 method = getattr(self, method_name)
168 if not callable(method):
169 raise TypeError(
170 f"Attribute {method_name} in {self.__class__.__name__} must be callable"
171 )
173 logger.debug(f"Validated implementation of {interface_name} with {len(required_methods)} methods")
175 def get_available_methods(self) -> Dict[str, MethodSignature]:
176 """Get all available methods for this interface."""
177 return _method_registry.get_methods(self.__class__.__name__)
179 def has_method_for_component(self, component: T, pattern: str) -> bool:
180 """Check if a method exists for a specific component and pattern."""
181 method_name = f"{pattern}_{component.value}"
182 return _method_registry.has_method(self.__class__.__name__, method_name)
184 def call_component_method(self, component: T, pattern: str, context: Any, **kwargs) -> Any:
185 """Dynamically call a component-specific method."""
186 method_name = f"{pattern}_{component.value}"
188 if not hasattr(self, method_name):
189 raise AttributeError(
190 f"Method {method_name} not found in {self.__class__.__name__}"
191 )
193 method = getattr(self, method_name)
194 return method(context, **kwargs)
197class InterfaceGenerator:
198 """
199 Factory for creating component-specific interfaces dynamically.
201 This class provides a high-level API for generating interfaces based on
202 component enums, with caching and type safety features.
203 """
205 def __init__(self):
206 self._interface_cache: Dict[str, Type] = {}
208 def create_interface(self,
209 component_enum: Type[T],
210 interface_name: Optional[str] = None,
211 method_patterns: Optional[list] = None,
212 base_classes: Optional[tuple] = None) -> Type[ComponentProcessorInterface]:
213 """
214 Create a component-specific interface class.
216 Args:
217 component_enum: The component enum to generate interface for
218 interface_name: Optional custom interface name
219 method_patterns: Optional custom method patterns
220 base_classes: Optional additional base classes
222 Returns:
223 Dynamically generated interface class
224 """
225 # Generate interface name if not provided
226 if interface_name is None:
227 interface_name = f"{component_enum.__name__}ProcessorInterface"
229 # Check cache
230 cache_key = f"{interface_name}_{id(component_enum)}"
231 if cache_key in self._interface_cache:
232 logger.debug(f"Returning cached interface {interface_name}")
233 return self._interface_cache[cache_key]
235 # Set default base classes
236 if base_classes is None:
237 base_classes = (ComponentProcessorInterface,)
239 # Create the interface class dynamically
240 interface_class = DynamicInterfaceMeta(
241 interface_name,
242 base_classes,
243 {},
244 component_enum=component_enum,
245 method_patterns=method_patterns
246 )
248 # Cache the interface
249 self._interface_cache[cache_key] = interface_class
251 logger.info(f"Created interface {interface_name} for {component_enum.__name__}")
252 return interface_class
254 def get_cached_interface(self, interface_name: str) -> Optional[Type]:
255 """Get a cached interface by name."""
256 for key, interface in self._interface_cache.items():
257 if key.startswith(interface_name):
258 return interface
259 return None
261 def clear_cache(self):
262 """Clear the interface cache."""
263 self._interface_cache.clear()
264 logger.debug("Cleared interface cache")
267# Global interface generator instance
268interface_generator = InterfaceGenerator()