ezmsg.sigproc.util.channels#

One way to say “split these channels into groups”.

Several sources attach a structured CoordinateAxis to the channel dimension carrying per-channel fields — e.g. ezmsg-blackrock’s ChannelMap emits a ch axis whose .data is a numpy struct array with x/y/size/label/array/bank/elec/headstage. Operations that treat channels in groups — per-bank rereferencing, block spatial filters — need to turn that metadata into index groups.

ChannelGroupSpec is the single spec type every such operation accepts, and resolve_channel_groups() is the single resolver:

spec

meaning

None

no grouping (caller’s default applies)

"bank"

group by that struct-array field

("array", "bank")

group by the tuple of those fields

[[0, 1, 2], [3, 4, 5]]

explicit index groups

fn(message, axis) -> groups

anything else

Groups are returned in first-appearance order along the channel axis so a resolved grouping is reproducible and readable against the channel table.

This module also owns the two ways a transformer can fold a channel axis into its per-message state hash: group_spec_fingerprint() (O(1), notices only that the metadata field appeared or vanished) and coord_value_fingerprint() (O(bytes), notices the values changing). Their docstrings explain which failure mode each one is for.

Module Attributes

ChannelGroupSpec

How to split a channel axis into groups.

Functions

array_value_fingerprint(arr)[source]#

Content digest of one array: (dtype, shape, checksum).

The shared primitive under coord_value_fingerprint(), also used directly by consumers that already hold the array (e.g. ConcatProcessor, fingerprinting each axis it caches).

zlib.crc32 rather than hash(arr.tobytes()) because the bottleneck is the hash, not the copy. Measured on a 256-channel ChannelMap axis (27.6 kB, Apple M-series): the tobytes() copy is 0.30 µs (94 GB/s) while CPython’s siphash over the result is 4.7 µs (5.5 GB/s); crc32 reads the array’s buffer directly at 29 GB/s for 0.94 µs total – 5.3x cheaper.

The tradeoff is a 32-bit checksum, so a collision means a missed state reset. dtype and shape ride along both because they are nearly free and because they carry most of the structural change a checksum could alias.

The dtype goes in as the np.dtype object, not str(dtype): numpy builds a structured dtype’s repr field by field, which costs 9.8 µs for the eight-field ChannelMap above – ten times the checksum it was annotating. The object is hashable and compares by value, so it does the same job for 0.02 µs.

crc32 needs a C-contiguous buffer, which a struct-array field view never is, so the gather is explicit here rather than left to fail.

Parameters:

arr (ndarray)

Return type:

tuple

channel_groups_from_field(message, axis=None, field='bank')[source]#

Group channel indices by one or more fields of a structured coordinate axis.

Parameters:

  • message (AxisArray) – Message whose axis coordinate is a structured CoordinateAxis (its .data is a structured numpy array).

  • axis (str | None) – Channel axis name. None defaults to the last dimension.

  • field (str | Sequence[str]) – Struct-array field to group by (e.g. "bank"), or a sequence of fields to group by their tuple (e.g. ("array", "bank")).

Returns:

Index groups, one per distinct value, in first-appearance order. None when the axis carries no usable structured field (no such axis, no .data, unstructured .data, a field is absent, or the per-channel length doesn’t match the data). Returning None rather than a single all-channel group lets callers distinguish “no metadata, fall back to my default” from “one bank”.

Return type:

list[ndarray] | None

coord_value_fingerprint(message, axis, fields=None)[source]#

Digest of the coordinate values on axis, restricted to fields.

The value-sensitive counterpart to group_spec_fingerprint(), for transformers that cache indices resolved against coordinate values (labels, regex matches, field matches). Folding this into _hash_message makes such a cache re-resolve when a source renames, reorders or swaps out channels without changing its key or channel count.

Parameters:

  • message (AxisArray) – The message whose axis is being fingerprinted.

  • axis (str | None) – Coordinate axis name. None defaults to the last dimension.

  • fields (Sequence[str] | None) – Struct-array fields the consumer actually matches against. None digests the whole coordinate array, which is what an unstructured (plain label) axis needs. A named field absent from the dtype contributes None, so gaining or losing it still registers.

Returns:

A hashable tuple, empty when the axis carries no coordinate data.

Return type:

tuple

Restricting to fields is about invalidation correctness, not speed: a source that recomputes float x/y positions each message would otherwise reset the state continuously, even for a selection that only ever reads label. It is sometimes also cheaper and sometimes not. Measured on a 256-channel ChannelMap (eight fields, 108 B itemsize, 27.6 kB total):

A wide field loses because extracting it is a strided gather (7-20 GB/s) while the whole axis is one contiguous read (29 GB/s). Fields are digested one at a time for the same reason numpy makes multi-field indexing a trap: arr[['array', 'bank']] returns a view that keeps the original itemsize, so its tobytes() is the entire 27.6 kB – asking for two of eight fields would otherwise cost more than asking for all of them.

fingerprint_stats()[source]#

