> ## 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.

# Python Streaming Inference

> Real-time chunk-by-chunk BCI processing with nimbus-bci. Learn streaming inference, chunk processing, and temporal aggregation for real-time BCI.

Process feature or embedding chunks as they arrive with `StreamingSession` (and `StreamingSessionSTS` for `NimbusSTS`).

<Note>
  This page is the **Python API**. Chunk-size guidelines, when to stream vs batch, and aggregation policy live on [Streaming inference configuration](/inference-configuration/streaming-inference). For app-facing decisions on frozen encoders, prefer [Personalizer](/personalizer/overview) and [BrainState](/personalizer/brain-state).
</Note>

## Supported heads

| Session | Heads | Notes |
| - | - | - |
| `StreamingSession` | `NimbusLDA`, `NimbusQDA`, `NimbusSoftmax` | Stateless; works on a fitted model snapshot |
| `StreamingSessionSTS` | `NimbusSTS` | Propagates latent state across chunks |

## Basic loop

```python theme={null}
from nimbus_bci import NimbusLDA, StreamingSession
from nimbus_bci.data import BCIMetadata

clf = NimbusLDA()
clf.fit(X_calibration, y_calibration)

metadata = BCIMetadata(
    sampling_rate=250.0,
    paradigm="motor_imagery",
    feature_type="csp",
    n_features=16,
    n_classes=4,
    chunk_size=125,  # samples per chunk — see config page for duration guidance
    temporal_aggregation="logvar",
)

session = StreamingSession(clf.model_, metadata)

for chunk in eeg_stream:  # each chunk: (n_features, chunk_size)
    result = session.process_chunk(chunk)
    print(result.prediction, result.confidence)

final = session.finalize_trial(method="weighted_vote")
print(final.prediction, final.confidence, final.entropy, final.n_chunks)
session.reset()  # required before the next trial
```

`process_chunk` returns a per-chunk result (`prediction`, `confidence`, `posterior`, …). `finalize_trial` aggregates chunks for the trial; `reset` clears session state.

## Aggregation methods

Pass `method=` to `finalize_trial`. Defaults and when to pick each method are documented on [Streaming inference configuration](/inference-configuration/streaming-inference).

```python theme={null}
final = session.finalize_trial(method="weighted_vote")   # default
final = session.finalize_trial(method="majority_vote")
final = session.finalize_trial(method="max_confidence")
final = session.finalize_trial(method="posterior_mean")
```

## Temporal aggregation

When chunks still have a time axis, `BCIMetadata.temporal_aggregation` reduces them before classification (for example `"logvar"` for CSP). Choose the feature type and aggregation on the [config page](/inference-configuration/streaming-inference); the Python session reads them from `BCIMetadata`.

## Bridge to BrainState

Convert a finalized `StreamingResult` for device / UI decisions:

```python theme={null}
from nimbus_bci import NimbusLDA, StreamingSession, brain_state_from_streaming_result
from nimbus_bci.data import BCIMetadata

clf = NimbusLDA().fit(X_cal, y_cal)
metadata = BCIMetadata(
    sampling_rate=250.0,
    paradigm="motor_imagery",
    feature_type="csp",
    n_features=X_cal.shape[1],
    n_classes=2,
    chunk_size=25,
    temporal_aggregation="logvar",
)
session = StreamingSession(clf.model_, metadata)

# ... session.process_chunk(chunk) for the trial ...
final = session.finalize_trial(method="weighted_vote")
state = brain_state_from_streaming_result(final, classes=["left", "right"])
print(state.intent, state.confidence, state.need_more_data)
session.reset()
```

Prefer `BrainState` for apps; keep `StreamingResult` for research diagnostics. See [BrainState](/personalizer/brain-state).

## StreamingSessionSTS

For non-stationary sessions, use `NimbusSTS` with `StreamingSessionSTS` so latent state propagates across chunks:

```python theme={null}
from nimbus_bci import NimbusSTS
from nimbus_bci.inference import StreamingSessionSTS
from nimbus_bci.data import BCIMetadata

clf = NimbusSTS(transition_cov=0.05, num_steps=100)
clf.fit(X_train, y_train)

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

session = StreamingSessionSTS(clf, metadata)

for chunk in eeg_stream:
    result = session.process_chunk(chunk)

final = session.finalize_trial()
session.reset()
```

Online updates after a trial still use the classifier’s `partial_fit` on trial-level features (your pipeline aggregates chunks). For STS state APIs and when to prefer STS, see [Bayesian STS](/models/rxsts).

## API notes

* Chunk shape must match `(metadata.n_features, metadata.chunk_size)`.
* Always `reset()` between trials so posteriors do not leak.
* Quality gates (confidence / entropy thresholds) are application policy — see [Streaming inference configuration](/inference-configuration/streaming-inference) and [BrainState](/personalizer/brain-state) presets.

## Next read

<Columns cols={2}>
  <Card title="Streaming configuration" icon="play" href="/inference-configuration/streaming-inference">
    Chunk size, aggregation policy, quality gates
  </Card>

  <Card title="BrainState" icon="brain" href="/personalizer/brain-state">
    App-facing intents, presets, and bridges
  </Card>

  <Card title="API Reference" icon="book" href="/python-sdk/api-reference">
    Full symbol reference
  </Card>

  <Card title="Real-time setup" icon="settings" href="/inference-configuration/real-time-setup">
    LSL, BrainFlow, and acquisition loops
  </Card>
</Columns>


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.