ezmsg.sigproc.aggregate#

Aggregation operations over arrays.

Note

AggregateTransformer and RangedAggregateTransformer support the Array API standard, enabling use with NumPy, CuPy, PyTorch, and other compatible array libraries. Operations not available on a given backend (nan-variants, trapezoid) fall back to NumPy automatically.

Functions

aggregate_slices(data, slices, axis_idx, operation, *, coordinates=None, index_to_coordinate=True)[source]#

Apply operation to each slice of data along axis_idx, stacked.

Where the groups come from is the caller’s business – coordinate bands resolved once, or bin boundaries recomputed per chunk – but running the aggregation is the same either way, and three operations need more than f(segment, axis) to do it correctly:

  • the nan-variants, which the Array API does not define, so they need a numpy fallback and a conversion back to the caller’s namespace;

  • TRAPEZOID, which integrates and so needs the axis’s x-coordinates, or it silently returns an integral in units of samples rather than of the axis;

  • ARGMIN/ARGMAX, which return a position within the slice. An index is rarely what anyone wants – “the peak is at 10.5 Hz” is useful, “the peak is at offset 7 of this band” is not – so it is converted back to the axis coordinate here.

The result has the same rank as data, with axis_idx reduced to one entry per slice, so a caller can drop it into the message it came from having replaced only that axis.

Parameters:

  • data – The array to slice. Any Array API namespace.

  • slices (Sequence[slice]) – One slice per output group, in output order. Slices index data along axis_idx; they need not be contiguous, and an empty sequence yields a zero-length result.

  • axis_idx (int) – Index of the axis being grouped.

  • operation (AggregationFunction) – The AggregationFunction to apply within each slice.

  • coordinates (NDArray | None) – The axis’s coordinate values, one per element of data along axis_idx. Required when needs_coordinates() is True for operation, ignored otherwise.

  • index_to_coordinate (bool) – Whether ARGMIN/ARGMAX results are converted from a within-slice index to an axis coordinate. False leaves the raw index and needs no coordinates. Every transformer in this package leaves this True; it exists for a caller that means to index back into the array it passed in.

Raises:

ValueError – if operation needs coordinates and none were given, or if coordinates does not span data along axis_idx. Either would otherwise produce a plausible-looking but wrong number.

axis_coordinates(message, axis_name)[source]#

The coordinate value of every element along axis_name.

Reads a coordinate axis’s values directly; evaluates a linear axis over its own length. Callers whose data does not line up with the message – anything carrying samples across message boundaries – must build their own vector instead.

Parameters:
Return type:

NDArray

needs_coordinates(operation)[source]#

Whether operation requires the axis’s x-coordinates.

Lets a caller skip building a coordinate vector it will not use, which is worth doing where that vector would have to be constructed per chunk. Accepts a single function or an iterable of them.

Parameters:

operation (AggregationFunction | Iterable[AggregationFunction])

Return type:

bool

ranged_aggregate(axis=None, bands=None, operation=AggregationFunction.MEAN)[source]#

Apply an aggregation operation over one or more bands.

Parameters:

  • axis (str | None) – The name of the axis along which to apply the bands.

  • bands (list[tuple[float, float]] | None) – [(band1_min, band1_max), (band2_min, band2_max), …] If not set then this acts as a passthrough node.

  • operation (AggregationFunction) – AggregationFunction to apply to each band.

Returns:

RangedAggregateTransformer

Return type:

RangedAggregateTransformer

Classes

class AggregateSettings(axis, operation=AggregationFunction.MEAN)[source]#

Bases: Settings

Settings for Aggregate.

Parameters:
axis: str#

The name of the axis to aggregate over. This axis will be removed from the output.

__init__(axis, operation=AggregationFunction.MEAN)#
Parameters:
Return type:

None

operation: AggregationFunction = 'mean'#

AggregationFunction to apply.

ARGMIN/ARGMAX return the coordinate of the extremum along axis – “the peak is at 14 Hz” – not its index, matching RangedAggregateTransformer and BinnedAggregateTransformer. An index would be unusable here anyway, since axis is removed from the output. Where axis carries no axis metadata the coordinates are 0, 1, 2, …, so the result is the index after all.

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

Bases: BaseTransformer[AggregateSettings, AxisArray, AxisArray]

Transformer that aggregates an entire axis using a specified operation.

Unlike RangedAggregateTransformer which aggregates over specific ranges/bands and preserves the axis (with one value per band), this transformer aggregates the entire axis and removes it from the output, reducing dimensionality by one.

Parameters:

settings (SettingsType)

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

Bases: BaseTransformerUnit[AggregateSettings, AxisArray, AxisArray, AggregateTransformer]

Unit that aggregates an entire axis using a specified operation.

Parameters:

settings (Settings | None)

SETTINGS#

alias of AggregateSettings

class AggregationFunction(*values)[source]#

Bases: OptionsEnum

Enum for aggregation functions available to be used in ranged_aggregate operation.

NONE = 'None (all)'#
MAX = 'max'#
MIN = 'min'#
MEAN = 'mean'#
MEDIAN = 'median'#
STD = 'std'#
SUM = 'sum'#
NANMAX = 'nanmax'#
NANMIN = 'nanmin'#
NANMEAN = 'nanmean'#
NANMEDIAN = 'nanmedian'#
NANSTD = 'nanstd'#
NANSUM = 'nansum'#
ARGMIN = 'argmin'#
ARGMAX = 'argmax'#
TRAPEZOID = 'trapezoid'#
class RangedAggregate(*args, settings=None, **kwargs)[source]#

Bases: BaseTransformerUnit[RangedAggregateSettings, AxisArray, AxisArray, RangedAggregateTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of RangedAggregateSettings

class RangedAggregateSettings(axis=None, bands=None, operation=AggregationFunction.MEAN)[source]#

Bases: Settings

Settings for RangedAggregate.

Parameters:
axis: str | None = None#

The name of the axis along which to apply the bands.

bands: list[tuple[float, float]] | None = None#

[(band1_min, band1_max), (band2_min, band2_max), …] If not set then this acts as a passthrough node.

operation: AggregationFunction = 'mean'#

AggregationFunction to apply to each band.

__init__(axis=None, bands=None, operation=AggregationFunction.MEAN)#
Parameters:
Return type:

None

class RangedAggregateState[source]#

Bases: object

slices: list[tuple[Any, ...]] | None = None#
out_axis: AxisBase | None = None#
ax_vec: NDArray | None = None#
class RangedAggregateTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[RangedAggregateSettings, AxisArray, AxisArray, RangedAggregateState]