feat: add circular/ring buffer and data wrapper, that can be used to facilitate a streaming viewer

examples/streaming.py

@@ -0,0 +1,33 @@
+# /// script
+# dependencies = [
+#     "ndv[pyqt,pygfx]",
+#     "imageio[tifffile]",
+# ]
+# ///
+import ndv
+from ndv.models import RingBuffer
+
+# some data we're going to stream (as if it was coming from a camera)
+data = ndv.data.cells3d()[:, 1]
+
+# a ring buffer to hold the data as it comes in
+# the ring buffer is a circular buffer that holds the last N frames
+rb = RingBuffer(max_capacity=20, dtype=(data.dtype, data.shape[-2:]))
+
+# pass the ring buffer to the viewer
+viewer = ndv.ArrayViewer(rb)
+viewer.show()
+
+
+# function that will be called after the app is running
+def stream() -> None:
+    # iterate over the data, add it to the ring buffer, and update the viewer index
+    for n, plane in enumerate(data):
+        rb.append(plane)
+        viewer.display_model.current_index.update({0: max(n, 20 - 1)})
+
+        ndv.process_events()  # force viewer updates for this example
+
+
+ndv.call_later(200, stream)
+ndv.run_app()

src/ndv/__init__.py

@@ -11,13 +11,21 @@
 from .controllers import ArrayViewer
 from .models import DataWrapper
 from .util import imshow
-from .views import run_app, set_canvas_backend, set_gui_backend
+from .views import (
+    call_later,
+    process_events,
+    run_app,
+    set_canvas_backend,
+    set_gui_backend,
+)
 
 __all__ = [
     "ArrayViewer",
     "DataWrapper",
+    "call_later",
     "data",
     "imshow",
+    "process_events",
     "run_app",
     "set_canvas_backend",
     "set_gui_backend",

src/ndv/controllers/_array_viewer.py

@@ -557,6 +557,9 @@
 
         display_model = self._data_model.display
         for key, data in response.data.items():
+            if data.size == 0:
+                # no data for this channel
+                continue
             if (lut_ctrl := self._lut_controllers.get(key)) is None:
                 if key is None:
                     model = display_model.default_lut

src/ndv/models/__init__.py

@@ -2,7 +2,7 @@
 
 from ._array_display_model import ArrayDisplayModel, ChannelMode
 from ._base_model import NDVModel
-from ._data_wrapper import DataWrapper
+from ._data_wrapper import DataWrapper, RingBufferWrapper
 from ._lut_model import (
     ClimPolicy,
     ClimsManual,
@@ -11,6 +11,7 @@
     ClimsStdDev,
     LUTModel,
 )
+from ._ring_buffer import RingBuffer
 from ._roi_model import RectangularROIModel
 
 __all__ = [
@@ -25,4 +26,6 @@
     "LUTModel",
     "NDVModel",
     "RectangularROIModel",
+    "RingBuffer",
+    "RingBufferWrapper",
 ]

src/ndv/models/_data_wrapper.py

@@ -15,6 +15,8 @@
 import numpy.typing as npt
 from psygnal import Signal
 
+from ._ring_buffer import RingBuffer
+
 if TYPE_CHECKING:
     from collections.abc import Container, Iterator
     from typing import Any, TypeAlias, TypeGuard
@@ -498,3 +500,62 @@
         if (torch := sys.modules.get("torch")) and isinstance(obj, torch.Tensor):
             return True
         return False
+
+
+class RingBufferWrapper(DataWrapper[RingBuffer]):
+    """Wrapper for ring buffer objects."""
+
+    def __init__(
+        self,
+        max_capacity: int | RingBuffer,
+        dtype: npt.DTypeLike = None,
+        *,
+        allow_overwrite: bool = True,
+    ):
+        if isinstance(max_capacity, RingBuffer):
+            if dtype is not None:  # pragma: no cover
+                raise ValueError(
+                    "Cannot specify dtype when passing an existing RingBuffer."
+                )
+            self._ring = max_capacity
+        else:
+            if dtype is None:
+                dtype = float
+            self._ring = RingBuffer(
+                max_capacity=max_capacity, dtype=dtype, allow_overwrite=allow_overwrite
+            )
+        self._ring.resized.connect(self.dims_changed)
+        super().__init__(self._ring)
+
+    @property
+    def dims(self) -> tuple[int, ...]:
+        """Return the dimensions of the data."""
+        return tuple(range(len(self._ring.shape)))
+
+    @property
+    def coords(self) -> Mapping:
+        """Return the coordinates for the data."""
+        shape = self._ring.shape
+        return {i: range(s) for i, s in enumerate(shape)}
+
+    @classmethod
+    def supports(cls, obj: Any) -> TypeGuard[np.ndarray]:
+        if isinstance(obj, RingBuffer):
+            return True
+        return False
+
+    def append(self, value: npt.ArrayLike) -> None:
+        """Append a value to the right end of the buffer."""
+        self._ring.append(value)
+
+    def appendleft(self, value: npt.ArrayLike) -> None:
+        """Append a value to the left end of the buffer."""
+        self._ring.appendleft(value)
+
+    def pop(self) -> np.ndarray:
+        """Pop a value from the right end of the buffer."""
+        return self._ring.pop()
+
+    def popleft(self) -> np.ndarray:
+        """Pop a value from the left end of the buffer."""
+        return self._ring.popleft()