Real-time Inference Setup

Setting up for real-time BCI applications requires proper configuration to achieve 10-25ms latency while maintaining accuracy. This guide covers essential setup steps and best practices for both Python and Julia SDKs.

Both SDKs support Bayesian models: Bayesian LDA, Bayesian QDA, Bayesian Softmax, and Bayesian STS (Python SDK only). Choose based on your accuracy, latency, and adaptivity requirements.

Quick Start

Minimal setup for real-time streaming:

Python
Julia

from nimbus_bci import NimbusLDA, StreamingSession
from nimbus_bci.data import BCIMetadata

# 1. Train classifier
clf = NimbusLDA()
clf.fit(X_train, y_train)

# 2. Configure streaming metadata
metadata = 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"
)

# 3. Initialize streaming session
session = StreamingSession(clf.model_, metadata)

# 4. Process chunks in real-time
for chunk in eeg_stream:
    result = session.process_chunk(chunk)
    execute_command(result.prediction, result.confidence)

using NimbusSDK

# 1. Authenticate
NimbusSDK.install_core("your-api-key")

# 2. Load model
model = load_model(NimbusLDA, "motor_imagery_4class_v1")

# 3. Configure streaming metadata
metadata = BCIMetadata(
    sampling_rate = 250.0,
    paradigm = :motor_imagery,
    feature_type = :csp,
    n_features = 16,
    n_classes = 4,
    chunk_size = 250  # 1 second chunks at 250 Hz
)

# 4. Initialize streaming session
session = init_streaming(model, metadata)

# 5. Process chunks in real-time
for chunk in eeg_stream
    result = process_chunk(session, chunk)
    execute_command(result.prediction, result.confidence)
end

Chunk Size Selection

The chunk_size parameter determines latency and accuracy tradeoff:

Fast Response (0.5 second chunks)

metadata = BCIMetadata(
    sampling_rate = 250.0,
    chunk_size = 125,  # 0.5s at 250 Hz
    # ... other params
)

Latency: ~10-15ms per chunk
Update rate: Every 0.5 seconds
Use case: Fast paced games, quick reactions
Trade-off: Less data per prediction, potentially lower confidence

Standard (1 second chunks) - Recommended

metadata = BCIMetadata(
    sampling_rate = 250.0,
    chunk_size = 250,  # 1s at 250 Hz
    # ... other params
)

Latency: ~15-20ms per chunk
Update rate: Every second
Use case: Most BCI applications
Trade-off: Good balance of speed and accuracy

Accurate (2 second chunks)

metadata = BCIMetadata(
    sampling_rate = 250.0,
    chunk_size = 500,  # 2s at 250 Hz
    # ... other params
)

Latency: ~20-25ms per chunk
Update rate: Every 2 seconds
Use case: High-stakes decisions, medical applications
Trade-off: Higher confidence but slower updates

Model Selection for Real-Time

Bayesian LDA vs Bayesian QDA vs Bayesian STS

Python
Julia

from nimbus_bci import NimbusLDA, NimbusQDA, NimbusSTS

# Bayesian LDA: Fastest, recommended for stationary real-time
clf_lda = NimbusLDA()
# Typical latency: 10-20ms
# Use when: Speed is priority, classes are well-separated, data is stationary

# Bayesian QDA: More flexible, slightly slower
clf_qda = NimbusQDA()
# Typical latency: 15-25ms
# Use when: Need flexibility for overlapping classes, data is stationary

# Bayesian STS: Adaptive, for non-stationary data
clf_sts = NimbusSTS()
# Typical latency: 20-30ms
# Use when: Long sessions (>30min), data drifts over time, need temporal adaptation

using NimbusSDK

# NimbusLDA: Faster, recommended for real-time
lda_model = load_model(NimbusLDA, "motor_imagery_4class_v1")
# Typical latency: 10-20ms
# Use when: Speed is priority, classes are well-separated

