Coverage for openhcs/core/components/framework.py: 63.5%
54 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +0000
1"""
2Core framework for generic component configuration.
4This module provides the foundational classes for configuring any enum as components
5with configurable multiprocessing axis and validation constraints.
6"""
8from dataclasses import dataclass
9from typing import Generic, TypeVar, Set, List, Optional, Type
10from enum import Enum
12T = TypeVar('T', bound=Enum)
15@dataclass(frozen=True)
16class ComponentConfiguration(Generic[T]):
17 """
18 Generic configuration for any enum-based component system.
20 This class encapsulates the configuration for a component system where:
21 - Components are defined by an enum
22 - One component serves as the multiprocessing axis
23 - Default variable components and group_by are specified
24 - Generic constraint validation is enforced: group_by ∉ variable_components
25 """
27 all_components: Set[T]
28 multiprocessing_axis: T
29 default_variable: List[T]
30 default_group_by: Optional[T]
32 def __post_init__(self):
33 """Validate configuration constraints."""
34 # Ensure multiprocessing_axis is in all_components
35 if self.multiprocessing_axis not in self.all_components: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true
36 raise ValueError(
37 f"multiprocessing_axis {self.multiprocessing_axis.value} "
38 f"must be in all_components"
39 )
41 # Ensure default_variable components are in all_components
42 for component in self.default_variable:
43 if component not in self.all_components: 43 ↛ 44line 43 didn't jump to line 44 because the condition on line 43 was never true
44 raise ValueError(
45 f"default_variable component {component.value} "
46 f"must be in all_components"
47 )
49 # Ensure default_group_by is in all_components (if specified)
50 if self.default_group_by and self.default_group_by not in self.all_components: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true
51 raise ValueError(
52 f"default_group_by {self.default_group_by.value} "
53 f"must be in all_components"
54 )
56 # Validate default combination
57 self.validate_combination(self.default_variable, self.default_group_by)
59 def validate_combination(self, variable: List[T], group_by: Optional[T]) -> None:
60 """
61 Validate that group_by is not in variable_components.
63 This enforces the core constraint: group_by ∉ variable_components
65 Args:
66 variable: List of variable components
67 group_by: Optional group_by component
69 Raises:
70 ValueError: If group_by is in variable_components
71 """
72 if group_by and group_by in variable: 72 ↛ 73line 72 didn't jump to line 73 because the condition on line 72 was never true
73 raise ValueError(
74 f"group_by {group_by.value} cannot be in variable_components "
75 f"{[v.value for v in variable]}"
76 )
78 def get_remaining_components(self) -> Set[T]:
79 """
80 Get components available for variable_components and group_by selection.
82 Returns all components except the multiprocessing_axis.
84 Returns:
85 Set of components available for variable/group_by selection
86 """
87 return self.all_components - {self.multiprocessing_axis}
89 def get_available_variable_components(self) -> List[T]:
90 """
91 Get all components that can be used as variable_components.
93 Returns:
94 List of components available as variable components
95 """
96 return list(self.get_remaining_components())
98 def get_available_group_by_components(self, exclude_variable: Optional[List[T]] = None) -> List[T]:
99 """
100 Get components that can be used as group_by, excluding variable components.
102 Args:
103 exclude_variable: Variable components to exclude from group_by options
105 Returns:
106 List of components available as group_by
107 """
108 remaining = self.get_remaining_components()
109 if exclude_variable:
110 remaining = remaining - set(exclude_variable)
111 return list(remaining)
114class ComponentConfigurationFactory:
115 """Factory for creating ComponentConfiguration instances."""
117 @staticmethod
118 def create_configuration(
119 component_enum: Type[T],
120 multiprocessing_axis: T,
121 default_variable: Optional[List[T]] = None,
122 default_group_by: Optional[T] = None
123 ) -> ComponentConfiguration[T]:
124 """
125 Create a ComponentConfiguration for the given enum with dynamic component resolution.
127 When multiprocessing_axis is specified, the remaining components are automatically
128 available for variable_components and group_by selection.
130 Args:
131 component_enum: The enum class defining all components
132 multiprocessing_axis: Component to use for multiprocessing
133 default_variable: Default variable components (auto-resolved if None)
134 default_group_by: Default group_by component (auto-resolved if None)
136 Returns:
137 ComponentConfiguration instance
138 """
139 all_components = set(component_enum)
141 # Dynamic resolution: remaining components = all_components - multiprocessing_axis
142 remaining_components = all_components - {multiprocessing_axis}
144 # Auto-resolve default_variable if not specified
145 if default_variable is None: 145 ↛ 147line 145 didn't jump to line 147 because the condition on line 145 was never true
146 # Use the first remaining component as default variable
147 default_variable = [list(remaining_components)[0]] if remaining_components else []
149 # Auto-resolve default_group_by if not specified
150 if default_group_by is None and len(remaining_components) > 1: 150 ↛ 152line 150 didn't jump to line 152 because the condition on line 150 was never true
151 # Use the second remaining component as default group_by (if available)
152 remaining_list = list(remaining_components)
153 # Ensure group_by is not in default_variable
154 for component in remaining_list:
155 if component not in default_variable:
156 default_group_by = component
157 break
159 return ComponentConfiguration(
160 all_components=all_components,
161 multiprocessing_axis=multiprocessing_axis,
162 default_variable=default_variable,
163 default_group_by=default_group_by
164 )
166 @staticmethod
167 def create_openhcs_default_configuration():
168 """
169 Create the default OpenHCS configuration.
171 This maintains backward compatibility with the current OpenHCS setup:
172 - Well as multiprocessing axis
173 - Site as default variable component
174 - Channel as default group_by
175 """
176 # Import here to avoid circular import with constants.py
177 from enum import Enum
179 class VariableComponents(Enum):
180 SITE = "site"
181 CHANNEL = "channel"
182 Z_INDEX = "z_index"
183 WELL = "well"
185 return ComponentConfigurationFactory.create_configuration(
186 VariableComponents,
187 multiprocessing_axis=VariableComponents.WELL,
188 default_variable=[VariableComponents.SITE],
189 default_group_by=VariableComponents.CHANNEL
190 )