ezmsg.dashboard.backend.services.stream_tap#

Watching what a publisher is actually sending, from the dashboard backend.

The backend already holds a live GraphContext, so it can create an ezmsg Subscriber on any topic directly: in its own event loop, with no unit injected into the graph and no edge added to the topology. That is the whole mechanism. There is no shared-memory hop and no helper process, because the thing that forces those on the desktop tools – a renderer living in a process that cannot be an ezmsg node – does not apply when the renderer is a browser at the other end of a socket the backend is already serving.

Two properties of ezmsg’s Subscriber are what make it safe to point this at a production graph:

  • leaky=True drops notifications instead of applying backpressure, and releases the publisher’s buffer for each one it drops (see Subscriber._handle_dropped_notification). A wedged browser tab cannot stall the graph it is watching. This is the single property worth checking first in any “tap a live pipeline” design, and ezmsg provides it at the client layer.

  • recv_zero_copy() yields the message itself, so nothing is pickled or deep-copied on the way in. We reduce inside the context manager and let go.

What leakiness costs is contiguity: a subscriber that falls behind loses whole messages, and a sweep drawn straight from those would join two ends of a gap into a smooth, entirely fictional trace. Hence the ring in stream_frames, written at message rate and read at frame rate, which reports the gap instead of hiding it.

ezmsg-tools supplies the AxisArray-to-plottable reduction (ezmsg.tools.plot.describe). It is an optional dependency: without it the message inspector still works on every topic – it needs nothing but the message – and only the plotting modes are unavailable. Nothing else in the dashboard imports it.

Module Attributes

LEAKY_MAX_QUEUE

How many messages a tap's subscriber may fall behind before ezmsg starts dropping them for us.

RING_SECONDS

Seconds of slack the ring stages between the recv loop and the pump.

RING_MAX_BYTES

Ceiling on one ring's footprint.

MAX_DRAWN_CHANNELS

a wider one is plotted too, a window at a time, and scrolled.

MAX_COLUMNS

Cap on payload columns per frame regardless of what a client asks for.

DEFAULT_WINDOW_SECONDS

Seconds of signal a sweep window shows by default.

SWEEP

View kinds a stream can be, decided by the data rather than by the client.

AUTO

What a client may ask for.

Functions

as_numpy(data)[source]#

A numpy view of a message’s payload, whatever framework produced it.

ezmsg does not require an AxisArray’s data to be a numpy array, and a pipeline running on MLX – or torch, or jax – hands over its own array type. Those satisfy numpy’s buffer or __array__ protocol, so this is usually a view and not a conversion: on Apple silicon’s unified memory, np.asarray of an MLX array aliases the very same buffer.

That aliasing is worth stating plainly, because writing through the result would write into the publisher’s array. Everything downstream of here either copies into the ring or copies explicitly; none of it writes in place.

Two things need handling beyond the happy path:

  • dtypes numpy has no equivalent for. bfloat16 is the one that matters: it fails the buffer protocol outright, and no dtype hint gets around it. See _bfloat16_via_bits(), and note in particular that the obvious fix – asking the framework to cast – silently corrupts data that arrived over the wire.

  • laziness. MLX does not evaluate until something reads the buffer, so this is where a publisher’s pending work gets forced. It is cheap here because the message had to be materialised to be delivered at all.

Deliberately does not fall back to running the producing framework’s compute on the payload. That is what the bfloat16 note above is about, and the failure it produces is invisible.

Raises:

StreamTapError – if the payload cannot be read as numeric data. The caller turns that into a plot error and leaves the message inspector working, which is the useful half for a stream nothing can draw.

Return type:

ndarray

Parameters:

data (Any)

availability_payload()[source]#

Whether plotting modes can run here, for the frontend to gate its UI on.

Return type:

dict[str, Any]

Classes

class StreamTap(topic, subscriber_factory, *, subscriber_release=None, ring_seconds=2.0, ring_max_bytes=33554432)[source]#

Bases: object

One subscriber on one topic, shared by every client watching it.

Sharing matters more than it looks: each extra Subscriber on a topic is another consumer the publisher has to serve and another set of buffers it has to hold, so opening one per browser tab would make the dashboard’s cost to the graph scale with how many people are looking at it.

Parameters:

  • topic (str)

  • subscriber_factory (Callable[..., Any])

  • subscriber_release (Callable[[Any], None] | None)

  • ring_seconds (float)

  • ring_max_bytes (int)

__init__(topic, subscriber_factory, *, subscriber_release=None, ring_seconds=2.0, ring_max_bytes=33554432)[source]#
Parameters:
Return type:

None

property topic: str#
property ref_count: int#
async start()[source]#
Return type:

None

async stop()[source]#
Return type:

None

acquire()[source]#
Return type:

None

release()[source]#
Return type:

None

snapshot_rate_hz()[source]#

Message rate over the recent window, or None before it means anything.

