diff --git a/src/phosphor/channel_plot.py b/src/phosphor/channel_plot.py index ac53708..43f198d 100644 --- a/src/phosphor/channel_plot.py +++ b/src/phosphor/channel_plot.py @@ -71,6 +71,11 @@ def __init__( # _buffer is set by the subclass before calling _init_rendering() self._buffer = None + # Trace colours, indexed by *absolute channel* so a channel keeps its + # colour as it scrolls -- see _channel_color. Subclasses override this + # with a configured palette before building graphics. + self._palette = list(CHANNEL_COLORS) + # 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 @@ -277,7 +282,7 @@ def _sync_overlays(self) -> None: self._label_overlay.set_view( getattr(buf, "channel_offset", 0), getattr(buf, "n_visible", 0), - getattr(buf, "channel_order", "top_down") == "top_down", + self._is_top_down(), y0, slope, z_scale, @@ -330,14 +335,89 @@ def _on_wheel_event(self, event) -> None: self._on_ctrl_scroll(delta) elif "Shift" in getattr(event, "modifiers", ()): # Shift+scroll → amplitude zoom - factor = 1.1 if delta > 0 else 0.9 - self._zoom_amplitude(factor) + self._zoom_amplitude(1.1 if self._shift_wheel_delta(event) > 0 else 0.9) else: # Unmodified scroll → channel scroll - buf = self._buffer - step = 1 if delta < 0 else -1 - buf.set_channel_offset(buf.channel_offset + step) - self._update_range_label() + step = self._channel_scroll_step(delta) + if step: + buf = self._buffer + buf.set_channel_offset(buf.channel_offset + step) + self._update_range_label() + + # ------------------------------------------------------------------ + # Row <-> channel mapping + # ------------------------------------------------------------------ + # + # A buffer lays its visible rows out along world-y, and "row 0" is the + # bottom of the canvas because that is what a plain arange of z-offsets + # produces. ``channel_order="top_down"`` reverses which channel lands on + # which row, so anything converting between a screen position and a channel + # has to go through here -- reading it off as ``channel_offset + row`` + # silently inverts the answer whenever top_down is in force, which is the + # sweep's default. + + def _is_top_down(self) -> bool: + """Whether the first visible channel is drawn at the top of the canvas. + + A buffer that does not declare an order (the spectrum) offsets its rows + with a plain arange, so *absent* means bottom-up rather than the sweep's + default. + """ + return getattr(self._buffer, "channel_order", "bottom_up") == "top_down" + + def _channel_at_row(self, row: int) -> int: + """Absolute channel drawn at *row*, counting up from the canvas bottom.""" + if self._is_top_down(): + row = self._buffer.n_visible - 1 - row + return self._buffer.channel_offset + row + + def _row_of_channel(self, channel: int) -> int: + """Row a channel is drawn at, counting up from the canvas bottom.""" + row = channel - self._buffer.channel_offset + if self._is_top_down(): + row = self._buffer.n_visible - 1 - row + return row + + def _channel_color(self, channel: int) -> tuple[float, float, float]: + """Palette colour for an absolute channel index, as RGB. + + Keyed to the channel rather than the on-screen row so a trace keeps its + colour while the window scrolls past it -- the point of a colour here + is to let the eye follow one channel, which a colour that belongs to + the row actively defeats. + """ + return tuple(self._palette[channel % len(self._palette)][:3]) + + @staticmethod + def _shift_wheel_delta(event) -> float: + """Wheel motion for Shift+scroll, from whichever axis it arrived on. + + Holding Shift makes the OS report a mouse wheel as *horizontal* + scrolling -- the convention that scrolls a document sideways -- so the + motion arrives in dx with dy pinned at 0. Reading dy alone then makes + every notch look negative and amplitude only ever zooms out. A trackpad + reports both axes natively and is unaffected, which is why this shows + up on a mouse only. + """ + dy = getattr(event, "dy", 0.0) or 0.0 + return dy if dy else (getattr(event, "dx", 0.0) or 0.0) + + @staticmethod + def _channel_scroll_step(delta: float) -> int: + """Channel-offset change for one wheel notch, 0 for no vertical motion. + + Scrolling down moves the window down the channel list, so the traces + travel with the fingers the way a document does. Split out from the + handler so the direction can be tested without a canvas, since it is + the kind of thing that is obvious in use and invisible in review. + + A horizontal trackpad swipe arrives as a wheel event carrying dx with + dy at 0, which is why 0 has to mean *stay*: taking it as a direction + makes sideways scrolling walk the channel window. + """ + if delta == 0: + return 0 + return 1 if delta > 0 else -1 def _on_pointer_move_event(self, event) -> None: self._handle_mouse_move(event) @@ -431,14 +511,13 @@ def _handle_mouse_move(self, event) -> None: best_dist = dist best_idx = i - ch_index = best_idx - abs_ch = buf.channel_offset + ch_index + abs_ch = self._channel_at_row(best_idx) labels = self._channel_labels 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}" + rgb = self._channel_color(abs_ch) + hex_color = f"#{int(rgb[0] * 255):02x}{int(rgb[1] * 255):02x}{int(rgb[2] * 255):02x}" html = f'\u25a0 {label}' from PySide6.QtCore import QPoint diff --git a/src/phosphor/spectrum_widget.py b/src/phosphor/spectrum_widget.py index 23939ef..202e0ca 100644 --- a/src/phosphor/spectrum_widget.py +++ b/src/phosphor/spectrum_widget.py @@ -8,7 +8,7 @@ from PySide6.QtWidgets import QWidget from .channel_plot import ChannelPlotWidget -from .constants import CHANNEL_COLORS, DEFAULT_N_VISIBLE +from .constants import DEFAULT_N_VISIBLE from .spectrum_buffer import SpectrumBuffer from .x_axis import XAxisWidget @@ -118,8 +118,8 @@ def _setup_graphics(self) -> None: buf = self.spectrum_buffer data = buf.get_multiline_data(self._display_freq_max) - n_vis = buf.n_visible - colors = [CHANNEL_COLORS[i % len(CHANNEL_COLORS)][:3] for i in range(n_vis)] + offset = buf.channel_offset + colors = [self._channel_color(offset + i) for i in range(buf.n_visible)] self._multi_line = subplot.add_multi_line( data, diff --git a/src/phosphor/sweep_buffer.py b/src/phosphor/sweep_buffer.py index 8e549f8..feb1ac8 100644 --- a/src/phosphor/sweep_buffer.py +++ b/src/phosphor/sweep_buffer.py @@ -74,12 +74,16 @@ 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 - raw_shape = (self.total_raw_samples, self.n_visible) + # Every channel is buffered, not just the visible window. Scrolling is + # then a change of which slice is drawn, with no reallocation and + # nothing lost -- the alternative is that a channel scrolled into view + # has no history, because it was discarded on arrival. + raw_shape = (self.total_raw_samples, self.n_channels) 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) + self.display_mins = np.zeros((self.n_columns, self.n_channels), dtype=np.float32) + self.display_maxs = np.zeros((self.n_columns, self.n_channels), dtype=np.float32) self.write_pos = 0 self.sweep_col = 0 @@ -149,13 +153,7 @@ def push_data(self, data: np.ndarray, timestamps=None) -> None: elif n_ch > self.n_channels: data = data[:, : self.n_channels] - # Select visible channels - 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: - pad = [(0, 0)] * vis_data.ndim - pad[1] = (0, self.n_visible - vis_data.shape[1]) - vis_data = np.pad(vis_data, pad) + vis_data = data.astype(np.float32, copy=False) # Truncate if more data than one full sweep if n_samples > self.total_raw_samples: @@ -215,19 +213,35 @@ def set_n_channels(self, n: int) -> None: self._allocate() def set_channel_offset(self, offset: int) -> None: + """Scroll the visible window. Cheap: no reallocation, nothing lost. + + Every channel is already buffered, so this only changes which slice is + drawn. The graphic keeps its shape -- ``n_visible`` rows either way -- + so there is no version bump and no rebuild; the next frame simply + redraws every column from the new window. + """ with self._lock: offset = max(0, min(offset, self.n_channels - self.n_visible)) if offset != self.channel_offset: self.channel_offset = offset - self._allocate() + self._mark_all_dirty() def set_n_visible(self, n: int) -> None: + """Change how many channels are drawn, keeping their history. + + The stored data is untouched -- only the height of the window over it + changes. The multiline graphic does change shape, so the version is + bumped to have it rebuilt, but that is a GPU-side rebuild rather than a + data reset: the traces reappear already populated. + """ with self._lock: n = max(1, min(n, self.n_channels)) if n != self.n_visible: self.n_visible = n self.channel_offset = min(self.channel_offset, self.n_channels - self.n_visible) - self._allocate() + self._ch_mid = np.zeros((self.n_visible, 1), dtype=np.float32) + self._mark_all_dirty() + self._version += 1 def set_display_dur(self, dur: float) -> None: with self._lock: @@ -261,6 +275,20 @@ def set_envelope(self, envelope: bool) -> None: def version(self) -> int: return self._version + def _mark_all_dirty(self) -> None: + """Force a full redraw on the next frame. Call while holding ``_lock``.""" + self._dirty_start = 0 + self._dirty_end = self.n_columns - 1 + + def _visible(self) -> slice: + """The channel slice currently on screen. + + Storage spans every channel; this is the window drawn from it. + Normalization and midpoints use it too, so the amplitude scale follows + what the user can see rather than channels off-screen. + """ + return slice(self.channel_offset, self.channel_offset + self.n_visible) + def _compute_y_scale(self) -> float: """Compute normalization scale from current buffer data. @@ -268,7 +296,11 @@ def _compute_y_scale(self) -> float: that normalized data fits within ±0.5, matching MultiLine z_offset_scale separation. Must be called while holding ``_lock``. """ - max_abs = max(float(np.abs(self.display_mins).max()), float(np.abs(self.display_maxs).max())) + vis = self._visible() + max_abs = max( + float(np.abs(self.display_mins[:, vis]).max()), + float(np.abs(self.display_maxs[:, vis]).max()), + ) return 0.5 / max(max_abs, 1e-12) def _compute_ch_mid(self, scale: float) -> np.ndarray: @@ -279,8 +311,9 @@ def _compute_ch_mid(self, scale: float) -> np.ndarray: translating channels as the user changes ``amplitude_scale``. Must be called while holding ``_lock``. """ - ch_min = self.display_mins.min(axis=0) - ch_max = self.display_maxs.max(axis=0) + vis = self._visible() + ch_min = self.display_mins[:, vis].min(axis=0) + ch_max = self.display_maxs[:, vis].max(axis=0) return (((ch_min + ch_max) / 2) * scale).astype(np.float32).reshape(-1, 1) def _build_multiline_array(self, mins, maxs, col_indices, scale) -> np.ndarray: @@ -288,6 +321,9 @@ def _build_multiline_array(self, mins, maxs, col_indices, scale) -> np.ndarray: Must be called while holding ``_lock``. """ + vis = self._visible() + mins = mins[:, vis] + maxs = maxs[:, vis] n_cols = mins.shape[0] out = np.zeros((self.n_visible, 2 * n_cols, 3), dtype=np.float32) col_x = col_indices.astype(np.float32) / max(self.n_columns - 1, 1) * self.display_dur @@ -325,9 +361,7 @@ def set_amplitude_scale(self, scale: float) -> None: if scale == self._amplitude_scale: return self._amplitude_scale = scale - # Force a full rebuild on the next animation frame. - self._dirty_start = 0 - self._dirty_end = self.n_columns - 1 + self._mark_all_dirty() def set_channel_order(self, order: str) -> None: if order not in ("top_down", "bottom_up"): @@ -336,8 +370,7 @@ def set_channel_order(self, order: str) -> None: if order == self.channel_order: return self.channel_order = order - self._dirty_start = 0 - self._dirty_end = self.n_columns - 1 + self._mark_all_dirty() def get_multiline_data(self) -> np.ndarray: """Full data shaped ``[n_visible, 2*n_columns, 3]`` for fastplotlib MultiLineGraphic. @@ -489,8 +522,10 @@ def _resize_display_dur(self, new_dur: float) -> None: # New write position: same total-sample count, different modulus new_write_pos = self._samples_since_alloc % new_total - # Create new (zeroed) buffer - new_raw = np.zeros((new_total, self.n_visible), dtype=np.float32) + # Create new (zeroed) buffer. Shape derived from the existing one so it + # keeps both the full channel width and, in envelope mode, the trailing + # (min, max) pair -- writing n_visible here silently mis-sized it. + new_raw = np.zeros((new_total,) + self.raw_buffer.shape[1:], dtype=np.float32) # Copy data preserving sample ages available = min(self._samples_since_alloc, old_total) @@ -523,8 +558,8 @@ def _resize_display_dur(self, new_dur: float) -> None: self.sweep_col = self._col_for_pos(new_write_pos) # Recompute display columns from new raw data - self.display_mins = np.zeros((new_n_columns, self.n_visible), dtype=np.float32) - self.display_maxs = np.zeros((new_n_columns, self.n_visible), dtype=np.float32) + self.display_mins = np.zeros((new_n_columns, self.n_channels), dtype=np.float32) + self.display_maxs = np.zeros((new_n_columns, self.n_channels), dtype=np.float32) self._recompute_columns(0, new_n_columns) self._dirty_start = None diff --git a/src/phosphor/sweep_widget.py b/src/phosphor/sweep_widget.py index 1e3ba28..4702a8f 100644 --- a/src/phosphor/sweep_widget.py +++ b/src/phosphor/sweep_widget.py @@ -106,6 +106,7 @@ def __init__(self, config: SweepConfig, parent: QWidget | None = None): # Create initial graphics self._cached_version = -1 + self._cached_offset = -1 self._multi_line = None self._z_offset_scale = 1.0 self._cursor_line = None @@ -157,6 +158,28 @@ def update_config(self, config: SweepConfig) -> None: # Graphics setup # ------------------------------------------------------------------ + def _visible_colors(self) -> list[tuple[float, float, float]]: + """RGB for each on-screen row, taken from its absolute channel.""" + buf = self.sweep_buffer + offset = buf.channel_offset + return [self._channel_color(offset + i) for i in range(buf.n_visible)] + + def _apply_channel_colors(self) -> None: + """Rewrite the MultiLine's colours after the window has scrolled. + + fastplotlib stores every line's vertices in one flat colour buffer, so + a row's colours are a contiguous run of ``stride`` entries -- vertices + plus the NaN separator that ends the line. + """ + ml = self._multi_line + buf = self.sweep_buffer + total = ml.colors.value.shape[0] + stride = total // max(buf.n_visible, 1) + rgba = np.ones((buf.n_visible, 4), dtype=np.float32) + rgba[:, :3] = np.asarray(self._visible_colors(), dtype=np.float32) + ml.colors[:] = np.repeat(rgba, stride, axis=0)[:total] + self._cached_offset = buf.channel_offset + def _setup_graphics(self) -> None: """Create or recreate MultiLineGraphic and cursor line.""" subplot = self._subplot @@ -176,9 +199,11 @@ def _setup_graphics(self) -> None: buf = self.sweep_buffer data = buf.get_multiline_data() - # Build per-channel colors (cycling through configured palette) + # Colours follow the absolute channel, so a trace keeps its colour + # while the window scrolls past it. n_vis = buf.n_visible - colors = [self._palette[i % len(self._palette)][:3] for i in range(n_vis)] + colors = self._visible_colors() + self._cached_offset = buf.channel_offset self._multi_line = subplot.add_multi_line( data, @@ -229,8 +254,18 @@ def _setup_event_pool(self) -> None: def _update_graphics(self) -> None: buf = self.sweep_buffer + if buf.channel_offset != self._cached_offset and buf.version == self._cached_version: + # Scrolling keeps the graphic, so its colours have to be rewritten + # in place to stay with their channels. Only on an actual scroll: + # this reuploads the whole colour buffer. + self._apply_channel_colors() + if buf.version != self._cached_version: - # Version changed (scroll, resize, display_dur change) → full rebuild + # The graphic's own shape changed -- channel count, visible count, + # sample rate, display duration -- so the MultiLine has to be + # rebuilt. Scrolling deliberately does not land here: it keeps + # n_visible rows and only marks the columns dirty, so it flows + # through the incremental path below and the plot never blanks. self._setup_graphics() return @@ -274,8 +309,7 @@ def _update_event_graphics(self) -> None: vis_end = buf.channel_offset + buf.n_visible if ev.channel < vis_start or ev.channel >= vis_end: continue - vis_idx = ev.channel - buf.channel_offset - y_center = vis_idx * self._z_offset_scale + y_center = self._row_of_channel(ev.channel) * self._z_offset_scale y_min = y_center - 0.45 * self._z_offset_scale y_max = y_center + 0.45 * self._z_offset_scale diff --git a/tests/test_interaction.py b/tests/test_interaction.py new file mode 100644 index 0000000..956d776 --- /dev/null +++ b/tests/test_interaction.py @@ -0,0 +1,189 @@ +"""Mouse interaction and trace colouring. + +Both are things that are obvious the moment you use the plot and invisible in +a diff: a scroll wheel that runs backwards, a modifier whose motion arrives on +the axis you did not read, a colour that belongs to the row instead of the +channel. The decisions are pulled out into small pure helpers precisely so +they can be pinned here without a canvas. +""" + +import numpy as np +import pytest + +from phosphor.channel_plot import ChannelPlotWidget +from phosphor.sweep_widget import SweepWidget + + +class Wheel: + """Stand-in for a rendercanvas wheel event.""" + + def __init__(self, dx=0.0, dy=0.0): + self.dx, self.dy = dx, dy + + +# ---- wheel direction ------------------------------------------------------- + + +def test_scroll_wheel_moves_the_channel_window_down_on_a_down_notch(qapp): + """Direction is easy to get backwards and only obvious in use.""" + step = ChannelPlotWidget._channel_scroll_step + assert step(1.0) == 1, "a positive wheel delta should advance the window" + assert step(-1.0) == -1 + assert step(0.0) == 0, "a horizontal swipe carries dy=0 and must not scroll" + + +def test_shift_scroll_reads_the_axis_the_os_actually_used(qapp): + """Holding Shift makes a mouse wheel arrive as horizontal scrolling. + + dy is then pinned at 0, every notch reads as negative, and amplitude only + ever zooms out -- on a mouse. A trackpad sends both axes and looks fine, + which is what makes this easy to miss. + """ + read = ChannelPlotWidget._shift_wheel_delta + + assert read(Wheel(dy=3.0)) == 3.0, "trackpad: vertical axis is used as-is" + assert read(Wheel(dy=-3.0)) == -3.0 + assert read(Wheel(dx=3.0)) == 3.0, "mouse+Shift: motion arrives on dx" + assert read(Wheel(dx=-3.0)) == -3.0, "and must still be able to be negative" + # dy wins when both are present, so a diagonal trackpad swipe is vertical. + assert read(Wheel(dx=-9.0, dy=1.0)) == 1.0 + assert read(Wheel()) == 0.0 + + +def test_trace_colour_belongs_to_the_channel_not_the_row(qapp): + """A colour that follows the row defeats the point of having one: the eye + cannot follow a channel across a scroll if its colour changes under it.""" + plot = ChannelPlotWidget.__new__(ChannelPlotWidget) # no canvas needed + plot._palette = [(1.0, 0.0, 0.0, 1.0), (0.0, 1.0, 0.0, 1.0), (0.0, 0.0, 1.0, 1.0)] + + assert plot._channel_color(0) == (1.0, 0.0, 0.0) + assert plot._channel_color(3) == (1.0, 0.0, 0.0), "palette cycles" + # Channel 4 is the same colour whether it is drawn in row 0 or row 3. + assert plot._channel_color(4) == (0.0, 1.0, 0.0) + + +# ---- colours stay with their channels -------------------------------------- + + +class FakeColors: + """fastplotlib keeps every line's vertices in one flat RGBA buffer.""" + + def __init__(self, n_rows: int, stride: int): + self.value = np.zeros((n_rows * stride, 4), dtype=np.float32) + + def __setitem__(self, key, value): + self.value[key] = value + + +class FakeBuffer: + def __init__(self, n_visible: int, channel_offset: int): + self.n_visible, self.channel_offset = n_visible, channel_offset + + +def make_widget(n_visible: int, offset: int, stride: int) -> SweepWidget: + """A SweepWidget with only the fields _apply_channel_colors touches. + + Built without __init__ because everything else it would construct needs a + GPU canvas, and the arithmetic under test is pure numpy. + """ + w = SweepWidget.__new__(SweepWidget) + w._palette = [(1.0, 0.0, 0.0, 1.0), (0.0, 1.0, 0.0, 1.0), (0.0, 0.0, 1.0, 1.0)] + w.sweep_buffer = FakeBuffer(n_visible, offset) + w._multi_line = type("ML", (), {})() + w._multi_line.colors = FakeColors(n_visible, stride) + return w + + +def test_visible_colours_are_taken_from_absolute_channels(): + w = make_widget(n_visible=3, offset=4, stride=2) + assert w._visible_colors() == [(0.0, 1.0, 0.0), (0.0, 0.0, 1.0), (1.0, 0.0, 0.0)] + + +def test_scrolling_rewrites_the_colour_buffer_row_by_row(): + """Each row owns a contiguous run of `stride` entries -- its vertices plus + the NaN separator that terminates the line.""" + stride = 4 + w = make_widget(n_visible=3, offset=1, stride=stride) + w._apply_channel_colors() + + written = w._multi_line.colors.value + assert written.shape == (3 * stride, 4) + for row, expected in enumerate([(0.0, 1.0, 0.0), (0.0, 0.0, 1.0), (1.0, 0.0, 0.0)]): + run = written[row * stride : (row + 1) * stride] + assert np.all(run[:, :3] == expected), f"row {row} is not one solid colour" + assert np.all(run[:, 3] == 1.0), "alpha must stay opaque" + + assert w._cached_offset == 1, "the applied offset is recorded, so it applies once" + + +def test_a_channel_keeps_its_colour_across_a_scroll(): + """The whole point: follow one trace as the window moves past it.""" + stride = 4 + before = make_widget(n_visible=3, offset=0, stride=stride) + before._apply_channel_colors() + # Channel 2 is drawn in row 2 here, and in row 0 after scrolling by two. + colour_of_channel_2 = before._multi_line.colors.value[2 * stride, :3].copy() + + after = make_widget(n_visible=3, offset=2, stride=stride) + after._apply_channel_colors() + + np.testing.assert_array_equal(after._multi_line.colors.value[0, :3], colour_of_channel_2) + + +def test_colour_rewrite_survives_a_single_visible_channel(): + """max(n_visible, 1) guards a division that would otherwise be by zero.""" + w = make_widget(n_visible=1, offset=7, stride=5) + w._apply_channel_colors() + assert np.all(w._multi_line.colors.value[:, :3] == (0.0, 1.0, 0.0)) + + +# ---- rows map back to the right channel ------------------------------------ + + +class OrderedBuffer: + def __init__(self, channel_offset, n_visible, channel_order=None): + self.channel_offset, self.n_visible = channel_offset, n_visible + if channel_order is not None: + self.channel_order = channel_order + + +def make_plot(buffer) -> ChannelPlotWidget: + plot = ChannelPlotWidget.__new__(ChannelPlotWidget) + plot._buffer = buffer + return plot + + +def test_top_down_puts_the_first_visible_channel_at_the_top(): + """The sweep's default. Row 0 is the bottom of the canvas, so the first + channel is the *last* row -- reading it off as offset + row inverts the + hover tooltip and drops event ticks on the wrong trace.""" + plot = make_plot(OrderedBuffer(channel_offset=10, n_visible=4, channel_order="top_down")) + + assert plot._channel_at_row(0) == 13, "bottom row holds the last channel" + assert plot._channel_at_row(3) == 10, "top row holds the first" + assert plot._row_of_channel(10) == 3 + assert plot._row_of_channel(13) == 0 + + +def test_bottom_up_counts_rows_with_the_channels(): + plot = make_plot(OrderedBuffer(channel_offset=10, n_visible=4, channel_order="bottom_up")) + + assert plot._channel_at_row(0) == 10 + assert plot._channel_at_row(3) == 13 + assert plot._row_of_channel(10) == 0 + + +def test_a_buffer_with_no_declared_order_is_bottom_up(): + """The spectrum offsets its rows with a plain arange and never sets + channel_order, so absent must not be read as the sweep's default.""" + plot = make_plot(OrderedBuffer(channel_offset=10, n_visible=4)) + + assert not plot._is_top_down() + assert plot._channel_at_row(0) == 10 + + +@pytest.mark.parametrize("order", ["top_down", "bottom_up"]) +def test_row_and_channel_mappings_are_inverses(order): + plot = make_plot(OrderedBuffer(channel_offset=7, n_visible=5, channel_order=order)) + for row in range(5): + assert plot._row_of_channel(plot._channel_at_row(row)) == row diff --git a/tests/test_sweep_buffer.py b/tests/test_sweep_buffer.py index dffaca2..e043836 100644 --- a/tests/test_sweep_buffer.py +++ b/tests/test_sweep_buffer.py @@ -160,3 +160,115 @@ def test_envelope_scale_and_midpoint_use_both_bounds(): 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) + + +# ---- scrolling keeps what it already drew ---------------------------------- +# +# Scrolling by one channel used to reallocate every buffer, so 31 of 32 rows +# were thrown away and redrawn from nothing even though their data had not +# changed. The fix is that storage spans every channel and the visible window +# is a view over it. + + +def channel_ramp(n_samples: int, n_channels: int) -> np.ndarray: + """Each channel holds its own index, so a row's identity is its value.""" + return np.tile(np.arange(n_channels, dtype=np.float32), (n_samples, 1)) + + +def test_all_channels_are_buffered_not_just_the_visible_ones(): + buf = make_buffer(n_channels=64, n_visible=8) + assert buf.raw_buffer.shape[1] == 64 + assert buf.display_mins.shape[1] == 64 + + +def test_scrolling_does_not_reallocate_or_rebuild(): + """A rebuild is what blanks the plot; the graphic keeps its shape here.""" + buf = make_buffer(n_channels=64, n_visible=8) + buf.push_data(channel_ramp(1000, 64)) + + version, raw = buf.version, buf.raw_buffer + buf.set_channel_offset(1) + + assert buf.version == version, "scrolling must not force a graphic rebuild" + assert buf.raw_buffer is raw, "scrolling must not reallocate the ring" + + +def test_scrolled_rows_keep_their_data(): + """The 31-of-32 case: everything that was on screen stays on screen. + + Asserted on the reduced columns rather than the multiline's y values, + because those are normalized to the visible window -- scrolling onto a + larger channel legitimately rescales every row, which would swamp the + thing being checked. + """ + buf = make_buffer(n_channels=64, n_visible=8) + buf.push_data(channel_ramp(1000, 64)) + + before = buf.display_maxs.copy() + buf.set_channel_offset(1) + + # Storage is untouched by scrolling; the window moved over it. + np.testing.assert_array_equal(buf.display_maxs, before) + # The seven rows that were already on screen are the same seven channels. + np.testing.assert_allclose(buf.display_maxs[:, 1:8].max(axis=0), np.arange(1, 8)) + # And the newly exposed row carries real data, not zeros. + assert buf.display_maxs[:, 8].max() == pytest.approx(8.0) + assert buf.get_multiline_data().shape[0] == 8 + + +def test_a_channel_scrolled_into_view_has_history_immediately(): + """The case no amount of copy-on-scroll could fix: the data has to have + been kept while the channel was off screen.""" + buf = make_buffer(n_channels=64, n_visible=4) + buf.push_data(channel_ramp(1000, 64)) + + buf.set_channel_offset(60) # jump well past anything ever displayed + data = buf.get_multiline_data() + + assert data.shape[0] == 4 + # Channels 60..63 were never visible, yet their columns are populated. + np.testing.assert_allclose(buf.display_maxs[:, 60:64].max(axis=0), [60.0, 61.0, 62.0, 63.0]) + + +def test_changing_visible_count_keeps_history_too(): + """Paging with /2 and x2 has the same problem and the same fix.""" + buf = make_buffer(n_channels=64, n_visible=8) + buf.push_data(channel_ramp(1000, 64)) + raw = buf.raw_buffer + + buf.set_n_visible(16) + + assert buf.raw_buffer is raw, "resizing the window must not reallocate" + assert buf.get_multiline_data().shape[0] == 16 + np.testing.assert_allclose(buf.display_maxs[:, :16].max(axis=0), np.arange(16)) + + +def test_normalization_follows_the_visible_window(): + """Amplitude scale must track what is on screen -- an off-screen channel + ten times larger should not flatten everything the user is looking at.""" + buf = make_buffer(n_channels=8, n_visible=2) + data = np.zeros((1000, 8), dtype=np.float32) + data[:, 0:2] = 1.0 + data[:, 7] = 100.0 # far off screen + buf.push_data(data) + + buf.get_multiline_data() + assert buf._compute_y_scale() == pytest.approx(0.5) # keyed to the visible 1.0 + + buf.set_channel_offset(6) # now channel 7 is visible + assert buf._compute_y_scale() == pytest.approx(0.005) + + +def test_time_zoom_preserves_the_envelope_pair_axis(): + """_resize_display_dur rebuilt the ring at the wrong rank in envelope mode, + so zooming time with the envelope on mis-sized the buffer.""" + buf = make_buffer(n_channels=4, n_visible=4, envelope=True) + env = np.zeros((1000, 4, 2), dtype=np.float32) + env[..., 0], env[..., 1] = -3.0, 3.0 + buf.push_data(env) + + buf.set_display_dur(0.5) + + assert buf.raw_buffer.ndim == 3 + assert buf.raw_buffer.shape[1:] == (4, 2) + assert buf.display_maxs.max() == pytest.approx(3.0)