ahadalii/Predictive_Maintenance_System
0
1"""2Machine Learning Model for Predictive Maintenance3Uses Random Forest Classifier for failure prediction4"""5 6import pandas as pd7import numpy as np8import pickle9from sklearn.ensemble import RandomForestClassifier10from sklearn.metrics import (11 accuracy_score, precision_score, recall_score, f1_score,12 confusion_matrix, classification_report, roc_auc_score13)14import warnings15warnings.filterwarnings('ignore')16 17class PredictiveMaintenanceModel:18 def __init__(self):19 """Initialize the model"""20 self.model = RandomForestClassifier(21 n_estimators=100,22 max_depth=10,23 min_samples_split=5,24 min_samples_leaf=2,25 random_state=42,26 class_weight='balanced' # Handle class imbalance27 )28 self.is_trained = False29 self.feature_importance_ = None30 31 def train(self, X_train, y_train):32 """Train the model"""33 print("Training Random Forest Classifier...")34 self.model.fit(X_train, y_train)35 self.is_trained = True36 37 # Get feature importance38 self.feature_importance_ = pd.DataFrame({39 'feature': X_train.columns,40 'importance': self.model.feature_importances_41 }).sort_values('importance', ascending=False)42 43 print("Model training complete!")44 return self45 46 def predict(self, X):47 """Make predictions"""48 if not self.is_trained:49 raise ValueError("Model must be trained first!")50 return self.model.predict(X)51 52 def predict_proba(self, X):53 """Get prediction probabilities"""54 if not self.is_trained:55 raise ValueError("Model must be trained first!")56 return self.model.predict_proba(X)57 58 def evaluate(self, X_test, y_test):59 """Evaluate the model"""60 if not self.is_trained:61 raise ValueError("Model must be trained first!")62 63 # Predictions64 y_pred = self.predict(X_test)65 y_pred_proba = self.predict_proba(X_test)[:, 1]66 67 # Metrics68 accuracy = accuracy_score(y_test, y_pred)69 precision = precision_score(y_test, y_pred, zero_division=0)70 recall = recall_score(y_test, y_pred, zero_division=0)71 f1 = f1_score(y_test, y_pred, zero_division=0)72 roc_auc = roc_auc_score(y_test, y_pred_proba)73 74 # Confusion matrix75 cm = confusion_matrix(y_test, y_pred)76 77 # Classification report78 report = classification_report(y_test, y_pred, zero_division=0)79 80 results = {81 'accuracy': accuracy,82 'precision': precision,83 'recall': recall,84 'f1_score': f1,85 'roc_auc': roc_auc,86 'confusion_matrix': cm,87 'classification_report': report,88 'y_pred': y_pred,89 'y_pred_proba': y_pred_proba90 }91 92 print("="*60)93 print("MODEL EVALUATION RESULTS")94 print("="*60)95 print(f"Accuracy: {accuracy:.4f}")96 print(f"Precision: {precision:.4f}")97 print(f"Recall: {recall:.4f}")98 print(f"F1-Score: {f1:.4f}")99 print(f"ROC-AUC: {roc_auc:.4f}")100 print("\nConfusion Matrix:")101 print(cm)102 print("\nClassification Report:")103 print(report)104 105 return results106 107 def predict_maintenance(self, X, tool_wear_values=None):108 """109 Predict maintenance needs and time to failure110 111 Args:112 X: Feature matrix113 tool_wear_values: Array of tool wear values (optional)114 115 Returns:116 DataFrame with predictions and maintenance recommendations117 """118 if not self.is_trained:119 raise ValueError("Model must be trained first!")120 121 # Get failure predictions122 failure_pred = self.predict(X)123 failure_proba = self.predict_proba(X)[:, 1]124 125 # Estimate time to failure based on tool wear126 # Average tool wear at failure is around 100-150 minutes based on EDA127 avg_tool_wear_at_failure = 120 # minutes128 129 if tool_wear_values is None:130 # Try to get tool wear from features if available131 if 'Tool wear [min]' in X.columns:132 tool_wear_values = X['Tool wear [min]'].values133 else:134 tool_wear_values = np.zeros(len(X))135 else:136 # Convert to numpy array if it's a list or scalar137 tool_wear_values = np.array(tool_wear_values)138 # If it's a scalar, make it an array139 if tool_wear_values.ndim == 0:140 tool_wear_values = np.array([tool_wear_values])141 142 # Ensure tool_wear_values is 1D array with same length as predictions143 if len(tool_wear_values) != len(failure_pred):144 # If single value provided, repeat it for all predictions145 if len(tool_wear_values) == 1:146 tool_wear_values = np.repeat(tool_wear_values, len(failure_pred))147 else:148 raise ValueError(f"tool_wear_values length ({len(tool_wear_values)}) doesn't match predictions length ({len(failure_pred)})")149 150 # Calculate estimated time to failure151 time_to_failure = np.maximum(0, avg_tool_wear_at_failure - tool_wear_values)152 153 # Determine maintenance urgency154 maintenance_status = []155 maintenance_urgency = []156 157 for i in range(len(failure_pred)):158 # Check if tool wear has exceeded the average failure threshold159 tool_wear_exceeded = tool_wear_values[i] >= avg_tool_wear_at_failure160 tool_wear_severely_exceeded = tool_wear_values[i] >= (avg_tool_wear_at_failure * 1.5) # 150% of threshold (180 min)161 162 # Combined risk assessment: consider both ML prediction AND tool wear163 if failure_pred[i] == 1:164 # Model predicts failure - highest priority165 maintenance_status.append("IMMEDIATE MAINTENANCE REQUIRED")166 maintenance_urgency.append("CRITICAL")167 elif tool_wear_severely_exceeded and failure_proba[i] > 0.2:168 # Severely exceeded tool wear AND significant failure probability169 maintenance_status.append("IMMEDIATE MAINTENANCE REQUIRED - Tool wear severely exceeded threshold")170 maintenance_urgency.append("CRITICAL")171 elif tool_wear_severely_exceeded:172 # Severely exceeded tool wear but low failure probability - still HIGH priority173 maintenance_status.append("URGENT MAINTENANCE NEEDED - Tool wear severely exceeded (monitor closely)")174 maintenance_urgency.append("HIGH")175 elif tool_wear_exceeded and failure_proba[i] > 0.3:176 # Tool wear exceeded AND moderate failure probability177 maintenance_status.append("IMMEDIATE MAINTENANCE REQUIRED - Tool wear exceeded threshold")178 maintenance_urgency.append("CRITICAL")179 elif tool_wear_exceeded:180 # Tool wear exceeded but low failure probability - HIGH priority181 maintenance_status.append("URGENT MAINTENANCE NEEDED - Tool wear exceeded threshold")182 maintenance_urgency.append("HIGH")183 elif failure_proba[i] > 0.7:184 # High failure probability regardless of tool wear185 maintenance_status.append("Maintenance needed soon")186 maintenance_urgency.append("HIGH")187 elif failure_proba[i] > 0.4:188 # Moderate failure probability189 maintenance_status.append("Schedule maintenance")190 maintenance_urgency.append("MEDIUM")191 elif time_to_failure[i] < 20:192 # Less than 20 minutes remaining193 maintenance_status.append("Monitor closely - Maintenance needed within 20 minutes")194 maintenance_urgency.append("MEDIUM")195 elif time_to_failure[i] < 60:196 # Less than 1 hour remaining197 maintenance_status.append("Monitor closely - Maintenance needed within 1 hour")198 maintenance_urgency.append("MEDIUM")199 else:200 # Low risk201 maintenance_status.append("No immediate maintenance needed")202 maintenance_urgency.append("LOW")203 204 results_df = pd.DataFrame({205 'Failure_Predicted': failure_pred,206 'Failure_Probability': failure_proba,207 'Time_to_Failure_Minutes': time_to_failure,208 'Maintenance_Status': maintenance_status,209 'Maintenance_Urgency': maintenance_urgency210 })211 212 return results_df213 214 215 def get_feature_importance(self):216 """Get feature importance"""217 if self.feature_importance_ is None:218 raise ValueError("Model must be trained first!")219 return self.feature_importance_220 221 def save_model(self, filepath='predictive_maintenance_model.pkl'):222 """Save the trained model"""223 if not self.is_trained:224 raise ValueError("Model must be trained first!")225 226 with open(filepath, 'wb') as f:227 pickle.dump(self.model, f)228 print(f"Model saved to {filepath}")229 230 def load_model(self, filepath='predictive_maintenance_model.pkl'):231 """Load a trained model"""232 with open(filepath, 'rb') as f:233 self.model = pickle.load(f)234 self.is_trained = True235 print(f"Model loaded from {filepath}")