Module 3: EEG Data Collection - Workbook

def calculate_voltage_resolution(adc_bits, voltage_range): """ Calculate the smallest detectable voltage change Parameters: - adc_bits: ADC resolution in bits - voltage_range: total input range in volts (e.g., ±250mV = 0.5V total) """ # Number of discrete levels levels = 2 ** # Voltage per bit resolution_volts = voltage_range / # Convert to microvolts resolution_uv = resolution_volts * return resolution_uv # Examples for different systems consumer_res = calculate_voltage_resolution(16, 0.5) # 16-bit, ±250mV research_res = calculate_voltage_resolution( , 0.5) # 24-bit, ±250mV print(f"Consumer system resolution: {consumer_res:.3f} μV") print(f"Research system resolution: {research_res:.3f} μV") # What's the minimum ADC bits needed to detect 0.1 μV changes with ±250mV range? min_bits =

def select_hardware_settings(recording_type): """ Recommend hardware settings based on recording type """ settings = { 'sampling_rate': None, 'filter_settings': {}, 'electrode_type': None, 'reference': None } if recording_type == 'ERP': # Event-related potentials need good temporal resolution settings['sampling_rate'] = # Hz (minimum 250) settings['filter_settings'] = { 'highpass': , # Hz (very low for slow components) 'lowpass': 100, # Hz 'notch': None # Often skip notch for ERPs } settings['electrode_type'] = 'Ag/AgCl wet' settings['reference'] = ' ' # Common for ERPs elif recording_type == 'BCI': # Brain-computer interface - real-time processing settings['sampling_rate'] = # Hz (balance quality/speed) settings['filter_settings'] = { 'highpass': 0.5, 'lowpass': , # Hz (focus on control signals) 'notch': 60 # Remove interference } settings['electrode_type'] = ' ' # For quick setup settings['reference'] = 'CAR' # Common average reference elif recording_type == 'Sleep': # Long duration, full spectrum needed settings['sampling_rate'] = # Hz (minimum for all stages) settings['filter_settings'] = { 'highpass': 0.3, 'lowpass': 35, 'notch': 60 } settings['electrode_type'] = ' ' # Comfortable for long recordings return settings

import numpy as np import matplotlib.pyplot as plt class ImpedanceChecker: def __init__(self, electrode_names, target_impedance=5): self.electrodes = electrode_names self.target = target_impedance # kΩ self.history = {name: [] for name in electrode_names} def measure_impedance(self, amplifier_connection): """ Measure impedance for all electrodes """ impedances = {} for electrode in self.electrodes: # Send test signal (usually 10-30 Hz) test_freq = # Hz test_current = # nA (typical) # Measure voltage response # Z = V / I (Ohm's law) measured_voltage = amplifier_connection.inject_current( electrode, test_current, test_freq ) # Calculate impedance in kΩ impedance_ohms = measured_voltage / (test_current * 1e-9) impedance_kohms = impedance_ohms / impedances[electrode] = impedance_kohms self.history[electrode].append(impedance_kohms) return impedances def get_quality_indicators(self, impedances): """ Categorize impedance quality """ quality = {} for electrode, z in impedances.items(): if z < self.target: quality[electrode] = ' ' # Good elif z < self.target * 2: quality[electrode] = 'acceptable' elif z < self.target * 4: quality[electrode] = ' ' # Needs work else: quality[electrode] = 'poor' return quality def suggest_fixes(self, impedances): """ Suggest specific fixes for high impedance """ suggestions = {} for electrode, z in impedances.items(): if z > self.target * 4: suggestions[electrode] = [ "Re-prep with more aggressive abrasion", "Check for hair interference", "Replace electrode if damaged" ] elif z > self.target * 2: suggestions[electrode] = [ "Add more conductive gel", "Press electrode more firmly", "Clean area with " ] elif z > self.target: suggestions[electrode] = [ "Minor adjustment needed", "Wait for gel to settle" ] return suggestions def plot_impedance_map(self, impedances): """ Create visual impedance map """ # This would create a topographical plot # Color coding: Green (<5kΩ), Yellow (5-10kΩ), Red (>10kΩ) pass # Common impedance problems and solutions impedance_problems = { 'all_high': { 'cause': ' ', # Ground electrode issue 'solution': 'Check ground electrode connection and impedance' }, 'increasing_over_time': { 'cause': ' ', # Gel drying 'solution': 'Add more gel or use different gel type' }, 'sudden_spike': { 'cause': 'Electrode lifted or moved', 'solution': ' ' } }

