Core Classes

Raster

class tad.Raster(events=<factory>, amplitudes=<factory>, triggers=None, dtype=dtype('float64'), allow_new_channels=True)

Bases: object

Store and manipulate rasterized (event-based) signals across channels.

This class is primarily designed for neuroscience-style spike rasters, but it generalizes to any multi-channel event timestamp data in dynamical systems.

Internally, timestamps are stored per channel as sorted 1D NumPy arrays (dtype typically float). This makes time-window slicing efficient using np.searchsorted and keeps representation compact.

Parameters:

• events (Dict[int | str, numpy.ndarray]) – Mapping from channel IDs to 1D arrays (or array-like) of event times. Arrays will be coerced to dtype, flattened, and sorted.

• dtype (numpy.dtype) – NumPy dtype used to store timestamps.

• allow_new_channels (bool) – If True, missing channels referenced by methods (e.g., insert) are created automatically. If False, referencing a missing channel raises KeyError.

• triggers (Any | List[Dict[str, Any]] | None) – Optional Triggers object or list of trigger dicts representing time intervals (e.g., stimulation periods). Can be a Triggers object from Triggers.py or a raw list of dicts for backward compatibility.

• amplitudes (Dict[int | str, ndarray])

__post_init__()

Normalize internal storage after dataclass initialization.

Coerces each channel’s timestamps to a 1D NumPy array of self.dtype and sorts them in ascending order.

Notes

• If events contains non-1D array-likes, they are flattened.

• Empty channels are represented by empty 1D arrays.

Return type:

None

allow_new_channels: bool = True

amplitudes: Dict[int | str, ndarray]

as_arrays(*, channels=None, sort_by_time=True, channel_dtype=None)

Export events as concatenated (times, channels) arrays.

Parameters:

• channels (Sequence[int | str] | None) – Subset/order of channels to export. If None, exports all channels in insertion order.

• sort_by_time (bool) – If True, sort the concatenated output by time (stable for equal times). If False, output is grouped by channel in the provided order.

• channel_dtype (dtype | None) – Optional dtype for returned channel array. If None: - uses int64 if all channel IDs are ints - otherwise uses object dtype

Returns:

• times (ndarray, shape (N,)) – Concatenated event times.

• ch_ids (ndarray, shape (N,)) – Channel identifiers per event.

Return type:

Tuple[ndarray, ndarray]

between(tstart, tstop, *, inclusive_stop=False)

Return a new Raster containing events in a time interval.

Parameters:

• tstart (float) – Start time of the interval.

• tstop (float) – Stop time of the interval.

• inclusive_stop (bool) – If False (default), returns events in [tstart, tstop). If True, returns events in [tstart, tstop].

Returns:

A new Raster with events restricted to the interval.

Return type:

Raster

Raises:

ValueError – If tstop < tstart or either boundary is not finite.

bin_counts(dt, *, tstart=None, tstop=None, channels=None, inclusive_stop=False, dtype=dtype('int64'))

Bin events into fixed-width time bins and return spike counts per channel.

This is a core primitive for downstream metrics (avalanches, binned MI/TE, PSTH-like quantities, population activity, etc.).

Parameters:

• dt (float) – Bin width (same units as timestamps). Must be positive and finite.

• tstart (float | None) – Start time of the binning window. If None, inferred as the global minimum event time across selected channels (or 0.0 if no events).

• tstop (float | None) – Stop time of the binning window. If None, inferred as the global maximum event time across selected channels (or tstart + dt if no events).

• channels (Sequence[int | str] | None) – Subset/order of channels to bin. If None, uses all channels in this raster. The returned count matrix row order follows this list (after optional sorting if you pass a sorted list yourself).

• inclusive_stop (bool) – If False (default), treat window as [tstart, tstop) and exclude events at exactly tstop. If True, include events at exactly tstop.

• dtype (dtype) – Integer dtype for returned counts.

Returns:

• bin_edges (ndarray, shape (n_bins + 1,)) – Bin edge times spanning the window.

• counts (ndarray, shape (n_channels, n_bins)) – Spike counts per channel per bin.

• ch_list (list) – Channel IDs corresponding to rows of counts.

Raises:

ValueError – If dt is not positive/finite, or if window bounds are invalid.

Return type:

Tuple[ndarray, ndarray, List[str | int]]

Notes

• Bins are constructed as consecutive intervals of width dt starting at tstart.

• If inclusive_stop=False, an event at exactly tstop is excluded.

