Publish gates: swallow only time-empty results, pass channel-empty through

src/ezmsg/sigproc/binned_aggregate.py

@@ -52,6 +52,7 @@
 
 from .aggregate import AGGREGATORS, AggregationFunction
 from .util.binning import BinSchedule, BinStep
+from .util.message import is_empty_along
 
 
 class BinnedAggregateSettings(ez.Settings):
@@ -192,8 +193,11 @@ async def on_signal(self, message: AxisArray) -> typing.AsyncGenerator:
 
         As with :obj:`Downsample`, most input chunks at a high input rate close
         no new bin, yielding a zero-length payload; broadcasting those wastes a
-        round-trip across SHM/socket.
+        round-trip across SHM/socket. Only emptiness along the binned axis is
+        suppressed: a message that is empty along other axes (e.g. all channels
+        sliced away upstream) still flows so downstream consumers keep its
+        cadence.
         """
         result = await self.processor.__acall__(message)
-        if result is not None and result.data.size > 0:
+        if result is not None and not is_empty_along(result, (self.SETTINGS.axis,)):
             yield self.OUTPUT_SIGNAL, result

src/ezmsg/sigproc/downsample.py

@@ -15,6 +15,8 @@
     slice_along_axis,
 )
 
+from .util.message import is_empty_along
+
 
 class DownsampleSettings(ez.Settings):
     """
@@ -115,9 +117,12 @@ async def on_signal(self, message: AxisArray) -> typing.AsyncGenerator:
         period, so ``DownsampleTransformer._process`` returns a payload with
         a zero-length axis. Suppressing the broadcast in that case avoids
         shipping an empty AxisArray across SHM/socket every input chunk.
+        Only emptiness along the downsampled axis is suppressed: a message
+        that is empty along other axes (e.g. all channels sliced away
+        upstream) still flows so downstream consumers keep its cadence.
         """
         result = await self.processor.__acall__(message)
-        if result is not None and result.data.size > 0:
+        if result is not None and not is_empty_along(result, (self.SETTINGS.axis,)):
             yield self.OUTPUT_SIGNAL, result
 
 

src/ezmsg/sigproc/resample.py

@@ -19,6 +19,7 @@
 
 from .util.axisarray_buffer import HybridAxisArrayBuffer, HybridAxisBuffer
 from .util.buffer import UpdateStrategy
+from .util.message import has_samples_along
 
 
 class ResampleSettings(ez.Settings):
@@ -452,7 +453,11 @@ async def gen_resampled(self):
             self._wake.clear()
             while True:
                 result: AxisArray = next(self.processor)
-                if np.prod(result.data.shape) == 0:
+                # A real chunk has a nonzero resample axis; an empty axis or the
+                # pre-init null template (which lacks the axis entirely) means
+                # "nothing ready". A chunk that is empty only along other axes
+                # (e.g. zero channels) is still real output and must be published.
+                if not has_samples_along(result, self.SETTINGS.axis):
                     break
                 yield self.OUTPUT_SIGNAL, result
                 ref_out = self.processor.state.reference_output

src/ezmsg/sigproc/resampleconcat.py

@@ -35,6 +35,7 @@
 from .concat import ConcatProcessor, ConcatSettings
 from .resample import ResampleProcessor, ResampleSettings
 from .util.buffer import UpdateStrategy
+from .util.message import has_samples_along
 
 
 class ResampleConcatSettings(ez.Settings):
@@ -163,7 +164,10 @@ def _drain(self) -> typing.Iterator[AxisArray]:
         """
         while True:
             result = next(self.processor)
-            if result is None or np.prod(result.data.shape) == 0:
+            # None / a missing or empty resample axis means "nothing ready"; a
+            # chunk that is empty only along other axes (e.g. the concatenated
+            # feature axis) is still a real output and must be published.
+            if result is None or not has_samples_along(result, self.SETTINGS.axis):
                 return
             yield result
 

src/ezmsg/sigproc/util/message.py

@@ -1,17 +1,46 @@
-"""
-Backwards-compatible re-exports from ezmsg.baseproc.util.message.
+"""Message (AxisArray) utilities.
 