{label: (calls, digests_computed, mapping_hits)} for this process.

Empty unless EZMSG_SIGPROC_FINGERPRINT_STATS=1. Counters are per-process, so a multi-process graph has to collect them from each worker.

digests_computed is the number that matters – how often the memo failed to save the O(bytes) work. mapping_hits separates the two shortcuts: the whole-axes-mapping check, which only fires when an upstream node passed the mapping through untouched, from the per-array check that does the real work (most nodes rebuild the mapping via replace(..., axes={...}) even when the axis objects inside it are unchanged).

Return type:

dict[str, tuple[int, int, int]]

group_spec_fields(spec)[source]#

The metadata field names spec groups by, or None if it needs none.

Explicit index groups, callables and None all return None: nothing about them can change with the message.

Deliberately O(1) — a field spec is discriminated by its first element, never by scanning all of them, because this runs on the per-message hash path. resolve_channel_groups() does the full homogeneity check once, at state reset, so a malformed spec still fails loudly.

Parameters:

spec (str | Sequence[str] | Sequence[Sequence[int]] | Callable[[AxisArray, str], Sequence[Sequence[int]] | None] | None)

Return type:

tuple[str, …] | None

group_spec_fingerprint(message, axis, spec)[source]#

O(1) summary of whether spec can resolve against this message.

Transformers fold this into their per-message state hash so a stream that gains or loses the grouping field re-resolves its groups, without paying to hash the field’s bytes on every message. Field values changing under a fixed key and channel count is deliberately not detected — a genuine remap arrives with a new key or channel count.

That concession is safe for a grouping (a stale grouping is arithmetic on the wrong partition, which a changed key or channel count would have caught) but not for every consumer of channel metadata. An operation whose cached state is a set of resolved indices – where a stale answer emits one channel’s samples under another channel’s label – wants coord_value_fingerprint() instead, which costs O(bytes) but actually tracks the values. The two are a deliberate pair; pick by whether a silent stale answer is recoverable downstream.

The two common specs – None and a single field name – are classified inline rather than through group_spec_fields(), because at this call rate the function call itself is a measurable share of the cost. Both shortcuts must agree with that function; everything less trivial defers to it.

Parameters:
Return type:

tuple

reset_fingerprint_stats()[source]#

Zero every counter in this process.

Return type:

None

resolve_channel_groups(message, axis, spec)[source]#

Turn a ChannelGroupSpec into validated index groups.

Returns None when spec is None or when a metadata-derived spec finds nothing to group by — in both cases the caller applies its own default (typically “all channels in one group”).

Parameters:
Return type:

list[ndarray] | None

validate_channel_groups(groups, n_channels)[source]#

Raise ValueError unless groups are in-range and pairwise disjoint.

Disjointness matters because every consumer of a grouping assumes a channel belongs to at most one group — a channel listed twice would be rereferenced twice, or would have two weight blocks written to the same output. Not every channel has to appear: omitted channels are the caller’s business (rereferencing passes them through unchanged).

An empty groups list, or empty groups within it, validates trivially.

Parameters:
Return type:

None

Classes

class AxisFingerprintMemo(label=None)[source]#

Bases: object

Per-consumer, identity-first fingerprints of a message’s coordinate axes.

A transformer that caches anything derived from axis values – resolved indices, output labels – has to notice when those values change under a fixed key and shape, and the honest check is O(bytes). This makes it O(1) in the case that actually occurs.

replace() carries axes, the axis objects and their .data arrays by reference, so a message threaded through a chain of transformers presents the same objects every time. Two is checks – first the whole axes mapping, then each array – settle it without touching the bytes. A miss just computes the digest, so this is a pure fast path: it can make the check cheaper, never wrong.

Measured across a 20-node graph checking one 256-channel ChannelMap axis: 99.7 µs to digest per node per message, 0.6 µs with this. After a cross-process hop every object is fresh, so it degrades to ~20 µs – see benchmarks/benchmark_axis_fingerprint.py.

The contract this assumes: a coordinate array is never mutated in place. Messages fan out to multiple graph branches, so mutating one is already unsafe; this turns that into a requirement.

One memo belongs to one consumer, which must pass the same names and exclude on every call – the whole-mapping shortcut caches a single answer per axes object and cannot tell that the question changed.

Parameters:

label (str | None)

__init__(label=None)[source]#
Parameters:

label (str | None)

Return type:

None

fingerprint(message, names=None, exclude=())[source]#

Digest the coordinate axes’ values, as ((name, digest), ...).

Parameters:

  • message (AxisArray) – Message whose axes are being fingerprinted.

  • names (Sequence[str] | None) – Restrict to these axes. None covers every coordinate axis, which is the safe default when the consumer’s own axis selection is not known until state reset.

  • exclude (Sequence[str]) – Axes to skip even when names is None – for an axis deliberately read live rather than cached.

Returns:

A hashable tuple; empty when no axis carries coordinate data. Axes without .data contribute nothing, since a LinearAxis compares by value for free.

Return type:

tuple