sample_eeg_generator.py

import numpy as np
from scipy import signal
import pandas as pd

class SampleEEGGenerator:
    """
    Generates research-grade sample EEG data for demonstration when authentic datasets are unavailable.
    Uses established neuroscience parameters and realistic signal characteristics.
    """
    
    def __init__(self):
        # Standard EEG frequency bands based on international 10-20 system
        self.frequency_bands = {
            'delta': (0.5, 4),    # Deep sleep, unconscious states
            'theta': (4, 8),      # Light sleep, meditation, memory
            'alpha': (8, 13),     # Relaxed wakefulness, eyes closed
            'beta': (13, 30),     # Active thinking, concentration
            'gamma': (30, 100)    # High-level cognitive processing
        }
        
        # Standard EEG electrode positions (10-20 system)
        self.electrode_positions = {
            'Fp1': {'region': 'frontal', 'hemisphere': 'left'},
            'Fp2': {'region': 'frontal', 'hemisphere': 'right'},
            'F3': {'region': 'frontal', 'hemisphere': 'left'},
            'F4': {'region': 'frontal', 'hemisphere': 'right'},
            'C3': {'region': 'central', 'hemisphere': 'left'},
            'C4': {'region': 'central', 'hemisphere': 'right'},
            'P3': {'region': 'parietal', 'hemisphere': 'left'},
            'P4': {'region': 'parietal', 'hemisphere': 'right'},
            'O1': {'region': 'occipital', 'hemisphere': 'left'},
            'O2': {'region': 'occipital', 'hemisphere': 'right'},
            'F7': {'region': 'temporal', 'hemisphere': 'left'},
            'F8': {'region': 'temporal', 'hemisphere': 'right'},
            'T3': {'region': 'temporal', 'hemisphere': 'left'},
            'T4': {'region': 'temporal', 'hemisphere': 'right'},
            'T5': {'region': 'temporal', 'hemisphere': 'left'},
            'T6': {'region': 'temporal', 'hemisphere': 'right'}
        }
    
    def generate_speech_eeg(self, duration=300, sampling_rate=250, num_channels=16):
        """
        Generate sample EEG data simulating speech comprehension experiments.
        Based on known neural responses to auditory language processing.
        
        Args:
            duration: Recording duration in seconds
            sampling_rate: EEG sampling rate in Hz
            num_channels: Number of EEG channels
            
        Returns:
            EEG data array (samples x channels), channel names, experimental info
        """
        num_samples = int(duration * sampling_rate)
        t = np.linspace(0, duration, num_samples)
        
        # Initialize EEG data
        eeg_data = np.zeros((num_samples, num_channels))
        
        # Get channel names (use first N electrodes)
        channel_names = list(self.electrode_positions.keys())[:num_channels]
        
        # Generate realistic EEG for each channel
        for i, channel in enumerate(channel_names):
            channel_info = self.electrode_positions[channel]
            region = channel_info['region']
            hemisphere = channel_info['hemisphere']
            
            # Base EEG signal for this channel
            channel_signal = self._generate_channel_signal(t, region, hemisphere, sampling_rate)
            
            # Add speech-specific responses
            speech_events = self._generate_speech_events(t, sampling_rate)
            channel_signal += speech_events
            
            # Add region-specific characteristics
            if region == 'temporal':
                # Temporal regions show strong auditory responses
                auditory_response = self._generate_auditory_response(t, sampling_rate)
                channel_signal += auditory_response * 1.5
            elif region == 'frontal':
                # Frontal regions show language processing
                language_response = self._generate_language_response(t, sampling_rate)
                channel_signal += language_response
            
            # Add realistic noise and artifacts
            noise = np.random.normal(0, 8, len(t))  # ~8μV noise
            channel_signal += noise
            
            # Occasional artifacts (eye blinks, muscle activity)
            artifacts = self._add_realistic_artifacts(t, sampling_rate)
            channel_signal += artifacts
            
            eeg_data[:, i] = channel_signal
        
        # Create experimental metadata
        exp_info = {
            'experiment_type': 'speech_comprehension',
            'sampling_rate': sampling_rate,
            'duration': duration,
            'num_channels': num_channels,
            'channel_names': channel_names,
            'paradigm': 'auditory_sentences',
            'subject_condition': 'healthy_adult',
            'data_type': 'research_simulation'
        }
        
        return eeg_data, channel_names, exp_info
    
    def _generate_channel_signal(self, t, region, hemisphere, sampling_rate):
        """Generate base EEG signal for a specific brain region."""
        signal_data = np.zeros_like(t)
        
        # Each region has characteristic frequency profiles
        if region == 'occipital':
            # Strong alpha rhythm (8-13 Hz) in visual cortex
            alpha_freq = np.random.uniform(9, 11)
            signal_data += 25 * np.sin(2 * np.pi * alpha_freq * t)
            
        elif region == 'frontal':
            # Beta activity during cognitive tasks
            beta_freq = np.random.uniform(15, 25)
            signal_data += 15 * np.sin(2 * np.pi * beta_freq * t)
            
        elif region == 'temporal':
            # Mixed alpha and theta for auditory processing
            theta_freq = np.random.uniform(5, 7)
            alpha_freq = np.random.uniform(8, 10)
            signal_data += 20 * np.sin(2 * np.pi * theta_freq * t)
            signal_data += 15 * np.sin(2 * np.pi * alpha_freq * t)
            
        elif region == 'central':
            # Mu rhythm (8-12 Hz) in sensorimotor areas
            mu_freq = np.random.uniform(9, 11)
            signal_data += 18 * np.sin(2 * np.pi * mu_freq * t)
            
        elif region == 'parietal':
            # Alpha and beta mix
            alpha_freq = np.random.uniform(8, 12)
            beta_freq = np.random.uniform(14, 20)
            signal_data += 20 * np.sin(2 * np.pi * alpha_freq * t)
            signal_data += 12 * np.sin(2 * np.pi * beta_freq * t)
        
        # Add phase variations and amplitude modulation
        phase_noise = np.random.uniform(0, 2*np.pi, 1)
        signal_data = signal_data * (1 + 0.1 * np.sin(2 * np.pi * 0.1 * t + phase_noise))
        
        return signal_data
    
    def _generate_speech_events(self, t, sampling_rate):
        """Generate event-related potentials for speech stimuli."""
        speech_signal = np.zeros_like(t)
        
        # Speech events every 2-3 seconds (realistic sentence presentation)
        event_interval = 2.5  # seconds
        num_events = int(len(t) / (event_interval * sampling_rate))
        
        for i in range(num_events):
            event_time = i * event_interval
            event_idx = int(event_time * sampling_rate)
            
            if event_idx < len(t) - 500:  # Ensure we don't go past array bounds
                # N400 component (semantic processing) around 400ms
                n400_start = event_idx + int(0.3 * sampling_rate)
                n400_end = event_idx + int(0.5 * sampling_rate)
                if n400_end < len(speech_signal):
                    n400_amplitude = np.random.uniform(-15, -25)  # Negative deflection
                    n400_time = np.linspace(0, 0.2, n400_end - n400_start)
                    n400_wave = n400_amplitude * np.exp(-n400_time * 5) * np.sin(2 * np.pi * 3 * n400_time)
                    speech_signal[n400_start:n400_end] += n400_wave
                
                # P600 component (syntactic processing) around 600ms
                p600_start = event_idx + int(0.55 * sampling_rate)
                p600_end = event_idx + int(0.75 * sampling_rate)
                if p600_end < len(speech_signal):
                    p600_amplitude = np.random.uniform(10, 20)  # Positive deflection
                    p600_time = np.linspace(0, 0.2, p600_end - p600_start)
                    p600_wave = p600_amplitude * np.exp(-p600_time * 4) * np.sin(2 * np.pi * 2 * p600_time)
                    speech_signal[p600_start:p600_end] += p600_wave
        
        return speech_signal
    
    def _generate_auditory_response(self, t, sampling_rate):
        """Generate auditory evoked responses."""
        auditory_signal = np.zeros_like(t)
        
        # Auditory steady-state responses at ~40Hz (gamma)
        gamma_freq = 40
        gamma_amplitude = 8
        auditory_signal += gamma_amplitude * np.sin(2 * np.pi * gamma_freq * t)
        
        # Modulate with slower rhythm (attention/engagement)
        attention_freq = 0.5
        modulation = 1 + 0.3 * np.sin(2 * np.pi * attention_freq * t)
        auditory_signal *= modulation
        
        return auditory_signal
    
    def _generate_language_response(self, t, sampling_rate):
        """Generate language-specific neural responses."""
        language_signal = np.zeros_like(t)
        
        # Beta-gamma coupling for language processing
        beta_freq = 20
        gamma_freq = 60
        
        beta_component = 12 * np.sin(2 * np.pi * beta_freq * t)
        gamma_component = 6 * np.sin(2 * np.pi * gamma_freq * t)
        
        # Cross-frequency coupling
        coupling = beta_component + gamma_component * (1 + 0.5 * beta_component)
        language_signal += coupling
        
        return language_signal
    
    def _add_realistic_artifacts(self, t, sampling_rate):
        """Add realistic EEG artifacts."""
        artifacts = np.zeros_like(t)
        
        # Eye blinks (~0.5 Hz, large amplitude)
        blink_times = np.random.poisson(0.3, len(t))  # ~0.3 blinks per sample
        for i, has_blink in enumerate(blink_times):
            if has_blink and i < len(t) - 100:
                blink_duration = int(0.2 * sampling_rate)  # 200ms blink
                blink_end = min(i + blink_duration, len(t))
                blink_amplitude = np.random.uniform(50, 150)  # Large artifact
                blink_shape = np.exp(-np.linspace(0, 5, blink_end - i))
                artifacts[i:blink_end] += blink_amplitude * blink_shape
        
        # Muscle artifacts (high frequency, random)
        muscle_noise = np.random.normal(0, 3, len(t))
        # High-pass filter to simulate EMG
        b, a = signal.butter(4, 20/(sampling_rate/2), 'high')
        muscle_filtered = signal.filtfilt(b, a, muscle_noise)
        artifacts += muscle_filtered * 0.5
        
        return artifacts
    
    def create_sample_dataset(self, num_subjects=5, duration_per_subject=300):
        """Create a complete sample dataset with multiple subjects."""
        dataset = []
        
        for subject_id in range(1, num_subjects + 1):
            # Generate slightly different parameters per subject
            sampling_rate = np.random.choice([250, 256, 500])
            num_channels = np.random.choice([16, 32, 64])
            
            eeg_data, channel_names, exp_info = self.generate_speech_eeg(
                duration=duration_per_subject,
                sampling_rate=sampling_rate,
                num_channels=num_channels
            )
            
            exp_info['subject_id'] = f'S{subject_id:03d}'
            exp_info['age'] = np.random.randint(20, 65)
            exp_info['handedness'] = np.random.choice(['right', 'left'], p=[0.9, 0.1])
            
            dataset.append({
                'subject_id': exp_info['subject_id'],
                'eeg_data': eeg_data,
                'channel_names': channel_names,
                'exp_info': exp_info
            })
        
        return dataset