0
# Machine Learning
1
2
Machine learning tools for decoding neural signals including common spatial patterns (CSP), temporal decoding, encoding models, and cross-validation utilities designed for neuroimaging data.
3
4
## Capabilities
5
6
### Spatial Filtering Methods
7
8
Learn spatial filters to extract discriminative patterns from multichannel neuroimaging data.
9
10
```python { .api }
11
class CSP:
12
"""Common Spatial Patterns for EEG/MEG classification."""
13
14
def __init__(self, n_components: int = 4, reg: Optional[Union[str, float]] = None,
15
log: Optional[bool] = None, cov_est: str = 'concat', transform_into: str = 'average_power',
16
norm_trace: bool = False, cov_method_params: Optional[Dict] = None,
17
rank: Optional[Union[str, int, Dict]] = None, component_order: str = 'mutual_info'):
18
"""
19
Initialize CSP transformer.
20
21
Parameters:
22
- n_components: Number of CSP components to keep
23
- reg: Regularization parameter for covariance estimation
24
- log: Apply log transformation to features
25
- cov_est: Covariance estimation method
26
- transform_into: Transform method ('average_power' or 'csp_space')
27
- norm_trace: Normalize covariance trace
28
- cov_method_params: Covariance estimation parameters
29
- rank: Data rank specification
30
- component_order: Component ordering method
31
"""
32
33
def fit(self, X: ArrayLike, y: ArrayLike) -> 'CSP':
34
"""
35
Fit CSP spatial filters.
36
37
Parameters:
38
- X: Training data (n_epochs, n_channels, n_times)
39
- y: Class labels (n_epochs,)
40
41
Returns:
42
Fitted CSP object
43
"""
44
45
def transform(self, X: ArrayLike) -> ArrayLike:
46
"""
47
Apply CSP transformation.
48
49
Parameters:
50
- X: Data to transform (n_epochs, n_channels, n_times)
51
52
Returns:
53
Transformed features (n_epochs, n_components)
54
"""
55
56
def fit_transform(self, X: ArrayLike, y: ArrayLike) -> ArrayLike:
57
"""
58
Fit CSP and transform data.
59
60
Parameters:
61
- X: Training data (n_epochs, n_channels, n_times)
62
- y: Class labels (n_epochs,)
63
64
Returns:
65
CSP features (n_epochs, n_components)
66
"""
67
68
def plot_patterns(self, info: Info, components: Optional[Union[int, List[int]]] = None,
69
ch_type: Optional[str] = None, layout: Optional[str] = None,
70
vmin: Optional[float] = None, vmax: Optional[float] = None,
71
cmap: Optional[str] = None, sensors: Union[bool, str] = True,
72
colorbar: bool = True, scale: Optional[float] = None,
73
scale_time: Optional[float] = None, unit: Optional[str] = None,
74
size: Union[int, Tuple] = 1, show_names: Union[bool, str] = False,
75
title: Optional[str] = None, show: bool = True,
76
outlines: str = 'head', image_interp: str = 'bilinear',
77
average: Optional[float] = None, sphere: Optional[float] = None) -> Figure:
78
"""
79
Plot CSP spatial patterns.
80
81
Returns:
82
Figure object
83
"""
84
85
class SPoC:
86
"""Source Power Comodulation."""
87
88
def __init__(self, n_components: int = 4, reg: Optional[Union[str, float]] = None,
89
log: Optional[bool] = None, cov_est: str = 'concat',
90
norm_trace: bool = False, cov_method_params: Optional[Dict] = None,
91
rank: Optional[Union[str, int, Dict]] = None):
92
"""
93
Initialize SPoC transformer.
94
95
Parameters:
96
- n_components: Number of components
97
- reg: Regularization parameter
98
- log: Apply log transformation
99
- cov_est: Covariance estimation method
100
- norm_trace: Normalize covariance trace
101
- cov_method_params: Covariance parameters
102
- rank: Data rank specification
103
"""
104
105
def fit(self, X: ArrayLike, y: ArrayLike) -> 'SPoC':
106
"""
107
Fit SPoC spatial filters.
108
109
Parameters:
110
- X: Training data (n_epochs, n_channels, n_times)
111
- y: Continuous target variable (n_epochs,)
112
113
Returns:
114
Fitted SPoC object
115
"""
116
117
def transform(self, X: ArrayLike) -> ArrayLike:
118
"""
119
Apply SPoC transformation.
120
121
Parameters:
122
- X: Data to transform (n_epochs, n_channels, n_times)
123
124
Returns:
125
SPoC features (n_epochs, n_components)
126
"""
127
128
class SSD:
129
"""Spatio-Spectral Decomposition."""
130
131
def __init__(self, info: Info, filt_params_signal: Dict, filt_params_noise: Dict,
132
reg: Optional[Union[str, float]] = None, n_components: Optional[int] = None,
133
picks: Optional[Union[str, List]] = None, sort_by_spectral_ratio: bool = True,
134
return_filtered: bool = False, n_fft: Optional[int] = None,
135
cov_method_params: Optional[Dict] = None, rank: Optional[Union[str, int, Dict]] = None):
136
"""
137
Initialize SSD transformer.
138
139
Parameters:
140
- info: Measurement info
141
- filt_params_signal: Signal band filter parameters
142
- filt_params_noise: Noise band filter parameters
143
- reg: Regularization parameter
144
- n_components: Number of components
145
- picks: Channel selection
146
- sort_by_spectral_ratio: Sort by spectral ratio
147
- return_filtered: Return filtered data
148
- n_fft: FFT length for PSD computation
149
- cov_method_params: Covariance parameters
150
- rank: Data rank specification
151
"""
152
153
def fit(self, X: ArrayLike) -> 'SSD':
154
"""Fit SSD spatial filters."""
155
156
def transform(self, X: ArrayLike) -> ArrayLike:
157
"""Apply SSD transformation."""
158
```
159
160
### Temporal Decoding
161
162
Decode information from neural signals with time-resolved analysis.
163
164
```python { .api }
165
class SlidingEstimator:
166
"""Time-resolved decoding with sliding window."""
167
168
def __init__(self, base_estimator, scoring: Optional[Union[str, callable]] = None,
169
n_jobs: int = 1, verbose: Optional[Union[bool, str, int]] = None):
170
"""
171
Initialize sliding estimator.
172
173
Parameters:
174
- base_estimator: Base classifier/regressor
175
- scoring: Scoring method
176
- n_jobs: Number of parallel jobs
177
- verbose: Verbosity level
178
"""
179
180
def fit(self, X: ArrayLike, y: ArrayLike) -> 'SlidingEstimator':
181
"""
182
Fit estimator at each time point.
183
184
Parameters:
185
- X: Training data (n_epochs, n_features, n_times)
186
- y: Target labels (n_epochs,)
187
188
Returns:
189
Fitted SlidingEstimator
190
"""
191
192
def predict(self, X: ArrayLike) -> ArrayLike:
193
"""
194
Predict at each time point.
195
196
Parameters:
197
- X: Test data (n_epochs, n_features, n_times)
198
199
Returns:
200
Predictions (n_epochs, n_times)
201
"""
202
203
def score(self, X: ArrayLike, y: ArrayLike) -> ArrayLike:
204
"""
205
Score at each time point.
206
207
Parameters:
208
- X: Test data (n_epochs, n_features, n_times)
209
- y: True labels (n_epochs,)
210
211
Returns:
212
Scores at each time point (n_times,)
213
"""
214
215
class GeneralizingEstimator:
216
"""Cross-temporal decoding (temporal generalization)."""
217
218
def __init__(self, base_estimator, scoring: Optional[Union[str, callable]] = None,
219
n_jobs: int = 1, verbose: Optional[Union[bool, str, int]] = None):
220
"""
221
Initialize generalizing estimator.
222
223
Parameters:
224
- base_estimator: Base classifier/regressor
225
- scoring: Scoring method
226
- n_jobs: Number of parallel jobs
227
- verbose: Verbosity level
228
"""
229
230
def fit(self, X: ArrayLike, y: ArrayLike) -> 'GeneralizingEstimator':
231
"""
232
Fit estimator at each time point.
233
234
Parameters:
235
- X: Training data (n_epochs, n_features, n_times)
236
- y: Target labels (n_epochs,)
237
238
Returns:
239
Fitted GeneralizingEstimator
240
"""
241
242
def score(self, X: ArrayLike, y: ArrayLike) -> ArrayLike:
243
"""
244
Score on all time point combinations.
245
246
Parameters:
247
- X: Test data (n_epochs, n_features, n_times)
248
- y: True labels (n_epochs,)
249
250
Returns:
251
Cross-temporal decoding scores (n_times, n_times)
252
"""
253
254
class TimeDelayingRidge:
255
"""Ridge regression with time delays for encoding models."""
256
257
def __init__(self, tmin: float, tmax: float, sfreq: float, alpha: float = 0.0,
258
reg_type: Optional[str] = None, n_jobs: int = 1, edge_correction: bool = True):
259
"""
260
Initialize time-delaying ridge regression.
261
262
Parameters:
263
- tmin: Minimum time delay
264
- tmax: Maximum time delay
265
- sfreq: Sampling frequency
266
- alpha: Regularization parameter
267
- reg_type: Regularization type
268
- n_jobs: Number of parallel jobs
269
- edge_correction: Apply edge correction
270
"""
271
272
def fit(self, X: ArrayLike, y: ArrayLike) -> 'TimeDelayingRidge':
273
"""
274
Fit time-delayed ridge regression.
275
276
Parameters:
277
- X: Features (n_times, n_features)
278
- y: Target (n_times, n_targets)
279
280
Returns:
281
Fitted TimeDelayingRidge
282
"""
283
284
def predict(self, X: ArrayLike) -> ArrayLike:
285
"""
286
Predict using fitted model.
287
288
Parameters:
289
- X: Features (n_times, n_features)
290
291
Returns:
292
Predictions (n_times, n_targets)
293
"""
294
```
295
296
### Feature Extraction and Preprocessing
297
298
Transform raw neural data into features suitable for machine learning.
299
300
```python { .api }
301
class Vectorizer:
302
"""Flatten multidimensional features."""
303
304
def __init__(self):
305
"""Initialize vectorizer."""
306
307
def fit(self, X: ArrayLike, y: Optional[ArrayLike] = None) -> 'Vectorizer':
308
"""Fit vectorizer (no-op)."""
309
return self
310
311
def transform(self, X: ArrayLike) -> ArrayLike:
312
"""
313
Flatten last two dimensions.
314
315
Parameters:
316
- X: Input data (..., n_channels, n_times)
317
318
Returns:
319
Flattened data (..., n_channels * n_times)
320
"""
321
322
class Scaler:
323
"""Feature scaling for neural data."""
324
325
def __init__(self, info: Info, scalings: Optional[Dict] = None, with_mean: bool = True,
326
with_std: bool = True):
327
"""
328
Initialize scaler.
329
330
Parameters:
331
- info: Measurement info
332
- scalings: Scaling factors by channel type
333
- with_mean: Center data
334
- with_std: Scale to unit variance
335
"""
336
337
def fit(self, X: ArrayLike, y: Optional[ArrayLike] = None) -> 'Scaler':
338
"""Fit scaling parameters."""
339
340
def transform(self, X: ArrayLike) -> ArrayLike:
341
"""Apply scaling transformation."""
342
343
class PSDEstimator:
344
"""Power spectral density features."""
345
346
def __init__(self, sfreq: float = 2 * np.pi, fmin: float = 0, fmax: float = np.inf,
347
bandwidth: Optional[float] = None, adaptive: bool = False, low_bias: bool = True,
348
n_jobs: int = 1, normalization: str = 'length', verbose: Optional[Union[bool, str, int]] = None):
349
"""
350
Initialize PSD estimator.
351
352
Parameters:
353
- sfreq: Sampling frequency
354
- fmin: Minimum frequency
355
- fmax: Maximum frequency
356
- bandwidth: Multitaper bandwidth
357
- adaptive: Use adaptive weighting
358
- low_bias: Reduce bias
359
- n_jobs: Number of parallel jobs
360
- normalization: Normalization method
361
- verbose: Verbosity level
362
"""
363
364
def fit(self, X: ArrayLike, y: Optional[ArrayLike] = None) -> 'PSDEstimator':
365
"""Fit PSD estimator."""
366
367
def transform(self, X: ArrayLike) -> ArrayLike:
368
"""Compute PSD features."""
369
370
class FilterEstimator:
371
"""Frequency filtering as preprocessing step."""
372
373
def __init__(self, info: Info, l_freq: Optional[float], h_freq: Optional[float],
374
l_trans_bandwidth: str = 'auto', h_trans_bandwidth: str = 'auto',
375
filter_length: str = 'auto', method: str = 'fir', n_jobs: int = 1):
376
"""
377
Initialize filter estimator.
378
379
Parameters:
380
- info: Measurement info
381
- l_freq: Low frequency cutoff
382
- h_freq: High frequency cutoff
383
- l_trans_bandwidth: Low transition bandwidth
384
- h_trans_bandwidth: High transition bandwidth
385
- filter_length: Filter length
386
- method: Filter method
387
- n_jobs: Number of parallel jobs
388
"""
389
390
def fit(self, X: ArrayLike, y: Optional[ArrayLike] = None) -> 'FilterEstimator':
391
"""Fit filter (no-op)."""
392
return self
393
394
def transform(self, X: ArrayLike) -> ArrayLike:
395
"""Apply filtering."""
396
```
397
398
### Cross-Validation and Model Evaluation
399
400
Specialized cross-validation and evaluation methods for neuroimaging data.
401
402
```python { .api }
403
def cross_val_multiscore(estimator, X: ArrayLike, y: Optional[ArrayLike] = None,
404
groups: Optional[ArrayLike] = None,
405
scoring: Optional[Union[str, List[str], Dict]] = None,
406
cv: Optional[Union[int, callable]] = None,
407
n_jobs: int = 1, verbose: int = 0,
408
fit_params: Optional[Dict] = None,
409
pre_dispatch: Union[int, str] = '2*n_jobs') -> ArrayLike:
410
"""
411
Cross-validation with multiple scoring metrics.
412
413
Parameters:
414
- estimator: Estimator to evaluate
415
- X: Feature data
416
- y: Target data
417
- groups: Group labels for grouped CV
418
- scoring: Scoring method(s)
419
- cv: Cross-validation strategy
420
- n_jobs: Number of parallel jobs
421
- verbose: Verbosity level
422
- fit_params: Parameters to pass to fit
423
- pre_dispatch: Controls job dispatching
424
425
Returns:
426
Cross-validation scores
427
"""
428
429
def get_coef(estimator, attr: str = 'filters_', inverse_transform: bool = False) -> ArrayLike:
430
"""
431
Extract coefficients from fitted estimator.
432
433
Parameters:
434
- estimator: Fitted estimator
435
- attr: Attribute name containing coefficients
436
- inverse_transform: Apply inverse transformation
437
438
Returns:
439
Coefficient array
440
"""
441
```
442
443
### Encoding Models
444
445
Model how stimulus features are encoded in neural responses.
446
447
```python { .api }
448
class ReceptiveField:
449
"""Temporal response function estimation."""
450
451
def __init__(self, tmin: float, tmax: float, sfreq: float, feature_names: Optional[List[str]] = None,
452
estimator: Optional[str] = None, scoring: str = 'corrcoef', n_jobs: int = 1,
453
edge_correction: bool = True, verbose: Optional[Union[bool, str, int]] = None):
454
"""
455
Initialize receptive field estimator.
456
457
Parameters:
458
- tmin: Minimum time lag
459
- tmax: Maximum time lag
460
- sfreq: Sampling frequency
461
- feature_names: Names of stimulus features
462
- estimator: Regression estimator to use
463
- scoring: Scoring method
464
- n_jobs: Number of parallel jobs
465
- edge_correction: Apply edge correction
466
- verbose: Verbosity level
467
"""
468
469
def fit(self, X: ArrayLike, y: ArrayLike) -> 'ReceptiveField':
470
"""
471
Fit receptive field model.
472
473
Parameters:
474
- X: Stimulus features (n_times, n_features)
475
- y: Neural responses (n_times, n_channels)
476
477
Returns:
478
Fitted ReceptiveField
479
"""
480
481
def predict(self, X: ArrayLike) -> ArrayLike:
482
"""
483
Predict neural responses.
484
485
Parameters:
486
- X: Stimulus features (n_times, n_features)
487
488
Returns:
489
Predicted responses (n_times, n_channels)
490
"""
491
492
def score(self, X: ArrayLike, y: ArrayLike) -> ArrayLike:
493
"""
494
Score model performance.
495
496
Parameters:
497
- X: Stimulus features
498
- y: True neural responses
499
500
Returns:
501
Scores for each channel
502
"""
503
504
class EMS:
505
"""EEG Motor Imagery Separability."""
506
507
def __init__(self, fmin: float = 8, fmax: float = 35, n_freq: int = 9,
508
n_steps: int = 8, return_spectrum: bool = False):
509
"""
510
Initialize EMS computation.
511
512
Parameters:
513
- fmin: Minimum frequency
514
- fmax: Maximum frequency
515
- n_freq: Number of frequencies
516
- n_steps: Number of time steps
517
- return_spectrum: Return full spectrum
518
"""
519
520
def fit(self, epochs_1: Epochs, epochs_2: Epochs) -> 'EMS':
521
"""
522
Compute EMS between two conditions.
523
524
Parameters:
525
- epochs_1: First condition epochs
526
- epochs_2: Second condition epochs
527
528
Returns:
529
Fitted EMS object
530
"""
531
532
def compute_ems(epochs_1: Epochs, epochs_2: Epochs, fmin: float = 8, fmax: float = 35,
533
n_freq: int = 9, n_steps: int = 8, return_spectrum: bool = False) -> Union[float, Tuple]:
534
"""
535
Compute EEG Motor Imagery Separability metric.
536
537
Parameters:
538
- epochs_1: First condition epochs
539
- epochs_2: Second condition epochs
540
- fmin: Minimum frequency
541
- fmax: Maximum frequency
542
- n_freq: Number of frequencies
543
- n_steps: Number of time steps
544
- return_spectrum: Return full spectrum
545
546
Returns:
547
EMS value or tuple with spectrum
548
"""
549
```
550
551
## Usage Examples
552
553
### CSP for Motor Imagery Classification
554
555
```python
556
import mne
557
from mne.decoding import CSP
558
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis
559
from sklearn.pipeline import Pipeline
560
561
# Load motor imagery epochs
562
epochs = mne.read_epochs('motor_imagery-epo.fiv')
563
564
# Get data and labels
565
X = epochs.get_data() # (n_epochs, n_channels, n_times)
566
y = epochs.events[:, 2] # Event IDs as labels
567
568
# Create CSP + LDA pipeline
569
csp = CSP(n_components=4, reg=None, log=True)
570
lda = LinearDiscriminantAnalysis()
571
pipeline = Pipeline([('csp', csp), ('lda', lda)])
572
573
# Fit and evaluate
574
from sklearn.model_selection import cross_val_score
575
scores = cross_val_score(pipeline, X, y, cv=5, scoring='accuracy')
576
print(f"Accuracy: {scores.mean():.3f} ± {scores.std():.3f}")
577
578
# Plot CSP patterns
579
csp.fit(X, y)
580
csp.plot_patterns(epochs.info, ch_type='eeg', size=1.5)
581
```
582
583
### Time-Resolved Decoding
584
585
```python
586
import mne
587
from mne.decoding import SlidingEstimator, cross_val_multiscore
588
from sklearn.linear_model import LogisticRegression
589
import numpy as np
590
591
# Load epochs
592
epochs = mne.read_epochs('sample-epo.fiv')
593
594
# Prepare data
595
X = epochs.get_data() # (n_epochs, n_channels, n_times)
596
y = epochs.events[:, 2]
597
598
# Create sliding estimator
599
clf = LogisticRegression(solver='liblinear')
600
time_decod = SlidingEstimator(clf, n_jobs=1, scoring='roc_auc')
601
602
# Perform cross-validation at each time point
603
scores = cross_val_multiscore(time_decod, X, y, cv=5, n_jobs=1)
604
605
# Plot decoding accuracy over time
606
times = epochs.times
607
plt.figure(figsize=(10, 4))
608
plt.plot(times, scores.mean(axis=0))
609
plt.fill_between(times, scores.mean(axis=0) - scores.std(axis=0),
610
scores.mean(axis=0) + scores.std(axis=0), alpha=0.3)
611
plt.xlabel('Time (s)')
612
plt.ylabel('ROC AUC')
613
plt.title('Time-resolved decoding')
614
plt.show()
615
```
616
617
### Cross-Temporal Decoding
618
619
```python
620
import mne
621
from mne.decoding import GeneralizingEstimator
622
from sklearn.linear_model import LogisticRegression
623
624
# Load epochs
625
epochs = mne.read_epochs('sample-epo.fiv')
626
X = epochs.get_data()
627
y = epochs.events[:, 2]
628
629
# Create generalizing estimator
630
clf = LogisticRegression(solver='liblinear')
631
time_gen = GeneralizingEstimator(clf, n_jobs=1, scoring='roc_auc')
632
633
# Fit and score
634
time_gen.fit(X, y)
635
scores = time_gen.score(X, y)
636
637
# Plot cross-temporal decoding matrix
638
times = epochs.times
639
fig, ax = plt.subplots(figsize=(8, 8))
640
im = ax.imshow(scores, extent=[times[0], times[-1], times[0], times[-1]],
641
cmap='RdYlBu_r', vmin=0.4, vmax=0.6, origin='lower')
642
ax.set_xlabel('Testing Time (s)')
643
ax.set_ylabel('Training Time (s)')
644
ax.set_title('Cross-temporal decoding')
645
plt.colorbar(im, ax=ax)
646
plt.show()
647
```
648
649
### Receptive Field Modeling
650
651
```python
652
import mne
653
from mne.decoding import ReceptiveField
654
import numpy as np
655
656
# Load continuous data
657
raw = mne.io.read_raw_fif('sample_audvis_raw.fiv', preload=True)
658
raw.filter(1, 8) # Filter for slow components
659
660
# Create synthetic stimulus (e.g., audio envelope)
661
n_times = len(raw.times)
662
stimulus = np.random.randn(n_times, 3) # 3 stimulus features
663
664
# Get MEG data
665
picks = mne.pick_types(raw.info, meg='grad')
666
meg_data = raw.get_data(picks=picks).T # (n_times, n_channels)
667
668
# Fit receptive field model
669
rf = ReceptiveField(tmin=-0.1, tmax=0.4, sfreq=raw.info['sfreq'],
670
scoring='corrcoef')
671
rf.fit(stimulus, meg_data)
672
673
# Get model scores
674
scores = rf.score(stimulus, meg_data)
675
print(f"Mean correlation: {scores.mean():.3f}")
676
677
# Plot receptive field patterns
678
rf.plot(picks=picks[:10], show=True)
679
```
680
681
### Feature Engineering Pipeline
682
683
```python
684
import mne
685
from mne.decoding import Scaler, Vectorizer, PSDEstimator
686
from sklearn.pipeline import Pipeline
687
from sklearn.ensemble import RandomForestClassifier
688
689
# Load epochs
690
epochs = mne.read_epochs('sample-epo.fiv')
691
X = epochs.get_data()
692
y = epochs.events[:, 2]
693
694
# Create feature extraction pipeline
695
pipeline = Pipeline([
696
('scaler', Scaler(epochs.info)), # Scale by channel type
697
('psd', PSDEstimator(sfreq=epochs.info['sfreq'],
698
fmin=1, fmax=40)), # PSD features
699
('vectorizer', Vectorizer()), # Flatten features
700
('classifier', RandomForestClassifier(n_estimators=100))
701
])
702
703
# Evaluate pipeline
704
from sklearn.model_selection import cross_val_score
705
scores = cross_val_score(pipeline, X, y, cv=5)
706
print(f"Accuracy: {scores.mean():.3f} ± {scores.std():.3f}")
707
```
708
709
## Types
710
711
```python { .api }
712
import numpy as np
713
from typing import Union, Optional, List, Dict, Tuple, Any, Callable
714
715
ArrayLike = Union[np.ndarray, List, Tuple]
716
Figure = Any # matplotlib.figure.Figure
717
```