File size: 9,080 Bytes
11898c7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 | #!/usr/bin/env python3
"""Validate the complete self-healing ML system."""
import logging
import sys
from pathlib import Path
# Add project root to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from pipelines.train import TrainingPipeline
from pipelines.inference import InferencePipeline
from pipelines.rollback import RollbackPipeline
from orchestration.controller import SelfHealingController
from healing.healing_actions import HealingActions
from monitoring.data_drift import DataDriftDetector
from decision_engine.policy_engine import PolicyEngine
from utils.config_loader import load_config, PipelineConfig
from loguru import logger
# Configure logging
logging.basicConfig(level=logging.INFO)
logger.add("logs/validation.log", rotation="1 day", retention="7 days")
class SystemValidator:
"""Validates all components of the self-healing ML system."""
def __init__(self):
"""Initialize validator."""
self.results = {
"components": {},
"integration": {},
"overall": "PASS"
}
def validate_component(self, name: str, func, *args, **kwargs) -> bool:
"""
Validate a single component.
Args:
name: Component name
func: Validation function
*args: Function arguments
**kwargs: Function keyword arguments
Returns:
True if validation passes
"""
try:
logger.info(f"Validating {name}...")
result = func(*args, **kwargs)
self.results["components"][name] = {
"status": "PASS",
"result": result
}
logger.success(f"✓ {name} passed")
return True
except Exception as e:
self.results["components"][name] = {
"status": "FAIL",
"error": str(e)
}
logger.error(f"✗ {name} failed: {e}")
self.results["overall"] = "FAIL"
return False
def validate_config_loader(self) -> bool:
"""Validate configuration loader."""
config = load_config("configs/pipeline.yaml")
return isinstance(config, PipelineConfig)
def validate_drift_detector(self) -> bool:
"""Validate drift detector."""
detector = DataDriftDetector(method="ks", threshold=0.05)
# Test data
import numpy as np
ref_data = np.random.normal(0, 1, 1000)
curr_data = np.random.normal(0, 1, 1000)
result = detector.detect_drift(ref_data, curr_data)
return result is not None
def validate_policy_engine(self) -> bool:
"""Validate policy engine."""
engine = PolicyEngine(config_path="configs/healing_policies.yaml")
# Test with sample signals
signals = {"data_drift": 0.3, "accuracy_drop": 0.15}
action, trace = engine.decide(signals)
return action is not None and trace is not None
def validate_healing_actions(self) -> bool:
"""Validate healing actions."""
config = load_config("configs/pipeline.yaml")
healing = HealingActions(config.model_dump())
# Test fallback action
result = healing.fallback()
return result["status"] in ["success", "failed"]
def validate_training_pipeline(self) -> bool:
"""Validate training pipeline."""
pipeline = TrainingPipeline()
# Create dummy data for testing
import pandas as pd
import numpy as np
data = pd.DataFrame({
"feature_1": np.random.randn(100),
"feature_2": np.random.randn(100),
"target": np.random.randint(0, 2, 100)
})
result = pipeline.run(data_path=None, data=data)
return result["status"] == "success"
def validate_inference_pipeline(self) -> bool:
"""Validate inference pipeline."""
pipeline = InferencePipeline()
# Get model info
info = pipeline.get_model_info()
return isinstance(info, dict)
def validate_rollback_pipeline(self) -> bool:
"""Validate rollback pipeline."""
pipeline = RollbackPipeline()
# List available models
models = pipeline.list_available_models()
return isinstance(models, list)
def validate_controller(self) -> bool:
"""Validate self-healing controller."""
controller = SelfHealingController()
# Get status
status = controller.get_status()
return status is not None
def validate_integration(self) -> bool:
"""Validate integration between components."""
logger.info("Validating integration...")
integration_tests = []
try:
# Test 1: Config -> Controller
config = load_config("configs/pipeline.yaml")
controller = SelfHealingController()
integration_tests.append(("Config->Controller", True))
# Test 2: Training -> Inference
train_pipeline = TrainingPipeline()
inference_pipeline = InferencePipeline()
integration_tests.append(("Training->Inference", True))
# Test 3: Detection -> Decision -> Healing
detector = DataDriftDetector()
policy_engine = PolicyEngine()
healing = HealingActions(config.model_dump())
integration_tests.append(("Detection->Decision->Healing", True))
# Record results
self.results["integration"] = {
test[0]: "PASS" if test[1] else "FAIL"
for test in integration_tests
}
all_passed = all(test[1] for test in integration_tests)
if all_passed:
logger.success("✓ Integration tests passed")
else:
logger.error("✗ Some integration tests failed")
return all_passed
except Exception as e:
logger.error(f"✗ Integration validation failed: {e}")
self.results["overall"] = "FAIL"
return False
def run_validation(self) -> dict:
"""Run complete validation suite."""
logger.info("Starting system validation")
logger.info("="*60)
# Validate individual components
component_tests = [
("Configuration Loader", self.validate_config_loader),
("Drift Detector", self.validate_drift_detector),
("Policy Engine", self.validate_policy_engine),
("Healing Actions", self.validate_healing_actions),
("Training Pipeline", self.validate_training_pipeline),
("Inference Pipeline", self.validate_inference_pipeline),
("Rollback Pipeline", self.validate_rollback_pipeline),
("Controller", self.validate_controller),
]
for name, test_func in component_tests:
self.validate_component(name, test_func)
# Validate integration
self.validate_integration()
# Generate summary
self._generate_summary()
return self.results
def _generate_summary(self):
"""Generate validation summary."""
logger.info("\n" + "="*60)
logger.info("VALIDATION SUMMARY")
logger.info("="*60)
# Component results
logger.info("\nComponents:")
for name, result in self.results["components"].items():
status = result["status"]
if status == "PASS":
logger.success(f" ✓ {name}: {status}")
else:
logger.error(f" ✗ {name}: {status}")
# Integration results
logger.info("\nIntegration:")
for name, status in self.results["integration"].items():
if status == "PASS":
logger.success(f" ✓ {name}: {status}")
else:
logger.error(f" ✗ {name}: {status}")
# Overall result
logger.info("\n" + "="*60)
if self.results["overall"] == "PASS":
logger.success("OVERALL: PASS - System is ready for production!")
else:
logger.error("OVERALL: FAIL - Some components need attention")
logger.info("="*60)
def main():
"""Run complete system validation."""
validator = SystemValidator()
results = validator.run_validation()
# Save results to file
import json
results_file = Path("validation_results.json")
with open(results_file, 'w') as f:
json.dump(results, f, indent=2, default=str)
logger.info(f"\nDetailed results saved to: {results_file}")
# Exit with appropriate code
if results["overall"] == "FAIL":
sys.exit(1)
else:
sys.exit(0)
if __name__ == "__main__":
main()
|