From 453b80335b2ce0665ad9e22b7fc2c30b57400440 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Tue, 8 Sep 2026 22:53:40 -0400 Subject: [PATCH 1/2] Rename chunk_dim to stream_dim, following ezmsg core Tracks the upstream rename of `AxisArray.chunk_dim`. `chunk` was overloaded -- xarray's `DataArray.chunk`, dask, zarr and HDF5 all use it for memory/IO layout, and ezmsg's own `array_chunker` / `chunk_len` mean "how much data per message" -- so the field now names its role instead: the dimension the stream extends along. Bumps the ezmsg floor to 3.10.0b3, the first release carrying the new name. Pure rename otherwise. 369 passed, 24 skipped. --- pyproject.toml | 6 +++--- src/ezmsg/learn/collection/sample_adapt_regressor.py | 4 ++-- src/ezmsg/learn/dim_reduce/adaptive_decomp.py | 12 ++++++------ src/ezmsg/learn/dim_reduce/incremental_decomp.py | 8 ++++---- src/ezmsg/learn/process/flatten.py | 8 ++++---- src/ezmsg/learn/process/sgd.py | 6 +++--- src/ezmsg/learn/process/slda.py | 8 ++++---- src/ezmsg/learn/util.py | 2 +- tests/dim_reduce/test_incremental_decomp.py | 2 +- tests/unit/test_axis_fingerprint_priming.py | 4 ++-- 10 files changed, 30 insertions(+), 30 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index b34f522..f6d4967 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,13 +9,13 @@ license = "MIT" requires-python = ">=3.10" dynamic = ["version"] dependencies = [ - # 3.10.0b2 for CoordinateAxis.fingerprint and AxisArray.chunk_dim, which + # 3.10.0b3 for CoordinateAxis.fingerprint and AxisArray.stream_dim, which # ezmsg-baseproc's default state hash reads. Pinned directly rather than # left to the transitive requirement: uv only enables pre-releases for a # package named with a pre-release marker in *this* file. - "ezmsg>=3.10.0b2", + "ezmsg>=3.10.0b3", "ezmsg-baseproc>=1.13.0", - "ezmsg-sigproc>=3.8.1", # Window/Resample must resolve chunk_dim from axis=None + "ezmsg-sigproc>=3.8.1", # Window/Resample must resolve stream_dim from axis=None "numpy", "scipy", "array-api-compat", diff --git a/src/ezmsg/learn/collection/sample_adapt_regressor.py b/src/ezmsg/learn/collection/sample_adapt_regressor.py index 405ef82..25b5c5e 100644 --- a/src/ezmsg/learn/collection/sample_adapt_regressor.py +++ b/src/ezmsg/learn/collection/sample_adapt_regressor.py @@ -177,7 +177,7 @@ class SampleAdaptRegressorSettings(ez.Settings): """.. deprecated:: 1.6 Scheduled for removal in 2.0. Resampling buffers along the dimension messages accumulate along, which now comes from - :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`.""" + :attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim`.""" def __post_init__(self) -> None: warn_axis_deprecated(self, "resample_axis", package="ezmsg-learn", removal="2.0") @@ -282,7 +282,7 @@ def configure(self) -> None: if use_window: self.WINDOW.apply_settings( WindowSettings( - # No `axis`: Window follows the stream's chunk_dim, which + # No `axis`: Window follows the stream's stream_dim, which # is what "time" was standing in for. newaxis="win", window_dur=self.SETTINGS.decode_window_dur, diff --git a/src/ezmsg/learn/dim_reduce/adaptive_decomp.py b/src/ezmsg/learn/dim_reduce/adaptive_decomp.py index f416c29..de8facb 100644 --- a/src/ezmsg/learn/dim_reduce/adaptive_decomp.py +++ b/src/ezmsg/learn/dim_reduce/adaptive_decomp.py @@ -17,7 +17,7 @@ BaseAdaptiveTransformer, BaseAdaptiveTransformerUnit, processor_state, - resolve_chunk_dim, + resolve_stream_dim, warn_axis_deprecated, ) from ezmsg.util.messages.axisarray import AxisArray, replace @@ -37,12 +37,12 @@ class AdaptiveDecompSettings(ez.Settings): ``None`` (default) decomposes every dimension except the one messages accumulate along, iterating over that one. Naming a dimension (e.g. - ``"ch"``) decomposes it and iterates over the chunk dimension instead. + ``"ch"``) decomposes it and iterates over the stream dimension instead. .. deprecated:: 1.6 The ``"!time"`` spelling -- "iterate over time" -- is scheduled for removal in 2.0. It hardcodes what - :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim` now + :attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim` now answers; leave this unset for the same behaviour.""" def __post_init__(self) -> None: @@ -101,7 +101,7 @@ def _calculate_axis_groups(self, message: AxisArray): # Iterate over the dimension messages accumulate along and collapse # every other one -- what "!time" spelled, with the dimension read # off the stream instead of assumed. - iter_axis = resolve_chunk_dim(message, self.STREAMING_DIMS) + iter_axis = resolve_stream_dim(message, self.STREAMING_DIMS) it_ax_ix = message.get_axis_idx(iter_axis) targ_axes = message.dims[:it_ax_ix] + message.dims[it_ax_ix + 1 :] off_targ_axes = [] @@ -116,8 +116,8 @@ def _calculate_axis_groups(self, message: AxisArray): targ_axes = [axis] # Iterate over the dimension messages accumulate along. This was a # hand-rolled `"win" if "win" in dims else "time"` guess, which is - # exactly what chunk_dim exists to answer. - iter_axis = resolve_chunk_dim(message, self.STREAMING_DIMS) + # exactly what stream_dim exists to answer. + iter_axis = resolve_stream_dim(message, self.STREAMING_DIMS) if iter_axis == axis: raise ValueError(f"Iterating axis ({iter_axis}) cannot be the same as the target axis ({axis})") it_ax_ix = message.get_axis_idx(iter_axis) diff --git a/src/ezmsg/learn/dim_reduce/incremental_decomp.py b/src/ezmsg/learn/dim_reduce/incremental_decomp.py index 3b87c49..c7ec538 100644 --- a/src/ezmsg/learn/dim_reduce/incremental_decomp.py +++ b/src/ezmsg/learn/dim_reduce/incremental_decomp.py @@ -85,7 +85,7 @@ def _initialize_processors( # Create windowing processor if update_interval is specified if settings.update_interval > 0: # Only the "!axis" spelling names the iteration dimension outright. - # Otherwise leave it to Window, which resolves chunk_dim from the + # Otherwise leave it to Window, which resolves stream_dim from the # message -- this is the "likely incorrect" hardcoded "time" that # used to be here, and there is no message to resolve from at this # point anyway. @@ -114,13 +114,13 @@ def _partial_fit_windowed(self, train_msg: AxisArray) -> None: axis_idx = train_msg.get_axis_idx("win") win_axis = train_msg.axes["win"] offsets = win_axis.value(np.asarray(range(train_msg.data.shape[axis_idx]))) - # Slicing "win" away leaves each sub-message no longer a chunk along + # Slicing "win" away leaves each sub-message no longer a stream along # it. Newer ezmsg clears the declaration for us, but say what these # slices *are* chunks along rather than leaving them undeclared: # successive windows advance along the within-window axis, which is # what the offset fix-up below re-anchors. Clearing it first keeps # this working on ezmsg versions that do not. - unbundled = replace(train_msg, chunk_dim=None) + unbundled = replace(train_msg, stream_dim=None) for ix, _msg in enumerate(unbundled.iter_over_axis("win")): _msg = replace( _msg, @@ -131,7 +131,7 @@ def _partial_fit_windowed(self, train_msg: AxisArray) -> None: offset=_msg.axes["time"].offset + offsets[ix], ), }, - chunk_dim="time", + stream_dim="time", ) self._procs["decomp"].partial_fit(_msg) diff --git a/src/ezmsg/learn/process/flatten.py b/src/ezmsg/learn/process/flatten.py index 6e247b2..54b1ef5 100644 --- a/src/ezmsg/learn/process/flatten.py +++ b/src/ezmsg/learn/process/flatten.py @@ -26,7 +26,7 @@ BaseStatefulTransformer, BaseTransformerUnit, processor_state, - resolve_chunk_dim, + resolve_stream_dim, ) from ezmsg.sigproc.flatten import ( FlattenSettings as SigprocFlattenSettings, @@ -122,7 +122,7 @@ class FlattenTransformer(BaseStatefulTransformer[FlattenSettings, AxisArray, Axi """ STREAMING_DIMS = ("win",) - """Fallback chunk dimension when the producer does not declare one. + """Fallback stream dimension when the producer does not declare one. The base class defaults to ``("time",)``, which is exactly wrong here: the canonical input is ``(win, time, ch[, feature])``, where ``win`` is what @@ -131,12 +131,12 @@ class FlattenTransformer(BaseStatefulTransformer[FlattenSettings, AxisArray, Axi stop this noticing a window-length change, while including ``win`` would rebuild the inner transformer every time the window count jittered. - Consulted only when :attr:`AxisArray.chunk_dim` is absent; a producer that + Consulted only when :attr:`AxisArray.stream_dim` is absent; a producer that declares it -- ezmsg-sigproc's ``Window`` does -- overrides this. """ def _reset_state(self, message: AxisArray) -> None: - preserve_axis = self.settings.preserve_axis or resolve_chunk_dim(message, self.STREAMING_DIMS) + preserve_axis = self.settings.preserve_axis or resolve_stream_dim(message, self.STREAMING_DIMS) sample_axis = self.settings.sample_axis or preserve_axis feature_axis = self.settings.feature_axis diff --git a/src/ezmsg/learn/process/sgd.py b/src/ezmsg/learn/process/sgd.py index bf4474b..1d2c06f 100644 --- a/src/ezmsg/learn/process/sgd.py +++ b/src/ezmsg/learn/process/sgd.py @@ -6,7 +6,7 @@ BaseAdaptiveTransformer, BaseAdaptiveTransformerUnit, processor_state, - resolve_chunk_dim, + resolve_stream_dim, ) from ezmsg.util.messages.axisarray import AxisArray from ezmsg.util.messages.util import replace @@ -37,7 +37,7 @@ class SGDDecoderState: class SGDDecoderTransformer(BaseAdaptiveTransformer[SGDDecoderSettings, AxisArray, ClassifierMessage, SGDDecoderState]): STREAMING_DIMS = ("win", "time") - """This decoder is fed windows, so a producer that declares no ``chunk_dim`` + """This decoder is fed windows, so a producer that declares no ``stream_dim`` is accumulating along ``win`` rather than ``time``. The base default would guess ``time`` and flatten the windows into the feature vector.""" @@ -96,7 +96,7 @@ def _process(self, message: AxisArray) -> ClassifierMessage | None: if np.any(np.isnan(message.data)): return None try: - chunk = resolve_chunk_dim(message, self.STREAMING_DIMS) + chunk = resolve_stream_dim(message, self.STREAMING_DIMS) chunk_idx = message.get_axis_idx(chunk) data = message.data if chunk_idx == 0 else np.moveaxis(message.data, chunk_idx, 0) X = data.reshape((data.shape[0], -1)) diff --git a/src/ezmsg/learn/process/slda.py b/src/ezmsg/learn/process/slda.py index 3840bc4..a88dd87 100644 --- a/src/ezmsg/learn/process/slda.py +++ b/src/ezmsg/learn/process/slda.py @@ -16,7 +16,7 @@ BaseStatefulTransformer, BaseTransformerUnit, processor_state, - resolve_configured_chunk_dim, + resolve_configured_stream_dim, warn_axis_deprecated, ) from ezmsg.util.messages.axisarray import AxisArray @@ -39,7 +39,7 @@ class SLDASettings(ez.Settings): Scheduled for removal in 2.0. The samples this classifies accumulate along one dimension, and the cached output template is keyed to it; that dimension now comes from - :attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`.""" + :attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim`.""" def __post_init__(self) -> None: warn_axis_deprecated(self, package="ezmsg-learn", removal="2.0") @@ -48,7 +48,7 @@ def __post_init__(self) -> None: @processor_state class SLDAState: axis: str = "" - """The resolved chunk dimension, fixed at reset so every later use agrees.""" + """The resolved stream dimension, fixed at reset so every later use agrees.""" lda: LDA out_template: typing.Optional[ClassifierMessage] = None @@ -56,7 +56,7 @@ class SLDAState: class SLDATransformer(BaseStatefulTransformer[SLDASettings, AxisArray, ClassifierMessage, SLDAState]): def _reset_state(self, message: AxisArray) -> None: - self.state.axis = resolve_configured_chunk_dim(self, message, self.settings.axis, legacy_default="time") + self.state.axis = resolve_configured_stream_dim(self, message, self.settings.axis, legacy_default="time") if self.settings.settings_path[-4:] == ".mat": # Expects a very specific format from a specific project. Not for general use. import scipy.io as sio diff --git a/src/ezmsg/learn/util.py b/src/ezmsg/learn/util.py index 5a113ed..5b5fcb0 100644 --- a/src/ezmsg/learn/util.py +++ b/src/ezmsg/learn/util.py @@ -24,7 +24,7 @@ def with_fingerprint(axis: AxisArray.CoordinateAxis) -> AxisArray.CoordinateAxis first consumer in every receiving process, on every message. Apply it to axes that describe the stream -- channel labels, class labels, - lag labels -- not to per-message coordinates along the chunk dimension, + lag labels -- not to per-message coordinates along the stream dimension, whose fingerprint no consumer reads. """ axis.fingerprint diff --git a/tests/dim_reduce/test_incremental_decomp.py b/tests/dim_reduce/test_incremental_decomp.py index ef0ef9c..3e16ebf 100644 --- a/tests/dim_reduce/test_incremental_decomp.py +++ b/tests/dim_reduce/test_incremental_decomp.py @@ -135,7 +135,7 @@ def test_initialization_pca(self, pca_test_data, update_interval): if update_interval > 0: assert "windowing" in transformer._procs win = transformer._procs["windowing"] - # Unset on purpose: Window resolves the stream's chunk_dim, which is + # Unset on purpose: Window resolves the stream's stream_dim, which is # what the old hardcoded "time" was standing in for (and got wrong # downstream of another windowing stage). assert win.settings.axis is None diff --git a/tests/unit/test_axis_fingerprint_priming.py b/tests/unit/test_axis_fingerprint_priming.py index ad4a5de..8649e4d 100644 --- a/tests/unit/test_axis_fingerprint_priming.py +++ b/tests/unit/test_axis_fingerprint_priming.py @@ -31,7 +31,7 @@ def signal(labels, n_time=32, fs=100.0, key="dev"): "ch": CoordinateAxis(data=np.array(labels), dims=["ch"]), }, key=key, - chunk_dim="time", + stream_dim="time", ) @@ -65,7 +65,7 @@ def test_priming_survives_the_transport(self, dtype): class TestCreatedAxesArePrimed: - """Messages here carry no ``chunk_dim``: released ezmsg-sigproc does not set + """Messages here carry no ``stream_dim``: released ezmsg-sigproc does not set it, so that is what these transformers actually receive today. It is why ``FlattenTransformer.STREAMING_DIMS`` names ``win`` -- the base class's ``("time",)`` fallback would exclude the lag dimension, which is the one From fc20490bfaeea6dc85ff5beadb4c19eb992d2b80 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 9 Sep 2026 03:01:06 -0400 Subject: [PATCH 2/2] Cap the test job at 30 minutes and supersede in-flight runs A wedged job used to run to GitHub's 6-hour default. One did: an ezmsg-simbiophys run hung immediately after `collected 161 items`, produced not one test dot, and sat on a runner for 3h09m until it was cancelled by hand -- long enough to starve the macOS queue for every other repo in the org. `timeout-minutes: 30` bounds that; the suite runs in well under a minute on every platform, so the margin is generous. The concurrency group makes a new push to a PR cancel the run already in flight for that ref, so a stuck job cannot hold a slot while its own replacement queues behind it. --- .github/workflows/python-tests.yml | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index fea84cb..8d692d2 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -11,6 +11,12 @@ on: - dev workflow_dispatch: +# A new push to a PR supersedes the run already in flight for that ref, so a +# wedged job cannot sit on a runner while its replacement queues behind it. +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + jobs: build: strategy: @@ -21,6 +27,7 @@ jobs: - "windows-latest" - "macos-latest" runs-on: ${{matrix.os}} + timeout-minutes: 30 steps: - uses: actions/checkout@v4 @@ -53,6 +60,7 @@ jobs: minimal-install: name: Minimal install (no extras) runs-on: ubuntu-latest + timeout-minutes: 30 steps: - uses: actions/checkout@v4