MNE-Python Integration

Use MNE-Python with nimbus-bci to build BCI pipelines from raw EEG to BrainState decisions.

Overview

MNE-Python handles:

Loading EEG data from various formats
Preprocessing (filtering, artifact removal)
Epoching and event extraction
Feature extraction (CSP, bandpower)

nimbus-bci handles:

Bayesian personalization (Personalizer) and standalone heads
Uncertainty quantification and gating (BrainState)
Online learning
Real-time inference
Heads: NimbusLDA, NimbusQDA, NimbusSoftmax, NimbusSTS

Dual path after MNE epochs

Product: epochs → frozen encoder (wrap / BrainDecode) → Personalizer → BrainState
Classical: epochs → CSP / bandpower → Personalizer(encoder=None) or NimbusLDA

Installation

Install both packages:

pip install nimbus-bci[mne]

This installs:

nimbus-bci - Bayesian personalization + heads
mne ≥ 1.6 - EEG preprocessing

Product path: epochs → encoder → Personalizer

Use MNE for acquisition / epoching, then hand trial tensors to a frozen trunk:

import mne
from nimbus_bci import Personalizer, wrap

raw = mne.io.read_raw_gdf("motor_imagery.gdf", preload=True)
raw.filter(8, 30)
events = mne.find_events(raw)
epochs = mne.Epochs(
    raw,
    events,
    event_id={"left_hand": 1, "right_hand": 2},
    tmin=0,
    tmax=4,
    baseline=None,
    preload=True,
)

# Trial tensor for your frozen trunk — shape depends on encode()
X = epochs.get_data()  # (n_epochs, n_channels, n_times)
y = epochs.events[:, 2]
# Map to semantic labels if needed
classes = ["left", "right"]

enc = wrap(model.encode, model_id="partner-encoder", embedding_dim=64)
adapter = Personalizer(
    encoder=enc,
    head="lda",
    classes=classes,
    preset="research",
)
adapter.fit(X_cal, y_cal)
states = adapter.predict(X_test)
print(states[0].intent, states[0].confidence, states[0].need_more_data)
adapter.save("subject_mne_profile")

Optional shortcuts: wrap_eegnet, wrap_braindecode — see Personalizer & Middleware.

Classical path: CSP → Personalizer / head

1. Load and Preprocess with MNE

import mne

# Load raw EEG data
raw = mne.io.read_raw_gdf("motor_imagery.gdf", preload=True)

# Filter to relevant frequency bands
raw.filter(8, 30)  # Mu + Beta bands for motor imagery

# Find events
events = mne.find_events(raw)

# Create epochs
epochs = mne.Epochs(
    raw,
    events,
    event_id={'left_hand': 1, 'right_hand': 2},
    tmin=0,
    tmax=4,
    baseline=None,
    preload=True
)

2. Extract Features

from nimbus_bci.compat import extract_csp_features

# Extract CSP features
csp_features, csp = extract_csp_features(epochs, n_components=8)

# csp_features shape: (n_epochs, n_components)
# csp: fitted CSP object for later use

3. Personalize (or fit a head)

from nimbus_bci import Personalizer, NimbusLDA
from sklearn.model_selection import cross_val_score

labels = epochs.events[:, 2]

# Product-style on classical features
adapter = Personalizer(
    encoder=None,
    head="lda",
    classes=["left", "right"],
    preset="research",
)
adapter.fit(csp_features, labels)
states = adapter.predict(csp_features[:5])

# Or standalone head for sklearn CV
clf = NimbusLDA()
scores = cross_val_score(clf, csp_features, labels, cv=5)
print(f"Accuracy: {scores.mean():.2%} (+/- {scores.std():.2%})")

Complete Motor Imagery Pipeline (classical)

import mne
from nimbus_bci import Personalizer
from nimbus_bci.compat import extract_csp_features
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# 1. Load data
raw = mne.io.read_raw_gdf("motor_imagery.gdf", preload=True)

# 2. Preprocessing
raw.filter(8, 30)  # Mu + Beta bands
raw.set_eeg_reference('average')  # Average reference

# 3. Artifact removal (optional)
raw.set_annotations(mne.preprocessing.annotate_muscle_zscore(raw))

