Skip to content

Features API

Auto-generated documentation for all feature classes.

Note

This page is generated by scripts/gen_feature_docs.py from the auto-discovered feature registry. Do not edit by hand.

cobrabox.infometrics.AmplitudeEntropy dataclass

AmplitudeEntropy(band_width: float)

Bases: BaseFeature[Data]

Compute amplitude entropy from time-series data using histogram-based probability estimation.

Amplitude entropy quantifies the randomness or unpredictability of the amplitude distribution in the data. It uses a histogram-based approach where the data is binned according to the specified band_width, and Shannon entropy is computed from the resulting probability distribution.

The entropy for each row (time point) is calculated as H = -sum(p_i * log2(p_i)), where p_i is the probability of the i-th bin. The final result is the mean entropy across all time points, returned as a scalar value.

Parameters:

Name Type Description Default
band_width float

The width of histogram bins for discretizing the data. Must be positive.

required

Returns:

Type Description

A 0-dimensional :class:~cobrabox.Data containing the mean amplitude

entropy as a scalar value.

Raises:

Type Description
ValueError

If band_width is not positive.

ValueError

If input data has fewer than 2 dimensions.

Example

import cobrabox as cb import numpy as np data = cb.Data.from_numpy(np.random.randn(50, 10), dims=["time", "space"]) result = cb.AmplitudeEntropy(band_width=0.5).apply(data) float(result.to_numpy()) # Scalar entropy value

cobrabox.signalstats.AmplitudeVariation dataclass

AmplitudeVariation()

Bases: BaseFeature[SignalData]

Compute amplitude variation over the time dimension.

Amplitude variation is the standard deviation of the EEG signal within the calculation window. A larger value indicates a more variable signal.

Returns:

Type Description

class:~cobrabox.Data with the time dimension removed. Shape is

(space,) for standard input, or (*extra_dims, space) if

additional dimensions are present (e.g. window_index). Values

are in the same units as the input signal.

Example

result = cb.AmplitudeVariation().apply(data)

cobrabox.transforms.AnalyticSignal dataclass

AnalyticSignal(
    feature: Literal[
        "analytic", "envelope", "phase", "frequency"
    ] = "analytic",
)

Bases: BaseFeature[SignalData]

Extract analytic-signal representations along the time axis.

Computes the analytic signal via :func:scipy.signal.hilbert along the time axis, then derives the requested representation. The output always has the same shape and dimensions as the input — the time dimension is preserved.

Parameters:

Name Type Description Default
feature Literal['analytic', 'envelope', 'phase', 'frequency']

Which representation to extract. One of:

  • 'analytic' — the complex analytic signal as a complex128 DataArray with the same dims and coords as the input.
  • 'envelope' — amplitude envelope (|analytic|), always non-negative.
  • 'phase' — instantaneous phase in radians (∈ [-pi, pi]).
  • 'frequency' — instantaneous frequency in Hz, computed as the time-derivative of the unwrapped phase divided by 2π. Requires data.sampling_rate to be set.
'analytic'

Returns:

Type Description

class:~cobrabox.SignalData with the same dims and coords as the input. Dtype is

complex128 for 'analytic', float64 for all other modes.

Raises:

Type Description
ValueError

If feature is not one of the four valid options.

ValueError

If feature='frequency' but data.sampling_rate is None.

Example

analytic = cb.AnalyticSignal().apply(data) env = cb.AnalyticSignal(feature="envelope").apply(data) phase = cb.AnalyticSignal(feature="phase").apply(data) freq = cb.AnalyticSignal(feature="frequency").apply(data)

cobrabox.signalstats.Autocorrelation dataclass

Autocorrelation(
    dim: str,
    fs: float,
    lag_steps: int | None = None,
    lag_ms: float | None = None,
)

Bases: BaseFeature[Data]

Compute normalized autocorrelation at a given lag along a required dimension.

MATLAB equivalent from mapping_seizure_dynamics/calc_feat: acf = autocorr(d,'NumLags',round(fs/1000*5)); autocorrel = acf(end);

Specify either lag_steps or lag_ms, not both. If neither is specified, the default lag of 5 ms is used.

Parameters:

Name Type Description Default
dim str

Dimension to compute autocorrelation along.

required
fs float

Sampling frequency in Hz. Must be positive.

required
lag_steps int | None

Lag in number of samples. Mutually exclusive with lag_ms.

None
lag_ms float | None

Lag in milliseconds. Mutually exclusive with lag_steps.

None

Returns:

Type Description

class:~cobrabox.Data with the dim dimension removed. Shape is the input shape

minus the size of dim; each element is the normalized autocorrelation at the

computed lag for that position in the remaining dimensions.

Example

result = cb.Autocorrelation(dim="time", fs=1000.0, lag_steps=5).apply(data)

cobrabox.spectral.BandPower dataclass

BandPower(
    bands: Literal["eeg"]
    | dict[str, list[float] | bool]
    | None = None,
    nperseg: int | None = None,
)

Bases: BaseFeature[SignalData]

Compute band power for specified frequency bands using Welch's method.

For each frequency band, integrates the power spectral density (PSD) estimated via Welch's method over the band's frequency range. Operates over the time dimension, returning one scalar power value per (space, band) pair.

Parameters:

Name Type Description Default
bands Literal['eeg'] | dict[str, list[float] | bool] | None

None (the default) computes a single "full" band integrating power over the entire spectrum (0 to Nyquist). "eeg" computes the five standard EEG bands, or pass a mapping of band name to frequency range [f_low, f_high] in Hz (True to use the default range for that band name):

  • delta: 1 - 4 Hz
  • theta: 4 - 8 Hz
  • alpha: 8 - 12 Hz
  • beta: 12 - 30 Hz
  • gamma: 30 - 45 Hz

An empty dict {} behaves like None.

None
nperseg int | None

Number of samples per Welch segment. Controls the trade-off between frequency resolution (larger → finer bins) and variance reduction (smaller → more segments to average). Defaults to min(n_time, 256) as chosen by :func:scipy.signal.welch.

None
Example

bp = cb.BandPower().apply(data) bp_full = cb.BandPower(bands=None).apply(data) bp_eeg = cb.BandPower(bands="eeg").apply(data) bp_custom = cb.BandPower(bands={"alpha": True, "ripple": [45, 80]}).apply(data) bp_fine = cb.BandPower(nperseg=512).apply(data)

Returns:

Type Description

class:~cobrabox.Data with dims (band_index, space) (plus a singleton

time dimension added by BaseFeature.apply). The band_index

coordinate holds the band names. Values are absolute power in units of

the input signal squared per Hz (signal² / Hz).

Raises:

Type Description
ValueError

If sampling_rate is not set on the input data.

ValueError

If nperseg is less than 2.

ValueError

If an unknown band name is provided or band spec is False.

ValueError

If an unknown bands preset string is provided.

cobrabox.decompositions.BandpassFilter dataclass

BandpassFilter(
    bands: Literal["eeg"] | dict[str, list[float]],
    ord: int = 3,
    keep_orig: bool = False,
)

Bases: BaseFeature[SignalData]

Filter a signal into frequency bands.

Applies a Butterworth bandpass filter for each band and stacks the results along a new band dimension.

Parameters:

Name Type Description Default
bands Literal['eeg'] | dict[str, list[float]]

Band specification. "eeg" computes the five standard EEG bands:

  • delta: 1 - 4 Hz
  • theta: 4 - 8 Hz
  • alpha: 8 - 12 Hz
  • beta: 12 - 30 Hz
  • gamma: 30 - 45 Hz

Or pass a mapping of band name to [low_hz, high_hz] frequency edges. None is not allowed.

required
ord int

Order of the filter. Defaults to 3.

3
keep_orig bool

Whether to keep the original signal as a "band" named "original". Defaults to False.

False

Raises:

Type Description
ValueError

If bands is None, an unknown preset string, or invalid.

ValueError

If the input Data has no known sampling_rate.

Returns:

Type Description

class:~cobrabox.SignalData: The bandpass-filtered signals stacked along a new band dimension. The band coordinate contains the band names (and "original" if keep_orig=True). The shape is the same as the input data with an additional band dimension.

Example

result = cb.BandpassFilter(bands="eeg").apply(data) result = cb.BandpassFilter(bands={"alpha": [8, 12]}).apply(data)

__post_init__

__post_init__() -> None

Validate parameters after initialization.

