# 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