# 4. Epoching
events = mne.find_events(raw)
event_id = {'left_hand': 1, 'right_hand': 2, 'feet': 3, 'tongue': 4}
epochs = mne.Epochs(
    raw, events, event_id,
    tmin=0, tmax=4,
    baseline=None,
    preload=True
)

# 5. Feature extraction
csp_features, csp = extract_csp_features(epochs, n_components=8)
labels = epochs.events[:, 2]

# 6. Train/test split
X_train, X_test, y_train, y_test = train_test_split(
    csp_features, labels, test_size=0.2, stratify=labels, random_state=42
)

# 7. Scale + Personalizer (encoder=None on classical features)
scaler = StandardScaler()
Z_train = scaler.fit_transform(X_train)
Z_test = scaler.transform(X_test)

adapter = Personalizer(
    encoder=None,
    head="lda",
    classes=["left", "right", "feet", "tongue"],
    preset="research",
)
adapter.fit(Z_train, y_train)
states = adapter.predict(Z_test)
print(states[0].intent, states[0].confidence)

# 8. Evaluate via the head (sklearn-compatible)
train_score = adapter.head.score(Z_train, y_train)
test_score = adapter.head.score(Z_test, y_test)
print(f"Train accuracy: {train_score:.2%}")
print(f"Test accuracy: {test_score:.2%}")

Optional Riemannian Feature Pipeline

For workflows that use covariance geometry, install the pyRiemann extra and build a sklearn pipeline from EEG epochs to a Nimbus head (then optional Personalizer on the resulting features):

pip install nimbus-bci[riemann]

import mne
from nimbus_bci import NimbusLDA
from nimbus_bci.riemann import make_riemann_nimbus_pipeline
from sklearn.model_selection import train_test_split

raw = mne.io.read_raw_gdf("motor_imagery.gdf", preload=True)
raw.filter(8, 30)
events = mne.find_events(raw)
epochs = mne.Epochs(raw, events, tmin=0, tmax=4, baseline=None, preload=True)

X_epochs = epochs.get_data()  # (n_trials, n_channels, n_times)
y = epochs.events[:, 2]

X_train, X_test, y_train, y_test = train_test_split(
    X_epochs, y, test_size=0.2, stratify=y, random_state=42
)

pipe = make_riemann_nimbus_pipeline(head=NimbusLDA())
pipe.fit(X_train, y_train)

probs = pipe.predict_proba(X_test)
predictions = pipe.predict(X_test)

This factory delegates covariance and tangent-space transforms to pyRiemann, then feeds the resulting feature rows into a Nimbus head. It is a composition helper, not a new Riemannian Bayesian model family.

CSP Feature Extraction

Basic CSP

from nimbus_bci.compat import extract_csp_features

# Extract CSP features
features, csp = extract_csp_features(
    epochs,
    n_components=8  # Number of CSP components
)

Custom CSP Parameters

from mne.decoding import CSP

# Create custom CSP
csp = CSP(
    n_components=8,
    reg='ledoit_wolf',  # Regularization
    log=True,
    norm_trace=False
)

# Fit and transform
X = epochs.get_data()
y = epochs.events[:, 2]
csp_features = csp.fit_transform(X, y)

Multi-Class CSP

For more than 2 classes:

from nimbus_bci.compat import extract_csp_features

# Works automatically for multi-class
features, csp = extract_csp_features(
    epochs,  # 4-class motor imagery
    n_components=8
)

# Train multi-class classifier
from nimbus_bci import NimbusLDA
clf = NimbusLDA()
clf.fit(features, epochs.events[:, 2])

Bandpower Features

Extract Bandpower

from nimbus_bci.compat import extract_bandpower_features

# Define frequency bands
bands = {
    'delta': (1, 4),
    'theta': (4, 8),
    'alpha': (8, 13),
    'beta': (13, 30),
    'gamma': (30, 45)
}

# Extract bandpower features
features, band_names = extract_bandpower_features(
    epochs,
    bands=bands,
    log_transform=True
)

# Train classifier
from nimbus_bci import NimbusQDA
clf = NimbusQDA()
clf.fit(features, epochs.events[:, 2])