from pylsl import StreamInfo, StreamOutlet, StreamInlet, resolve_stream import numpy as np import time class EEGStreamer: def __init__(self, device_name, n_channels, sampling_rate): """ Initialize LSL stream for EEG data """ # Create stream info self.info = StreamInfo( name=' ', # Stream name type=' ', # Data type (EEG) channel_count=n_channels, nominal_srate=sampling_rate, channel_format=' ', # float32 source_id=device_name ) # Add channel labels channels = self.info.desc().append_child("channels") channel_names = [f'Ch{i+1}' for i in range(n_channels)] for c in channel_names: channels.append_child("channel") \ .append_child_value("label", c) \ .append_child_value("unit", " ") \ .append_child_value("type", "EEG") # Create outlet self.outlet = StreamOutlet(self.info) def stream_data(self, eeg_buffer): """ Send data chunk through LSL """ # Check if anyone is listening if self.outlet.have_consumers(): # Send sample self.outlet. (eeg_buffer.tolist()) def add_markers(self, marker_stream): """ Send event markers synchronized with EEG """ # Create marker outlet marker_info = StreamInfo( name=' ', type='Markers', channel_count=1, nominal_srate=0, # Irregular sampling channel_format=' ', # string source_id='marker_stream' ) marker_outlet = StreamOutlet(marker_info) return marker_outlet class EEGReceiver: def __init__(self, stream_name=None): """ Connect to LSL stream and receive data """ # Resolve EEG stream print("Looking for EEG stream...") streams = resolve_stream('type', ' ') if len(streams) == 0: raise RuntimeError("No EEG stream found") # Create inlet self.inlet = StreamInlet(streams[0]) # Get stream info info = self.inlet.info() self.n_channels = info. () self.srate = info.nominal_srate() print(f"Connected to {info.name()} - {self.n_channels} channels @ {self.srate} Hz") def receive_chunk(self, max_samples=1024): """ Receive a chunk of data """ # Pull chunk chunk, timestamps = self.inlet. (max_samples=max_samples) if chunk: return np.array(chunk), np.array(timestamps) else: return None, None def receive_continuous(self, duration, callback=None): """ Continuously receive data for specified duration """ start_time = time.time() all_data = [] all_timestamps = [] while (time.time() - start_time) < duration: chunk, timestamps = self.receive_chunk() if chunk is not None: all_data.append(chunk) all_timestamps.append(timestamps) # Process in real-time if callback provided if callback: callback(chunk, timestamps) # Concatenate all data if all_data: data = np. (all_data, axis=0) timestamps = np.concatenate(all_timestamps) return data, timestamps else: return None, None

