File size: 9,648 Bytes
0b794cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 | """
Machine Learning Model for Predictive Maintenance
Uses Random Forest Classifier for failure prediction
"""
import pandas as pd
import numpy as np
import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import (
accuracy_score, precision_score, recall_score, f1_score,
confusion_matrix, classification_report, roc_auc_score
)
import warnings
warnings.filterwarnings('ignore')
class PredictiveMaintenanceModel:
def __init__(self):
"""Initialize the model"""
self.model = RandomForestClassifier(
n_estimators=100,
max_depth=10,
min_samples_split=5,
min_samples_leaf=2,
random_state=42,
class_weight='balanced' # Handle class imbalance
)
self.is_trained = False
self.feature_importance_ = None
def train(self, X_train, y_train):
"""Train the model"""
print("Training Random Forest Classifier...")
self.model.fit(X_train, y_train)
self.is_trained = True
# Get feature importance
self.feature_importance_ = pd.DataFrame({
'feature': X_train.columns,
'importance': self.model.feature_importances_
}).sort_values('importance', ascending=False)
print("Model training complete!")
return self
def predict(self, X):
"""Make predictions"""
if not self.is_trained:
raise ValueError("Model must be trained first!")
return self.model.predict(X)
def predict_proba(self, X):
"""Get prediction probabilities"""
if not self.is_trained:
raise ValueError("Model must be trained first!")
return self.model.predict_proba(X)
def evaluate(self, X_test, y_test):
"""Evaluate the model"""
if not self.is_trained:
raise ValueError("Model must be trained first!")
# Predictions
y_pred = self.predict(X_test)
y_pred_proba = self.predict_proba(X_test)[:, 1]
# Metrics
accuracy = accuracy_score(y_test, y_pred)
precision = precision_score(y_test, y_pred, zero_division=0)
recall = recall_score(y_test, y_pred, zero_division=0)
f1 = f1_score(y_test, y_pred, zero_division=0)
roc_auc = roc_auc_score(y_test, y_pred_proba)
# Confusion matrix
cm = confusion_matrix(y_test, y_pred)
# Classification report
report = classification_report(y_test, y_pred, zero_division=0)
results = {
'accuracy': accuracy,
'precision': precision,
'recall': recall,
'f1_score': f1,
'roc_auc': roc_auc,
'confusion_matrix': cm,
'classification_report': report,
'y_pred': y_pred,
'y_pred_proba': y_pred_proba
}
print("="*60)
print("MODEL EVALUATION RESULTS")
print("="*60)
print(f"Accuracy: {accuracy:.4f}")
print(f"Precision: {precision:.4f}")
print(f"Recall: {recall:.4f}")
print(f"F1-Score: {f1:.4f}")
print(f"ROC-AUC: {roc_auc:.4f}")
print("\nConfusion Matrix:")
print(cm)
print("\nClassification Report:")
print(report)
return results
def predict_maintenance(self, X, tool_wear_values=None):
"""
Predict maintenance needs and time to failure
Args:
X: Feature matrix
tool_wear_values: Array of tool wear values (optional)
Returns:
DataFrame with predictions and maintenance recommendations
"""
if not self.is_trained:
raise ValueError("Model must be trained first!")
# Get failure predictions
failure_pred = self.predict(X)
failure_proba = self.predict_proba(X)[:, 1]
# Estimate time to failure based on tool wear
# Average tool wear at failure is around 100-150 minutes based on EDA
avg_tool_wear_at_failure = 120 # minutes
if tool_wear_values is None:
# Try to get tool wear from features if available
if 'Tool wear [min]' in X.columns:
tool_wear_values = X['Tool wear [min]'].values
else:
tool_wear_values = np.zeros(len(X))
else:
# Convert to numpy array if it's a list or scalar
tool_wear_values = np.array(tool_wear_values)
# If it's a scalar, make it an array
if tool_wear_values.ndim == 0:
tool_wear_values = np.array([tool_wear_values])
# Ensure tool_wear_values is 1D array with same length as predictions
if len(tool_wear_values) != len(failure_pred):
# If single value provided, repeat it for all predictions
if len(tool_wear_values) == 1:
tool_wear_values = np.repeat(tool_wear_values, len(failure_pred))
else:
raise ValueError(f"tool_wear_values length ({len(tool_wear_values)}) doesn't match predictions length ({len(failure_pred)})")
# Calculate estimated time to failure
time_to_failure = np.maximum(0, avg_tool_wear_at_failure - tool_wear_values)
# Determine maintenance urgency
maintenance_status = []
maintenance_urgency = []
for i in range(len(failure_pred)):
# Check if tool wear has exceeded the average failure threshold
tool_wear_exceeded = tool_wear_values[i] >= avg_tool_wear_at_failure
tool_wear_severely_exceeded = tool_wear_values[i] >= (avg_tool_wear_at_failure * 1.5) # 150% of threshold (180 min)
# Combined risk assessment: consider both ML prediction AND tool wear
if failure_pred[i] == 1:
# Model predicts failure - highest priority
maintenance_status.append("IMMEDIATE MAINTENANCE REQUIRED")
maintenance_urgency.append("CRITICAL")
elif tool_wear_severely_exceeded and failure_proba[i] > 0.2:
# Severely exceeded tool wear AND significant failure probability
maintenance_status.append("IMMEDIATE MAINTENANCE REQUIRED - Tool wear severely exceeded threshold")
maintenance_urgency.append("CRITICAL")
elif tool_wear_severely_exceeded:
# Severely exceeded tool wear but low failure probability - still HIGH priority
maintenance_status.append("URGENT MAINTENANCE NEEDED - Tool wear severely exceeded (monitor closely)")
maintenance_urgency.append("HIGH")
elif tool_wear_exceeded and failure_proba[i] > 0.3:
# Tool wear exceeded AND moderate failure probability
maintenance_status.append("IMMEDIATE MAINTENANCE REQUIRED - Tool wear exceeded threshold")
maintenance_urgency.append("CRITICAL")
elif tool_wear_exceeded:
# Tool wear exceeded but low failure probability - HIGH priority
maintenance_status.append("URGENT MAINTENANCE NEEDED - Tool wear exceeded threshold")
maintenance_urgency.append("HIGH")
elif failure_proba[i] > 0.7:
# High failure probability regardless of tool wear
maintenance_status.append("Maintenance needed soon")
maintenance_urgency.append("HIGH")
elif failure_proba[i] > 0.4:
# Moderate failure probability
maintenance_status.append("Schedule maintenance")
maintenance_urgency.append("MEDIUM")
elif time_to_failure[i] < 20:
# Less than 20 minutes remaining
maintenance_status.append("Monitor closely - Maintenance needed within 20 minutes")
maintenance_urgency.append("MEDIUM")
elif time_to_failure[i] < 60:
# Less than 1 hour remaining
maintenance_status.append("Monitor closely - Maintenance needed within 1 hour")
maintenance_urgency.append("MEDIUM")
else:
# Low risk
maintenance_status.append("No immediate maintenance needed")
maintenance_urgency.append("LOW")
results_df = pd.DataFrame({
'Failure_Predicted': failure_pred,
'Failure_Probability': failure_proba,
'Time_to_Failure_Minutes': time_to_failure,
'Maintenance_Status': maintenance_status,
'Maintenance_Urgency': maintenance_urgency
})
return results_df
def get_feature_importance(self):
"""Get feature importance"""
if self.feature_importance_ is None:
raise ValueError("Model must be trained first!")
return self.feature_importance_
def save_model(self, filepath='predictive_maintenance_model.pkl'):
"""Save the trained model"""
if not self.is_trained:
raise ValueError("Model must be trained first!")
with open(filepath, 'wb') as f:
pickle.dump(self.model, f)
print(f"Model saved to {filepath}")
def load_model(self, filepath='predictive_maintenance_model.pkl'):
"""Load a trained model"""
with open(filepath, 'rb') as f:
self.model = pickle.load(f)
self.is_trained = True
print(f"Model loaded from {filepath}") |