ezmsg.tools.plot.describe#

Reading a stream’s shape well enough to plot it.

Every consumer that puts ezmsg data on a phosphor widget has to answer the same questions – how many channels, at what rate, what are they called, and is this a signal or an envelope – and each has been answering them slightly differently. This module answers them once.

It deliberately does not import phosphor or Qt: the inputs are dims, axes and attrs, and the outputs are plain numbers and arrays. That keeps it usable from a topic subscriber, from a shared-memory mirror, and from a test with neither.

Functions

describe_axisarray(msg, *, time_axis='time', label_fields=('label',))[source]#

Describe a stream from one of its AxisArray messages.

Parameters:
Return type:

StreamShape

describe_mirror(mirror, *, time_axis='time', label_fields=('label',))[source]#

Describe a stream from a connected EZShmMirror.

Returns None until the writer has published both a valid buffer header and its metadata – the two arrive independently, and a description built from only one of them would be missing either the shape or the names.

Parameters:
Return type:

StreamShape | None

flatten_for_plot(data, shape)[source]#

Reshape a block to what a plot’s push_data expects.

(n_samples, n_channels, k) when the stream carries a k-wide metric tuple, (n_samples, n_channels) otherwise, with any other dimensions folded into channels.

The metric case is the reason this exists. Folding a (time, ch, 2) block into (time, ch * 2) – which is what a naive reshape does – renders as twice as many traces, alternating the two metrics, with every channel label off by a factor of two. It looks like data, so nothing complains.

Parameters:
Return type:

ndarray

metric_axis(dims, axes)[source]#

Describe the trailing per-sample tuple, or None if there is not one.

Recognised by labels, not by name or width. The name only narrows the search; the labels are what distinguish a (min, max) envelope from a (mean, std) dispersion pair, which is the same shape and must not be drawn the same way.

Returns a spec for any tuple in METRIC_KINDS, including ones no renderer here supports yet – describing a stream is not the same as being able to draw it, and a caller that only wants to know what arrived should not have to catch an exception. See require_sweep_renderable() for the capability check.

Parameters:
Return type:

MetricSpec | None

require_sweep_renderable(shape)[source]#

Raise if a sweep plot cannot draw this stream’s metric axis.

Raises:

UnsupportedMetricError – for a recognised metric a sweep cannot draw.

Parameters:

shape (StreamShape)

Return type:

None

stream_axis(msg, *fallbacks)[source]#

Which dimension of msg the stream accumulates along.

Prefers the producer’s own declaration (stream_dim) and falls back to the first of fallbacks the message actually has, which is what these tools did before the field existed.

The fallback is a guess, and "time" is the wrong guess downstream of a windowing stage: a (win, time, ch) message has a time dimension, but it is the within-window lag, so a sweep plot keyed on it draws each window’s interior along the x-axis and treats the windows as channels – and reads an offset that does not advance with the stream.

Parameters:
Return type:

str | None

Classes

class MetricSpec(axis, labels, kind)[source]#

Bases: NamedTuple

A trailing per-sample tuple: what it is called and what it holds.

Parameters:
axis: str#

Name of the trailing axis.

labels: tuple[str, ...]#

Its coordinate values, lowercased, in order.

kind: str#

The entry in METRIC_KINDS these labels matched.

class StreamShape(n_channels, srate, channel_labels, metric, unit)[source]#

Bases: NamedTuple

What a plot needs to know about an incoming stream.

Parameters:
n_channels: int#

Channels, excluding any envelope axis.

srate: float#

Samples per second of the pushed stream. For an envelope this is the bucket rate, not the rate before decimation – which is what a sweep buffer must be sized with, or its ring is longer than the data arriving to fill it.

channel_labels: list[str] | None#

One name per channel, or None if the stream does not say.

metric: MetricSpec | None#

The trailing per-sample tuple, if the stream carries one.

unit: str | None#

