Create app.py

app.py

@@ -0,0 +1,284 @@
+# Import required libraries
+import gradio as gr
+import numpy as np
+import mne
+import matplotlib.pyplot as plt
+from scipy import signal
+from scipy.stats import skew, kurtosis
+import pandas as pd
+from pathlib import Path
+import tempfile
+import os
+from huggingface_hub import HfApi, HfFolder
+import warnings
+import logging
+
+# Configure logging system
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Initialize Hugging Face API with token from environment variable
+def initialize_hf_api():
+    try:
+        hf_token = os.getenv('HF_TOKEN')
+        if hf_token:
+            hf_api = HfApi(token=hf_token)
+            HfFolder.save_token(hf_token)
+            logger.info("Successfully initialized Hugging Face API")
+            return hf_api
+        else:
+            logger.warning("HF_TOKEN not found in environment variables")
+            return None
+    except Exception as e:
+        logger.error(f"Error initializing Hugging Face API: {str(e)}")
+        return None
+
+class EEGProcessor:
+    """Class for processing and analyzing EEG signals"""
+    
+    def __init__(self):
+        """Initialize the EEG processor with default parameters"""
+        self.sampling_rate = 250  # Default sampling rate in Hz
+        self.freq_bands = {
+            'delta': (0.5, 4),   # Delta band (0.5-4 Hz)
+            'theta': (4, 8),     # Theta band (4-8 Hz)
+            'alpha': (8, 13),    # Alpha band (8-13 Hz)
+            'beta': (13, 30),    # Beta band (13-30 Hz)
+            'gamma': (30, 50)    # Gamma band (30-50 Hz)
+        }
+
+    def load_eeg(self, file_path):
+        """
+        Load and validate EEG data from file
+        Args:
+            file_path: Path to the EEG file
+        Returns:
+            mne.io.Raw: Loaded EEG data
+        """
+        try:
+            # Validate file existence and size
+            if not os.path.exists(file_path):
+                raise ValueError("File does not exist")
+            
+            # Check file size (100MB limit)
+            if os.path.getsize(file_path) > 100 * 1024 * 1024:
+                raise ValueError("File size exceeds 100MB limit")
+
+            # Load EEG data using MNE
+            raw = mne.io.read_raw_edf(file_path, preload=True)
+            self.sampling_rate = raw.info['sfreq']
+            logger.info(f"Successfully loaded EEG file: {file_path}")
+            return raw
+            
+        except Exception as e:
+            logger.error(f"Error loading EEG file: {str(e)}")
+            raise ValueError(f"Error loading EEG file: {str(e)}")
+
+    def preprocess_signal(self, raw):
+        """
+        Apply preprocessing steps to the EEG signal
+        Args:
+            raw: Raw EEG data
+        Returns:
+            mne.io.Raw: Preprocessed EEG data
+        """
+        try:
+            logger.info("Starting signal preprocessing")
+            
+            # Apply bandpass filter (0.5-50 Hz)
+            raw.filter(l_freq=0.5, h_freq=50., fir_design='firwin')
+            logger.info("Applied bandpass filter")
+            
+            # Remove power line interference
+            raw.notch_filter(freqs=[50, 60])
+            logger.info("Applied notch filter")
+            
+            return raw
+            
+        except Exception as e:
+            logger.error(f"Error in signal preprocessing: {str(e)}")
+            raise
+
+    def extract_features(self, raw):
+        """
+        Extract time and frequency domain features from EEG data
+        Args:
+            raw: Preprocessed EEG data
+        Returns:
+            dict: Extracted features
+        """
+        try:
+            logger.info("Starting feature extraction")
+            data = raw.get_data()
+            features = {}
+            
+            # Calculate time domain features
+            features['mean'] = np.mean(data, axis=1)
+            features['variance'] = np.var(data, axis=1)
+            features['skewness'] = skew(data, axis=1)
+            features['kurtosis'] = kurtosis(data, axis=1)
+            
+            # Calculate frequency domain features
+            for band_name, (low_freq, high_freq) in self.freq_bands.items():
+                band_power = self._calculate_band_power(data, low_freq, high_freq)
+                features[f'{band_name}_power'] = band_power
+            
+            logger.info("Feature extraction completed successfully")
+            return features
+            
+        except Exception as e:
+            logger.error(f"Error in feature extraction: {str(e)}")
+            raise
+
+    def _calculate_band_power(self, data, low_freq, high_freq):
+        """
+        Calculate power in specific frequency band
+        Args:
+            data: EEG data
+            low_freq: Lower frequency bound
+            high_freq: Upper frequency bound
+        Returns:
+            float: Band power value
+        """
+        try:
+            # Calculate power spectral density
+            freqs, psd = signal.welch(data, fs=self.sampling_rate)
+            
+            # Find frequencies within band
+            idx = np.logical_and(freqs >= low_freq, freqs <= high_freq)
+            
+            # Calculate band power using trapezoidal integration
+            band_power = np.trapz(psd[:, idx], freqs[idx])
+            return band_power
+            
+        except Exception as e:
+            logger.error(f"Error calculating band power: {str(e)}")
+            raise
+
+def create_visualization(raw, features):
+    """
+    Create visualization plots for EEG analysis
+    Args:
+        raw: EEG data
+        features: Extracted features
+    Returns:
+        matplotlib.figure.Figure: Figure containing plots
+    """
+    try:
+        # Create figure with three subplots
+        fig, (ax1, ax2, ax3) = plt.subplots(3, 1, figsize=(12, 15))
+        
+        # Plot 1: Raw EEG signal
+        data = raw.get_data()
+        times = np.arange(data.shape[1]) / raw.info['sfreq']
+        ax1.plot(times, data.T)
+        ax1.set_title('Raw EEG Signal')
+        ax1.set_xlabel('Time (s)')
+        ax1.set_ylabel('Amplitude')
+        
+        # Plot 2: Power spectrum
+        freqs, psd = signal.welch(data, fs=raw.info['sfreq'])
+        ax2.semilogy(freqs, psd.T)
+        ax2.set_title('Power Spectrum')
+        ax2.set_xlabel('Frequency (Hz)')
+        ax2.set_ylabel('Power Spectral Density')
+        
+        # Plot 3: Band powers
+        band_powers = {k: v for k, v in features.items() if 'power' in k}
+        ax3.bar(band_powers.keys(), band_powers.values())
+        ax3.set_title('Band Powers')
+        ax3.set_xlabel('Frequency Band')
+        ax3.set_ylabel('Power')
+        
+        plt.tight_layout()
+        return fig
+        
+    except Exception as e:
+        logger.error(f"Error creating visualization: {str(e)}")
+        raise
+
+def process_eeg(file):
+    """
+    Main processing function for the Gradio interface
+    Args:
+        file: Uploaded file object
+    Returns:
+        tuple: (matplotlib figure, feature analysis string)
+    """
+    try:
+        logger.info("Starting EEG processing")
+        
+        # Create temporary file for processing
+        with tempfile.NamedTemporaryFile(suffix='.edf', delete=False) as tmp_file:
+            tmp_file.write(file.read())
+            tmp_path = tmp_file.name
+        
+        logger.info("Temporary file created")
+        
+        # Initialize processor and process EEG
+        processor = EEGProcessor()
+        raw = processor.load_eeg(tmp_path)
+        raw = processor.preprocess_signal(raw)
+        features = processor.extract_features(raw)
+        
+        # Create visualizations and feature summary
+        fig = create_visualization(raw, features)
+        feature_df = pd.DataFrame({k: [v] for k, v in features.items()})
+        
+        # Clean up temporary file
+        os.unlink(tmp_path)
+        logger.info("Processing completed successfully")
+        
+        return fig, feature_df.to_string()
+        
+    except Exception as e:
+        logger.error(f"Error in EEG processing: {str(e)}")
+        raise gr.Error(f"Error processing EEG: {str(e)}")
+
+# Create Gradio interface
+def create_interface():
+    """Create and configure the Gradio interface"""
+    try:
+        # Initialize Hugging Face API
+        initialize_hf_api()
+        
+        # Create Gradio blocks interface
+        with gr.Blocks(title="EEG Signal Analysis") as iface:
+            gr.Markdown("# EEG Signal Analysis Tool")
+            gr.Markdown("Upload an EEG file (.edf format) for analysis")
+            
+            with gr.Row():
+                file_input = gr.File(
+                    label="Upload EEG File",
+                    file_types=[".edf"],
+                    type="binary"
+                )
+            
+            with gr.Row():
+                analyze_btn = gr.Button("Analyze EEG")
+            
+            with gr.Row():
+                plot_output = gr.Plot(label="EEG Analysis Plots")
+                feature_output = gr.Textbox(label="Feature Analysis", lines=10)
+            
+            # Set up button click event
+            analyze_btn.click(
+                fn=process_eeg,
+                inputs=[file_input],
+                outputs=[plot_output, feature_output]
+            )
+        
+        return iface
+        
+    except Exception as e:
+        logger.error(f"Error creating interface: {str(e)}")
+        raise
+
+# Launch the application
+if __name__ == "__main__":
+    try:
+        iface = create_interface()
+        iface.launch()
+    except Exception as e:
+        logger.error(f"Application startup failed: {str(e)}")
+        raise