From 0df682fa5a09d13cf27a79c8ba9775f8e72702b4 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 21:53:25 -0400 Subject: [PATCH 1/4] Scroll without resetting the plot Scrolling one channel threw away every buffer and started over, even though 31 of 32 rows held exactly the data they had a moment earlier. The cause was upstream of the reallocation: push_data sliced arriving data to the visible window, so channels off screen were discarded on arrival and a channel scrolled into view had no history to show. No amount of copy-on-scroll fixes that -- the data was never kept. So storage now spans every channel and the visible window is a view over it. set_channel_offset stops reallocating: it moves the window, marks the columns dirty, and leaves the version alone so the MultiLine is not rebuilt. A channel scrolled into view is already populated, including one that has never been displayed. set_n_visible gets the same benefit. It still bumps the version, because the graphic genuinely changes shape, but that is now a GPU-side rebuild rather than a data reset -- /2 and x2 come back with their traces already drawn. Normalization follows the visible window rather than the whole buffer. Without that, an off-screen channel ten times larger than the rest would flatten everything the user is actually looking at. Cost, measured at 256 channels over a 5 s window with 32 visible: buffered push/10ms scroll+draw full rate 30 kHz 158 MB 0.10 ms 0.36 ms envelope 2 ms bins 9 MB 0.06 ms 0.33 ms The reduction now runs over all 256 channels rather than the visible 32, which sounded expensive and is not: 0.10 ms per 10 ms of data is about 1% of a core. Memory is the real price, and it is why the envelope and this go together -- 158 MB is defensible, 9 MB is nothing. Also fixes a bug this work surfaced: _resize_display_dur rebuilt the ring as (new_total, n_visible), which was both the wrong width and, since the envelope landed, the wrong rank. Zooming time with the envelope on mis-sized the buffer. The shape is now derived from the buffer being replaced. Nine tests. Four fail if scrolling is put back to reallocating. --- src/phosphor/sweep_buffer.py | 83 ++++++++++++++++++-------- src/phosphor/sweep_widget.py | 6 +- tests/test_sweep_buffer.py | 112 +++++++++++++++++++++++++++++++++++ 3 files changed, 176 insertions(+), 25 deletions(-) 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..6f187c9 100644 --- a/src/phosphor/sweep_widget.py +++ b/src/phosphor/sweep_widget.py @@ -230,7 +230,11 @@ def _update_graphics(self) -> None: buf = self.sweep_buffer 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 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) From 06bc44d8687dce9a842a91ab116958683cb94958 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 22:03:39 -0400 Subject: [PATCH 2/4] Flip the scroll wheel direction, and ignore horizontal swipes Scrolling down now moves the channel window down the list, so the traces travel with the fingers the way a document does. A horizontal trackpad swipe arrives as a wheel event carrying dx with dy at 0, which the old branch read as a direction -- so a sideways swipe walked the channel window. Zero now means stay. Direction is split into a static helper so it can be tested without a canvas: it is obvious in use and invisible in review. --- src/phosphor/channel_plot.py | 26 ++++++++++++++++++++++---- tests/test_overlays.py | 13 +++++++++++++ 2 files changed, 35 insertions(+), 4 deletions(-) diff --git a/src/phosphor/channel_plot.py b/src/phosphor/channel_plot.py index ac53708..caae939 100644 --- a/src/phosphor/channel_plot.py +++ b/src/phosphor/channel_plot.py @@ -334,10 +334,28 @@ def _on_wheel_event(self, event) -> None: self._zoom_amplitude(factor) 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() + + @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) diff --git a/tests/test_overlays.py b/tests/test_overlays.py index c580e96..6e9c75c 100644 --- a/tests/test_overlays.py +++ b/tests/test_overlays.py @@ -220,3 +220,16 @@ def spy(self, *args): _, y_top2, _, y_bot2 = lines[0] assert y_bot2 == y_bot assert y_top2 < y_top + + +# ---- 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.""" + from phosphor.channel_plot import ChannelPlotWidget + + 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" From 7213f696471e8d34892421064c593be3339bc802 Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 22:26:41 -0400 Subject: [PATCH 3/4] Keep a trace's colour with its channel, and fix Shift+wheel on a mouse Two things that only show up when you actually drive the plot. Shift+scroll zoomed out whichever way the wheel turned, but only on a mouse. Holding Shift makes the OS report a 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 made every notch negative. A trackpad sends both axes, which is why it looked fine there. Trace colour was keyed to the on-screen row, so scrolling repainted every trace in its neighbour's colour -- which defeats the point of a colour now that scrolling no longer resets the plot: the eye cannot follow a channel whose colour changes under it. Colour is now keyed to the absolute channel, via a _channel_color on the base widget shared by the sweep, the spectrum, and the hover tooltip. The tooltip swatch had its own palette and matched neither, so it now agrees with the trace it names. Scrolling rewrites the MultiLine's flat colour buffer in place rather than rebuilding the graphic, so the plot still does not blank. --- src/phosphor/channel_plot.py | 36 ++++++++- src/phosphor/spectrum_widget.py | 6 +- src/phosphor/sweep_widget.py | 35 +++++++- tests/test_interaction.py | 136 ++++++++++++++++++++++++++++++++ tests/test_overlays.py | 13 --- 5 files changed, 204 insertions(+), 22 deletions(-) create mode 100644 tests/test_interaction.py diff --git a/src/phosphor/channel_plot.py b/src/phosphor/channel_plot.py index caae939..20b83cf 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 @@ -330,8 +335,7 @@ 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 step = self._channel_scroll_step(delta) @@ -340,6 +344,30 @@ def _on_wheel_event(self, event) -> None: buf.set_channel_offset(buf.channel_offset + step) self._update_range_label() + 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. @@ -455,8 +483,8 @@ def _handle_mouse_move(self, event) -> None: 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_widget.py b/src/phosphor/sweep_widget.py index 6f187c9..e746701 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,6 +254,12 @@ 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: # The graphic's own shape changed -- channel count, visible count, # sample rate, display duration -- so the MultiLine has to be diff --git a/tests/test_interaction.py b/tests/test_interaction.py new file mode 100644 index 0000000..9e2af35 --- /dev/null +++ b/tests/test_interaction.py @@ -0,0 +1,136 @@ +"""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 + +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)) diff --git a/tests/test_overlays.py b/tests/test_overlays.py index 6e9c75c..c580e96 100644 --- a/tests/test_overlays.py +++ b/tests/test_overlays.py @@ -220,16 +220,3 @@ def spy(self, *args): _, y_top2, _, y_bot2 = lines[0] assert y_bot2 == y_bot assert y_top2 < y_top - - -# ---- 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.""" - from phosphor.channel_plot import ChannelPlotWidget - - 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" From c6e8ddc866dfc7290e61caa1d7b60ebcb378a31c Mon Sep 17 00:00:00 2001 From: Chadwick Boulay Date: Wed, 5 Aug 2026 23:13:20 -0400 Subject: [PATCH 4/4] Map screen rows back to channels through the channel order The hover tooltip named the channel mirrored about the middle of the window: rows are laid out along world-y from the canvas bottom, and it read the channel off as channel_offset + row. That only holds bottom-up, and the sweep defaults to channel_order='top_down'. Event ticks had the same bug from the other direction, so an event on a channel drew its mark on the channel mirrored opposite -- silently, since a tick on a trace looks plausible wherever it lands. Both now go through _channel_at_row / _row_of_channel on the base widget, which is the only place the order is interpreted. The label overlay was already correct and now shares the same source of truth; that fixes the spectrum as a side effect, since it declares no order but offsets its rows bottom-up, and the old getattr default assumed top_down. --- src/phosphor/channel_plot.py | 39 ++++++++++++++++++++++++-- src/phosphor/sweep_widget.py | 3 +- tests/test_interaction.py | 53 ++++++++++++++++++++++++++++++++++++ 3 files changed, 90 insertions(+), 5 deletions(-) diff --git a/src/phosphor/channel_plot.py b/src/phosphor/channel_plot.py index 20b83cf..43f198d 100644 --- a/src/phosphor/channel_plot.py +++ b/src/phosphor/channel_plot.py @@ -282,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, @@ -344,6 +344,40 @@ def _on_wheel_event(self, event) -> None: 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. @@ -477,8 +511,7 @@ 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}" diff --git a/src/phosphor/sweep_widget.py b/src/phosphor/sweep_widget.py index e746701..4702a8f 100644 --- a/src/phosphor/sweep_widget.py +++ b/src/phosphor/sweep_widget.py @@ -309,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 index 9e2af35..956d776 100644 --- a/tests/test_interaction.py +++ b/tests/test_interaction.py @@ -8,6 +8,7 @@ """ import numpy as np +import pytest from phosphor.channel_plot import ChannelPlotWidget from phosphor.sweep_widget import SweepWidget @@ -134,3 +135,55 @@ def test_colour_rewrite_survives_a_single_visible_channel(): 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