diff --git a/src/phosphor/__init__.py b/src/phosphor/__init__.py index 7cb5226..4709801 100644 --- a/src/phosphor/__init__.py +++ b/src/phosphor/__init__.py @@ -3,14 +3,17 @@ from .__version__ import __version__ as __version__ from .channel_plot import ChannelPlotWidget from .controls import ChannelPlotControlsWidget +from .overlays import ChannelLabelOverlay, ScaleBarOverlay from .scatter_widget import ScatterConfig, ScatterWidget from .spectrum_widget import SpectrumConfig, SpectrumWidget from .sweep_buffer import SweepEvent from .sweep_widget import SweepConfig, SweepWidget __all__ = [ + "ChannelLabelOverlay", "ChannelPlotControlsWidget", "ChannelPlotWidget", + "ScaleBarOverlay", "ScatterConfig", "ScatterWidget", "SpectrumConfig", diff --git a/src/phosphor/channel_plot.py b/src/phosphor/channel_plot.py index 264268a..ac53708 100644 --- a/src/phosphor/channel_plot.py +++ b/src/phosphor/channel_plot.py @@ -2,11 +2,16 @@ from __future__ import annotations +import logging + import fastplotlib as fpl from PySide6.QtCore import Qt from PySide6.QtWidgets import QLabel, QToolTip, QVBoxLayout, QWidget from .constants import CHANNEL_COLORS +from .overlays import ChannelLabelOverlay, ScaleBarOverlay + +logger = logging.getLogger(__name__) __all__ = ["ChannelPlotWidget"] @@ -15,7 +20,7 @@ class ChannelPlotWidget(QWidget): """Base widget for fastplotlib-rendered multichannel plots. Provides canvas setup, channel scrolling (scroll / ↑↓ / PgUp/PgDn / [ ]), - amplitude zoom (Shift+scroll / - = A), and range label overlay. + amplitude zoom (Shift+scroll / - =), and range label overlay. Subclass contract: @@ -40,7 +45,6 @@ def __init__( super().__init__(parent) self._n_channels = n_channels self._channel_labels = channel_labels - self._autoscale_enabled = True # Layout layout = QVBoxLayout(self) @@ -67,6 +71,20 @@ def __init__( # _buffer is set by the subclass before calling _init_rendering() self._buffer = None + # Overlays, created on demand. Kept as None until asked for so a plot + # that never wants them pays nothing. + self._label_overlay: ChannelLabelOverlay | None = None + self._scale_bar_overlay: ScaleBarOverlay | None = None + # Cached world->screen-y projection, keyed on the inputs that can change + # it. map_world_to_screen is a pygfx round-trip; at 60 fps it is worth + # not repeating while nothing has moved. + self._projection: tuple[float, float] | None = None + self._projection_key: tuple | None = None + self._scale_bar_text: str = "" + # Identity of the label list last pushed to the overlay, so a relabel + # via update_config is picked up without copying the list every frame. + self._label_source: object = None + # ------------------------------------------------------------------ # Subclass hook: call after buffer is ready # ------------------------------------------------------------------ @@ -106,7 +124,14 @@ def _update_graphics(self) -> None: raise NotImplementedError def _apply_auto_scale(self) -> None: - """Set camera to fit content. Override in subclass for fast path.""" + """Frame the camera on the current content. Override for a fast path. + + Called every frame, unconditionally. It is the only thing that moves + the camera -- ``_init_rendering`` disables the pan/zoom controller -- + so skipping it does not hand control to the user, it strands the view. + Time and amplitude zoom work by changing what the data occupies and + letting this follow. + """ self._subplot.auto_scale(maintain_aspect=False, zoom=1.0) def _on_ctrl_scroll(self, delta: float) -> None: @@ -118,8 +143,152 @@ def _on_ctrl_scroll(self, delta: float) -> None: def _animation_callback(self) -> None: self._update_graphics() - if self._autoscale_enabled: - self._apply_auto_scale() + self._apply_auto_scale() + self._sync_overlays() + + # ------------------------------------------------------------------ + # Overlays + # ------------------------------------------------------------------ + + def set_channel_labels(self, labels: list[str] | None) -> None: + """Replace the per-channel labels used by the tooltip and the overlay. + + Labels often arrive after the plot is built -- a source announces its + channel names on its own schedule -- so this exists to avoid callers + assigning to ``_channel_labels`` and hoping the overlay notices. + """ + self._channel_labels = list(labels) if labels is not None else None + self._push_channel_labels() + + @property + def channel_labels_visible(self) -> bool: + """Whether per-trace identifiers are currently drawn.""" + return self._label_overlay is not None and self._label_overlay.isVisible() + + def set_channel_labels_visible(self, visible: bool) -> None: + """Show or hide per-trace identifiers drawn over the canvas. + + Uses the ``channel_labels`` the plot was configured with. Positioning + and font sizing follow the live camera, so labels stay on their traces + as the user scrolls, pages, or zooms. + + Worth having a way to turn off: the labels sit on opaque chips, so on a + dense trace they cover signal, and once the user knows which channel is + which they mostly want them gone. + """ + if not visible: + if self._label_overlay is not None: + self._label_overlay.hide() + return + if self._label_overlay is None: + self._label_overlay = ChannelLabelOverlay(self._fpl_widget) + self._label_overlay.setGeometry(self._fpl_widget.rect()) + self._push_channel_labels() + self._label_overlay.show() + self._label_overlay.raise_() + self._sync_overlays() + + def set_scale_bar_text(self, text: str | None) -> None: + """Show a calibration bar one row-height tall, labelled *text*. + + The caller supplies the text because what a row-height means depends on + the signal's units, which this widget does not know. ``None`` or an + empty string hides the bar. + """ + if not text: + if self._scale_bar_overlay is not None: + self._scale_bar_overlay.set_bar(0.0, "") + self._scale_bar_overlay.hide() + return + if self._scale_bar_overlay is None: + self._scale_bar_overlay = ScaleBarOverlay(self._fpl_widget) + self._scale_bar_overlay.setGeometry(self._fpl_widget.rect()) + self._scale_bar_text = text + self._scale_bar_overlay.show() + self._scale_bar_overlay.raise_() + self._sync_overlays() + + def _push_channel_labels(self) -> None: + """Forward the plot's labels to the overlay when they have been replaced. + + Keyed on identity rather than value: update_config swaps the list, and + comparing 256 strings every frame to notice would cost more than the + feature does. + """ + if self._label_overlay is None or self._channel_labels is self._label_source: + return + self._label_source = self._channel_labels + self._label_overlay.set_labels(list(self._channel_labels or [])) + + def _screen_y_projection(self) -> tuple[float | None, float | None]: + """Affine ``screen_y = y0 + slope * world_y`` for the current camera. + + Returns ``(None, None)`` if the camera cannot be probed, which the + overlay handles by falling back to its analytic layout. + """ + subplot = getattr(self, "_subplot", None) + if subplot is None: + return None, None + buf = self._buffer + size = self._fpl_widget.size() + camera = getattr(subplot, "camera", None) + key = ( + size.width(), + size.height(), + getattr(buf, "n_visible", None), + getattr(self, "_z_offset_scale", 1.0), + float(camera.height) if camera is not None else None, + float(camera.world.position[1]) if camera is not None else None, + ) + if self._projection is not None and key == self._projection_key: + return self._projection + + try: + # x is irrelevant to screen-y for the axis-aligned ortho camera. + y_at_0 = float(subplot.map_world_to_screen((0.0, 0.0, 0.0))[1]) + y_at_1 = float(subplot.map_world_to_screen((0.0, 1.0, 0.0))[1]) + except Exception: # noqa: BLE001 - pygfx call with no enumerated failures + # In a paint path: losing the labels' precise placement is better + # than losing the frame. + logger.debug("map_world_to_screen failed; overlays fall back to analytic layout", exc_info=True) + return None, None + + self._projection = (y_at_0, y_at_1 - y_at_0) + self._projection_key = key + return self._projection + + def _sync_overlays(self) -> None: + """Keep overlays sized to the canvas and in step with the view.""" + if self._label_overlay is None and self._scale_bar_overlay is None: + return + rect = self._fpl_widget.rect() + for overlay in (self._label_overlay, self._scale_bar_overlay): + if overlay is not None and overlay.size() != self._fpl_widget.size(): + overlay.setGeometry(rect) + + buf = self._buffer + if buf is None: + return + y0, slope = self._screen_y_projection() + z_scale = getattr(self, "_z_offset_scale", 1.0) or 1.0 + + if self._label_overlay is not None: + self._push_channel_labels() + self._label_overlay.set_view( + getattr(buf, "channel_offset", 0), + getattr(buf, "n_visible", 0), + getattr(buf, "channel_order", "top_down") == "top_down", + y0, + slope, + z_scale, + ) + + if self._scale_bar_overlay is not None: + text = self._scale_bar_text + # A row spans ±0.5 in normalized units, so half a row's world + # distance is the bar the caller's text describes. + length = 0.0 if slope is None else 0.5 * abs(slope) + self._scale_bar_overlay.set_bar(length, text) # ------------------------------------------------------------------ # Event handlers (fpl native events) @@ -151,11 +320,6 @@ def _on_key_down(self, key: str) -> None: elif key == "=": self._zoom_amplitude(1.25) - elif key in ("a", "A"): - self._autoscale_enabled = not self._autoscale_enabled - if self._autoscale_enabled: - self._apply_auto_scale() - self._update_range_label() def _on_wheel_event(self, event) -> None: @@ -194,6 +358,36 @@ def set_mouse_enabled(self, enabled: bool) -> None: self._renderer.remove_event_handler(self._on_wheel_event, "wheel") self._renderer.remove_event_handler(self._on_pointer_move_event, "pointer_move") + def set_max_fps(self, fps: float | None) -> None: + """Cap the canvas render rate. + + A scrolling trace reads smooth well below the display refresh, and + render cost falls roughly in proportion, so this is the cheapest + performance knob a caller has. ``None`` leaves whatever rendercanvas + was already doing; ``fps <= 0`` uncaps (``update_mode="fastest"``). + + Safe to call before or after the canvas exists; a backend that predates + ``set_update_mode`` is warned about once and otherwise ignored, since + losing the cap is a performance regression rather than a broken plot. + """ + if fps is None: + return + canvas = getattr(self._figure, "canvas", None) + set_update_mode = getattr(canvas, "set_update_mode", None) + if set_update_mode is None: + logger.warning("Canvas has no set_update_mode; leaving the render rate at its default.") + return + try: + if fps <= 0: + set_update_mode("fastest") + else: + # "continuous" is the mode that honours max_fps; "ondemand" + # ignores it and redraws only on events, which a live sweep + # would then never do. + set_update_mode("continuous", max_fps=float(fps)) + except Exception: + logger.exception("Failed to set the canvas render rate; leaving it at its default.") + # ------------------------------------------------------------------ # Amplitude zoom helper # ------------------------------------------------------------------ @@ -211,7 +405,6 @@ def _zoom_amplitude(self, factor: float) -> None: return camera = self._subplot.camera camera.world.scale_y *= factor - self._autoscale_enabled = False # ------------------------------------------------------------------ # Mouse hover tooltip @@ -245,7 +438,7 @@ def _handle_mouse_move(self, event) -> None: label = labels[abs_ch] if labels and abs_ch < len(labels) else f"Ch {abs_ch}" rgba = CHANNEL_COLORS[ch_index % len(CHANNEL_COLORS)] - hex_color = f"#{int(rgba[0]*255):02x}{int(rgba[1]*255):02x}{int(rgba[2]*255):02x}" + hex_color = f"#{int(rgba[0] * 255):02x}{int(rgba[1] * 255):02x}{int(rgba[2] * 255):02x}" html = f'\u25a0 {label}' from PySide6.QtCore import QPoint diff --git a/src/phosphor/constants.py b/src/phosphor/constants.py index 875ba5a..021a629 100644 --- a/src/phosphor/constants.py +++ b/src/phosphor/constants.py @@ -8,6 +8,10 @@ CURSOR_GAP_COLUMNS = 5 BG_COLOR = (0.10, 0.10, 0.12, 1.0) CURSOR_COLOR = (0.25, 0.25, 0.28, 0.85) +# Cursor line width in pixels. The geometry-derived width is in time-world +# units and collapses to roughly one pixel, which is hard to see against a +# dense trace; this is applied on top. +CURSOR_THICKNESS = 2.0 DEFAULT_MAX_EVENTS = 500 # max stored events (deque capacity) EVENT_POOL_SIZE = 64 # max simultaneously rendered event ticks diff --git a/src/phosphor/controls.py b/src/phosphor/controls.py index c68b1c9..3ea75ff 100644 --- a/src/phosphor/controls.py +++ b/src/phosphor/controls.py @@ -54,6 +54,12 @@ def __init__( layout.setContentsMargins(2, 1, 2, 1) layout.setSpacing(3) + # Named insertion points, so a subclass can put its own widgets next to + # the group they belong with. Without these the only way to extend this + # toolbar is to scan the layout for a QLabel with the right text, which + # breaks the moment a label is renamed or reordered. + self._slots: dict[str, int] = {} + # Channel scrolling: up/down by one, and page up/down by n_visible. layout.addWidget(self._make_label("Channel")) self._btn_ch_up = self._make_button("\u2191", "Scroll up one channel", self._on_ch_up) @@ -64,6 +70,7 @@ def __init__( layout.addWidget(self._btn_ch_down) layout.addWidget(self._btn_pg_up) layout.addWidget(self._btn_pg_down) + self._mark_slot("channel", layout) layout.addWidget(self._make_separator()) @@ -79,6 +86,7 @@ def __init__( layout.addWidget(self._spin_visible) layout.addWidget(self._make_button("/2", "Halve visible channels", self._on_visible_halve)) layout.addWidget(self._make_button("x2", "Double visible channels", self._on_visible_double)) + self._mark_slot("visible", layout) layout.addWidget(self._make_separator()) @@ -86,6 +94,7 @@ def __init__( layout.addWidget(self._make_label("Amplitude")) layout.addWidget(self._make_button("\u2212", "Shrink amplitude", lambda: self._plot._zoom_amplitude(0.8))) layout.addWidget(self._make_button("+", "Grow amplitude", lambda: self._plot._zoom_amplitude(1.25))) + self._mark_slot("amplitude", layout) # Time zoom — only present if the plot supports it (sweep / spectrum). if hasattr(plot, "_time_zoom"): @@ -93,19 +102,25 @@ def __init__( layout.addWidget(self._make_label("Time")) layout.addWidget(self._make_button("\u2212", "Zoom time out (longer span)", lambda: plot._time_zoom(2.0))) layout.addWidget(self._make_button("+", "Zoom time in (shorter span)", lambda: plot._time_zoom(0.5))) + self._mark_slot("time", layout) layout.addWidget(self._make_separator()) - # Autoscale toggle. - self._btn_auto = QtWidgets.QToolButton() - self._btn_auto.setText("Auto") - self._btn_auto.setCheckable(True) - self._btn_auto.setChecked(plot._autoscale_enabled) - self._btn_auto.setToolTip("Toggle camera autoscale (key: A)") - self._btn_auto.toggled.connect(self._on_auto_toggled) - layout.addWidget(self._btn_auto) - + # Channel-label toggle. The labels sit on opaque chips, so on a dense + # trace they cover signal; once the user knows which channel is which, + # they mostly want them gone. + self._btn_labels = QtWidgets.QToolButton() + self._btn_labels.setText("Labels") + self._btn_labels.setCheckable(True) + self._btn_labels.setChecked(plot.channel_labels_visible) + self._btn_labels.setToolTip("Show per-channel labels over the traces") + self._btn_labels.toggled.connect(self._plot.set_channel_labels_visible) + layout.addWidget(self._btn_labels) + self._mark_slot("labels", layout) + + self.add_controls(layout) layout.addStretch(1) + self._mark_slot("end", layout) # Periodically resync widget state with the buffer (other inputs — # keyboard, mouse — also mutate it). 200 ms is plenty for a panel. @@ -114,6 +129,40 @@ def __init__( self._sync_timer.timeout.connect(self._sync_from_buffer) self._sync_timer.start() + # ---- extension points --------------------------------------------- + + def add_controls(self, layout: QtWidgets.QHBoxLayout) -> None: + """Append subclass widgets, before the trailing stretch. + + Called once during ``__init__``. Override for controls that belong at + the end of the bar; use :meth:`insert_after` for ones that belong beside + an existing group. + """ + + def insert_after(self, slot: str, widget: QtWidgets.QWidget) -> None: + """Insert *widget* immediately after a named group. + + Slots are ``"channel"``, ``"visible"``, ``"amplitude"``, ``"time"`` + (absent unless the plot supports time zoom), ``"labels"``, and + ``"end"``. Later slots shift as earlier ones grow, so inserting is + order-independent. + + :raises KeyError: for an unknown slot, naming the ones that exist -- + silently dropping the widget would be worse, and ``"time"`` + genuinely is conditional. + """ + if slot not in self._slots: + raise KeyError(f"unknown control slot {slot!r}; available: {sorted(self._slots)}") + index = self._slots[slot] + self.layout().insertWidget(index, widget) + for name, pos in self._slots.items(): + if pos >= index: + self._slots[name] = pos + 1 + + def _mark_slot(self, name: str, layout: QtWidgets.QHBoxLayout) -> None: + """Record the current end of *layout* as the named insertion point.""" + self._slots[name] = layout.count() + # ---- helpers ------------------------------------------------------ def _make_button(self, text: str, tip: str, slot) -> QtWidgets.QToolButton: @@ -173,13 +222,6 @@ def _on_visible_double(self) -> None: buf.set_n_visible(min(buf.n_channels, buf.n_visible * 2)) self._plot._update_range_label() - # ---- autoscale ---------------------------------------------------- - - def _on_auto_toggled(self, on: bool) -> None: - self._plot._autoscale_enabled = on - if on: - self._plot._apply_auto_scale() - # ---- periodic resync --------------------------------------------- def _sync_from_buffer(self) -> None: @@ -191,7 +233,7 @@ def _sync_from_buffer(self) -> None: self._spin_visible.setRange(1, buf.n_channels) self._spin_visible.setValue(buf.n_visible) self._spin_visible.blockSignals(False) - if self._btn_auto.isChecked() != self._plot._autoscale_enabled: - self._btn_auto.blockSignals(True) - self._btn_auto.setChecked(self._plot._autoscale_enabled) - self._btn_auto.blockSignals(False) + if self._btn_labels.isChecked() != self._plot.channel_labels_visible: + self._btn_labels.blockSignals(True) + self._btn_labels.setChecked(self._plot.channel_labels_visible) + self._btn_labels.blockSignals(False) diff --git a/src/phosphor/overlays.py b/src/phosphor/overlays.py new file mode 100644 index 0000000..9a944cc --- /dev/null +++ b/src/phosphor/overlays.py @@ -0,0 +1,263 @@ +"""Qt overlays painted on top of a plot canvas. + +fastplotlib draws the data; these draw the things that have to sit *on* the +data and stay legible — per-trace identifiers and an amplitude calibration bar. +Qt rather than GPU graphics because text is the point: crisp glyphs at arbitrary +sizes, hinted and antialiased by the platform, with no vertex churn as the view +scrolls. + +Each is parented to the figure's Qt widget (the same trick +:class:`~phosphor.channel_plot.ChannelPlotWidget` already uses for its range +label), click-through so the plot keeps receiving wheel and hover events, and +told where the traces are by the owning plot — see +``ChannelPlotWidget._sync_overlays``. They hold no reference to the plot, which +is what keeps them reusable and testable. +""" + +from __future__ import annotations + +import math + +from PySide6 import QtCore, QtGui, QtWidgets + +__all__ = ["ChannelLabelOverlay", "ScaleBarOverlay"] + +# Font size bounds in pixels. Between them the label is scaled to about half the +# per-trace spacing, so it reads as centred in its row. +# +# The upper bound is not cosmetic. Each label sits on an opaque chip, so at a +# low channel count -- where rows are hundreds of pixels tall -- scaling with the +# row would paint a large block over the very signal the label refers to. Past +# ordinary reading size a bigger label conveys nothing, so it stops growing. +MIN_LABEL_FONT_PX = 8 +MAX_LABEL_FONT_PX = 14 + +LABEL_TEXT_COLOR = (205, 205, 210) +LABEL_CHIP_COLOR = (25, 25, 30, 185) + +# Scale bar layout, in canvas pixels. +SCALE_BAR_RIGHT_MARGIN = 14 +# Gap from the bar's foot to the bottom of the canvas. The bar sits down here +# rather than centred because the middle of the plot is where the traces are; +# a calibration mark belongs out of the way, in the corner, like a map's. +SCALE_BAR_BOTTOM_MARGIN = 24 +SCALE_BAR_SERIF_HALF = 4 +SCALE_BAR_TEXT_GAP = 6 +SCALE_BAR_MIN_DRAW_PX = 1.0 +SCALE_BAR_FONT_PX = 10 + + +class _CanvasOverlay(QtWidgets.QWidget): + """Shared setup: transparent, click-through, no background.""" + + def __init__(self, parent: QtWidgets.QWidget | None = None) -> None: + super().__init__(parent) + # Click-through so the plot keeps receiving wheel/hover, and no opaque + # background so only the painted marks show. + self.setAttribute(QtCore.Qt.WidgetAttribute.WA_TransparentForMouseEvents) + self.setAttribute(QtCore.Qt.WidgetAttribute.WA_NoSystemBackground) + self.setAttribute(QtCore.Qt.WidgetAttribute.WA_TranslucentBackground) + + +class ChannelLabelOverlay(_CanvasOverlay): + """Per-trace identifiers, drawn beside each channel's baseline. + + Feed it the *full* label list via :meth:`set_labels` and the current + scroll/zoom window via :meth:`set_view`; it indexes the list by absolute + channel so scrolling needs no reslicing by the caller. + """ + + def __init__( + self, + parent: QtWidgets.QWidget | None = None, + *, + min_font_px: int = MIN_LABEL_FONT_PX, + max_font_px: int = MAX_LABEL_FONT_PX, + ) -> None: + super().__init__(parent) + self._min_font_px = int(min_font_px) + self._max_font_px = int(max_font_px) + self._labels: list[str] = [] + self._offset = 0 + self._n_visible = 0 + self._top_down = True + # Affine canvas-y mapping supplied by the owner: + # screen_y(world_y) = y0 + slope * world_y + # None until the first render can measure the camera. + self._y0: float | None = None + self._slope: float | None = None + self._z_scale = 1.0 + + def set_labels(self, labels: list[str]) -> None: + """Set the full per-absolute-channel label list.""" + labels = list(labels) + if labels == self._labels: + return + self._labels = labels + self.update() + + def set_view( + self, + offset: int, + n_visible: int, + top_down: bool, + y0: float | None = None, + slope: float | None = None, + z_scale: float = 1.0, + ) -> None: + """Update the visible window and the world→screen-y mapping. + + ``y0``/``slope`` come from the plot's actual camera, so labels land on + the trace baselines rather than on a re-derived guess. Without them the + overlay falls back to reconstructing the layout analytically, which is + right until someone pans or zooms. Repaints only on change. + """ + state = (offset, n_visible, top_down, y0, slope, z_scale) + if state == (self._offset, self._n_visible, self._top_down, self._y0, self._slope, self._z_scale): + return + self._offset, self._n_visible, self._top_down = offset, n_visible, top_down + self._y0, self._slope, self._z_scale = y0, slope, z_scale + self.update() + + def _row_geometry(self, n: int, h: int) -> tuple[float, float]: + """``(y0, slope)`` mapping world-y to canvas-y for the current view. + + Prefers the measured camera mapping. The fallback reproduces the + autoscaled layout: rows at ``y = row * z_scale`` with a symmetric margin + of ``max((n - 1) * z_scale * 0.05, 0.5)``, filling the height. + """ + z = self._z_scale or 1.0 + if self._y0 is not None and self._slope is not None and self._slope != 0.0: + return self._y0, self._slope + top = (n - 1) * z + margin = max(top * 0.05, 0.5) + span = top + 2 * margin + # Higher world-y maps nearer the top, so the slope is negative. + slope = -h / span + y0 = (top + margin) / span * h + return y0, slope + + def paintEvent(self, event) -> None: + n = self._n_visible + h = self.height() + if not self._labels or n <= 0 or h <= 0: + return + + z = self._z_scale or 1.0 + y0, slope = self._row_geometry(n, h) + row_span_px = abs(slope * z) + if row_span_px <= 0: + return + font_px = max(self._min_font_px, min(round(float(0.5 * row_span_px)), self._max_font_px)) + + painter = QtGui.QPainter(self) + painter.setRenderHint(QtGui.QPainter.RenderHint.TextAntialiasing) + font = QtGui.QFont() + font.setStyleHint(QtGui.QFont.StyleHint.Monospace) + font.setPixelSize(font_px) + painter.setFont(font) + fm = painter.fontMetrics() + + # Once a row is shorter than the text is tall, labelling every row would + # overlap into an unreadable smear; skip rows instead. + step = max(1, math.ceil(fm.height() / row_span_px)) + + text_color = QtGui.QColor(*LABEL_TEXT_COLOR) + chip_color = QtGui.QColor(*LABEL_CHIP_COLOR) + + for i in range(0, n, step): + abs_ch = self._offset + i + if abs_ch >= len(self._labels): + break + text = self._labels[abs_ch] + if not text: + continue + # top_down puts visible row 0 at the highest world-y. + world_y = ((n - 1 - i) if self._top_down else i) * z + y_center = y0 + slope * world_y + baseline = round(float(y_center + fm.ascent() / 2.0 - fm.descent() / 2.0)) + tw = fm.horizontalAdvance(text) + + painter.setPen(QtCore.Qt.PenStyle.NoPen) + painter.setBrush(chip_color) + painter.drawRect(0, baseline - fm.ascent(), tw + 6, fm.height()) + painter.setPen(text_color) + painter.drawText(3, baseline, text) + + painter.end() + + +class ScaleBarOverlay(_CanvasOverlay): + """A labelled vertical calibration bar against the canvas' right edge. + + The owner supplies both the pixel length and the text, because what a + row-height *means* is a property of the signal's units, which a plotting + widget does not know. At high channel counts the bar is too short to read; + it becomes a usable ruler once the user pages down to fewer traces. + """ + + def __init__(self, parent: QtWidgets.QWidget | None = None) -> None: + super().__init__(parent) + self._length_px: float = 0.0 + self._text: str = "" + + def set_bar(self, length_px: float, text: str) -> None: + """Set the bar's pixel length and label. + + A non-positive length or an empty label hides it. Repaints on change. + """ + length_px = max(0.0, float(length_px)) + if length_px == self._length_px and text == self._text: + return + self._length_px, self._text = length_px, text + self.update() + + def paintEvent(self, event) -> None: + length = self._length_px + h, w = self.height(), self.width() + if not self._text or length < SCALE_BAR_MIN_DRAW_PX or h <= 0 or w <= 0: + return + + x = w - SCALE_BAR_RIGHT_MARGIN + # Anchored at its foot, growing upward: the bottom edge stays put as the + # amplitude scale changes, so the bar reads as a ruler standing on the + # canvas floor rather than something that drifts when you zoom. + y_bot = h - SCALE_BAR_BOTTOM_MARGIN + y_top = y_bot - length + y_center = (y_top + y_bot) / 2.0 + + painter = QtGui.QPainter(self) + painter.setRenderHint(QtGui.QPainter.RenderHint.Antialiasing) + painter.setRenderHint(QtGui.QPainter.RenderHint.TextAntialiasing) + + line_color = QtGui.QColor(*LABEL_TEXT_COLOR) + chip_color = QtGui.QColor(*LABEL_CHIP_COLOR) + pen = QtGui.QPen(line_color) + pen.setWidth(2) + painter.setPen(pen) + + # Vertical bar with serifs at both ends. + painter.drawLine(int(x), round(float(y_top)), int(x), round(float(y_bot))) + painter.drawLine( + int(x - SCALE_BAR_SERIF_HALF), round(float(y_top)), int(x + SCALE_BAR_SERIF_HALF), round(float(y_top)) + ) + painter.drawLine( + int(x - SCALE_BAR_SERIF_HALF), round(float(y_bot)), int(x + SCALE_BAR_SERIF_HALF), round(float(y_bot)) + ) + + # Label, right-aligned to the left of the bar and vertically centred. + font = QtGui.QFont() + font.setPixelSize(SCALE_BAR_FONT_PX) + painter.setFont(font) + fm = painter.fontMetrics() + tw = fm.horizontalAdvance(self._text) + text_right = x - SCALE_BAR_SERIF_HALF - SCALE_BAR_TEXT_GAP + text_left = text_right - tw + baseline = round(float(y_center + fm.ascent() / 2.0 - fm.descent() / 2.0)) + + painter.setPen(QtCore.Qt.PenStyle.NoPen) + painter.setBrush(chip_color) + painter.drawRect(int(text_left - 3), int(baseline - fm.ascent()), tw + 6, fm.height()) + painter.setPen(line_color) + painter.drawText(int(text_left), baseline, self._text) + painter.end() diff --git a/src/phosphor/sweep_buffer.py b/src/phosphor/sweep_buffer.py index 2f79644..8e549f8 100644 --- a/src/phosphor/sweep_buffer.py +++ b/src/phosphor/sweep_buffer.py @@ -29,7 +29,12 @@ def __init__( max_events: int = DEFAULT_MAX_EVENTS, channel_order: str = "top_down", amplitude_scale: float = 1.0, + envelope: bool = False, ): + # When True, pushed data is ``(n_samples, n_channels, 2)`` -- a + # already-reduced (min, max) pair per sample rather than a raw value. + # See :meth:`push_data`. + self.envelope = bool(envelope) self.n_channels = n_channels self.srate = srate self.display_dur = display_dur @@ -69,7 +74,10 @@ def _allocate(self): self.n_columns = min(self._configured_n_columns, self.total_raw_samples) self.samples_per_column = self.total_raw_samples / self.n_columns - self.raw_buffer = np.zeros((self.total_raw_samples, self.n_visible), dtype=np.float32) + raw_shape = (self.total_raw_samples, self.n_visible) + if self.envelope: + raw_shape += (2,) + self.raw_buffer = np.zeros(raw_shape, dtype=np.float32) self.display_mins = np.zeros((self.n_columns, self.n_visible), dtype=np.float32) self.display_maxs = np.zeros((self.n_columns, self.n_visible), dtype=np.float32) @@ -91,6 +99,21 @@ def _allocate(self): def push_data(self, data: np.ndarray, timestamps=None) -> None: """Push new samples. data shape: (n_samples, n_channels). Thread-safe. + In ``envelope`` mode the shape is ``(n_samples, n_channels, 2)``, where + the trailing pair is an already-reduced ``(min, max)`` for that sample's + interval. That is the shape a min/max decimator upstream produces, and + feeding it here means the reduction happens once, near the source, + instead of shipping the full-rate signal across a process boundary only + to discard most of it. Column reduction then takes the min of the mins + and the max of the maxes, which is the same envelope at a coarser + resolution -- so the display is identical to what the raw signal would + have drawn, as long as the upstream buckets are finer than a column. + + The buffer's ``srate`` describes the stream being pushed, so in envelope + mode it is the *bucket* rate, not the rate before decimation. Sizing + from the pre-decimation rate would make the ring ``factor`` times longer + than the data arriving to fill it, and the sweep would sit mostly empty. + *timestamps* sets the time basis for event alignment: - ``None`` — elapsed time increments by ``n_samples / srate``. @@ -103,14 +126,26 @@ def push_data(self, data: np.ndarray, timestamps=None) -> None: return with self._lock: - n_samples = data.shape[0] - n_ch = data.shape[1] if data.ndim > 1 else 1 - if data.ndim == 1: + if self.envelope: + if data.ndim != 3 or data.shape[2] != 2: + raise ValueError( + f"envelope buffer expects (n_samples, n_channels, 2), got {data.shape}. " + "Build the buffer with envelope=False to push raw samples." + ) + elif data.ndim == 1: data = data[:, np.newaxis] + elif data.ndim != 2: + raise ValueError(f"expected (n_samples, n_channels), got {data.shape}. Did you mean envelope=True?") - # Handle channel count mismatch + n_samples = data.shape[0] + n_ch = data.shape[1] + + # Handle channel count mismatch. Pad/trim the channel axis only; + # any trailing envelope axis rides along untouched. if n_ch < self.n_channels: - data = np.pad(data, ((0, 0), (0, self.n_channels - n_ch))) + pad = [(0, 0)] * data.ndim + pad[1] = (0, self.n_channels - n_ch) + data = np.pad(data, pad) elif n_ch > self.n_channels: data = data[:, : self.n_channels] @@ -118,7 +153,9 @@ def push_data(self, data: np.ndarray, timestamps=None) -> None: end_ch = min(self.channel_offset + self.n_visible, self.n_channels) vis_data = data[:, self.channel_offset : end_ch].astype(np.float32) if vis_data.shape[1] < self.n_visible: - vis_data = np.pad(vis_data, ((0, 0), (0, self.n_visible - vis_data.shape[1]))) + pad = [(0, 0)] * vis_data.ndim + pad[1] = (0, self.n_visible - vis_data.shape[1]) + vis_data = np.pad(vis_data, pad) # Truncate if more data than one full sweep if n_samples > self.total_raw_samples: @@ -204,6 +241,18 @@ def set_srate(self, srate: float) -> None: self.srate = srate self._allocate() + def set_envelope(self, envelope: bool) -> None: + """Switch between raw and pre-reduced (min, max) input. + + Reallocates, since the raw buffer changes rank -- so whatever is + currently displayed is discarded. Only worth calling when the source + itself changed shape. + """ + with self._lock: + if bool(envelope) != self.envelope: + self.envelope = bool(envelope) + self._allocate() + # ------------------------------------------------------------------ # Properties and MultiLine data (called from render/UI thread) # ------------------------------------------------------------------ @@ -380,8 +429,15 @@ def _recompute_columns(self, first_col: int, n_cols: int) -> None: chunk = self.raw_buffer[start:end] with warnings.catch_warnings(): warnings.simplefilter("ignore", RuntimeWarning) - mins = np.nanmin(chunk, axis=0) - maxs = np.nanmax(chunk, axis=0) + if self.envelope: + # Min over the lower bounds, max over the upper ones -- + # the same envelope this column would have had from raw + # samples, provided the upstream buckets are finer. + mins = np.nanmin(chunk[..., 0], axis=0) + maxs = np.nanmax(chunk[..., 1], axis=0) + else: + mins = np.nanmin(chunk, axis=0) + maxs = np.nanmax(chunk, axis=0) # Replace NaN results (all-NaN columns) with 0 self.display_mins[col] = np.nan_to_num(mins, nan=0.0) self.display_maxs[col] = np.nan_to_num(maxs, nan=0.0) diff --git a/src/phosphor/sweep_widget.py b/src/phosphor/sweep_widget.py index e79f593..1e3ba28 100644 --- a/src/phosphor/sweep_widget.py +++ b/src/phosphor/sweep_widget.py @@ -11,9 +11,11 @@ from .constants import ( CURSOR_COLOR, CURSOR_GAP_COLUMNS, + CURSOR_THICKNESS, DEFAULT_DISPLAY_DUR, DEFAULT_LINE_THICKNESS, DEFAULT_MAX_EVENTS, + DEFAULT_MAX_FPS, DEFAULT_N_COLUMNS, DEFAULT_N_VISIBLE, EVENT_POOL_SIZE, @@ -46,6 +48,21 @@ class SweepConfig: colors: list[tuple[float, float, float, float]] | None = None # Initial waveform-amplitude multiplier (1.0 = data autoscale only). amplitude_scale: float = 1.0 + # Pushed data is ``(n_samples, n_channels, 2)`` holding an already-reduced + # (min, max) pair per sample, rather than raw values. Lets a producer + # decimate near the source -- see ``SweepBuffer.push_data``. + envelope: bool = False + # Sweeping cursor appearance. phosphor sizes the cursor from the column + # gap, which is in time-world units and collapses to about a pixel; these + # let a caller ask for something more visible without reaching into the + # fastplotlib graphic after every rebuild. + cursor_thickness: float = CURSOR_THICKNESS + cursor_color: tuple[float, float, float, float] | str = CURSOR_COLOR + # Target canvas render rate. A scrolling trace reads smooth well below the + # display refresh, and halving the rate roughly halves render cost, so this + # is the cheapest knob there is. ``0`` uncaps; ``None`` leaves fastplotlib's + # own default alone. + max_fps: float | None = DEFAULT_MAX_FPS class SweepWidget(ChannelPlotWidget): @@ -81,6 +98,7 @@ def __init__(self, config: SweepConfig, parent: QWidget | None = None): max_events=config.max_events, channel_order=config.channel_order, amplitude_scale=config.amplitude_scale, + envelope=config.envelope, ) self._buffer = self.sweep_buffer self._palette = list(config.colors) if config.colors is not None else list(SOFT_CHANNEL_COLORS) @@ -97,6 +115,7 @@ def __init__(self, config: SweepConfig, parent: QWidget | None = None): # Start rendering self._init_rendering() + self.set_max_fps(config.max_fps) # ------------------------------------------------------------------ # Public API @@ -129,6 +148,9 @@ def update_config(self, config: SweepConfig) -> None: if config.display_dur != buf.display_dur: buf.set_display_dur(config.display_dur) self._time_axis.set_range(config.display_dur) + if config.envelope != buf.envelope: + buf.set_envelope(config.envelope) + self.set_max_fps(config.max_fps) self._update_range_label() # ------------------------------------------------------------------ @@ -168,7 +190,12 @@ def _setup_graphics(self) -> None: # Cursor: vertical line at the sweep position. # Span the MultiLine's y-extent with small margin. sweep_x = buf.sweep_col / max(buf.n_columns - 1, 1) * buf.display_dur - cursor_color = CURSOR_COLOR[:3] + cursor_color = self._config.cursor_color + if isinstance(cursor_color, (tuple, list)): + cursor_color = tuple(cursor_color)[:3] + # The geometry-derived width is in time-world units and collapses to + # about a pixel, so the configured thickness is the floor rather than + # the other way round. gap_w = CURSOR_GAP_COLUMNS / max(buf.n_columns - 1, 1) * buf.display_dur y_bottom = 0.0 y_top = (n_vis - 1) * self._z_offset_scale @@ -181,7 +208,7 @@ def _setup_graphics(self) -> None: dtype=np.float32, ), colors=cursor_color, - thickness=max(1.0, gap_w * 2), + thickness=max(self._config.cursor_thickness, gap_w * 2), ) self._setup_event_pool() diff --git a/src/phosphor/x_axis.py b/src/phosphor/x_axis.py index 9ae34e7..5217ef7 100644 --- a/src/phosphor/x_axis.py +++ b/src/phosphor/x_axis.py @@ -36,7 +36,7 @@ def __init__(self, range_max: float, unit: str = "s", parent: QWidget | None = N self._log = False self.setFixedHeight(24) bg = BG_COLOR - self.setStyleSheet(f"background-color: rgb({int(bg[0]*255)},{int(bg[1]*255)},{int(bg[2]*255)});") + self.setStyleSheet(f"background-color: rgb({int(bg[0] * 255)},{int(bg[1] * 255)},{int(bg[2] * 255)});") def set_range(self, range_max: float) -> None: self._range_max = range_max diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..b4cbbc3 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,25 @@ +"""Qt fixture for widget tests. + +The overlays are plain ``QWidget`` painters -- no GPU, no fastplotlib -- so they +can be exercised on the offscreen platform plugin. That keeps the geometry maths +(which is where the bugs live) under test on a headless runner, without needing +a rendering backend. +""" + +import os + +import pytest + + +@pytest.fixture(scope="session") +def qapp(): + """A QApplication on the offscreen platform, or skip.""" + os.environ.setdefault("QT_QPA_PLATFORM", "offscreen") + QtWidgets = pytest.importorskip("PySide6.QtWidgets", reason="PySide6 not installed") + app = QtWidgets.QApplication.instance() + if app is None: + try: + app = QtWidgets.QApplication([]) + except Exception as exc: # noqa: BLE001 - any platform failure means skip + pytest.skip(f"cannot start QApplication: {exc}") + return app diff --git a/tests/test_overlays.py b/tests/test_overlays.py new file mode 100644 index 0000000..c580e96 --- /dev/null +++ b/tests/test_overlays.py @@ -0,0 +1,222 @@ +"""Canvas overlay geometry. + +The overlays are simple to draw and easy to get subtly wrong: a label a few +pixels off its trace, or one that stops updating when the view scrolls, is the +kind of thing that survives a screenshot review. The maths is all in +``_row_geometry`` and the change detection, so that is what is pinned here. +""" + +import pytest + +from phosphor.overlays import ( + MAX_LABEL_FONT_PX, + MIN_LABEL_FONT_PX, + ChannelLabelOverlay, + ScaleBarOverlay, +) + + +@pytest.fixture() +def labels(qapp) -> ChannelLabelOverlay: + ov = ChannelLabelOverlay() + ov.resize(400, 300) + return ov + + +@pytest.fixture() +def bar(qapp) -> ScaleBarOverlay: + ov = ScaleBarOverlay() + ov.resize(400, 300) + return ov + + +# ---- world -> screen mapping ----------------------------------------------- + + +def test_measured_projection_is_used_verbatim(labels): + """When the plot supplies the camera's mapping, use it -- do not re-derive.""" + labels.set_view(0, 4, True, y0=100.0, slope=-25.0, z_scale=1.0) + assert labels._row_geometry(4, 300) == (100.0, -25.0) + + +def test_analytic_fallback_spans_the_canvas(labels): + """With no camera mapping, rows plus a 5% margin must fill the height.""" + labels.set_view(0, 5, True, y0=None, slope=None, z_scale=1.0) + y0, slope = labels._row_geometry(5, 300) + + # Rows sit at world-y 0..4 with margin max(4*0.05, 0.5) = 0.5 either side, + # so world -0.5 maps to the bottom (y=300) and 4.5 to the top (y=0). + assert y0 + slope * -0.5 == pytest.approx(300.0) + assert y0 + slope * 4.5 == pytest.approx(0.0) + assert slope < 0 # higher world-y is nearer the top + + +def test_degenerate_slope_falls_back(labels): + """A zero slope would collapse every label onto one line.""" + labels.set_view(0, 4, True, y0=10.0, slope=0.0, z_scale=1.0) + _, slope = labels._row_geometry(4, 300) + assert slope != 0.0 + + +def test_single_channel_fallback_does_not_divide_by_zero(labels): + labels.set_view(0, 1, True, z_scale=1.0) + y0, slope = labels._row_geometry(1, 300) + assert slope == pytest.approx(-300.0) # margin 0.5 either side of one row + assert y0 == pytest.approx(150.0) + + +# ---- repaint gating -------------------------------------------------------- + + +def test_set_view_repaints_only_on_change(labels, monkeypatch): + calls = [] + monkeypatch.setattr(labels, "update", lambda: calls.append(1)) + + labels.set_view(0, 4, True, 100.0, -25.0, 1.0) + assert len(calls) == 1 + labels.set_view(0, 4, True, 100.0, -25.0, 1.0) + assert len(calls) == 1, "identical view should not repaint" + labels.set_view(1, 4, True, 100.0, -25.0, 1.0) + assert len(calls) == 2, "scrolling by one channel must repaint" + + +def test_set_labels_repaints_only_on_change(labels, monkeypatch): + labels.set_labels(["a", "b"]) + calls = [] + monkeypatch.setattr(labels, "update", lambda: calls.append(1)) + + labels.set_labels(["a", "b"]) + assert calls == [] + labels.set_labels(["a", "c"]) + assert len(calls) == 1 + + +def test_scale_bar_repaints_only_on_change(bar, monkeypatch): + bar.set_bar(40.0, "100 uV") + calls = [] + monkeypatch.setattr(bar, "update", lambda: calls.append(1)) + + bar.set_bar(40.0, "100 uV") + assert calls == [] + bar.set_bar(41.0, "100 uV") + assert len(calls) == 1 + bar.set_bar(41.0, "200 uV") + assert len(calls) == 2 + + +# ---- painting is defensive ------------------------------------------------- + + +def test_painting_with_no_data_is_a_no_op(labels, bar): + """Rendering starts before the first view arrives; must not raise.""" + labels.render(labels.grab()) # no labels, no view + bar.render(bar.grab()) # no bar length or text + + +def test_label_paint_survives_extremes(labels): + """Very tall and very short rows both have to draw without raising.""" + labels.set_labels([f"ch{i}" for i in range(512)]) + + labels.set_view(0, 1, True, 150.0, -300.0, 1.0) # one enormous row + labels.grab() + + labels.set_view(0, 512, True, 300.0, -0.6, 1.0) # sub-pixel rows + labels.grab() + + +def test_font_bounds_are_sane(): + assert 0 < MIN_LABEL_FONT_PX < MAX_LABEL_FONT_PX + # Ordinary reading size. Bigger conveys nothing and the opaque chip behind + # each label would cover the signal it refers to. + assert MAX_LABEL_FONT_PX <= 16 + + +def test_font_is_capped_on_tall_rows(qapp, monkeypatch): + """Few visible channels means rows hundreds of pixels tall; the label must + not scale with them.""" + from PySide6 import QtGui + + sizes = [] + original = QtGui.QPainter.setFont + + def spy(self, font): + sizes.append(font.pixelSize()) + return original(self, font) + + monkeypatch.setattr(QtGui.QPainter, "setFont", spy) + + ov = ChannelLabelOverlay() + ov.resize(400, 1000) + ov.set_labels(["ch0", "ch1"]) + ov.set_view(0, 2, True, 1000.0, -500.0, 1.0) # 500 px rows + ov.grab() + + assert sizes and max(sizes) <= MAX_LABEL_FONT_PX + + +def test_font_bounds_are_overridable(qapp): + ov = ChannelLabelOverlay(min_font_px=6, max_font_px=9) + assert (ov._min_font_px, ov._max_font_px) == (6, 9) + + +def test_labels_are_indexed_by_absolute_channel(labels, monkeypatch): + """Scrolling changes which labels are drawn without the caller reslicing.""" + drawn: list[str] = [] + labels.set_labels([f"ch{i}" for i in range(64)]) + labels.set_view(0, 4, True, 300.0, -75.0, 1.0) + + # Intercept the text calls rather than reading pixels. + from PySide6 import QtGui + + original = QtGui.QPainter.drawText + + def spy(self, *args): + if args and isinstance(args[-1], str): + drawn.append(args[-1]) + return original(self, *args) + + monkeypatch.setattr(QtGui.QPainter, "drawText", spy) + + labels.grab() + assert drawn == ["ch0", "ch1", "ch2", "ch3"] + + drawn.clear() + labels.set_view(10, 4, True, 300.0, -75.0, 1.0) + labels.grab() + assert drawn == ["ch10", "ch11", "ch12", "ch13"] + + +def test_scale_bar_sits_in_the_lower_right(bar, monkeypatch): + """Out of the way of the traces, and anchored at its foot so the bottom + edge does not move when the amplitude scale changes.""" + from PySide6 import QtGui + + from phosphor.overlays import SCALE_BAR_BOTTOM_MARGIN, SCALE_BAR_RIGHT_MARGIN + + lines: list[tuple] = [] + original = QtGui.QPainter.drawLine + + def spy(self, *args): + lines.append(args) + return original(self, *args) + + monkeypatch.setattr(QtGui.QPainter, "drawLine", spy) + + w, h = bar.width(), bar.height() + bar.set_bar(60.0, "100 uV") + bar.grab() + + # First drawLine is the vertical bar: (x, y_top, x, y_bot). + x, y_top, _, y_bot = lines[0] + assert x == w - SCALE_BAR_RIGHT_MARGIN + assert y_bot == h - SCALE_BAR_BOTTOM_MARGIN + assert y_bot - y_top == pytest.approx(60.0, abs=1) + assert y_top > h / 2, "bar should sit below the vertical midpoint" + + # Growing the bar moves its top, not its foot. + lines.clear() + bar.set_bar(120.0, "200 uV") + bar.grab() + _, y_top2, _, y_bot2 = lines[0] + assert y_bot2 == y_bot + assert y_top2 < y_top diff --git a/tests/test_sweep_buffer.py b/tests/test_sweep_buffer.py new file mode 100644 index 0000000..dffaca2 --- /dev/null +++ b/tests/test_sweep_buffer.py @@ -0,0 +1,162 @@ +"""SweepBuffer's CPU-side reduction, including the envelope input mode. + +``SweepBuffer`` is pure numpy and threading -- no canvas, no GPU -- so it can be +exercised headlessly, which is where the reduction logic actually lives. + +The property that matters for envelope mode is *equivalence*: pushing a +pre-reduced (min, max) stream must draw what the raw signal would have drawn. +If that does not hold, moving decimation upstream changes what the user sees, +and the whole point was that it should not. +""" + +import numpy as np +import pytest + +from phosphor.sweep_buffer import SweepBuffer + + +def make_buffer(**kwargs) -> SweepBuffer: + defaults = dict(n_channels=4, srate=1000.0, display_dur=1.0, n_columns=10, n_visible=4) + defaults.update(kwargs) + return SweepBuffer(**defaults) + + +def minmax_decimate(raw: np.ndarray, factor: int) -> np.ndarray: + """(n, ch) -> (n // factor, ch, 2), the shape an upstream decimator emits.""" + n = raw.shape[0] // factor * factor + buckets = raw[:n].reshape(-1, factor, raw.shape[1]) + return np.stack([buckets.min(axis=1), buckets.max(axis=1)], axis=-1) + + +# ---- raw mode is unchanged -------------------------------------------------- + + +def test_raw_push_reduces_to_columns(): + buf = make_buffer() + raw = np.zeros((1000, 4), dtype=np.float32) + raw[15, 0] = 5.0 # column 0 spans samples 0..99 + raw[150, 1] = -3.0 # column 1 spans 100..199 + buf.push_data(raw) + + assert buf.display_maxs[0, 0] == pytest.approx(5.0) + assert buf.display_mins[1, 1] == pytest.approx(-3.0) + + +def test_raw_push_still_accepts_1d(): + buf = make_buffer(n_channels=1, n_visible=1) + buf.push_data(np.ones(1000, dtype=np.float32)) + assert buf.display_maxs.max() == pytest.approx(1.0) + + +# ---- envelope mode ---------------------------------------------------------- + + +def test_envelope_matches_raw_for_the_same_signal(): + """The equivalence that justifies decimating upstream at all. + + Note the envelope buffer is configured at the *envelope* rate, not the raw + one: ``srate`` describes the stream being pushed. Both buffers then span the + same wall-clock second over the same ten columns, so each column covers the + same 100 raw samples, and the two reductions must agree exactly. + """ + rng = np.random.default_rng(0) + raw = rng.standard_normal((1000, 4)).astype(np.float32) + factor = 10 + + from_raw = make_buffer(srate=1000.0) + from_raw.push_data(raw) + + from_env = make_buffer(srate=1000.0 / factor, envelope=True) + from_env.push_data(minmax_decimate(raw, factor)) + + np.testing.assert_allclose(from_env.display_mins, from_raw.display_mins) + np.testing.assert_allclose(from_env.display_maxs, from_raw.display_maxs) + + +def test_envelope_srate_is_the_bucket_rate(): + """Sizing from the pre-decimation rate is the easy mistake: the ring would + be `factor` times too long and the sweep would sit mostly empty.""" + buf = make_buffer(srate=100.0, display_dur=1.0, envelope=True) + assert buf.total_raw_samples == 100 + assert buf.raw_buffer.shape == (100, 4, 2) + + +def test_envelope_preserves_a_spike_stride_decimation_would_lose(): + """The motivating case, end to end through the buffer.""" + raw = np.zeros((1000, 1), dtype=np.float32) + raw[37, 0] = 100.0 # missed by raw[::10] + + buf = make_buffer(n_channels=1, n_visible=1, envelope=True) + buf.push_data(minmax_decimate(raw, 10)) + + assert buf.display_maxs.max() == pytest.approx(100.0) + + +def test_envelope_allocates_a_rank_3_raw_buffer(): + assert make_buffer(envelope=True).raw_buffer.shape == (1000, 4, 2) + assert make_buffer().raw_buffer.shape == (1000, 4) + + +def test_envelope_rejects_raw_shaped_data(): + """A silent misread here would plot half the channels at wrong values.""" + buf = make_buffer(envelope=True) + with pytest.raises(ValueError, match="expects .*n_channels, 2"): + buf.push_data(np.zeros((100, 4), dtype=np.float32)) + + +def test_raw_mode_rejects_envelope_shaped_data(): + buf = make_buffer() + with pytest.raises(ValueError, match="Did you mean envelope=True"): + buf.push_data(np.zeros((100, 4, 2), dtype=np.float32)) + + +def test_envelope_channel_padding_keeps_the_pair_axis(): + """Fewer channels than configured pads the channel axis only.""" + buf = make_buffer(n_channels=4, n_visible=4, envelope=True) + data = np.ones((100, 2, 2), dtype=np.float32) + buf.push_data(data) + # Channels 0-1 carry the pushed value; 2-3 were padded with zeros. + assert buf.display_maxs[0, 0] == pytest.approx(1.0) + assert buf.display_maxs[0, 3] == pytest.approx(0.0) + + +def test_envelope_channel_trimming(): + buf = make_buffer(n_channels=2, n_visible=2, envelope=True) + buf.push_data(np.ones((100, 5, 2), dtype=np.float32)) + assert buf.display_maxs.shape[1] == 2 + + +def test_envelope_wraps_the_ring_like_raw(): + """More samples than one sweep must wrap, not overflow.""" + buf = make_buffer(envelope=True) + env = np.zeros((1500, 4, 2), dtype=np.float32) + env[..., 1] = 2.0 + buf.push_data(env) # 1.5x the 1000-sample ring + assert buf.display_maxs.max() == pytest.approx(2.0) + + +def test_set_envelope_reallocates(): + buf = make_buffer() + assert buf.raw_buffer.ndim == 2 + buf.set_envelope(True) + assert buf.raw_buffer.ndim == 3 + assert buf.envelope + # Idempotent: no reallocation, and the version does not churn. + version = buf.version + buf.set_envelope(True) + assert buf.version == version + + +def test_envelope_scale_and_midpoint_use_both_bounds(): + """_compute_y_scale/_compute_ch_mid read display_mins/maxs, so an envelope + must fill both -- a bug filling only one would autoscale to half range.""" + buf = make_buffer(n_channels=1, n_visible=1, envelope=True) + env = np.zeros((100, 1, 2), dtype=np.float32) + env[..., 0] = -4.0 + env[..., 1] = 4.0 + buf.push_data(env) + + assert buf.display_mins.min() == pytest.approx(-4.0) + assert buf.display_maxs.max() == pytest.approx(4.0) + # ±0.5 normalization over a ±4 range. + assert buf._compute_y_scale() == pytest.approx(0.125)