Files
LEDMatrix/test/test_scroll_helper.py
T
ChuckandClaude Opus 5.5 ed0753c1ca perf: cheap per-frame and per-fetch savings (#725)
Six small savings with no behaviour change: the odds fetch no longer pretty-prints every response for a debug line; the scroll integer-slice path drops a redundant full-frame np.ascontiguousarray; ledmatrix-web.service gets MALLOC_ARENA_MAX=2 like the display unit; core ESPN responses are parsed via response_json (orjson when installed); and the scroll frame stats go to INFO only for degraded windows plus a 5-minute heartbeat.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-03 14:26:03 -04:00

566 lines
22 KiB
Python

"""
Tests for src/common/scroll_helper.py
Covers ScrollHelper: create_scrolling_image, update_scroll_position,
get_visible_portion, calculate_dynamic_duration, set_* methods,
reset_scroll, clear_cache, get_scroll_info.
"""
import numpy as np
import pytest
import time
from unittest.mock import patch
from PIL import Image
from src.common.scroll_helper import (
ScrollHelper,
format_frame_stats,
frame_stats,
)
DISPLAY_W = 64
DISPLAY_H = 32
@pytest.fixture
def helper():
return ScrollHelper(display_width=DISPLAY_W, display_height=DISPLAY_H)
def _make_image(width: int = 64, height: int = 32, color=(255, 0, 0)) -> Image.Image:
img = Image.new("RGB", (width, height), color)
return img
# ---------------------------------------------------------------------------
# __init__ / initial state
# ---------------------------------------------------------------------------
class TestScrollHelperInit:
def test_initial_scroll_position(self, helper):
assert helper.scroll_position == 0.0
def test_initial_scroll_complete_false(self, helper):
assert helper.scroll_complete is False
def test_display_dimensions(self, helper):
assert helper.display_width == DISPLAY_W
assert helper.display_height == DISPLAY_H
# ---------------------------------------------------------------------------
# create_scrolling_image
# ---------------------------------------------------------------------------
class TestCreateScrollingImage:
def test_empty_content_returns_blank_image(self, helper):
result = helper.create_scrolling_image([])
assert isinstance(result, Image.Image)
assert helper.total_scroll_width == 0
def test_single_item_creates_image(self, helper):
img = _make_image(width=100)
result = helper.create_scrolling_image([img])
assert isinstance(result, Image.Image)
assert result.width > DISPLAY_W # includes leading gap
def test_multiple_items_wider_image(self, helper):
items = [_make_image(width=50), _make_image(width=50)]
result = helper.create_scrolling_image(items)
# Should be wider than two items alone
assert result.width > 100
def test_scroll_position_reset(self, helper):
helper.scroll_position = 500.0
helper.create_scrolling_image([_make_image()])
assert helper.scroll_position == 0.0
def test_cached_array_set(self, helper):
helper.create_scrolling_image([_make_image()])
assert helper.cached_array is not None
def test_scroll_complete_reset(self, helper):
helper.scroll_complete = True
helper.create_scrolling_image([_make_image()])
assert helper.scroll_complete is False
def test_total_scroll_width_matches_image(self, helper):
img = _make_image(width=200)
result = helper.create_scrolling_image([img])
assert helper.total_scroll_width == result.width
# ---------------------------------------------------------------------------
# set_scrolling_image
# ---------------------------------------------------------------------------
class TestSetScrollingImage:
def test_sets_cached_image(self, helper):
img = _make_image(width=200)
helper.set_scrolling_image(img)
assert helper.cached_image is img
def test_sets_cached_array(self, helper):
img = _make_image(width=200)
helper.set_scrolling_image(img)
assert helper.cached_array is not None
def test_scroll_width_matches_image(self, helper):
img = _make_image(width=300)
helper.set_scrolling_image(img)
assert helper.total_scroll_width == 300
def test_none_clears_cache(self, helper):
helper.set_scrolling_image(_make_image())
helper.set_scrolling_image(None)
assert helper.cached_image is None
# ---------------------------------------------------------------------------
# update_scroll_position (time-based mode)
# ---------------------------------------------------------------------------
class TestUpdateScrollPosition:
def test_position_advances_over_time(self, helper):
helper.create_scrolling_image([_make_image(width=200)])
helper.scroll_speed = 100.0 # 100 px/s
helper.last_update_time = time.time() - 0.1 # pretend 100ms elapsed
initial = helper.scroll_position
helper.update_scroll_position()
assert helper.scroll_position > initial
def test_no_advance_without_image(self, helper):
helper.update_scroll_position() # no image, should not crash
assert helper.scroll_position == 0.0
def test_zero_width_content_stays_zero(self, helper):
helper.create_scrolling_image([]) # empty → width 0
helper.update_scroll_position()
assert helper.scroll_position == 0.0
def test_scroll_complete_clamped(self, helper):
helper.create_scrolling_image([_make_image(width=100)])
# Force position past the end
helper.scroll_position = helper.total_scroll_width + 50
helper.total_distance_scrolled = helper.total_scroll_width + 50
helper.update_scroll_position()
assert helper.scroll_complete is True
assert helper.scroll_position <= helper.total_scroll_width
# ---------------------------------------------------------------------------
# get_visible_portion
# ---------------------------------------------------------------------------
class TestGetVisiblePortion:
def test_returns_none_without_image(self, helper):
assert helper.get_visible_portion() is None
def test_returns_image_sized_to_display(self, helper):
helper.create_scrolling_image([_make_image(width=200)])
visible = helper.get_visible_portion()
assert visible is not None
assert visible.width == DISPLAY_W
assert visible.height == DISPLAY_H
def test_different_positions_give_different_images(self, helper):
helper.create_scrolling_image([_make_image(width=300)])
img1 = helper.get_visible_portion()
helper.scroll_position = 50
img2 = helper.get_visible_portion()
# Images should differ (colour from scrolled content)
# Just verify both are valid PIL images with correct size
assert img1.width == img2.width == DISPLAY_W
@pytest.mark.parametrize("start_x", [0, 1, 37, 200 - DISPLAY_W])
def test_integer_slice_is_byte_identical_to_a_contiguous_copy(
self, helper, start_x):
# The integer path dropped np.ascontiguousarray() before tobytes():
# a column slice of the strip is not C-contiguous, and tobytes()
# must still give the same C-order bytes the copy did.
rng = np.random.default_rng(start_x)
strip = rng.integers(0, 256, (DISPLAY_H, 200, 3), dtype=np.uint8)
helper.cached_array = strip
view = strip[:, start_x:start_x + DISPLAY_W]
assert not view.flags["C_CONTIGUOUS"]
frame = helper._get_visible_portion_integer(start_x, start_x + DISPLAY_W)
assert frame.tobytes() == np.ascontiguousarray(view).tobytes()
assert view.tobytes() == np.ascontiguousarray(view).tobytes()
# ---------------------------------------------------------------------------
# reset_scroll / clear_cache
# ---------------------------------------------------------------------------
class TestResetAndClear:
def test_reset_restores_position(self, helper):
helper.create_scrolling_image([_make_image(width=200)])
helper.scroll_position = 100.0
helper.reset_scroll()
assert helper.scroll_position == 0.0
def test_reset_clears_complete_flag(self, helper):
helper.scroll_complete = True
helper.reset_scroll()
assert helper.scroll_complete is False
def test_reset_alias(self, helper):
helper.scroll_position = 50.0
helper.reset()
assert helper.scroll_position == 0.0
def test_clear_cache(self, helper):
helper.create_scrolling_image([_make_image()])
helper.clear_cache()
assert helper.cached_image is None
assert helper.cached_array is None
assert helper.total_scroll_width == 0
# ---------------------------------------------------------------------------
# calculate_dynamic_duration
# ---------------------------------------------------------------------------
class TestCalculateDynamicDuration:
def test_returns_min_when_disabled(self, helper):
helper.dynamic_duration_enabled = False
helper.min_duration = 30
result = helper.calculate_dynamic_duration()
assert result == 30
def test_returns_min_when_no_content(self, helper):
helper.total_scroll_width = 0
helper.min_duration = 30
result = helper.calculate_dynamic_duration()
assert result == 30
def test_respects_min_duration(self, helper):
helper.create_scrolling_image([_make_image(width=50)])
helper.min_duration = 60
helper.max_duration = 300
helper.scroll_speed = 500.0 # very fast → very short time
result = helper.calculate_dynamic_duration()
assert result >= 60
def test_respects_max_duration(self, helper):
helper.create_scrolling_image([_make_image(width=5000)])
helper.min_duration = 10
helper.max_duration = 60
helper.scroll_speed = 1.0 # very slow → very long time
result = helper.calculate_dynamic_duration()
assert result <= 60
def test_time_based_calculation(self, helper):
helper.create_scrolling_image([_make_image(width=200)])
helper.scroll_speed = 100.0
helper.min_duration = 1
helper.max_duration = 600
helper.frame_based_scrolling = False
result = helper.calculate_dynamic_duration()
assert isinstance(result, int)
assert result > 0
# ---------------------------------------------------------------------------
# set_* configuration methods
# ---------------------------------------------------------------------------
class TestSetMethods:
def test_set_scroll_speed_time_based(self, helper):
helper.frame_based_scrolling = False
helper.set_scroll_speed(50.0)
assert helper.scroll_speed == 50.0
def test_set_scroll_speed_clamped_low(self, helper):
helper.frame_based_scrolling = False
helper.set_scroll_speed(0.0)
assert helper.scroll_speed >= 1.0
def test_set_scroll_speed_clamped_high(self, helper):
helper.frame_based_scrolling = False
helper.set_scroll_speed(10000.0)
assert helper.scroll_speed <= 500.0
def test_set_scroll_delay(self, helper):
helper.set_scroll_delay(0.05)
assert helper.scroll_delay == 0.05
def test_set_scroll_delay_clamped(self, helper):
helper.set_scroll_delay(0.0001)
assert helper.scroll_delay >= 0.001
def test_set_target_fps(self, helper):
helper.set_target_fps(60.0)
assert helper.target_fps == 60.0
def test_set_target_fps_clamped(self, helper):
helper.set_target_fps(1000.0)
assert helper.target_fps <= 200.0
def test_set_sub_pixel_scrolling(self, helper):
helper.set_sub_pixel_scrolling(True)
assert helper.sub_pixel_scrolling is True
helper.set_sub_pixel_scrolling(False)
assert helper.sub_pixel_scrolling is False
def test_set_frame_based_scrolling(self, helper):
helper.set_frame_based_scrolling(True)
assert helper.frame_based_scrolling is True
def test_set_dynamic_duration_settings(self, helper):
helper.set_dynamic_duration_settings(enabled=True, min_duration=20, max_duration=120, buffer=0.2)
assert helper.dynamic_duration_enabled is True
assert helper.min_duration == 20
assert helper.max_duration == 120
assert helper.duration_buffer == pytest.approx(0.2)
# ---------------------------------------------------------------------------
# get_scroll_info
# ---------------------------------------------------------------------------
class TestGetScrollInfo:
def test_returns_dict(self, helper):
info = helper.get_scroll_info()
assert isinstance(info, dict)
def test_required_keys(self, helper):
info = helper.get_scroll_info()
for key in ("scroll_position", "total_distance_scrolled", "scroll_speed",
"scroll_complete", "dynamic_duration"):
assert key in info
def test_scroll_position_reflected(self, helper):
helper.scroll_position = 42.0
info = helper.get_scroll_info()
assert info["scroll_position"] == 42.0
class TestFrameStatsPercentiles:
"""The stats line is the instrument this whole scroll change is measured
with, so its median and p95 have to be the real ones.
Both are also thresholds: stalls are counted at 1.5x the median and skips
at 0.5x, so an off-by-one in the median biases the counts as well as the
printed numbers.
"""
# 100 samples of 1..100ms. True median 50.5ms (the mean of the two middle
# samples, not the upper one at 51ms); nearest-rank p95 is the 95th
# sample at 95ms, not the 96th at 96ms.
HUNDRED = [i / 1000.0 for i in range(1, 101)]
def test_even_window_median_averages_both_middle_samples(self):
assert frame_stats(self.HUNDRED)["median"] == pytest.approx(0.0505)
def test_p95_uses_nearest_rank(self):
assert frame_stats(self.HUNDRED)["p95"] == pytest.approx(0.095)
def test_odd_window_median_is_the_middle_sample(self):
times = [i / 1000.0 for i in range(1, 102)] # 101 samples
assert frame_stats(times)["median"] == pytest.approx(0.051)
def test_single_sample_window_does_not_index_out_of_range(self):
stats = frame_stats([0.010])
assert stats["median"] == pytest.approx(0.010)
assert stats["p95"] == pytest.approx(0.010)
assert stats["min"] == stats["max"] == pytest.approx(0.010)
def test_two_sample_window(self):
stats = frame_stats([0.010, 0.020])
assert stats["median"] == pytest.approx(0.015)
assert stats["p95"] == pytest.approx(0.020)
def test_input_order_does_not_matter(self):
assert frame_stats(list(reversed(self.HUNDRED))) == frame_stats(self.HUNDRED)
def test_caller_window_is_not_mutated(self):
times = [0.030, 0.010, 0.020]
frame_stats(times)
assert times == [0.030, 0.010, 0.020]
def test_stall_threshold_follows_the_median(self):
# Ten 10ms frames and two 30ms stalls: median 10ms, so >15ms is a
# stall -- exactly the two.
stats = frame_stats([0.010] * 10 + [0.030] * 2)
assert stats["stalls"] == 2
assert stats["skips"] == 0
def test_skips_are_frames_that_never_reached_the_panel(self):
stats = frame_stats([0.010] * 10 + [0.002] * 3)
assert stats["skips"] == 3
assert stats["stalls"] == 0
def test_fps_is_the_reciprocal_of_the_mean(self):
assert frame_stats([0.010] * 50)["fps"] == pytest.approx(100.0)
def test_formatted_line_reports_the_corrected_values(self):
line = format_frame_stats(self.HUNDRED)
assert "median 50.50ms" in line, line
assert "p95 95.00ms" in line, line
assert "over 100 frames" in line, line
def test_log_frame_rate_emits_the_line_and_clears_the_window(self, helper):
helper._window = [0.010] * 20
helper.last_frame_time = time.time() # arm the clock; see below
helper.last_fps_log_time = 0.0 # force the 5s boundary
with patch.object(helper.logger, "info") as info:
helper.log_frame_rate()
assert info.called
assert "Scroll frame stats" in info.call_args[0][0]
assert helper._window == []
class TestFrameStatsLogLevel:
"""The stats line is INFO only when a window is degraded, on the window
that recovers, and as a 5-minute heartbeat; every other window is DEBUG.
Every 5s from every scroller at INFO was most of a healthy rig's journal.
"""
HEALTHY = [0.010] * 500
# 10% of frames a whole refresh late: fps 90.9 against a locked 100.
SLOW = [0.010] * 450 + [0.020] * 50
# 2% stalled: barely moves the mean, but it is the judder to see.
STALLING = [0.010] * 490 + [0.025] * 10
def _log_window(self, helper, window):
helper._window = list(window)
helper.last_frame_time = time.time()
helper.last_fps_log_time = 0.0
with patch.object(helper.logger, "info") as info, \
patch.object(helper.logger, "debug") as debug, \
patch.object(helper.logger, "isEnabledFor", return_value=True):
helper.log_frame_rate()
return info, debug
def test_degraded_predicate(self):
from src.common.scroll_helper import frame_stats_degraded
assert not frame_stats_degraded(frame_stats(self.HEALTHY))
assert frame_stats_degraded(frame_stats(self.SLOW))
assert frame_stats_degraded(frame_stats(self.STALLING))
# One stall in 500 is a normal wobble, not degraded.
assert not frame_stats_degraded(
frame_stats([0.010] * 499 + [0.025]))
def test_first_window_is_info_as_a_heartbeat(self, helper):
info, debug = self._log_window(helper, self.HEALTHY)
assert info.called and not debug.called
def test_healthy_window_after_the_heartbeat_is_debug(self, helper):
helper._stats_last_info_log = time.time()
info, debug = self._log_window(helper, self.HEALTHY)
assert not info.called
assert "Scroll frame stats" in debug.call_args[0][0]
@pytest.mark.parametrize("window", ["SLOW", "STALLING"])
def test_degraded_window_is_info(self, helper, window):
helper._stats_last_info_log = time.time()
info, debug = self._log_window(helper, getattr(self, window))
assert "Scroll frame stats" in info.call_args[0][0]
assert not debug.called
def test_recovery_window_is_info_then_quiet(self, helper):
helper._stats_last_info_log = time.time()
self._log_window(helper, self.SLOW)
info, _ = self._log_window(helper, self.HEALTHY)
assert info.called, "the recovery was not reported"
info, debug = self._log_window(helper, self.HEALTHY)
assert not info.called and debug.called
def test_heartbeat_comes_back_after_the_interval(self, helper):
from src.common.scroll_helper import STATS_HEARTBEAT_INTERVAL
helper._stats_last_info_log = time.time() - STATS_HEARTBEAT_INTERVAL - 1
info, _ = self._log_window(helper, self.HEALTHY)
assert info.called
def test_reset_scroll_does_not_rearm_the_heartbeat(self, helper):
helper._stats_last_info_log = time.time()
helper.reset_scroll()
info, _ = self._log_window(helper, self.HEALTHY)
assert not info.called
class TestIdleGapIsNotAFrame:
"""The first frame of a scroll has no predecessor, so timing one measures
the idle gap since the last scroll rather than a frame.
Left in, that gap lands in the max field of otherwise healthy windows and
counts as one stall per scroll start -- about 0.2% at 500 frames to a
window, which is the same order as the real stall rates it sits next to.
The stall rate is the number used to judge whether a scroll change worked,
so it has to be clean.
"""
def test_clock_starts_unarmed(self, helper):
assert helper.last_frame_time is None
def test_first_frame_seeds_the_clock_without_sampling(self, helper):
helper.log_frame_rate()
assert helper.last_frame_time is not None
assert helper._window == []
assert helper.frame_times == []
def test_second_frame_is_sampled(self, helper):
helper.log_frame_rate()
helper.log_frame_rate()
assert len(helper._window) == 1
def test_reset_scroll_disarms_the_clock(self, helper):
helper.log_frame_rate()
assert helper.last_frame_time is not None
helper.reset_scroll()
assert helper.last_frame_time is None
def test_gap_longer_than_the_log_interval_is_dropped(self, helper):
"""Covers callers that scroll without ever calling reset_scroll()."""
helper.last_frame_time = time.time() - 137.0 # a real observed gap
helper.log_frame_rate()
assert helper._window == []
assert helper.frame_times == []
def test_a_gap_does_not_reach_the_stats_line(self, helper):
helper.last_frame_time = time.time() - 137.0
helper.last_fps_log_time = 0.0 # the 5s boundary is also due
with patch.object(helper.logger, "info") as info:
helper.log_frame_rate()
assert not info.called, "the idle gap was reported as a frame"
def test_window_of_only_gaps_logs_nothing(self, helper):
"""A window whose every sample was dropped has nothing to report --
and reporting the gap itself is the bug this guards."""
helper.last_fps_log_time = 0.0
with patch.object(helper.logger, "info") as info:
helper.log_frame_rate() # seeds
helper.last_frame_time = time.time() - 137.0
helper.log_frame_rate() # dropped
assert not info.called
def test_seeding_restarts_the_window_timer(self, helper):
"""A new scroll should not open by reporting a one-frame window."""
helper.last_fps_log_time = 0.0 # boundary long overdue
helper.log_frame_rate() # seeds
with patch.object(helper.logger, "info") as info:
helper.log_frame_rate() # first real frame
assert not info.called
assert len(helper._window) == 1
def test_a_dropped_gap_restarts_the_window_timer_too(self, helper):
"""The size-guard path is the one scrollers that never call
reset_scroll() take, so it must restart the window like the sentinel
does, or their next scroll opens with a one-frame stats line."""
helper.log_frame_rate() # seeds
helper.last_frame_time = time.time() - 137.0
helper.last_fps_log_time = 0.0 # boundary long overdue
with patch.object(helper.logger, "info") as info:
helper.log_frame_rate() # the gap, dropped
helper.log_frame_rate() # first real frame
assert not info.called, "a one-frame window was reported"
assert len(helper._window) == 1
def test_a_normal_frame_still_counts(self, helper):
helper.last_frame_time = time.time() - 0.010
helper.log_frame_rate()
assert len(helper._window) == 1
assert helper._window[0] == pytest.approx(0.010, abs=0.005)