Spaces:
Runtime error
Runtime error
| """ | |
| 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!") | |