Skip to content

API.md

Latest commit

 

History

History
407 lines (309 loc) · 10.6 KB

API.md

File metadata and controls

407 lines (309 loc) · 10.6 KB
Raw

Viewer API Cheatsheet

viewer is a local Python/ImGui app for synchronized time-based data browsing.

Public Imports

Most demo code imports from the top-level package:

from viewer import (
    show,
    Cache,
    BaseStream,
    Stream,
    Span,
    TimeSeries,
    TraceView,
    Ephys,
    EphysView,
    ProbeView,
    HeatmapView,
    Spikes,
    RasterView,
    Video,
    VideoView,
    EventBars,
)

Transforms and data-loading helpers live in submodules:

from viewer.transforms import CAR, Lowpass, Highpass, Bandpass, FFT
from viewer.utils import load_env, load_bin, load_prb, load_probeinterface, tsg_to_spikes

Run examples from the repository root:

uv run python demo_timeseries.py
uv run python demo_raster.py
uv run python demo_probe.py
uv run python demo_heatmap.py
uv run python demo_video.py
uv run python demo_ephys_fft.py

Some demos expect local data under scripts/, and the ephys demos use paths from .env.

Core Pattern

Every app is a list of (stream, view) pairs:

from viewer import TimeSeries, TraceView, show

signal = TimeSeries(
    name="pupil",
    values=pupil_values,      # shape: (samples,) or (samples, channels)
    ts=pupil_ts,              # regular timestamps in seconds
    fs=30.0,
    chunk_samples=300,
)

show([(signal, TraceView())], title="Pupil", span=10.0)

show(...) links the x-axis across all stream plots, adds transport controls, loads chunks in background worker threads, and opens a settings panel for each stream.

show(
    streams,                         # list[tuple[Stream, View]]
    title="Viewer",
    window_size=(1480, 900),
    event_bars=None,
    overlays=(),
    span=2.0,                        # initial visible seconds
    max_workers=3,                   # chunk-loading threads
)

Stream names must be unique. Time is in seconds. The cache keeps the current chunk plus the previous and next chunks for each visible stream.

Time Series Traces

Use TimeSeries for regular sampled continuous data backed by NumPy, Zarr, memmap, or any array-like object with shape, dtype, and slicing.

import zarr
from viewer import TimeSeries, TraceView, show

root = zarr.open("scripts/exp1.zarr", mode="r")
ephys_grp = root["ephys"]
pupil_grp = root["behavior/pupil"]

ephys = TimeSeries(
    "ephys",
    values=ephys_grp["values"],
    ts=ephys_grp["ts"],
    fs=ephys_grp.attrs["fs"],
    chunk_samples=ephys_grp["values"].chunks[0],
)

pupil = TimeSeries(
    "pupil",
    values=pupil_grp["values"],
    ts=pupil_grp["ts"],
    fs=pupil_grp.attrs["fs"],
    chunk_samples=pupil_grp["values"].chunks[0],
)

show([(ephys, TraceView()), (pupil, TraceView())])

TraceView(width=1.0, gain=1.0, spacing=1.0) draws each channel as a separate row. gain scales the signal and spacing controls channel offsets.

Spike Rasters

Use Spikes for sorted spike times and matching integer unit labels.

from viewer import Spikes, RasterView, show

spikes = Spikes(
    name="units",
    ts=spike_times,                  # sorted 1-D seconds
    spike_units=spike_units,         # same length as spike_times
    chunk_duration=10.0,
    unit_ids=unit_ids,               # optional display/order list
)

show([(spikes, RasterView(metadata=unit_metadata, sort_by="unit_display_y"))])

RasterView accepts numeric metadata for sorting and coloring:

RasterView(
    metadata={
        "unit_ids": unit_ids,
        "rate": firing_rates,
        "unit_display_y": display_y,
    },
    unit_ids=unit_ids,
    tick_height=5.0,
    width=1.0,
    cmap="cmocean:phase",
    sort_by="unit_display_y",
    color_by="rate",
)

Metadata values can be arrays aligned with unit_ids, or dict-like objects keyed by unit id. If rate is present, RasterView uses it as the default color field.

Video Plus Raster

Use Video for a file-backed video stream. Pair it with VideoView, then add other streams with matching time bases.

from pathlib import Path
import numpy as np
from viewer import Video, VideoView, Spikes, RasterView, show

video = Video(
    "wake video",
    Path("scripts/data/A5044-240404A_wake.avi"),
    chunk_duration=1.0,
    scale=0.5,
)

keep = spike_times <= video.t_max
spikes = Spikes(
    "units",
    ts=spike_times[keep],
    spike_units=spike_units[keep],
    chunk_duration=5.0,
    unit_ids=unit_ids,
)

show(
    [(video, VideoView()), (spikes, RasterView(metadata=metadata))],
    title="Video + Raster Demo",
    span=8.0,
    max_workers=2,
)

VideoView(fill=False) fits the frame inside the plot. Set fill=True to crop/fill the plot area.

Probe/Ephys Traces

Use Ephys for high-channel-count sample-clocked data. It differs from TimeSeries by using probe geometry for plotting and by assuming time starts at 0.

import numpy as np
from viewer import Ephys, EphysView, ProbeView, show

geometry = {
    "channel_ids": np.arange(n_channels),
    "shank_ids": shank_ids,
    "x": x_positions,
    "y": y_positions,
}

ephys = Ephys(
    "probe",
    values,                          # shape: (samples, channels)
    geometry,
    fs=1000.0,
    chunk_samples=2_000,
    scale=0.195,                     # applied on read
    offset=0.0,
    units="uV",
)

