elektro

What is Elektro?

Elektro is the Python client for the Elektro electrophysiology backend — a service for storing, organizing and analysing ephys data (signals, recordings, stimuli, experiments) and the biophysical neuron models behind them.

It splits every object into two halves and keeps them in sync for you: the metadata (a dataset's name, its folder, the experiment it belongs to) lives in a GraphQL API, while the heavy numeric payload (the actual signal, arrays, tables, meshes) is streamed to and from object storage as Zarr and Parquet. You hand Elektro a NumPy array; it uploads the bytes, registers the metadata, and gives you back a typed object whose .data is a lazy xarray.DataArray — so you can ask for dataset.data.max() on a multi-gigabyte recording and only the bytes you touch are ever pulled down.

It answers questions like "store this recording and give it back to me as an array I can slice without downloading the whole thing" — and does it as one line of typed, async-first Python.

Note: Elektro is built for the Arkitekt ecosystem. Configuration and authentication come from Fakts, the transport from Rath, and its domain types plug straight into Arkitekt actions and workflows.

Installation

pip install elektro

Elektro requires Python ≥ 3.11. Optional extras pull in heavier, domain-specific dependencies only when you need them:

pip install "elektro[table]"   # pyarrow + pandas + duckdb — query Parquet stores
pip install "elektro[neuron]"  # the NEURON simulator — build & run biophysical models
pip install "elektro[sparse]"  # sporadik — upload spike rasters as sparse datasets

Inside an arkitekt app, pip install "arkitekt[rekuest,elektro]" installs it alongside arkitekt.

Scope

Elektro is responsible for the client side of ephys data and nothing else:

1. Modelling — it exposes the Elektro domain (array datasets, folders, experiments, neuron models, …) as typed, validated Python objects generated from the GraphQL schema.
2. Moving data — array-, table- and file-like values you pass into a mutation are transparently uploaded to object storage before the GraphQL call, and read back lazily afterwards. You work with NumPy/xarray; Elektro handles Zarr, Parquet and credentials.
3. Composing — it bundles a GraphQL client (rath) and an object-storage client (datalayer) into one Elektro app, usable sync or async.

It does not run the backend, define the schema, or mint your credentials — those come from the Elektro server and from Fakts / Arkitekt. Elektro is the piece that turns a configured, authenticated connection into typed objects and lazy arrays.

Core concepts

The domain types you work with (all registered as structures in elektro/arkitekt.py, beside the service):

Heavy payloads enter the API as store-backed scalars (elektro/scalars.py). You pass a plain in-memory value; Elektro uploads it and substitutes a store reference:

How it fits together

An Elektro app is a small composition of two clients (elektro/elektro.py):

        ┌──────────────────── Elektro ────────────────────┐
        │                                                  │
        │   rath: ElektroRath          datalayer: DataLayer│
        │   (GraphQL metadata)         (object storage)    │
        └───────┬──────────────────────────────┬──────────┘
                │                               │
   write path   │                               │
   create_array_dataset(data=np.ndarray, …)     │
                │                               │
        ┌───────▼────────┐   upload bytes   ┌───▼──────────────┐
        │ UploadMiddleware├─────────────────►│  Zarr / Parquet  │
        │ (intercepts the │                  │  object storage  │
        │  array variable)│◄─────────────────┤  (via obstore)   │
        └───────┬────────┘   store id        └───▲──────────────┘
                │                                 │
        ┌───────▼────────┐                        │  lazy reads
        │ GraphQL mutation│  carries store id      │
        │ via ElektroRath │                        │
        └───────┬────────┘                        │
                │                                  │
   read path    │  ArrayDataset                    │
      dataset.data  ──── DatasetTrait ───────────►│  xr.DataArray
                       (only the bytes you touch are pulled)

The write path: you call create_array_dataset(data=array, …); UploadMiddleware (elektro/middleware/upload.py) pulls every array out of the variables — the level-0 data and each scales entry alike — uploads them to Zarr stores through the datalayer, and the GraphQL mutation only ever carries the resulting store ids. The read path: the returned ArrayDataset mixes in DatasetTrait, whose .data property opens level 0's store as a lazy xarray.DataArray — nothing is downloaded until you compute. A store belongs to a pyramid level, not to the dataset: level_data(2) and multi_scale_data() reach the rest.

Quickstart

Every GraphQL operation is a method of the Elektro client, in a blocking and an a-prefixed async flavour (create_folder / acreate_folder, get_array_dataset / aget_array_dataset, …). Nothing is looked up behind your back: you call through the client you were given.

In an arkitekt app

Add elektro_service to your app and take the client by annotation — arkitekt injects it. elektro.specs holds lens types such as SingleChannelTrace that declare what an action needs:

import numpy as np
import xarray as xr

from arkitekt import App, Task, run
from elektro import Elektro, elektro_service
from elektro.specs import SingleChannelTrace, carried_axes, ensure

app = App("rectify", "0.1.0", services=[elektro_service])


@app.action
def rectify(trace: SingleChannelTrace, elektro: Elektro, task: Task) -> SingleChannelTrace:
    """Rectify Trace"""
    source = trace.data  # a lazy xarray.DataArray
    task.progress(50, f"Rectifying {source.shape}")
    result = elektro.create_array_dataset(
        data=xr.DataArray(np.abs(np.asarray(source)), dims=source.dims),
        scales=[],
        name="rectified",
        axes=carried_axes(trace, source.dims),
    )
    return ensure(result.lens(), SingleChannelTrace)


if __name__ == "__main__":
    run(app)

Datasets, lenses, folders, experiments, neuron models and the other domain types travel between actions by id (@elektro/arraydataset, @elektro/lens, …), so an action can take and return them directly.

From a script

import numpy as np
from arkitekt import easy
from elektro import elektro_service

with easy("my-script", elektro_service) as elektro:
    # Create a folder (metadata only)
    folder = elektro.create_folder(name="my_experiment")

    # Upload a 1-D signal — the array is stored in Zarr, the metadata in GraphQL.
    # `axes` is required: an axis is what gives a dimension a type.
    dataset = elektro.create_array_dataset(
        data=np.random.random((1000,)),
        scales=[],
        name="signal_1",
        axes=["t"],
        folder=folder.id,
    )

    dataset.id           # the new dataset's id
    dataset.data.shape   # (1000,) — a lazy xarray.DataArray, materialized on access

    # Fetch one back
    again = elektro.get_array_dataset(dataset.id)

In async code use async with aeasy("my-script", elektro_service) as elektro: and the async twins:

folder = await elektro.acreate_folder(name="my_experiment")
dataset = await elektro.acreate_array_dataset(
    data=np.random.random((1000,)), scales=[], name="signal_1", axes=["t"], folder=folder.id
)

Outside Arkitekt you can construct the client yourself — elektro = Elektro(rath=ElektroRath(...), datalayer=DataLayer(...)) — and use it as a context manager (with elektro: ...). See tests/conftest.py for a full manual wiring against a local deployment. An object a call returns remembers the client that fetched it, so its own follow-up calls (dataset.data, system.graph(), dataset.lens(...)) go through that client; trait methods that call the API also take an explicit elektro=.

Working with data

Lazy arrays. An ArrayDataset (anything mixing in DatasetTrait) exposes its payload as an xarray.DataArray that is only fetched on demand:

data = dataset.data               # level 0, as a lazy dask-backed xr.DataArray
peak = data.max().compute()       # pulls only what it needs
overview = dataset.level_data(3)          # a coarser pyramid level, same axis names
dataset.multi_scale_data()                # every level, finest first
window = dataset.lens(t=(0, 30_000), c=2) # a Lens: an immutable selection, in samples
window.data                               # the sliced xr.DataArray

Time lives in the graph, not in columns. elektro uses the model mikro uses for space, for time. A dataset lives in a unit-less sample grid; physical time enters once, as a clock plus one edge — a sampling law (t = sample · period + t_start) or, for irregular data, a time lookup. Offsets between clocks are edges too, so nothing stores a composed answer:

# CS first: the session is a clock, a signal is a dataset, the timing is one edge.
# What the values measure is a ValueUnit anchor, not an axis. Bare axis names cover
# the conventional ones ("t", "c", "sweep", ...).
clock = elektro.create_coordinate_system(
    name="session 12", registrations=[],
    axes=[PhysicalAxisInput(name="t", type=AxisType.TIME, unit="second")],
)
probe = elektro.create_array_dataset(
    data=np.zeros((90_000, 4)), scales=[], name="probe A", axes=["t", "c"],
    anchors=[CoordinateAnchorInput(axis_anchors=[], value_unit=ValueUnitInput(unit="uV"))],
)
# `source` is the grid being timed: a coordinate system id, never the dataset's own id.
# Passing the dataset *object* is fine -- it resolves to its grid client-side.
law = elektro.create_sampling_law(
    source=probe.intrinsic_system.id, clock=clock.id, sampling_rate="30 kHz", t_start="2 s",
)
law.as_matrix()                                  # [[1/30000, 0, 2], [0, 0, 1]]
probe.intrinsic_system.axis_values_in(clock, "t", 5)  # the first five sample times, in s

Declared axes are checked against the array before it is uploaded — a missing declaration, a rank mismatch and a transposed (c, t) are all refused client-side, as the server would refuse them afterwards.

Marks. An AnnotationCollection owns the space its shapes are drawn in; an edge says what they mark (a dataset's samples, a segment's clock, an experiment's world):

marks = elektro.create_annotation_collection(
    name="artifacts", axes=["t"],
    derived_from=[DatasetDerivedFromInput(dataset=probe.id, transform=IdentityTransformInput())],
)
elektro.create_annotation(kind=AnnotationKind.EPOCH, vectors=[[100.0], [250.0]], collection=marks.id)

Experiments are mikro's scenes for time: a world (usually a session clock) and layers over data the experiment does not own. A layer carries view state only; where it sits in time is the graph's answer, so there is no per-layer offset -- a segment's clock is placed once, with create_clock_offset, and everything timed on it moves together.

experiment = elektro.create_experiment_from_coordinate_system(coordinate_system=clock.id)
# ...or by hand, one layer per kind:
experiment = elektro.create_experiment(name="session 12", coordinate_system=clock.id)
elektro.create_trace_layer(experiment=experiment.id, dataset=probe.id, channel_index=2)
elektro.create_spikes_layer(experiment=experiment.id, sparse_dataset=raster.id,
                    color_bys=[ColorByInput(table=units.id, column="depth")])
elektro.create_events_layer(experiment=experiment.id, table_dataset=ttl.id, label_column="label")

experiment.data   # xr.Dataset: `traces` (trace, time), timed through the graph, world units

A session is a clock and the datasets timed onto it; create_session writes one sampling law per dataset in one call (time_unit is required client-side -- the server's default is seconds). A simulated run is a session too: there is no run object. Each output states what computed it as a simulation spoke (SimulationStateInput: the neuron model, dt, duration) beside its RecordingSite / StimulusSite, on the dataset's {} anchor; the server refuses outputs of one clock that disagree on it. A model's runs read back as get_neuron_model_sessions(model.id).sessions (clock + simulated datasets).

clock = elektro.create_session(name="sweep 1", datasets=[v_soma.id, i_inj.id], time_unit="millisecond",
                       sampling=SamplingInput(rate="40 kHz", t_start="0 ms"))

Spike rasters are sparse datasets over (unit, t): the unit axis identified by a units table, the TIME axis by nothing (a sampling law places it). Declarations are checked against the matrix before it is uploaded (elektro/sparse.py):

raster = elektro.create_sparse_dataset(
    name="spikes", store=[spikes.tocsr(), spikes.tocsc()],
    axes=[SparseAxisInput(name="unit", identified_by=[TableIdentifiesInput(table=units.id)]),
          SparseAxisInput(name="t", type=AxisType.TIME)],
)
elektro.create_sampling_law(source=raster.coordinate_system.id, clock=clock.id,
                    sampling_rate="30 kHz", t_start="0 s")

Tables. create_table_dataset takes a dict of columns, a DataFrame, an Arrow table or a parquet path; columns is partial -- the rest (names, DuckDB dtypes, order) is read off the data. A table reads back as a lazy relation (requires elektro[table]):

ttl = elektro.create_table_dataset(
    name="TTL", data={"t": onsets_s, "label": labels},
    columns=[ColumnInput(name="t", axis_type=AxisType.TIME, unit="second")],
)
ttl.data.df()                     # a duckdb relation over the parquet on S3, materialised

Files & meshes. Big-file and media stores download to disk on request:

path = bigfile_store.download()              # → local file path
path = media_store.download("frame.png")     # presigned download

Units. Physical quantities are real pint quantities via kanne (Duration, Frequency, ElectricPotential, Unit, …), coerced on the wire by CoercePintLink so the server always receives canonical units.

Neuron modelling

Elektro models biophysical neurons as a typed hierarchy — ModelConfig → Cell → Topology (sections & connections) → Biophysics (mechanisms & conductances) → Compartment. The input traits (elektro/traits.py) make these easy to build and inspect:

config.as_input()            # ModelConfig → its GraphQL input form
biophysics.as_dataframe()    # inspect compartments/mechanisms as a pandas DataFrame

With the elektro[neuron] extra installed, elektro/neuron/ parses NEURON .mod mechanism files and runs simulations directly from a ModelConfig. arun_simulation publishes the run as a session: each recorded array becomes its own ArrayDataset (recordings carry a simulation spoke, stimuli -- the run's input -- only their site), then create_session times them all onto one new clock with a sampling law. It returns a PublishedRun (clock, recordings, stimuli), a client-side bundle, not a server object.

In the Arkitekt ecosystem

• elektro_service (elektro/arkitekt.py) — registers Elektro as an Arkitekt service. It builds a fully wired Elektro client from a Fakts config (FaktsAuthLink for auth, AIOHttpLink/GraphQLWSLink for transport, a DataLayer for storage), so App(services=[elektro_service]) gets a ready-to-use client with no manual setup.
• Structures (the same module's registry) — the domain types (ArrayDataset, Lens, Folder, Experiment, NeuronModel, …) are registered as structures on an arkitekt-spec AppRegistry, each under an identifier like @elektro/arraydataset, so they can be passed in and out of actions and workflows (expand/shrink/search handled for you).

Under the hood the transport is Rath (ElektroRath, a link chain of file extraction, dicting, pint coercion, auth and an HTTP/WebSocket split), the composition is koil, and object storage goes through obstore + Zarr/Parquet.

Public API

from elektro import Elektro, elektro_service

# the building blocks of the client
from elektro.rath import ElektroRath
from elektro.datalayer import DataLayer

# generated types and inputs (operations are methods of `Elektro`)
from elektro.api.schema import ArrayDataset, Folder, CoordinateAnchorInput, SamplingInput

elektro.api.schema is generated from the GraphQL schema by turms and regenerated whenever the schema changes — it is the source of truth for the full set of available operations (as methods of ElektroApi, which Elektro mixes in) and types.

Development

uv sync
uv run pytest

Integration tests spin up a real Elektro + MinIO deployment with testcontainers and are marked @pytest.mark.integration; tests needing the NEURON simulator are marked @pytest.mark.neuron. Run just the fast suite with uv run pytest -m "not integration and not neuron".

See RELEASING.md for the semantic-release flow (main → stable, next → rc prereleases, N.x → maintenance).