# NimbusQDA: More flexible, slightly slower
gmm_model = load_model(NimbusQDA, "motor_imagery_4class_gmm_v1")
# Typical latency: 15-25ms
# Use when: Need flexibility for overlapping classes

# Note: NimbusSTS (Bayesian STS) is currently available in Python SDK only

For most real-time applications with stationary data, Bayesian LDA (NimbusLDA) is recommended due to its faster inference while maintaining good accuracy. For long sessions with non-stationary data, consider NimbusSTS (Python SDK).

Aggregation Strategies

Choose how to combine multiple chunks for final decisions:

Weighted Vote (Recommended)

# High-confidence chunks contribute more
final_result = finalize_trial(session; method=:weighted_vote)

Best for: Most applications
Advantage: Balances all chunks by confidence
Typical accuracy: Best overall performance

Majority Vote

# Each chunk has equal weight
final_result = finalize_trial(session; method=:majority_vote)

Best for: When all chunks have similar quality
Advantage: Simple, robust to outliers
Typical accuracy: Good with consistent signal

Latest Chunk Only

# Use only the most recent chunk
final_result = finalize_trial(session; method=:latest)

Best for: Ultra-responsive applications
Advantage: Fastest response to state changes
Typical accuracy: Lower, but most reactive

Performance Optimization

JIT Compilation Warmup

Julia’s JIT compiler makes the first inference slower. Warm up before real use:

using NimbusSDK

# Load model
model = load_model(NimbusLDA, "motor_imagery_4class_v1")
session = init_streaming(model, metadata)

# Warmup: Run a few dummy inferences
println("Warming up...")
dummy_chunk = randn(16, 250)  # Same shape as real data
for i in 1:10
    process_chunk(session, dummy_chunk)
end
println("✓ Ready for real-time processing")

# Now process real data (much faster)
for chunk in eeg_stream
    result = process_chunk(session, chunk)  # Fast after warmup
end

Memory Pre-allocation

Avoid allocations in the real-time loop:

using NimbusSDK

# Pre-allocate buffers
chunk_buffer = zeros(Float64, 16, 250)
results_buffer = Vector{StreamingResult}(undef, 100)

# Real-time loop with minimal allocation
for (i, raw_chunk) in enumerate(eeg_stream)
    # Reuse pre-allocated buffer
    chunk_buffer .= raw_chunk
    
    # Process (minimal allocation)
    result = process_chunk(session, chunk_buffer)
    results_buffer[i] = result
    
    # Handle result
    handle_bci_command(result)
end

Confidence Thresholding

Only act on high-confidence predictions:

# Configure threshold based on application
confidence_threshold = 0.75  # Adjust based on needs

for chunk in eeg_stream
    result = process_chunk(session, chunk)
    
    # Only execute if confidence exceeds threshold
    if result.confidence > confidence_threshold
        execute_command(result.prediction)
    else
        # Optional: provide feedback about low confidence
        display_uncertain_state()
    end
end

Real-Time Monitoring

Performance Tracking

Monitor system performance during operation:

using NimbusSDK
using Statistics

# Initialize tracker
latencies = Float64[]
confidences = Float64[]

for (i, chunk) in enumerate(eeg_stream)
    # Time the inference
    t_start = time()
    result = process_chunk(session, chunk)
    t_end = time()
    
    latency_ms = (t_end - t_start) * 1000
    push!(latencies, latency_ms)
    push!(confidences, result.confidence)
    
    # Report every 50 chunks
    if i % 50 == 0
        println("\n=== Performance Report ===")
        println("Mean latency: $(round(mean(latencies[end-49:end]), digits=1)) ms")
        println("Max latency: $(round(maximum(latencies[end-49:end]), digits=1)) ms")
        println("Mean confidence: $(round(mean(confidences[end-49:end]), digits=3))")
    end
end

Quality Monitoring

Track signal quality in real-time:

using NimbusSDK

# Track confidence trends
confidence_window = CircularBuffer{Float64}(20)

