analysis.py

"""

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'")