Declare chunk_dim and prime the channel fingerprint

.github/workflows/python-tests.yml

@@ -4,7 +4,9 @@ on:
   push:
     branches: [main]
   pull_request:
-    branches: [main]
+    branches:
+      - main
+      - dev
   workflow_dispatch:
 
 jobs:

pyproject.toml

@@ -7,7 +7,8 @@ authors = [
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
-    "ezmsg>=3.9.0",
+    # 3.10.0b2 for AxisArray.chunk_dim and CoordinateAxis.fingerprint.
+    "ezmsg>=3.10.0b2",
     "ezmsg-baseproc>=1.7.0",
     "numpy>=1.26.4",
     "pylsl>=1.18.4b1",

src/ezmsg/lsl/inlet.py

@@ -495,6 +495,13 @@ def _setup_after_open(self) -> bool:
             # No structured metadata — fall back to numeric string labels.
             ch_labels = [str(i + 1) for i in range(n_ch)]
             ch_ax = AxisArray.CoordinateAxis(data=np.array(ch_labels), dims=["ch"])
+        # Compute the channel fingerprint once, now. It is cached on the axis and
+        # pickled with it, and every message from this connection reuses this same
+        # axis object, so one checksum covers the whole stream. Left cold it would
+        # be computed by the first stateful consumer in this process -- and, since
+        # unpickling builds a new axis object per message, by the first consumer in
+        # every other process, on every message, until the inlet reconnects.
+        ch_ax.fingerprint
         # Pre-allocate a message template.
         fs = inlet_info.nominal_srate()
         time_ax = (
@@ -509,6 +516,12 @@ def _setup_after_open(self) -> bool:
             dims=["time", "ch"],
             axes={"time": time_ax, "ch": ch_ax},
             key=key,
+            # Messages append along `time` whether the stream is regular (a
+            # LinearAxis whose offset advances) or irregular (a CoordinateAxis of
+            # per-sample timestamps). Either way its extent is just however many
+            # samples arrived, and consumers must leave it out of the state they
+            # cache against the stream's configuration.
+            chunk_dim="time",
             attrs={
                 "lsl_uid": uid,
                 "lsl_source_id": source_id,

tests/test_inlet.py

@@ -390,6 +390,50 @@ def test_inlet_stamps_stream_identity_onto_every_message():
     assert attrs["lsl_hostname"] == "rpi5"
 
 
+class TestTheTemplateIsReadyForConsumers:
+    """Two things only the source can supply, both set once per connection.
+
+    ``chunk_dim`` names the dimension messages append along -- the one whose
+    length is just however many samples arrived, and which a consumer must
+    therefore leave out of the state it caches against the stream's
+    configuration. ``fingerprint`` is the channel axis's content digest, cached
+    on the axis and pickled with it, so priming it here spares the first
+    consumer in every process from recomputing it on every message.
+    """
+
+    @pytest.mark.parametrize("srate", [50.0, 0.0], ids=["regular", "irregular"])
+    def test_the_template_declares_its_chunk_dim(self, srate):
+        producer = LSLInletProducer(settings=LSLInletSettings())
+        _connect(producer, _FakeStreamInfo(srate=srate))
+        assert producer._state.msg_template.chunk_dim == "time"
+
+    @pytest.mark.parametrize("srate", [50.0, 0.0], ids=["regular", "irregular"])
+    def test_the_channel_axis_carries_its_fingerprint(self, srate):
+        producer = LSLInletProducer(settings=LSLInletSettings())
+        _connect(producer, _FakeStreamInfo(srate=srate))
+        ch_ax = producer._state.msg_template.axes["ch"]
+        assert "_fingerprint" in ch_ax.__dict__
+        assert ch_ax.fingerprint is not None
+
+    def test_it_survives_the_transport(self):
+        import pickle
+
+        producer = LSLInletProducer(settings=LSLInletSettings())
+        _connect(producer)
+        landed = pickle.loads(pickle.dumps(producer._state.msg_template))
+        assert landed.chunk_dim == "time"
+        assert "_fingerprint" in landed.axes["ch"].__dict__
+
+    def test_a_reconnect_reprimes_for_the_new_channel_set(self):
+        producer = LSLInletProducer(settings=LSLInletSettings())
+        _connect(producer, _FakeStreamInfo(n_ch=2))
+        first = producer._state.msg_template.axes["ch"].fingerprint
+        _connect(producer, _FakeStreamInfo(n_ch=5))
+        second = producer._state.msg_template.axes["ch"]
+        assert "_fingerprint" in second.__dict__
+        assert second.fingerprint != first
+
+
 def test_setup_retries_when_the_description_cannot_be_fetched(caplog):
     """The desc is fetched over the wire, so a connect can open but not complete."""