for chunk in eeg_stream
    result = process_chunk(session, chunk)
    
    # Add to rolling window
    push!(confidence_window, result.confidence)
    
    # Check for degrading quality
    if length(confidence_window) >= 20
        recent_conf = mean(confidence_window)
        
        if recent_conf < 0.6
            @warn "Signal quality degrading"
            println("Recent mean confidence: $(round(recent_conf, digits=2))")
            println("Recommendation: Check electrode contact")
        end
    end
end

Application-Specific Setups

Motor Imagery Game Control

using NimbusSDK

# Setup for gaming
model = load_model(NimbusLDA, "motor_imagery_4class_v1")

# Fast chunks for responsive gameplay
metadata = BCIMetadata(
    sampling_rate = 250.0,
    chunk_size = 125,  # 0.5s for quick response
    paradigm = :motor_imagery,
    feature_type = :csp,
    n_features = 16,
    n_classes = 4
)

session = init_streaming(model, metadata)

# Game control loop
for chunk in game_eeg_stream
    result = process_chunk(session, chunk)
    
    # Lower threshold for responsive gameplay
    if result.confidence > 0.65
        update_game_state(result.prediction)
    end
end

P300 Speller

using NimbusSDK

# P300 detection setup
model = load_model(NimbusLDA, "p300_binary_v1")

# Post-stimulus window
metadata = BCIMetadata(
    sampling_rate = 250.0,
    chunk_size = 200,  # 0.8s post-stimulus
    paradigm = :p300,
    feature_type = :erp,
    n_features = 12,
    n_classes = 2  # target vs non-target
)

session = init_streaming(model, metadata)

# Process stimulus presentations
for (stimulus, chunk) in p300_stream
    result = process_chunk(session, chunk)
    
    # Higher threshold for spelling accuracy
    if result.prediction == 1 && result.confidence > 0.85
        selected_character = stimulus
        println("Selected: $selected_character")
    end
end

Wheelchair Control (Safety-Critical)

using NimbusSDK

# Safety-critical setup
model = load_model(NimbusLDA, "motor_imagery_4class_v1")

# Longer chunks for higher confidence
metadata = BCIMetadata(
    sampling_rate = 250.0,
    chunk_size = 500,  # 2s for safety
    paradigm = :motor_imagery,
    feature_type = :csp,
    n_features = 16,
    n_classes = 4
)

session = init_streaming(model, metadata)

# Safety-critical control loop
for chunk in wheelchair_stream
    result = process_chunk(session, chunk)
    
    # Very high threshold for safety
    if result.confidence > 0.90
        execute_wheelchair_command(result.prediction)
    else
        # Require confirmation for low confidence
        request_manual_confirmation(result.prediction, result.confidence)
    end
end

Error Handling

Robust Real-Time Loop

using NimbusSDK

session = init_streaming(model, metadata)
error_count = 0
max_errors = 10

while is_bci_active()
    try
        chunk = get_next_chunk()
        
        # Process with timeout protection
        result = process_chunk(session, chunk)
        
        # Reset error counter on success
        error_count = 0
        
        # Handle result
        handle_bci_result(result)
        
    catch e
        error_count += 1
        @error "Chunk processing failed" exception=e error_count
        
        if error_count >= max_errors
            @error "Too many consecutive errors - stopping"
            break
        end
        
        # Brief pause before retry
        sleep(0.1)
    end
end

System Requirements

Minimum Requirements

CPU: Modern multi-core processor (Intel i5 or equivalent)
RAM: 4GB available
Julia: Version 1.9 or later
OS: Windows, macOS, or Linux

Recommended for Real-Time

CPU: Intel i7 or equivalent (or better)
RAM: 8GB available
Julia: Version 1.10+
Storage: SSD for faster model loading
OS: Linux for best real-time performance

Latency Expectations

Hardware	NimbusLDA Latency	NimbusQDA Latency
i5, 8GB RAM	15-25ms	20-30ms
i7, 16GB RAM	10-20ms	15-25ms
High-end workstation	10-15ms	12-20ms

