ezmsg.dashboard.backend.stream_frames#
Staging and wire-encoding for live sample data on its way to the browser.
Everything here is plain numpy with no ezmsg, asyncio or FastAPI in sight, which is what makes the interesting parts – ring wraparound, overflow accounting, and envelope reduction – testable without a running graph.
Three ideas carry the module:
A ring, not a queue. A tap reads messages at whatever rate the publisher sends them and the browser reads frames at its refresh rate; those are unrelated clocks. The ring absorbs the difference. It is deliberately small – it stages one pump interval plus slack, not the display history, which lives in the browser – so sizing it is about jitter, not about how much past the user wants to see.
Overflow is reported, never hidden. When a reader falls far enough behind
that the writer laps it, the samples in between are gone. Silently handing over
whatever is currently in the ring would draw a seamless trace across a gap that
really happened, which is worse than useless on a plot someone is using to
decide whether a pipeline is healthy. SampleRing.read() says so instead.
One payload shape. Everything a sweep draws goes out as (n_out,
n_channels, 2) min/max pairs, whether it was decimated, passed through, or
arrived as a native (min, max) envelope from an upstream aggregator. A single
shape means a single code path in the renderer, and the degenerate case
(min == max) draws as the polyline it is.
That shape is also, exactly, the memory layout of an RG32F texture with one
row per column and one texel per channel, which is how the browser holds the
ring. A frame therefore uploads with texSubImage2D untouched – no transpose,
no per-sample work on either side of the wire.
Module Attributes
Bytes the header length prefix occupies, little-endian |
Functions
- decode_binary_frame(frame)[source]#
Inverse of
encode_binary_frame(), for tests and for debugging.
- encode_binary_frame(header, payload)[source]#
[u32 header_len][utf-8 JSON header][float32 payload].Binary rather than JSON because the payload is the bulk of the traffic and JSON would roughly quadruple it while forcing the browser to parse numbers one at a time; this way the client wraps the tail in a
Float32Arraywith no per-sample work at all. The header stays JSON because it is small, it changes shape as modes are added, and being able to read a frame in devtools is worth more than the bytes.
- envelope_pairs(block, max_columns)[source]#
Reduce
(n, n_channels, width)samples to(n_out, n_channels, 2).The trailing pair is what makes a frame an
RG32Ftexture upload on the browser side: one row per output column, one texel per channel, red and green holding the column’s minimum and maximum.Decimating here rather than in the browser is the whole reason a 30 kHz stream fits down a websocket: the client only ever has as many columns as it has pixels, so the wire cost is set by the plot’s width instead of by the sample rate.
Min/max is the only correct reduction for this. Subsampling drops the extremes, which on neural data means the spikes – the part someone opened the plot to look at – vanish at exactly the zoom levels where they matter.
NaNs propagate into the column that contains them, which draws as a gap. That is deliberate: a stream producing NaNs should look broken.
- ring_capacity_for(*, srate, n_channels, width, seconds, max_bytes, minimum=256)[source]#
How many samples of slack to stage, bounded by a memory ceiling.
The ring only has to cover one pump interval plus jitter, so
secondsis small by design.max_bytesis what stops a 30 kHz 256-channel stream from turning a modest duration into hundreds of megabytes; hitting it costs slack, not correctness, since running short of slack surfaces as a reported overflow.
Classes
- class RingCursor(total_read=0)[source]#
Bases:
objectOne reader’s position in a
SampleRing.Held per websocket client rather than per ring so that several browsers watching one topic share a single subscriber without stealing each other’s samples.
- Parameters:
total_read (int)
- class SampleRing(capacity, n_channels, width)[source]#
Bases:
objectFixed-capacity circular buffer of
(n_channels, width)samples.widthis the stream’s native per-sample tuple: 1 for an ordinary signal, 2 when an upstream aggregator already reduced it to (min, max). Keeping the native width here rather than expanding everything to pairs on write is what keeps the hot path – one slice assignment per message – cheap.Positions are tracked as monotonic totals rather than an index plus a wrap counter. The two are equivalent, but totals make “how far behind is this reader” a subtraction instead of a case analysis, and there is no fixed-size header forcing the counter to wrap the way there is across a shared-memory boundary.
- write(block)[source]#
Append
(n_samples, n_channels, width)samples, wrapping as needed.A block longer than the whole ring keeps only its tail: the samples it would have overwritten within this one call are already unreachable, so copying them in first would be work no reader could ever observe.
- read(cursor, *, max_samples=None)[source]#
Take everything
cursorhas not seen yet.Returns
(block, overflowed, first_sample_index).blockis always a copy: the caller is about to await a socket write, by which point the writer may well have overwritten the region it came from.first_sample_indexcounts from the first sample this ring ever accepted, which is what lets the client put a real time axis under the data even across a gap.