ezmsg.sigproc.flatten#

Flatten non-time dimensions of an AxisArray into a single axis.

The most common use is collapsing a (time, ch, feature) stream into (time, ch_x_feature) for transports that only support 2-D data (e.g. LSL).

The merged axis is always emitted as a numpy structured CoordinateAxis whose rows are the cartesian product of the source-axis entries. Per cartesian combination, the merge rule is dict-union with later-wins on conflicts:

  • Source axes with a structured CoordinateAxis contribute all their fields to the merged row (so an upstream ch axis carrying (x, y, label, bank, elec, device) propagates every field through).

  • Source axes with a simple CoordinateAxis contribute one virtual field named after the axis (e.g. "feature") whose value is the axis label at that position.

  • Source axes without a CoordinateAxis contribute one virtual uint32 field with a 1-based position index.

A canonical "label" field is always present on the output struct. Its value is the cartesian-product join of each source axis’s primary label (the axis’s own "label" field if struct, else the axis value or position index), separated by FlattenSettings.label_separator (default "/"). This "label" overwrites any inherited label field from the source axes — the original per-axis labels are still available via the named fields (ch, feature, etc.).

Functions

axis_labels(axis_data)[source]#

Normalize a CoordinateAxis .data array to a list of labels.

Return type:

list

normalize_axis_label(label)[source]#

Return a hashable string-or-tuple representation of a coord label.

Handles structured-dtype entries (with a "label" field, or any other named fields) and numpy scalar types. Public so other ezmsg packages — notably ezmsg.learn.process.flatten — can share the same convention.

Classes

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

Bases: BaseTransformerUnit[FlattenSettings, AxisArray, AxisArray, FlattenTransformer]

Parameters:

settings (Settings | None)

SETTINGS#

alias of FlattenSettings

class FlattenSettings(preserve_axis=None, sample_axis=None, flatten_axes=None, output_axis='ch', label_separator='/')[source]#

Bases: Settings

Settings for Flatten.

Parameters:

  • preserve_axis (str | None)

  • sample_axis (str | None)

  • flatten_axes (tuple[str, ...] | None)

  • output_axis (str)

  • label_separator (str)

preserve_axis: str | None = None#

Axis kept as the leading dim of the output (typically "time"). Defaults to message.dims[0].

sample_axis: str | None = None#

Optional rename for preserve_axis on the output (e.g. set to "time" when the input’s preserved dim is named "win"). Defaults to preserve_axis (no rename).

flatten_axes: tuple[str, ...] | None = None#

Tuple of axes to collapse, in fastest-varying-last order. Defaults to all non-preserve dims, in input order, so a (time, ch, feature) input flattens with ch slow / feature fast — matching numpy’s C-order reshape.

output_axis: str = 'ch'#

Name of the merged axis on the output.

label_separator: str = '/'#

Separator for cartesian-product labels in the canonical "label" field of the output struct. Default "/". Set to "" to concatenate without a delimiter.

__init__(preserve_axis=None, sample_axis=None, flatten_axes=None, output_axis='ch', label_separator='/')#
Parameters:

  • preserve_axis (str | None)

  • sample_axis (str | None)

  • flatten_axes (tuple[str, ...] | None)

  • output_axis (str)

  • label_separator (str)

Return type:

None

class FlattenState[source]#

Bases: object

hash: int = -1#
preserve_axis: str = ''#
sample_axis: str = ''#
output_axis: str = ''#
flatten_axes: tuple[str, ...] = ()#
rest_axes: tuple[str, ...] = ()#
perm: tuple[int, ...] | None = None#
target_shape: tuple[int, ...] = ()#
output_axis_obj: CoordinateAxis | None = None#
output_dims: tuple[str, ...] = ()#
class FlattenTransformer(*args, **kwargs)[source]#

Bases: BaseStatefulTransformer[FlattenSettings, AxisArray, AxisArray, FlattenState]