If True, it is included by nudging the histogram right edge. - Uses np.searchsorted-based histogramming per channel for correctness and speed on sorted spike times.

channels()

Return the list of channel IDs currently stored.

Returns:

List of channel IDs.

Return type:

list

clear(channel=None)

Clear events from one channel or from all channels.

Parameters:

channel (int | str | None) – If provided, clears only that channel. If None, clears all channels.

Return type:

None

copy()

Return a deep copy of this Raster (arrays are copied).

Returns:

Deep copy of the raster.

Return type:

Raster

dtype: dtype = dtype('float64')

classmethod empty(channels=None, *, dtype=dtype('float64'), allow_new_channels=True)

Create an empty raster, optionally pre-defining channels.

Parameters:

• channels (Iterable[int | str] | None) – Iterable of channel IDs to create with no events.

• dtype (dtype) – NumPy dtype used to store timestamps.

• allow_new_channels (bool) – If True, missing channels referenced by methods can be created automatically.

Returns:

A new empty Raster instance.

Return type:

Raster

events: Dict[int | str, ndarray]

get_trigger_intervals()

Return trigger intervals as a list of (start, end) tuples.

Supports both Triggers objects and backward-compatible lists of trigger dictionaries.

Return type:

List[Tuple[float, float]]

insert(channel, t)

Insert a timestamp into a channel, preserving sorted order.

Parameters:

• channel (int | str) – Channel ID.

• t (float) – Timestamp to insert.

Return type:

None

Notes

• Insertion into a NumPy array is O(n) due to shifting; for very frequent online insertion at large scale, a buffered strategy can be added later.

• Duplicates are allowed.

insert_channel(channel, times=None, *, overwrite=False, sort=True)

Insert a new channel (optionally with initial events).

Parameters:

• channel (int | str) – Channel ID to insert.

• times (ndarray | Iterable[float] | None) – Optional array-like of timestamps for the channel. If None, the channel is created empty.

• overwrite (bool) – If False (default), raises if the channel already exists. If True, replaces the channel’s events with times (or empty).

• sort (bool) – If True (default), sorts provided times.

Raises:

• KeyError – If channel exists and overwrite=False.

• ValueError – If any timestamp is not finite.

Return type:

None

insert_timestamparray(channel, times, *, assume_sorted=False, sort_result=True)

Insert multiple timestamps into a channel.

Parameters:

• channel (int | str) – Channel ID.

• times (ndarray | Iterable[float]) – Array-like timestamps to insert (NumPy array or iterable).

• assume_sorted (bool) – If True, assumes times is already sorted ascending (saves a sort).

• sort_result (bool) – If True (default), ensures the channel remains sorted after insertion.

Raises:

ValueError – If any timestamp is not finite.

Return type:

None

Notes

• Uses concatenation + sort as a simple, robust approach.

• Complexity is O((n+m) log(n+m)) due to sorting; for very large n with small m, a merge approach could be implemented later.

classmethod load(path, *, h5=True, group='/raster')

Load a Raster from disk in either HDF5 or JSON format.

Parameters:

• path (str | Path) – Input file path.

• h5 (bool) – If True (default), load from HDF5 using h5py. If False, load from JSON.

• group (str) – HDF5 group path (only used if h5=True).

Returns:

Loaded Raster instance.

Return type:

Raster

Raises:

• ImportError – If h5=True but h5py is not installed.

• ValueError – If schema/version are not recognized.

• KeyError – If the HDF5 group does not exist.

merge(other, *, on_missing_channels='union', in_place=False)

Merge another Raster into this one (concatenate events per channel).

Parameters:

• other (Raster) – Raster to merge into this raster.

• on_missing_channels (str) – Channel set handling: - “union”: result contains union of channels from both rasters - “intersection”: result contains only channels present in both

• in_place (bool) – If True, modifies and returns self. If False, returns a merged copy.

Returns:

Merged raster.

Return type:

Raster

Raises:

ValueError – If on_missing_channels is not recognized.

Notes

• Event times are concatenated and then sorted within each channel.

• other is coerced to self.dtype in the result.

n_channels()

Return the number of channels currently stored.

Returns:

Number of channels.

Return type:

int

plot(*, ax=None, channels=None, tstart=None, tstop=None, inclusive_stop=False, y_gap=1.0, tick_halfheight=None, linewidth=1.0, sort_channels=True, show=False, plot_triggers=False, trigger_color='red', trigger_alpha=0.5, trigger_linewidth=2.0)

