from nimbus_bci import NimbusLDAimport numpy as np# Generate sample data (in practice, use your preprocessed EEG features)np.random.seed(42)n_samples, n_features = 100, 16X_train = np.random.randn(n_samples, n_features)y_train = np.random.randint(0, 4, n_samples) # 4-class problem# Create and fit classifierclf = NimbusLDA()clf.fit(X_train, y_train)# Predict on new dataX_test = np.random.randn(20, n_features)predictions = clf.predict(X_test)probabilities = clf.predict_proba(X_test)print(f"Predictions: {predictions}")print(f"Probabilities shape: {probabilities.shape}") # (20, 4)
Key concepts:NimbusLDA() creates a Bayesian LDA classifier, fit() trains the model on labeled data, predict() returns class predictions, and predict_proba() returns full posterior distributions.
Common hyperparameters:mu_scale controls prior precision on class means (higher = more regularization), and class_prior_alpha sets the Dirichlet prior on class probabilities.
5
Online Learning
Update models incrementally with new data:
from nimbus_bci import NimbusLDA# Initial trainingclf = NimbusLDA()clf.fit(X_train, y_train)# Later: update with new data (no retraining from scratch)X_new = np.random.randn(10, n_features)y_new = np.random.randint(0, 4, 10)clf.partial_fit(X_new, y_new)print("Model updated with new data")
partial_fit() enables adaptive BCI systems that learn from user feedback during operation.
6
Optional: Active Learning
Use active learning to label only the most informative calibration trials:
from nimbus_bci.active_learning import calibration_sufficient, suggest_next_trialprevious = clf.get_model()ranked = suggest_next_trial( clf, X_pool, strategy="bald", n=4, num_posterior_samples=256,)# Cue and label the selected pool rows, then update.X_new, y_new = collect_labels_for(ranked.indices)clf.partial_fit(X_new, y_new)status = calibration_sufficient( clf, X_pool, criterion="posterior_stability", previous=previous, threshold=0.02,)
Active learning expects feature rows shaped (n_trials, n_features). See Active Learning for strategy guidance and stopping criteria.
7
Batch Inference with Diagnostics
Use batch inference for comprehensive diagnostics:
from nimbus_bci import predict_batch, NimbusLDAfrom nimbus_bci.data import BCIData, BCIMetadataimport numpy as np# Train modelclf = NimbusLDA()clf.fit(X_train, y_train)# Prepare data for batch inferencemetadata = BCIMetadata( sampling_rate=250.0, paradigm="motor_imagery", feature_type="csp", n_features=16, n_classes=4)# BCIData expects (n_features, n_samples, n_trials). predict_batch()# requires at least 2 time samples, so duplicate tabular features across# a short temporal axis for this synthetic example.X_test_reshaped = np.repeat(X_test.T[:, np.newaxis, :], 2, axis=1)data = BCIData(X_test_reshaped, metadata)# Run batch inferenceresult = predict_batch(clf.model_, data)print(f"Predictions: {result.predictions}")print(f"Mean entropy: {result.mean_entropy:.2f} bits")print(f"Balance: {result.balance:.2%}")print(f"Latency: {result.latency_ms:.1f}ms")
Diagnostics available:entropy (prediction uncertainty), balance (class distribution balance), and latency_ms (inference time per trial).
8
Streaming Inference
Process data in real-time chunks:
from nimbus_bci import NimbusLDA, StreamingSessionfrom nimbus_bci.data import BCIMetadata# Train modelclf = NimbusLDA()clf.fit(X_train, y_train)# Setup streaming sessionmetadata = BCIMetadata( sampling_rate=250.0, paradigm="motor_imagery", feature_type="csp", n_features=16, n_classes=4, chunk_size=125, # 500ms chunks at 250 Hz temporal_aggregation="logvar")session = StreamingSession(clf.model_, metadata)# Simulate streaming data (in practice, from real-time EEG)n_chunks = 4for i in range(n_chunks): # Each chunk: (n_features, chunk_size) chunk = np.random.randn(16, 125) result = session.process_chunk(chunk) print(f"Chunk {i+1}: class {result.prediction} (conf: {result.confidence:.2%})")# Finalize trial with aggregationfinal = session.finalize_trial(method="weighted_vote")print(f"\nFinal prediction: class {final.prediction}")print(f"Entropy: {final.entropy:.2f} bits")
Streaming inference enables real-time BCI applications with sub-20ms latency per chunk.
from nimbus_bci import NimbusSTS# Train with moderate drift trackingclf = NimbusSTS( transition_cov=0.05, # Controls drift adaptation speed num_steps=100)clf.fit(X_train, y_train)# Stateful prediction with time propagationpredictions = []for x_t in X_stream: # Time-ordered samples # Advance state one time step clf.propagate_state() # Predict using current state pred = clf.predict(x_t.reshape(1, -1))[0] predictions.append(pred)# Online learning with delayed feedback (canonical BCI paradigm)for trial in online_session: # 1. Advance time clf.propagate_state() # 2. Predict prediction = clf.predict(trial.features.reshape(1, -1))[0] # 3. User performs action execute_action(prediction) # 4. Get feedback true_label = wait_for_feedback() # 5. Update state clf.partial_fit(trial.features.reshape(1, -1), [true_label])# State inspection and transferz_mean, z_cov = clf.get_latent_state()print(f"Current state: {z_mean}")# Save state for next sessionimport picklewith open("model_with_state.pkl", "wb") as f: pickle.dump({'model': clf, 'state': (z_mean, z_cov)}, f)
Key Difference: Unlike static models (LDA, QDA, Softmax), NimbusSTS maintains a latent state that evolves over time, allowing it to adapt to non-stationary distributions. Use propagate_state() to explicitly advance time.
from nimbus_bci import estimate_normalization_params, apply_normalization# Estimate normalization from training datanorm_params = estimate_normalization_params(X_train, method="zscore")# Apply to all dataX_train_norm = apply_normalization(X_train, norm_params)X_test_norm = apply_normalization(X_test, norm_params)# Train on normalized dataclf = NimbusLDA()clf.fit(X_train_norm, y_train)# Predict on normalized test datapredictions = clf.predict(X_test_norm)
Always use the same normalization parameters for training and testing data. Estimate from training set, then apply to test set.
from nimbus_bci import NimbusQDA# QDA often works well for P300 (overlapping distributions)clf = NimbusQDA()clf.fit(erp_features, labels) # 0=non-target, 1=target# Predict with confidenceprobs = clf.predict_proba(new_erp_features)high_confidence = probs[:, 1] > 0.8 # Target probability > 80%
from nimbus_bci import NimbusLDA# Initial trainingclf = NimbusLDA()clf.fit(X_calibration, y_calibration)# During use: adapt to userfor trial in online_session: trial_features = trial.features.reshape(1, -1) prediction = clf.predict(trial_features)[0] # Get feedback (e.g., from user or task) true_label = get_feedback() # Update model clf.partial_fit(trial_features, [true_label])
Not directly. nimbus-bci expects preprocessed features (for example CSP, bandpower, or ERP features). See Preprocessing Requirements.
Which classifier should I start with?
Start with NimbusLDA for fast baselines, especially motor imagery. Try NimbusQDA for overlapping distributions (such as P300), and NimbusSTS for non-stationary sessions.
