ezmsg.sigproc.affinetransform#

Affine transformations via matrix multiplication: y = Ax or y = Ax + B.

For full matrix transformations where channels are mixed (off-diagonal weights), use AffineTransformTransformer or the AffineTransform unit.

For simple per-channel scaling and offset (diagonal weights only), use LinearTransformTransformer from ezmsg.sigproc.linear instead, which is more efficient as it avoids matrix multiplication.

Functions

affine_transform(weights, axis=None, right_multiply=True, channel_clusters=None, min_cluster_size=32)[source]#

Perform affine transformations on streaming data.

Parameters:

  • weights (ndarray | str | Path | Callable[[int], ndarray]) – An array of weights, a path to a file with weights compatible with np.loadtxt, or a callable that accepts n_in and returns an ndarray of shape (n_in, n_out).

  • axis (str | None) – The name of the axis to apply the transformation to. Defaults to the leading (0th) axis in the array.

  • right_multiply (bool) – Set False to transpose the weights before applying.

  • channel_clusters (list[list[int]] | None) – Optional explicit channel cluster specification. See AffineTransformSettings.channel_clusters.

  • min_cluster_size (int) – Minimum channels per cluster; smaller clusters are merged. See AffineTransformSettings.min_cluster_size.

Returns:

AffineTransformTransformer.

Return type:

AffineTransformTransformer

common_rereference(mode='mean', axis=None, include_current=True, channel_clusters=None)[source]#

Perform common average referencing (CAR) on streaming data.

Parameters:

  • mode (str) – The statistical mode to apply – either “mean” or “median”

  • axis (str | None) – The name of the axis to apply the transformation to.

  • include_current (bool) – Set False to exclude each channel from participating in the calculation of its reference.

  • channel_clusters (list[list[int]] | None) – Optional channel clusters for per-cluster rereferencing. See CommonRereferenceSettings.channel_clusters.

Returns:

CommonRereferenceTransformer

Return type:

CommonRereferenceTransformer

Classes

class AffineTransform(*args, settings=None, **kwargs)[source]#

Bases: BaseTransformerUnit[AffineTransformSettings, AxisArray, AxisArray, AffineTransformTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of AffineTransformSettings

class AffineTransformSettings(weights, axis=None, right_multiply=True, channel_clusters=None, min_cluster_size=32)[source]#

Bases: Settings

Settings for AffineTransform.

Parameters:
weights: ndarray | str | Path | Callable[[int], ndarray]#

An array of weights, a path to a file with weights compatible with np.loadtxt, or a callable that accepts n_in: int and returns an ndarray of shape (n_in, n_out).

axis: str | None = None#

The name of the axis to apply the transformation to. Defaults to the leading (0th) axis in the array.

right_multiply: bool = True#

Set False to transpose the weights before applying.

channel_clusters: list[list[int]] | None = None#

Optional explicit input channel cluster specification for block-diagonal optimization.

Each element is a list of input channel indices forming one cluster. The corresponding output indices are derived automatically from the non-zero columns of the weight matrix for those input rows.

When provided, the weight matrix is decomposed into per-cluster sub-matrices and multiplied separately, which is faster when cross-cluster weights are zero.

If None, block-diagonal structure is auto-detected from the zero pattern of the weights.

min_cluster_size: int = 32#

Minimum number of channels per cluster for the block-diagonal optimization. Clusters smaller than this are greedily merged together to avoid excessive Python loop overhead. Set to 1 to disable merging.

__init__(weights, axis=None, right_multiply=True, channel_clusters=None, min_cluster_size=32)#
Parameters:
Return type:

None

class AffineTransformState[source]#

Bases: object

weights: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#
new_axis: AxisBase | None = None#
n_out: int = 0#
clusters: list | None = None#

list of (in_indices_xp, out_indices_xp, sub_weights_xp) tuples when block-diagonal.

class AffineTransformTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[AffineTransformSettings, AxisArray, AxisArray, AffineTransformState]

Apply affine transformation via matrix multiplication: y = Ax or y = Ax + B.

Use this transformer when you need full matrix transformations that mix channels (off-diagonal weights), such as spatial filters or projections.

For simple per-channel scaling and offset where each output channel depends only on its corresponding input channel (diagonal weight matrix), use LinearTransformTransformer instead, which is more efficient.

The weights matrix can include an offset row (stacked as [A|B]) where the input is automatically augmented with a column of ones to compute y = Ax + B.

set_weights(weights, *, recalc_clusters=False)[source]#

Replace weight values, optionally recalculating cluster decomposition.

weights must be in canonical orientation (right_multiply already applied by the caller or by _reset_state). The array may live in any Array-API namespace (NumPy, CuPy, etc.).

Parameters:

  • weights – Weight matrix in canonical orientation.

  • recalc_clusters – When True, re-run block-diagonal cluster detection and store the new decomposition. When False (default), reuse the existing cluster structure and only update weight values.

Return type:

None

class CommonRereference(*args, settings=None, **kwargs)[source]#

Bases: BaseTransformerUnit[CommonRereferenceSettings, AxisArray, AxisArray, CommonRereferenceTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of CommonRereferenceSettings

class CommonRereferenceSettings(mode='mean', axis=None, include_current=True, channel_clusters=None)[source]#

Bases: Settings

Settings for CommonRereference

Parameters:
mode: str = 'mean'#

The statistical mode to apply – either “mean” or “median”.

axis: str | None = None#

The name of the axis to apply the transformation to.

include_current: bool = True#

Set False to exclude each channel from participating in the calculation of its reference.

__init__(mode='mean', axis=None, include_current=True, channel_clusters=None)#
Parameters:
Return type:

None

channel_clusters: list[list[int]] | None = None#

Optional channel clusters for per-cluster rereferencing. Each element is a list of channel indices forming one cluster. The common reference is computed independently within each cluster. If None, all channels form a single cluster.

class CommonRereferenceState[source]#

Bases: object

clusters: list | None = None#

list of xp arrays of channel indices, one per cluster.

class CommonRereferenceTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[CommonRereferenceSettings, AxisArray, AxisArray, CommonRereferenceState]