Documentation Index Fetch the complete documentation index at: https://docs.nimbusbci.com/llms.txt
Use this file to discover all available pages before exploring further.
MNE-Python Integration
The nimbus-bci library integrates seamlessly with MNE-Python, the leading Python package for EEG/MEG analysis. This allows you to build complete BCI pipelines from raw EEG to classification.
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 classification
Uncertainty quantification
Online learning
Real-time inference
Model families: NimbusLDA, NimbusQDA, NimbusSoftmax, NimbusSTS
Installation
Install both packages:
pip install nimbus-bci[mne]
This installs:
nimbus-bci - Bayesian classifiers
mne ≥ 1.6 - EEG preprocessing
Basic Workflow
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
)
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. Train Classifier
from nimbus_bci import NimbusLDA
# Get labels
labels = epochs.events[:, 2 ]
# Train classifier
clf = NimbusLDA()
clf.fit(csp_features, labels)
# Evaluate
from sklearn.model_selection import cross_val_score
scores = cross_val_score(clf, csp_features, labels, cv = 5 )
print ( f "Accuracy: { scores.mean() :.2%} (+/- { scores.std() :.2%} )" )
Complete Motor Imagery Pipeline
import mne
from nimbus_bci import NimbusLDA
from nimbus_bci.compat import extract_csp_features
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
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. Create pipeline
pipe = make_pipeline(
StandardScaler(),
NimbusLDA( mu_scale = 3.0 )
)
# 8. Train
pipe.fit(X_train, y_train)
# 9. Evaluate
train_score = pipe.score(X_train, y_train)
test_score = pipe.score(X_test, y_test)
print ( f "Train accuracy: { train_score :.2%} " )
print ( f "Test accuracy: { test_score :.2%} " )
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
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 ])
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
Streaming Inference Real-time BCI with chunk processing
API Reference Complete API documentation
sklearn Integration Advanced sklearn patterns
Examples Working code examples