-New code should import directly from ezmsg.baseproc instead.
+Also re-exports sample-message symbols from ezmsg.baseproc.util.message for
+backwards compatibility; new code should import those directly from
+ezmsg.baseproc instead.
 """
 
+import typing
+
 from ezmsg.baseproc.util.message import (
     SampleMessage,
     SampleTriggerMessage,
     is_sample_message,
 )
+from ezmsg.util.messages.axisarray import AxisArray
 
 __all__ = [
     "SampleMessage",
     "SampleTriggerMessage",
+    "has_samples_along",
+    "is_empty_along",
     "is_sample_message",
 ]
+
+
+def is_empty_along(message: AxisArray, dims: typing.Iterable[str]) -> bool:
+    """True iff any of the named dims is present in ``message`` with zero length.
+
+    Publish gates use this instead of ``data.size == 0`` so a message that is
+    empty only along *other* axes — e.g. an upstream selection removed every
+    channel while time samples remain — still flows downstream, preserving the
+    stream's cadence for consumers that align or merge multiple sources.
+    Dims not present in the message are ignored.
+    """
+    return any(d in message.dims and message.data.shape[message.get_axis_idx(d)] == 0 for d in dims)
+
+
+def has_samples_along(message: AxisArray, dim: str) -> bool:
+    """True iff ``dim`` is present in ``message`` with nonzero length.
+
+    Stricter than ``not is_empty_along(...)``: the dim must exist. Drain loops
+    use this to decide whether a chunk is real output, so that a placeholder
+    lacking the axis entirely (e.g. ResampleProcessor's pre-init null template,
+    ``dims=[""]``) counts as "nothing ready" rather than a publishable chunk.
+    """
+    return dim in message.dims and message.data.shape[message.get_axis_idx(dim)] > 0

src/ezmsg/sigproc/window.py

@@ -20,6 +20,7 @@
     sliding_win_oneaxis,
 )
 
+from .util.message import is_empty_along
 from .util.profile import profile_subpub
 from .util.sparse import sliding_win_oneaxis as sparse_sliding_win_oneaxis
 
@@ -289,7 +290,10 @@ async def on_signal(self, message: AxisArray) -> typing.AsyncGenerator:
         xp = get_namespace(message.data)
         try:
             ret = self.processor(message)
-            if ret.data.size > 0:
+            # Swallow only when no complete windows (or, in pass-through mode, no
+            # samples) came out; emptiness along other axes (e.g. all channels
+            # sliced away upstream) still flows to preserve stream cadence.
+            if not is_empty_along(ret, (self.SETTINGS.newaxis or "win", self.SETTINGS.axis or ret.dims[0])):
                 if self.SETTINGS.newaxis is not None or self.SETTINGS.window_dur is None:
                     # Multi-win mode or pass-through mode.
                     yield self.OUTPUT_SIGNAL, ret

tests/unit/test_empty_gate.py

@@ -0,0 +1,129 @@
+"""Publish gates must swallow only time-empty results, not channel-empty ones.
+
+A unit that suppresses empty publishes (Downsample, BinnedAggregate, Window) or
+uses emptiness as a drain-loop terminator (ResampleUnit, ResampleConcat) should
+key on the time-like axis, not ``data.size``. A message that is empty only along
+other axes -- e.g. an upstream Slicer with ``on_empty="warn"`` removed every
+channel while time samples remain -- must still flow so downstream consumers
+keep the stream's cadence.
+"""
+
+import asyncio
+
+import numpy as np
+from ezmsg.util.messages.axisarray import AxisArray
+
+from ezmsg.sigproc.binned_aggregate import BinnedAggregate, BinnedAggregateSettings
+from ezmsg.sigproc.downsample import Downsample, DownsampleSettings
+from ezmsg.sigproc.resampleconcat import ResampleConcat, ResampleConcatSettings
+from ezmsg.sigproc.util.message import has_samples_along, is_empty_along
+from ezmsg.sigproc.window import Window, WindowSettings
+from tests.helpers.empty_time import make_msg
+
+
+def _null_template() -> AxisArray:
+    """ResampleProcessor's pre-init placeholder: no time axis at all."""
+    return AxisArray(data=np.array([]), dims=[""], axes={}, key="null")
+
+
+def _drive(unit, msg):
+    async def _run():
+        return [m async for m in unit.on_signal(msg)]
+
+    return asyncio.run(_run())
+
+
+def test_is_empty_along():
+    msg = make_msg(n_time=10, n_ch=0)
+    assert is_empty_along(msg, ("ch",))
+    assert not is_empty_along(msg, ("time",))
+    assert is_empty_along(msg, ("time", "ch"))
+    # Dims not present in the message are ignored.
+    assert not is_empty_along(msg, ("win",))
+    msg = make_msg(n_time=0, n_ch=3)
+    assert is_empty_along(msg, ("time",))
+    assert not is_empty_along(msg, ("ch",))
+    msg = make_msg(n_time=0, n_ch=0)
+    assert is_empty_along(msg, ("time",))
+    assert is_empty_along(msg, ("ch",))
+
+
+def test_has_samples_along():
+    assert has_samples_along(make_msg(n_time=10, n_ch=0), "time")
+    assert not has_samples_along(make_msg(n_time=10, n_ch=0), "ch")
+    assert not has_samples_along(make_msg(n_time=0, n_ch=3), "time")
+    # A message lacking the dim entirely (e.g. the resample pre-init null
+    # template) has no samples along it.
+    assert not has_samples_along(_null_template(), "time")
+
+
+def test_downsample_gate():
+    unit = Downsample(DownsampleSettings(axis="time", target_rate=50.0))
+    unit.create_processor()
+    # Zero channels but nonzero time: the (5, 0) result must be published.
+    published = _drive(unit, make_msg(n_time=10, n_ch=0, fs=100.0))
+    assert len(published) == 1
+    _, msg_out = published[0]
+    assert msg_out.data.shape == (5, 0)
+    # Zero time: swallowed.
+    assert _drive(unit, make_msg(n_time=0, n_ch=0, fs=100.0)) == []
+
+
+def test_binned_aggregate_gate():
+    unit = BinnedAggregate(BinnedAggregateSettings(axis="time", bin_duration=0.02))
+    unit.create_processor()
+    # 40 samples @ 1 kHz close 2 bins; zero channels must not suppress the publish.
+    published = _drive(unit, make_msg(n_time=40, n_ch=0, fs=1000.0))
+    assert len(published) == 1
+    _, msg_out = published[0]
+    assert msg_out.data.shape == (2, 0)
+    # 10 more samples close no bin: swallowed.
+    assert _drive(unit, make_msg(n_time=10, n_ch=0, fs=1000.0)) == []
+
+
+def test_window_gate():
+    settings = WindowSettings(axis="time", newaxis="win", window_dur=0.1, window_shift=0.1)
+    unit = Window(settings)
+    unit.create_processor()
+    # 30 samples @ 100 Hz -> 3 complete 10-sample windows; zero channels must
+    # not suppress the publish.
+    published = _drive(unit, make_msg(n_time=30, n_ch=0, fs=100.0))
+    assert len(published) == 1
+    _, msg_out = published[0]
+    assert msg_out.dims == ["win", "time", "ch"]
+    assert msg_out.data.shape == (3, 10, 0)
+    # 5 more samples complete no window: swallowed.
+    assert _drive(unit, make_msg(n_time=5, n_ch=0, fs=100.0)) == []
+
+
+class _StubProcessor:
+    """Stands in for ResampleConcatProcessor: yields canned chunks."""
+
+    def __init__(self, items):
+        self._items = iter(items)
+
+    def __next__(self):
+        return next(self._items)
+
+
+def test_resampleconcat_drain_gate():
+    unit = ResampleConcat(ResampleConcatSettings())
+    # A chunk that is empty along the concatenated feature axis is a real
+    # output; only an empty resample ("time") axis terminates the drain.
+    unit.processor = _StubProcessor(
+        [
+            make_msg(n_time=5, n_ch=0),
+            make_msg(n_time=0, n_ch=3),
+            make_msg(n_time=5, n_ch=3),  # must NOT be reached
+        ]
+    )
+    drained = list(unit._drain())
+    assert len(drained) == 1
+    assert drained[0].data.shape == (5, 0)
+    # None also terminates the drain.
+    unit.processor = _StubProcessor([None])
+    assert list(unit._drain()) == []
+    # The pre-init null template (no time axis at all) also terminates the
+    # drain instead of being published.
+    unit.processor = _StubProcessor([_null_template()])
+    assert list(unit._drain()) == []