Streaming updates#

ndv can be used to visualize data that is continuously updated, such as images from a camera or a live data stream. The following document shows some examples of such implementation.

Basic streaming, with no history#

To visualize a live data stream, simply create an ndv.ArrayViewer controller with an empty buffer matching your data shape. Then, when new data is available, update the buffer in place with the new data. Calling update() on the ArrayDisplayModel.current_index will force the display to fetch your new data:

examples/streaming.py

# /// script
# dependencies = [
#     "ndv[pyqt,pygfx]",
#     "imageio[tifffile]",
# ]
# ///
"""Example of streaming data."""

import numpy as np

import ndv

# some data we're going to stream (as if it was coming from a camera)
data = ndv.data.cells3d()[:, 1]

# a buffer to hold the current frame in the viewer
buffer = np.zeros_like(data[0])
viewer = ndv.ArrayViewer(buffer)
viewer.show()


# function that will be called after the app is running
def stream(nframes: int = len(data) * 4) -> None:
    for i in range(nframes):
        # iterate over the data, update the buffer *in place*,
        buffer[:] = data[i % len(data)]
        # and update the viewer index to redraw
        viewer.display_model.current_index.update()
        ndv.process_events()  # force viewer updates for this example


ndv.call_later(200, stream)
ndv.run_app()

Streaming, remembering the last N frames#

To visualize a live data stream while keeping the last N frames in memory, you can use the ndv.models.RingBuffer class. It offers a convenient append() method to add new data, and takes care of updating the "apparent" shape of the data (as far as the viewer is concerned):

examples/streaming_with_history.py

# /// script
# dependencies = [
#     "ndv[pyqt,pygfx]",
#     "imageio[tifffile]",
# ]
# ///
"""Example of streaming data, retaining the last N frames."""

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
N = 50
rb = RingBuffer(max_capacity=N, 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
    for n, plane in enumerate(data):
        rb.append(plane)
        # and update the viewer index to redraw (and possibly move the slider)
        viewer.display_model.current_index.update({0: max(n, N - 1)})

        ndv.process_events()  # force viewer updates for this example


ndv.call_later(200, stream)
ndv.run_app()