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:
Returns:

AffineTransformTransformer.

Return type:

AffineTransformTransformer

common_rereference(mode='mean', axis=None, include_current=True, channel_clusters=None, cluster_by_field=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.

  • cluster_by_field (str | None) – Optionally derive clusters from a structured channel-axis field (e.g. "bank"). See CommonRereferenceSettings.cluster_by_field.

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 | RereferenceKind | Callable[[int], ndarray]#

An array of weights; a path to a file with weights compatible with np.loadtxt; a RereferenceKind or its string value (e.g. "car") to build a deterministic rereference matrix over channel_clusters; or a callable that accepts n_in: int and returns an ndarray of shape (n_in, n_out).

Note: if you simply want streaming CAR, CommonRereference in this module is usually the better choice (per-sample mean subtraction instead of a matmul, plus median support). Kind-based weights are useful for discovering the available deterministic transforms and for workflows that start from such a matrix and later replace it externally via AffineTransformTransformer.set_weights(). For variants (leave-one-out, minimum cluster size) pass a callable, e.g. lambda n: car_matrix(n, clusters=..., include_current=False).

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, cluster_by_field=None)[source]#

Bases: Settings

Settings for CommonRereference

Parameters:

  • mode (str)

  • axis (str | None)

  • include_current (bool)

  • channel_clusters (list[list[int]] | None)

  • cluster_by_field (str | None)

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.

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.

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

  • mode (str)

  • axis (str | None)

  • include_current (bool)

  • channel_clusters (list[list[int]] | None)

  • cluster_by_field (str | None)

Return type:

None

cluster_by_field: str | None = None#

Derive channel_clusters automatically from a structured field of the channel coordinate axis (e.g. "bank" to rereference within each electrode bank). Used only when channel_clusters is None and the axis actually carries that field; otherwise falls back to a single all-channel cluster. Explicit channel_clusters always takes precedence. Note: the any changes to the field values are not detected. The channel->field map is expected to be static for the stream’s life.

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]