From df7cb3b99c8fdc5e330c1b7f31147725c31e74fc Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 14:17:08 -0400 Subject: [PATCH 01/10] Add a shared adapter for putting ezmsg streams on phosphor plots Three places were doing this independently -- sigmon, viewer, and a downstream shmem viewer -- and each had a different subset of it right. The one that matters: both CLIs computed channel count as `data.size // n_samples` and reshaped to match, which turns a (time, ch, 2) min/max envelope into twice as many traces alternating lower and upper bounds, against labels now off by a factor of two. It renders, so nothing complains. ezmsg.tools.plot.describe is the pure half: given dims, axes and attrs, work out channel count, rate, labels, unit, and whether this is an envelope. It imports neither Qt nor phosphor, so it serves a topic subscriber, a shmem mirror, and a test with neither, and it is where the reshaping rule lives. An envelope is identified by its axis *labels*, not its name or width. A 2-wide trailing axis could as easily be (mean, std), which would be nonsense drawn as bounds, so the coordinates have to say min and max. The name only narrows the search, since ezmsg-sigproc's BinnedAggregate lets callers rename it. ezmsg.tools.plot.ShmemSweepWidget is the Qt half: mirror a ring, build the plot on first data, poll at the render cadence, and reconfigure in place when the stream changes shape rather than rebuilding -- a rebuild flashes the plot every time a user narrows their channel selection. It rebuilds only when the rate or envelope mode changes, since those invalidate the buffer's whole layout. The rate it hands the buffer is the ring header's, which for an envelope is already the bucket rate. That is the one that is easy to get wrong: sizing from the pre-decimation rate makes the ring `factor` times longer than the data arriving to fill it, and the sweep sits mostly empty. sigmon and viewer both move onto describe, which is what fixes the envelope handling in each and is the evidence the abstraction is not just one consumer's widget with the serial numbers filed off. 13 tests, none needing Qt or a GPU. --- pyproject.toml | 4 + src/ezmsg/tools/plot/__init__.py | 29 ++++ src/ezmsg/tools/plot/describe.py | 209 +++++++++++++++++++++++++ src/ezmsg/tools/plot/shmem_sweep.py | 233 ++++++++++++++++++++++++++++ src/ezmsg/tools/sigmon/cli.py | 26 ++-- src/ezmsg/tools/viewer/cli.py | 35 +++-- tests/test_plot_describe.py | 169 ++++++++++++++++++++ 7 files changed, 675 insertions(+), 30 deletions(-) create mode 100644 src/ezmsg/tools/plot/__init__.py create mode 100644 src/ezmsg/tools/plot/describe.py create mode 100644 src/ezmsg/tools/plot/shmem_sweep.py create mode 100644 tests/test_plot_describe.py diff --git a/pyproject.toml b/pyproject.toml index 7c88fcd..466cdd4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,9 @@ perfmon = [ "dash-bootstrap-components>=1.6.0", "ezmsg-baseproc>=1.1.0", ] +# NOTE: ezmsg.tools.plot needs phosphor's envelope support, set_channel_labels, +# set_channel_labels_visible and SweepConfig.max_fps. Bump this floor to the +# release that carries them before publishing. sigmon = [ "PySide6>=6.7", "pygraphviz>=1.14", @@ -99,5 +102,6 @@ known-first-party = ["ezmsg.tools"] known-third-party = ["ezmsg"] [tool.uv.sources] +phosphor = { path = "../phosphor", editable = true } # Uncomment to use development version of ezmsg from git #ezmsg = { git = "https://github.com/ezmsg-org/ezmsg.git", branch = "dev" } diff --git a/src/ezmsg/tools/plot/__init__.py b/src/ezmsg/tools/plot/__init__.py new file mode 100644 index 0000000..93710a1 --- /dev/null +++ b/src/ezmsg/tools/plot/__init__.py @@ -0,0 +1,29 @@ +"""Putting ezmsg streams onto phosphor plots. + +:mod:`.describe` is the pure half -- given dims, axes and attrs, work out what +is being plotted -- and imports neither Qt nor phosphor, so it is usable from a +topic subscriber, a shared-memory mirror, or a test with neither. +:mod:`.shmem_sweep` is the Qt widget built on it. + +Requires the ``viewer`` or ``sigmon`` extra (PySide6 and phosphor). Importing +this package pulls in Qt, so import :mod:`.describe` directly if that is all +you need. +""" + +from .describe import ( + StreamShape, + describe_axisarray, + describe_mirror, + envelope_axis, + flatten_for_plot, +) +from .shmem_sweep import ShmemSweepWidget + +__all__ = [ + "ShmemSweepWidget", + "StreamShape", + "describe_axisarray", + "describe_mirror", + "envelope_axis", + "flatten_for_plot", +] diff --git a/src/ezmsg/tools/plot/describe.py b/src/ezmsg/tools/plot/describe.py new file mode 100644 index 0000000..b251f42 --- /dev/null +++ b/src/ezmsg/tools/plot/describe.py @@ -0,0 +1,209 @@ +"""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. +""" + +from __future__ import annotations + +import typing + +import numpy as np + +from ..chmeta import channel_names + +__all__ = [ + "ENVELOPE_AXIS_CANDIDATES", + "StreamShape", + "describe_axisarray", + "describe_mirror", + "envelope_axis", + "flatten_for_plot", +] + +# Axis names an upstream min/max decimator might use for its (min, max) pair. +# ezmsg-sigproc's BinnedAggregate calls it "metric" by default but the name is a +# setting, so recognising a couple of obvious alternatives costs nothing. +ENVELOPE_AXIS_CANDIDATES = ("metric", "minmax", "bound") + +# Aggregation-function labels that make an axis an envelope rather than, say, +# a (mean, std) pair -- which is 2-wide and named the same way but must not be +# drawn as an upper and lower bound. +_ENVELOPE_LABELS = ("min", "max") + + +class StreamShape(typing.NamedTuple): + """What a plot needs to know about an incoming stream.""" + + 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.""" + + envelope: bool + """Whether each sample carries a (min, max) pair on a trailing axis.""" + + unit: str | None + """The signal's amplitude unit, if it declares one.""" + + +def envelope_axis(dims: typing.Sequence[str], axes: typing.Mapping[str, typing.Any]) -> str | None: + """Name of the trailing (min, max) axis, or None if this is a plain signal. + + Identified by its *labels* rather than its name or width. A length-2 + trailing axis could as easily be (mean, std), which would be nonsense to + draw as an envelope, so the coordinate values have to say ``min`` and + ``max``. The name is only used to narrow the search. + """ + if not dims: + return None + name = dims[-1] + if name not in ENVELOPE_AXIS_CANDIDATES: + return None + data = _axis_data(axes.get(name)) + if data is None or len(data) != 2: + return None + return name if tuple(str(v).lower() for v in data) == _ENVELOPE_LABELS else None + + +def _axis_data(axis: typing.Any) -> np.ndarray | None: + """Coordinate values of an axis given either as a dict or an ezmsg object.""" + if axis is None: + return None + if isinstance(axis, dict): + data = axis.get("data") + else: + data = getattr(axis, "data", None) + return None if data is None else np.asarray(data) + + +def _axis_gain(axis: typing.Any) -> float | None: + if axis is None: + return None + gain = axis.get("gain") if isinstance(axis, dict) else getattr(axis, "gain", None) + return None if gain in (None, 0) else float(gain) + + +def _describe( + dims: typing.Sequence[str], + axes: typing.Mapping[str, typing.Any], + attrs: typing.Mapping[str, typing.Any], + shape: typing.Sequence[int], + srate: float | None, + *, + time_axis: str = "time", + label_fields: typing.Sequence[str] = ("label",), +) -> StreamShape: + dims = list(dims) + env_axis = envelope_axis(dims, axes) + + # Channel count is everything that is not time and not the envelope pair. + n_channels = 1 + for name, size in zip(dims, shape): + if name in (time_axis, env_axis): + continue + n_channels *= int(size) + + if srate is None: + gain = _axis_gain(axes.get(time_axis)) + srate = 1.0 / gain if gain else 0.0 + + ch_data = _axis_data(axes.get("ch")) + labels = None + if ch_data is not None and ch_data.dtype.fields is not None: + labels = channel_names(ch_data, n_channels, fields=label_fields) + + unit = attrs.get("unit") if attrs else None + return StreamShape( + n_channels=max(1, n_channels), + srate=float(srate or 0.0), + channel_labels=labels, + envelope=env_axis is not None, + unit=None if unit is None else str(unit), + ) + + +def describe_axisarray( + msg: typing.Any, + *, + time_axis: str = "time", + label_fields: typing.Sequence[str] = ("label",), +) -> StreamShape: + """Describe a stream from one of its ``AxisArray`` messages.""" + return _describe( + msg.dims, + msg.axes, + getattr(msg, "attrs", None) or {}, + msg.data.shape, + None, + time_axis=time_axis, + label_fields=label_fields, + ) + + +def describe_mirror( + mirror: typing.Any, + *, + time_axis: str = "time", + label_fields: typing.Sequence[str] = ("label",), +) -> StreamShape | None: + """Describe a stream from a connected :class:`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. + """ + meta = mirror.meta + if meta is None or not meta.bvalid or meta.ndim < 2: + return None + axes = mirror.axes + if axes is None: + return None + shape = tuple(int(v) for v in meta.shape[: meta.ndim]) + # The ring rolls the buffered axis to the front; dims record the sender's + # order, so rebuild the order the buffer is actually in. + dims = list(mirror.dims or []) + if time_axis in dims: + dims.insert(0, dims.pop(dims.index(time_axis))) + return _describe( + dims, + axes, + mirror.attrs or {}, + shape, + float(meta.srate), + time_axis=time_axis, + label_fields=label_fields, + ) + + +def flatten_for_plot(data: np.ndarray, shape: StreamShape) -> np.ndarray: + """Reshape a block to what phosphor's ``push_data`` expects. + + ``(n_samples, ..., 2)`` for an envelope, ``(n_samples, n_channels)`` + otherwise, with any extra dimensions folded into channels. + + The envelope 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 lower and upper bounds, with + every channel label off by a factor of two. It looks like data, so nothing + complains. + """ + if data.size == 0: + return data.reshape((0, shape.n_channels, 2) if shape.envelope else (0, shape.n_channels)) + n_samples = data.shape[0] + if shape.envelope: + return data.reshape(n_samples, shape.n_channels, 2) + return data.reshape(n_samples, shape.n_channels) diff --git a/src/ezmsg/tools/plot/shmem_sweep.py b/src/ezmsg/tools/plot/shmem_sweep.py new file mode 100644 index 0000000..11ce89a --- /dev/null +++ b/src/ezmsg/tools/plot/shmem_sweep.py @@ -0,0 +1,233 @@ +"""A sweep plot fed from a shared-memory ring. + +:class:`~ezmsg.tools.shmem.shmem.ShMemCircBuff` writes samples into shared +memory and :class:`~ezmsg.tools.shmem.shmem_mirror.EZShmMirror` reads them back +in another process; this is the Qt widget that sits on the far end and draws +them. It exists so that consumers stop writing their own: the lazy build, the +poll timer, the metadata handling and the envelope unpacking are the same +problem every time, and each reimplementation has got a different subset of it +right. + +Why shared memory at all: the plot needs every sample the pipeline produces, and +routing a 30 kHz multichannel stream through ezmsg's message transport to reach +a GUI in another process costs a serialization round-trip per message. The ring +is written once and read in place. +""" + +from __future__ import annotations + +import logging +import typing + +import numpy as np +from phosphor import ChannelPlotControlsWidget +from phosphor.sweep_widget import SweepConfig, SweepWidget +from PySide6 import QtCore, QtWidgets + +from ..shmem.shmem_mirror import EZShmMirror +from .describe import StreamShape, describe_mirror, flatten_for_plot + +logger = logging.getLogger(__name__) + +__all__ = ["ShmemSweepWidget"] + +# Poll rate used when the render rate is uncapped, so there is no draw cadence +# to match. +DEFAULT_POLL_HZ: float = 60.0 + + +class ShmemSweepWidget(QtWidgets.QWidget): + """Mirrors a shmem ring and draws it, building the plot on first data. + + The plot cannot be built up front: channel count, sample rate and channel + names are properties of the stream, and the stream may not exist yet when + the window opens. So this shows a placeholder, polls, and builds once the + writer has published something real. + """ + + def __init__( + self, + shmem_name: str, + *, + display_dur: float = 5.0, + n_visible: int | None = None, + poll_hz: float | None = None, + max_fps: float | None = None, + n_columns: int | None = None, + label_fields: typing.Sequence[str] = ("label",), + show_controls: bool = True, + placeholder_text: str = "Waiting for data…", + parent: QtWidgets.QWidget | None = None, + ) -> None: + super().__init__(parent) + self._display_dur = display_dur + self._n_visible = n_visible + self._max_fps = max_fps + self._n_columns = n_columns + self._label_fields = tuple(label_fields) + self._show_controls = show_controls + + layout = QtWidgets.QVBoxLayout(self) + layout.setContentsMargins(0, 0, 0, 0) + self._layout = layout + + self._placeholder = QtWidgets.QLabel(placeholder_text) + self._placeholder.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) + layout.addWidget(self._placeholder) + + self._sweep: SweepWidget | None = None + self._controls: ChannelPlotControlsWidget | None = None + self._shape: StreamShape | None = None + + self._shmem_name = shmem_name + self._mirror = EZShmMirror(shmem_name) + + poll_hz = self._effective_poll_hz(poll_hz, max_fps) + self._timer = QtCore.QTimer(self) + self._timer.setInterval(max(1, int(1000.0 / poll_hz))) + self._timer.timeout.connect(self._on_tick) + self._timer.start() + + self._idle_ticks = 0 + self._idle_log_every = max(1, int(3.0 * poll_hz)) + + # ---- Public API ---------------------------------------------------- + + @property + def sweep(self) -> SweepWidget | None: + """The inner plot, or None before the first data arrives.""" + return self._sweep + + @property + def stream_shape(self) -> StreamShape | None: + """What the widget most recently understood the stream to be.""" + return self._shape + + def shutdown(self) -> None: + """Stop polling, release the mirror, and close the figure. + + The figure has to go before the Qt widget is destroyed, or rendercanvas + keeps painting into a deleted canvas. + """ + self._timer.stop() + self._mirror.disconnect() + self._close_figure() + + # ---- Internals ----------------------------------------------------- + + @staticmethod + def _effective_poll_hz(poll_hz: float | None, max_fps: float | None) -> float: + """Resolve how often to read the ring. + + An explicit rate wins. Otherwise match the render cap: reading faster + than the plot draws buys nothing but copies. With no cap there is no + cadence to match, so fall back to a default. + """ + if poll_hz is not None and poll_hz > 0: + return float(poll_hz) + if max_fps is not None and max_fps > 0: + return float(max_fps) + return DEFAULT_POLL_HZ + + def _close_figure(self) -> None: + if self._sweep is None: + return + figure = getattr(self._sweep, "_figure", None) + if figure is not None: + try: + figure.close() + except Exception: + logger.exception("closing the sweep figure raised; continuing teardown") + + def _on_tick(self) -> None: + samples, _overflow = self._mirror.auto_view() + + shape = describe_mirror(self._mirror, label_fields=self._label_fields) + if shape is None or shape.srate <= 0: + self._idle_ticks += 1 + if self._idle_ticks % self._idle_log_every == 0: + logger.info("Waiting for data on shmem %r — nothing published yet.", self._shmem_name) + return + if self._idle_ticks: + logger.info("Connected to shmem %r; data is flowing.", self._shmem_name) + self._idle_ticks = 0 + + self._apply_shape(shape) + + if samples is not None and samples.size: + self._sweep.push_data(np.ascontiguousarray(flatten_for_plot(samples, shape), dtype=np.float32)) + + def _apply_shape(self, shape: StreamShape) -> None: + """Build the plot, or reconfigure it if the stream changed underneath.""" + previous, self._shape = self._shape, shape + if self._sweep is None: + self._build(shape) + return + if previous == shape: + return + # A changed rate or envelope mode means the buffer's whole layout is + # wrong; anything else resizes in place, which avoids the plot flashing + # every time a user narrows the channel selection. + if previous is None or previous.srate != shape.srate or previous.envelope != shape.envelope: + self._build(shape) + else: + self._sweep.update_config(self._config_for(shape)) + self._sweep.set_channel_labels(self._labels_for(shape)) + + def _labels_for(self, shape: StreamShape) -> list[str]: + labels = shape.channel_labels + if labels is not None and len(labels) >= shape.n_channels: + return list(labels[: shape.n_channels]) + return [f"ch{i}" for i in range(shape.n_channels)] + + def _config_for(self, shape: StreamShape) -> SweepConfig: + kwargs: dict[str, typing.Any] = {} + if self._n_columns is not None: + kwargs["n_columns"] = self._n_columns + if self._max_fps is not None: + kwargs["max_fps"] = self._max_fps + return SweepConfig( + n_channels=shape.n_channels, + # For an envelope this is the bucket rate, which is what the buffer + # must be sized with -- describe_mirror reads it off the ring header, + # so it is already post-decimation. + srate=shape.srate, + display_dur=self._display_dur, + n_visible=self._n_visible if self._n_visible is not None else shape.n_channels, + channel_labels=self._labels_for(shape), + envelope=shape.envelope, + **kwargs, + ) + + def _build(self, shape: StreamShape) -> None: + logger.info( + "Building sweep: %d channels @ %.1f Hz%s", + shape.n_channels, + shape.srate, + " (min/max envelope)" if shape.envelope else "", + ) + # Keep whatever time span the user had scrolled to across a rebuild. + if self._sweep is not None: + buf = getattr(self._sweep, "sweep_buffer", None) + dur = getattr(buf, "display_dur", None) + if dur: + self._display_dur = dur + self._close_figure() + self._layout.removeWidget(self._sweep) + self._sweep.deleteLater() + self._sweep = None + if self._controls is not None: + self._layout.removeWidget(self._controls) + self._controls.deleteLater() + self._controls = None + if self._placeholder is not None: + self._layout.removeWidget(self._placeholder) + self._placeholder.deleteLater() + self._placeholder = None + + self._sweep = SweepWidget(self._config_for(shape), parent=self) + self._layout.addWidget(self._sweep) + self._sweep.set_channel_labels_visible(True) + if self._show_controls: + self._controls = ChannelPlotControlsWidget(self._sweep, parent=self) + self._layout.addWidget(self._controls) diff --git a/src/ezmsg/tools/sigmon/cli.py b/src/ezmsg/tools/sigmon/cli.py index ac798e7..d591d04 100644 --- a/src/ezmsg/tools/sigmon/cli.py +++ b/src/ezmsg/tools/sigmon/cli.py @@ -18,6 +18,7 @@ from PySide6.QtGui import QKeySequence, QShortcut from PySide6.QtWidgets import QApplication, QMainWindow, QSplitter, QWidget +from ezmsg.tools.plot.describe import describe_axisarray, flatten_for_plot from ezmsg.tools.sigmon.dag_widget import DAGWidget logger = logging.getLogger(__name__) @@ -89,6 +90,8 @@ def __init__( # Channel metadata cached from the first message of each topic. self._channel_labels: list[str] | None = None self._channel_positions: np.ndarray | None = None + # What describe_axisarray made of the stream; rebuilt on topic change. + self._shape = None # Cached parameters for rebuilding the primary (sweep/spectrum) widget. self._primary_config: SweepConfig | SpectrumConfig | None = None self._showing_scatter = False @@ -104,6 +107,7 @@ def _on_node_selected(self, topic: str) -> None: self._first_message = True self._channel_labels = None self._channel_positions = None + self._shape = None self._primary_config = None self._showing_scatter = False @@ -121,16 +125,13 @@ def _create_plot_widget(self, msg) -> None: labels = self._channel_labels if "time" in msg.dims: - time_axis = msg.get_axis("time") - srate = 1.0 / time_axis.gain - time_idx = msg.get_axis_idx("time") - n_samples = msg.shape[time_idx] - n_channels = msg.data.size // n_samples - + shape = describe_axisarray(msg) + self._shape = shape config = SweepConfig( - n_channels=n_channels, - srate=srate, - channel_labels=labels, + n_channels=shape.n_channels, + srate=shape.srate, + channel_labels=labels or shape.channel_labels, + envelope=shape.envelope, ) widget = SweepWidget(config) @@ -217,10 +218,9 @@ def _push_message(self, msg) -> None: if isinstance(widget, SweepWidget): time_idx = msg.get_axis_idx("time") if "time" in msg.dims else 0 - n_samples = msg.shape[time_idx] - n_channels = msg.data.size // n_samples if n_samples > 0 else 1 - data_2d = np.moveaxis(msg.data, time_idx, 0).reshape(n_samples, n_channels) - widget.push_data(data_2d.astype(np.float32)) + shape = self._shape or describe_axisarray(msg) + data = flatten_for_plot(np.moveaxis(msg.data, time_idx, 0), shape) + widget.push_data(data.astype(np.float32)) elif isinstance(widget, SpectrumWidget): freq_idx = msg.get_axis_idx("freq") if "freq" in msg.dims else 0 diff --git a/src/ezmsg/tools/viewer/cli.py b/src/ezmsg/tools/viewer/cli.py index ea4d713..2751fc3 100644 --- a/src/ezmsg/tools/viewer/cli.py +++ b/src/ezmsg/tools/viewer/cli.py @@ -19,6 +19,8 @@ ) from PySide6.QtWidgets import QApplication, QMainWindow, QWidget +from ezmsg.tools.plot.describe import describe_axisarray, flatten_for_plot + logger = logging.getLogger(__name__) GRAPH_IP = "127.0.0.1" @@ -105,6 +107,8 @@ def __init__( self._first_message = True self._channel_labels: list[str] | None = None self._channel_positions: np.ndarray | None = None + # What describe_axisarray made of the stream; rebuilt on topic change. + self._shape = None # ------------------------------------------------------------------ # Data handling @@ -121,19 +125,17 @@ def _create_plot_widget(self, msg) -> None: labels = self._channel_labels if self._mode == PlotMode.timeseries: - if "time" in msg.dims: - time_axis = msg.get_axis("time") - srate = 1.0 / time_axis.gain - time_idx = msg.get_axis_idx("time") - n_samples = msg.shape[time_idx] - n_channels = msg.data.size // n_samples - else: - logger.warning("No 'time' dimension — using shape[0] as time") - n_samples = msg.shape[0] - n_channels = msg.data.size // n_samples if n_samples > 0 else 1 - srate = 1000.0 - - config = SweepConfig(n_channels=n_channels, srate=srate, channel_labels=labels) + shape = describe_axisarray(msg) + if not shape.srate: + logger.warning("No usable 'time' axis — assuming 1 kHz") + shape = shape._replace(srate=1000.0) + self._shape = shape + config = SweepConfig( + n_channels=shape.n_channels, + srate=shape.srate, + channel_labels=labels or shape.channel_labels, + envelope=shape.envelope, + ) widget = SweepWidget(config) elif self._mode == PlotMode.spectral: @@ -166,13 +168,12 @@ def _push_message(self, msg) -> None: if isinstance(widget, SweepWidget): time_idx = msg.get_axis_idx("time") if "time" in msg.dims else 0 - n_samples = msg.shape[time_idx] - n_channels = msg.data.size // n_samples if n_samples > 0 else 1 - data_2d = np.moveaxis(msg.data, time_idx, 0).reshape(n_samples, n_channels) + shape = self._shape or describe_axisarray(msg) + data = flatten_for_plot(np.moveaxis(msg.data, time_idx, 0), shape) # Pass the AxisArray time-axis offset so the sweep buffer # tracks the same clock as the event timestamps. ts = msg.get_axis("time").offset if "time" in msg.dims else None - widget.push_data(data_2d.astype(np.float32), timestamps=ts) + widget.push_data(data.astype(np.float32), timestamps=ts) elif isinstance(widget, SpectrumWidget): freq_idx = msg.get_axis_idx("freq") if "freq" in msg.dims else 0 diff --git a/tests/test_plot_describe.py b/tests/test_plot_describe.py new file mode 100644 index 0000000..cac7176 --- /dev/null +++ b/tests/test_plot_describe.py @@ -0,0 +1,169 @@ +"""Working out what a stream is, so a plot can draw it. + +The case that matters is the envelope. A ``(time, ch, 2)`` block folded naively +into two dimensions renders as twice as many traces, alternating lower and upper +bounds, against channel labels that are now off by a factor of two -- and it +looks like data, so nothing complains. That is what both CLIs did before this +module existed, and it is what these tests exist to prevent coming back. +""" + +import numpy as np +import pytest +from ezmsg.util.messages.axisarray import AxisArray + +from ezmsg.tools.plot.describe import ( + describe_axisarray, + envelope_axis, + flatten_for_plot, +) + +CHANNEL_DTYPE = np.dtype([("bank", "U2"), ("elec", " AxisArray.CoordinateAxis: + data = np.zeros(n, dtype=CHANNEL_DTYPE) + for i in range(n): + data["bank"][i] = "A" + data["elec"][i] = i + 1 + data["label"][i] = f"e{i}" + return AxisArray.CoordinateAxis(data=data, dims=["ch"], unit="") + + +def metric_axis(labels=("min", "max")) -> AxisArray.CoordinateAxis: + return AxisArray.CoordinateAxis(data=np.array(list(labels)), dims=["metric"], unit="") + + +def signal(n_time=10, n_ch=4, fs=30000.0, unit="uV") -> AxisArray: + return AxisArray( + data=np.zeros((n_time, n_ch), dtype=np.float32), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=fs), "ch": ch_axis(n_ch)}, + attrs={"unit": unit}, + key="sig", + ) + + +def envelope(n_time=10, n_ch=4, fs=1000.0, labels=("min", "max")) -> AxisArray: + return AxisArray( + data=np.zeros((n_time, n_ch, 2), dtype=np.float32), + dims=["time", "ch", "metric"], + axes={"time": AxisArray.TimeAxis(fs=fs), "ch": ch_axis(n_ch), "metric": metric_axis(labels)}, + attrs={"unit": "uV"}, + key="env", + ) + + +# ---- plain signals --------------------------------------------------------- + + +def test_describes_a_plain_signal(): + shape = describe_axisarray(signal(n_ch=8)) + assert (shape.n_channels, shape.srate, shape.envelope) == (8, 30000.0, False) + assert shape.channel_labels == [f"e{i}" for i in range(8)] + assert shape.unit == "uV" + + +def test_extra_dimensions_fold_into_channels(): + """A (time, ch, band) block has no envelope axis, so band multiplies out.""" + msg = AxisArray( + data=np.zeros((10, 4, 3), dtype=np.float32), + dims=["time", "ch", "band"], + axes={"time": AxisArray.TimeAxis(fs=100.0), "ch": ch_axis(4)}, + key="multi", + ) + shape = describe_axisarray(msg) + assert shape.n_channels == 12 + assert not shape.envelope + + +def test_missing_ch_metadata_yields_no_labels(): + msg = AxisArray( + data=np.zeros((10, 4), dtype=np.float32), + dims=["time", "ch"], + axes={"time": AxisArray.TimeAxis(fs=100.0)}, + key="bare", + ) + assert describe_axisarray(msg).channel_labels is None + + +# ---- envelopes ------------------------------------------------------------- + + +def test_describes_an_envelope(): + shape = describe_axisarray(envelope(n_ch=4)) + assert shape.envelope + # The pair axis must not be counted as channels. + assert shape.n_channels == 4 + assert shape.channel_labels == ["e0", "e1", "e2", "e3"] + + +def test_envelope_is_identified_by_labels_not_width(): + """A 2-wide trailing axis is not automatically an envelope: (mean, std) is + the same shape and would be nonsense drawn as bounds.""" + dims = ["time", "ch", "metric"] + assert envelope_axis(dims, {"metric": metric_axis(("min", "max"))}) == "metric" + assert envelope_axis(dims, {"metric": metric_axis(("mean", "std"))}) is None + assert envelope_axis(dims, {"metric": metric_axis(("MIN", "MAX"))}) == "metric" + + +def test_envelope_axis_must_be_trailing_and_named(): + assert envelope_axis(["time", "metric", "ch"], {"metric": metric_axis()}) is None + assert envelope_axis(["time", "ch", "other"], {"other": metric_axis()}) is None + assert envelope_axis([], {}) is None + + +def test_envelope_axis_of_wrong_width_is_rejected(): + wide = AxisArray.CoordinateAxis(data=np.array(["min", "max", "mean"]), dims=["metric"], unit="") + assert envelope_axis(["time", "ch", "metric"], {"metric": wide}) is None + + +def test_axes_may_be_plain_dicts(): + """EZShmMirror hands back dicts, not ezmsg axis objects.""" + as_dict = {"kind": "coord", "unit": "", "dims": ["metric"], "data": np.array(["min", "max"])} + assert envelope_axis(["time", "ch", "metric"], {"metric": as_dict}) == "metric" + + +# ---- reshaping ------------------------------------------------------------- + + +def test_envelope_keeps_its_pair_axis(): + """The regression this module exists for.""" + shape = describe_axisarray(envelope(n_ch=4)) + raw = np.arange(10 * 4 * 2, dtype=np.float32).reshape(10, 4, 2) + + out = flatten_for_plot(raw, shape) + + assert out.shape == (10, 4, 2) + # Naive flattening would have produced (10, 8) with min/max interleaved. + np.testing.assert_array_equal(out, raw) + + +def test_plain_signal_flattens_to_two_dimensions(): + shape = describe_axisarray(signal(n_ch=4)) + out = flatten_for_plot(np.zeros((10, 4), dtype=np.float32), shape) + assert out.shape == (10, 4) + + +def test_extra_dimensions_flatten_into_channels(): + msg = AxisArray( + data=np.zeros((10, 4, 3), dtype=np.float32), + dims=["time", "ch", "band"], + axes={"time": AxisArray.TimeAxis(fs=100.0), "ch": ch_axis(4)}, + key="multi", + ) + shape = describe_axisarray(msg) + assert flatten_for_plot(np.zeros((10, 4, 3), dtype=np.float32), shape).shape == (10, 12) + + +def test_empty_blocks_keep_their_rank(): + """A zero-length block still has to match the shape of its neighbours.""" + env = describe_axisarray(envelope(n_ch=4)) + sig = describe_axisarray(signal(n_ch=4)) + assert flatten_for_plot(np.zeros((0, 4, 2), dtype=np.float32), env).shape == (0, 4, 2) + assert flatten_for_plot(np.zeros((0, 4), dtype=np.float32), sig).shape == (0, 4) + + +def test_envelope_srate_is_the_post_decimation_rate(): + """What a sweep buffer must be sized with. Using the pre-decimation rate + makes the ring far longer than the data arriving to fill it.""" + assert describe_axisarray(envelope(fs=1000.0)).srate == pytest.approx(1000.0) From 41ef5bc1cd2b8cb374251d57dd104963aa2c1db4 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 14:30:16 -0400 Subject: [PATCH 02/10] Recognise metric kinds by vocabulary, refuse the ones we cannot draw A trailing 2-wide axis was being treated as a min/max envelope, which is right for BinnedAggregate(MIN, MAX) and wrong for anything else with the same shape. (mean, std) is the obvious other case: same rank, same width, completely different drawing -- a semi-transparent band from mean-std to mean+std with an opaque line at the mean, and a column reduction that averages rather than takes extremes. So the axis is now identified by a vocabulary. METRIC_KINDS maps label tuples to a kind; SWEEP_RENDERABLE_METRICS says which of those a sweep can actually draw, which today is minmax alone. Adding a kind to the first is cheap; adding it to the second means teaching a renderer. Describing and drawing are separated deliberately. describe_* reports any recognised metric, including ones nothing here can draw, because a caller asking what arrived should not have to catch an exception to find out. require_sweep_renderable is the capability check, and it names the kind, the labels and what is supported rather than just refusing. The widget reports and stops rather than raising out of its Qt timer slot, which would repeat the failure every tick for as long as the window stayed open. The message goes where the plot would have been. Labels are matched case-insensitively, and an unrecognised vocabulary is not an error -- it falls through to being ordinary extra dimensions, which is the pre-existing behaviour for any trailing axis. --- src/ezmsg/tools/plot/__init__.py | 14 ++- src/ezmsg/tools/plot/describe.py | 139 +++++++++++++++++++++------- src/ezmsg/tools/plot/shmem_sweep.py | 34 ++++++- src/ezmsg/tools/sigmon/cli.py | 7 +- src/ezmsg/tools/viewer/cli.py | 7 +- tests/test_plot_describe.py | 59 ++++++++---- 6 files changed, 205 insertions(+), 55 deletions(-) diff --git a/src/ezmsg/tools/plot/__init__.py b/src/ezmsg/tools/plot/__init__.py index 93710a1..40ce497 100644 --- a/src/ezmsg/tools/plot/__init__.py +++ b/src/ezmsg/tools/plot/__init__.py @@ -11,19 +11,29 @@ """ from .describe import ( + METRIC_KINDS, + SWEEP_RENDERABLE_METRICS, + MetricSpec, StreamShape, + UnsupportedMetricError, describe_axisarray, describe_mirror, - envelope_axis, flatten_for_plot, + metric_axis, + require_sweep_renderable, ) from .shmem_sweep import ShmemSweepWidget __all__ = [ + "METRIC_KINDS", + "SWEEP_RENDERABLE_METRICS", + "MetricSpec", "ShmemSweepWidget", "StreamShape", + "UnsupportedMetricError", "describe_axisarray", "describe_mirror", - "envelope_axis", "flatten_for_plot", + "metric_axis", + "require_sweep_renderable", ] diff --git a/src/ezmsg/tools/plot/describe.py b/src/ezmsg/tools/plot/describe.py index b251f42..15abf5f 100644 --- a/src/ezmsg/tools/plot/describe.py +++ b/src/ezmsg/tools/plot/describe.py @@ -19,23 +19,65 @@ from ..chmeta import channel_names __all__ = [ - "ENVELOPE_AXIS_CANDIDATES", + "METRIC_AXIS_CANDIDATES", + "METRIC_KINDS", + "SWEEP_RENDERABLE_METRICS", + "MetricSpec", "StreamShape", + "UnsupportedMetricError", "describe_axisarray", "describe_mirror", - "envelope_axis", "flatten_for_plot", + "metric_axis", + "require_sweep_renderable", ] -# Axis names an upstream min/max decimator might use for its (min, max) pair. +# Axis names an upstream aggregator might use for its per-sample tuple. # ezmsg-sigproc's BinnedAggregate calls it "metric" by default but the name is a # setting, so recognising a couple of obvious alternatives costs nothing. -ENVELOPE_AXIS_CANDIDATES = ("metric", "minmax", "bound") +METRIC_AXIS_CANDIDATES = ("metric", "minmax", "bound", "stat") -# Aggregation-function labels that make an axis an envelope rather than, say, -# a (mean, std) pair -- which is 2-wide and named the same way but must not be -# drawn as an upper and lower bound. -_ENVELOPE_LABELS = ("min", "max") +# Label tuples we recognise, and what to call the thing they describe. Keyed on +# labels rather than width because width says nothing: (min, max) and +# (mean, std) are both 2-wide and mean entirely different things, and drawing +# one as the other is silently wrong rather than visibly broken. +# +# Adding a kind here is the cheap half. The expensive half is teaching a +# renderer to draw it -- see SWEEP_RENDERABLE_METRICS. +METRIC_KINDS: dict[tuple[str, ...], str] = { + ("min", "max"): "minmax", + ("mean", "std"): "mean_std", + ("mean", "sem"): "mean_sem", +} + +# What a sweep plot can actually draw today. +# +# "minmax" maps onto phosphor's envelope input directly: the pair *is* the band, +# so the existing column reduction (min of mins, max of maxes) is correct. +# +# A dispersion pair like "mean_std" needs different drawing -- a semi-transparent +# band from mean-std to mean+std with an opaque line at the mean -- and +# different column reduction, since averaging a mean is not the same as taking +# extremes. Recognised here so it fails with an explanation instead of being +# drawn as if it were an envelope. +SWEEP_RENDERABLE_METRICS = frozenset({"minmax"}) + + +class MetricSpec(typing.NamedTuple): + """A trailing per-sample tuple: what it is called and what it holds.""" + + axis: str + """Name of the trailing axis.""" + + labels: tuple[str, ...] + """Its coordinate values, lowercased, in order.""" + + kind: str + """The entry in :data:`METRIC_KINDS` these labels matched.""" + + +class UnsupportedMetricError(NotImplementedError): + """A recognised metric axis that no renderer here can draw yet.""" class StreamShape(typing.NamedTuple): @@ -53,30 +95,58 @@ class StreamShape(typing.NamedTuple): channel_labels: list[str] | None """One name per channel, or None if the stream does not say.""" - envelope: bool - """Whether each sample carries a (min, max) pair on a trailing axis.""" + 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 + def envelope(self) -> bool: + """Whether each sample carries a (min, max) pair -- phosphor's envelope.""" + return self.metric is not None and self.metric.kind == "minmax" -def envelope_axis(dims: typing.Sequence[str], axes: typing.Mapping[str, typing.Any]) -> str | None: - """Name of the trailing (min, max) axis, or None if this is a plain signal. - Identified by its *labels* rather than its name or width. A length-2 - trailing axis could as easily be (mean, std), which would be nonsense to - draw as an envelope, so the coordinate values have to say ``min`` and - ``max``. The name is only used to narrow the search. +def metric_axis(dims: typing.Sequence[str], axes: typing.Mapping[str, typing.Any]) -> MetricSpec | None: + """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 :data:`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 :func:`require_sweep_renderable` for + the capability check. """ if not dims: return None name = dims[-1] - if name not in ENVELOPE_AXIS_CANDIDATES: + if name not in METRIC_AXIS_CANDIDATES: return None data = _axis_data(axes.get(name)) - if data is None or len(data) != 2: + if data is None: return None - return name if tuple(str(v).lower() for v in data) == _ENVELOPE_LABELS else None + labels = tuple(str(v).lower() for v in data) + kind = METRIC_KINDS.get(labels) + return None if kind is None else MetricSpec(axis=name, labels=labels, kind=kind) + + +def require_sweep_renderable(shape: StreamShape) -> None: + """Raise if a sweep plot cannot draw this stream's metric axis. + + :raises UnsupportedMetricError: for a recognised metric a sweep cannot draw. + """ + metric = shape.metric + if metric is None or metric.kind in SWEEP_RENDERABLE_METRICS: + return + raise UnsupportedMetricError( + f"stream carries a {metric.kind!r} metric axis {metric.labels} on {metric.axis!r}, " + f"which a sweep plot cannot draw yet (supported: {sorted(SWEEP_RENDERABLE_METRICS)}). " + "Aggregate the stream differently upstream, or add rendering for it." + ) def _axis_data(axis: typing.Any) -> np.ndarray | None: @@ -108,12 +178,13 @@ def _describe( label_fields: typing.Sequence[str] = ("label",), ) -> StreamShape: dims = list(dims) - env_axis = envelope_axis(dims, axes) + metric = metric_axis(dims, axes) + metric_name = metric.axis if metric is not None else None - # Channel count is everything that is not time and not the envelope pair. + # Channel count is everything that is neither time nor the metric tuple. n_channels = 1 for name, size in zip(dims, shape): - if name in (time_axis, env_axis): + if name in (time_axis, metric_name): continue n_channels *= int(size) @@ -131,7 +202,7 @@ def _describe( n_channels=max(1, n_channels), srate=float(srate or 0.0), channel_labels=labels, - envelope=env_axis is not None, + metric=metric, unit=None if unit is None else str(unit), ) @@ -190,20 +261,20 @@ def describe_mirror( def flatten_for_plot(data: np.ndarray, shape: StreamShape) -> np.ndarray: - """Reshape a block to what phosphor's ``push_data`` expects. + """Reshape a block to what a plot's ``push_data`` expects. - ``(n_samples, ..., 2)`` for an envelope, ``(n_samples, n_channels)`` - otherwise, with any extra dimensions folded into channels. + ``(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 envelope case is the reason this exists. Folding a ``(time, ch, 2)`` + 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 lower and upper bounds, with - every channel label off by a factor of two. It looks like data, so nothing + 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. """ + width = len(shape.metric.labels) if shape.metric is not None else None + tail = (shape.n_channels,) if width is None else (shape.n_channels, width) if data.size == 0: - return data.reshape((0, shape.n_channels, 2) if shape.envelope else (0, shape.n_channels)) - n_samples = data.shape[0] - if shape.envelope: - return data.reshape(n_samples, shape.n_channels, 2) - return data.reshape(n_samples, shape.n_channels) + return data.reshape((0,) + tail) + return data.reshape((data.shape[0],) + tail) diff --git a/src/ezmsg/tools/plot/shmem_sweep.py b/src/ezmsg/tools/plot/shmem_sweep.py index 11ce89a..58c8e3a 100644 --- a/src/ezmsg/tools/plot/shmem_sweep.py +++ b/src/ezmsg/tools/plot/shmem_sweep.py @@ -25,7 +25,13 @@ from PySide6 import QtCore, QtWidgets from ..shmem.shmem_mirror import EZShmMirror -from .describe import StreamShape, describe_mirror, flatten_for_plot +from .describe import ( + StreamShape, + UnsupportedMetricError, + describe_mirror, + flatten_for_plot, + require_sweep_renderable, +) logger = logging.getLogger(__name__) @@ -78,6 +84,7 @@ def __init__( self._sweep: SweepWidget | None = None self._controls: ChannelPlotControlsWidget | None = None self._shape: StreamShape | None = None + self._error: str | None = None self._shmem_name = shmem_name self._mirror = EZShmMirror(shmem_name) @@ -113,8 +120,24 @@ def shutdown(self) -> None: self._mirror.disconnect() self._close_figure() + @property + def error(self) -> str | None: + """Why the widget gave up, or None if it has not.""" + return self._error + # ---- Internals ----------------------------------------------------- + def _fail(self, message: str) -> None: + """Stop polling and say why, in the widget and in the log.""" + if self._error is not None: + return + self._error = message + logger.error("%s", message) + self._timer.stop() + if self._placeholder is not None: + self._placeholder.setText(message) + self._placeholder.setWordWrap(True) + @staticmethod def _effective_poll_hz(poll_hz: float | None, max_fps: float | None) -> float: """Resolve how often to read the ring. @@ -152,6 +175,15 @@ def _on_tick(self) -> None: logger.info("Connected to shmem %r; data is flowing.", self._shmem_name) self._idle_ticks = 0 + try: + require_sweep_renderable(shape) + except UnsupportedMetricError as exc: + # Stop rather than draw it as something it is not. Reported once and + # the timer stopped, because raising out of a Qt slot would repeat + # this every tick for as long as the window is open. + self._fail(str(exc)) + return + self._apply_shape(shape) if samples is not None and samples.size: diff --git a/src/ezmsg/tools/sigmon/cli.py b/src/ezmsg/tools/sigmon/cli.py index d591d04..900564d 100644 --- a/src/ezmsg/tools/sigmon/cli.py +++ b/src/ezmsg/tools/sigmon/cli.py @@ -18,7 +18,11 @@ from PySide6.QtGui import QKeySequence, QShortcut from PySide6.QtWidgets import QApplication, QMainWindow, QSplitter, QWidget -from ezmsg.tools.plot.describe import describe_axisarray, flatten_for_plot +from ezmsg.tools.plot.describe import ( + describe_axisarray, + flatten_for_plot, + require_sweep_renderable, +) from ezmsg.tools.sigmon.dag_widget import DAGWidget logger = logging.getLogger(__name__) @@ -126,6 +130,7 @@ def _create_plot_widget(self, msg) -> None: if "time" in msg.dims: shape = describe_axisarray(msg) + require_sweep_renderable(shape) self._shape = shape config = SweepConfig( n_channels=shape.n_channels, diff --git a/src/ezmsg/tools/viewer/cli.py b/src/ezmsg/tools/viewer/cli.py index 2751fc3..e4edac1 100644 --- a/src/ezmsg/tools/viewer/cli.py +++ b/src/ezmsg/tools/viewer/cli.py @@ -19,7 +19,11 @@ ) from PySide6.QtWidgets import QApplication, QMainWindow, QWidget -from ezmsg.tools.plot.describe import describe_axisarray, flatten_for_plot +from ezmsg.tools.plot.describe import ( + describe_axisarray, + flatten_for_plot, + require_sweep_renderable, +) logger = logging.getLogger(__name__) @@ -129,6 +133,7 @@ def _create_plot_widget(self, msg) -> None: if not shape.srate: logger.warning("No usable 'time' axis — assuming 1 kHz") shape = shape._replace(srate=1000.0) + require_sweep_renderable(shape) self._shape = shape config = SweepConfig( n_channels=shape.n_channels, diff --git a/tests/test_plot_describe.py b/tests/test_plot_describe.py index cac7176..1f2dc72 100644 --- a/tests/test_plot_describe.py +++ b/tests/test_plot_describe.py @@ -12,9 +12,11 @@ from ezmsg.util.messages.axisarray import AxisArray from ezmsg.tools.plot.describe import ( + UnsupportedMetricError, describe_axisarray, - envelope_axis, flatten_for_plot, + metric_axis, + require_sweep_renderable, ) CHANNEL_DTYPE = np.dtype([("bank", "U2"), ("elec", " AxisArray.CoordinateAxis: return AxisArray.CoordinateAxis(data=data, dims=["ch"], unit="") -def metric_axis(labels=("min", "max")) -> AxisArray.CoordinateAxis: +def metric_ax(labels=("min", "max")) -> AxisArray.CoordinateAxis: return AxisArray.CoordinateAxis(data=np.array(list(labels)), dims=["metric"], unit="") @@ -47,7 +49,7 @@ def envelope(n_time=10, n_ch=4, fs=1000.0, labels=("min", "max")) -> AxisArray: return AxisArray( data=np.zeros((n_time, n_ch, 2), dtype=np.float32), dims=["time", "ch", "metric"], - axes={"time": AxisArray.TimeAxis(fs=fs), "ch": ch_axis(n_ch), "metric": metric_axis(labels)}, + axes={"time": AxisArray.TimeAxis(fs=fs), "ch": ch_axis(n_ch), "metric": metric_ax(labels)}, attrs={"unit": "uV"}, key="env", ) @@ -97,30 +99,55 @@ def test_describes_an_envelope(): assert shape.channel_labels == ["e0", "e1", "e2", "e3"] -def test_envelope_is_identified_by_labels_not_width(): - """A 2-wide trailing axis is not automatically an envelope: (mean, std) is - the same shape and would be nonsense drawn as bounds.""" +def test_metric_kind_comes_from_labels_not_width(): + """A 2-wide trailing axis says nothing on its own: (min, max) and + (mean, std) are the same shape and mean entirely different things.""" dims = ["time", "ch", "metric"] - assert envelope_axis(dims, {"metric": metric_axis(("min", "max"))}) == "metric" - assert envelope_axis(dims, {"metric": metric_axis(("mean", "std"))}) is None - assert envelope_axis(dims, {"metric": metric_axis(("MIN", "MAX"))}) == "metric" + assert metric_axis(dims, {"metric": metric_ax(("min", "max"))}).kind == "minmax" + assert metric_axis(dims, {"metric": metric_ax(("mean", "std"))}).kind == "mean_std" + assert metric_axis(dims, {"metric": metric_ax(("MIN", "MAX"))}).kind == "minmax" + # Not a vocabulary we know: treat as ordinary extra dimensions. + assert metric_axis(dims, {"metric": metric_ax(("p5", "p95"))}) is None -def test_envelope_axis_must_be_trailing_and_named(): - assert envelope_axis(["time", "metric", "ch"], {"metric": metric_axis()}) is None - assert envelope_axis(["time", "ch", "other"], {"other": metric_axis()}) is None - assert envelope_axis([], {}) is None +def test_only_minmax_is_renderable_today(): + """Others are recognised so they fail with an explanation rather than + being drawn as if they were an envelope.""" + minmax = describe_axisarray(envelope()) + require_sweep_renderable(minmax) # does not raise -def test_envelope_axis_of_wrong_width_is_rejected(): + dispersion = describe_axisarray(envelope(labels=("mean", "std"))) + assert dispersion.metric.kind == "mean_std" + assert not dispersion.envelope + with pytest.raises(UnsupportedMetricError, match="mean_std"): + require_sweep_renderable(dispersion) + + +def test_unrenderable_metric_still_describes_cleanly(): + """Describing is not drawing: a caller that only wants to know what + arrived should not have to catch anything.""" + shape = describe_axisarray(envelope(n_ch=4, labels=("mean", "std"))) + assert shape.n_channels == 4 + assert shape.metric.labels == ("mean", "std") + assert shape.channel_labels == ["e0", "e1", "e2", "e3"] + + +def test_metric_axis_must_be_trailing_and_named(): + assert metric_axis(["time", "metric", "ch"], {"metric": metric_ax()}) is None + assert metric_axis(["time", "ch", "other"], {"other": metric_ax()}) is None + assert metric_axis([], {}) is None + + +def test_metric_axis_of_unknown_width_is_rejected(): wide = AxisArray.CoordinateAxis(data=np.array(["min", "max", "mean"]), dims=["metric"], unit="") - assert envelope_axis(["time", "ch", "metric"], {"metric": wide}) is None + assert metric_axis(["time", "ch", "metric"], {"metric": wide}) is None def test_axes_may_be_plain_dicts(): """EZShmMirror hands back dicts, not ezmsg axis objects.""" as_dict = {"kind": "coord", "unit": "", "dims": ["metric"], "data": np.array(["min", "max"])} - assert envelope_axis(["time", "ch", "metric"], {"metric": as_dict}) == "metric" + assert metric_axis(["time", "ch", "metric"], {"metric": as_dict}).kind == "minmax" # ---- reshaping ------------------------------------------------------------- From 07aa40a028a9d99c61c8d738b8b0c78fb5ea63f6 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 15:38:51 -0400 Subject: [PATCH 03/10] Stop committing a local path source for phosphor `phosphor = { path = "../phosphor", editable = true }` is a developer convenience for testing against an unreleased phosphor. It went in with the adapter commit by accident, and CI has no sibling checkout: error: Distribution not found at: file:///home/runner/work/ezmsg-tools/phosphor A bad path source fails resolution for *every* job, including ones that never touch the package -- the failure above is `uv sync --only-group docs`, which needs neither phosphor nor Qt. Left a note next to the existing commented-out ezmsg git source saying how to add one locally and that it should not be committed. --- pyproject.toml | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 466cdd4..2cd2857 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,6 +102,10 @@ known-first-party = ["ezmsg.tools"] known-third-party = ["ezmsg"] [tool.uv.sources] -phosphor = { path = "../phosphor", editable = true } +# Local path sources are a developer convenience and must not be committed: CI +# has no sibling checkouts, so a path here fails resolution for every job, +# including ones that never touch the package (`uv sync --only-group docs`). +# Add them locally with `uv add ../phosphor --editable --frozen` and drop the +# change before committing. # Uncomment to use development version of ezmsg from git #ezmsg = { git = "https://github.com/ezmsg-org/ezmsg.git", branch = "dev" } From ca54c31255134fab74f69e4eb0936eec5693ab61 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 17:34:15 -0400 Subject: [PATCH 04/10] Give ShmemSweepWidget subclass hooks, and make the rebuild rule testable on_plot_built and on_frame are for the parts a host application has that this widget cannot: overlays and diagnostics parented to a plot that does not exist until the stream does, and readouts needing units the widget has no way to interpret. Without them a subclass has to override _on_tick and _build, which are internals. on_frame fires whether or not samples arrived, so a subclass sees a steady cadence rather than one that stalls whenever the source goes quiet. _needs_rebuild pulls the rebuild-versus-resize decision out of _apply_shape so it can be tested without a ring, a GPU or a display. It is worth pinning: rebuilding throws away the figure and flashes the plot, which happens every time a user narrows their channel selection if the rule is too eager, and a rate or metric change that is *not* caught leaves the buffer laid out for the wrong data. --- src/ezmsg/tools/plot/shmem_sweep.py | 42 +++++++++++++-- tests/test_shmem_sweep.py | 79 +++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 4 deletions(-) create mode 100644 tests/test_shmem_sweep.py diff --git a/src/ezmsg/tools/plot/shmem_sweep.py b/src/ezmsg/tools/plot/shmem_sweep.py index 58c8e3a..7bb7178 100644 --- a/src/ezmsg/tools/plot/shmem_sweep.py +++ b/src/ezmsg/tools/plot/shmem_sweep.py @@ -125,6 +125,26 @@ def error(self) -> str | None: """Why the widget gave up, or None if it has not.""" return self._error + # ---- Subclass hooks ------------------------------------------------- + + def on_plot_built(self) -> None: + """Called after the inner plot is created or recreated. + + Attach anything parented to the plot here -- overlays, diagnostics -- + rather than in ``__init__``: the plot does not exist until the stream + does, and it is thrown away and rebuilt if the stream's rate or metric + changes underneath. + """ + + def on_frame(self, shape: StreamShape) -> None: + """Called once per poll tick, after any new samples are pushed. + + For state that has to track the plot but that this widget cannot + compute -- anything needing units, or a host application's own + readouts. Called whether or not samples arrived, so a subclass sees a + steady cadence. + """ + # ---- Internals ----------------------------------------------------- def _fail(self, message: str) -> None: @@ -138,6 +158,20 @@ def _fail(self, message: str) -> None: self._placeholder.setText(message) self._placeholder.setWordWrap(True) + @staticmethod + def _needs_rebuild(previous: StreamShape | None, shape: StreamShape) -> bool: + """Whether a stream change invalidates the buffer's layout. + + Rebuilding throws away the figure and flashes the plot, which is + unpleasant every time a user narrows their channel selection -- so it + is reserved for changes the buffer cannot absorb. A different sample + rate resizes the ring, and a change of metric changes the rank of what + is stored; a channel count or relabel is handled in place. + """ + if previous is None: + return True + return previous.srate != shape.srate or previous.envelope != shape.envelope + @staticmethod def _effective_poll_hz(poll_hz: float | None, max_fps: float | None) -> float: """Resolve how often to read the ring. @@ -189,6 +223,8 @@ def _on_tick(self) -> None: if samples is not None and samples.size: self._sweep.push_data(np.ascontiguousarray(flatten_for_plot(samples, shape), dtype=np.float32)) + self.on_frame(shape) + def _apply_shape(self, shape: StreamShape) -> None: """Build the plot, or reconfigure it if the stream changed underneath.""" previous, self._shape = self._shape, shape @@ -197,10 +233,7 @@ def _apply_shape(self, shape: StreamShape) -> None: return if previous == shape: return - # A changed rate or envelope mode means the buffer's whole layout is - # wrong; anything else resizes in place, which avoids the plot flashing - # every time a user narrows the channel selection. - if previous is None or previous.srate != shape.srate or previous.envelope != shape.envelope: + if self._needs_rebuild(previous, shape): self._build(shape) else: self._sweep.update_config(self._config_for(shape)) @@ -263,3 +296,4 @@ def _build(self, shape: StreamShape) -> None: if self._show_controls: self._controls = ChannelPlotControlsWidget(self._sweep, parent=self) self._layout.addWidget(self._controls) + self.on_plot_built() diff --git a/tests/test_shmem_sweep.py b/tests/test_shmem_sweep.py new file mode 100644 index 0000000..a54a648 --- /dev/null +++ b/tests/test_shmem_sweep.py @@ -0,0 +1,79 @@ +"""ShmemSweepWidget's decision logic. + +The widget itself needs a live ring, a GPU and a display, so what is covered +here is the part that does not: how often to read, and the rule for when a +stream change means rebuilding the plot rather than resizing it. +""" + +import pytest + +# Reaching the widget needs Qt and a rendering backend; a headless runner has +# neither. Keyed on the import, so it still runs wherever they exist. +_mod = pytest.importorskip( + "ezmsg.tools.plot.shmem_sweep", + reason="needs PySide6 + phosphor (the 'viewer' or 'sigmon' extra)", + # Not the default: since pytest 9.1 importorskip only skips on + # ModuleNotFoundError, and rendercanvas raises a plain ImportError from a + # module that is very much installed. + exc_type=ImportError, +) + +from ezmsg.tools.plot.describe import MetricSpec, StreamShape # noqa: E402 + +DEFAULT_POLL_HZ = _mod.DEFAULT_POLL_HZ +_poll = _mod.ShmemSweepWidget._effective_poll_hz + +MINMAX = MetricSpec("metric", ("min", "max"), "minmax") + + +def shape(**kw) -> StreamShape: + base = dict(n_channels=4, srate=1000.0, channel_labels=None, metric=None, unit=None) + base.update(kw) + return StreamShape(**base) + + +@pytest.mark.parametrize( + ("poll_hz", "max_fps", "expected"), + [ + (None, 30, 30.0), # match the draw cadence + (45, 30, 45.0), # explicit wins + (120, None, 120.0), # explicit honoured with no cap + (None, None, DEFAULT_POLL_HZ), # nothing to match + (None, 0, DEFAULT_POLL_HZ), # uncapped is not a cadence of zero + (None, -1, DEFAULT_POLL_HZ), + (0, 30, 30.0), # non-positive poll means "not set" + (-5, None, DEFAULT_POLL_HZ), + ], +) +def test_poll_rate_precedence(poll_hz, max_fps, expected): + assert _poll(poll_hz, max_fps) == expected + + +def test_poll_rate_is_always_float(): + """The interval is computed as 1000/poll_hz, so an int here would still + work -- but the coercion is what keeps that true for odd inputs.""" + assert isinstance(_poll(45, 30), float) + + +# ---- rebuild vs resize ----------------------------------------------------- +# +# Rebuilding throws away the figure and flashes the plot, so it is reserved for +# changes that invalidate the buffer's layout. Everything else resizes in place. + + +@pytest.mark.parametrize( + ("before", "after", "rebuild"), + [ + (shape(), shape(n_channels=8), False), # narrower selection: resize + (shape(), shape(channel_labels=["a"] * 4), False), # relabel: resize + (shape(), shape(srate=2000.0), True), # rate change invalidates the ring + (shape(), shape(metric=MINMAX), True), # envelope changes the rank + (shape(metric=MINMAX), shape(), True), + ], +) +def test_rebuild_only_when_the_layout_is_invalid(before, after, rebuild): + assert _mod.ShmemSweepWidget._needs_rebuild(before, after) is rebuild + + +def test_first_shape_always_builds(): + assert _mod.ShmemSweepWidget._needs_rebuild(None, shape()) is True From df19bdead2a0540c722d56efedb29e518129383a Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 17:58:34 -0400 Subject: [PATCH 05/10] Record the dim order the ring actually holds ShMemCircBuff wrote two descriptions of the same buffer and they disagreed. on_message rolls the buffered axis to the front, and meta.shape was already written in that order -- but the metadata blob recorded msg.dims, the order the message arrived in. For the common (time, ch) case those coincide, so it went unnoticed; for a source sending (ch, time) the header said one thing and dims said another. describe_mirror was compensating by re-rolling dims itself, which is the wrong place: a reader has no way to know it should, and any other consumer of mirror.dims would have got it wrong. The sink knows what it stored, so it says so. Fixed at the source and the compensation removed. A test pins that a transposed source is described by the buffer's order. --- src/ezmsg/tools/plot/describe.py | 9 +++------ src/ezmsg/tools/shmem/shmem.py | 7 ++++++- tests/test_plot_describe.py | 26 ++++++++++++++++++++++++++ 3 files changed, 35 insertions(+), 7 deletions(-) diff --git a/src/ezmsg/tools/plot/describe.py b/src/ezmsg/tools/plot/describe.py index 15abf5f..5210e38 100644 --- a/src/ezmsg/tools/plot/describe.py +++ b/src/ezmsg/tools/plot/describe.py @@ -244,13 +244,10 @@ def describe_mirror( if axes is None: return None shape = tuple(int(v) for v in meta.shape[: meta.ndim]) - # The ring rolls the buffered axis to the front; dims record the sender's - # order, so rebuild the order the buffer is actually in. - dims = list(mirror.dims or []) - if time_axis in dims: - dims.insert(0, dims.pop(dims.index(time_axis))) + # dims and meta.shape describe the same ordering -- the sink records the + # order the ring actually holds, not the order the message arrived in. return _describe( - dims, + list(mirror.dims or []), axes, mirror.attrs or {}, shape, diff --git a/src/ezmsg/tools/shmem/shmem.py b/src/ezmsg/tools/shmem/shmem.py index d80ebcf..b0bccd2 100644 --- a/src/ezmsg/tools/shmem/shmem.py +++ b/src/ezmsg/tools/shmem/shmem.py @@ -378,7 +378,12 @@ def _update_aux_if_needed(self, msg: AxisArray) -> bool: ): return False - blob, dropped = encode_aux(msg.dims, msg.axes, msg.attrs, msg.key, self.SETTINGS.axis) + # The ring rolls the buffered axis to the front (see on_message), and + # meta.shape already describes that order, so dims must too -- a reader + # given the sender's original order would have to know to re-roll it, + # which is knowledge it has no way to arrive at. + rolled_dims = [self.SETTINGS.axis] + [d for d in msg.dims if d != self.SETTINGS.axis] + blob, dropped = encode_aux(rolled_dims, msg.axes, msg.attrs, msg.key, self.SETTINGS.axis) if dropped: dropped_set = frozenset(dropped) if self.STATE.warned_dropped_attrs != dropped_set: diff --git a/tests/test_plot_describe.py b/tests/test_plot_describe.py index 1f2dc72..cc71dd6 100644 --- a/tests/test_plot_describe.py +++ b/tests/test_plot_describe.py @@ -194,3 +194,29 @@ def test_envelope_srate_is_the_post_decimation_rate(): """What a sweep buffer must be sized with. Using the pre-decimation rate makes the ring far longer than the data arriving to fill it.""" assert describe_axisarray(envelope(fs=1000.0)).srate == pytest.approx(1000.0) + + +def test_transposed_source_is_described_by_the_buffer_order(): + """The sink rolls the buffered axis to the front, so a source that sent + (ch, time) is held as (time, ch) -- and dims must say so, since a reader + has no way to know it should re-roll them.""" + from ezmsg.tools.plot.describe import describe_mirror + + class FakeMeta: + bvalid, ndim, srate = True, 3, 1000.0 + shape = (2000, 4, 2) # rolled: time first + + class FakeMirror: + meta = FakeMeta() + # What ShMemCircBuff now records: the order the ring actually holds. + dims = ["time", "ch", "metric"] + axes = { + "ch": {"kind": "coord", "data": np.zeros(4, dtype=CHANNEL_DTYPE)}, + "metric": {"kind": "coord", "data": np.array(["min", "max"])}, + } + attrs = {"unit": "uV"} + + shape = describe_mirror(FakeMirror()) + assert shape.n_channels == 4 + assert shape.envelope + assert shape.srate == pytest.approx(1000.0) From 2edae7f6ea1e2979710d297482b288feb9d2a2cf Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 19:26:08 -0400 Subject: [PATCH 06/10] Hide the old sweep before closing its figure Rebuilding logs a rendercanvas "Present finish error: Internal C++ object already deleted". The plot works either way, but the sequence is genuinely loose: figure.close() is followed by deleteLater, which defers the C++ destruction to the next event-loop turn, leaving a window in which rendercanvas can try to present a frame into a widget on its way out. Hiding first narrows it -- a hidden widget receives no paint events -- and is correct ordering regardless of whether it silences the message. Unverified: reproducing needs a live canvas, and there is no GUI harness here. If the message persists, the next thing to try is stopping the canvas scheduler explicitly before close rather than relying on close to do it. --- src/ezmsg/tools/plot/shmem_sweep.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/ezmsg/tools/plot/shmem_sweep.py b/src/ezmsg/tools/plot/shmem_sweep.py index 7bb7178..c4b2645 100644 --- a/src/ezmsg/tools/plot/shmem_sweep.py +++ b/src/ezmsg/tools/plot/shmem_sweep.py @@ -187,8 +187,19 @@ def _effective_poll_hz(poll_hz: float | None, max_fps: float | None) -> float: return DEFAULT_POLL_HZ def _close_figure(self) -> None: + """Stop the old canvas drawing, then close it. + + Hidden first: a hidden widget receives no paint events, which narrows + the window in which rendercanvas can try to present a frame into a Qt + object that is on its way out. deleteLater defers the actual C++ + destruction to the next event-loop turn, so that window is real. + """ if self._sweep is None: return + try: + self._sweep.hide() + except Exception: + logger.debug("hiding the sweep before teardown raised; continuing", exc_info=True) figure = getattr(self._sweep, "_figure", None) if figure is not None: try: From efb8d3b685f69bc42d5be3b4a1b25ac1efcc942e Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 19:55:05 -0400 Subject: [PATCH 07/10] Say which shared-memory segment was created, and why it took retries A channel-selection change logs two "Created shmem at sm_..." lines back to back, and there was no way to tell from them what the two were: the names are sha256-hashed to fit the platform's length limit, so both look like noise. They are the data ring, reallocated because the frame shape changed, and the metadata blob, republished because the ch axis lost channels -- both necessary, neither obvious. Created data ring gen 1 (1000x64) (256000 bytes) at sm_PkyA3Ki... Created stream metadata gen 2 (4448 bytes) at sm_hzb4AM68... The generation and shape are the useful part: they say which reset this was and how big the result is, which is what someone chasing a blanked plot or a memory question actually wants. Also fixes the retry counter, which was never incremented, so every line claimed "in 0 attempts" whether or not a stale name had been cleaned up first -- a number that could only ever read zero. Now the retries are mentioned only when there were some, which is the case worth noticing. --- src/ezmsg/tools/shmem/shmem.py | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/src/ezmsg/tools/shmem/shmem.py b/src/ezmsg/tools/shmem/shmem.py index b0bccd2..d34d52e 100644 --- a/src/ezmsg/tools/shmem/shmem.py +++ b/src/ezmsg/tools/shmem/shmem.py @@ -178,12 +178,16 @@ class ShMemCircBuffState(ez.State): warned_dropped_attrs: typing.Optional[frozenset] = None -def _persist_create_shmem(name: str, size: int) -> SharedMemory: +def _persist_create_shmem(name: str, size: int, purpose: str = "") -> SharedMemory: """ Create a shared memory object, retrying if necessary. Args: name: The name of the shared memory object. size: The size of the shared memory object. + purpose: What this segment is for, for the log line. Names are hashed + to fit the platform's length limit, so without this a reader cannot + tell the data ring from the metadata blob -- and a shape change + recreates both, back to back. Returns: The SharedMemory object. """ @@ -198,13 +202,15 @@ def _persist_create_shmem(name: str, size: int) -> SharedMemory: ) break except FileExistsError: + n_attempts += 1 tmp_shmem = SharedMemory( name=name, create=False, ) tmp_shmem.close() tmp_shmem.unlink() - ez.logger.info(f"Created shmem at {name} in {n_attempts} attempts after {time.time() - t0:.2f} s.") + retried = f" after {n_attempts} stale-name retries," if n_attempts else "" + ez.logger.info(f"Created {purpose or 'shmem'} ({size} bytes) at {name}{retried} in {time.time() - t0:.3f} s.") return result @@ -331,7 +337,7 @@ def _reset_meta(self, reset_generation: bool = True) -> None: # Create the metadata shared memory object. meta_size = int(ctypes.sizeof(ShmemArrMeta)) short_name = shorten_shmem_name(self.SETTINGS.shmem_name) - self.STATE.meta_shmem = _persist_create_shmem(short_name, meta_size) + self.STATE.meta_shmem = _persist_create_shmem(short_name, meta_size, purpose="shmem header") if self.SETTINGS.shmem_name is None: # If the name is None, then we need to get the name from the shared memory object. @@ -404,7 +410,7 @@ def _update_aux_if_needed(self, msg: AxisArray) -> bool: # 0 means "nothing published", so skip it when the uint32 wraps. generation = (self.STATE.meta_struct.meta_generation + 1) % (2**32) or 1 aux_name = shorten_shmem_name(self.SETTINGS.shmem_name + "/meta" + str(generation)) - self.STATE.aux_shmem = _persist_create_shmem(aux_name, len(blob)) + self.STATE.aux_shmem = _persist_create_shmem(aux_name, len(blob), purpose=f"stream metadata gen {generation}") self.STATE.aux_shmem.buf[: len(blob)] = blob # Order matters: the segment is fully written before the header names it, @@ -514,7 +520,12 @@ def _reset_buffer(self, msg: AxisArray) -> None: buff_size = int(n_frames * np.prod(frame_shape) * msg.data.itemsize) buff_shm_name = self.SETTINGS.shmem_name + "/buffer" + str(self.STATE.meta_struct.buffer_generation) short_name = shorten_shmem_name(buff_shm_name) - self.STATE.buffer_shmem = _persist_create_shmem(short_name, buff_size) + self.STATE.buffer_shmem = _persist_create_shmem( + short_name, + buff_size, + purpose=f"data ring gen {self.STATE.meta_struct.buffer_generation} " + f"({'x'.join(str(d) for d in (n_frames,) + frame_shape)})", + ) self.STATE.buffer_arr = np.ndarray( self.STATE.meta_struct.shape[: self.STATE.meta_struct.ndim], dtype=np.dtype(self.STATE.meta_struct.dtype.decode("utf8")), From 12dbaf402b49bd47049907fde08673c8c6950cf8 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 21:47:11 -0400 Subject: [PATCH 08/10] Keep phosphor optional, and say so when it is missing phosphor is what viewer and sigmon plot with; nothing else here needs it. In core it would put fastplotlib, PySide6 and a GPU stack behind every install -- so ezmsg-performance-monitor, a Dash web app, would ship Qt -- and raise the core numpy floor from 1.26 to 2.4.2. It stays in the two extras, now at >=0.7.0 for the envelope input, set_channel_labels, set_channel_labels_visible and SweepConfig.max_fps. What made core look necessary is real, but it is not about imports: console scripts are installed unconditionally, because [project.scripts] has no notion of extras. `pip install ezmsg-tools` puts ezmsg-signal-monitor on the PATH and running it gives a bare ModuleNotFoundError with no hint that an extra exists or what it is called. perfmon has the identical problem with dash. Both entry points now go through a guard that names the extra: ezmsg-signal-monitor needs the optional 'sigmon' dependencies (could not import 'ezmsg.qt'). Install them with: pip install 'ezmsg-tools[sigmon]' It reports the module that was actually missing rather than assuming the extra is the whole story -- an ImportError from inside the CLI is a different problem, and saying which one it was keeps the message honest. Making phosphor properly optional then exposed a bug in ezmsg.tools.plot. Its __init__ eagerly imported the Qt widget, so importing the deliberately Qt-free .describe submodule pulled in a GPU stack anyway -- a submodule import runs its parent's __init__ first. It never showed while phosphor was always installed. ShmemSweepWidget is now resolved lazily via PEP 562 __getattr__, so the public name still works where phosphor is present and .describe is importable where it is not. Verified on a clean install with neither phosphor nor Qt present. --- pyproject.toml | 13 +++++---- src/ezmsg/tools/_entry.py | 42 +++++++++++++++++++++++++++++ src/ezmsg/tools/perfmon/__init__.py | 12 +++++++++ src/ezmsg/tools/plot/__init__.py | 27 +++++++++++++++---- src/ezmsg/tools/sigmon/__init__.py | 12 +++++++++ 5 files changed, 94 insertions(+), 12 deletions(-) create mode 100644 src/ezmsg/tools/_entry.py diff --git a/pyproject.toml b/pyproject.toml index 2cd2857..8cf0b64 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,27 +49,26 @@ perfmon = [ "dash-bootstrap-components>=1.6.0", "ezmsg-baseproc>=1.1.0", ] -# NOTE: ezmsg.tools.plot needs phosphor's envelope support, set_channel_labels, -# set_channel_labels_visible and SweepConfig.max_fps. Bump this floor to the -# release that carries them before publishing. +# ezmsg.tools.plot needs phosphor's envelope input, set_channel_labels, +# set_channel_labels_visible and SweepConfig.max_fps -- all v0.7.0. sigmon = [ "PySide6>=6.7", "pygraphviz>=1.14", "typer>=0.15.1", "ezmsg-qt>=0.2.1", - "phosphor>=0.5.0", + "phosphor>=0.7.0", "pandas", ] viewer = [ "PySide6>=6.7", "typer>=0.15.1", "ezmsg-qt>=0.2.1", - "phosphor>=0.5.0", + "phosphor>=0.7.0", ] [project.scripts] -ezmsg-performance-monitor = "ezmsg.tools.perfmon.cli:main" -ezmsg-signal-monitor = "ezmsg.tools.sigmon.cli:main" +ezmsg-performance-monitor = "ezmsg.tools.perfmon:main" +ezmsg-signal-monitor = "ezmsg.tools.sigmon:main" [build-system] requires = ["hatchling", "hatch-vcs"] diff --git a/src/ezmsg/tools/_entry.py b/src/ezmsg/tools/_entry.py new file mode 100644 index 0000000..680c1b0 --- /dev/null +++ b/src/ezmsg/tools/_entry.py @@ -0,0 +1,42 @@ +"""Launching a console script whose dependencies live in an optional extra. + +Console scripts are installed unconditionally -- ``[project.scripts]`` has no +notion of extras -- so ``pip install ezmsg-tools`` puts commands on the PATH +whose imports are not satisfied. Run one and you get a bare +``ModuleNotFoundError: dash`` with no hint that an extra exists or what it is +called. + +The alternative would be promoting those dependencies to the core install, so +that a Dash web app drags in Qt and a GPU stack for everyone. A clear message +is the cheaper fix. +""" + +import importlib +import os +import sys +import typing + +__all__ = ["run_cli"] + + +def run_cli(module: str, extra: str) -> typing.NoReturn: + """Import ``module`` and call its ``main()``, or explain what is missing. + + :param module: Dotted path of the CLI module to run. + :param extra: The extra that declares this command's dependencies. + """ + try: + cli = importlib.import_module(module) + except ImportError as exc: + command = os.path.basename(sys.argv[0]) or module + missing = getattr(exc, "name", None) + # Name the module that was actually missing rather than assuming the + # extra is the whole story: an ImportError from inside the CLI is a + # different problem, and saying which one it was keeps this honest. + detail = f" (could not import {missing!r})" if missing else "" + raise SystemExit( + f"{command} needs the optional '{extra}' dependencies{detail}.\n" + f"Install them with:\n\n" + f" pip install 'ezmsg-tools[{extra}]'\n" + ) from exc + sys.exit(cli.main()) diff --git a/src/ezmsg/tools/perfmon/__init__.py b/src/ezmsg/tools/perfmon/__init__.py index e69de29..34a25eb 100644 --- a/src/ezmsg/tools/perfmon/__init__.py +++ b/src/ezmsg/tools/perfmon/__init__.py @@ -0,0 +1,12 @@ +"""Performance monitor: a Dash app over ezmsg's profiler output. + +The console script points here rather than at :mod:`.cli` so that a missing +``perfmon`` extra produces an explanation instead of a traceback -- see +:mod:`ezmsg.tools._entry`. +""" + +from .._entry import run_cli + + +def main() -> None: + run_cli("ezmsg.tools.perfmon.cli", "perfmon") diff --git a/src/ezmsg/tools/plot/__init__.py b/src/ezmsg/tools/plot/__init__.py index 40ce497..8b6ff5c 100644 --- a/src/ezmsg/tools/plot/__init__.py +++ b/src/ezmsg/tools/plot/__init__.py @@ -3,13 +3,19 @@ :mod:`.describe` is the pure half -- given dims, axes and attrs, work out what is being plotted -- and imports neither Qt nor phosphor, so it is usable from a topic subscriber, a shared-memory mirror, or a test with neither. -:mod:`.shmem_sweep` is the Qt widget built on it. +:mod:`.shmem_sweep` is the Qt widget built on it, and needs the ``viewer`` or +``sigmon`` extra. -Requires the ``viewer`` or ``sigmon`` extra (PySide6 and phosphor). Importing -this package pulls in Qt, so import :mod:`.describe` directly if that is all -you need. +``ShmemSweepWidget`` is resolved lazily so that importing this package, or +anything under it, does not pull in Qt. Eagerly importing it here would make +``from ezmsg.tools.plot.describe import ...`` fail without phosphor installed, +since importing a submodule runs its parent's ``__init__`` first -- which would +put a GPU stack behind a module that deliberately has no rendering dependency +at all. """ +import typing + from .describe import ( METRIC_KINDS, SWEEP_RENDERABLE_METRICS, @@ -22,7 +28,9 @@ metric_axis, require_sweep_renderable, ) -from .shmem_sweep import ShmemSweepWidget + +if typing.TYPE_CHECKING: # pragma: no cover - import for type checkers only + from .shmem_sweep import ShmemSweepWidget __all__ = [ "METRIC_KINDS", @@ -37,3 +45,12 @@ "metric_axis", "require_sweep_renderable", ] + + +def __getattr__(name: str) -> typing.Any: + """Resolve the Qt widget on first use (PEP 562).""" + if name == "ShmemSweepWidget": + from .shmem_sweep import ShmemSweepWidget + + return ShmemSweepWidget + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/src/ezmsg/tools/sigmon/__init__.py b/src/ezmsg/tools/sigmon/__init__.py index e69de29..b54d427 100644 --- a/src/ezmsg/tools/sigmon/__init__.py +++ b/src/ezmsg/tools/sigmon/__init__.py @@ -0,0 +1,12 @@ +"""Signal monitor: a graph inspector with live plots. + +The console script points here rather than at :mod:`.cli` so that a missing +``sigmon`` extra produces an explanation instead of a traceback -- see +:mod:`ezmsg.tools._entry`. +""" + +from .._entry import run_cli + + +def main() -> None: + run_cli("ezmsg.tools.sigmon.cli", "sigmon") From 1aa408ee391be45eea8069f453490c36f5b6b617 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 23:34:50 -0400 Subject: [PATCH 09/10] Require phosphor 0.8.0 --- pyproject.toml | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 8cf0b64..e8f2d9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,21 +49,19 @@ perfmon = [ "dash-bootstrap-components>=1.6.0", "ezmsg-baseproc>=1.1.0", ] -# ezmsg.tools.plot needs phosphor's envelope input, set_channel_labels, -# set_channel_labels_visible and SweepConfig.max_fps -- all v0.7.0. sigmon = [ "PySide6>=6.7", "pygraphviz>=1.14", "typer>=0.15.1", "ezmsg-qt>=0.2.1", - "phosphor>=0.7.0", + "phosphor>=0.8.0", "pandas", ] viewer = [ "PySide6>=6.7", "typer>=0.15.1", "ezmsg-qt>=0.2.1", - "phosphor>=0.7.0", + "phosphor>=0.8.0", ] [project.scripts] From 8b8cea79dc81bff6843acde6144b641f5ff65ce7 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 23:47:33 -0400 Subject: [PATCH 10/10] fix test --- tests/test_shmem_mirror.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/tests/test_shmem_mirror.py b/tests/test_shmem_mirror.py index 30305ff..e0d591d 100644 --- a/tests/test_shmem_mirror.py +++ b/tests/test_shmem_mirror.py @@ -76,12 +76,14 @@ async def on_signal(self, message: AxisArray) -> typing.AsyncGenerator: CHANNEL_COUNT = 128 CHUNK_SIZE = 64 TOTAL_DURATION = 5.0 +STARTUP_TIMEOUT = 20.0 +SHUTDOWN_TIMEOUT = 20.0 def app(file_path) -> None: change_type = "dtype" chunk_rate = 10.0 - chunk_size = SR // chunk_rate + chunk_size = int(SR // chunk_rate) n_messages = int(TOTAL_DURATION * chunk_rate) comps = { @@ -139,6 +141,9 @@ def test_shmem_mirror_switch_buffer(): START_TIME = time.time() while get_chunk(mirror) is None: + assert ( + time.time() - START_TIME < STARTUP_TIMEOUT + ), f"No data in shared memory after {STARTUP_TIMEOUT} s; the pipeline never produced a message." time.sleep(0.1) print(f"*** Pipeline started in {time.time() - START_TIME:.2f} seconds") @@ -146,7 +151,11 @@ def test_shmem_mirror_switch_buffer(): data_received = collect_data(mirror, TOTAL_DURATION) # Stop bolt and LSL stream - app_thread.join() + app_thread.join(timeout=SHUTDOWN_TIMEOUT) + assert not app_thread.is_alive(), ( + f"Pipeline still running {SHUTDOWN_TIMEOUT} s after the data collection window; " + "it never reached its terminating message count." + ) messages: typing.List[AxisArray] = [_ for _ in message_log(file_path)] file_path.unlink(missing_ok=True)