kernel.sdk

Installation

pip install kernel_sdk

Requires Python 3.11 or newer, and the kortex acquisition driver running on the same machine. kortex is what talks to the hardware; the SDK is a client of kortex.

Connecting

With no arguments the client reads the websocket address and auth secret from the local kortex config, so there is nothing to configure:

from kernel.sdk import SdkClient

client = SdkClient()          # connects immediately, waits for Flow to finish booting
...
client.stop()                 # the client cannot be reused after this

To reach a specific kortex — a remote machine, or one without the config file — pass the address plus either a secret or a pre-encoded token:

client = SdkClient("ws://localhost:13254", secret=secret)

Pass autostart=False to construct without connecting, then call client.start(block=True, timeout=10) yourself. start() also subscribes to the USB and system-state metadata that the device-control methods check, and blocks while Flow boots if a device is attached. It is idempotent, so calling it on an already-connected client is harmless.

Quick start

from kernel.sdk import SdkClient, MomentNumber, Wavelength

client = SdkClient()
print("available modules:", client.get_available_modules())

def on_data(timestamps, data):
    # timestamps: (t,)   data: (t, 48, 3, 48, 6)
    print(timestamps.shape, data.shape)

future = client.on_moments(MomentNumber.First, Wavelength.Red, on_data)
...
future.cancel()
client.stop()

Shape reference

Every subscription hands your callback a NumPy array whose first axis is time. This table is the fastest way to see what you will receive.

Arrays are np.float32. The first callback carries however many samples were buffered when you subscribed; after that t is normally 1.

on_moments

One moment at one wavelength, with the noise floor and bad channels already removed. Axes are (time, source_module, source_id, detector_module, detector_id).

from kernel.sdk import MomentNumber, Wavelength

def on_data(timestamps, data):
    # mean photon count for module 0's own sources and detectors
    block = data[:, 0, :, 0, :]
    print(block.shape)   # (t, 3, 6)

client.on_moments(MomentNumber.Zeroth, Wavelength.Red, on_data)

Raises ValueError if that moment stream is not available, which normally means no Flow device is connected.

on_moments_by_module

A convenience wrapper that keeps only detectors on one module, averaged across that module's three sources — the short-separation channels. Pass a module id from get_available_modules().

def on_data(timestamps, data):
    print(data.shape)    # (t, 6) — one value per detector

client.on_moments_by_module(MomentNumber.First, Wavelength.Red, module=0, callback=on_data)

Here 0.0 rather than -inf marks a detector with no data. If every detector reads 0.0, that module is not connected or is having problems.

on_all_moments new

All six moment streams (3 moments × 2 wavelengths) in a single callback, already time-aligned by kortex. This is the one to reach for when you want the full picture: you get them all, every time, with no client-side buffering and no reconciling timestamps across six separate subscriptions.

from kernel.sdk import MomentNumber, Wavelength

def on_data(timestamps, data):
    # (time, moment, wavelength, source_module, source_id, detector_module, detector_id)
    print(data.shape)                              # (t, 3, 2, 48, 3, 48, 6)
    first_red = data[:, MomentNumber.First, 0]     # (t, 48, 3, 48, 6)
    second_ir = data[:, MomentNumber.Second, 1]

client.on_all_moments(on_data)

Raises ValueError if any of the six streams is unavailable — it will not silently give you a partial stack.

on_eeg

EEG voltage and impedance together. Axes are (time, channel_id, data_type), where the last axis is 0 for voltage in volts and 1 for impedance in ohms.

def on_data(timestamps, data):
    voltage   = data[:, :, 0]     # (t, 32) volts
    impedance = data[:, :, 1]     # (t, 32) ohms

client.on_eeg(on_data)

The channel axis is a fixed 32 wide regardless of how many channels the attached hardware populates; unpopulated channels read -inf. Use get_eeg_channels() to map indices to labels.

on_retained_channels

Per-module fraction of channels currently passing quality checks, from 0.0 to 1.0. This is the signal behind the coupling check — a good proxy for headset fit. -1.0 means no data for that module.

def on_data(timestamps, retained):
    print(retained[-1, modules])   # latest sample, only the modules you care about

client.on_retained_channels(on_data)

on_task

Events emitted by task code using TaskClient — see the TaskClient reference. The callback receives a dict with two parallel lists.

def on_events(timestamps, events):
    for name, value in zip(events["events"], events["values"]):
        print(name, value)     # e.g. "start_trial" 3

client.on_task(on_events)

on_flow_state new

React to the device changing state instead of polling get_flow_system_state(). The callback receives a FlowSystemState, or None before the device has reported one.

from kernel.sdk import FlowSystemState

def on_state(state):
    if state is FlowSystemState.Faulted:
        print("Flow faulted — power cycle required")

futures = client.on_flow_state(on_state)
...
for f in futures:
    f.cancel()

Note this returns a list of futures, unlike the data subscriptions.

Callbacks & threading

