nimbus-bci

Nimbus Personalizer — Bayesian personalization head for frozen neural EEG embeddings.
Plug any frozen trunk that emits encode(X) → Z, adapt online with a Nimbus head, ship apps on BrainState — without owning or retraining the encoder.

0.6 product shape: the Personalizer surface — cheap online adaptation (~8–10× cheaper than batch refit, ~22–131× cheaper than trunk retraining at matched n), portable profiles, and paradigm recipes (MI · P300 · Riemannian).

Also ships sklearn-compatible heads for classical feature pipelines (CSP, Riemann, …): Bayesian LDA / QDA / Softmax, plus NimbusSTS (EKF latent state + point-estimated emissions). Those heads power Personalizer and can be used alone.

Docs · Personalizer · API reference · PyPI · Contact

Install

pip install nimbus-bci
# optional: pip install nimbus-bci[softmax]   # JAX Softmax head
# optional: pip install nimbus-bci[riemann]   # pyRiemann pipelines
# optional: pip install nimbus-bci[mne]       # MNE helpers
# optional: pip install nimbus-bci[train]     # torch trunk-arm trainers (LoRA / FT)
# optional: pip install nimbus-bci[viz]       # matplotlib plotting helpers
# optional: pip install nimbus-bci[all]       # everything above

From source: pip install -e ".[all]" after cloning.

Upgrading from 0.5? See Migration notes in CHANGELOG.md (ships in the sdist; also linked from docs.nimbusbci.com).

5-minute quickstart

Run the example (no external data needed):

python examples/quickstart.py

Or follow along:

import numpy as np
from nimbus_bci import Personalizer, wrap

# 1. Wrap any frozen trunk (EEGNet, REVE, ...) — here a random projection.
rng = np.random.default_rng(42)
proj = rng.standard_normal((6, 16))
enc = wrap(lambda X: X @ proj, model_id="toy", embedding_dim=16)

# 2. Fit a Bayesian head on ~30 calibration trials.
#    for_deployment = strict gating + LDA + minimal params (the common case).
#    For research/experimentation: Personalizer.for_research(enc, classes, ...).
#    For classical features (no encoder): Personalizer.for_features(classes).
#    To PIN the accept rate for your trunk (default-gate accept rates span
#    7-78% across trunks), fit + calibrate the gate in one call:
#    adapter.fit_calibrated(X_cal, y_cal, X_val, y_val, target_accept=0.7)
adapter = Personalizer.for_deployment(enc, ["left_hand", "right_hand"],
                                       paradigm="motor_imagery")
adapter.fit(X_cal, y_cal)

# 3. Predict → BrainState with trial rejection.
states = adapter.predict(X_test)         # list[BrainState]
accepted = [s for s in states if not s.rejected]  # confident trials
# s.intent, s.confidence, s.uncertainty, s.rejected, s.rejection_reason

# 4. Adapt online (exact conjugate update — no retraining).
adapter.partial_fit(X_new, y_new)

# 5. Save / load.
adapter.save("profile"); loaded = Personalizer.load("profile", encoder=enc)

The three regimes

# A. Classical features (CSP log-variance, tangent space, hand-crafted) —
#    no encoder, features go straight to the Bayesian head.
p = Personalizer.for_features(["left", "right"]).fit(X_csp, y)

# B. Frozen deep trunk (EEGNet, foundation model) — embeddings → head,
#    strict trial gating out of the box.
p = Personalizer.for_deployment(enc, ["left", "right"]).fit(X_cal, y_cal)
states = p.predict(X_test)          # BrainState per trial, rejected flagged

# C. Riemannian geometry & P300 (OAS / xDAWN tangent space) —
#    weak-trunk / raw epochs; for_riemann defaults to label-free geodesic
#    tracking (tracking_alpha=0.05). Opt out with tracking_alpha=0.0.
p = Personalizer.for_riemann(["left", "right"]).fit(X_cal, y_cal)
# Or for P300 spellers (tracking opt-in; recommend α=0.05 when enabling):
# Personalizer.for_p300(["NonTarget", "Target"], tracking_alpha=0.05)
p.adapt(epoch)                      # label-free reference update (0 labels needed!)

# D. Streaming adaptation loop — predict, optionally partial_fit on new
#    labels; the conjugate head update is exact and cheap (no retraining).
#    Reference: examples/middleware_streaming_adaptation.py
for trial in stream:
    state = p.predict(trial)                # BrainState
    if label_available(state):
        p.partial_fit(trial, label)         # ~1/30-1/131 the wall cost of retraining