Output Shape

features, band_names = extract_bandpower_features(epochs, bands)
print(features.shape)   # (n_epochs, n_channels * n_bands)
print(band_names)       # Names from the bands dict

P300 ERP Features

Extract ERP Amplitudes

import mne
from nimbus_bci import NimbusQDA
import numpy as np

# Load P300 data
raw = mne.io.read_raw_fif("p300_oddball.fif", preload=True)
raw.filter(0.1, 30)

# Epoch around stimuli
events = mne.find_events(raw)
epochs = mne.Epochs(
    raw, events,
    event_id={'target': 1, 'non_target': 2},
    tmin=-0.2, tmax=0.8,
    baseline=(-0.2, 0),
    preload=True
)

# Extract ERP features (mean amplitude in time windows)
def extract_erp_features(epochs):
    data = epochs.get_data()  # (n_epochs, n_channels, n_times)
    
    # Define time windows of interest
    p300_window = (0.3, 0.5)  # 300-500ms
    times = epochs.times
    
    # Find time indices
    idx = np.where((times >= p300_window[0]) & (times <= p300_window[1]))[0]
    
    # Mean amplitude in window
    features = data[:, :, idx].mean(axis=2)  # (n_epochs, n_channels)
    
    return features

erp_features = extract_erp_features(epochs)

# Train classifier
clf = NimbusQDA()
clf.fit(erp_features, epochs.events[:, 2])

Convert Between Formats

MNE Epochs to BCIData

from nimbus_bci.compat import from_mne_epochs

# Convert MNE Epochs to BCIData
data = from_mne_epochs(
    epochs,
    paradigm="motor_imagery",
    feature_type="raw"  # or "csp", "bandpower"
)

# Use with batch inference only when the model was trained with matching
# raw-channel features and metadata.n_features.
from nimbus_bci import predict_batch
# result = predict_batch(raw_feature_model, data)

BCIData to MNE Epochs

from nimbus_bci.compat import to_mne_epochs

# Convert BCIData back to MNE Epochs
epochs_reconstructed = to_mne_epochs(
    bci_data,
    info=original_epochs.info
)

Complete BCI Pipeline

Create end-to-end pipeline:

from nimbus_bci.compat import create_bci_pipeline
from nimbus_bci import NimbusLDA

# Create complete pipeline
pipeline = create_bci_pipeline(
    NimbusLDA,
    feature_extraction="csp",
    n_csp_components=8
)

# Load and preprocess with MNE
raw = mne.io.read_raw_gdf("data.gdf", preload=True)
raw.filter(8, 30)
events = mne.find_events(raw)
epochs = mne.Epochs(raw, events, tmin=0, tmax=4, preload=True)

# Extract features
X = epochs.get_data()
y = epochs.events[:, 2]

# Train pipeline
pipeline.fit(X, y)

# Predict
predictions = pipeline.predict(X)

Real-Time BCI with MNE

Online Processing

import mne
from nimbus_bci import NimbusLDA, StreamingSession
from nimbus_bci.data import BCIMetadata
from nimbus_bci.compat import extract_csp_features

# 1. Calibration phase
print("=== Calibration ===")
raw = mne.io.read_raw_gdf("calibration.gdf", preload=True)
raw.filter(8, 30)
events = mne.find_events(raw)
epochs = mne.Epochs(raw, events, tmin=0, tmax=4, preload=True)

# Train CSP and classifier
csp_features, csp = extract_csp_features(epochs, n_components=8)
clf = NimbusLDA()
clf.fit(csp_features, epochs.events[:, 2])

# 2. Online phase
print("=== Online ===")

# Setup streaming
metadata = BCIMetadata(
    sampling_rate=250.0,
    paradigm="motor_imagery",
    feature_type="csp",
    n_features=8,
    n_classes=4,
    chunk_size=125,
    temporal_aggregation="logvar"
)

session = StreamingSession(clf.model_, metadata)

# Process real-time data
from mne.io import RawArray

