NeuroStream-Edge

A modular, interface-driven runtime for continuous biosignal processing and real-time inference.

---

Overview

NeuroStream-Edge is an open-source runtime for building real-time biosignal processing pipelines. It focuses on the infrastructure between a continuous sensor stream and machine learning inference — providing modular components for streaming, buffering, window generation, runtime orchestration, storage, and performance analysis.

The runtime is designed to support multiple biosignal modalities — EEG, ECG, EMG, and PPG — while remaining independent of any specific downstream task or machine-learning architecture.

---

Highlights

• Interface-driven architecture — every layer exposes an abstract base class. The runtime depends only on interfaces, never on concrete implementations.
• Streaming-first, fixed-memory execution — pre-allocated ring buffers with O(1) writes and circular overwrite semantics. No growth, no GC pressure in the hot path.
• Model-agnostic & signal-agnostic — bring your own ONNX / TorchScript model or implement the InferenceEngine interface.
• Minimal core dependencies — the runtime requires only NumPy. Every heavy backend (ONNX, PyTorch, BrainFlow, LSL) is an optional extra with a clear ImportError and lazy import.
• Observable by design — a decoupled Observer-based event system feeds a logger, profiler, and metrics collector without coupling into the execution path.
• Online runtime and offline pipeline are intentionally separated — reflecting the different engineering requirements of low-latency inference and large-scale dataset processing.

---

Status

Alpha. All eight core modules are implemented and covered by 95 passing tests (~78% line coverage). The runtime is functional end-to-end with a reference simulated pipeline, but is not yet benchmarked against external baselines and the public API is still stabilising. See the Roadmap for what is planned.

Module	Interface	Role	Status
acquisition	SignalSource	Continuous sample streams from sensors or simulators	✅
streaming	Buffer	Fixed-capacity, overwrite-on-full buffers	✅
windowing	WindowGenerator	Overlapping inference-ready windows	✅
runtime	—	Pipeline orchestration + event system	✅
models	InferenceEngine	Model-agnostic prediction backends	✅
storage	StorageBackend	Offline dataset recording / playback	✅
observability	Observer	Decoupled logging, profiling, metrics	✅
benchmarking	—	Throughput & latency characterisation	✅

---

Installation

From PyPI

pip install neurostream-edge

From source

git clone https://github.com/tarunc997/NeuroStream-Edge.git
cd NeuroStream-Edge
pip install -e .

Optional extras

Heavy backends are never installed automatically — install only what you need:

pip install "neurostream-edge[onnx]"      # ONNX Runtime
pip install "neurostream-edge[torch]"     # TorchScript
pip install "neurostream-edge[brainflow]" # BrainFlow hardware adapters
pip install "neurostream-edge[lsl]"       # Lab Streaming Layer
pip install "neurostream-edge[dev]"       # pytest, pytest-cov

Extras can be combined, e.g. pip install "neurostream-edge[onnx,dev]".

---

Quick start

A minimal online pipeline: SimulatedSignalSource → RingBuffer → SlidingWindowGenerator → DummyInferenceEngine.

import time
import neurostream as ns

# 1. Build components (dependency injection)
source  = ns.SimulatedSignalSource(sample_rate=250.0, n_channels=8, seed=42)
buffer  = ns.RingBuffer(capacity=2500, n_channels=8)           # 10 s @ 250 Hz
window  = ns.SlidingWindowGenerator(buffer, window_size=250,   # 1 s window
                                    stride=50)                 # 0.2 s stride
engine  = ns.DummyInferenceEngine(n_outputs=3)

# 2. Optional observability
bus = ns.EventBus()
metrics  = ns.Metrics()
profiler = ns.Profiler()
bus.subscribe("*", metrics)
bus.subscribe("*", profiler)

# 3. Wire everything into the runtime
runtime = ns.Runtime(source=source, buffer=buffer, window=window,
                     inference=engine, event_bus=bus, read_chunk=10)

# 4. Run
runtime.start()
runtime.run_for(2.0)   # run for 2 seconds
runtime.stop()

print(f"Predictions: {len(runtime.predictions)}")
print(f"Throughput : { {k: f'{v:.1f}/s' for k, v in metrics.throughput().items()} }")

More examples live in examples/:

• online_pipeline.py — full online inference pipeline
• offline_recording.py — record a stream to disk and replay it
• benchmark_buffers.py — compare buffer backends

---

Architecture

The online inference runtime and offline data pipeline are intentionally separated, reflecting the different engineering requirements of low-latency inference and large-scale dataset processing.

Online runtime

Continuous Signal
        │
        ▼
Acquisition Layer
        │
        ▼
Streaming Layer
 (Ring Buffers)
        │
        ▼
Windowing Layer
        │
        ▼
Inference Runtime
        │
        ▼
Model Interface
        │
        ▼
Prediction

Pipeline context

                Runtime (orchestrator)
                        │
       ┌────────────────┼────────────────┐
       ▼                ▼                ▼
 Acquisition       Streaming        Windowing
 (SignalSource)     (Buffer)     (WindowGenerator)
       │                │                │
       └────────────────┼────────────────┘
                        ▼
                 Models (InferenceEngine)
                        │
                  Storage (offline)

   Observability (Logger · Profiler · Metrics) ← Runtime events
   Benchmarking  (per-component measurement)

Design principles

• Streaming-first — designed around continuous signals, not static datasets.
• Interface-driven — the runtime depends only on abstractions; components are interchangeable (Strategy pattern).
• Composition over inheritance — no per-modality EEGRuntime / ECGRuntime classes; one runtime serves every signal type.
• Fixed-memory execution — storage is allocated once and never grows.
• Storage-independent & model-agnostic — backends are pluggable.

Allocation note: the streaming write path is O(1) and allocation-free for contiguous writes. Read paths (read(), latest()) allocate copies to avoid exposing internal buffers.

---

Repository structure

docs/                # architectural & per-module reference
examples/            # runnable scripts
neurostream/
├── acquisition/     # signal sources & adapters
├── streaming/       # fixed-memory buffers
├── windowing/       # sliding-window generation
├── runtime/         # pipeline orchestration & events
├── models/          # inference engines & factory
├── storage/         # dataset backends
├── observability/   # logger, profiler, metrics
└── benchmarking/    # benchmarks & latency profiling
tests/               # 95 tests

---

Documentation

Detailed per-module guides cover each layer's purpose, its abstract interface, the concrete implementations and their trade-offs, and runnable usage examples.

Module	Guide
Acquisition	docs/modules/acquisition.md
Streaming	docs/modules/streaming.md
Windowing	docs/modules/windowing.md
Models	docs/modules/models.md
Storage	docs/modules/storage.md
Runtime	docs/modules/runtime.md
Observability	docs/modules/observability.md
Benchmarking	docs/modules/benchmarking.md

Start at docs/index.md for the architecture overview.

---

Testing

The suite uses real concurrency — actual background threads, real timing, real predictions — rather than mocks, so integration behaviour is exercised directly.

pip install "neurostream-edge[dev]"
pytest

With a coverage report:

pytest --cov=neurostream --cov-report=term-missing

---

Scope

NeuroStream-Edge is infrastructure, not an end-user application. It is not tied to a specific machine-learning task such as seizure detection, sleep staging, emotion recognition, or motor imagery. Instead, it provides the runtime required to support continuous biosignal inference across multiple research and deployment scenarios.

---

Roadmap

• End-to-end reference pipeline on a public dataset (e.g. TUH EEG, BCI Comp IV)
• Benchmark against a naive baseline using the built-in benchmarking/ layer
• Convenience quickstart() constructor for low-friction onboarding
• Config-driven pipelines + neurostream run CLI (YAML → wired runtime)
• Plugin entry points ([project.entry-points]) for third-party engines/sources
• Explicit backpressure / drop policies (block / drop-newest / drop-oldest)
• Source & storage factories to match the existing InferenceFactory
• Async / queued EventBus option to fully decouple observers from the tick
• CITATION.cff and first tagged release

---

Contributing

Contributions are welcome. The project follows an interface-driven architecture — every layer exposes an abstract base class, and the runtime depends only on interfaces. Before adding new functionality, open an issue to discuss scope.

When contributing, please:

1. Keep the runtime free of concrete-component dependencies (depend on interfaces).
2. Add or update tests for any behavioural change.
3. Run pytest and ensure the full suite passes before submitting a PR.
4. Follow the existing numpy-style docstring conventions.

---

License

MIT License © Tarun Chilkur