from brainflow.board_shim import BoardShim, BrainFlowInputParams, BoardIds from brainflow.data_filter import DataFilter, FilterTypes, DetrendOperations class BrainFlowEEG: def __init__(self, board_type='SYNTHETIC'): """ Initialize BrainFlow for EEG acquisition """ # Set up board parameters self.params = BrainFlowInputParams() # Select board if board_type == 'SYNTHETIC': self.board_id = BoardIds. .value elif board_type == 'OPENBCI': self.board_id = BoardIds.CYTON_BOARD.value self.params.serial_port = ' ' # COM port elif board_type == 'MUSE': self.board_id = BoardIds. .value # Create board object self.board = BoardShim(self.board_id, self.params) # Get board info self.sampling_rate = BoardShim.get_sampling_rate(self.board_id) self.eeg_channels = BoardShim.get_eeg_channels(self.board_id) self.timestamp_channel = BoardShim.get_timestamp_channel(self.board_id) def start_stream(self, buffer_size=450000): """ Start data acquisition """ # Prepare session self.board. () # Start stream with buffer self.board.start_stream(buffer_size) print(f"Streaming started - {len(self.eeg_channels)} channels @ {self.sampling_rate} Hz") def get_data(self, num_samples=None): """ Get data from the buffer """ if num_samples: # Get specific number of samples data = self.board. (num_samples) else: # Get all available data data = self.board.get_board_data() return data def apply_filters(self, data): """ Apply real-time filters to data """ # Detrend to remove DC offset for channel in self.eeg_channels: DataFilter.detrend(data[channel], DetrendOperations. .value) # Bandpass filter for channel in self.eeg_channels: DataFilter.perform_bandpass( data[channel], self.sampling_rate, , # Low freq , # High freq 4, # Order FilterTypes.BUTTERWORTH.value, 0 # No ripple ) # Notch filter for powerline for channel in self.eeg_channels: DataFilter.perform_bandstop( data[channel], self.sampling_rate, , # Center frequency - 60 Hz 4.0, # Width 4, # Order FilterTypes.BUTTERWORTH.value, 0 ) return data def stop_stream(self): """ Stop acquisition and release resources """ self.board. () self.board.release_session() # Real-time processing example def process_chunk(eeg_chunk, channels): """ Process EEG data in real-time """ # Calculate instantaneous power alpha_channels = [] for ch_idx in channels: # Extract alpha band (8-13 Hz) alpha_power = DataFilter.get_band_power( eeg_chunk[ch_idx], sampling_rate, , # Alpha range ) alpha_channels.append(alpha_power) return np.mean(alpha_channels)

import numpy as np import pandas as pd from psychopy import visual, core, event class EEGExperiment: def __init__(self, subject_id, condition): self.subject_id = subject_id self.condition = condition self.results = [] # Timing parameters (in seconds) self.timing = { 'fixation': , # Pre-stimulus baseline 'stimulus': , # Stimulus duration 'response_window': , # Max response time 'iti': (1.0, 2.0), # Inter-trial interval (jittered) 'break_every': # Trials between breaks } # Experimental parameters self.n_blocks = 4 self.trials_per_block = 50 self.conditions = ['congruent', 'incongruent'] def create_trial_list(self): """ Create balanced trial list with proper randomization """ trials = [] for block in range(self.n_blocks): block_trials = [] # Create equal number of each condition for condition in self.conditions: n_trials = self.trials_per_block // len(self.conditions) block_trials.extend([condition] * n_trials) # Randomize within block np.random. (block_trials) # Add to main list with block info for i, trial_type in enumerate(block_trials): trial = { 'block': block + 1, 'trial': i + 1, 'condition': trial_type, 'iti': np.random.uniform(*self.timing['iti']) } trials.append(trial) return pd.DataFrame(trials) def send_trigger(self, trigger_code, lsl_outlet=None): """ Send trigger to EEG system """ # Trigger codes triggers = { 'experiment_start': 1, 'block_start': 10, 'fixation': 20, 'stimulus_congruent': , 'stimulus_incongruent': , 'response_correct': 40, 'response_incorrect': 41, 'response_miss': 42, 'block_end': 90, 'experiment_end': 99 } if lsl_outlet: # Send via LSL lsl_outlet.push_sample([trigger_code]) else: # Send via parallel port (if available) # parallel.setData(trigger_code) pass # Log trigger timestamp = core.getTime() return {'trigger': trigger_code, 'timestamp': timestamp} def run_trial(self, trial_info, window, lsl_outlet): """ Run a single trial with proper timing """ trial_data = trial_info.copy() # Fixation period (baseline) fixation = visual.TextStim(window, text='+') fixation.draw() window.flip() self.send_trigger( , lsl_outlet) # Fixation trigger core.wait(self.timing['fixation']) # Present stimulus if trial_info['condition'] == 'congruent': stimulus = visual.TextStim(window, text='< < < < <') trigger = self.triggers['stimulus_congruent'] else: stimulus = visual.TextStim(window, text='< < > < <') trigger = self.triggers['stimulus_incongruent'] stimulus.draw() window.flip() stim_onset = core.getTime() self.send_trigger(trigger, lsl_outlet) # Collect response response = event.waitKeys( maxWait=self.timing['response_window'], keyList=['left', 'right'], timeStamped=core.Clock() ) # Process response if response: key, rt = response[0] trial_data['response'] = key trial_data['rt'] = rt * 1000 # Convert to ms # Check accuracy correct_response = 'left' if '<' in trial_info['condition'] else 'right' trial_data['correct'] = (key == correct_response) # Send response trigger if trial_data['correct']: self.send_trigger( , lsl_outlet) else: self.send_trigger(41, lsl_outlet) else: # No response trial_data['response'] = 'miss' trial_data['rt'] = None trial_data['correct'] = False self.send_trigger(42, lsl_outlet) # Inter-trial interval window.flip() # Clear screen core.wait(trial_info['iti']) return trial_data # Artifact control strategies artifact_controls = { 'eye_movements': { 'strategy': 'Central fixation cross', 'implementation': ' ' }, 'muscle_tension': { 'strategy': ' ', 'implementation': 'Regular breaks, comfortable seating' }, 'alpha_contamination': { 'strategy': 'Keep eyes open', 'implementation': ' ' } }

