Spaces:
Runtime error
Runtime error
| """ | |
| Test API Connection with Trained Model | |
| Verify that: | |
| 1. Model loads correctly | |
| 2. predict_soh() works | |
| 3. API endpoints work | |
| 4. End-to-end prediction flow | |
| """ | |
| import pandas as pd | |
| import numpy as np | |
| import pickle | |
| import json | |
| from predict_soh import predict_soh | |
| from api_service import get_model_info_internal | |
| def test_model_loading(): | |
| """Test that model loads correctly.""" | |
| print("="*80) | |
| print("TEST 1: MODEL LOADING") | |
| print("="*80) | |
| try: | |
| with open('./models/nasa_soh_model.pkl', 'rb') as f: | |
| model_data = pickle.load(f) | |
| model = model_data['model'] | |
| feature_cols = model_data['feature_cols'] | |
| metrics = model_data['metrics'] | |
| print(f"✅ Model loaded successfully") | |
| print(f" Model type: {type(model).__name__}") | |
| print(f" Feature count: {len(feature_cols)}") | |
| print(f" Validation R²: {metrics['val_r2']:.6f}") | |
| print(f" Validation MAE: {metrics['val_mae']:.6f}") | |
| print(f" OOB Score: {metrics.get('oob_score', 'N/A')}") | |
| return True, model_data | |
| except Exception as e: | |
| print(f"❌ Model loading failed: {e}") | |
| return False, None | |
| def test_prediction_function(): | |
| """Test predict_soh() function.""" | |
| print("\n" + "="*80) | |
| print("TEST 2: PREDICTION FUNCTION") | |
| print("="*80) | |
| try: | |
| # Load test data (B0007, first 100 cycles) | |
| test_data = pd.read_csv('./data/nasa_converted/B0007.csv') | |
| test_subset = test_data[test_data['cycle_index'] <= 100].copy() | |
| print(f"Test data: {len(test_subset)} cycles") | |
| print(f"Columns: {test_subset.columns.tolist()}") | |
| # Run prediction | |
| predictions = predict_soh( | |
| test_subset, | |
| model_path='./models/nasa_soh_model.pkl', | |
| pipeline_path='./models/feature_pipeline_causal.pkl' | |
| ) | |
| print(f"\n✅ Prediction successful!") | |
| print(f" Current SOH: {predictions['current_soh']['point_estimate']:.4f}") | |
| print(f" Health Score: {predictions['health_score']:.2f}/100") | |
| print(f" Remaining Life: {predictions['remaining_life']['years']:.2f} years") | |
| print(f" Degradation Rate: {predictions['degradation_rate']['per_year_pct']:.2f}%/year") | |
| print(f" Data Quality: {predictions['data_quality']['quality_score']:.2f}") | |
| print(f" Confidence: {predictions['confidence']:.4f}") | |
| # Test forecasts | |
| print(f"\n Forecasts:") | |
| for key, forecast in predictions['forecasts'].items(): | |
| cycles = key.replace('soh_at_', '').replace('_cycles', '') | |
| print(f" {cycles} cycles: {forecast['point_estimate']:.4f} [{forecast['lower_95']:.4f}, {forecast['upper_95']:.4f}]") | |
| return True, predictions | |
| except Exception as e: | |
| print(f"❌ Prediction failed: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return False, None | |
| def test_api_info(): | |
| """Test API model info endpoint.""" | |
| print("\n" + "="*80) | |
| print("TEST 3: API MODEL INFO") | |
| print("="*80) | |
| try: | |
| info = get_model_info_internal() | |
| print(f"✅ Model info retrieved successfully") | |
| print(f" Model Type: {info['model_type']}") | |
| print(f" Training Data: {info['training_data']}") | |
| print(f" Validation Data: {info['validation_data']}") | |
| print(f" R² Score: {info['metrics']['r2']:.6f}") | |
| print(f" MAE: {info['metrics']['mae']:.6f}") | |
| print(f" RMSE: {info['metrics']['rmse']:.6f}") | |
| print(f" Feature Count: {info['feature_count']}") | |
| print(f" Version: {info['version']}") | |
| return True, info | |
| except Exception as e: | |
| print(f"❌ API info failed: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return False, None | |
| def test_full_pipeline(): | |
| """Test complete prediction pipeline.""" | |
| print("\n" + "="*80) | |
| print("TEST 4: FULL PREDICTION PIPELINE") | |
| print("="*80) | |
| try: | |
| # Simulate BMS data (new battery, not in training set) | |
| bms_data = pd.DataFrame({ | |
| 'cycle_index': list(range(1, 51)), | |
| 'voltage': np.linspace(4.2, 3.8, 50) + np.random.normal(0, 0.02, 50), | |
| 'current': np.random.choice([1.5, -1.5], 50), | |
| 'temperature': 25 + np.random.normal(0, 1, 50), | |
| 'capacity': np.linspace(2.0, 1.9, 50) + np.random.normal(0, 0.01, 50) | |
| }) | |
| print(f"Simulated BMS data: {len(bms_data)} cycles") | |
| predictions = predict_soh( | |
| bms_data, | |
| model_path='./models/nasa_soh_model.pkl', | |
| pipeline_path='./models/feature_pipeline_causal.pkl', | |
| initial_capacity=2.0 | |
| ) | |
| print(f"\n✅ Full pipeline test successful!") | |
| print(f" Current Cycle: {predictions['current_cycle']}") | |
| print(f" Current SOH: {predictions['current_soh']['point_estimate']:.4f}") | |
| print(f" 95% Interval: [{predictions['current_soh']['lower_95']:.4f}, {predictions['current_soh']['upper_95']:.4f}]") | |
| print(f" Health Score: {predictions['health_score']:.2f}/100") | |
| print(f" RUL (cycles): {predictions['remaining_life']['cycles']:.1f}") | |
| print(f" RUL (years): {predictions['remaining_life']['years']:.2f}") | |
| return True | |
| except Exception as e: | |
| print(f"❌ Full pipeline test failed: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| return False | |
| if __name__ == "__main__": | |
| print("="*80) | |
| print("EXPERT MODEL - API CONNECTION TESTS") | |
| print("="*80) | |
| results = {} | |
| # Test 1: Model loading | |
| success, model_data = test_model_loading() | |
| results['model_loading'] = success | |
| # Test 2: Prediction function | |
| success, predictions = test_prediction_function() | |
| results['prediction_function'] = success | |
| # Test 3: API info | |
| success, api_info = test_api_info() | |
| results['api_info'] = success | |
| # Test 4: Full pipeline | |
| success = test_full_pipeline() | |
| results['full_pipeline'] = success | |
| # Summary | |
| print("\n" + "="*80) | |
| print("TEST SUMMARY") | |
| print("="*80) | |
| for test_name, success in results.items(): | |
| status = "✅ PASS" if success else "❌ FAIL" | |
| print(f" {test_name}: {status}") | |
| all_passed = all(results.values()) | |
| if all_passed: | |
| print("\n" + "="*80) | |
| print("✅ ALL TESTS PASSED - MODEL IS PRODUCTION-READY") | |
| print("="*80) | |
| print("\nNext steps:") | |
| print(" 1. Start API service: python api_service.py") | |
| print(" 2. API will be available at: http://localhost:8000") | |
| print(" 3. Test endpoint: POST /api/predict") | |
| print(" 4. Get model info: GET /api/model-info") | |
| else: | |
| print("\n" + "="*80) | |
| print("⚠️ SOME TESTS FAILED - CHECK ERRORS ABOVE") | |
| print("="*80) | |