""" Production Validation Module Business logic gates and quality checks for battery assets. MIT/Stanford Quality: - Multi-level validation (technical, business, quality) - Confidence scoring - Clear error messages - Audit trail """ import pandas as pd import numpy as np from typing import Dict, List, Optional from datetime import datetime class ValidationError(Exception): """Custom exception for validation failures.""" pass class BatteryValidator: """ Production-grade validator for battery assets. """ def __init__(self): self.validation_history = [] def validate_data_quality(self, battery_data: pd.DataFrame) -> Dict: """ Validate technical data quality. Returns: -------- dict: {is_valid: bool, score: float, issues: list} """ issues = [] warnings = [] # Check required columns required_cols = ['cycle_index', 'voltage', 'current', 'temperature', 'capacity'] missing = [col for col in required_cols if col not in battery_data.columns] if missing: issues.append(f"Missing required columns: {missing}") return {'is_valid': False, 'score': 0.0, 'issues': issues, 'warnings': warnings} # Check minimum data points if len(battery_data) < 10: issues.append(f"Insufficient data: {len(battery_data)} cycles (minimum 10 required)") elif len(battery_data) < 50: warnings.append(f"Limited data: {len(battery_data)} cycles (recommend 50+ for accuracy)") # Check value ranges v_min, v_max = battery_data['voltage'].min(), battery_data['voltage'].max() if v_min < 0 or v_max > 5: issues.append(f"Voltage out of range: {v_min:.2f}V - {v_max:.2f}V (expected 0-5V)") if v_min == v_max: issues.append("Voltage is constant (no variation detected)") # Check capacity c_min, c_max = battery_data['capacity'].min(), battery_data['capacity'].max() if c_min < 0: issues.append(f"Negative capacity detected: {c_min:.3f}") if c_min == c_max: warnings.append("Capacity is constant (no degradation detected)") # Check temperature t_min, t_max = battery_data['temperature'].min(), battery_data['temperature'].max() if t_min < -20 or t_max > 80: warnings.append(f"Unusual temperature range: {t_min:.1f}°C - {t_max:.1f}°C") # Data completeness completeness = 1 - battery_data.isna().sum().sum() / (len(battery_data) * len(battery_data.columns)) if completeness < 0.9: warnings.append(f"Data completeness: {completeness*100:.1f}% (recommend >90%)") # Calculate quality score score = 1.0 score *= completeness score *= (1 - len(issues) * 0.2) # Penalty for issues score *= (1 - len(warnings) * 0.05) # Small penalty for warnings score = max(0, min(1, score)) is_valid = len(issues) == 0 and score >= 0.7 return { 'is_valid': is_valid, 'score': score, 'completeness': completeness, 'issues': issues, 'warnings': warnings } def validate_for_certification( self, battery_data: pd.DataFrame, prediction_result: Dict ) -> Dict: """ Validate if asset qualifies for "Certified" status. Certification requirements: - Data quality score >= 0.8 - Minimum 50 cycles - Prediction confidence >= 0.7 - Health score available Returns: -------- dict: {can_certify: bool, stage: str, reason: str} """ # Data quality check quality = self.validate_data_quality(battery_data) if not quality['is_valid']: return { 'can_certify': False, 'stage': 'Pending', 'reason': f"Data quality issues: {', '.join(quality['issues'])}" } # Minimum cycles check if len(battery_data) < 50: return { 'can_certify': False, 'stage': 'Monitoring', 'reason': f"Insufficient cycle history: {len(battery_data)} cycles (minimum 50)" } # Prediction confidence check confidence = prediction_result.get('confidence', 0) if confidence < 0.7: return { 'can_certify': False, 'stage': 'Under Review', 'reason': f"Low prediction confidence: {confidence:.2f} (minimum 0.70)" } # Quality score check if quality['score'] < 0.8: return { 'can_certify': False, 'stage': 'Under Review', 'reason': f"Quality score below threshold: {quality['score']:.2f} (minimum 0.80)" } # All checks passed return { 'can_certify': True, 'stage': 'Certified', 'reason': f"All validation criteria met (Quality: {quality['score']:.2f}, Confidence: {confidence:.2f})" } def validate_business_logic( self, asset_metadata: Dict ) -> Dict: """ Validate business logic constraints. Returns: -------- dict: {is_valid: bool, warnings: list} """ warnings = [] # Check capacity range capacity = asset_metadata.get('capacity', 0) if capacity > 10000: # > 10 MWh warnings.append("Exceptionally large capacity (>10 MWh) - verify data") elif capacity < 0.1: # < 100 Wh warnings.append("Very small capacity (<0.1 kWh) - likely lab cell") # Check project stage consistency stage = asset_metadata.get('project_stage', '') category = asset_metadata.get('category', []) if stage == 'Certified' and 'Research' in category: warnings.append("Research data marked as Certified - consider 'Monitoring' stage") # Check realistic degradation rate deg_rate = asset_metadata.get('degradation_rate', 0) if deg_rate > 50: warnings.append(f"Extremely high degradation rate: {deg_rate:.1f}%/year") elif deg_rate < 0.1: warnings.append("Very low degradation detected - verify data quality") return { 'is_valid': True, # Warnings don't block, just inform 'warnings': warnings } def log_validation(self, asset_id: str, validation_result: Dict): """Log validation for audit trail.""" self.validation_history.append({ 'timestamp': datetime.utcnow().isoformat(), 'asset_id': asset_id, 'result': validation_result }) if __name__ == "__main__": # Test validation print("Testing production validator...") validator = BatteryValidator() # Test 1: Good quality data print("\n1. NASA B0005 (should pass):") nasa_data = pd.read_csv('./data/nasa_converted/B0005.csv') result = validator.validate_data_quality(nasa_data) print(f" Valid: {result['is_valid']}") print(f" Quality Score: {result['score']:.2f}") print(f" Issues: {len(result['issues'])}") # Test 2: Certification check print("\n2. Certification check:") mock_prediction = {'confidence': 0.85, 'health_score': 85} cert_result = validator.validate_for_certification(nasa_data, mock_prediction) print(f" Can Certify: {cert_result['can_certify']}") print(f" Stage: {cert_result['stage']}") print(f" Reason: {cert_result['reason']}") # Test 3: Business logic print("\n3. Business logic:") mock_asset = { 'capacity': 1.89, 'project_stage': 'Certified', 'category': ['Research', 'Validated'], 'degradation_rate': 15.5 } business_result = validator.validate_business_logic(mock_asset) print(f" Valid: {business_result['is_valid']}") print(f" Warnings: {len(business_result['warnings'])}") if business_result['warnings']: for w in business_result['warnings']: print(f" - {w}") print("\n✅ Validation system test complete!")