class EventMarkerSystem: def __init__(self): self.marker_log = [] self.sync_pulses = [] def create_marker_scheme(self): """ Design a comprehensive marker scheme """ marker_scheme = { # Experiment structure 'exp_start': 1, 'exp_end': 2, 'block_start': 10, 'block_end': 11, # Trial events 'trial_start': 20, 'fixation_onset': 21, 'stimulus_onset': 30, 'stimulus_offset': 31, 'response_window_start': 40, 'response_window_end': 41, # Stimulus types (30-39) 'stim_category_A': 32, 'stim_category_B': 33, 'stim_target': 35, 'stim_distractor': 36, # Responses (50-59) 'response_correct': 50, 'response_incorrect': 51, 'response_miss': 52, 'response_early': 53, # Special events 'break_start': 90, 'break_end': 91, 'sync_pulse': , # For timing verification 'artifact_detected': 99 } return marker_scheme def send_marker(self, marker_code, timestamp=None): """ Send marker with timing verification """ if timestamp is None: timestamp = time.time() # Record in log self.marker_log.append({ 'code': marker_code, 'timestamp': timestamp, 'sample': None # Will be filled by EEG system }) # Send to multiple systems self._send_to_lsl(marker_code) self._send_to_parallel(marker_code) # For critical markers, send verification pulse if marker_code in [30, 50, 51, 52]: # Stimulus/response markers time.sleep(0.001) # 1ms delay self._send_to_parallel( ) # Clear/sync pulse def verify_timing(self, eeg_timestamps): """ Check marker timing accuracy """ timing_errors = [] for i, marker in enumerate(self.marker_log[:-1]): next_marker = self.marker_log[i + 1] # Calculate expected vs actual intervals expected_interval = next_marker['timestamp'] - marker['timestamp'] # Find markers in EEG data marker_sample = self._find_marker_in_eeg(marker['code'], eeg_timestamps) next_sample = self._find_marker_in_eeg(next_marker['code'], eeg_timestamps) if marker_sample and next_sample: actual_interval = (next_sample - marker_sample) / # sampling_rate # Calculate timing error error = abs(actual_interval - expected_interval) * 1000 # ms if error > : # Threshold in ms timing_errors.append({ 'marker1': marker['code'], 'marker2': next_marker['code'], 'error_ms': error }) return timing_errors def generate_sync_pattern(self, duration=10): """ Generate synchronization pattern for offline alignment """ # Create unique pattern of pulses sync_times = [] patterns = [100, 200, 100, 300, 100, 100, 200] # ms intervals current_time = 0 for interval in patterns: current_time += interval / 1000 # Convert to seconds sync_times.append(current_time) return sync_times # Timing considerations timing_best_practices = { 'stimulus_duration': { 'minimum': , # ms - at least 2 screen refreshes 'recommended': 500, # ms 'reason': 'Ensure reliable ERPs' }, 'marker_precision': { 'requirement': ' ms', # 1-2ms precision needed 'method': 'Hardware triggers or LSL timestamps' }, 'jitter': { 'purpose': ' ', 'range': '±500ms from mean ITI' } }