while True:
    # Acquire raw EEG chunk (your acquisition code)
    raw_chunk = acquire_eeg_chunk(duration=0.5)  # 500ms
    
    # Create MNE RawArray for preprocessing
    info = mne.create_info(
        ch_names=['C3', 'Cz', 'C4'],  # Your channels
        sfreq=250.0,
        ch_types='eeg'
    )
    raw_mne = RawArray(raw_chunk, info)
    
    # Apply same preprocessing
    raw_mne.filter(8, 30)
    
    # Convert raw EEG into the same feature space used for training.
    # This should return (n_features, chunk_size), matching metadata.
    feature_chunk = extract_streaming_csp_chunk(raw_mne.get_data(), csp)
    
    # Process chunk
    result = session.process_chunk(feature_chunk)
    print(f"Prediction: {result.prediction} ({result.confidence:.2%})")

Cross-Session Transfer

Handle different sessions with normalization:

import mne
from nimbus_bci import NimbusLDA
from nimbus_bci import estimate_normalization_params, apply_normalization
from nimbus_bci.compat import extract_csp_features

# Session 1 (training)
raw1 = mne.io.read_raw_gdf("session1.gdf", preload=True)
raw1.filter(8, 30)
epochs1 = mne.Epochs(raw1, mne.find_events(raw1), tmin=0, tmax=4, preload=True)
features1, csp = extract_csp_features(epochs1, n_components=8)

# Estimate normalization from session 1
norm_params = estimate_normalization_params(features1, method="zscore")
features1_norm = apply_normalization(features1, norm_params)

# Train on normalized features
clf = NimbusLDA()
clf.fit(features1_norm, epochs1.events[:, 2])

# Session 2 (testing)
raw2 = mne.io.read_raw_gdf("session2.gdf", preload=True)
raw2.filter(8, 30)
epochs2 = mne.Epochs(raw2, mne.find_events(raw2), tmin=0, tmax=4, preload=True)

# Apply same CSP
X2 = epochs2.get_data()
features2 = csp.transform(X2)

# Apply same normalization
features2_norm = apply_normalization(features2, norm_params)

# Predict
predictions = clf.predict(features2_norm)
accuracy = (predictions == epochs2.events[:, 2]).mean()
print(f"Cross-session accuracy: {accuracy:.2%}")

Advanced Preprocessing

ICA for Artifact Removal

import mne
from nimbus_bci import NimbusLDA
from nimbus_bci.compat import extract_csp_features

# Load data
raw = mne.io.read_raw_gdf("data.gdf", preload=True)
raw.filter(1, 40)

# Run ICA
ica = mne.preprocessing.ICA(n_components=15, random_state=42)
ica.fit(raw)

# Find and remove eye blinks
eog_indices, eog_scores = ica.find_bads_eog(raw)
ica.exclude = eog_indices

# Apply ICA
raw_clean = ica.apply(raw.copy())

# Continue with clean data
raw_clean.filter(8, 30)
epochs = mne.Epochs(raw_clean, mne.find_events(raw_clean), tmin=0, tmax=4, preload=True)
features, csp = extract_csp_features(epochs)

clf = NimbusLDA()
clf.fit(features, epochs.events[:, 2])

Automated Artifact Rejection

import mne

# Load and filter
raw = mne.io.read_raw_gdf("data.gdf", preload=True)
raw.filter(8, 30)

# Epoch
events = mne.find_events(raw)
epochs = mne.Epochs(raw, events, tmin=0, tmax=4, preload=True)

# Automated rejection
reject_criteria = dict(
    eeg=100e-6  # 100 µV
)

epochs.drop_bad(reject=reject_criteria)

print(f"Kept {len(epochs)} / {len(events)} epochs")

# Continue with clean epochs
from nimbus_bci.compat import extract_csp_features
features, csp = extract_csp_features(epochs)

Visualization

Plot CSP Patterns

import mne
from nimbus_bci.compat import extract_csp_features
import matplotlib.pyplot as plt

# Extract CSP
features, csp = extract_csp_features(epochs, n_components=8)

# Plot CSP patterns
csp.plot_patterns(epochs.info, show=True)

# Plot CSP filters
csp.plot_filters(epochs.info, show=True)

Plot Classification Results

from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
import matplotlib.pyplot as plt