Plot the raster using matplotlib.

Each event is drawn as a vertical tick at its timestamp, arranged by channel along the y-axis.

Parameters:

• ax (Axes | None) – Matplotlib Axes to draw on. If None, creates a new figure and axes.

• channels (Sequence[int | str] | None) – Subset/order of channels to plot. If None, plot all channels.

• tstart (float | None) – Optional time window start. If provided with tstop, the plot is restricted to that interval.

• tstop (float | None) – Optional time window stop.

• inclusive_stop (bool) – If True, include events at exactly tstop when windowing.

• y_gap (float) – Vertical spacing between channels.

• tick_halfheight (float | None) – Half-height of the vertical tick marks. If None, defaults to 0.4 * y_gap.

• linewidth (float) – Line width of tick marks.

• sort_channels (bool) – If True, try to sort channel IDs for display.

• show (bool) – If True, calls plt.show() before returning.

• plot_triggers (bool) – If True, draw trigger intervals from self.triggers as horizontal lines above the raster.

• trigger_color (str) – Color used for trigger interval lines.

• trigger_alpha (float) – Transparency of the trigger lines.

• trigger_linewidth (float) – Line width used for trigger interval lines.

Returns:

The Axes containing the plot.

Return type:

matplotlib.axes.Axes

pop(channel, index=-1)

Remove and return one timestamp from a channel.

Parameters:

• channel (int | str) – Channel ID.

• index (int) – Index of the event to remove (default: -1, last event).

Returns:

The removed timestamp.

Return type:

float

Raises:

IndexError – If the channel has no events.

pop_channel(channel)

Remove an entire channel and return its timestamps.

Parameters:

channel (int | str) – Channel ID to remove.

Returns:

Sorted 1D array of timestamps that were stored in the channel.

Return type:

ndarray

Raises:

KeyError – If the channel does not exist.

relabel_channels(mapping, *, on_conflict='raise', in_place=False)

Relabel channels according to a mapping.

Parameters:

• mapping (Mapping[int | str, int | str]) – Dictionary-like mapping {old_channel: new_channel}. Channels not present in mapping are left unchanged.

• on_conflict (str) – What to do if multiple old channels map to the same new channel, or if a new channel already exists. - “raise”: raise ValueError - “merge”: merge events into the target channel (and keep sorted)

• in_place (bool) – If True, modifies and returns self. If False, returns a relabeled copy.

Returns:

Raster with relabeled channels.

Return type:

Raster

Raises:

ValueError – If conflicts occur and on_conflict="raise".

save(path, *, h5=True, group='/raster', indent=2, compression='gzip', compression_opts=4, overwrite=True, save_amplitudes=False, save_triggers=False)

Save this Raster to disk in either HDF5 or JSON format.

Parameters:

• path (str | Path) – Output file path.

• h5 (bool) – If True (default), save to HDF5 using h5py. If False, save to JSON.

• group (str) – HDF5 group path (only used if h5=True).

• indent (int) – JSON indentation level (only used if h5=False). Use None for compact JSON.

• compression (None | str) – HDF5 compression (only used if h5=True).

• compression_opts (int) – HDF5 compression level/options (only used if h5=True).

• overwrite (bool) – If True, overwrite the target HDF5 group if it exists (only used if h5=True).

• save_amplitudes (bool) – If True, saves the amplitudes of the spikes

• save_triggers (bool)

Raises:

• ImportError – If h5=True but h5py is not installed.

• TypeError – If channel IDs are not int or str.

Return type:

None

shift(dt, *, in_place=False)

Shift all timestamps by a constant offset.

Parameters:

• dt (float) – Time offset to add to every timestamp (can be negative).

• in_place (bool) – If True, modifies and returns self. If False, returns a shifted copy.

Returns:

Shifted raster.

Return type:

Raster

Notes

This operation preserves sorted order within each channel because adding a constant is order-preserving.

triggers: Any | List[Dict[str, Any]] | None = None

CData

class tad.CData(fsample, channel_ids, electrode_labels=None, mask=None, time_vector=None, model_name=None)

Bases: AData, ABC

Abstract base class for computational/simulation data sources.

CData provides a place to store explicit time vectors and model outputs, but keeps the core contract defined by AData: export to a Raster.

Parameters:

• fsample (float)

• channel_ids (Sequence[Any])

• electrode_labels (Optional[Sequence[Any]])

• mask (Optional[Sequence[bool]])

• time_vector (Optional[np.ndarray])