Return type:

float | None

status_payload()[source]#
Return type:

dict[str, Any]

inspect_payload()[source]#
Return type:

dict[str, Any]

property description: _Description | None#
property description_generation: int#
pending_samples(cursor)[source]#

How many samples this cursor has not read yet.

Lets a client decide whether a read is worth making before it makes one, which is what keeps the frame rate from dictating the time base.

Return type:

int

Parameters:

cursor (RingCursor)

read_sweep(cursor, target_columns)[source]#

(pairs, overflowed, first_sample_index, n_samples) or None if idle.

target_columns is how many columns this block should occupy, which the caller derives from the window’s time base – not the width of the plot. Passing the plot width here instead would make every frame fill the whole window regardless of how much time it covered.

Return type:

tuple[ndarray, bool, int, int] | None

Parameters:
read_latest(seen_generation)[source]#

The most recent spectrum/scatter frame, if it is newer than seen.

Return type:

tuple[ndarray, int] | None

Parameters:

seen_generation (int)

reset_cursor(cursor)[source]#

Point a cursor at the newest sample, discarding the backlog.

Used when a client first attaches or after a shape change: the samples already in the ring belong to the previous plot, and prepending them to a freshly-configured one would draw the old stream’s tail on the new stream’s axes.

Return type:

None

Parameters:

cursor (RingCursor)

class StreamTapClient(tap, mode, max_columns, window_seconds=2.0, cursor=<factory>, seen_description_generation=-1, seen_latest_generation=-1, column_remainder=0.0)[source]#

Bases: object

One websocket’s view of a tap: what it wants, and where it has read to.

Parameters:
tap: StreamTap#
mode: str#
max_columns: int#
window_seconds: float = 2.0#
cursor: RingCursor#
seen_description_generation: int = -1#
seen_latest_generation: int = -1#
column_remainder: float = 0.0#

Fractional column carried between frames. Rounding each frame’s column count independently would let the time base drift against the sample clock, slowly and invisibly.

property topic: str#
set_max_columns(max_columns)[source]#
Return type:

None

Parameters:

max_columns (int)

set_window_seconds(window_seconds)[source]#
Return type:

None

Parameters:

window_seconds (float)

effective_columns(srate)[source]#

How many columns the requested window is actually worth.

The plot’s pixel budget is a ceiling, not a target. A slow stream cannot fill it: two seconds of a 100 Hz signal is 200 samples, and a column can never stand for less than one sample.

Spreading those 200 samples across a 2000-column plot is what the earlier version did, and it silently showed twenty seconds while the caption said two – with the window control apparently dead, because every window short enough to matter produced the same one-sample column. Shrinking the column count instead keeps the window honest; the plot simply draws fewer, wider columns, which is all the data there is.

Return type:

int

Parameters:

srate (float)

samples_per_column(srate)[source]#

How many source samples one output column stands for.

A stream with no sample rate advances one column per message, so there is nothing to average and this is 1.

Return type:

float

Parameters:

srate (float)

shown_seconds(srate)[source]#

How much time the plot will actually span.

Usually the requested window, but not always: a very slow stream runs into the two-column floor, and half a sample of a 1 Hz signal is not something anything can draw. The client captions from this rather than from the request, so the axis never claims a span the data does not have.

Return type:

float | None

Parameters:

srate (float)

take_description()[source]#

The stream description, but only once per change.

Return type:

dict[str, Any] | None

take_frame()[source]#

The next data frame for this client, or None if there is nothing new.

Returns the header and payload rather than sending them, so the framing and the mode dispatch can be tested without a websocket on either end.

Callers must consume take_description() first: the header refers to a description generation, and a frame that arrives before the description it belongs to would be drawn on the previous stream’s axes.

Return type:

tuple[dict[str, Any], ndarray] | None

__init__(tap, mode, max_columns, window_seconds=2.0, cursor=<factory>, seen_description_generation=-1, seen_latest_generation=-1, column_remainder=0.0)#
Parameters:
Return type:

None

class StreamTapManager(subscriber_factory, *, subscriber_release=None)[source]#

Bases: object

Refcounted registry of live taps, one per topic.

Parameters:

  • subscriber_factory (Callable[..., Any])

  • subscriber_release (Callable[[Any], None] | None)

__init__(subscriber_factory, *, subscriber_release=None)[source]#
Parameters:
Return type:

None

property active_topics: list[str]#
async shutdown()[source]#
Return type:

None

client(topic, *, mode, max_columns, window_seconds=2.0)[source]#
Return type:

AsyncIterator[StreamTapClient]

Parameters:

Exceptions

exception StreamTapError[source]#

Bases: RuntimeError

A tap could not be opened, or cannot render the stream it found.

exception StreamTapUnavailableError[source]#

Bases: StreamTapError

Taps are not available at all – no GraphContext, or no ‘viz’ extra.