""" Comprehensive Exploratory Data Analysis (EDA) for AI4I 2020 Predictive Maintenance Dataset This script performs 10-15 different analyses as required for the project. """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import warnings warnings.filterwarnings('ignore') # Set style for better visualizations sns.set_style("whitegrid") plt.rcParams['figure.figsize'] = (12, 6) class EDAAnalysis: def __init__(self, data_path='ai4i2020.csv'): """Initialize the EDA analysis class""" self.df = pd.read_csv(data_path) self.prepare_data() def prepare_data(self): """Prepare data for analysis""" # Create a copy for analysis self.df_clean = self.df.copy() # Calculate temperature difference (important feature) self.df_clean['Temperature difference [K]'] = ( self.df_clean['Process temperature [K]'] - self.df_clean['Air temperature [K]'] ) # Calculate power (Rotational speed * Torque) self.df_clean['Power [W]'] = ( self.df_clean['Rotational speed [rpm]'] * self.df_clean['Torque [Nm]'] / 9.5488 # Conversion factor ) # Failure type names self.failure_types = { 'TWF': 'Tool Wear Failure', 'HDF': 'Heat Dissipation Failure', 'PWF': 'Power Failure', 'OSF': 'Overstrain Failure', 'RNF': 'Random Failure' } def analysis_1_summary_statistics(self): """Analysis 1: Summary statistics for numerical features""" print("="*80) print("ANALYSIS 1: SUMMARY STATISTICS") print("="*80) numerical_cols = [ 'Air temperature [K]', 'Process temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]', 'Tool wear [min]', 'Temperature difference [K]', 'Power [W]' ] summary = self.df_clean[numerical_cols].describe() print(summary) print("\n") return summary def analysis_2_missing_values(self): """Analysis 2: Missing value analysis""" print("="*80) print("ANALYSIS 2: MISSING VALUE ANALYSIS") print("="*80) missing = self.df_clean.isnull().sum() missing_pct = (missing / len(self.df_clean)) * 100 missing_df = pd.DataFrame({ 'Missing Count': missing, 'Missing Percentage': missing_pct }) missing_df = missing_df[missing_df['Missing Count'] > 0] if len(missing_df) == 0: print("✓ No missing values found in the dataset!") else: print(missing_df) print("\n") return missing_df def analysis_3_data_types(self): """Analysis 3: Data types and unique value counts""" print("="*80) print("ANALYSIS 3: DATA TYPES AND UNIQUE VALUES") print("="*80) info_df = pd.DataFrame({ 'Column': self.df_clean.columns, 'Data Type': self.df_clean.dtypes, 'Unique Values': [self.df_clean[col].nunique() for col in self.df_clean.columns], 'Non-Null Count': self.df_clean.count().values }) print(info_df.to_string(index=False)) print("\n") # Categorical value counts print("Type Distribution:") print(self.df_clean['Type'].value_counts()) print("\n") return info_df def analysis_4_target_distribution(self): """Analysis 4: Target variable (Machine failure) distribution""" print("="*80) print("ANALYSIS 4: TARGET VARIABLE DISTRIBUTION") print("="*80) failure_counts = self.df_clean['Machine failure'].value_counts() failure_pct = self.df_clean['Machine failure'].value_counts(normalize=True) * 100 print("Machine Failure Distribution:") print(f" No Failure (0): {failure_counts[0]} ({failure_pct[0]:.2f}%)") print(f" Failure (1): {failure_counts[1]} ({failure_pct[1]:.2f}%)") print("\n") # Failure types breakdown print("Failure Types Breakdown:") for ft in ['TWF', 'HDF', 'PWF', 'OSF', 'RNF']: count = self.df_clean[ft].sum() pct = (count / len(self.df_clean)) * 100 print(f" {self.failure_types[ft]}: {count} ({pct:.2f}%)") print("\n") return failure_counts def analysis_5_correlation_analysis(self): """Analysis 5: Correlation analysis""" print("="*80) print("ANALYSIS 5: CORRELATION ANALYSIS") print("="*80) numerical_cols = [ 'Air temperature [K]', 'Process temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]', 'Tool wear [min]', 'Temperature difference [K]', 'Power [W]', 'Machine failure' ] corr_matrix = self.df_clean[numerical_cols].corr() print("Correlation with Machine Failure:") failure_corr = corr_matrix['Machine failure'].sort_values(ascending=False) print(failure_corr) print("\n") return corr_matrix def analysis_6_outlier_detection(self): """Analysis 6: Outlier detection using IQR method""" print("="*80) print("ANALYSIS 6: OUTLIER DETECTION (IQR METHOD)") print("="*80) numerical_cols = [ 'Air temperature [K]', 'Process temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]', 'Tool wear [min]' ] outlier_summary = {} for col in numerical_cols: Q1 = self.df_clean[col].quantile(0.25) Q3 = self.df_clean[col].quantile(0.75) IQR = Q3 - Q1 lower_bound = Q1 - 1.5 * IQR upper_bound = Q3 + 1.5 * IQR outliers = self.df_clean[(self.df_clean[col] < lower_bound) | (self.df_clean[col] > upper_bound)] outlier_count = len(outliers) outlier_pct = (outlier_count / len(self.df_clean)) * 100 outlier_summary[col] = { 'Lower Bound': lower_bound, 'Upper Bound': upper_bound, 'Outlier Count': outlier_count, 'Outlier Percentage': outlier_pct } print(f"{col}:") print(f" Bounds: [{lower_bound:.2f}, {upper_bound:.2f}]") print(f" Outliers: {outlier_count} ({outlier_pct:.2f}%)") print("\n") return outlier_summary def analysis_7_feature_distributions(self): """Analysis 7: Feature distribution analysis""" print("="*80) print("ANALYSIS 7: FEATURE DISTRIBUTION ANALYSIS") print("="*80) numerical_cols = [ 'Air temperature [K]', 'Process temperature [K]', 'Rotational speed [rpm]', 'Torque [Nm]', 'Tool wear [min]' ] dist_stats = {} for col in numerical_cols: dist_stats[col] = { 'Mean': self.df_clean[col].mean(), 'Median': self.df_clean[col].median(), 'Mode': self.df_clean[col].mode()[0] if len(self.df_clean[col].mode()) > 0 else None, 'Std': self.df_clean[col].std(), 'Skewness': self.df_clean[col].skew(), 'Kurtosis': self.df_clean[col].kurtosis() } print(f"{col}:") print(f" Mean: {dist_stats[col]['Mean']:.2f}") print(f" Median: {dist_stats[col]['Median']:.2f}") print(f" Std Dev: {dist_stats[col]['Std']:.2f}") print(f" Skewness: {dist_stats[col]['Skewness']:.2f}") print(f" Kurtosis: {dist_stats[col]['Kurtosis']:.2f}") print() print("\n") return dist_stats def analysis_8_failure_by_type(self): """Analysis 8: Failure analysis by machine type""" print("="*80) print("ANALYSIS 8: FAILURE ANALYSIS BY MACHINE TYPE") print("="*80) failure_by_type = self.df_clean.groupby('Type')['Machine failure'].agg([ 'count', 'sum', 'mean' ]).round(4) failure_by_type.columns = ['Total Machines', 'Failures', 'Failure Rate'] failure_by_type['Failure Rate'] = failure_by_type['Failure Rate'] * 100 print(failure_by_type) print("\n") return failure_by_type def analysis_9_tool_wear_analysis(self): """Analysis 9: Tool wear analysis and relationship with failures""" print("="*80) print("ANALYSIS 9: TOOL WEAR ANALYSIS") print("="*80) print("Tool Wear Statistics:") print(f" Mean: {self.df_clean['Tool wear [min]'].mean():.2f} minutes") print(f" Median: {self.df_clean['Tool wear [min]'].median():.2f} minutes") print(f" Max: {self.df_clean['Tool wear [min]'].max():.2f} minutes") print(f" Min: {self.df_clean['Tool wear [min]'].min():.2f} minutes") print() # Tool wear vs failure tool_wear_failure = self.df_clean.groupby('Machine failure')['Tool wear [min]'].agg([ 'mean', 'median', 'std', 'min', 'max' ]) print("Tool Wear by Failure Status:") print(tool_wear_failure) print("\n") return tool_wear_failure def analysis_10_temperature_analysis(self): """Analysis 10: Temperature analysis""" print("="*80) print("ANALYSIS 10: TEMPERATURE ANALYSIS") print("="*80) temp_stats = self.df_clean.groupby('Machine failure')[ ['Air temperature [K]', 'Process temperature [K]', 'Temperature difference [K]'] ].agg(['mean', 'std']) print("Temperature Statistics by Failure Status:") print(temp_stats) print("\n") return temp_stats def analysis_11_power_analysis(self): """Analysis 11: Power and rotational speed analysis""" print("="*80) print("ANALYSIS 11: POWER AND ROTATIONAL SPEED ANALYSIS") print("="*80) power_stats = self.df_clean.groupby('Machine failure')[ ['Rotational speed [rpm]', 'Torque [Nm]', 'Power [W]'] ].agg(['mean', 'std', 'min', 'max']) print("Power Statistics by Failure Status:") print(power_stats) print("\n") return power_stats def analysis_12_pairwise_relationships(self): """Analysis 12: Pairwise feature relationships""" print("="*80) print("ANALYSIS 12: PAIRWISE FEATURE RELATIONSHIPS") print("="*80) key_features = [ 'Tool wear [min]', 'Temperature difference [K]', 'Rotational speed [rpm]', 'Torque [Nm]', 'Machine failure' ] pairwise_corr = self.df_clean[key_features].corr() print("Pairwise Correlations:") print(pairwise_corr) print("\n") return pairwise_corr def analysis_13_failure_type_analysis(self): """Analysis 13: Detailed failure type analysis""" print("="*80) print("ANALYSIS 13: DETAILED FAILURE TYPE ANALYSIS") print("="*80) failure_types = ['TWF', 'HDF', 'PWF', 'OSF', 'RNF'] for ft in failure_types: failed_machines = self.df_clean[self.df_clean[ft] == 1] if len(failed_machines) > 0: print(f"\n{self.failure_types[ft]} ({ft}):") print(f" Count: {len(failed_machines)}") print(f" Avg Tool Wear: {failed_machines['Tool wear [min]'].mean():.2f} min") print(f" Avg Temp Diff: {failed_machines['Temperature difference [K]'].mean():.2f} K") print(f" Avg Rotational Speed: {failed_machines['Rotational speed [rpm]'].mean():.2f} rpm") print(f" Avg Torque: {failed_machines['Torque [Nm]'].mean():.2f} Nm") print("\n") def analysis_14_time_to_failure_estimation(self): """Analysis 14: Time to failure estimation based on tool wear""" print("="*80) print("ANALYSIS 14: TIME TO FAILURE ESTIMATION") print("="*80) # Analyze tool wear progression for machines that failed failed_machines = self.df_clean[self.df_clean['Machine failure'] == 1] if len(failed_machines) > 0: avg_tool_wear_at_failure = failed_machines['Tool wear [min]'].mean() median_tool_wear_at_failure = failed_machines['Tool wear [min]'].median() print(f"Average Tool Wear at Failure: {avg_tool_wear_at_failure:.2f} minutes") print(f"Median Tool Wear at Failure: {median_tool_wear_at_failure:.2f} minutes") print() # Estimate time remaining for machines not yet failed non_failed = self.df_clean[self.df_clean['Machine failure'] == 0] if len(non_failed) > 0: non_failed['Estimated Time to Failure'] = ( avg_tool_wear_at_failure - non_failed['Tool wear [min]'] ) non_failed['Estimated Time to Failure'] = non_failed['Estimated Time to Failure'].clip(lower=0) print("Time to Failure Estimates (for non-failed machines):") print(f" Machines needing immediate maintenance (< 10 min): " f"{(non_failed['Estimated Time to Failure'] < 10).sum()}") print(f" Machines needing maintenance soon (10-50 min): " f"{((non_failed['Estimated Time to Failure'] >= 10) & (non_failed['Estimated Time to Failure'] < 50)).sum()}") print(f" Machines with time remaining (> 50 min): " f"{(non_failed['Estimated Time to Failure'] >= 50).sum()}") print("\n") def analysis_15_grouped_aggregations(self): """Analysis 15: Grouped aggregations by type and failure status""" print("="*80) print("ANALYSIS 15: GROUPED AGGREGATIONS") print("="*80) grouped = self.df_clean.groupby(['Type', 'Machine failure']).agg({ 'Tool wear [min]': ['mean', 'std', 'max'], 'Temperature difference [K]': ['mean', 'std'], 'Rotational speed [rpm]': ['mean', 'std'], 'Torque [Nm]': ['mean', 'std'] }) print("Grouped Statistics by Type and Failure Status:") print(grouped) print("\n") return grouped def run_all_analyses(self): """Run all EDA analyses""" print("\n" + "="*80) print("COMPREHENSIVE EXPLORATORY DATA ANALYSIS") print("AI4I 2020 Predictive Maintenance Dataset") print("="*80 + "\n") results = {} results['summary_stats'] = self.analysis_1_summary_statistics() results['missing_values'] = self.analysis_2_missing_values() results['data_types'] = self.analysis_3_data_types() results['target_distribution'] = self.analysis_4_target_distribution() results['correlation'] = self.analysis_5_correlation_analysis() results['outliers'] = self.analysis_6_outlier_detection() results['distributions'] = self.analysis_7_feature_distributions() results['failure_by_type'] = self.analysis_8_failure_by_type() results['tool_wear'] = self.analysis_9_tool_wear_analysis() results['temperature'] = self.analysis_10_temperature_analysis() results['power'] = self.analysis_11_power_analysis() results['pairwise'] = self.analysis_12_pairwise_relationships() self.analysis_13_failure_type_analysis() self.analysis_14_time_to_failure_estimation() results['grouped'] = self.analysis_15_grouped_aggregations() print("="*80) print("EDA ANALYSIS COMPLETE!") print("="*80) return results if __name__ == "__main__": # Run EDA analysis eda = EDAAnalysis('ai4i2020.csv') results = eda.run_all_analyses() # Save processed data for modeling eda.df_clean.to_csv('processed_data.csv', index=False) print("\nProcessed data saved to 'processed_data.csv'")