• model_name (Optional[str])

EData

class tad.EData(fsample, channel_ids, electrode_labels=None, mask=None, fname=None, recording_system=None)

Bases: AData, ABC

Abstract base class for experimental data sources.

EData adds minimal experimental-specific metadata (like a source filename), but remains backend-agnostic (it does not force SpikeInterface).

Parameters:

• fsample (float)

• channel_ids (Sequence[Any])

• electrode_labels (Optional[Sequence[Any]])

• mask (Optional[Sequence[bool]])

• fname (Optional[str])

• recording_system (Optional[str])

MCSData

class tad.MCSData(fname, fsample=None, load_recording=True, load_digital=False, generate_probe=False, probe_data=None)

Bases: EData

MCS recording handler.

Parameters:

• fname (str) – Path to the .h5 file.

• fsample (float, optional) – Sampling frequency (Hz). If None, read from the recording.

• load_recording (bool, default=True) – Load the analog recording stream into SpikeInterface.

• load_digital (bool, default=False) – Load a digital channel from the HDF5 file.

• generate_probe (bool, default=False) – Generate a probe object from probe_data (if you use this in your pipeline).

• probe_data (dict, optional) – Probe configuration.

apply_filter(*args, **kwargs)

Apply a SpikeInterface filter to the recording.

Parameters:

• bandpass (array-like or float) – For ‘bandpass’, provide [low_freq, high_freq]. For ‘highpass’, provide a single cutoff or the format expected by SpikeInterface.

• btype ({'bandpass', 'highpass'}, default='bandpass') – Filter type.

Returns:

recording – The filtered recording.

Return type:

spikeinterface.BaseRecording

Raises:

ValueError – If recording is not loaded or bandpass is not provided.

Notes

The original code had a check against the builtin filter; this cleanup removes that non-functional check without changing behavior of the actual filtering.

blank_period(*args, **kwargs)

Exclude (blank) a time window from analysis.

Parameters:

• tstart (float) – Start time (s).

• tstop (float) – Stop time (s).

choose_mask(*args, **kwargs)

Open a GUI to select channels to include (updates self.mask).

Parameters:

• tmin (float, default=0) – Start time (s) for trace preview.

• tmax (float, default=10) – Stop time (s) for trace preview.

• show (bool, default=True) – Show plot.

Notes

This method keeps the original 8x8 grid layout and electrode label mapping: - col = lab // 10 - 1 - row = lab % 10 - 1

convert_digital()

Convert raw digital recording to a small integer state representation.

Returns:

Converted digital signal as int32.

Return type:

np.ndarray

Notes

Preserves the original transformation: - anchor by first sample - log2(abs(x - a + 1)) - values > 2 are set to 0

detect_digital_falling_edge()

Detect falling edges (index positions) in the converted digital signal.

Returns:

Sample indices where digital_recording[i] < digital_recording[i-1].

Return type:

list of int

detect_digital_rising_edge()

Detect rising edges (index positions) in the converted digital signal.

Returns:

Sample indices where digital_recording[i] > digital_recording[i-1].

Return type:

list of int

detect_spikes(*args, **kwargs)

Detect spikes (peaks) in the recording.

Parameters:

• recording (spikeinterface.BaseRecording, optional) – Recording object to use for detect spikes. If None, uses self.recording. This allows using an artifact-removed recording if desired.

• method (str, default='by_channel') – Peak detection method passed to detect_peaks.

• peak_sign (str, default='neg') – ‘neg’ or ‘pos’ depending on spike polarity.

• detect_threshold (float, default=5) – Detection threshold.

• exclude_sweep_ms (float, default=0.2) – Exclusion window in ms.

• noise_levels (array or None) – array of noise levels array precomputed in uV.

• detect_noise_levels (bool,) – if no array of noise level is provided, they can be computed, if set to True.

• job_kwargs – if not None, explicitly uses parallel computing, e.g. {‘n_jobs’: n}.

Returns:

Always returns 1 (kept for backward compatibility).

Return type:

int

export_history_json(path, indent=2)

Export processing history to a JSON file.

Parameters:

• path (str) – Output path.

• indent (int, default=2) – JSON indentation.

Return type:

None

get_probe()

Get the probe object.

Returns:

The probe object if generated, else None.

Return type:

MEAProbe

get_raster(*args, **kwargs)

Export a Raster object using detected peaks and current selection.

Parameters:

• tstart (float, optional) – Start time in seconds. Defaults to start of recording.