probe = ProbeView(geometry, visible_channels=np.arange(0, 32))
show([(ephys, EphysView(probe=probe, gain=1 / 40))], title="Probe")

Geometry arrays must be ordered like data columns. channel_ids are labels for those columns; visible_channels selects by channel_ids. If you omit visible_channels, no channels are initially selected, and users can enable channels in the settings panel.

ProbeView also understands optional geometry keys:

  • display_y or display_index: explicit plot row positions.
  • colors, color_hex, or channel_colors: one color per channel.

EphysView(probe=..., width=1.0, gain=1.0, envelope_threshold=2.0) switches from raw traces to min/max envelopes when the viewport has many samples per pixel.

Heatmaps

HeatmapView draws a 2-D TimeSeries or transform output where rows are time and columns are bins.

from viewer import TimeSeries, HeatmapView, show

spec_stream = TimeSeries(
    "spectrogram",
    spectrogram.astype("float32"),    # shape: (time, frequency)
    times,
    fs,
    chunk_samples=240,
)

show([
    (spec_stream, HeatmapView(freqs, y_label="Frequency (Hz)", cmap="Viridis")),
])

Constructor options:

HeatmapView(
    y=None,                           # bin centers
    y_edges=None,                     # explicit edges override y
    y_label="Bin",
    cmap="Viridis",
    scale_min=0.0,
    scale_max=1.0,
    auto_scale=True,
    label_fmt="",
)

If y or y_edges is omitted, the view uses transform-provided chunk["y"], then stream.y, then integer bin numbers.

Transform Pipeline

TimeSeries and Ephys support .pipe(..., name="new stream"). The new stream shares the same source data but has its own name, metadata, and cache entries.

from viewer.transforms import Bandpass, FFT
from viewer import EphysView, HeatmapView, ProbeView, show

fft_stream = raw_ephys.pipe(
    Bandpass(1.0, 225.0),
    FFT(
        channel=100,
        window_s=0.2,
        step_s=0.01,
        freq_min=1.0,
        freq_max=200.0,
        log_power=True,
    ),
    name="fft1",
)

show([
    (raw_ephys, EphysView(probe=ProbeView(geometry), gain=1 / 40)),
    (fft_stream, HeatmapView(y_label="Frequency (Hz)", cmap="Viridis")),
])

Available transforms:

  • CAR(mode="median" | "mean"): common average reference.
  • Lowpass(freq, order=4, pad_s=None, zero_phase=True).
  • Highpass(freq, order=4, pad_s=None, zero_phase=True).
  • Bandpass(low, high, order=4, pad_s=None, zero_phase=True).
  • FFT(channel=0, window_s=0.5, step_s=0.05, power=True, freq_min=0.0, freq_max=None, log_power=False).

Filters use SciPy SOS Butterworth filters. FFT returns a heatmap-friendly stream with frequency centers in stream.y.

Event Bars And Overlays

EventBars can be drawn as a linked top panel with event_bars=..., or as shaded overlays on every plot with overlays=(bars,).

from viewer import EventBars, show

behavior = EventBars(
    starts=[0, 5, 10],
    ends=[5, 10, 15],
    labels=["quiet", "run", "rest"],
    label_order=["quiet", "run", "rest"],
    colors={
        "quiet": "#6b8cff",
        "run": "#47d18c",
        "rest": "#ffc857",
    },
)

show(streams, event_bars=behavior, overlays=(behavior,), span=5.0)

Any custom overlay only needs a draw_overlay() method. Built-in views call it while their ImPlot plot is active.

Utility Workflows

Load paths from a .env file:

from pathlib import Path
from viewer.utils import load_env

data_path = Path(load_env()["EPHYS_DATA_PATH"])

Load a Neuropixels-style binary file with the project defaults (int16, 384 channels):

from viewer.utils import load_bin

data = load_bin(data_path / "eeg" / "eeg.dat")

Load probe geometry:

from viewer.utils import load_prb, load_probeinterface

geometry = load_prb(kilosort_output_dir)
geometry = load_probeinterface(data_path / "concat" / "probe.json")

Convert a pynapple-style time-support group to spike arrays:

from viewer.utils import tsg_to_spikes

spike_times, spike_units = tsg_to_spikes(tsg)

Custom Streams And Views

Most users only need the built-in streams and views. For new data types, follow the existing protocol.

A stream should provide:

name: str
t_min: float
t_max: float
n_chunks: int
chunk_nbytes: int
transforms: object | None

def at(t: float) -> int: ...
def read(chunk_idx: int) -> dict: ...
def chunks_in(span: Span) -> range: ...

BaseStream already implements span, chunks_in(...), len(stream), and integer stream[i] access if you implement at(...), read(...), and chunk metadata.

A view should provide:

def draw_plot(stream, chunks, t, view_t0, view_t1, overlays, *, time_axis="clock"):
    ...

def draw_settings(stream, cache):
    ...

Chunk payload shape depends on the view:

  • TraceView and HeatmapView: data, t_start, t_stop, sample_start, sample_stop, fs, dt, nbytes.
  • EphysView: same as traces, with data shaped (samples, channels).
  • RasterView: ts, data for unit ids, t_start, t_stop, nbytes.
  • VideoView: ts, frame_idx, data shaped (frames, height, width, 3), fs, dt, nbytes.

UI Controls

  • Space: play/pause.
  • Left/right arrows: jump by a quarter of the visible span.
  • r: reset cursor and span.
  • Transport bar: playback speed, clock/seconds axis mode, scrub slider, debug popup.
  • Settings panel: per-stream visibility, cache/chunk debug, view settings, and transform settings.