Coverage for openhcs/core/components/metaprogramming.py: 0.0%
101 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +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 ABC, abstractmethod
11from typing import Any, Dict, Type, TypeVar, Generic, Optional, Set, Callable, Union
12from enum import Enum
13import inspect
15logger = logging.getLogger(__name__)
17T = TypeVar('T', bound=Enum)
20class MethodSignature:
21 """Represents a dynamically generated method signature."""
23 def __init__(self, name: str, return_type: Type = Any, **kwargs):
24 self.name = name
25 self.return_type = return_type
26 self.parameters = kwargs
28 def __repr__(self):
29 params = ", ".join(f"{k}: {v.__name__ if hasattr(v, '__name__') else v}"
30 for k, v in self.parameters.items())
31 return f"{self.name}({params}) -> {self.return_type.__name__ if hasattr(self.return_type, '__name__') else self.return_type}"
34class ComponentMethodRegistry:
35 """Registry for tracking dynamically generated methods."""
37 def __init__(self):
38 self._methods: Dict[str, Dict[str, MethodSignature]] = {}
40 def register_method(self, interface_name: str, method: MethodSignature):
41 """Register a method for an interface."""
42 if interface_name not in self._methods:
43 self._methods[interface_name] = {}
44 self._methods[interface_name][method.name] = method
45 logger.debug(f"Registered method {method.name} for interface {interface_name}")
47 def get_methods(self, interface_name: str) -> Dict[str, MethodSignature]:
48 """Get all methods for an interface."""
49 return self._methods.get(interface_name, {})
51 def has_method(self, interface_name: str, method_name: str) -> bool:
52 """Check if an interface has a specific method."""
53 return (interface_name in self._methods and
54 method_name in self._methods[interface_name])
57# Global method registry
58_method_registry = ComponentMethodRegistry()
61class DynamicInterfaceMeta(type):
62 """
63 Metaclass that dynamically generates interface methods based on component enums.
65 This metaclass inspects the component enum during class creation and generates
66 abstract methods for each component, enabling truly generic component processing.
67 """
69 def __new__(mcs, name, bases, namespace, component_enum=None, method_patterns=None, **kwargs):
70 """
71 Create a new interface class with dynamically generated methods.
73 Args:
74 name: Class name
75 bases: Base classes
76 namespace: Class namespace
77 component_enum: Enum class to generate methods from
78 method_patterns: List of method patterns to generate
79 **kwargs: Additional arguments
80 """
81 # Default method patterns if not specified
82 if method_patterns is None:
83 method_patterns = ['process', 'validate', 'get_keys']
85 # Generate methods if component_enum is provided
86 if component_enum is not None:
87 logger.info(f"Generating dynamic interface {name} for enum {component_enum.__name__}")
88 mcs._generate_methods(namespace, component_enum, method_patterns, name)
90 # Create the class
91 cls = super().__new__(mcs, name, bases, namespace)
93 # Register the interface
94 if component_enum is not None:
95 cls._component_enum = component_enum
96 cls._method_patterns = method_patterns
97 logger.info(f"Created dynamic interface {name} with {len(_method_registry.get_methods(name))} methods")
99 return cls
101 @staticmethod
102 def _generate_methods(namespace: Dict[str, Any], component_enum: Type[Enum],
103 method_patterns: list, interface_name: str):
104 """Generate abstract methods for each component and pattern combination."""
105 for component in component_enum:
106 component_name = component.value
108 for pattern in method_patterns:
109 method_name = f"{pattern}_{component_name}"
111 # Create method signature
112 signature = MethodSignature(
113 name=method_name,
114 return_type=Any,
115 context=Any,
116 **{f"{component_name}_value": str}
117 )
119 # Register the method
120 _method_registry.register_method(interface_name, signature)
122 # Create abstract method
123 def create_abstract_method(method_name=method_name):
124 @abstractmethod
125 def abstract_method(self, context: Any, **kwargs) -> Any:
126 """Dynamically generated abstract method."""
127 raise NotImplementedError(f"Method {method_name} must be implemented")
129 abstract_method.__name__ = method_name
130 abstract_method.__qualname__ = f"{interface_name}.{method_name}"
131 return abstract_method
133 # Add method to namespace
134 namespace[method_name] = create_abstract_method()
135 logger.debug(f"Generated abstract method {method_name} for {interface_name}")
138class ComponentProcessorInterface(metaclass=DynamicInterfaceMeta):
139 """
140 Base interface for component processors with dynamically generated methods.
142 This class uses the DynamicInterfaceMeta metaclass to automatically generate
143 abstract methods based on the component enum, providing a truly generic
144 interface that adapts to any component configuration.
145 """
147 def __init__(self, component_enum: Type[T]):
148 """
149 Initialize the processor interface.
151 Args:
152 component_enum: The component enum to process
153 """
154 self.component_enum = component_enum
155 self._validate_implementation()
157 def _validate_implementation(self):
158 """Validate that all required methods are implemented."""
159 interface_name = self.__class__.__name__
160 required_methods = _method_registry.get_methods(interface_name)
162 for method_name, signature in required_methods.items():
163 if not hasattr(self, method_name):
164 raise NotImplementedError(
165 f"Class {self.__class__.__name__} must implement method {method_name}"
166 )
168 method = getattr(self, method_name)
169 if not callable(method):
170 raise TypeError(
171 f"Attribute {method_name} in {self.__class__.__name__} must be callable"
172 )
174 logger.debug(f"Validated implementation of {interface_name} with {len(required_methods)} methods")
176 def get_available_methods(self) -> Dict[str, MethodSignature]:
177 """Get all available methods for this interface."""
178 return _method_registry.get_methods(self.__class__.__name__)
180 def has_method_for_component(self, component: T, pattern: str) -> bool:
181 """Check if a method exists for a specific component and pattern."""
182 method_name = f"{pattern}_{component.value}"
183 return _method_registry.has_method(self.__class__.__name__, method_name)
185 def call_component_method(self, component: T, pattern: str, context: Any, **kwargs) -> Any:
186 """Dynamically call a component-specific method."""
187 method_name = f"{pattern}_{component.value}"
189 if not hasattr(self, method_name):
190 raise AttributeError(
191 f"Method {method_name} not found in {self.__class__.__name__}"
192 )
194 method = getattr(self, method_name)
195 return method(context, **kwargs)
198class InterfaceGenerator:
199 """
200 Factory for creating component-specific interfaces dynamically.
202 This class provides a high-level API for generating interfaces based on
203 component enums, with caching and type safety features.
204 """
206 def __init__(self):
207 self._interface_cache: Dict[str, Type] = {}
209 def create_interface(self,
210 component_enum: Type[T],
211 interface_name: Optional[str] = None,
212 method_patterns: Optional[list] = None,
213 base_classes: Optional[tuple] = None) -> Type[ComponentProcessorInterface]:
214 """
215 Create a component-specific interface class.
217 Args:
218 component_enum: The component enum to generate interface for
219 interface_name: Optional custom interface name
220 method_patterns: Optional custom method patterns
221 base_classes: Optional additional base classes
223 Returns:
224 Dynamically generated interface class
225 """
226 # Generate interface name if not provided
227 if interface_name is None:
228 interface_name = f"{component_enum.__name__}ProcessorInterface"
230 # Check cache
231 cache_key = f"{interface_name}_{id(component_enum)}"
232 if cache_key in self._interface_cache:
233 logger.debug(f"Returning cached interface {interface_name}")
234 return self._interface_cache[cache_key]
236 # Set default base classes
237 if base_classes is None:
238 base_classes = (ComponentProcessorInterface,)
240 # Create the interface class dynamically
241 interface_class = DynamicInterfaceMeta(
242 interface_name,
243 base_classes,
244 {},
245 component_enum=component_enum,
246 method_patterns=method_patterns
247 )
249 # Cache the interface
250 self._interface_cache[cache_key] = interface_class
252 logger.info(f"Created interface {interface_name} for {component_enum.__name__}")
253 return interface_class
255 def get_cached_interface(self, interface_name: str) -> Optional[Type]:
256 """Get a cached interface by name."""
257 for key, interface in self._interface_cache.items():
258 if key.startswith(interface_name):
259 return interface
260 return None
262 def clear_cache(self):
263 """Clear the interface cache."""
264 self._interface_cache.clear()
265 logger.debug("Cleared interface cache")
268# Global interface generator instance
269interface_generator = InterfaceGenerator()