• tstop (float, optional) – Stop time in seconds. Defaults to end of recording.

• include_amplitudes (bool, default=True) – If True, include spike amplitudes in the raster in uV.

• include_triggers (bool, default=True) – If True, include triggers from the recording in the raster.

Returns:

Raster object with amplitudes and triggers fields populated.

Return type:

Raster

get_traces(tstart=0, tstop=10, channel_ids=None, return_in_uV=True)

Retrieve traces from the recording.

Parameters:

• tstart (float, default=0) – Start time in seconds.

• tstop (float, default=10) – Stop time in seconds.

• channel_ids (list, optional) – Subset of channel ids to extract.

• return_in_uV (bool, default=True) – Convert to microvolts.

Returns:

Traces array of shape (n_samples, n_channels).

Return type:

np.ndarray

get_triggers(*args, **kwargs)

Build Triggers from the digital recording in a given time window.

Parameters:

• method (str, default='artifact') – Method to obtain triggers. - ‘artifact’: detect artifact blocks from the analog recording using a threshold. - ‘first_passage’: detect the first threshold crossing of each event and generate a fixed-duration slot. - ‘digital_trigger’: use the digital recording directly, optionally with a custom interpretor.

• tstart (float, optional) – Start time in seconds. Defaults to the start of the recording.

• tstop (float, optional) – Stop time in seconds. Defaults to the end of the recording.

• interpretor (callable, optional) – Custom function to interpret the (windowed) digital signal into trigger slots. It must accept (digital_recording, triggers, fsample) or (digital_recording, triggers, fsample, dt_after_trigger). Trigger times created by the custom interpreter are assumed to be relative to tstart; get_triggers will shift them to absolute time.

• dt_after_trigger (float, optional) – Extra argument forwarded to interpretor when provided.

• artifact_threshold (float, optional) – Threshold multiplier used by artifact and first_passage methods.

• mean_noise_level (float, optional) – Mean noise level used by artifact and first_passage methods.

• refractory_trigger_period (float, optional) – Required for first_passage. Minimum time in seconds between consecutive triggers.

• stim_on_time (float, optional) – Required for first_passage. Duration in seconds for the generated trigger slot.

• moving_avg_window (float, optional) – Time in ms of the moving average window when artifact or first passage methods are chosen.

Returns:

Trigger container with the detected interval slots.

Return type:

Triggers

Raises:

• TypeError – If method is not a string.

• ValueError – If required arguments for the selected method are missing, invalid, or if the requested time window is outside the recording range.

load_mask_and_labels(fname)

Load mask, channel_ids and electrode_labels from a JSON or CSV file.

The file is expected to contain the same number of entries as the current instance (same channel count). This method does NOT reorder channels; it assumes the file lines/order match your current channel ordering.

Parameters:

fname (str) – Input filename (.json or .csv)

Returns:

0 if OK.

Return type:

int

plot_raster(ax)

Plot a raster of detected spikes into an existing axes.

Parameters:

ax (matplotlib.axes.Axes) – Target axes.

Returns:

Always returns 1 (backward compatible).

Return type:

int

plot_traces_in_grid(tmin=0, tmax=10, n_subsample=None, show=True)

Plot channel traces in a grid.

Parameters:

• tmin (float, default=0) – Start time (s).

• tmax (float, default=10) – Stop time (s).

• n_subsample (Optional[int]) – 1/n_subsample factor, represents the number of time samples skipped in plotting.

• show (bool, default=True) – Show plot.

Returns:

Always returns 1 (backward compatible).

Return type:

int

remove_artifacts_from_trigger(*args, **kwargs)

Remove stimulation artifacts around triggers using SpikeInterface.

Parameters:

• ms_before (float, default=0.1) – Time (ms) before each trigger to remove.

• ms_after (float, default=0.4) – Time (ms) after each trigger to remove.

• mode (str, default='zeros') – Artifact replacement mode (as supported by pre.remove_artifacts).

Returns:

recording_clean – Recording after artifact removal.

Return type:

spikeinterface.BaseRecording

Raises:

ValueError – If artifact removal has already been performed or triggers are missing.

save_mask_and_labels(fname, csv_format=False)

Save current mask, channel_ids and electrode_labels to a JSON or CSV file.

Parameters:

• fname (str) – Output filename.

• csv_format (bool) – If True, write a CSV with 3 columns: channel_id, electrode_label, mask. If False, write JSON.

Returns:

0 if OK.

Return type:

int

set_mask(mask)

Return type:

None