# Get predictions
predictions = clf.predict(features)
labels = epochs.events[:, 2]

# Confusion matrix
cm = confusion_matrix(labels, predictions)
disp = ConfusionMatrixDisplay(cm, display_labels=['Left', 'Right', 'Feet', 'Tongue'])
disp.plot()
plt.show()

Best Practices

1. Consistent Preprocessing

Apply same preprocessing to all data:

def preprocess_raw(raw):
    """Standard preprocessing pipeline."""
    raw = raw.copy()
    raw.filter(8, 30)
    raw.set_eeg_reference('average')
    return raw

# Use for all sessions
raw_train = preprocess_raw(mne.io.read_raw_gdf("train.gdf", preload=True))
raw_test = preprocess_raw(mne.io.read_raw_gdf("test.gdf", preload=True))

2. Save Preprocessing Objects

Save CSP and normalization for later use:

import joblib

# Save CSP transformer
joblib.dump(csp, 'csp_transformer.pkl')

# Save normalization params
joblib.dump(norm_params, 'norm_params.pkl')

# Load later
csp = joblib.load('csp_transformer.pkl')
norm_params = joblib.load('norm_params.pkl')

3. Validate Data Quality

Check data quality before training:

from nimbus_bci import diagnose_preprocessing

# Diagnose features
from nimbus_bci.data import BCIData, BCIMetadata

metadata = BCIMetadata(
    sampling_rate=250.0,
    paradigm="motor_imagery",
    feature_type="csp",
    n_features=features.shape[1],
    n_classes=len(set(labels))
)
bci_data = BCIData(features.T[:, None, :], metadata, labels)

report = diagnose_preprocessing(
    bci_data
)

print(f"Quality score: {report.quality_score}")
print(f"Errors: {report.errors}")
print(f"Warnings: {report.warnings}")

Next Read

Personalizer & Middleware

Encoder contract and BrainState

sklearn Integration

CV / GridSearch on heads

Streaming Inference

Real-time BCI with chunk processing

API Reference

Complete API documentation

​Overview

​Installation

​Product path: epochs → encoder → Personalizer

​Classical path: CSP → Personalizer / head

​1. Load and Preprocess with MNE

​2. Extract Features

​3. Personalize (or fit a head)

​Complete Motor Imagery Pipeline (classical)

​Optional Riemannian Feature Pipeline

​CSP Feature Extraction

​Basic CSP

​Custom CSP Parameters

​Multi-Class CSP

​Bandpower Features

​Extract Bandpower

​Output Shape

​P300 ERP Features

​Extract ERP Amplitudes

​Convert Between Formats

​MNE Epochs to BCIData

​BCIData to MNE Epochs

​Complete BCI Pipeline

​Real-Time BCI with MNE

​Online Processing

​Cross-Session Transfer

​Advanced Preprocessing

​ICA for Artifact Removal

​Automated Artifact Rejection

​Visualization

​Plot CSP Patterns

​Plot Classification Results

​Best Practices

​1. Consistent Preprocessing

​2. Save Preprocessing Objects

​3. Validate Data Quality

​Next Read

Personalizer & Middleware

sklearn Integration

Streaming Inference

API Reference

Overview

Installation

Product path: epochs → encoder → Personalizer

Classical path: CSP → Personalizer / head

1. Load and Preprocess with MNE

2. Extract Features

3. Personalize (or fit a head)

Complete Motor Imagery Pipeline (classical)

Optional Riemannian Feature Pipeline

CSP Feature Extraction

Basic CSP

Custom CSP Parameters

Multi-Class CSP

Bandpower Features

Extract Bandpower

Output Shape

P300 ERP Features

Extract ERP Amplitudes

Convert Between Formats

MNE Epochs to BCIData

BCIData to MNE Epochs

Complete BCI Pipeline

Real-Time BCI with MNE

Online Processing

Cross-Session Transfer

Advanced Preprocessing

ICA for Artifact Removal

Automated Artifact Rejection

Visualization

Plot CSP Patterns

Plot Classification Results

Best Practices

1. Consistent Preprocessing

2. Save Preprocessing Objects

3. Validate Data Quality

Next Read