Coverage for src/pytribeam/GUI/config_ui/validator.py: 0%
102 statements
« prev ^ index » next coverage.py v7.6.1, created at 2026-06-16 18:30 +0000
1"""Configuration validation.
3This module provides validation logic for pipeline configurations,
4separating validation concerns from UI concerns.
5"""
7from dataclasses import dataclass
8from typing import List, Optional, Dict, Tuple, Union
10import pytribeam.factory as factory
11import pytribeam.types as tbt
12import pytribeam.utilities as ut
13from pytribeam.GUI.common.errors import ValidationError
14from pytribeam.GUI.config_ui.pipeline_model import PipelineConfig
17@dataclass
18class ValidationResult:
19 """Result of a validation check.
21 Attributes:
22 success: Whether validation passed
23 step_name: Name of step validated
24 message: Detailed message (usually for failures)
25 exception: Original exception if validation failed
26 settings: Validated settings object (if applicable)
27 """
29 success: bool
30 step_name: str
31 message: str = ""
32 exception: Optional[Exception] = None
33 settings: Union[
34 tbt.CustomSettings,
35 tbt.EBSDSettings,
36 tbt.EDSSettings,
37 tbt.ImageSettings,
38 tbt.FIBSettings,
39 tbt.LaserSettings,
40 ] = None
42 def __str__(self) -> str:
43 """Get human-readable string representation."""
44 status = "passed" if self.success else "failed"
45 msg = f": {self.message}" if self.message else ""
46 return f"{self.step_name} - {status}{msg}"
48 def __bool__(self) -> bool:
49 """Allow using result in boolean context."""
50 return self.success
53class ConfigValidator:
54 """Validates pipeline configurations against schema.
56 This class handles validation of configuration files, checking that
57 all required fields are present and values are valid according to
58 the pytribeam schema.
59 """
61 def __init__(self):
62 """Initialize validator."""
63 self._yml_format = ut.yml_format(version=1.0)
65 def set_version(self, version: float):
66 """Set configuration version for validation.
68 Args:
69 version: Configuration version number
70 """
71 self._yml_format = ut.yml_format(version=version)
73 def validate_general(self, config_dict: Dict) -> ValidationResult:
74 """Validate general configuration section.
76 Args:
77 config_dict: Dictionary containing at least 'general' key
79 Returns:
80 ValidationResult indicating success or failure
81 """
82 try:
83 general_set = factory.general(
84 config_dict,
85 yml_format=self._yml_format,
86 )
87 return ValidationResult(
88 success=True,
89 step_name="general",
90 settings=general_set,
91 )
92 except KeyError as e:
93 return ValidationResult(
94 success=False,
95 step_name="general",
96 message=f"Missing required field: {e}",
97 exception=e,
98 )
99 except Exception as e:
100 return ValidationResult(
101 success=False,
102 step_name="general",
103 message=f"{type(e).__name__}: {str(e)}",
104 exception=e,
105 )
107 def validate_step(
108 self,
109 microscope: tbt.Microscope,
110 step_name: str,
111 step_config: Dict,
112 general_settings,
113 ) -> ValidationResult:
114 """Validate a single pipeline step.
116 Args:
117 microscope: Connected microscope object
118 step_name: Name of step being validated
119 step_config: Step configuration dictionary
120 general_settings: Validated general settings
122 Returns:
123 ValidationResult indicating success or failure
124 """
125 try:
126 step_settings = factory.step(
127 microscope,
128 step_name=step_name,
129 step_settings=step_config,
130 general_settings=general_settings,
131 yml_format=self._yml_format,
132 )
133 return ValidationResult(
134 success=True, step_name=step_name, settings=step_settings
135 )
136 except KeyError as e:
137 return ValidationResult(
138 success=False,
139 step_name=step_name,
140 message=f"Missing required field: {e}",
141 exception=e,
142 )
143 except Exception as e:
144 return ValidationResult(
145 success=False,
146 step_name=step_name,
147 message=f"{type(e).__name__}: {str(e)}",
148 exception=e,
149 )
151 def validate_full_pipeline(
152 self,
153 config_dict: Dict,
154 microscope: Optional[tbt.Microscope] = None,
155 ) -> List[ValidationResult]:
156 """Validate complete pipeline configuration.
158 Args:
159 config_dict: Complete configuration dictionary
160 microscope: Optional connected microscope (created if not provided)
162 Returns:
163 List of ValidationResult objects, one per validated component
164 """
165 results = []
167 # Validate general first
168 general_db = config_dict.get("general", {})
169 general_result = self.validate_general(general_db)
170 results.append(general_result)
172 if not general_result.success:
173 return results
175 # If no steps, we're done
176 if "steps" not in config_dict or not config_dict["steps"]:
177 return results
179 # Get general settings for step validation
180 try:
181 general_set = factory.general(
182 config_dict["general"],
183 yml_format=self._yml_format,
184 )
185 except Exception as e:
186 # This shouldn't happen since we validated above
187 results.append(
188 ValidationResult(
189 success=False,
190 step_name="General (re-validation)",
191 message=f"Failed to re-load general settings: {e}",
192 exception=e,
193 )
194 )
195 return results
197 # Create or use provided microscope connection
198 should_disconnect = False
199 if microscope is None:
200 try:
201 microscope = tbt.Microscope()
202 ut.connect_microscope(
203 microscope,
204 quiet_output=True,
205 connection_host=general_set.connection.host,
206 connection_port=general_set.connection.port,
207 )
208 should_disconnect = True
209 except Exception as e:
210 results.append(
211 ValidationResult(
212 success=False,
213 step_name="Microscope Connection",
214 message=f"Failed to connect: {e}",
215 exception=e,
216 )
217 )
218 return results
220 # Validate each step
221 try:
222 for step_name, step_config in config_dict["steps"].items():
223 result = self.validate_step(
224 microscope,
225 step_name,
226 step_config,
227 general_set,
228 )
229 results.append(result)
230 finally:
231 # Clean up microscope connection if we created it
232 if should_disconnect and microscope:
233 try:
234 ut.disconnect_microscope(microscope)
235 except Exception:
236 pass # Ignore disconnection errors
238 return results
240 def validate_pipeline_model(
241 self,
242 pipeline: PipelineConfig,
243 ) -> List[ValidationResult]:
244 """Validate a PipelineConfig model.
246 Convenience method that converts model to dictionary and validates.
248 Args:
249 pipeline: PipelineConfig to validate
250 microscope: Optional connected microscope
252 Returns:
253 List of ValidationResult objects
254 """
255 config_dict = pipeline.to_dict()
256 return self.validate_full_pipeline(config_dict)
258 def check_duplicate_names(self, pipeline: PipelineConfig) -> ValidationResult:
259 """Check for duplicate step names.
261 Args:
262 pipeline: PipelineConfig to check
264 Returns:
265 ValidationResult indicating if names are unique
266 """
267 is_valid, duplicates = pipeline.validate_step_names()
269 if is_valid:
270 return ValidationResult(
271 success=True,
272 step_name="Step Names",
273 )
274 else:
275 dup_list = ", ".join(duplicates)
276 return ValidationResult(
277 success=False,
278 step_name="Step Names",
279 message=f"Duplicate names found: {dup_list}",
280 )
282 def check_has_steps(self, pipeline: PipelineConfig) -> ValidationResult:
283 """Check that pipeline has at least one step.
285 Args:
286 pipeline: PipelineConfig to check
288 Returns:
289 ValidationResult indicating if pipeline has steps
290 """
291 if pipeline.get_step_count() > 0:
292 return ValidationResult(
293 success=True,
294 step_name="Step Count",
295 )
296 else:
297 return ValidationResult(
298 success=False,
299 step_name="Step Count",
300 message="Pipeline must have at least one step",
301 )
303 def validate_pipeline_structure(
304 self,
305 pipeline: PipelineConfig,
306 ) -> List[ValidationResult]:
307 """Validate pipeline structure without schema validation.
309 Checks basic requirements like unique names and presence of steps.
310 This is faster than full validation and doesn't require microscope connection.
312 Args:
313 pipeline: PipelineConfig to validate
315 Returns:
316 List of ValidationResult objects
317 """
318 results = []
320 # Check step count
321 results.append(self.check_has_steps(pipeline))
323 # Check duplicate names
324 results.append(self.check_duplicate_names(pipeline))
326 return results
328 @staticmethod
329 def get_summary(results: List[ValidationResult]) -> Tuple[bool, str]:
330 """Get summary of validation results.
332 Args:
333 results: List of ValidationResult objects
335 Returns:
336 Tuple of (all_passed, summary_message)
337 """
338 all_passed = all(r.success for r in results)
339 lines = [str(r) for r in results]
340 summary = "\n".join(lines)
342 if all_passed:
343 summary = f"✓ All checks passed\n\n{summary}"
344 else:
345 failed_count = sum(1 for r in results if not r.success)
346 summary = f"✗ {failed_count} check(s) failed\n\n{summary}"
348 return all_passed, summary