diff --git a/pyproject.toml b/pyproject.toml index c42ab9a..d141149 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ readme = "README.md" requires-python = ">=3.10" dynamic = ["version"] dependencies = [ - "ezmsg[axisarray]>=3.10.0b2", + "ezmsg[axisarray]>=3.10.0b3", "typing-extensions>=4.0.0", ] diff --git a/src/ezmsg/baseproc/__init__.py b/src/ezmsg/baseproc/__init__.py index 7e654a4..ce54559 100644 --- a/src/ezmsg/baseproc/__init__.py +++ b/src/ezmsg/baseproc/__init__.py @@ -99,13 +99,6 @@ get_base_transformer_type, ) from .util.asio import CoroutineExecutionError, SyncToAsyncGeneratorWrapper, run_coroutine_sync -from .util.chunkdim import ( - STREAMING_DIMS, - resolve_chunk_dim, - resolve_configured_chunk_dim, - resolve_feature_dim, - resolve_transform_dim, -) from .util.deprecation import ( DEFAULT_REMOVAL, suppress_axis_deprecation, @@ -128,6 +121,13 @@ sanitize_settings_value, ) from .util.profile import profile_method, profile_subpub +from .util.streamdim import ( + STREAMING_DIMS, + resolve_configured_stream_dim, + resolve_feature_dim, + resolve_stream_dim, + resolve_transform_dim, +) from .util.typeresolution import check_message_type_compatibility, resolve_typevar __all__ = [ @@ -202,9 +202,9 @@ "SampleMessage", "SampleTriggerMessage", "is_sample_message", - "resolve_chunk_dim", - "resolve_configured_chunk_dim", + "resolve_configured_stream_dim", "resolve_feature_dim", + "resolve_stream_dim", "resolve_transform_dim", "suppress_axis_deprecation", "warn_axis_deprecated", diff --git a/src/ezmsg/baseproc/stateful.py b/src/ezmsg/baseproc/stateful.py index bec64b7..925dd60 100644 --- a/src/ezmsg/baseproc/stateful.py +++ b/src/ezmsg/baseproc/stateful.py @@ -14,8 +14,8 @@ ) from .protocols import MessageInType, MessageOutType, SettingsType, StateType from .util.asio import run_coroutine_sync -from .util.chunkdim import STREAMING_DIMS as _STREAMING_DIMS from .util.message import is_sample_message +from .util.streamdim import STREAMING_DIMS as _STREAMING_DIMS from .util.typeresolution import resolve_typevar @@ -32,7 +32,7 @@ def _shape_slice(dims: list[str], exclude: tuple[str, ...]) -> slice | None: """A slice selecting the dimensions whose *length* feeds the hash. Only worth having when the excluded dimensions sit at one end, which is the - case for every layout in practice -- the chunk dimension leads (``time, ch``; + case for every layout in practice -- the stream dimension leads (``time, ch``; ``win, time, ch``) or, after a transpose, trails. Anything else returns ``None`` and simply declines the fast path rather than paying a comprehension per message to reproduce ``shape``. @@ -92,16 +92,16 @@ def _build_witness( # a process boundary, where unpickling hands out a new object per message but # the fingerprint rides along already computed. kept = tuple((dim, axes.get(dim), _axis_value(axes.get(dim))) for dim in dims if dim not in exclude) - # The chunk axis is a new object every message on any path -- its offset + # The stream axis is a new object every message on any path -- its offset # advances -- so it is compared by value always. - chunked = tuple((dim, getattr(axes.get(dim), "gain", None)) for dim in dims if dim in exclude) - w_dims, w_key, w_chunk = list(dims), message.key, message.chunk_dim + streamed = tuple((dim, getattr(axes.get(dim), "gain", None)) for dim in dims if dim in exclude) + w_dims, w_key, w_stream = list(dims), message.key, message.stream_dim - if len(kept) == 1 and len(chunked) == 1 and chunked[0][1] is not None and kept[0][2] is not None: - # One coordinate axis to pin down and one chunk axis carrying the sample + if len(kept) == 1 and len(streamed) == 1 and streamed[0][1] is not None and kept[0][2] is not None: + # One coordinate axis to pin down and one stream axis carrying the sample # rate. This is `(time, ch)`, and `(win, time, ch)` once `time` is also # excluded -- between them, nearly every message in a graph. - (kept_dim, kept_axis, kept_value), (chunk_dim, chunk_gain) = kept[0], chunked[0] + (kept_dim, kept_axis, kept_value), (stream_dim, stream_gain) = kept[0], streamed[0] kept_ix = dims.index(kept_dim) def validate( @@ -109,13 +109,13 @@ def validate( _kd: str = kept_dim, _ka: typing.Any = kept_axis, _kv: typing.Any = kept_value, - _cd: str = chunk_dim, - _cg: float = chunk_gain, + _sd: str = stream_dim, + _sg: float = stream_gain, _kix: int = kept_ix, _klen: int = shape[kept_ix], _dims: list[str] = w_dims, _key: str = w_key, - _chunk: str | None = w_chunk, + _stream: str | None = w_stream, _check_key: bool = include_key, ) -> bool: axes = msg.axes @@ -124,14 +124,14 @@ def validate( if axis is not _ka and _axis_value(axis) != _kv: return False return ( - axes[_cd].gain == _cg + axes[_sd].gain == _sg and msg.data.shape[_kix] == _klen - and msg.chunk_dim == _chunk + and msg.stream_dim == _stream and msg.dims == _dims and (not _check_key or msg.key == _key) ) except (AttributeError, KeyError, IndexError): - # The layout shifted out from under the specialisation: the chunk + # The layout shifted out from under the specialisation: the stream # axis stopped being linear (an irregular-rate stream switches to # a CoordinateAxis), a dimension lost its axis, or the data lost a # dimension. Decline and let the full hash sort it out. Costs @@ -143,12 +143,12 @@ def validate( def validate( msg: typing.Any, _kept: tuple = kept, - _chunked: tuple = chunked, + _streamed: tuple = streamed, _sl: slice = sl, _ks: tuple = shape[sl], _dims: list[str] = w_dims, _key: str = w_key, - _chunk: str | None = w_chunk, + _stream: str | None = w_stream, _check_key: bool = include_key, ) -> bool: axes = msg.axes @@ -156,7 +156,7 @@ def validate( incoming = axes.get(dim) if incoming is not axis and (value is None or _axis_value(incoming) != value): return False - for dim, gain in _chunked: + for dim, gain in _streamed: # No `is not None` shortcut on the axis: an excluded dimension # *losing* its axis drops a term from the hash, so absence has to # compare unequal to a gain rather than be skipped. @@ -164,7 +164,7 @@ def validate( return False return ( msg.data.shape[_sl] == _ks - and msg.chunk_dim == _chunk + and msg.stream_dim == _stream and msg.dims == _dims and (not _check_key or msg.key == _key) ) @@ -186,7 +186,7 @@ class Stateful(ABC, typing.Generic[StateType]): Recomputing the hash means walking the dims, reaching into the axes and building a tuple to hash -- and in a steady stream the answer is the same every time. A producer that builds its per-stream axes once and replaces only - the chunk axis per message (the template idiom every ezmsg source uses) hands + the stream axis per message (the template idiom every ezmsg source uses) hands every consumer the *same coordinate axis object* for the life of the stream, so identity is enough to prove the hash cannot have changed. @@ -195,14 +195,14 @@ class Stateful(ABC, typing.Generic[StateType]): """ STREAMING_DIMS: typing.ClassVar[tuple[str, ...]] = _STREAMING_DIMS - """Fallback chunk dimension for messages that do not declare one. + """Fallback stream dimension for messages that do not declare one. - Consulted only when :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim` + Consulted only when :attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim` is ``None``. ``("time",)`` is right for a raw signal and wrong downstream of a windowing stage, where the message is ``(win, time, ch)`` and ``win`` is what grows; such a processor sets ``("win",)``. - Prefer teaching the producer to declare ``chunk_dim``. That puts the answer + Prefer teaching the producer to declare ``stream_dim``. That puts the answer in the one place that knows it, rather than asking each consumer to guess about a message it did not create. """ @@ -231,10 +231,10 @@ def _hash_message(self, message: typing.Any) -> int: """ Check if the message metadata indicates a need for state reset. - For a message that declares :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`, + For a message that declares :attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim`, the default keys on everything describing the stream's *shape and - identity* but not its per-chunk extent: the message key, its dims, the - length of every dimension except the one it is a chunk along, the + identity* but not its per-message extent: the message key, its dims, the + length of every dimension except the one it streams along, the coordinate values on those dimensions, and the gain and offset of any linear axis among them. See :meth:`_message_hash`. @@ -272,13 +272,13 @@ def _message_hash( Folds in, in dimension order: * ``message.key`` (unless *include_key* is False) and ``message.dims`` - * for each dimension other than the chunk dimension: its length, plus + * for each dimension other than the stream dimension: its length, plus either the coordinate axis's :attr:`~ezmsg.util.messages.axisarray.CoordinateAxis.fingerprint` or a linear axis's ``gain`` **and** ``offset`` - * for the chunk dimension: only the ``gain`` + * for the stream dimension: only the ``gain`` - ``offset`` is dropped for the chunk dimension alone, where it simply + ``offset`` is dropped for the stream dimension alone, where it simply counts off elapsed samples. Everywhere else it locates the axis and a change in it is a configuration change: a spectrum whose ``freq`` axis moves from 5-25 Hz to 70-90 Hz keeps the same gain and the same length, @@ -289,7 +289,7 @@ def _message_hash( channels that are no longer there, and the first samples of the new ones come out dominated by the old ones' history. - The chunk dimension is ``message.chunk_dim`` when declared, else + The stream dimension is ``message.stream_dim`` when declared, else :attr:`STREAMING_DIMS`. Naming a dimension the message does not have is harmless -- nothing matches, so nothing is excluded. @@ -297,7 +297,7 @@ def _message_hash( reset-once-then-never behaviour those processors had before. :param exclude_dims: Further dimensions to leave out, *in addition to* - the chunk dimension. Use for a processor whose state genuinely does + the stream dimension. Use for a processor whose state genuinely does not depend on a dimension's identity. :param include_key: Set False for a processor whose state depends only on shape, so that switching streams does not force a reset. @@ -323,13 +323,13 @@ def _message_hash( # The producer renamed the dims and so is the only party that reliably # knows which one grows; fall back to the class default when it is silent. - chunk_dim = message.chunk_dim - if chunk_dim is None: + stream_dim = message.stream_dim + if stream_dim is None: exclude = self.STREAMING_DIMS if exclude_dims is None else (*self.STREAMING_DIMS, *exclude_dims) elif exclude_dims is None: - exclude = (chunk_dim,) + exclude = (stream_dim,) else: - exclude = (chunk_dim, *exclude_dims) + exclude = (stream_dim, *exclude_dims) # Hoisted out of the loop: this runs on every message of every stream, # so the repeated attribute lookups are worth removing. A tuple rather diff --git a/src/ezmsg/baseproc/util/deprecation.py b/src/ezmsg/baseproc/util/deprecation.py index 6cb155b..7c1c059 100644 --- a/src/ezmsg/baseproc/util/deprecation.py +++ b/src/ezmsg/baseproc/util/deprecation.py @@ -8,18 +8,18 @@ the same coordinate, forever. Which dimension that is belongs to the producer, and -:attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim` is where it says so. +:attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim` is where it says so. A setting that lets a consumer disagree can only be used to be wrong, so it is -going away; see :func:`~ezmsg.baseproc.util.chunkdim.resolve_chunk_dim`. +going away; see :func:`~ezmsg.baseproc.util.streamdim.resolve_stream_dim`. During the deprecation window the setting is still honoured, so nothing changes behaviour until it is removed. Two warnings partition the call sites: * This module's construction-time :class:`FutureWarning` fires for *every* use, including a harmless ``axis="time"`` on a raw stream. It means "delete this". -* :func:`~ezmsg.baseproc.util.chunkdim.resolve_configured_chunk_dim`'s runtime +* :func:`~ezmsg.baseproc.util.streamdim.resolve_configured_stream_dim`'s runtime warning fires only when the configured axis disagrees with a *declared* - ``chunk_dim``. It means "deleting this will change what this stage computes". + ``stream_dim``. It means "deleting this will change what this stage computes". To find every remaining call site in a pipeline, run its tests with ``-W error::FutureWarning``. @@ -128,8 +128,8 @@ def warn_axis_deprecated( f"{type(settings).__name__}.{field} is deprecated and will be removed in " f"{package} {removal}. This processor carries state between messages, which " f"is only meaningful along the dimension they accumulate along; that " - f"dimension now comes from AxisArray.chunk_dim. Drop the setting. " - f"If the stream's chunk_dim is wrong, fix it at the producer.", + f"dimension now comes from AxisArray.stream_dim. Drop the setting. " + f"If stream_dim is wrong, fix it at the producer.", FutureWarning, stacklevel=_user_stacklevel(), ) diff --git a/src/ezmsg/baseproc/util/chunkdim.py b/src/ezmsg/baseproc/util/streamdim.py similarity index 80% rename from src/ezmsg/baseproc/util/chunkdim.py rename to src/ezmsg/baseproc/util/streamdim.py index b2f643b..44209a4 100644 --- a/src/ezmsg/baseproc/util/chunkdim.py +++ b/src/ezmsg/baseproc/util/streamdim.py @@ -6,7 +6,7 @@ :meth:`~ezmsg.util.messages.axisarray.AxisArray.transpose` and downstream of any windowing stage, where a ``(time, ch)`` stream becomes ``(win, time, ch)``. -:attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim` is the producer's +:attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim` is the producer's declaration of which dimension messages accumulate along -- the one party that reliably knows. These helpers turn that declaration into the axis a given kind of processor should use, and they live here because @@ -17,10 +17,10 @@ Three rules, because one does not fit every case: -* :func:`resolve_chunk_dim` -- for state carried *between* messages. +* :func:`resolve_stream_dim` -- for state carried *between* messages. * :func:`resolve_feature_dim` -- for a static axis (channels, components). * :func:`resolve_transform_dim` -- for a transform that consumes a regularly - sampled axis, which downstream of a windowing stage is *not* the chunk one. + sampled axis, which downstream of a windowing stage is *not* the stream one. """ import typing @@ -30,18 +30,18 @@ __all__ = [ "STREAMING_DIMS", - "resolve_chunk_dim", - "resolve_configured_chunk_dim", + "resolve_configured_stream_dim", "resolve_feature_dim", + "resolve_stream_dim", "resolve_transform_dim", ] STREAMING_DIMS: tuple[str, ...] = ("time",) -"""Default fallback chunk dimension, matching ``BaseStatefulTransformer``.""" +"""Default fallback stream dimension, matching ``BaseStatefulTransformer``.""" -def resolve_chunk_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = STREAMING_DIMS) -> str: +def resolve_stream_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = STREAMING_DIMS) -> str: """The dimension successive messages accumulate along. This is the axis a processor that carries state *between* messages must @@ -52,7 +52,7 @@ def resolve_chunk_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = to message N+1's head at the same coordinate, forever. The producer renamed the dims and so is the only party that reliably knows - which one grows; ``message.chunk_dim`` is that declaration. When a producer + which one grows; ``message.stream_dim`` is that declaration. When a producer is silent, *streaming_dims* supplies the guess -- ``("time",)`` is right for a raw signal and wrong downstream of a windowing stage, where the message is ``(win, time, ch)`` and ``win`` is what grows. @@ -60,15 +60,15 @@ def resolve_chunk_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = ``dims[0]`` is the last resort only. It is a position, not a meaning, and it breaks under :meth:`~ezmsg.util.messages.axisarray.AxisArray.transpose`. """ - if message.chunk_dim is not None: - return message.chunk_dim + if message.stream_dim is not None: + return message.stream_dim for name in streaming_dims: if name in message.dims: return name return message.dims[0] -def resolve_configured_chunk_dim( +def resolve_configured_stream_dim( processor: typing.Any, message: AxisArray, configured: str | None, @@ -78,26 +78,26 @@ def resolve_configured_chunk_dim( *configured* wins when set -- an explicit axis is an instruction, and removing that escape hatch would break every pipeline that passes the - common ``axis="time"``. But when the producer *declared* a different chunk + common ``axis="time"``. But when the producer *declared* a different stream dimension, that disagreement is worth surfacing exactly once: the processor's cross-message state is about to be carried along an axis whose length is fixed, which is a different operation from the one the caller almost certainly meant. - The warning fires only against a declared ``chunk_dim``, never against the + The warning fires only against a declared ``stream_dim``, never against the :attr:`STREAMING_DIMS` guess -- warning on a guess would fire on every correctly-configured windowed pipeline whose producer is merely silent. :param legacy_default: The dimension this processor's ``axis`` setting used to default to, for the stages whose default was a hardcoded ``"time"`` - rather than a positional guess. Flipping those to follow ``chunk_dim`` - changes results wherever the chunk dimension is not ``"time"`` -- most + rather than a positional guess. Flipping those to follow ``stream_dim`` + changes results wherever the stream dimension is not ``"time"`` -- most obviously downstream of a windowing stage, where it is ``"win"`` -- and unlike an explicitly configured axis there is nothing in the settings to warn about. Passing the old default here surfaces exactly that population, once, and is dropped when the setting is removed. """ - resolved = resolve_chunk_dim(message, getattr(processor, "STREAMING_DIMS", STREAMING_DIMS)) + resolved = resolve_stream_dim(message, getattr(processor, "STREAMING_DIMS", STREAMING_DIMS)) if configured is None: if ( legacy_default is not None @@ -108,40 +108,40 @@ def resolve_configured_chunk_dim( processor._legacy_axis_default_warned = True ez.logger.warning( f"{type(processor).__name__} used to operate on axis={legacy_default!r} by default; it now " - f"follows the stream's chunk_dim={resolved!r}. This changes its output. The old behaviour was " + f"follows the declared stream_dim={resolved!r}. This changes its output. The old behaviour was " f"carrying state across messages along {legacy_default!r}, whose length does not grow, so this " f"is a fix -- but pass axis={legacy_default!r} explicitly to keep the previous behaviour." ) return resolved if ( - message.chunk_dim is not None - and configured != message.chunk_dim + message.stream_dim is not None + and configured != message.stream_dim and configured in message.dims - and not getattr(processor, "_chunk_dim_mismatch_warned", False) + and not getattr(processor, "_stream_dim_mismatch_warned", False) ): - processor._chunk_dim_mismatch_warned = True + processor._stream_dim_mismatch_warned = True ez.logger.warning( f"{type(processor).__name__} is configured with axis={configured!r} but messages declare " - f"chunk_dim={message.chunk_dim!r}. State carried between messages will be applied along " - f"{configured!r}, whose length does not grow. Set axis=None to follow the declared chunk dimension." + f"stream_dim={message.stream_dim!r}. State carried between messages will be applied along " + f"{configured!r}, whose length does not grow. Set axis=None to follow the declared stream dimension." ) return configured def resolve_feature_dim(message: AxisArray, position: int = -1) -> str: - """The dimension at *position*, skipping the chunk dimension. + """The dimension at *position*, skipping the stream dimension. For processors whose axis is a *static* one -- channels, coordinate - components, feature labels. ``chunk_dim`` is emphatically not the answer - here, but the naive ``dims[position]`` can silently *be* the chunk + components, feature labels. ``stream_dim`` is emphatically not the answer + here, but the naive ``dims[position]`` can silently *be* the stream dimension: a ``(ch, time)`` stream makes ``dims[-1]`` the accumulating axis, and an affine transform would then matmul across time while a slicer would discard samples. - Falls back to ``dims[position]`` when the chunk dimension is all there is, + Falls back to ``dims[position]`` when the stream dimension is all there is, which keeps 1-D messages working rather than raising on them. """ - candidates = [d for d in message.dims if d != message.chunk_dim] + candidates = [d for d in message.dims if d != message.stream_dim] if not candidates: return message.dims[position] return candidates[position] @@ -150,26 +150,26 @@ def resolve_feature_dim(message: AxisArray, position: int = -1) -> str: def resolve_transform_dim(message: AxisArray, streaming_dims: typing.Iterable[str] = STREAMING_DIMS) -> str: """The regularly-sampled dimension a transform consumes. - Neither :func:`resolve_chunk_dim` nor :func:`resolve_feature_dim` fits a + Neither :func:`resolve_stream_dim` nor :func:`resolve_feature_dim` fits a stage like :obj:`~ezmsg.sigproc.spectrum.Spectrum`, which needs the axis whose ``gain`` is a sample period and whose extent is the transform length: - * On a raw ``(time, ch)`` stream that *is* the chunk dimension. + * On a raw ``(time, ch)`` stream that *is* the stream dimension. * On windowed ``(win, time, ch)`` it is ``time`` -- ``win`` is what accumulates, but each window's spectrum is taken over ``time``. - So: prefer the innermost non-chunk dimension carrying a ``LinearAxis``, and - fall back to the chunk dimension when there is none. ``ch`` carries a + So: prefer the innermost non-stream dimension carrying a ``LinearAxis``, and + fall back to the stream dimension when there is none. ``ch`` carries a ``CoordinateAxis`` (or no axis at all), so the raw case falls through correctly rather than transforming across channels. """ - chunk_dim = resolve_chunk_dim(message, streaming_dims) + stream_dim = resolve_stream_dim(message, streaming_dims) for name in reversed(message.dims): - if name == chunk_dim: + if name == stream_dim: continue if isinstance(message.axes.get(name), AxisArray.LinearAxis): return name - return chunk_dim + return stream_dim def with_fingerprint(axis: AxisArray.CoordinateAxis) -> AxisArray.CoordinateAxis: @@ -188,7 +188,7 @@ def with_fingerprint(axis: AxisArray.CoordinateAxis) -> AxisArray.CoordinateAxis A primed one arrives with the answer already attached. Apply it to axes that describe the stream -- channel labels, frequency - labels, feature labels -- not to per-message coordinates along the chunk + labels, feature labels -- not to per-message coordinates along the stream dimension, whose fingerprint no consumer reads and whose data is new every message anyway. """ diff --git a/tests/test_baseproc.py b/tests/test_baseproc.py index 6133c5b..cc0361f 100644 --- a/tests/test_baseproc.py +++ b/tests/test_baseproc.py @@ -1270,7 +1270,7 @@ def _msg(n_time, labels, fs=100.0, key="dev", coord_time=False): "ch": AxisArray.CoordinateAxis(data=np.array(labels), dims=["ch"]), }, key=key, - chunk_dim="time", + stream_dim="time", ) @staticmethod @@ -1286,7 +1286,7 @@ def _win_msg(n_win, n_time=10, labels=("c0", "c1")): "ch": AxisArray.CoordinateAxis(data=np.array(labels), dims=["ch"]), }, key="dev", - chunk_dim="win", + stream_dim="win", ) def _resets(self, proc, messages): @@ -1338,10 +1338,10 @@ def test_coordinate_streaming_axis_does_not_reset(self): msgs = [self._msg(5, ["c0", "c1"], coord_time=True) for _ in range(3)] assert self._resets(proc, msgs) == [0] - def test_chunk_dim_names_the_dimension_that_grows(self): + def test_stream_dim_names_the_dimension_that_grows(self): """Downstream of a windowing stage, `win` grows and `time` is fixed. - A consumer cannot infer this -- `time` is the chunk dimension on a raw + A consumer cannot infer this -- `time` is the stream dimension on a raw signal and a fixed within-window axis here -- so the producer declares it and the consumer needs to know nothing. """ @@ -1437,15 +1437,15 @@ def spectrum(freq_offset): "ch": AxisArray.CoordinateAxis(data=np.array(["c0", "c1"]), dims=["ch"]), }, key="dev", - chunk_dim="time", + stream_dim="time", ) proc = self.Probe() assert proc._hash_message(spectrum(5.0)) != proc._hash_message(spectrum(70.0)) assert proc._hash_message(spectrum(5.0)) == proc._hash_message(spectrum(5.0)) - def test_exclude_dims_is_additive_to_the_chunk_dim(self): - """Naming a dimension to ignore must not un-exclude the chunk dim.""" + def test_exclude_dims_is_additive_to_the_stream_dim(self): + """Naming a dimension to ignore must not un-exclude the stream dim.""" import numpy as np def msg(n_time, labels): @@ -1457,7 +1457,7 @@ def msg(n_time, labels): "ch": AxisArray.CoordinateAxis(data=np.array(labels), dims=["ch"]), }, key="dev", - chunk_dim="time", + stream_dim="time", ) proc = self.Probe() diff --git a/tests/test_hash_witness.py b/tests/test_hash_witness.py index a6edc1b..81dfcda 100644 --- a/tests/test_hash_witness.py +++ b/tests/test_hash_witness.py @@ -9,7 +9,7 @@ mutation a real one can undergo. Three bugs were found this way and each has a named test below: a witness blind -to ``exclude_dims``, a specialised validator that assumed the chunk axis stayed +to ``exclude_dims``, a specialised validator that assumed the stream axis stayed linear, and a generic one that skipped an excluded dimension whose axis had disappeared. """ @@ -50,20 +50,20 @@ def msg( *, fs: float = 100.0, key: str = "dev", - n_chunk: int = 8, + n_stream: int = 8, offset: float = 0.0, ch_axis: CoordinateAxis | None = None, - chunk_dim: str | None = "time", + stream_dim: str | None = "time", ) -> AxisArray: return AxisArray( - np.zeros((n_chunk, len(labels)), np.float32), + np.zeros((n_stream, len(labels)), np.float32), dims=["time", "ch"], axes={ "time": AxisArray.TimeAxis(fs=fs, offset=offset), "ch": ch_axis if ch_axis is not None else CoordinateAxis(data=np.array(labels), dims=["ch"]), }, key=key, - **({"chunk_dim": chunk_dim} if chunk_dim else {}), + **({"stream_dim": stream_dim} if stream_dim else {}), ) @@ -75,7 +75,7 @@ def test_a_reused_axis_object_hits_and_agrees(self): first = probe._message_hash(msg(["a", "b"], ch_axis=hoisted)) probe._hash = first for step in range(1, 6): - m = msg(["a", "b"], ch_axis=hoisted, offset=step * 0.1, n_chunk=8 + step) + m = msg(["a", "b"], ch_axis=hoisted, offset=step * 0.1, n_stream=8 + step) assert probe._message_hash(m) == recomputed(m) == first def test_a_rebuilt_axis_with_equal_content_still_agrees(self): @@ -105,22 +105,22 @@ def test_a_real_change_is_not_masked(self, mutate): assert after == recomputed(mutate()) assert after != before - def test_withdrawing_chunk_dim_changes_nothing_here(self): + def test_withdrawing_stream_dim_changes_nothing_here(self): """Undeclared falls back to ``STREAMING_DIMS``, which names the same dimension for a ``(time, ch)`` stream -- so the hash is unchanged, and the witness has to agree rather than assume a declaration change matters.""" probe = Probe() before = probe._message_hash(msg(["a", "b"])) probe._hash = before - undeclared = msg(["a", "b"], chunk_dim=None) + undeclared = msg(["a", "b"], stream_dim=None) assert probe._message_hash(undeclared) == recomputed(undeclared) == before - def test_chunk_size_jitter_does_not_disturb_it(self): + def test_stream_size_jitter_does_not_disturb_it(self): probe = Probe() hoisted = CoordinateAxis(data=np.array(["a", "b"]), dims=["ch"]) - first = probe._message_hash(msg(["a", "b"], ch_axis=hoisted, n_chunk=8)) + first = probe._message_hash(msg(["a", "b"], ch_axis=hoisted, n_stream=8)) probe._hash = first - assert probe._message_hash(msg(["a", "b"], ch_axis=hoisted, n_chunk=37)) == first + assert probe._message_hash(msg(["a", "b"], ch_axis=hoisted, n_stream=37)) == first class TestTheBugsTheFuzzFound: @@ -132,7 +132,7 @@ def test_exclude_dims_is_part_of_the_witness(self): probe._hash = probe._message_hash(m) assert probe._message_hash(m, exclude_dims=("ch",)) == recomputed(m, exclude_dims=("ch",)) - def test_a_chunk_axis_that_stops_being_linear(self): + def test_a_stream_axis_that_stops_being_linear(self): """An irregular-rate stream swaps its TimeAxis for a CoordinateAxis. The specialised validator reads ``.gain`` directly and must not raise.""" probe = Probe() @@ -146,7 +146,7 @@ def test_a_chunk_axis_that_stops_being_linear(self): "ch": hoisted, }, key="dev", - chunk_dim="time", + stream_dim="time", ) assert probe._message_hash(irregular) == recomputed(irregular) @@ -164,14 +164,14 @@ def test_an_excluded_dimension_losing_its_axis(self): "feat": AxisArray.LinearAxis(gain=2.0, offset=0.0), }, key="dev", - chunk_dim="time", + stream_dim="time", ) without = AxisArray( np.zeros((8, 2, 2), np.float32), dims=["time", "ch", "feat"], axes={"time": AxisArray.TimeAxis(fs=100.0), "ch": hoisted}, key="dev", - chunk_dim="time", + stream_dim="time", ) kwargs = {"exclude_dims": ("feat",)} probe._hash = probe._message_hash(with_axis, **kwargs) @@ -220,16 +220,16 @@ def test_fuzz_the_fast_path_never_disagrees(seed: int): rng = random.Random(seed) labels_pool = [["a", "b", "c"], ["x", "y", "z"], ["a", "b", "c", "d"], ["a", "b"]] - def build(dims, chunk, labels, fs, key, n_chunk, offset, coord_time): + def build(dims, stream, labels, fs, key, n_stream, offset, coord_time): shape, axes = [], {} for dim in dims: if dim == "ch": shape.append(len(labels)) axes["ch"] = CoordinateAxis(data=np.array(labels), dims=["ch"]) elif dim in ("time", "win"): - shape.append(n_chunk) + shape.append(n_stream) axes[dim] = ( - CoordinateAxis(data=np.arange(n_chunk).astype(float), dims=[dim], unit="s") + CoordinateAxis(data=np.arange(n_stream).astype(float), dims=[dim], unit="s") if coord_time else AxisArray.TimeAxis(fs=fs, offset=offset) ) @@ -240,13 +240,13 @@ def build(dims, chunk, labels, fs, key, n_chunk, offset, coord_time): axes[dim] = CoordinateAxis(data=np.array([f"{dim}0", f"{dim}1"]), dims=[dim]) elif roll < 0.7: axes[dim] = AxisArray.LinearAxis(gain=rng.choice([1.0, 2.0]), offset=rng.choice([0.0, 5.0])) - extra = {"chunk_dim": chunk} if chunk in dims else {} + extra = {"stream_dim": stream} if stream in dims else {} return AxisArray(np.zeros(shape, np.float32), dims=list(dims), axes=axes, key=key, **extra) checked = 0 for _ in range(120): probe = Probe() - dims, chunk = rng.choice(DIMSETS) + dims, stream = rng.choice(DIMSETS) kwargs = rng.choice(CALL_KWARGS) labels, fs, key = rng.choice(labels_pool), rng.choice([100.0, 200.0]), rng.choice(["dev", "dev2"]) hoisted = CoordinateAxis(data=np.array(labels), dims=["ch"]) @@ -260,12 +260,12 @@ def build(dims, chunk, labels, fs, key, n_chunk, offset, coord_time): elif roll < 0.32: key = rng.choice(["dev", "dev2"]) elif roll < 0.38: - dims, chunk = rng.choice(DIMSETS) + dims, stream = rng.choice(DIMSETS) elif roll < 0.44: kwargs = rng.choice(CALL_KWARGS) elif roll < 0.48: probe.state = {} # a stateful_op restore mid-stream - message = build(dims, chunk, labels, fs, key, rng.choice([8, 13, 21]), step * 0.1, rng.random() < 0.15) + message = build(dims, stream, labels, fs, key, rng.choice([8, 13, 21]), step * 0.1, rng.random() < 0.15) # Half the time the producer hands back the same axis object. if "ch" in message.axes and rng.random() < 0.5: if len(labels) == message.data.shape[message.dims.index("ch")]: @@ -273,7 +273,7 @@ def build(dims, chunk, labels, fs, key, n_chunk, offset, coord_time): got = probe._message_hash(message, **kwargs) assert got == recomputed( message, **kwargs - ), f"stale hash: dims={message.dims} chunk_dim={message.chunk_dim} kwargs={kwargs}" + ), f"stale hash: dims={message.dims} stream_dim={message.stream_dim} kwargs={kwargs}" probe._hash = got checked += 1 assert checked == 120 * 14 diff --git a/tests/test_chunkdim.py b/tests/test_streamdim.py similarity index 69% rename from tests/test_chunkdim.py rename to tests/test_streamdim.py index 23f7a63..d804910 100644 --- a/tests/test_chunkdim.py +++ b/tests/test_streamdim.py @@ -2,7 +2,7 @@ ``dims[0]`` is not "the streaming axis" and ``dims[-1]`` is not "the channel axis" -- both are positions, and both break under transpose and downstream of a -windowing stage. ``AxisArray.chunk_dim`` is the producer's declaration of which +windowing stage. ``AxisArray.stream_dim`` is the producer's declaration of which dimension messages accumulate along; these helpers turn it into the axis a given kind of processor should use. @@ -14,9 +14,9 @@ from ezmsg.util.messages.axisarray import AxisArray, CoordinateAxis from ezmsg.baseproc import ( - resolve_chunk_dim, - resolve_configured_chunk_dim, + resolve_configured_stream_dim, resolve_feature_dim, + resolve_stream_dim, resolve_transform_dim, ) @@ -27,9 +27,9 @@ def _ch_axis(n): return CoordinateAxis(data=np.array([f"ch{i}" for i in range(n)]), dims=["ch"]) -def transposed(n_ch=3, n_time=8, chunk_dim="time"): +def transposed(n_ch=3, n_time=8, stream_dim="time"): """``(ch, time)``: the first dim is static, the accumulating one is second.""" - kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + kwargs = {"stream_dim": stream_dim} if stream_dim else {} return AxisArray( np.arange(n_ch * n_time, dtype=float).reshape(n_ch, n_time), dims=["ch", "time"], @@ -50,12 +50,12 @@ def windowed(n_win=4, n_time=8, n_ch=3): "ch": _ch_axis(n_ch), }, key="dev", - chunk_dim="win", + stream_dim="win", ) -def raw(n_time=8, n_ch=3, chunk_dim="time"): - kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} +def raw(n_time=8, n_ch=3, stream_dim="time"): + kwargs = {"stream_dim": stream_dim} if stream_dim else {} return AxisArray( np.zeros((n_time, n_ch), dtype=float), dims=["time", "ch"], @@ -65,47 +65,47 @@ def raw(n_time=8, n_ch=3, chunk_dim="time"): ) -class TestResolveChunkDim: +class TestResolveStreamDim: def test_the_declaration_wins_over_position(self): msg = transposed() - assert msg.dims[0] != msg.chunk_dim, "the fixture must distinguish the two" - assert resolve_chunk_dim(msg) == "time" + assert msg.dims[0] != msg.stream_dim, "the fixture must distinguish the two" + assert resolve_stream_dim(msg) == "time" def test_it_follows_a_windowing_stage_onto_win(self): - assert resolve_chunk_dim(windowed()) == "win" + assert resolve_stream_dim(windowed()) == "win" - def test_an_undeclared_chunk_dim_falls_back_to_streaming_dims(self): - assert resolve_chunk_dim(transposed(chunk_dim=None)) == "time" + def test_an_undeclared_stream_dim_falls_back_to_streaming_dims(self): + assert resolve_stream_dim(transposed(stream_dim=None)) == "time" def test_the_streaming_dims_fallback_is_configurable(self): msg = windowed() - object.__setattr__(msg, "chunk_dim", None) - assert resolve_chunk_dim(msg, ("win",)) == "win" + object.__setattr__(msg, "stream_dim", None) + assert resolve_stream_dim(msg, ("win",)) == "win" def test_dims_zero_is_the_last_resort_only(self): """Nothing declared and nothing recognised: the position is all there is.""" msg = AxisArray(np.zeros((4, 2)), dims=["a", "b"], key="dev") - assert resolve_chunk_dim(msg) == "a" + assert resolve_stream_dim(msg) == "a" class TestResolveFeatureDim: - def test_it_skips_the_chunk_dim_on_a_transposed_stream(self): + def test_it_skips_the_stream_dim_on_a_transposed_stream(self): """``dims[-1]`` here is ``time``. A slicer defaulting to it would drop samples, and an affine transform would matmul across time.""" msg = transposed() - assert msg.dims[-1] == msg.chunk_dim, "the fixture must make the naive guess wrong" + assert msg.dims[-1] == msg.stream_dim, "the fixture must make the naive guess wrong" assert resolve_feature_dim(msg) == "ch" def test_it_is_unchanged_on_a_conventional_stream(self): assert resolve_feature_dim(raw()) == "ch" - def test_position_zero_skips_the_chunk_dim_too(self): - """RangedAggregate's case: ``dims[0]`` is usually the chunk dim, which is + def test_position_zero_skips_the_stream_dim_too(self): + """RangedAggregate's case: ``dims[0]`` is usually the stream dim, which is the worst possible default for an axis that must carry band values.""" assert resolve_feature_dim(windowed(), 0) == "time" - def test_a_chunk_only_message_falls_back_rather_than_raising(self): - msg = AxisArray(np.zeros(8), dims=["time"], axes={"time": AxisArray.TimeAxis(fs=FS)}, chunk_dim="time") + def test_a_stream_only_message_falls_back_rather_than_raising(self): + msg = AxisArray(np.zeros(8), dims=["time"], axes={"time": AxisArray.TimeAxis(fs=FS)}, stream_dim="time") assert resolve_feature_dim(msg) == "time" @@ -114,7 +114,7 @@ def test_windowed_input_transforms_within_the_window(self): """``win`` accumulates, but each window's spectrum is over ``time``.""" assert resolve_transform_dim(windowed()) == "time" - def test_raw_input_falls_through_to_the_chunk_dim(self): + def test_raw_input_falls_through_to_the_stream_dim(self): """``ch`` carries a CoordinateAxis, not a LinearAxis, so it is not a candidate and the rule lands back on ``time``.""" assert resolve_transform_dim(raw()) == "time" @@ -123,14 +123,14 @@ def test_it_holds_under_transposition(self): assert resolve_transform_dim(transposed()) == "time" -class TestResolveConfiguredChunkDim: +class TestResolveConfiguredStreamDim: def test_an_explicit_axis_still_wins(self): """The escape hatch stays open: an explicit axis is an instruction.""" class Proc: STREAMING_DIMS = ("time",) - assert resolve_configured_chunk_dim(Proc(), windowed(), "time") == "time" + assert resolve_configured_stream_dim(Proc(), windowed(), "time") == "time" def test_a_disagreement_warns_once(self, caplog): class Proc: @@ -139,15 +139,15 @@ class Proc: proc = Proc() with caplog.at_level("WARNING"): for _ in range(3): - resolve_configured_chunk_dim(proc, windowed(), "time") - assert sum("chunk_dim" in r.message for r in caplog.records) == 1 + resolve_configured_stream_dim(proc, windowed(), "time") + assert sum("stream_dim" in r.message for r in caplog.records) == 1 def test_agreement_is_silent(self, caplog): class Proc: STREAMING_DIMS = ("time",) with caplog.at_level("WARNING"): - resolve_configured_chunk_dim(Proc(), raw(), "time") + resolve_configured_stream_dim(Proc(), raw(), "time") assert not caplog.records def test_a_mere_guess_never_warns(self, caplog): @@ -158,7 +158,7 @@ class Proc: STREAMING_DIMS = ("time",) msg = windowed() - object.__setattr__(msg, "chunk_dim", None) + object.__setattr__(msg, "stream_dim", None) with caplog.at_level("WARNING"): - resolve_configured_chunk_dim(Proc(), msg, "win") + resolve_configured_stream_dim(Proc(), msg, "win") assert not caplog.records