Source code in src/cobrabox/decompositions/_bandpass_filter.py
def __post_init__(self) -> None:
    """Validate parameters after initialization."""
    if self.ord <= 0:
        raise ValueError(f"ord must be positive, got {self.ord}")
    if self.bands is None:
        raise ValueError("bands cannot be None; pass 'eeg' or a band mapping")
    if isinstance(self.bands, str):
        if self.bands not in _BAND_PRESETS:
            raise ValueError(
                f"Unknown bands preset {self.bands!r}. "
                f"Available presets: {list(_BAND_PRESETS)}."
            )
        resolved: Mapping[str, Sequence[float]] = dict(_BAND_PRESETS[self.bands])
    else:
        resolved = self.bands
    if not resolved:
        raise ValueError("bands cannot be empty")
    for band_name, freqs in resolved.items():
        if len(freqs) != 2:
            raise ValueError(f"Band '{band_name}' must have exactly 2 frequencies [low, high]")
        low, high = freqs
        if low < 0 or high < 0:
            raise ValueError(
                f"Band '{band_name}' frequencies must be non-negative, got [{low}, {high}]"
            )
        if low >= high:
            raise ValueError(
                f"Band '{band_name}' low frequency must be less than high, got [{low}, {high}]"
            )

cobrabox.connectivity.Coherence dataclass

Coherence(nperseg: int | None = None)

Bases: BaseFeature[SignalData]

Compute magnitude-squared coherence for all pairwise channel combinations.

Uses Welch's method (50 % overlap, Hann window) to estimate the magnitude-squared coherence between each unique pair of spatial channels, then averages over frequency bins to produce a single scalar per pair. Because coherence is symmetric the result is a symmetric NxN matrix; the diagonal is set to NaN (self-coherence is uninformative).

Extra dimensions beyond 'space' and 'time' (e.g. window_index) are preserved — coherence is computed along the 'time' axis for every combination of those extra dimensions.

Parameters:

Name Type Description Default
nperseg int | None

Samples per FFT segment. Defaults to min(256, n_time). Must be >= 2 and <= n_time.

None
Example

data = cb.load_dataset("dummy_random")[0] coh = cb.Coherence().apply(data) coh.data.dims ('space_to', 'space_from')

Returns:

Type Description

class:~cobrabox.Data with dims (*extra_dims, space_to, space_from) (plus a

singleton time dimension added by BaseFeature.apply). Both

space_to and space_from carry the original channel coordinates.

Values are in [0, 1]; the diagonal is NaN (self-coherence). The matrix

is symmetric: result[i, j] == result[j, i].

Raises:

Type Description
ValueError

If nperseg is provided and less than 2.

ValueError

If input data lacks a 'space' dimension.

ValueError

If fewer than 2 spatial channels are present.

ValueError

If computed nperseg is less than 2 or exceeds time samples.

cobrabox.windowing.ConcatAggregate dataclass

ConcatAggregate()

Bases: AggregatorFeature

Aggregate a stream of per-window Data by stacking along a new 'window' dimension.

Collects all windows and concatenates them along a new 'window' dimension, preserving the per-window results rather than reducing them. Per-window pipeline history is propagated to the result.

When the windows carry window_start/window_end in extra — as those produced by :class:SlidingWindow do — the window coordinate is labelled with each window's start time on the original time axis rather than a bare index, and a non-dimension window_end coordinate carries the matching end times. This lets you plot against real time and locate an event ("the seizure starts at 4.2 s") on the window axis. Splitters that do not report window positions fall back to an integer index from 0.

Returns:

Type Description

A new Data object with an additional leading window dimension.

All metadata from the original data is preserved, including

sampling_rate. History includes all per-window operations followed

by "ConcatAggregate".

Example

chord = ( ... cb.SlidingWindow(window_size=100, step_size=50) ... | cb.LineLength() ... | cb.ConcatAggregate() ... ) result = chord.apply(data) result.data.dims # ('window', ...) result.data.window.values[:3] # 100 Hz data, step of 50 samples array([0. , 0.5, 1. ])

cobrabox.spectral.ContinuousWaveletTransform dataclass

ContinuousWaveletTransform(
    wavelet: _CwtWavelet = "morl",
    scales: list[float] | None = None,
    n_scales: int = 64,
    scaling: Literal[
        "magnitude", "power", "complex"
    ] = "magnitude",
)

Bases: BaseFeature[SignalData]

Continuous wavelet transform (CWT) scalogram.

Applies pywt.cwt along the time axis for each spatial channel, producing a 2-D time-scale representation. The output preserves the time dimension and adds a scale dimension, making it directly composable with downstream features that operate along time or space.

Parameters:

Name Type Description Default
wavelet _CwtWavelet

Continuous wavelet name. Common choices for EEG analysis:

  • "morl" (default) — real Morlet; captures oscillatory components with strong time-frequency localisation.
  • "mexh" — Mexican Hat (second derivative of a Gaussian); highlights singularities and sharp transitions.
  • "cmor{B}-{C}" — complex Morlet (e.g. "cmor1.5-1.0"); gives complex coefficients with better frequency resolution.

See pywt.wavelist(kind='continuous') for all built-in options.

'morl'
scales list[float] | None

Explicit array of scales to analyse. If None (default), n_scales linearly-spaced integer scales [1, 2, ..., n_scales] are used. Larger scales correspond to lower frequencies.

None
n_scales int

Number of scales when scales is None. Defaults to 64.

64
scaling Literal['magnitude', 'power', 'complex']

Output scaling. One of:

  • "magnitude" (default) — |CWT|; real-valued, non-negative.
  • "power" — |CWT|²; power scalogram.
  • "complex" — raw complex coefficients (dtype complex128).
'magnitude'

Returns:

Name Type Description

class:~cobrabox.SignalData with dims (*extra_dims, "space", "scale", "time")

where
  • scale holds the raw pywt scale values.
  • frequency is a non-index coordinate on the scale dimension giving the pseudo-frequency in Hz (from data.sampling_rate; falls back to cycles/sample when sampling_rate is None).
  • time preserves the original time coordinates from the input.

Raises:

Type Description
ValueError

If wavelet is not a valid continuous wavelet, if scales is empty or contains non-positive values, if n_scales < 1, or if scaling is not one of the three valid options.

Example

data = cb.load_dataset("dummy_random")[0] cwt = cb.ContinuousWaveletTransform(n_scales=32).apply(data) cwt.data.dims ('space', 'scale', 'time') "frequency" in cwt.data.coords True

cobrabox.spectral.Cordance dataclass

Cordance(
    bands: dict[str, list[float] | bool] | None = None,
    nperseg: int | None = None,
    threshold: float = 0.5,
    output: Literal[
        "cordance", "concordance", "discordance"
    ] = "cordance",
    nan_on_zero: bool = False,
)

Bases: BaseFeature[SignalData]

Compute cordance, a qEEG measure combining absolute and relative bandpower.

Cordance (Leuchter et al., 1994) integrates absolute and relative spectral power into a single index per channel per frequency band.