Regime A/B/C are the shipped surface. A governed session runtime (drift gates, attribution, label budgets, quarantine) with subconscious error-potential self-supervision is in research and will ship separately.

The full parameter surface (user=, preset=, head choice) via direct construction — anything that maps inputs to embeddings (n, d) plugs in; Nimbus does not own DL architectures:

adapter = Personalizer(
    encoder=enc, head="lda", user="subject_001",
    classes=["left", "right"], preset="strict",
)

Full walkthrough with copy-paste synthetic data: Quickstart. Feeding real EEG? Read the epochs ingest contract first — epoching, baseline, filtering, sfreq discipline, axis conventions: Preparing epochs.

Optional shortcuts for common external trunks: wrap_eegnet, and from nimbus_bci.middleware.adapters import wrap_braindecode (see encoder contract).

Examples in this repo: middleware_personalizer.py, middleware_streaming_adaptation.py, middleware_geometry_gateway.py, middleware_mi_session.py, middleware_braindecode.py, middleware_reve.py, middleware_recommend_adapt.py.
Bench EEGNet demo (repo-only, needs nimbusbench[dl]): nimbusbench/scripts/middleware_eegnet.py.

When to adapt (thin helper)

Optional mean-shift check before spending labels on partial_fit. You choose the threshold tau.

dec = adapter.recommend_adapt(X_stream, tau=your_tau)  # stay vs stream update
if dec.should_adapt:
    adapter.partial_fit(X_lab, y_lab)

Trial rejection (BrainState.rejected / need_more_data) is a different gate — accept/reject the intent for the app.
Details: BrainState · recipe: examples/middleware_recommend_adapt.py.

A governed adaptation session — drift gates, label budgets, quarantine, and rollback — is under research and will ship as a separate release once its evidence matures.

Classical heads (CSP / features)

Same Bayesian heads without an encoder — inputs are already features:

from nimbus_bci import NimbusLDA

clf = NimbusLDA()
clf.fit(X_train, y_train)
proba = clf.predict_proba(X_test)
clf.partial_fit(X_new, y_new)

Also: streaming (StreamingSession), active learning (CalibrationSession), MNE helpers, optional Riemann pipelines. See docs.nimbusbci.com and examples/e2e_mne_csp_nimbus.py.

Working with precomputed embeddings

If your pipeline already computed embeddings — frozen backbone → per-trial embeddings (N, F) — declare the contract so the Personalizer consumes X directly instead of re-encoding it through the attached trunk:

from nimbus_bci import Personalizer, wrap

adapter = Personalizer(
    encoder=wrap(module),        # attached for model_id / lineage — never called
    x_kind="embeddings",         # every X below is ALREADY the encoder output
    head="softmax",
    classes=["left", "right"],
)
adapter.fit(Z_cal, y_cal)                        # (N, F) — head + transforms fit directly
adapter.partial_fit(Z_stream, y_stream)          # online head updates
adapter.adapt(Z_stream, y_stream, level="head")  # supervised head adaptation

The rule: trunk adaptation needs raw encoder inputs. LoRA / BaLoRA / fine-tuning backpropagate through the trunk, and precomputed embeddings have no gradient path to trunk weights — so adapt(..., level="trunk"), adapt_trunk(...), adapt_async(...), fit_tent(...), trunk MC sampling, and Riemannian reference tracking (update_riemann_reference(...)) raise PersonalizerInputError under x_kind="embeddings". To adapt the trunk, feed raw epochs with the default x_kind="epochs". x_kind="embeddings" with encoder=None is also legal — that is the classical feature-space flow (Personalizer.for_features(...)).

Profiles round-trip the contract: export_profile() records x_kind and the observed embed_dim in UserProfile.meta; from_profile(...) restores them, refuses an x_kind mismatch, and validates the feature dim of the next fit against the recorded embed_dim.

Evidence

Public Personalizer story (heads, cost vs fine-tune, multi-trunk / REVE):
Evidence · Overview.

License

Two-track proprietary license — see the LICENSE.txt shipped with the package (and on PyPI). Non-commercial use is free under the Non-Commercial terms; commercial use requires a paid license. Contact: hello@nimbusbci.com.

Contributing & security

Licensees with source access: follow the repository CONTRIBUTING.md and SECURITY.md. Report vulnerabilities privately to hello@nimbusbci.com — do not open a public issue.