diff --git a/pyproject.toml b/pyproject.toml index 7c88fcd..e8f2d9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,19 +54,19 @@ sigmon = [ "pygraphviz>=1.14", "typer>=0.15.1", "ezmsg-qt>=0.2.1", - "phosphor>=0.5.0", + "phosphor>=0.8.0", "pandas", ] viewer = [ "PySide6>=6.7", "typer>=0.15.1", "ezmsg-qt>=0.2.1", - "phosphor>=0.5.0", + "phosphor>=0.8.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"] @@ -99,5 +99,10 @@ known-first-party = ["ezmsg.tools"] known-third-party = ["ezmsg"] [tool.uv.sources] +# 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" } 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 new file mode 100644 index 0000000..8b6ff5c --- /dev/null +++ b/src/ezmsg/tools/plot/__init__.py @@ -0,0 +1,56 @@ +"""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, and needs the ``viewer`` or +``sigmon`` extra. + +``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, + MetricSpec, + StreamShape, + UnsupportedMetricError, + describe_axisarray, + describe_mirror, + flatten_for_plot, + metric_axis, + require_sweep_renderable, +) + +if typing.TYPE_CHECKING: # pragma: no cover - import for type checkers only + from .shmem_sweep import ShmemSweepWidget + +__all__ = [ + "METRIC_KINDS", + "SWEEP_RENDERABLE_METRICS", + "MetricSpec", + "ShmemSweepWidget", + "StreamShape", + "UnsupportedMetricError", + "describe_axisarray", + "describe_mirror", + "flatten_for_plot", + "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/plot/describe.py b/src/ezmsg/tools/plot/describe.py new file mode 100644 index 0000000..5210e38 --- /dev/null +++ b/src/ezmsg/tools/plot/describe.py @@ -0,0 +1,277 @@ +"""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__ = [ + "METRIC_AXIS_CANDIDATES", + "METRIC_KINDS", + "SWEEP_RENDERABLE_METRICS", + "MetricSpec", + "StreamShape", + "UnsupportedMetricError", + "describe_axisarray", + "describe_mirror", + "flatten_for_plot", + "metric_axis", + "require_sweep_renderable", +] + +# 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. +METRIC_AXIS_CANDIDATES = ("metric", "minmax", "bound", "stat") + +# 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): + """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.""" + + 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 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 METRIC_AXIS_CANDIDATES: + return None + data = _axis_data(axes.get(name)) + if data is None: + return 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: + """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) + metric = metric_axis(dims, axes) + metric_name = metric.axis if metric is not None else None + + # 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, metric_name): + 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, + metric=metric, + 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]) + # 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( + list(mirror.dims or []), + 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 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. + """ + 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,) + 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 new file mode 100644 index 0000000..c4b2645 --- /dev/null +++ b/src/ezmsg/tools/plot/shmem_sweep.py @@ -0,0 +1,310 @@ +"""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, + UnsupportedMetricError, + describe_mirror, + flatten_for_plot, + require_sweep_renderable, +) + +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._error: str | 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() + + @property + 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: + """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 _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. + + 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: + """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: + 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 + + 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: + 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 + if self._sweep is None: + self._build(shape) + return + if previous == shape: + return + if self._needs_rebuild(previous, shape): + 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) + self.on_plot_built() diff --git a/src/ezmsg/tools/shmem/shmem.py b/src/ezmsg/tools/shmem/shmem.py index d80ebcf..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. @@ -378,7 +384,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: @@ -399,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, @@ -509,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")), 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") diff --git a/src/ezmsg/tools/sigmon/cli.py b/src/ezmsg/tools/sigmon/cli.py index ac798e7..900564d 100644 --- a/src/ezmsg/tools/sigmon/cli.py +++ b/src/ezmsg/tools/sigmon/cli.py @@ -18,6 +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, + require_sweep_renderable, +) from ezmsg.tools.sigmon.dag_widget import DAGWidget logger = logging.getLogger(__name__) @@ -89,6 +94,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 +111,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 +129,14 @@ 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) + require_sweep_renderable(shape) + 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 +223,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..e4edac1 100644 --- a/src/ezmsg/tools/viewer/cli.py +++ b/src/ezmsg/tools/viewer/cli.py @@ -19,6 +19,12 @@ ) from PySide6.QtWidgets import QApplication, QMainWindow, QWidget +from ezmsg.tools.plot.describe import ( + describe_axisarray, + flatten_for_plot, + require_sweep_renderable, +) + logger = logging.getLogger(__name__) GRAPH_IP = "127.0.0.1" @@ -105,6 +111,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 +129,18 @@ 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) + require_sweep_renderable(shape) + 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 +173,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..cc71dd6 --- /dev/null +++ b/tests/test_plot_describe.py @@ -0,0 +1,222 @@ +"""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 ( + UnsupportedMetricError, + describe_axisarray, + flatten_for_plot, + metric_axis, + require_sweep_renderable, +) + +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_ax(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_ax(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_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 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_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 + + 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 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 metric_axis(["time", "ch", "metric"], {"metric": as_dict}).kind == "minmax" + + +# ---- 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) + + +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) 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) 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