Algorithm (Leuchter 1994 threshold-based): 1. Compute per-band absolute power (AP) via :class:~cobrabox.spectral.band_power.BandPower (Welch's method). 2. Compute relative power (RP) as the ratio of each band's absolute power to the total power across all requested bands. 3. Normalize AP and RP by dividing by the maximum across channels: Anorm = AP / max(AP), Rnorm = RP / max(RP). 4. Classify each channel using a threshold of 0.5: - Concordant: both Anorm > 0.5 and Rnorm > 0.5 - Discordant: Anorm < 0.5 and Rnorm > 0.5 5. Compute cordance scores as deviation from the threshold: - Concordance = (Anorm - 0.5) + (Rnorm - 0.5) - Discordance = (0.5 - Anorm) + (Rnorm - 0.5)

By default, this returns a combined "cordance" value where concordant channels get positive scores and discordant channels get negative scores. Use the output parameter to get concordance or discordance maps separately.

Parameters:

Name Type Description Default
bands dict[str, list[float] | bool] | None

Mapping of band name to frequency range [f_low, f_high] in Hz, or True to use the default range for that band name. If None or empty, all five default bands are computed:

  • delta: 1 - 4 Hz
  • theta: 4 - 8 Hz
  • alpha: 8 - 12 Hz
  • beta: 12 - 30 Hz
  • gamma: 30 - 45 Hz
None
nperseg int | None

Number of samples per Welch segment. Controls the trade-off between frequency resolution and variance reduction. Defaults to min(n_time, 256) as chosen by :func:scipy.signal.welch.

None
threshold float

Threshold for classifying concordant vs discordant channels. Default is 0.5 (50% of normalized maximum) as in the original Leuchter 1994 algorithm.

0.5
output Literal['cordance', 'concordance', 'discordance']

What to return: - "cordance" (default): Combined score where concordant channels are positive and discordant channels are negative. Channels that are neither concordant nor discordant return 0. - "concordance": Only concordance scores (0 for non-concordant). - "discordance": Only discordance scores (0 for non-discordant).

'cordance'
nan_on_zero bool

If True, channels with zero total bandpower output NaN instead of raising an error. Useful for batch processing where some channels may be silent. Defaults to False.

False
Example

cord = cb.Cordance().apply(data) cord_theta = cb.Cordance(bands={"theta": True}).apply(data) disc = cb.Cordance(output="discordance").apply(data)

Returns:

Type Description

class:~cobrabox.Data with dims (band_index, space). The band_index

coordinate holds the band names. Values are unitless cordance scores:

positive for concordant channels, negative for discordant channels.

References

Leuchter, A. F., Cook, I. A., Lufkin, R. B., Dunkin, J., Newton, T. F., Cummings, J. L., ... & Walter, D. O. (1994). Cordance: a new method for assessment of cerebral perfusion and metabolism using quantitative electroencephalography. NeuroImage, 1(3), 208-219.

US Patent 5,309,923 (1994). Method and apparatus for determining brain activity including the nature of brain lesions by electroencephalography.

cobrabox.connectivity.Correlation dataclass

Correlation(dim: str = 'time', method: str = 'pearson')

Bases: BaseFeature[Data]

Compute pairwise Pearson or Spearman correlation between all channel pairs.

Correlates channels along a chosen dimension (default: "time") producing a symmetric N x N matrix. Input data must be exactly 2-dimensional: one dimension is the channel axis, the other is the axis along which correlation is computed.

The diagonal (self-correlation) is set to 1.0. The result is a plain :class:~cobrabox.Data object (output_type = Data) because the correlation axis is consumed.

Parameters:

Name Type Description Default
dim str

Name of the dimension to correlate along. Defaults to "time". Must be present in the input data. The remaining dimension becomes both <other_dim>_to and <other_dim>_from in the output.

'time'
method str

Correlation method to use. "pearson" (default) uses :func:numpy.corrcoef. "spearman" converts values to ranks along dim first, then applies :func:numpy.corrcoef.

'pearson'

Raises:

Type Description
ValueError

If method is not "pearson" or "spearman".

ValueError

If input data is not exactly 2-dimensional.

ValueError

If dim is not present in the input data's dimensions.

Example

data = cb.load_dataset("dummy_random")[0] corr = cb.Correlation().apply(data) corr.data.dims ('space_to', 'space_from')

Returns:

Type Description

class:~cobrabox.Data with dims (<other_dim>_to, <other_dim>_from).

Both coordinate axes carry the original channel labels. Values are in

[-1, 1]; the diagonal is 1.0. The matrix is symmetric.

cobrabox.connectivity.Covariance dataclass

Covariance(dim: str = 'time')

Bases: BaseFeature[Data]

Compute pairwise sample covariance between all channel pairs.

Computes the sample covariance matrix along a chosen dimension (default: "time") producing a symmetric N x N matrix. Input data must be exactly 2-dimensional: one dimension is the channel axis, the other is the axis along which covariance is computed.

The diagonal contains the sample variance of each channel (ddof=1). The result is a plain :class:~cobrabox.Data object (output_type = Data) because the covariance axis is consumed.

Parameters:

Name Type Description Default
dim str

Name of the dimension to compute covariance along. Defaults to "time". Must be present in the input data. The remaining dimension becomes both <other_dim>_to and <other_dim>_from in the output.

'time'

Raises:

Type Description
ValueError

If input data is not exactly 2-dimensional.

ValueError

If dim is not present in the input data's dimensions.

Example

data = cb.load_dataset("dummy_random")[0] cov = cb.Covariance().apply(data) cov.data.dims ('space_to', 'space_from')

Returns:

Type Description

class:~cobrabox.Data with dims (<other_dim>_to, <other_dim>_from).

Both coordinate axes carry the original channel labels. The diagonal

contains per-channel sample variance; off-diagonal entries are

covariances. The matrix is symmetric.

cobrabox.connectivity.DirectedTransferFunction dataclass

DirectedTransferFunction(
    var_order: int | None = None, n_freqs: int = 128
)

Bases: BaseFeature[SignalData]

Estimate the Directed Transfer Function (DTF) between channels via a VAR model.

Like :class:PartialDirectedCoherence, the DTF is derived from a multivariate autoregressive fit. Where PDC normalizes the coefficient matrix A(f) column-wise (capturing direct causal influence), DTF inverts A(f) to the transfer function H(f) and normalizes its rows (capturing total causal influence, including indirect paths).

For each frequency bin, DTF[i, j, f] represents the normalized total directional influence from channel j to channel i. Values lie in [0, 1] and the rows of the magnitude-squared DTF sum to 1.

Parameters:

Name Type Description Default
var_order int | None

Number of lags for the VAR model. None (default) lets statsmodels pick the optimal order via AIC.

None
n_freqs int

Number of frequency bins in [0, sr/2]. Default 128.

128

Returns:

Type Description

class:~cobrabox.Data with dims ("space_to", "space_from", "frequency").

References

Kamiński, M., & Blinowska, K. J. (1991). A new method of the description of the information flow in the brain structures. Biological Cybernetics 65(3), 203-210.

cobrabox.spectral.DiscreteWaveletTransform dataclass

DiscreteWaveletTransform(
    wavelet: _DwtWavelet = "db4",
    level: int | None = None,
    mode: Literal[
        "zero",
        "constant",
        "symmetric",
        "periodic",
        "smooth",
        "periodization",
        "reflect",
        "antisymmetric",
        "antireflect",
    ] = "symmetric",
)

Bases: BaseFeature[SignalData]

Multi-level discrete wavelet decomposition (DWT).

Uses pywt.wavedec to decompose the time axis into one approximation and level detail coefficient arrays. Because each decomposition level has a different number of coefficients, shorter arrays are right-padded with NaN to match the length of the finest-level (level 1) detail coefficients, producing a rectangular output that can be passed to downstream features.

Parameters:

Name Type Description Default
wavelet _DwtWavelet

Discrete wavelet name (e.g. "db4", "haar", "sym5"). See pywt.wavelist(kind='discrete') for all valid options. Daubechies-4 ("db4") is a standard choice for EEG analysis.

'db4'
level int | None

Decomposition level. None (default) uses the maximum level allowed by the signal length and wavelet filter length.

None
mode Literal['zero', 'constant', 'symmetric', 'periodic', 'smooth', 'periodization', 'reflect', 'antisymmetric', 'antireflect']

Signal extension mode for boundary handling (e.g. "symmetric", "periodization", "zero"). See pywt.Modes for options.

'symmetric'

Returns:

Type Description

class:~cobrabox.Data with dims ``(*extra_dims, "space", "wavelet_level",

"coef_index")`` where:

  • wavelet_level has string coordinates following the pywt.wavedec output order: ["approx", "detail_L", ..., "detail_1"] (coarsest approximation first, then details from coarsest to finest).
  • coef_index runs from 0 to the length of the finest detail minus 1. Shorter levels are NaN-padded on the right.

Raises:

Type Description
ValueError

If wavelet is not a valid discrete wavelet name, if level < 1, or if level exceeds the maximum possible for the signal length.

Example

data = cb.load_dataset("dummy_random")[0] dwt = cb.DiscreteWaveletTransform(wavelet="db4", level=4).apply(data) dwt.data.dims ('space', 'wavelet_level', 'coef_index') list(dwt.data.coords["wavelet_level"].values) ['approx', 'detail_4', 'detail_3', 'detail_2', 'detail_1']

cobrabox.decompositions.EMD dataclass

EMD(
    max_imfs: int | None = None,
    method: Literal[
        "sift", "mask_sift", "iterated_mask_sift"
    ] = "sift",
    keep_orig: bool = False,
)

Bases: BaseFeature[SignalData]

Decompose a signal into Intrinsic Mode Functions (IMFs) using Empirical Mode Decomposition.

Applies EMD along the time axis and stacks the resulting IMFs along a new imf dimension with coordinates imf0, imf1, etc. The last component is always the residual (labelled residual).

The IMFs sum to the original signal: result.sum(dim='imf') == original.

When processing multi-channel data, different channels may produce different numbers of IMFs. Missing IMFs are filled with NaN (not zero) so that operations like mean(dim='imf') give correct results without being diluted by fake zeros.

Parameters:

Name Type Description Default
max_imfs int | None

Maximum number of IMFs to compute (not including residual). If None, the algorithm determines the number adaptively. Defaults to None.

None
method Literal['sift', 'mask_sift', 'iterated_mask_sift']

Sift method to use. One of:

  • 'sift' — standard EMD (default)
  • 'mask_sift' — mask sift for improved mode separation
  • 'iterated_mask_sift' — iterated mask sift
'sift'
keep_orig bool

Whether to keep the original signal as an IMF named original. Defaults to False.

False

Raises:

Type Description
ValueError

If method is not a valid sift method.

ValueError

If max_imfs is not positive.

Returns:

Type Description

class:~cobrabox.SignalData: The IMFs stacked along a new imf dimension. The imf coordinate contains labels imf0, imf1, ..., residual (and original if keep_orig=True). Shape is the same as input with an additional imf dimension.

An n_imfs dict in attrs tracks the number of actual IMFs extracted per channel (excluding residual and original). Keys are channel names (e.g. "Fp1" or "Fp1/A" for multi-dimensional), or "signal" for 1D data. This is useful when channels produce different numbers of IMFs and missing IMFs are filled with NaN.

Example

result = cb.EMD().apply(data) result = cb.EMD(max_imfs=5, method="mask_sift").apply(data) result = cb.EMD(keep_orig=True).apply(data)

__post_init__

__post_init__() -> None

Validate parameters after initialization.

Source code in src/cobrabox/decompositions/_emd.py
def __post_init__(self) -> None:
    """Validate parameters after initialization."""
    if self.method not in _SIFT_METHODS:
        raise ValueError(
            f"Invalid method {self.method!r}. "
            f"Must be one of: {', '.join(repr(m) for m in _SIFT_METHODS)}."
        )
    if self.max_imfs is not None and self.max_imfs <= 0:
        raise ValueError(f"max_imfs must be positive, got {self.max_imfs}")

cobrabox.connectivity.EnvelopeCorrelation dataclass

EnvelopeCorrelation(
    orthogonalize: str | bool = "pairwise",
    absolute: bool = False,
)

Bases: BaseFeature[SignalData]

Compute amplitude envelope correlation (AEC) between all channel pairs.

Estimates the correlation between the amplitude envelopes of each pair of spatial channels. Envelopes are extracted via the analytic signal (Hilbert transform). When orthogonalize='pairwise', zero-lag contributions are removed between each pair before correlating, which reduces the influence of volume conduction.

Delegates to :func:mne_connectivity.envelope_correlation with log=False.

Parameters:

Name Type Description Default
orthogonalize str | bool

Whether and how to orthogonalise signal pairs before correlating. 'pairwise' (default) removes zero-lag correlations per pair. False skips orthogonalisation.

'pairwise'
absolute bool

If True, take the absolute value of correlation coefficients before symmetrising. Defaults to False (unlike the mne-connectivity default of True).

False

Returns:

Type Description

class:~cobrabox.Data with dims (space_to, space_from) containing

Pearson correlation values between amplitude envelopes.

Raises:

Type Description
ValueError

If data has dimensions other than time and space.

ValueError

If data has fewer than 2 spatial channels.

References

Hipp, J. F., Hawellek, D. J., Corbetta, M., Siegel, M., & Engel, A. K. (2012). Large-scale cortical correlation structure of spontaneous oscillatory activity. Nature Neuroscience, 15(6), 884-890. https://doi.org/10.1038/nn.3101

Example

aec = cb.EnvelopeCorrelation().apply(data) aec_signed = cb.EnvelopeCorrelation(absolute=False).apply(data) aec_no_orth = cb.EnvelopeCorrelation(orthogonalize=False).apply(data)

cobrabox.signalstats.EpileptogenicityIndex dataclass

EpileptogenicityIndex(
    window_duration: float = 1.0,
    bias: float = 0.5,
    threshold: float = 30.0,
    integration_window: float = 5.0,
    tau: float = 1.0,
)

Bases: BaseFeature[SignalData]

Compute the Epileptogenicity Index (EI) per channel (Bartolomei et al., 2008).

EI quantifies the epileptogenicity of each recorded brain structure by combining two features of the ictal SEEG signal: the spectral property (presence of high-frequency rapid discharges) and the temporal property (how early in the seizure process the discharge appears relative to other channels). Higher EI indicates a more epileptogenic structure.

The algorithm operates in three stages:

  1. Energy Ratio ER[n] — a sliding-window periodogram is computed along each channel. At each window position the ratio of high- to low-frequency band power is computed::

    ER[n] = (E_beta + E_gamma) / (E_theta + E_alpha)

Frequency bands (Table 2 of the paper): θ 3.5-7.4 Hz · alpha 7.4-12.4 Hz · beta 12.4-24 Hz · gamma 24 Hz-Nyquist

  1. Page-Hinkley (CUSUM) detection — applied per channel to find the detection time N_d (onset of the rapid discharge). Channels where no alarm fires are assigned N_d = last ER sample, which yields near-zero EI after normalisation.

  2. EI formula and normalisation — given the earliest detection time N_0 (reference channel) and integration window H samples::

    EI_i = sum(ER[N_di : N_di + H]) / (N_di - N_0 + tau)

Values are divided by the per-recording maximum to obtain a [0, 1] scale.

Reference

Bartolomei F, Chauvel P, Wendling F. Epileptogenicity of brain structures in human temporal lobe epilepsy: a quantified study from intracerebral EEG. Brain (2008), 131:1818-1830. DOI: 10.1093/brain/awn111

Parameters:

Name Type Description Default
window_duration float

Duration of the ER sliding window in seconds. Longer windows give better frequency resolution but coarser temporal resolution. Default: 1.0 s.

1.0
bias float

Page-Hinkley bias v. Higher values suppress small ER fluctuations and reduce false detections. Default: 0.5.

0.5
threshold float

Page-Hinkley alarm threshold λ. Higher values reduce false alarms at the cost of delayed or missed detections. Default: 30.0.

30.0
integration_window float

Duration H (seconds) over which ER[n] is summed after detection to capture the rapid-discharge energy. Default: 5.0 s.

5.0
tau float

Small constant τ preventing division by zero when a channel is the first to fire (delay = 0). Default: 1.0.

1.0

Returns:

Type Description

class:~cobrabox.Data with dim (space,), values normalised to [0, 1]. If no

rapid discharge is detected in any channel, all values are 0.

Raises:

Type Description
ValueError

If data does not have exactly time and space dimensions.

ValueError

If data.sampling_rate is not set.

ValueError

If any channel's signal is shorter than one ER window.

Example

ei = cb.EpileptogenicityIndex().apply(seizure_data) ei_vals = ei.data # (space,) values in [0, 1]

cobrabox.transforms.FourierTransform dataclass

FourierTransform(
    norm: Literal["real", "psd"] | None = None,
    cutoff: float | None = None,
)

Bases: BaseFeature[SignalData]

Real-valued FFT along the time axis.

Produces a frequency-domain representation of every channel. Output dims are (*non_time_dims, "frequency"); time is consumed. When data.sampling_rate is available the frequency coordinate is in Hz, otherwise it falls back to cycles-per-sample.

Parameters:

Name Type Description Default
norm Literal['real', 'psd'] | None

If None (default), returns the raw complex coefficients. If "real" returns |FFT| as a real-valued array. If "psd" returns the power spectral density |FFT|^2.

None
cutoff float | None

None (default) or float (in Hz). Largest frequency returned. If None or cutoff > Nyquist freq, parameter is ignored and output contains frequencies up to Nyquist frequency self.sampling_rate // 2. Given value must be positive.

None

Returns:

Type Description

class:~cobrabox.Data with dims (*non_time_dims, "frequency"). Dtype is

complex128 when norm=None and float64 otherwise.

Raises:

Type Description
ValueError

If the input Data has no known sampling_rate.

ValueError

If norm is not one of [None, 'real', 'psd'].

ValueError

If negative cutoff is given.

UserWarning

If data does not have sampling_rate defined.

UserWarning

If cutoff larger than Nyquist frequency given.

__post_init__

__post_init__() -> None

Validate parameters after initialization.

Source code in src/cobrabox/transforms/_fourier_transform.py
def __post_init__(self) -> None:
    """Validate parameters after initialization."""
    if self.cutoff is not None and self.cutoff < 0:
        raise ValueError(f"cutoff must be positive, got {self.cutoff}")

    if self.norm not in [None, "real", "psd"]:
        raise ValueError("norm must be one of [None, 'real', 'psd']")

cobrabox.surrogates.FourierTransformSurrogates dataclass

FourierTransformSurrogates(
    n_surrogates: int,
    multivariate: bool = True,
    return_data: bool = True,
    random_state: Generator | int | None = None,
)

Bases: SplitterFeature[SignalData]

Generate Fourier transform surrogates (preserving autocorrelation) of SignalData.

Creates surrogate time series by randomizing the phases of the Fourier transform while preserving the power spectrum (and thus autocorrelation). This generates null-hypothesis data for testing whether observed effects exceed what would be expected from linear, stationary, Gaussian processes with the same correlation structure.

Parameters:

Name Type Description Default
n_surrogates int

Number of surrogate time series to generate. Must be a non-negative integer.

required
multivariate bool

If True (default), applies the same random phases to all series. This ensures that cross-correlations are approximately preserved.

True
return_data bool

If True (default), the generator yields the original data first, followed by the surrogates. This gives length 1 + n_surrogates.

True
random_state Generator | int | None

Initialiser for the pseudorandom number generator to ensure reproducibility. Can be a Generator, int seed, or None.

None

Returns:

Type Description

Iterator yielding SignalData objects containing the surrogate time series

(and optionally the original data first).

Raises:

Type Description
ValueError

If n_surrogates is not an integer.

ValueError

If n_surrogates is negative.

References

Theiler, J., Eubank, S., Longtin, A., Galdrikian, B., & Farmer, J. D. (1992). Testing for nonlinearity in time series: the method of surrogate data. Physica D: Nonlinear Phenomena, 58(1-4), 77-94. https://doi.org/10.1016/0167-2789(92)90102-S

Example

import cobrabox as cb data = cb.SignalData.from_numpy( ... np.random.randn(100, 10), ... dims=["time", "space"], ... sampling_rate=100.0 ... ) surrogates = list(cb.FourierTransformSurrogates( ... n_surrogates=5, random_state=42 ... )(data))

cobrabox.infometrics.FractalDimension dataclass

FractalDimension(
    method: Literal["higuchi", "katz"] = "higuchi",
    k_max: int = 10,
)

Bases: BaseFeature[SignalData]

Compute fractal dimension over the time dimension.

Selects the algorithm via the method parameter:

  • "higuchi" — Higuchi Fractal Dimension. Constructs k sub-series for each interval k = 1…k_max, computes their normalised average curve length L(k), then estimates the fractal dimension as the slope of log(L(k)) vs log(1/k) via least-squares regression. Tunable via k_max. Values close to 1 indicate a smooth signal; values close to 2 indicate a highly irregular signal.

  • "katz" — Katz Fractal Dimension. Models the signal as a 2-D curve and estimates fractal dimension from the total Euclidean path length, step count, and maximum planar distance from the first sample. Parameter-free, O(N).

Parameters:

Name Type Description Default
method Literal['higuchi', 'katz']

Which fractal dimension algorithm to use. One of "higuchi" (default) or "katz".

'higuchi'
k_max int

Maximum interval for the Higuchi algorithm. Must be >= 2. Default is 10. Ignored when method="katz" (a warning is issued when a non-default value is passed but unused).

10

Returns:

Type Description

class:~cobrabox.Data with the time dimension removed. Shape is

(space,) for standard input. Values are dimensionless floats

typically in [1, 2] for Higuchi and >= 1 for Katz.

Raises:

Type Description
ValueError

If method is not one of the valid options, or if k_max < 2 when using the Higuchi method.

References

Higuchi, T. (1988). Approach to an irregular time series on the basis of the fractal theory. Physica D, 31(2), 277-283.

Katz, M. J. (1988). Fractals and the analysis of waveforms. Computers in Biology and Medicine, 18(3), 145-156.

Example

result = cb.FractalDimension().apply(data) # Higuchi, k_max=10 result = cb.FractalDimension(method="higuchi", k_max=20).apply(data) result = cb.FractalDimension(method="katz").apply(data)

cobrabox.connectivity.GrangerCausality dataclass

GrangerCausality(
    coords: list[str] | list[int] | None = None,
    lag: int | None = None,
    maxlag: int = 1,
)

Bases: BaseFeature[SignalData]

Compute the Granger causality matrix across channels.

For each ordered pair (i, j), measures whether channel j Granger-causes channel i via the log-ratio of prediction-error variances of two VAR models (restricted: past of i only; unrestricted: past of both i and j). Self-causality is undefined and reported as NaN on the diagonal.

Parameters:

Name Type Description Default
coords list[str] | list[int] | None

Channels to include. None (default) computes the full (K, K) matrix across all space coordinates. Pass a list of coordinate names to restrict the output (each pair is an independent OLS solve, so this is a real compute saving).

None
lag int | None

Specific lag order to test. When set, returns a 2-D matrix.

None
maxlag int

Maximum lag to test when lag is None. The output gains a lag_index dimension when maxlag > 1. Default: 1.

1

Returns:

Type Description

class:~cobrabox.Data with dims (space_to, space_from) (or

(space_to, space_from, lag_index) when multiple lags are tested).

result.sel(space_to=A, space_from=B) is the causal influence

B → A. Diagonal entries are NaN.

Raises:

Type Description
ValueError

If maxlag < 1, lag < 1 when provided, or coords is an empty list.

Example

Full matrix at lag 2

result = cb.GrangerCausality(lag=2).apply(data)

Restricted to two channels at the default lag

result = cb.GrangerCausality(coords=["F3", "F4"]).apply(data)

Range of lags

result = cb.GrangerCausality(maxlag=4).apply(data)

cobrabox.transforms.InverseFourierTransform dataclass

InverseFourierTransform(
    n: int | None = None, sampling_rate: float | None = None
)

Bases: BaseFeature[Data]

Inverse of :class:~cobrabox.transforms.fourier_transform.FourierTransform.

Takes an array with a frequency dimension (complex coefficients) and returns a time-domain reconstruction.

Parameters:

Name Type Description Default
n int | None

Length of the output time axis. If None, defaults to 2 * (n_freq - 1).

None
sampling_rate float | None

Sampling rate to attach to the reconstructed time axis. When None the resulting Data has no sampling rate.

None

Returns:

Type Description

class:~cobrabox.Data with the frequency dim replaced by time.

cobrabox.infometrics.LempelZiv dataclass

LempelZiv()

Bases: BaseFeature[SignalData]

Compute Lempel-Ziv Complexity (LZC) over the time dimension.

The signal is binarised relative to its mean (values above the mean → 1, values at or below → 0). LZC counts distinct patterns in the resulting binary sequence and normalises by the theoretical maximum, giving a value in (0, 1] where higher values indicate greater irregularity.

The counting algorithm follows Lempel & Ziv (1976) as implemented in NeuroKit2 (https://github.com/neuropsychology/NeuroKit, MIT licence).

Returns:

Type Description

class:~cobrabox.Data with the time dimension removed. Shape is

(space,) for standard input. Values are dimensionless floats in

(0, 1].

References

Lempel, A., & Ziv, J. (1976). On the complexity of finite sequences. IEEE Transactions on Information Theory, 22(1), 75-81.

Example

result = cb.LempelZiv().apply(data)

cobrabox.signalstats.LineLength dataclass

LineLength()

Bases: BaseFeature[SignalData]

Compute line length over the time dimension.

Line length is the sum of absolute differences between consecutive timepoints. A larger value indicates a more rapidly varying signal.

Returns:

Type Description

class:~cobrabox.Data with the time dimension removed. Shape is

(space,) for standard input, or (*extra_dims, space) if

additional dimensions are present (e.g. window_index). Values

are in the same units as the input signal.

Example

result = cb.LineLength().apply(data)

cobrabox.signalstats.Max dataclass

Max(dim: str)

Bases: BaseFeature[Data]

Compute the maximum value across a dimension.

Parameters:

Name Type Description Default
dim str

Name of the dimension to reduce over (e.g. "time").

required

Returns:

Type Description

class:~cobrabox.Data with dim removed. Shape is the input shape

minus the reduced dimension. Values are the per-position maximum

in the same units as the input data.

Raises:

Type Description
ValueError

If the specified dimension is not found in the data.

Example

result = cb.Max(dim="time").apply(data)

cobrabox.signalstats.Mean dataclass

Mean(dim: str)

Bases: BaseFeature[Data]

Compute the mean value across a dimension.

Parameters:

Name Type Description Default
dim str

Name of the dimension to reduce over (e.g. "time").

required

Returns:

Type Description

class:~cobrabox.Data with dim removed. Shape is the input shape

minus the reduced dimension. Values are the per-position arithmetic

mean in the same units as the input signal.

Raises:

Type Description
ValueError

If the specified dimension is not found in the data.

Example

result = cb.Mean(dim="time").apply(data)

cobrabox.windowing.MeanAggregate dataclass

MeanAggregate()

Bases: AggregatorFeature

Aggregate a stream of per-window Data by averaging across windows.

Collects all windows, stacks them along a temporary 'window' dimension, and reduces with mean. Per-window pipeline history is propagated to the result.

Returns:

Type Description

A new Data object with the same shape as each individual window

result (the window stacking dimension is reduced away). All

metadata from the original data is preserved, including

sampling_rate. History includes all per-window operations

followed by "MeanAggregate".

Example

chord = ( ... cb.SlidingWindow(window_size=100, step_size=50) ... | cb.LineLength() ... | cb.MeanAggregate() ... ) result = chord.apply(data)

cobrabox.signalstats.Min dataclass

Min(dim: str)

Bases: BaseFeature[Data]

Compute the minimum value across a dimension.

Parameters:

Name Type Description Default
dim str

Name of the dimension to reduce over (e.g. "time").

required

Returns:

Type Description

class:~cobrabox.Data with dim removed. Shape is the input shape

minus the reduced dimension. Values are the per-position minimum

in the same units as the input signal.

Example

result = cb.Min(dim="time").apply(data)

cobrabox.connectivity.MutualInformation dataclass

MutualInformation(
    dim: str = "time",
    other_dim: str | None = None,
    bins: int | None = None,
    equiprobable_bins: bool = True,
    log_base: float = 2.0,
)

Bases: BaseFeature[SignalData]

Compute mutual information (MI) between all pairs of series along a specified dimension (by default 'time').

Mutual information is a measure of the mutual dependence between two variables. It is defined as MI(X, Y) = H(X) + H(Y) - H(X, Y), where H(X) is the entropy of X and H(X, Y) is the joint entropy of X and Y. The entropy is computed by discretizing the data into bins and applying the formula H(X) = -sum(p_i * log(p_i)), where p_i is the empirical probability of the i-th bin.

The number of bins can be specified by the user or determined heuristically based on the size of the data along the specified dimension as $n^{1/3}$. The bins can be equiprobable (i.e., each bin has the same number of samples) or equidistant (i.e., each bin has the same width).

Parameters:

Name Type Description Default
dim str

The dimension along which to compute MI. Default is "time".

'time'
other_dim str | None

The other dimension to consider when computing MI. This is mandatory when the input data has more than two dimensions. Default is None.

None
bins int | None

The number of bins to use for discretizing the data. If None, it will be determined heuristically based on the size of the data along the specified dimension. Default is None.

None
equiprobable_bins bool

Whether to use equiprobable bins (True) or equidistant bins (False) for discretizing the data. Default is True.

True
log_base float

The logarithm base to use for computing MI. Default is 2.0.

2.0

Returns:

Name Type Description
A

class:~cobrabox.Data with dimensions "space_from" and "space_to"

containing the pairwise mutual information matrix. The shape is

(..., n_space, n_space) where n_space is the size of the dimension

specified by other_dim.

Raises:

Type Description
ValueError

If bins is not a positive integer.

ValueError

If dim is not a string.

ValueError

If other_dim is not a string or None.

ValueError

If dim is not found in the data.

ValueError

If other_dim is not found in the data.

ValueError

If other_dim must be specified for data with more than 2 dimensions.

References

Shannon, C. E. (1948). A mathematical theory of communication. Bell System Technical Journal, 27(3), 379-423.

Example

import cobrabox as cb import numpy as np data = cb.SignalData.from_numpy( ... np.random.randn(10, 100), ... dims=["space", "time"], ... sampling_rate=100.0 ... ) result = cb.MutualInformation().apply(data) result.data.shape (10, 10)

cobrabox.infometrics.Nonreversibility dataclass

Nonreversibility()

Bases: BaseFeature[SignalData]

Compute dc_norm: normalised deviation from causal normality (time-irreversibility).

Fits a VAR(1) model in forward and reverse time directions:

Forward:  X_t = A @ X_{t-1} + noise
Reverse:  X_t = B @ X_{t+1} + noise

Both A and B are rescaled so their spectral radius is strictly less than 1. dc_norm quantifies the asymmetry between forward and reverse dynamics:

dc_norm = ||A - B^T||_F / (||A + B^T||_F + ||A - B^T||_F)

Result is bounded in [0, 1). A dc_norm of 0 indicates perfect time-reversibility; larger values indicate stronger irreversibility.

Returns:

Type Description

class:~cobrabox.Data with a single spatial coordinate ('dc_norm').

The time dimension is removed.

Raises:

Type Description
ValueError

If 'time' dimension is missing or has fewer than 2 timepoints.

ValueError

If 'space' dimension has fewer than 2 channels.

Example

result = cb.Nonreversibility().apply(data)

References

TODO: Add citation for the dc_norm / time-irreversibility VAR(1) measure.

cobrabox.transforms.NotchFilter dataclass

NotchFilter(
    freq: float, q: float = 30.0, zero_phase: bool = True
)

Bases: BaseFeature[SignalData]

Apply a notch filter to remove a specific frequency.

Attenuates a narrow band centered on freq using :func:scipy.signal.iirnotch followed by filtering along the time axis. All channels are filtered independently.

Parameters:

Name Type Description Default
freq float

Center frequency in Hz to remove (e.g. 50 or 60 Hz for power-line noise).

required
q float

Quality factor controlling the notch width. Higher values create narrower notches. Defaults to 30.0.

30.0
zero_phase bool

If True (default), uses :func:scipy.signal.filtfilt for zero-phase (forward-backward) filtering with no phase distortion — the MNE-style default for EEG preprocessing. If False, uses :func:scipy.signal.lfilter, a causal forward-only IIR filter that introduces phase distortion but may be preferable for causal or online processing.

True

Raises:

Type Description
ValueError

If freq is not positive, q is not positive, freq exceeds the Nyquist frequency, or the input has no sampling_rate.

Returns:

Type Description

class:~cobrabox.SignalData with the same shape, dimensions, and

metadata as the input.

Example

result = cb.NotchFilter(freq=50.0).apply(data) result = cb.NotchFilter(freq=60.0, q=30.0).apply(data) result = cb.NotchFilter(freq=50.0, zero_phase=False).apply(data)

cobrabox.connectivity.PartialCorrelation dataclass

PartialCorrelation(
    coords: list[str] | list[int] | None = None,
    control_vars: list[str] | list[int] | None = None,
)

Bases: BaseFeature[SignalData]

Compute the partial-correlation matrix across channels.

Each off-diagonal entry result[i, j] is the correlation between channels i and j after partialling out the influence of every other channel in the matrix (and any explicit control_vars). Computed via inversion of the full correlation matrix — conditioning therefore requires all channels, so coords only filters the output; the precision-matrix inversion is still O(K³) in the full channel count.

Parameters:

Name Type Description Default
coords list[str] | list[int] | None

Channels to include in the output. None (default) returns the full (K, K) matrix; pass a list of coordinate names to restrict the output (compute cost is unchanged).

None
control_vars list[str] | list[int] | None

Additional coordinates to control for that are not part of the output matrix. When None and coords is set, no extra controls are added (only the channels in coords enter the precision matrix).

None

Returns:

Type Description

class:~cobrabox.Data with dims (space_to, space_from) and shape

(len(coords), len(coords)). Symmetric, with 1 on the diagonal.

Raises:

Type Description
ValueError

If the correlation matrix is singular (perfect collinearity or too few samples), if coords is an empty list, or if any named coordinate is missing from the space dimension.

Example

Full matrix

pc = cb.PartialCorrelation().apply(data)

Two-channel subset, control on an extra channel

pc = cb.PartialCorrelation( ... coords=["F3", "F4"], control_vars=["Cz"] ... ).apply(data)

cobrabox.connectivity.PartialDirectedCoherence dataclass

PartialDirectedCoherence(
    var_order: int | None = None, n_freqs: int = 128
)

Bases: BaseFeature[SignalData]

Estimate the Partial Directed Coherence (PDC) between channels via a VAR model.

Fits a multivariate autoregressive (VAR) model to the input time series and derives the PDC spectrum from the estimated coefficient matrices.

For each frequency bin, PDC[i, j, f] represents the normalized directional influence from channel j to channel i. Values lie in [0, 1] and sum to 1 across the space_to (sink) dimension for each source channel and frequency bin: sum_i PDC[i, j, f]^2 = 1 for all j, f.

Parameters:

Name Type Description Default
var_order int | None

Number of lags for the VAR model. None (default) lets :func:statsmodels.tsa.api.VAR pick the optimal order via AIC.

None
n_freqs int

Number of frequency bins in [0, sr/2]. Default 128.

128

Returns:

Type Description

class:~cobrabox.Data with dims ("space_to", "space_from", "frequency").

Raises:

Type Description
ValueError

If data.sampling_rate is None, the input is not 2-D, there are fewer than 2 channels, or var_order < 1 / n_freqs < 1.

References

Baccalá, L. A., & Sameshima, K. (2001). Partial directed coherence: a new concept in neural structure determination. Biological Cybernetics 84(6), 463-474.

cobrabox.connectivity.PhaseLockingValue dataclass

PhaseLockingValue(
    coords: list[str] | list[int] | None = None,
)

Bases: BaseFeature[SignalData]

Compute the phase-locking-value (PLV) matrix across channels.

For each ordered pair (i, j), computes the PLV between channels i and j via instantaneous phase differences (extracted with the Hilbert transform). PLV is symmetric, takes values in [0, 1], and is 1 on the diagonal.

Parameters:

Name Type Description Default
coords list[str] | list[int] | None

Channels to include. None (default) computes the full (K, K) matrix across all space coordinates; pass a list of coordinate names to restrict the output. Restricting saves both the Hilbert transform (run on the selected channels only) and the pairwise loop.

None

Returns:

Type Description

class:~cobrabox.Data with dims (space_to, space_from). Values are

symmetric and in [0, 1].

Raises:

Type Description
ValueError

If coords is an empty list or any coordinate is not found in the space dimension.

Example

Full matrix

plv = cb.PhaseLockingValue().apply(data)

Single pair

plv = cb.PhaseLockingValue(coords=["F3", "F4"]).apply(data)

cobrabox.spectral.PowerSpectralDensity dataclass

PowerSpectralDensity(cutoff: float | None = None)

Bases: BaseFeature[SignalData]

Power spectral density via FFT.

Convenience wrapper around :class:~cobrabox.FourierTransform with norm="psd". Computes the one-sided PSD as |FFT|^2 / (N * sampling_rate) with the interior bins doubled to account for the discarded negative frequencies.

Parameters:

Name Type Description Default
cutoff float | None

Optional upper frequency limit in Hz. Frequencies above cutoff are discarded. If None (default) the full spectrum up to the Nyquist frequency is returned.

None

Returns:

Type Description

class:~cobrabox.Data with dims (*non_time_dims, "frequency").

Dtype is float64.

Raises:

Type Description
ValueError

If cutoff is negative.

UserWarning

If data has no sampling_rate.

UserWarning

If cutoff exceeds the Nyquist frequency.

cobrabox.connectivity.ReciprocalConnectivity dataclass

ReciprocalConnectivity(
    freq_band: tuple[float, float] | None = (30.0, 80.0),
    normalize: bool = False,
)

Bases: BaseFeature[Data]

Compute per-channel Reciprocal Connectivity (RC) from a directed connectivity matrix.

Reciprocal Connectivity quantifies the net directional role of each channel: positive values indicate a net sink (more influence received than sent), and negative values indicate a net source (more influence sent than received).

For each channel i::

RC[i] = in_strength[i] - out_strength[i]

where (with mat[i, j] = flow from j to i):

  • in_strength[i] = mean over j of mat[i, j, band]
  • out_strength[i] = mean over j of mat[j, i, band]

Matrix-only input. To compute RC from a raw signal, pipe it through a directed connectivity feature first::

rc = (PartialDirectedCoherence() | ReciprocalConnectivity()).apply(signal)

Parameters:

Name Type Description Default
freq_band tuple[float, float] | None

Frequency band (fmin, fmax) in Hz to average over. Required when the input matrix has a "frequency" dimension; pass None when the input is already a single matrix. Default (30.0, 80.0) Hz.

(30.0, 80.0)
normalize bool

If True, z-score the matrix (excluding the diagonal) before computing in/out strengths.

False

Returns:

Type Description

class:~cobrabox.Data with dim ("space",), shape (n_channels,).

Raises:

Type Description
ValueError

If the input lacks "space_to" and "space_from" dimensions, if the matrix is asymmetric vs symmetric requirements are violated, or if freq_band is incompatible with the input.

Example

rc = cb.ReciprocalConnectivity(freq_band=(30.0, 80.0)).apply(pdc_matrix) rc = (cb.PartialDirectedCoherence() | rc).apply(signal_data)

cobrabox.infometrics.RecurrenceMatrix dataclass

RecurrenceMatrix(
    rec_metric: RecMetric = "cosine",
    fc_options: list[str | int | float] = list(),
)

Bases: BaseFeature[SignalData]

Compute a pairwise recurrence (self-similarity) matrix from a time-series.

Behaviour depends on the shape of the input and the contents of fc_options:

2-D input (N, T) — multivariate time-series:

  • fc_options=[] (default): each time-point is used as a state vector directly. Output shape (T, T).
  • fc_options=['pearson']: FC computed per window with default window_size=10 and overlap=0.5.
  • fc_options=['pearson', 50]: FC with window_size=50, default overlap=0.5.
  • fc_options=['pearson', 50, 0.25]: FC with window_size=50 and overlap=0.25.
  • Supported FC metrics: 'pearson', 'spearman', 'MI', 'PLV', 'AEC'.
  • A UserWarning is raised if window_size < 5.

3-D input (N, N, T) — time-series of FC matrices:

Already-computed FC matrices; fc_options is ignored. Output shape (T, T).

Parameters:

Name Type Description Default
rec_metric RecMetric

Pairwise similarity metric. One of 'cosine' (default), 'correlation', 'euclidean'.

'cosine'
fc_options list[str | int | float]

List controlling window-based FC computation. [] = state-vector mode. [fc_metric] = FC with defaults. Supported: 'pearson', 'spearman', 'MI', 'PLV', 'AEC'. [fc_metric, window_size] = FC with given window size. [fc_metric, window_size, overlap] = full control.

list()

Returns:

Type Description

class:~cobrabox.Data of shape (n, n) with dims ('t1', 't2').

Raises:

Type Description
ValueError

If a required dimension is missing.

ValueError

If window_size >= n_time.

ValueError

If metric values are invalid.

Example

state-vector mode (default)

rec = cb.RecurrenceMatrix().apply(data)

window/FC mode — just fc_metric, rest default

rec = cb.RecurrenceMatrix('cosine', ['pearson']).apply(data)

window/FC mode — full control

rec = cb.RecurrenceMatrix('cosine', ['pearson', 50, 0.25]).apply(data)

3-D pre-computed FC time-series

rec = cb.RecurrenceMatrix('cosine').apply(fc_data)

References

Eckmann, J. P., Kamphorst, S. O., & Ruelle, D. (1987). Recurrence plots of dynamical systems. Europhysics Letters, 4(9), 973-977. https://doi.org/10.1209/0295-5075/4/9/004

Marwan, N., Romano, M. C., Thiel, M., & Kurths, J. (2007). Recurrence plots for the analysis of complex systems. Physics Reports, 438(5-6), 237-329. https://doi.org/10.1016/j.physrep.2006.11.001

cobrabox.decompositions.SVD dataclass

SVD(
    dim: str,
    n_components: int = 10,
    center: bool = True,
    zscore: bool = False,
    mask: DataArray | None = None,
    return_unstacked_V: bool = True,
    output: _SVDOutputMode = "V",
)

Bases: BaseFeature[Data]

Compute truncated SVD over one dimension.

The input data.data may be N-D (fMRI, EEG, etc.). We treat dim as the sample axis and stack all other dims into a single feature axis internally.

Parameters:

Name Type Description Default
dim str

Sample dimension (e.g. "time", "trial").

required
n_components int

Number of components to return.

10
center bool

If True, subtract each feature's baseline (mean over dim) before SVD. Intuition: removes offsets/mean image so components describe fluctuations rather than the average level.

True
zscore bool

If True, after centering, also normalize each feature by its variability (divide by std over dim). Intuition: prevents a few "loud"/high-variance features from dominating; components reflect shared patterns rather than amplitude. Implies centering.

False
mask DataArray | None

Optional boolean selector over non-dim dims. True keeps a feature (voxel/channel/etc.), False removes it. Use mask to restrict SVD to a subset of features, for instance to exclude irrelevant/noisy features (e.g. non-brain voxels, bad EEG channels), speed up computation, and focus on a region/sensor set.

None
return_unstacked_V bool

If True, return V reshaped back to the original non-dim dims.

True
output _SVDOutputMode

Which primary output to return as xr.DataArray: - "V" (default): feature patterns/maps, shape (component, ...) - "U": sample scores/timecourses, shape (dim, component)

'V'

Returns:

Type Description

class:~cobrabox.Data: If output="V": (component, ...) feature patterns/maps If output="U": (dim, component) sample scores/timecourses

Extra outputs are stored in result.attrs["svd"]: S, Vh, mean, std, and whichever of U/V is not returned.

Example

fMRI (time,x,y,z) + mask -> returns V(component,x,y,z)

result = cb.SVD(dim="time", n_components=10, mask=brain_mask).apply(data)

EEG (time,channel) -> returns V(component,channel)

result = cb.SVD(dim="time", n_components=10).apply(eeg_data)

EEG time-frequency (time,channel,freq) + zscore -> V(component,channel,freq)

result = cb.SVD(dim="time", n_components=10, zscore=True).apply(tfr_data)

Trial-wise (trial,time,channel) -> V(component,time,channel), U stored in attrs

result = cb.SVD(dim="trial", n_components=10).apply(epochs_data)

No unstacking -> main output is Vh(component,features)

out = cb.SVD(dim="time", n_components=10, return_unstacked_V=False).apply(data) svd = out.data.attrs["svd"] U = svd["U"] # sample scores / timecourses (if output="V") S = svd["S"] # singular values

Raises:

Type Description
ValueError

If dim is not found in the input data dimensions.

ValueError

If n_components is not positive.

ValueError

If output is not one of {"V", "U"}.

References

Golub, G. H., & Kahan, W. (1965). Calculating the singular values and pseudo-inverse of a matrix. Journal of the Society for Industrial and Applied Mathematics: Series B, Numerical Analysis, 2(2), 205-224.

cobrabox.infometrics.SampleEntropy dataclass

SampleEntropy(
    m: int = 2, r: float | None = None, log_base: float = 2
)

Bases: BaseFeature[SignalData]

Sample Entropy feature.

Sample entropy quantifies the regularity of a time-series. It is the negative logarithm of the conditional probability that two sequences similar for m points remain similar when one more point is included. Lower values indicate more regular (predictable) signals, while higher values indicate greater complexity.

By default, the binary logarithm (base 2) is used, diverging from the original definition which uses the natural logarithm. This can be configured via the log_base parameter.

The feature works on any SignalData that contains a time dimension. The time dimension is collapsed and all other dimensions are preserved.

Parameters:

Name Type Description Default
m int

Embedding dimension (length of compared sequences). Must be >= 1.

2
r float | None

Tolerance for matching sequences. If None a default of 0.2 * std(signal) is used, where std is the standard deviation of the time series.

None
log_base float

Base of the logarithm used in the entropy calculation. Defaults to 2 (binary logarithm). Use np.e for the natural logarithm (original definition) or 10 for base-10 logarithm.

2

Returns:

Name Type Description
A

class:~cobrabox.Data containing the sample entropy with the

time dimension collapsed.

Raises:

Type Description
ValueError

If embedding dimension m is less than 1.

ValueError

If logarithm base is invalid (<= 0 or == 1).

ValueError

If time series length is not greater than embedding dimension m.

Example

entropy = cb.SampleEntropy(m=2).apply(data) # base-2 (default) entropy_nat = cb.SampleEntropy(m=2, log_base=np.e).apply(data) # natural log

cobrabox.windowing.SlidingWindow dataclass

SlidingWindow(window_size: int = 10, step_size: int = 5)

Bases: SplitterFeature[SignalData]

Yield one Data per sliding window over the time dimension.

Lazily generates windows to avoid materialising all windows in memory at once.

Parameters:

Name Type Description Default
window_size int

Number of timepoints per window. Must be >= 1.

10
step_size int

Step between window starts in timepoints. Must be >= 1.

5

Returns:

Type Description

Generator of Data objects. Each yielded item has the same

dimensions as the input with the time axis sliced to

window_size samples. The string "SlidingWindow" is appended

to history on each yielded window. All other metadata is

preserved.

Each window also carries its position on the original time axis in

extra: window_start (time of the window's first sample) and

window_end (time of its last sample), taken from the input's

time coordinate — in seconds when a sampling rate is known, in

sample indices otherwise. These survive per-window features that

consume the time dimension, which is how ConcatAggregate can

label the stacked window axis with real times.

Example

windows = list(cb.SlidingWindow(window_size=100, step_size=50)(data)) len(windows) # number of windows depends on data length windows[0].data.sizes["time"] 100 windows[1].extra["window_start"] # 100 Hz data, step of 50 samples 0.5

cobrabox.windowing.SlidingWindowReduce dataclass

SlidingWindowReduce(
    window_size: int = 10,
    step_size: int = 5,
    dim: str = "time",
    agg: Literal[
        "mean", "std", "sum", "min", "max"
    ] = "mean",
)

Bases: BaseFeature[SignalData]

Sliding window with automatic per-window reduction.

Creates sliding windows over a dimension, applies an aggregation function to each window, and stacks results along a new "window" dimension.

This combines windowing + aggregation into a single feature, avoiding the need for a Chord when you just want reduced window statistics.

Parameters:

Name Type Description Default
window_size int

Number of samples per window. Must be >= 1.

10
step_size int

Step between window starts in samples. Must be >= 1.

5
dim str

Name of the dimension to window over and reduce (default: "time").

'time'
agg Literal['mean', 'std', 'sum', 'min', 'max']

Aggregation function to apply to each window. One of: "mean", "std", "sum", "min", "max".

'mean'

Returns:

Type Description

Data with the reduced dimension removed and a new "window" dimension

added. The "window" dimension has length equal to the number of

windows that fit in the input data. It is coordinate-labelled with each

window's start position on the reduced axis (seconds when windowing

over time with a known sampling rate), and a non-dimension

window_end coordinate carries the matching end positions.

Example

Mean of each 100-sample window, stepping by 50

result = cb.SlidingWindowReduce( ... window_size=100, step_size=50, dim="time", agg="mean" ... ).apply(data) result.data.dims ('window', 'channel') # time dimension is reduced, window dim added result.data.window.values[:3] # 100 Hz data, step of 50 samples array([0. , 0.5, 1. ])

cobrabox.spectral.Spectrogram dataclass

Spectrogram(
    nperseg: int | None = None,
    noverlap: int | None = None,
    window: str = "hann",
    scaling: str = "log",
)

Bases: BaseFeature[SignalData]

Compute the power spectrogram for each spatial channel.

Uses scipy.signal.spectrogram (Welch-style STFT) to estimate the short-time power spectral density (or magnitude spectrum) for every channel independently. Extra dimensions (e.g. window_index) are preserved: the spectrogram is computed along the time axis for each slice of those dimensions.

Parameters:

Name Type Description Default
nperseg int | None

Samples per FFT segment. Controls frequency resolution. Defaults to min(256, n_time). Must be >= 2 and <= n_time.

None
noverlap int | None

Overlap between consecutive segments in samples. Defaults to nperseg // 2 (50 % overlap). Must be < nperseg.

None
window str

Window function name passed to scipy (e.g. "hann", "hamming"). Defaults to "hann".

'hann'
scaling str

Output scaling. One of:

  • "log" (default) — 10 · log₁₀(PSD) in dB re 1 V²/Hz. Values where PSD ≈ 0 are clamped to avoid -inf.
  • "density" — power spectral density in V²/Hz.
  • "spectrum" — power spectrum in V² (not normalised by bandwidth).
  • "magnitude" — |STFT| (absolute value of complex STFT coefficients, computed via scipy.signal.stft).
'log'

Returns:

Type Description

class:~cobrabox.SignalData with dims ``(*extra_dims, "space", "frequency",

"time")wherefrequencyis in Hz andtime`` holds the window

centre times in seconds (as returned by scipy).

Raises:

Type Description
ValueError

If scaling is not one of the four valid options, if nperseg is invalid, or if noverlap >= nperseg.

Example

data = cb.load_dataset("dummy_random")[0] sg = cb.Spectrogram().apply(data) sg.data.dims ('space', 'frequency', 'time')

cobrabox.signalstats.SpikeCount dataclass

SpikeCount()

Bases: BaseFeature[Data]

Calculate spikes in the input data using the IQR method.

Detects outliers as values falling outside ±1.5*IQR from Q1/Q3. Returns a scalar count of detected spikes.

Returns:

Type Description

class:~cobrabox.Data with shape (), dims (),

containing the spike count as a scalar float value.

Raises:

Type Description
ValueError

If input data is empty.

Example

result = SpikeCount().apply(data)