These are per-chunk latencies. Total system latency includes signal acquisition and any application-specific processing.

Troubleshooting

High Latency Issues

# Profile to find bottlenecks
latencies = []

for chunk in test_stream[1:100]
    t_start = time()
    result = process_chunk(session, chunk)
    latency = (time() - t_start) * 1000
    
    push!(latencies, latency)
end

println("Statistics:")
println("  Mean: $(round(mean(latencies), digits=1)) ms")
println("  Median: $(round(median(latencies), digits=1)) ms")
println("  95th percentile: $(round(quantile(latencies, 0.95), digits=1)) ms")
println("  Max: $(round(maximum(latencies), digits=1)) ms")

# First chunk is usually slower (JIT compilation)
println("\nExcluding first chunk:")
println("  Mean: $(round(mean(latencies[2:end]), digits=1)) ms")

Common Solutions:

Run warmup inferences before real use
Use NimbusLDA instead of NimbusQDA for speed
Reduce chunk_size if acceptable
Check system resource usage
Close other applications

Low Confidence Issues

# Diagnose signal quality
using NimbusSDK

# Collect diagnostic trial
diagnostic_data = collect_diagnostic_trial()

# Check preprocessing quality
report = diagnose_preprocessing(diagnostic_data)

println("Preprocessing quality: $(round(report.quality_score * 100, digits=1))%")

if !isempty(report.warnings)
    println("\nWarnings:")
    for warning in report.warnings
        println("  • $warning")
    end
end

if !isempty(report.recommendations)
    println("\nRecommendations:")
    for rec in report.recommendations
        println("  • $rec")
    end
end

Next Read

Streaming Inference

Complete streaming inference guide

Batch Processing

Offline analysis workflows

Error Handling

Robust error recovery patterns

Code Examples

Working real-time examples

Tip: Start with standard 1-second chunks and adjust based on your application’s speed/accuracy requirements.

Getting Started

Core Concepts

Models & Algorithms

Configuration & Setup

Examples

Development

Python SDK

Julia SDK

​Real-time Inference Setup

​Quick Start

​Chunk Size Selection

​Fast Response (0.5 second chunks)

​Standard (1 second chunks) - Recommended

​Accurate (2 second chunks)

​Model Selection for Real-Time

​Bayesian LDA vs Bayesian QDA vs Bayesian STS

​Aggregation Strategies

​Weighted Vote (Recommended)

​Majority Vote

​Latest Chunk Only

​Performance Optimization

​JIT Compilation Warmup

​Memory Pre-allocation

​Confidence Thresholding

​Real-Time Monitoring

​Performance Tracking

​Quality Monitoring

​Application-Specific Setups

​Motor Imagery Game Control

​P300 Speller

​Wheelchair Control (Safety-Critical)

​Error Handling

​Robust Real-Time Loop

​System Requirements

​Minimum Requirements

​Recommended for Real-Time

​Latency Expectations

​Troubleshooting

​High Latency Issues

​Low Confidence Issues

​Next Read

Streaming Inference

Batch Processing

Error Handling

Code Examples

Real-time Inference Setup

Quick Start

Chunk Size Selection

Fast Response (0.5 second chunks)

Standard (1 second chunks) - Recommended

Accurate (2 second chunks)

Model Selection for Real-Time

Bayesian LDA vs Bayesian QDA vs Bayesian STS

Aggregation Strategies

Weighted Vote (Recommended)

Majority Vote

Latest Chunk Only

Performance Optimization

JIT Compilation Warmup

Memory Pre-allocation

Confidence Thresholding

Real-Time Monitoring

Performance Tracking

Quality Monitoring

Application-Specific Setups

Motor Imagery Game Control

P300 Speller

Wheelchair Control (Safety-Critical)

Error Handling

Robust Real-Time Loop

System Requirements

Minimum Requirements

Recommended for Real-Time

Latency Expectations

Troubleshooting

High Latency Issues

Low Confidence Issues

Next Read