Keep callbacks to a handful of microseconds: push onto a queue.Queue and do the real work on your own thread or process.

import queue, threading

packets = queue.Queue()
client.on_all_moments(lambda ts, data: packets.put((ts, data)))

def consume():
    while True:
        ts, data = packets.get()
        expensive_analysis(ts, data)

threading.Thread(target=consume, daemon=True).start()

Callbacks run on a shared worker pool, so they may run concurrently with each other and are not called on your main thread. Guard any shared state you touch.

get_available_modules

The module ids present on the connected headset. Use these to index the module axes of moments arrays and as the module argument to on_moments_by_module.

modules = client.get_available_modules()   # e.g. [0, 1, 2, 5, 8, ...]

Raises FlowDisconnected, FlowNotBooted, or FlowFaulted when the device is not ready, and TimeoutError if the metadata does not arrive within a second.

EEG channels & layout new

get_eeg_channels() returns channel labels in the same order as the channel axis of on_eeg data. get_eeg_layout() returns (label, x, y) per channel for topographic plotting.

labels = client.get_eeg_channels()          # ["Fp1", "Fp2", ...]
layout = client.get_eeg_layout()            # [("Fp1", 0.31, 0.86), ...]

def on_data(timestamps, data):
    for i, label in enumerate(labels):
        print(label, data[-1, i, 0])        # latest voltage per named channel

Raises ValueError if neither source reports any EEG channels.

Flow probe geometry new

Where the Flow sources and detectors sit. get_flow_locs() gives true 3-D head-space positions; get_flow_layout() gives the flattened 2-D projection used for topographic plots. Both return the same structure:

locs = client.get_flow_locs()
for module_id, coords in locs["sources"].items():
    print(module_id, coords)      # module id is an int; coords is (3, 3) xyz

layout = client.get_flow_layout()
xy = layout["detectors"][0]       # (6, 2) for module 0

Module ids are integers, and only populated modules appear. NaN marks a source or detector position the device did not report. Values from a .patch.json in the kortex pre-session directory, if present, override what the device reports.

Device state

The two methods fetch state on demand (with a cached fast path); the two properties read the cached value without blocking. To be notified of changes rather than polling, use on_flow_state().

Initialize, tune, lasers

initialize_flow() is run once after powering on the device; tune() is run after the headset is on the participant. Both take a CommandTarget that must match between them.

from kernel.sdk import CommandTarget

client.initialize_flow(org_id, api_key)                      # uploads to Kernel
client.initialize_flow(target=CommandTarget.LocalOffline)    # stays on this machine

client.tune()
client.turn_lasers_on()
...
client.turn_lasers_off()

org_id and api_key are required for every target except LocalOffline, which skips the cloud handshake entirely.

Coupling check

An interactive helper that samples retained channels and prompts the operator to adjust the headset until the fit is good. Returns True on good coupling, False if the operator chooses to continue anyway, and raises RuntimeError if they abort.

Online workflow

An end-to-end example that walks an operator through initializing, tuning, checking coupling, and streaming moments, uploading to Kernel. It prompts at each step with input().

from kernel.sdk import run_workflow

run_workflow(org_id="...", api_key="...")

Treat it as a worked example to copy and adapt rather than as production API — it hardcodes one module and prints data to stdout.

Offline workflow new

The same workflow with nothing leaving the machine. It initializes and tunes with CommandTarget.LocalOffline, records to local disk, waits for kortex's offline pipeline, and returns the path to the generated SNIRF file. No organization id or API key is needed.

from kernel.sdk import run_workflow_offline

snirf_path = run_workflow_offline(subject_id="P01", duration=300)
print(snirf_path)   # .../snirf_output/....snirf

Driving a recording yourself

If you want the recording without the prompts, use the three pieces directly:

from kernel.sdk import SdkClient, CommandTarget

client = SdkClient()
try:
    client.initialize_flow(target=CommandTarget.LocalOffline)
    client.tune(target=CommandTarget.LocalOffline)
    client.turn_lasers_on()

    session_id = client.start_offline_recording("P01")
    ...                                     # run your task here
    client.stop_offline_recording()
finally:
    client.turn_lasers_off()

snirf_path = client.wait_for_offline_snirf(session_id)

Pass duration in seconds to have kortex stop writing on its own; otherwise call stop_offline_recording(). Either way the SNIRF file is produced asynchronously, so wait_for_offline_snirf() is what tells you it is ready. It raises RuntimeError if the pipeline reports an error or finds no session data.

Enums

Exceptions

Everything the SDK raises derives from SDKError, so a single except SDKError catches all of it.

from kernel.sdk import SDKError, FlowFaulted

try:
    client.tune()
except FlowFaulted:
    print("power cycle Flow")
except SDKError as e:
    print("sdk error:", e)

EventFuture

Every on_* method returns an EventFuture — the handle for the subscription it created.

future = client.on_eeg(on_data)
...
future.cancel()

on_flow_state() returns a list of futures; cancel each one.