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.

Both transformers here take a ChannelGroupSpec as channel_groups: explicit index groups, or the name of a channel-metadata field ("bank", "array", …) to group by. For CommonRereference the groups say which channels share a reference. For AffineTransform they only shape the construction of deterministic weights — the block structure the matmul exploits is always read off the weight matrix itself.

Module Attributes

KERNELS

Valid values for AffineTransformSettings.kernel.

Functions

affine_transform(weights, axis=None, right_multiply=True, channel_groups=None, kernel='auto')[source]#

Perform affine transformations on streaming data.

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), or a callable that accepts n_in and returns an ndarray of shape (n_in, n_out). See AffineTransformSettings.weights; for streaming CAR prefer common_rereference().
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_groups (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | None) – Channel grouping used to build kind- or callable-based weights. See AffineTransformSettings.channel_groups.
kernel (str) – Matmul kernel selection. See AffineTransformSettings.kernel.

Returns:

AffineTransformTransformer.

Return type:

AffineTransformTransformer

common_rereference(mode='mean', axis=None, include_current=True, channel_groups=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_groups (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | None) – Which channels share a reference – explicit index groups or a channel-metadata field name (e.g. "bank"). See CommonRereferenceSettings.channel_groups.

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_groups=None, kernel='auto')[source]#

Bases: Settings

Settings for AffineTransform.

Parameters:

weights (ndarray | str | Path | RereferenceKind | Callable[[int], ndarray])
axis (str | None)
right_multiply (bool)
channel_groups (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | None)
kernel (str)

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_groups; or a callable accepting n_in: int (optionally also the resolved groups) and returning 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 group size) pass a callable, e.g. lambda n, groups: car_matrix(n, groups=groups, 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.

How to group input channels when building the weight matrix.

Applies only when weights is a RereferenceKind or a callable — e.g. weights="car", channel_groups="bank" builds a per-bank common average reference from the channel axis metadata. See ChannelGroupSpec for the accepted forms.

It has no effect when weights is an explicit array or file: the block structure exploited by the matmul is always derived from the weight matrix itself, so a grouping that disagreed with the weights could never change the result (ezmsg-org/ezmsg-sigproc#198).

kernel: str = 'auto'#

"auto" (default) picks between a dense matmul and a block-diagonal one from the structure of the weights and the message size; "dense" and "blocks" force the choice. See ezmsg.sigproc.util.blockdiag for the cost model behind "auto".

Type:: Matmul kernel

__init__(weights, axis=None, right_multiply=True, channel_groups=None, kernel='auto')#

Parameters:

weights (ndarray | str | Path | RereferenceKind | Callable[[int], ndarray])
axis (str | None)
right_multiply (bool)
channel_groups (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | None)
kernel (str)

Return type:

None

class AffineTransformState[source]#

Bases: object

weights: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#: Full weight matrix for the dense kernel; None when blocks are in use.

blocks: list | None = None#: list of (in_slice, out_slice, sub_weights) for the block-diagonal kernel.

in_perm: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#: Channel gather that makes the blocks contiguous, or None if they already are.

out_perm: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#: Output-channel counterpart of in_perm, used to slice the weights.

out_inv_perm: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#: Gather that undoes out_perm on the result.

out_dtype: DTypeLike | None = None#: Result dtype of a block matmul, resolved once against the message dtype.

fill_in_place: bool = False#: Whether the backend accepts matmul(..., out=); else blocks are concatenated.

device: object = None#: Device the output buffer is allocated on, resolved once at reset.

new_axis: AxisBase | None = None#

n_out: int = 0#

n_in: int = 0#: Channels expected on the message; blocks require it to equal weights.shape[0].

n_samples: int = 1#: Representative samples per message, for the kernel cost model.

dispatched: bool = False#

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_structure=False)[source]#

Replace weight values, optionally re-deriving the matmul kernel.

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_structure (bool) – When True, re-derive the block-diagonal structure from weights and re-choose the kernel. When False (default), keep the existing block layout and only refresh the values – appropriate for an adaptive filter whose sparsity pattern is fixed. Note that re-deriving reads the whole matrix on the host, so avoid it on a hot path with device-resident weights.

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

Bases: Settings

Settings for CommonRereference

Parameters:

mode (str)
axis (str | None)
include_current (bool)
channel_groups (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | 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.

__init__(mode='mean', axis=None, include_current=True, channel_groups=None)#

Parameters:

mode (str)
axis (str | None)
include_current (bool)
channel_groups (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | None)

Return type:

None

include_current: bool = True#: Set False to exclude each channel from participating in the calculation of its reference.

Which channels share a reference. The common reference is computed independently within each group; channels in no group pass through unchanged.

None (default) references every channel against one common average. Pass explicit index groups, or the name of a channel-metadata field to group by – channel_groups="bank" rereferences within each electrode bank. See ChannelGroupSpec.

A field-derived grouping is resolved when the stream gains or loses that field, not when its values change: the channel-to-field map is expected to be static for a given stream key and channel count.

class CommonRereferenceState[source]#

Bases: object

single: bool = False#: Whether one reference covers every channel – the no-allocation fast path.

passthrough: bool = False#: Leave-one-out with nothing to reference against; emit the input unchanged.

project: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#: (n_ch, n_groups) group-mean projector; None on the single path.

spread: ndarray[tuple[Any, ...], dtype[_ScalarT]] | None = None#: (n_groups, n_ch) indicator that broadcasts each group’s reference back.

scale: ndarray[tuple[Any, ...], dtype[_ScalarT]] | float = 1.0#: Per-channel N/(N-1) leave-one-out gain, or 1.0 when it is uniform.

groups: list | None = None#: Per-group index arrays; used only by the median path.

out_dtype: DTypeLike | None = None#

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

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

Subtract a common reference, computed over all channels or within groups.

mode="mean" is expressed as two skinny matmuls rather than a per-group gather/scatter loop, so cost is independent of whether a group’s channels are contiguous and there is no Python loop per message.

Channels belonging to no group pass through unchanged, matching car_matrix(), which leaves them identity.

Floating-point input keeps its dtype; integer input promotes to float. (An earlier version promoted float32 to float64, doubling the bandwidth of every downstream stage.)