ezmsg.event.kernel

Kernel abstractions for sparse event processing.

Kernels can be applied to sparse events to produce either: 1. Dense signals (via SparseKernelInserter) 2. Binned activation features (via BinnedKernelActivation)

Functions

alpha_kernel(t, sigma)

Alpha function kernel: k(t) = (t/sigma^2) * exp(-t/sigma) for t >= 0.

Peaks at t = sigma. Normalized so that integral from 0 to inf equals 1.

Parameters:

• t (ndarray[tuple[Any, ...], dtype[_ScalarT]])

• sigma (float)

Return type:

ndarray[tuple[Any, …], dtype[_ScalarT]]

boxcar_kernel(t, sigma)

Boxcar (rectangular) kernel: k(t) = 1/(2*sigma) for abs(t) < sigma.

Symmetric (acausal). Width is 2*sigma. Normalized so that integral equals 1.

Parameters:

• t (ndarray[tuple[Any, ...], dtype[_ScalarT]])

• sigma (float)

Return type:

ndarray[tuple[Any, …], dtype[_ScalarT]]

causal_boxcar_kernel(t, sigma)

Causal boxcar kernel: k(t) = 1/sigma for 0 <= t < sigma.

Normalized so that integral equals 1.

Parameters:

• t (ndarray[tuple[Any, ...], dtype[_ScalarT]])

• sigma (float)

Return type:

ndarray[tuple[Any, …], dtype[_ScalarT]]

exponential_kernel(t, sigma)

Causal exponential decay kernel: k(t) = exp(-t/sigma) / sigma for t >= 0.

Normalized so that integral from 0 to inf equals 1.

Parameters:

• t (ndarray[tuple[Any, ...], dtype[_ScalarT]])

• sigma (float)

Return type:

ndarray[tuple[Any, …], dtype[_ScalarT]]

gaussian_kernel(t, sigma)

Gaussian kernel: k(t) = exp(-t^2 / (2*sigma^2)) / (sigma * sqrt(2*pi)).

Symmetric (acausal). Normalized so that integral equals 1.

Parameters:

• t (ndarray[tuple[Any, ...], dtype[_ScalarT]])

• sigma (float)

Return type:

ndarray[tuple[Any, …], dtype[_ScalarT]]

Classes

class ArrayKernel(data, pre_samples=0)

Bases: Kernel

Kernel from explicit array (e.g., spike waveforms).

Parameters:

• data (ndarray[tuple[Any, ...], dtype[_ScalarT]]) – 1D array of kernel values.

• pre_samples (int) – Number of samples before t=0. Default 0 (causal kernel). For a waveform centered at t=0, use pre_samples = len(data) // 2.

__init__(data, pre_samples=0)

Parameters:

• data (ndarray[tuple[Any, ...], dtype[_ScalarT]])

• pre_samples (int)

property length: int

Total kernel length in samples.

property pre_samples: int

Number of samples before t=0 (for acausal kernels).

property data: ndarray[tuple[Any, ...], dtype[float64]]

Raw kernel data array.

evaluate(t)

Evaluate kernel at time offsets.

Parameters:

t (ndarray[tuple[Any, ...], dtype[floating]])

Return type:

ndarray[tuple[Any, …], dtype[floating]]

class FunctionalKernel(func, sigma, fs, truncate_at=5.0, causal=True)

Bases: Kernel

Kernel from a function (e.g., exponential decay, Gaussian).

The function should accept (t, sigma) where t is time in samples and sigma is the time constant in samples.

Parameters:

• func (Callable[[ndarray[tuple[Any, ...], dtype[_ScalarT]], float], ndarray[tuple[Any, ...], dtype[_ScalarT]]]) – Kernel function f(t, sigma) -> values.

• sigma (float) – Time constant in seconds.

• fs (float) – Sample rate in Hz (for converting sigma to samples).

• truncate_at (float) – Truncate kernel at this many time constants. Default 5.0.

• causal (bool) – If True, kernel is zero for t < 0. Default True.

Example

>>> kernel = FunctionalKernel(
...     func=lambda t, s: (t >= 0) * np.exp(-t / s) / s,
...     sigma=0.010,  # 10ms
...     fs=30000,
... )

__init__(func, sigma, fs, truncate_at=5.0, causal=True)

Parameters:

• func (Callable[[ndarray[tuple[Any, ...], dtype[_ScalarT]], float], ndarray[tuple[Any, ...], dtype[_ScalarT]]])

• sigma (float)

• fs (float)

• truncate_at (float)

• causal (bool)

property length: int

Total kernel length in samples.

property pre_samples: int

Number of samples before t=0 (for acausal kernels).

property sigma: float

Time constant in seconds.

property sigma_samples: float

Time constant in samples.

evaluate(t)

Evaluate kernel function at time offsets.

Parameters:

t (ndarray[tuple[Any, ...], dtype[floating]])

Return type:

ndarray[tuple[Any, …], dtype[floating]]

class Kernel

Bases: ABC

Base class for kernels applied to sparse events.

A kernel defines a shape that gets inserted/convolved at event locations. Supports both causal (forward-looking) and acausal (symmetric) kernels.

abstract property length: int

Total kernel length in samples.

abstract property pre_samples: int

Number of samples before t=0 (for acausal kernels).

abstractmethod evaluate(t)

Evaluate kernel at time offsets t (in samples relative to event).

Parameters:

t (ndarray[tuple[Any, ...], dtype[floating]]) – Array of time offsets in samples. t=0 is the event time.

Returns:

Kernel values at the given offsets.

Return type:

ndarray[tuple[Any, …], dtype[floating]]

property is_causal: bool

True if kernel is zero for t < 0.

property post_samples: int

Number of samples at and after t=0.

class MultiKernel(kernels, default_key=None)

Bases: object

Dictionary of kernels indexed by event value.

Useful when different event types (e.g., waveform IDs 1, 2, 3) should produce different kernel shapes.

Parameters:

• kernels (dict[int, Kernel]) – Dictionary mapping event values to Kernel objects.

• default_key (int | None) – Key to use for unknown event values. Default is first key.

__init__(kernels, default_key=None)

Parameters:

• kernels (dict[int, Kernel])

• default_key (int | None)

get(value)

Get kernel for event value, falling back to default.

Parameters:

value (int)

Return type:

Kernel

property max_length: int

Maximum kernel length across all kernels.

property max_pre_samples: int

Maximum pre_samples across all kernels.

property max_post_samples: int

Maximum post_samples across all kernels.

property keys: list[int]

Available kernel keys.