Python Streaming Inference

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

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")

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()

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()