From 13b24a490c3b04d2333b498b3749adc0ce7fa3bb Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Tue, 8 Sep 2026 22:52:59 -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. 91 passed. --- pyproject.toml | 4 ++-- src/ezmsg/tools/plot/describe.py | 8 +++---- src/ezmsg/tools/shmem/aux_meta.py | 8 +++---- src/ezmsg/tools/shmem/shmem.py | 8 +++---- src/ezmsg/tools/shmem/shmem_mirror.py | 4 ++-- tests/test_plot_describe.py | 14 ++++++------- tests/test_shmem_aux_meta.py | 30 +++++++++++++-------------- tests/test_shmem_sink.py | 16 +++++++------- 8 files changed, 46 insertions(+), 46 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index bb452d3..61c02fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,10 +9,10 @@ readme = "README.md" requires-python = ">=3.11" dynamic = ["version"] dependencies = [ - # 3.10.0b2 for AxisArray.chunk_dim, which the shmem sink follows to pick + # 3.10.0b3 for AxisArray.stream_dim, which the shmem sink follows to pick # the buffered axis, and CoordinateAxis.fingerprint, which aux_meta uses # to compare axes without touching their bytes. - "ezmsg>=3.10.0b2", + "ezmsg>=3.10.0b3", "numpy>=1.26.0", "typer>=0.24.1", ] diff --git a/src/ezmsg/tools/plot/describe.py b/src/ezmsg/tools/plot/describe.py index e957fb6..08b5867 100644 --- a/src/ezmsg/tools/plot/describe.py +++ b/src/ezmsg/tools/plot/describe.py @@ -112,7 +112,7 @@ def stream_axis(msg: typing.Any, *fallbacks: str) -> str | None: """Which dimension of *msg* the stream accumulates along. Prefers the producer's own declaration - (:attr:`~ezmsg.util.messages.axisarray.AxisArray.chunk_dim`) and falls back + (:attr:`~ezmsg.util.messages.axisarray.AxisArray.stream_dim`) and falls back to the first of *fallbacks* the message actually has, which is what these tools did before the field existed. @@ -122,10 +122,10 @@ def stream_axis(msg: typing.Any, *fallbacks: str) -> str | None: window's interior along the x-axis and treats the windows as channels -- and reads an offset that does not advance with the stream. """ - chunk_dim = getattr(msg, "chunk_dim", None) + stream_dim = getattr(msg, "stream_dim", None) dims = getattr(msg, "dims", ()) or () - if chunk_dim is not None and chunk_dim in dims: - return chunk_dim + if stream_dim is not None and stream_dim in dims: + return stream_dim for name in fallbacks: if name in dims: return name diff --git a/src/ezmsg/tools/shmem/aux_meta.py b/src/ezmsg/tools/shmem/aux_meta.py index e6af6dd..4282589 100644 --- a/src/ezmsg/tools/shmem/aux_meta.py +++ b/src/ezmsg/tools/shmem/aux_meta.py @@ -31,7 +31,7 @@ {"kind": "linear", "unit": str, "gain": float, "offset": float} {"kind": "coord", "unit": str, "dims": list[str], "data": np.ndarray} -``chunk_dim`` carries the source message's declaration of which dimension it +``stream_dim`` carries the source message's declaration of which dimension it accumulates along, or ``None`` from a producer that declares nothing. It is what the sink used to choose ``buffered_axis``, recorded so a consumer can tell the two apart -- an operator may have overridden the buffered axis. @@ -102,7 +102,7 @@ def encode_aux( attrs: typing.Mapping[str, typing.Any], key: str, buffered_axis: str, - chunk_dim: typing.Optional[str] = None, + stream_dim: typing.Optional[str] = None, ) -> tuple[bytes, list[str]]: """Serialize an AxisArray's static metadata. @@ -124,7 +124,7 @@ def encode_aux( "attrs": plain_attrs, "key": key, "buffered_axis": buffered_axis, - "chunk_dim": chunk_dim, + "stream_dim": stream_dim, } return pickle.dumps(payload, protocol=pickle.HIGHEST_PROTOCOL), dropped @@ -148,7 +148,7 @@ def decode_aux(blob: bytes) -> dict: ) # Additive keys are defaulted rather than required, so a blob from a writer # that predates them still decodes. See the module docstring. - payload.setdefault("chunk_dim", None) + payload.setdefault("stream_dim", None) return payload diff --git a/src/ezmsg/tools/shmem/shmem.py b/src/ezmsg/tools/shmem/shmem.py index 853a8d5..3c30191 100644 --- a/src/ezmsg/tools/shmem/shmem.py +++ b/src/ezmsg/tools/shmem/shmem.py @@ -159,7 +159,7 @@ class ShMemCircBuffSettings(ez.Settings): conn: typing.Optional[multiprocessing.connection.Connection] = None axis: typing.Optional[str] = None - """Dimension to buffer along. ``None`` follows the message's ``chunk_dim``. + """Dimension to buffer along. ``None`` follows the message's ``stream_dim``. The ring is a history of the stream, so this has to be the dimension messages accumulate along; buffering a static one would store the same @@ -171,7 +171,7 @@ class ShMemCircBuffSettings(ez.Settings): window count ended up inside ``frame_shape`` -- reallocating the buffer whenever the window count jittered -- and the reported sample rate was the within-window rate, a 10x error in the viewer's time base for a 10-sample - window. Set explicitly only for a producer that declares no ``chunk_dim``. + window. Set explicitly only for a producer that declares no ``stream_dim``. """ @@ -407,7 +407,7 @@ def _update_aux_if_needed(self, msg: AxisArray) -> bool: # which is knowledge it has no way to arrive at. buff_axis = self.STATE.buff_axis rolled_dims = [buff_axis] + [d for d in msg.dims if d != buff_axis] - blob, dropped = encode_aux(rolled_dims, msg.axes, msg.attrs, msg.key, buff_axis, chunk_dim=msg.chunk_dim) + blob, dropped = encode_aux(rolled_dims, msg.axes, msg.attrs, msg.key, buff_axis, stream_dim=msg.stream_dim) if dropped: dropped_set = frozenset(dropped) if self.STATE.warned_dropped_attrs != dropped_set: @@ -457,7 +457,7 @@ def _resolve_axis(self, msg: AxisArray) -> typing.Optional[str]: An explicit setting wins so an operator can still drive a producer that declares nothing; otherwise the message decides. """ - axis = self.SETTINGS.axis if self.SETTINGS.axis is not None else msg.chunk_dim + axis = self.SETTINGS.axis if self.SETTINGS.axis is not None else msg.stream_dim if axis is None: axis = "time" if "time" in msg.dims else None return axis diff --git a/src/ezmsg/tools/shmem/shmem_mirror.py b/src/ezmsg/tools/shmem/shmem_mirror.py index 7f269bf..8e90056 100644 --- a/src/ezmsg/tools/shmem/shmem_mirror.py +++ b/src/ezmsg/tools/shmem/shmem_mirror.py @@ -121,14 +121,14 @@ def dims(self) -> typing.Optional[typing.List[str]]: return None if self._aux is None else self._aux["dims"] @property - def chunk_dim(self) -> typing.Optional[str]: + def stream_dim(self) -> typing.Optional[str]: """Which dimension the *source* declared it accumulates along. Distinct from the buffered axis: an operator can override that, and a producer on ezmsg < 3.10 declares nothing, in which case this is None. """ self._refresh_aux() - return None if self._aux is None else self._aux.get("chunk_dim") + return None if self._aux is None else self._aux.get("stream_dim") @property def buffered_axis(self) -> typing.Optional[str]: diff --git a/tests/test_plot_describe.py b/tests/test_plot_describe.py index 5bbbce1..3e562dd 100644 --- a/tests/test_plot_describe.py +++ b/tests/test_plot_describe.py @@ -233,8 +233,8 @@ class TestStreamAxis: """ @staticmethod - def _windowed(chunk_dim="win"): - kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + def _windowed(stream_dim="win"): + kwargs = {"stream_dim": stream_dim} if stream_dim else {} return AxisArray( np.zeros((4, 10, 3), np.float32), dims=["win", "time", "ch"], @@ -247,7 +247,7 @@ def test_it_prefers_the_declaration(self): assert stream_axis(self._windowed(), "time") == "win" def test_it_falls_back_when_nothing_is_declared(self): - assert stream_axis(self._windowed(chunk_dim=None), "time") == "time" + assert stream_axis(self._windowed(stream_dim=None), "time") == "time" def test_the_fallback_order_is_honoured(self): msg = AxisArray( @@ -259,7 +259,7 @@ def test_the_fallback_order_is_honoured(self): assert stream_axis(msg, "time", "freq") == "freq" def test_a_declaration_naming_an_absent_dim_is_ignored(self): - """`chunk_dim` is validated at construction, but a message can reach a + """`stream_dim` is validated at construction, but a message can reach a viewer after a transform that dropped the dimension without updating it. Falling back beats indexing on a name that is not there.""" msg = AxisArray( @@ -267,9 +267,9 @@ def test_a_declaration_naming_an_absent_dim_is_ignored(self): dims=["time", "ch"], axes={"time": AxisArray.TimeAxis(fs=100.0)}, key="dev", - chunk_dim="time", + stream_dim="time", ) - object.__setattr__(msg, "chunk_dim", "win") + object.__setattr__(msg, "stream_dim", "win") assert stream_axis(msg, "time") == "time" def test_none_when_nothing_matches(self): @@ -282,6 +282,6 @@ def test_a_plain_stream_is_unaffected(self): dims=["time", "ch"], axes={"time": AxisArray.TimeAxis(fs=100.0)}, key="dev", - chunk_dim="time", + stream_dim="time", ) assert stream_axis(msg, "time") == "time" diff --git a/tests/test_shmem_aux_meta.py b/tests/test_shmem_aux_meta.py index b0c2a2d..496ee82 100644 --- a/tests/test_shmem_aux_meta.py +++ b/tests/test_shmem_aux_meta.py @@ -438,17 +438,17 @@ def test_reader_rejects_a_foreign_header_loudly(): # --------------------------------------------------------------------------- -# chunk_dim: which dimension the stream accumulates along +# stream_dim: which dimension the stream accumulates along # --------------------------------------------------------------------------- -def _windowed(n_win: int = 4, n_lag: int = 10, n_ch: int = 3, chunk_dim: str | None = "win") -> AxisArray: +def _windowed(n_win: int = 4, n_lag: int = 10, n_ch: int = 3, stream_dim: str | None = "win") -> AxisArray: """`(win, time, ch)` -- what a windowing stage emits. `time` here is the *within-window* lag dimension. Both are LinearAxes, so - nothing about the message distinguishes them except `chunk_dim`. + nothing about the message distinguishes them except `stream_dim`. """ - kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + kwargs = {"stream_dim": stream_dim} if stream_dim else {} return AxisArray( np.zeros((n_win, n_lag, n_ch), np.float32), dims=["win", "time", "ch"], @@ -462,25 +462,25 @@ def _windowed(n_win: int = 4, n_lag: int = 10, n_ch: int = 3, chunk_dim: str | N ) -class TestTheBlobCarriesChunkDim: +class TestTheBlobCarriesStreamDim: def test_it_round_trips(self): msg = _windowed() - blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "win", chunk_dim=msg.chunk_dim) - assert decode_aux(blob)["chunk_dim"] == "win" + blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "win", stream_dim=msg.stream_dim) + assert decode_aux(blob)["stream_dim"] == "win" def test_none_from_a_producer_that_declares_nothing(self): - msg = _windowed(chunk_dim=None) - blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "win", chunk_dim=msg.chunk_dim) - assert decode_aux(blob)["chunk_dim"] is None + msg = _windowed(stream_dim=None) + blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "win", stream_dim=msg.stream_dim) + assert decode_aux(blob)["stream_dim"] is None def test_it_is_distinct_from_the_buffered_axis(self): """An operator can override which axis the ring buffers; the source's own declaration is recorded separately so a consumer can tell.""" msg = _windowed() - blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "time", chunk_dim=msg.chunk_dim) + blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "time", stream_dim=msg.stream_dim) payload = decode_aux(blob) assert payload["buffered_axis"] == "time" - assert payload["chunk_dim"] == "win" + assert payload["stream_dim"] == "win" def test_a_blob_written_before_the_key_existed_still_decodes(self): """Adding a key must not break a mixed-version link -- that pairing is @@ -488,13 +488,13 @@ def test_a_blob_written_before_the_key_existed_still_decodes(self): import pickle msg = _windowed() - blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "win", chunk_dim="win") + blob, _ = encode_aux(list(msg.dims), msg.axes, msg.attrs, msg.key, "win", stream_dim="win") payload = pickle.loads(blob) - del payload["chunk_dim"] # what an older writer emits + del payload["stream_dim"] # what an older writer emits old_blob = pickle.dumps(payload, protocol=pickle.HIGHEST_PROTOCOL) decoded = decode_aux(old_blob) - assert decoded["chunk_dim"] is None + assert decoded["stream_dim"] is None assert decoded["buffered_axis"] == "win" diff --git a/tests/test_shmem_sink.py b/tests/test_shmem_sink.py index 47fd0be..5541578 100644 --- a/tests/test_shmem_sink.py +++ b/tests/test_shmem_sink.py @@ -120,13 +120,13 @@ def test_shmem_change(change_type: str): # --------------------------------------------------------------------------- -def _windowed_msg(n_win: int = 4, n_lag: int = 10, n_ch: int = 3, chunk_dim: str | None = "win") -> AxisArray: +def _windowed_msg(n_win: int = 4, n_lag: int = 10, n_ch: int = 3, stream_dim: str | None = "win") -> AxisArray: """`(win, time, ch)` -- what a windowing stage emits. `time` is the *within-window* lag dimension. Both it and `win` are - LinearAxes, so nothing distinguishes them but `chunk_dim`. + LinearAxes, so nothing distinguishes them but `stream_dim`. """ - kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} + kwargs = {"stream_dim": stream_dim} if stream_dim else {} return AxisArray( np.zeros((n_win, n_lag, n_ch), np.float32), dims=["win", "time", "ch"], @@ -140,8 +140,8 @@ def _windowed_msg(n_win: int = 4, n_lag: int = 10, n_ch: int = 3, chunk_dim: str ) -def _plain_msg(n_time: int = 20, n_ch: int = 3, chunk_dim: str | None = "time") -> AxisArray: - kwargs = {"chunk_dim": chunk_dim} if chunk_dim else {} +def _plain_msg(n_time: int = 20, n_ch: int = 3, stream_dim: str | None = "time") -> AxisArray: + kwargs = {"stream_dim": stream_dim} if stream_dim else {} return AxisArray( np.zeros((n_time, n_ch), np.float32), dims=["time", "ch"], @@ -180,10 +180,10 @@ def test_an_explicit_setting_still_wins(self): def test_an_undeclared_producer_falls_back_to_time(self): """Nothing better is available. A windowed producer that declares no - `chunk_dim` still gets the old, wrong answer -- the fix is for it to + `stream_dim` still gets the old, wrong answer -- the fix is for it to declare one, which every ezmsg source now does.""" - assert _sink()._resolve_axis(_plain_msg(chunk_dim=None)) == "time" - assert _sink()._resolve_axis(_windowed_msg(chunk_dim=None)) == "time" + assert _sink()._resolve_axis(_plain_msg(stream_dim=None)) == "time" + assert _sink()._resolve_axis(_windowed_msg(stream_dim=None)) == "time" def test_a_message_with_neither_is_skipped(self): msg = AxisArray( From 4c3b1af1e19f8ed973d0b8f811e4994bc8f55e24 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 9 Sep 2026 03:01:07 -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 | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index f6c517b..6165099 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -9,6 +9,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: @@ -19,6 +25,7 @@ jobs: - "windows-latest" - "macos-latest" runs-on: ${{matrix.os}} + timeout-minutes: 30 steps: - uses: actions/checkout@v4