The signal’s amplitude unit, if it declares one.

property envelope: bool#

Whether each sample carries a (min, max) pair – phosphor’s envelope.

Exceptions

exception UnsupportedMetricError[source]#

Bases: NotImplementedError

A recognised metric axis that no renderer here can draw yet.

stream_axis(msg, *fallbacks)[source]#

Which dimension of msg the stream accumulates along.

Prefers the producer’s own declaration (stream_dim) and falls back to the first of fallbacks the message actually has, which is what these tools did before the field existed.

The fallback is a guess, and "time" is the wrong guess downstream of a windowing stage: a (win, time, ch) message has a time dimension, but it is the within-window lag, so a sweep plot keyed on it draws each window’s interior along the x-axis and treats the windows as channels – and reads an offset that does not advance with the stream.

Parameters:
Return type:

str | None

class MetricSpec(axis, labels, kind)[source]#

Bases: NamedTuple

A trailing per-sample tuple: what it is called and what it holds.

Parameters:
axis: str#

Name of the trailing axis.

labels: tuple[str, ...]#

Its coordinate values, lowercased, in order.

kind: str#

The entry in METRIC_KINDS these labels matched.

class StreamShape(n_channels, srate, channel_labels, metric, unit)[source]#

Bases: NamedTuple

What a plot needs to know about an incoming stream.

Parameters:
n_channels: int#

Channels, excluding any envelope axis.

srate: float#

Samples per second of the pushed stream. For an envelope this is the bucket rate, not the rate before decimation – which is what a sweep buffer must be sized with, or its ring is longer than the data arriving to fill it.

channel_labels: list[str] | None#

One name per channel, or None if the stream does not say.

metric: MetricSpec | None#

The trailing per-sample tuple, if the stream carries one.

unit: str | None#

The signal’s amplitude unit, if it declares one.

property envelope: bool#

Whether each sample carries a (min, max) pair – phosphor’s envelope.

exception UnsupportedMetricError[source]#

Bases: NotImplementedError

A recognised metric axis that no renderer here can draw yet.

describe_axisarray(msg, *, time_axis='time', label_fields=('label',))[source]#

Describe a stream from one of its AxisArray messages.

Parameters:
Return type:

StreamShape

describe_mirror(mirror, *, time_axis='time', label_fields=('label',))[source]#

Describe a stream from a connected EZShmMirror.

Returns None until the writer has published both a valid buffer header and its metadata – the two arrive independently, and a description built from only one of them would be missing either the shape or the names.

Parameters:
Return type:

StreamShape | None

flatten_for_plot(data, shape)[source]#

Reshape a block to what a plot’s push_data expects.

(n_samples, n_channels, k) when the stream carries a k-wide metric tuple, (n_samples, n_channels) otherwise, with any other dimensions folded into channels.

The metric case is the reason this exists. Folding a (time, ch, 2) block into (time, ch * 2) – which is what a naive reshape does – renders as twice as many traces, alternating the two metrics, with every channel label off by a factor of two. It looks like data, so nothing complains.

Parameters:
Return type:

ndarray

metric_axis(dims, axes)[source]#

Describe the trailing per-sample tuple, or None if there is not one.

Recognised by labels, not by name or width. The name only narrows the search; the labels are what distinguish a (min, max) envelope from a (mean, std) dispersion pair, which is the same shape and must not be drawn the same way.

Returns a spec for any tuple in METRIC_KINDS, including ones no renderer here supports yet – describing a stream is not the same as being able to draw it, and a caller that only wants to know what arrived should not have to catch an exception. See require_sweep_renderable() for the capability check.

Parameters:
Return type:

MetricSpec | None

require_sweep_renderable(shape)[source]#

Raise if a sweep plot cannot draw this stream’s metric axis.

Raises:

UnsupportedMetricError – for a recognised metric a sweep cannot draw.

Parameters:

shape (StreamShape)

Return type:

None