diff --git a/inference/test_model.py b/inference/test_model.py new file mode 100644 index 0000000..29a0fd1 --- /dev/null +++ b/inference/test_model.py @@ -0,0 +1,334 @@ +""" +Flight Event Prediction - Model Testing Script + +This script tests trained Random Forest models for flight event prediction. +It loads saved models, runs inference on test data, and evaluates performance. + +Outputs: + - Prediction results: predictions_output.csv + - Evaluation metrics: accuracy, precision, recall +""" + +import pandas as pd +import numpy as np +from pathlib import Path +from sklearn.metrics import accuracy_score, precision_score, recall_score, classification_report, confusion_matrix +import joblib +import logging +from datetime import datetime + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s - %(levelname)s - %(message)s' +) +logger = logging.getLogger(__name__) + + +class FlightEventModelTester: + """ + Tester for trained Random Forest flight event prediction models. + """ + + def __init__(self): + """Initialize the model tester.""" + self.model = None + self.feature_columns = [ + 'altitude', + 'heading', + 'vertical_speed', + 'velocity', + 'roll', + 'pitch', + 'yaw', + 'g_force' + ] + self.target_column = 'event_label' + + def load_model(self, model_path: Path): + """ + Load trained Random Forest model from disk. + + Args: + model_path: Path to saved model (.pkl file) + """ + logger.info(f"Loading model from: {model_path}") + + if not model_path.exists(): + raise FileNotFoundError(f"Model file not found: {model_path}") + + self.model = joblib.load(model_path) + logger.info(f"✓ Model loaded successfully: {type(self.model).__name__}") + + # Display model parameters + if hasattr(self.model, 'n_estimators'): + logger.info(f" - Number of estimators: {self.model.n_estimators}") + if hasattr(self.model, 'max_depth'): + logger.info(f" - Max depth: {self.model.max_depth}") + + def load_test_data(self, data_path: Path) -> pd.DataFrame: + """ + Load test dataset for inference. + + Args: + data_path: Path to test data CSV file + + Returns: + DataFrame containing test data + """ + logger.info(f"\nLoading test data from: {data_path}") + + if not data_path.exists(): + raise FileNotFoundError(f"Test data file not found: {data_path}") + + df = pd.read_csv(data_path) + logger.info(f"✓ Loaded {len(df)} test samples") + + # Verify required feature columns exist + missing_cols = [col for col in self.feature_columns if col not in df.columns] + if missing_cols: + raise ValueError(f"Missing required feature columns: {missing_cols}") + + # Check if labels exist (for evaluation) + has_labels = self.target_column in df.columns + if has_labels: + logger.info(f"✓ Test data contains ground truth labels") + logger.info("\nActual label distribution:") + print(df[self.target_column].value_counts().to_string()) + else: + logger.warning("⚠ Test data does not contain labels (evaluation will be skipped)") + + return df + + def run_inference(self, df: pd.DataFrame) -> np.ndarray: + """ + Run inference on test data to generate predictions. + + Args: + df: DataFrame containing test features + + Returns: + Array of predicted labels + """ + if self.model is None: + raise ValueError("No model loaded. Call load_model() first.") + + logger.info("\n" + "="*60) + logger.info("Running Inference") + logger.info("="*60) + + # Extract features + X_test = df[self.feature_columns] + + logger.info(f"\nGenerating predictions for {len(X_test)} samples...") + predictions = self.model.predict(X_test) + + logger.info("✓ Inference completed") + logger.info(f"\nPredicted label distribution:") + unique, counts = np.unique(predictions, return_counts=True) + for label, count in zip(unique, counts): + logger.info(f" {label}: {count} ({count/len(predictions)*100:.2f}%)") + + return predictions + + def evaluate_predictions(self, y_true: pd.Series, y_pred: np.ndarray) -> dict: + """ + Evaluate model predictions against ground truth labels. + + Args: + y_true: Ground truth labels + y_pred: Predicted labels + + Returns: + Dictionary containing evaluation metrics + """ + logger.info("\n" + "="*60) + logger.info("Model Evaluation") + logger.info("="*60) + + # Calculate metrics + accuracy = accuracy_score(y_true, y_pred) + precision = precision_score(y_true, y_pred, average='weighted', zero_division=0) + recall = recall_score(y_true, y_pred, average='weighted', zero_division=0) + + metrics = { + 'accuracy': accuracy, + 'precision': precision, + 'recall': recall, + 'timestamp': datetime.now().isoformat() + } + + # Display evaluation metrics + logger.info("\n📊 Evaluation Metrics:") + logger.info("="*60) + logger.info(f" Accuracy: {accuracy:.4f} ({accuracy*100:.2f}%)") + logger.info(f" Precision: {precision:.4f} ({precision*100:.2f}%)") + logger.info(f" Recall: {recall:.4f} ({recall*100:.2f}%)") + logger.info("="*60) + + # Display detailed classification report + logger.info("\n📋 Detailed Classification Report:") + print("\n" + classification_report(y_true, y_pred, zero_division=0)) + + # Display confusion matrix + logger.info("🔢 Confusion Matrix:") + cm = confusion_matrix(y_true, y_pred) + classes = sorted(y_true.unique()) + + # Create formatted confusion matrix + cm_df = pd.DataFrame(cm, index=classes, columns=classes) + cm_df.index.name = 'Actual' + cm_df.columns.name = 'Predicted' + print("\n" + cm_df.to_string()) + + return metrics + + def save_predictions(self, df: pd.DataFrame, predictions: np.ndarray, output_path: Path): + """ + Save prediction results to CSV file. + + Args: + df: Original test DataFrame + predictions: Predicted labels + output_path: Path to save predictions + """ + logger.info(f"\n💾 Saving predictions to: {output_path}") + + # Create output DataFrame with original data and predictions + output_df = df.copy() + output_df['predicted_event_label'] = predictions + + # Add prediction confidence if available (for Random Forest) + if hasattr(self.model, 'predict_proba'): + probabilities = self.model.predict_proba(df[self.feature_columns]) + max_probabilities = np.max(probabilities, axis=1) + output_df['prediction_confidence'] = max_probabilities + + # Save to CSV + output_path.parent.mkdir(parents=True, exist_ok=True) + output_df.to_csv(output_path, index=False) + + logger.info(f"✓ Saved {len(output_df)} predictions") + logger.info(f" Columns: {', '.join(output_df.columns.tolist())}") + + def save_metrics(self, metrics: dict, output_path: Path): + """ + Save evaluation metrics to JSON file. + + Args: + metrics: Dictionary of evaluation metrics + output_path: Path to save metrics + """ + import json + + logger.info(f"\n💾 Saving metrics to: {output_path}") + + output_path.parent.mkdir(parents=True, exist_ok=True) + + with output_path.open('w') as f: + json.dump(metrics, f, indent=2) + + logger.info("✓ Metrics saved successfully") + + +def main(): + """Main execution function for model testing.""" + # Define paths relative to inference folder + inference_dir = Path(__file__).parent + + # Model paths (try both bestModel.pkl and finalModel.pkl) + models_dir = inference_dir / 'Models' + model_paths = [ + models_dir / 'bestModel.pkl', + models_dir / 'finalModel.pkl' + ] + + # Data paths (try multiple locations) + data_candidates = [ + inference_dir / 'dataset' / 'test.csv', # Processed test set + inference_dir / 'Data' / 'labeled_flight_data.csv', # Full labeled dataset + inference_dir / 'labeled_flight_data.csv' # Alternative location + ] + + # Output paths + output_predictions = inference_dir / 'predictions_output.csv' + output_metrics = inference_dir / 'evaluation_metrics.json' + + logger.info("="*60) + logger.info("Flight Event Prediction - Model Testing") + logger.info("="*60) + + try: + # Initialize tester + tester = FlightEventModelTester() + + # Load model (try available model files) + model_loaded = False + for model_path in model_paths: + if model_path.exists(): + tester.load_model(model_path) + model_loaded = True + break + + if not model_loaded: + raise FileNotFoundError( + f"No model files found. Searched: {', '.join(str(p) for p in model_paths)}\n" + "Please train a model first using train_model.py" + ) + + # Load test data (try available data files) + test_data = None + data_path_used = None + for data_path in data_candidates: + if data_path.exists(): + test_data = tester.load_test_data(data_path) + data_path_used = data_path + break + + if test_data is None: + raise FileNotFoundError( + f"No test data found. Searched: {', '.join(str(p) for p in data_candidates)}\n" + "Please ensure labeled or processed data is available." + ) + + # Run inference + predictions = tester.run_inference(test_data) + + # Evaluate if ground truth labels are available + metrics = None + if tester.target_column in test_data.columns: + y_true = test_data[tester.target_column] + metrics = tester.evaluate_predictions(y_true, predictions) + + # Save metrics + tester.save_metrics(metrics, output_metrics) + else: + logger.info("\n⚠ Skipping evaluation (no ground truth labels available)") + + # Save predictions + tester.save_predictions(test_data, predictions, output_predictions) + + # Final summary + logger.info("\n" + "="*60) + logger.info("✅ Testing Complete!") + logger.info("="*60) + logger.info(f"\nModel used: {model_path}") + logger.info(f"Test data: {data_path_used}") + logger.info(f"Predictions saved: {output_predictions}") + + if metrics: + logger.info(f"Metrics saved: {output_metrics}") + logger.info(f"\n📊 Final Results:") + logger.info(f" Accuracy: {metrics['accuracy']:.4f}") + logger.info(f" Precision: {metrics['precision']:.4f}") + logger.info(f" Recall: {metrics['recall']:.4f}") + + logger.info("\n✓ All outputs saved in inference/ folder") + + except Exception as e: + logger.error(f"\n❌ Testing failed: {e}") + raise + + +if __name__ == '__main__': + main() \ No newline at end of file