mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 14:25:08 +00:00
perf(scroll): pace frames to the panel — 44→100 fps, stalls 14% → 0.02% (#523)
* perf(scroll): pace frames to the panel, not to a fixed sleep Scrolling ran at 44-46 fps on a 2x128x64 chain and 14-17% of frames took 41-53ms, which reads as judder. Four independent causes, each measured on the hardware; details and the diagnostic recipe are in docs/SCROLL_PERFORMANCE.md. The high-FPS loop slept a flat 8ms after every render. display() has already blocked on the panel's vsync by then, so that sleep was added to a wait that had happened: ~4ms of render plus 8ms put each iteration at ~12ms against a 10ms refresh grid, so every swap missed a refresh and the loop settled at 50fps while asking for 125 -- with no headroom, so a further 14% of frames slipped again. It now sleeps only the remainder, with a 1ms floor so plugin threads still get the GIL. ScrollHelper stepped position on a wall clock at 1/scroll_delay steps per second. Plugins set scroll_delay to the frame period, so that comparison sat exactly on its own threshold: a frame arriving a hair early moved zero pixels and rendered an identical frame, dirty-tracking skipped the swap, it returned in ~2ms, and the beat repeated. No scroll_delay value tunes that out -- a shorter delay trades stalled frames for periodic double-steps. Both modes now accumulate elapsed time at the same configured speed, so position stays proportional to real time. Sub-pixel blending goes back to off by default. It renders a half-step by mixing two adjacent columns, which on a coarse panel showing pixel-font text alternates crisp and smeared frames and reads as shimmer -- visibly worse than integer stepping on the hardware. Vegas mode still opts in. disk_cache uses orjson when importable, falling back to the stdlib. Encoding a ~1MB record drops from 14.8ms to 5.4ms end-to-end, and that work holds the GIL while a marquee is on screen. display_manager also checksummed the whole framebuffer twice per frame (dirty tracking, then the preview snapshot); the snapshot now takes the checksum the caller already computed. New src/common/scroll_config.py resolves scroll settings in one place. Five ticker plugins each hand-rolled this and disagreed: odds-ticker ranked the deprecated scroll_pixels_per_second above the documented scroll_speed/delay pair, and because that key carries a schema default the documented settings were dead for every user (ChuckBuilds/ledmatrix-plugins#408), while ledmatrix-leaderboard read the same key only as a fallback. The resolver also warns when a speed will not advance a whole number of pixels per refresh, which is the property that actually determines whether a scroll looks smooth. scripts/build_rgbmatrix_nogil.sh rebuilds the rgbmatrix binding so it releases the GIL. Upstream declares SwapOnVSync without nogil, unlike SetPixel/Clear/Fill beside it, so the render thread held the GIL for the whole vsync wait and starved background threads into long uninterruptible bursts. The script patches, builds and self-verifies into a scratch tree; --install backs up the original and rolls back if the service does not come back healthy. Measured after: 100 fps locked, no stalls observed, render thread down from 51% to 19% of one core. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(display): keep the panel swap locked to vsync while scrolling Dirty tracking skipped SwapOnVSync for byte-identical frames. That is the right call for static content, but SwapOnVSync is also what paces the render loop, so skipping it skips the wait for the panel: a duplicate frame returns in ~8ms instead of ~10ms on a 100Hz panel, advances the strip only 0.8px instead of 1.0px, and so makes the next frame more likely to repeat as well. The effect sustains itself once it starts. Measured over 20 minutes on a 2x128x64 chain, both scrollers configured identically at 100 px/s: leaderboard 10ms x35, 11ms x3 (clean) odds-ticker 10ms x26, 8ms x7, 15ms x5 (~20% duplicates mid-scroll) The duplicates were not end-of-cycle idling -- 38% of fast frames fell within 90s of a scroll completion against 35% of normal frames, a null result. The trigger is per-frame work: odds does more of it, and more variably, so it is first to land a frame that advances less than a whole pixel. Pushing an identical frame costs one canvas copy. Falling out of vsync lock costs smooth motion. Static content is untouched, because is_currently_scrolling() expires on its own inactivity threshold -- covered by test_stale_scrolling_state_stops_forcing_pushes so a plugin that stops scrolling without saying so cannot pin the panel into always-push. Also de-flakes test_snapshot_still_written_on_skip, which asserted a strict mtime increase between two writes that can land in the same filesystem tick; it failed about two runs in three on Windows regardless of the code under test. The file is now backdated before the check. 156 tests pass on the Pi. Not yet confirmed by eye on the panel. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scroll): report the frame-time tail, and stop the row-major blit Two problems, both found by looking at the panel rather than the metric. The frame-stats line reported ONE instantaneous frame every 5 seconds -- about 1 frame in 500 -- printed beside a 100-frame average. Both hide exactly the fault they are used to chase: a 2ms duplicate and a 21ms double-wait average to precisely 10ms, so a ticker stalling on half its frames still reports a healthy "Avg FPS: 100.0". That reading cost several rounds of chasing the wrong layer. The line now aggregates every frame since the last log and reports median, p95, max, min, and explicit stall and skip rates (past 1.5x the median missed a refresh; under half never reached the panel, because dirty tracking skipped the swap so the frame never waited on vsync). On the hardware this now reads: leaderboard 100.0 fps over 501 frames | median 10.00ms p95 10.05ms max 10.34ms | stalls 0 (0.0%) skips 0 (0.0%) The binding rebuild's blit patch becomes opt-in (RGB_PATCH_BLIT=1, default off). Reordering that loop to row-major changes what a torn frame looks like: column-major tearing shows as a vertical seam, row-major as a horizontal split between the panel's upper and lower halves. On a 1/32 scan panel that reads as a one-pixel fold across the middle of every panel, which is what was reported on hardware and what went away when the blit was reverted. All of the measured gain comes from the SwapOnVSync change, so the risky half is simply not worth taking; the header says so. Also fixes --install resolving its paths against $HOME, which is /root under sudo, so it looked in /root/rgbmatrix-nogil-build and died with "no built module found" on a machine where the build had just succeeded. It now resolves SUDO_USER's home. Both build paths are verified on the Pi: default yields one GIL-release site, RGB_PATCH_BLIT=1 yields two. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * feat(scroll): let users pick a crisp speed for their own panel Whole-pixel motion was previously only available at multiples of the refresh rate -- 100, 200, 300 px/s on a 100Hz panel. 100 px/s crosses a 256px panel in 2.6s, which is brisk for reading, and everything slower had to blend (blur) or repeat frames unevenly (judder). There was no way to ask for 50 px/s and get clean motion. SwapOnVSync takes a framerate_fraction the display manager never passed. It holds each frame for N panel refreshes; the panel keeps refreshing at its full rate throughout, so holding costs nothing in flicker and only changes how often a NEW image is presented. That turns 50 px/s into one whole pixel every second refresh instead of half a pixel every refresh. The crisp speeds are therefore refresh_hz / hold * pixels_per_frame, and that ladder depends on the panel: a Pi Zero on a long chain has a different set of good speeds from a Pi 4 on a short one. crisp_ladder() enumerates them and solve_crisp() picks the best match for a requested speed. solve_crisp weights motion quality rather than picking the numerically nearest entry, which matters more than it sounds. Asked for 30 px/s, nearest-by-value answers 28.6 -- 2px jumps at 14fps -- over 33.3, which is single-pixel motion at 33fps and obviously better on the panel. The target is also clamped into the ladder's range first, because relative error saturates near 1.0 for a target far outside it and the quality penalty would otherwise answer "10000 px/s" with the slowest entry. configure() snaps to the ladder and applies the hold when given a display manager. Without one the hold silently cannot happen and motion falls back to fractional pixels, so it warns rather than failing quietly. set_frame_hold() resets to 1 when scrolling stops, so one plugin's pacing cannot leak into whatever is on screen next. scripts/scroll_speeds.py is the user-facing part: it prints the ladder for the configured rate, measures what the panel ACTUALLY manages (--measure, for hardware that cannot reach its configured limit), highlights the nearest option to a wanted speed, and demos one live. It never starts or stops the display service itself -- doing that inside a script stranded the panel twice today. Speeds below ~20 px/s remain stepped regardless. That is the pixel pitch, not a software limit. Also fixes the dirty-tracking test spy, which stubbed SwapOnVSync with a single-argument function and would have masked the new call as a failed push, and rewrites a configure() test that had started passing for the wrong reason: it asserted a judder warning, which snapping now prevents, and was matching the unrelated "hold could not be applied" warning instead. 183 tests pass on the Pi. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scroll): tie the frame hold to the scroll, not the plugin The hold applied in configure() never reached the panel. Plugins share one display manager, and set_scrolling_state(False) -- fired whenever ANY other plugin finishes its scroll -- reset the hold to 1. A hold set once at plugin construction was therefore always gone by the time that plugin rendered. The symptom was a log line that lied. ledmatrix-stocks reported Scroll configured: 50.0 px/s (1px every 2 refreshes = 50.0 fps, smooth) while the panel measured 100.0 fps, median 10.00ms. Config, resolution and snapping were all correct; only the pacing silently was not applied. set_scrolling_state(is_scrolling, frame_hold=1) now carries it, so the hold lives exactly as long as the scroll that asked for it. configure() reports the value as ScrollSettings.frame_hold instead of applying it -- applying it behind the caller's back could never have been right on a shared display manager. Existing callers are unaffected; the default keeps one frame per refresh. Verified on hardware: stocks at 50 px/s now measures 50.0 fps over 251 frames | median 20.00ms p95 20.09ms | stalls 0 skips 0 20.00ms being exactly two refreshes, with the panel still refreshing at 100Hz underneath so flicker is unchanged. test_another_plugin_stopping_does_not_strand_a_hold pins the interaction that broke this. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(scroll,cache): resolve CodeRabbit review on #523 Eight findings, all reproduced before fixing. scroll_config.configure() read the refresh rate *after* resolve() had already used it. resolve() fills in target_fps, pixels_per_frame and the judder warning from that rate, so on a 60Hz panel every one of them described 100Hz -- and with snap_to_crisp=False nothing downstream corrected it, so set_target_fps() paced the helper to 100 FPS. The rate is now settled first, and falls back to the global config rather than straight to the default. refresh_hz_from_config() used `(cfg.get("display") or {}).get(...)`, which raises AttributeError when either level is truthy but not a mapping -- out of a function whose whole contract is a rate or a default. The frame-stats line reported the upper-middle sample as the median and the 96th sorted sample as p95 of 100. Both are also thresholds (stalls at 1.5x the median, skips at 0.5x), so the counts were biased too. The arithmetic is now in frame_stats()/format_frame_stats(), testable without a clock. configure()'s docstring and docs/SCROLL_PERFORMANCE.md still said it applies the frame hold and warns when it cannot. It deliberately does neither since "tie the frame hold to the scroll, not the plugin"; a caller following the old text would omit set_scrolling_state() and slow snapped speeds would still present every refresh. disk_cache had no policy for non-finite floats: orjson writes null, the stdlib writes NaN/Infinity, and orjson then rejects those legacy files so DiskCache.get deleted them as corrupt. One behaviour on both paths now -- write null, keep legacy records readable. allow_nan=False detects the values; the replacement walk runs only when there is one, so the ordinary write path is byte-identical and pays nothing. build_rgbmatrix_nogil.sh picked the build artifact with a glob piped to `head -1`, which sorts cpython-311 ahead of cpython-313, so a stale .so staged in from the source tree was installed as core.so while the GIL check -- which reads the generated core.cpp, not the .so -- still passed. It now requires the current interpreter's exact ABI name and fails closed. Its systemctl calls were also unchecked under `set -uo pipefail`: a failed stop left the old service running, the following start succeeded as a no-op, and the health check reported SUCCESS for a binding that was never loaded. orjson floor raised to 3.11.6 for CVE-2025-67221 (unbounded recursion in dumps); it covers the project's Python 3.10-3.13 range. Adds test/test_cache_nonfinite_floats.py (14) plus regression tests in test_scroll_config.py and test_scroll_helper.py. 9 of the cache tests and 9 of the scroll_config tests fail against the pre-fix code. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9 * test(harness): keep the visual double's signature tied to production Moves set_scrolling_state's frame_hold into the test double here, where DisplayManager gains it, rather than in #534 where it arrived a PR early. CodeRabbit flagged the #534 version correctly: a double that accepts an argument production does not lets the call pass every harness run and raise TypeError on the panel, which is the one failure a safety harness exists to prevent. The drift has now gone both ways across two branches -- double behind production on this branch, double ahead of it on #534 -- so it is pinned instead of remembered. test_display_double_parity.py compares the two signatures and fails with the direction of the drift named. It reads the files with ast rather than importing them, because display_manager imports rgbmatrix at module scope and this check should hold on a laptop and in CI as well as on a Pi. Plugins begin passing frame_hold in ledmatrix-plugins#462, which is why production and the double both need it before that lands. Full suite: 3889 passed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9 --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,158 @@
|
||||
"""Non-finite floats round-trip identically with and without orjson.
|
||||
|
||||
JSON has no NaN or Infinity. The stdlib emits them anyway as an extension;
|
||||
orjson refuses to and writes null. Cache files outlive the decision of which
|
||||
encoder is installed, so both halves of that gap are pinned here:
|
||||
|
||||
* writing -- installing orjson must not silently change what gets cached,
|
||||
so the stdlib path writes null too;
|
||||
* reading -- records already on disk carrying NaN or Infinity must stay
|
||||
readable, or installing orjson turns each of them into a "corrupted cache
|
||||
file" that DiskCache.get logs as an error and deletes.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import json
|
||||
import math
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from src.cache import disk_cache as disk_cache_module # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stdlib_cache():
|
||||
"""The module as it loads on a host with no orjson wheel."""
|
||||
# A None entry in sys.modules makes `import orjson` raise ImportError,
|
||||
# which is the branch we want, whether or not orjson is really installed.
|
||||
with mock.patch.dict(sys.modules, {"orjson": None}):
|
||||
module = importlib.reload(disk_cache_module)
|
||||
assert module.orjson is None
|
||||
yield module
|
||||
importlib.reload(disk_cache_module)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def orjson_cache():
|
||||
"""The module as it loads with orjson available."""
|
||||
module = importlib.reload(disk_cache_module)
|
||||
if module.orjson is None:
|
||||
pytest.skip("orjson is not installed on this host")
|
||||
return module
|
||||
|
||||
|
||||
NON_FINITE = {"nan": float("nan"), "inf": float("inf"), "ninf": float("-inf")}
|
||||
|
||||
|
||||
def _reject_constant(name):
|
||||
"""Make json.loads as strict as orjson about NaN/Infinity tokens."""
|
||||
raise AssertionError(f"non-spec JSON constant in output: {name}")
|
||||
|
||||
|
||||
class TestWritePolicy:
|
||||
"""Non-finite floats become null on whichever encoder is in use."""
|
||||
|
||||
def _assert_nulled(self, module):
|
||||
record = module._loads(module._dumps(dict(NON_FINITE, ok=1.5)))
|
||||
assert record["nan"] is None
|
||||
assert record["inf"] is None
|
||||
assert record["ninf"] is None
|
||||
# Finite values are untouched.
|
||||
assert record["ok"] == 1.5
|
||||
|
||||
def test_stdlib_writes_null(self, stdlib_cache):
|
||||
self._assert_nulled(stdlib_cache)
|
||||
|
||||
def test_orjson_writes_null(self, orjson_cache):
|
||||
self._assert_nulled(orjson_cache)
|
||||
|
||||
def test_stdlib_emits_spec_compliant_json(self, stdlib_cache):
|
||||
# The point of the write half: bytes written without orjson must still
|
||||
# parse once orjson is installed later. json.loads accepts the
|
||||
# extension tokens, so it cannot show this -- assert on the bytes, and
|
||||
# on a strict reader when there is one.
|
||||
raw = stdlib_cache._dumps(dict(NON_FINITE))
|
||||
assert b"NaN" not in raw
|
||||
assert b"Infinity" not in raw
|
||||
assert json.loads(raw, parse_constant=_reject_constant) == {
|
||||
"nan": None, "inf": None, "ninf": None}
|
||||
|
||||
def test_nested_non_finite_are_replaced(self, stdlib_cache):
|
||||
data = {"a": [1.0, float("nan"), {"b": float("inf")}], "c": (float("-inf"),)}
|
||||
record = stdlib_cache._loads(stdlib_cache._dumps(data))
|
||||
assert record["a"] == [1.0, None, {"b": None}]
|
||||
assert record["c"] == [None]
|
||||
|
||||
def test_finite_payloads_are_byte_identical_to_the_old_encoder(self, stdlib_cache):
|
||||
# allow_nan=False must not change ordinary output.
|
||||
data = {"x": 1, "y": [1.5, "s", True, None], "z": {"k": 2.25}}
|
||||
assert stdlib_cache._dumps(data) == json.dumps(
|
||||
data, cls=stdlib_cache.DateTimeEncoder).encode("utf-8")
|
||||
|
||||
|
||||
class TestReplaceNonFinite:
|
||||
def test_leaves_ordinary_values_alone(self):
|
||||
for value in (1, 1.5, "s", True, None, [], {}):
|
||||
assert disk_cache_module._replace_nonfinite(value) == value
|
||||
|
||||
def test_replaces_every_non_finite_float(self):
|
||||
for value in NON_FINITE.values():
|
||||
assert disk_cache_module._replace_nonfinite(value) is None
|
||||
|
||||
def test_walks_nested_containers(self):
|
||||
assert disk_cache_module._replace_nonfinite(
|
||||
{"a": [{"b": float("nan")}]}) == {"a": [{"b": None}]}
|
||||
|
||||
|
||||
class TestLegacyRecordsStayReadable:
|
||||
"""Files written before orjson arrived still load."""
|
||||
|
||||
LEGACY = b'{"timestamp": 1000.0, "value": NaN, "other": Infinity}'
|
||||
|
||||
def test_stdlib_reads_legacy_tokens(self, stdlib_cache):
|
||||
record = stdlib_cache._loads(self.LEGACY)
|
||||
assert math.isnan(record["value"])
|
||||
assert math.isinf(record["other"])
|
||||
|
||||
def test_orjson_falls_back_for_legacy_tokens(self, orjson_cache):
|
||||
record = orjson_cache._loads(self.LEGACY)
|
||||
assert math.isnan(record["value"])
|
||||
assert math.isinf(record["other"])
|
||||
|
||||
def test_genuinely_malformed_files_still_raise(self, stdlib_cache):
|
||||
with pytest.raises(json.JSONDecodeError):
|
||||
stdlib_cache._loads(b'{"a": ')
|
||||
|
||||
def test_orjson_still_raises_for_malformed_files(self, orjson_cache):
|
||||
with pytest.raises(json.JSONDecodeError):
|
||||
orjson_cache._loads(b'{"a": ')
|
||||
|
||||
|
||||
class TestDiskCacheEndToEnd:
|
||||
def _cache(self, module, tmp_path):
|
||||
return module.DiskCache(str(tmp_path))
|
||||
|
||||
def test_legacy_file_is_not_deleted_as_corrupt(self, orjson_cache, tmp_path):
|
||||
cache = self._cache(orjson_cache, tmp_path)
|
||||
path = cache.get_cache_path("legacy")
|
||||
Path(path).write_bytes(
|
||||
b'{"timestamp": %d, "value": NaN}' % int(__import__("time").time()))
|
||||
|
||||
record = cache.get("legacy", max_age=None)
|
||||
|
||||
assert record is not None, "legacy NaN record was treated as corrupt"
|
||||
assert math.isnan(record["value"])
|
||||
assert Path(path).exists(), "legacy NaN record was deleted"
|
||||
|
||||
def test_round_trip_through_set_and_get(self, stdlib_cache, tmp_path):
|
||||
cache = self._cache(stdlib_cache, tmp_path)
|
||||
cache.set("k", {"timestamp": __import__("time").time(),
|
||||
"value": float("nan")})
|
||||
record = cache.get("k", max_age=None)
|
||||
assert record is not None
|
||||
assert record["value"] is None
|
||||
@@ -47,12 +47,17 @@ class _SwapSpy:
|
||||
def __init__(self, matrix):
|
||||
self.matrix = matrix
|
||||
self.count = 0
|
||||
self.last_frame_hold = None
|
||||
self._orig = matrix.SwapOnVSync
|
||||
|
||||
def __enter__(self):
|
||||
def counting(canvas):
|
||||
def counting(canvas, *args):
|
||||
# *args carries framerate_fraction, which display_manager passes so
|
||||
# a frame can be held for several refreshes. Signature must match
|
||||
# the real binding or the spy hides a TypeError as a failed push.
|
||||
self.count += 1
|
||||
return self._orig(canvas)
|
||||
self.last_frame_hold = args[0] if args else 1
|
||||
return self._orig(canvas, *args)
|
||||
self.matrix.SwapOnVSync = counting
|
||||
return self
|
||||
|
||||
@@ -109,6 +114,11 @@ class TestDirtyTracking:
|
||||
dm.draw.rectangle([0, 0, 30, 8], fill=(255, 255, 0))
|
||||
dm.update_display() # push + snapshot write (first frame)
|
||||
assert os.path.exists(dm._snapshot_path)
|
||||
# Backdate the file so the "was it bumped?" check below cannot be
|
||||
# defeated by filesystem mtime granularity -- on Windows two writes in
|
||||
# the same tick get identical timestamps, which made this test fail
|
||||
# roughly two runs in three regardless of the code under test.
|
||||
os.utime(dm._snapshot_path, (time.time() - 60, time.time() - 60))
|
||||
first_mtime = os.path.getmtime(dm._snapshot_path)
|
||||
|
||||
# Age the write/touch bookkeeping past TOUCH_INTERVAL so the next
|
||||
@@ -125,6 +135,62 @@ class TestDirtyTracking:
|
||||
assert os.path.getmtime(dm._snapshot_path) > first_mtime
|
||||
|
||||
|
||||
class TestScrollLock:
|
||||
"""Dirty tracking must not skip the panel push while a scroll is running.
|
||||
|
||||
SwapOnVSync is what paces the render loop, so skipping it also skips the
|
||||
wait for the panel. A duplicate frame therefore returns early -- ~8ms
|
||||
instead of ~10ms on a 100Hz panel -- which advances the strip only 0.8px
|
||||
instead of 1.0px, which makes the NEXT frame more likely to be a duplicate
|
||||
too. That is self-sustaining: measured at ~20% duplicate frames mid-scroll
|
||||
on the odds ticker against essentially zero on a lighter plugin with
|
||||
identical scroll settings. Pushing an identical frame costs one canvas
|
||||
copy; falling out of vsync lock costs smooth motion.
|
||||
"""
|
||||
|
||||
def test_identical_frames_still_push_while_scrolling(self, dm):
|
||||
dm.draw.rectangle([0, 0, 12, 12], fill=(0, 0, 255))
|
||||
dm.update_display()
|
||||
dm.set_scrolling_state(True)
|
||||
try:
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
dm.update_display()
|
||||
dm.update_display()
|
||||
assert spy.count == 3, "scrolling must stay locked to the panel"
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
def test_identical_frames_are_skipped_when_not_scrolling(self, dm):
|
||||
"""The optimisation still applies to static content."""
|
||||
dm.set_scrolling_state(False)
|
||||
dm.draw.rectangle([0, 0, 14, 14], fill=(255, 0, 255))
|
||||
dm.update_display()
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
dm.update_display()
|
||||
assert spy.count == 0
|
||||
|
||||
def test_stale_scrolling_state_stops_forcing_pushes(self, dm):
|
||||
"""A plugin that stops scrolling without saying so must not pin the
|
||||
panel into always-push forever. is_currently_scrolling() expires on
|
||||
its own inactivity threshold, and the skip has to come back with it."""
|
||||
dm.draw.rectangle([0, 0, 16, 16], fill=(0, 255, 255))
|
||||
dm.update_display()
|
||||
dm.set_scrolling_state(True)
|
||||
try:
|
||||
# Backdate the activity marker past the inactivity threshold.
|
||||
dm._scrolling_state['last_scroll_activity'] = (
|
||||
time.time() - dm._scrolling_state['scroll_inactivity_threshold'] - 1.0)
|
||||
assert dm.is_currently_scrolling() is False
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
dm.update_display()
|
||||
assert spy.count == 0
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
|
||||
class TestKillSwitch:
|
||||
def test_dirty_tracking_can_be_disabled(self, dm):
|
||||
dm._dirty_tracking_enabled = False
|
||||
@@ -160,3 +226,89 @@ class TestKillSwitch:
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(pytest.main([__file__, "-v"]))
|
||||
|
||||
|
||||
class TestFrameHold:
|
||||
"""Holding a frame for N refreshes is how a scroll runs slower than one
|
||||
pixel per refresh without fractional pixel positions."""
|
||||
|
||||
def test_hold_reaches_swap_on_vsync(self, dm):
|
||||
dm.set_scrolling_state(True)
|
||||
dm.set_frame_hold(3)
|
||||
try:
|
||||
dm.draw.rectangle([0, 0, 9, 9], fill=(120, 0, 200))
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
assert spy.count == 1
|
||||
assert spy.last_frame_hold == 3
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
def test_default_is_every_refresh(self, dm):
|
||||
dm.set_scrolling_state(True)
|
||||
try:
|
||||
dm.draw.rectangle([0, 0, 11, 11], fill=(0, 120, 200))
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
assert spy.last_frame_hold == 1
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
def test_hold_resets_when_scrolling_stops(self, dm):
|
||||
"""One plugin's pacing must not leak into whatever is on screen next."""
|
||||
dm.set_scrolling_state(True)
|
||||
dm.set_frame_hold(5)
|
||||
dm.set_scrolling_state(False)
|
||||
dm.set_scrolling_state(True)
|
||||
try:
|
||||
dm.draw.rectangle([0, 0, 13, 13], fill=(200, 120, 0))
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
assert spy.last_frame_hold == 1
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
@pytest.mark.parametrize("bad,expected", [(0, 1), (-4, 1), (None, 1), ("x", 1)])
|
||||
def test_unusable_holds_are_ignored_or_floored(self, dm, bad, expected):
|
||||
dm.set_frame_hold(bad)
|
||||
assert dm._frame_hold == expected
|
||||
|
||||
|
||||
class TestFrameHoldLifetime:
|
||||
"""The hold must last exactly as long as the scroll that asked for it.
|
||||
|
||||
Plugins share one display manager. A hold applied at plugin construction is
|
||||
wiped the moment any *other* plugin finishes scrolling, so by the time the
|
||||
first plugin renders it is back to one pixel per refresh -- the speed reads
|
||||
correct in the log and is wrong on the panel.
|
||||
"""
|
||||
|
||||
def test_scrolling_state_carries_the_hold(self, dm):
|
||||
dm.set_scrolling_state(True, frame_hold=4)
|
||||
try:
|
||||
dm.draw.rectangle([0, 0, 7, 7], fill=(10, 200, 10))
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
assert spy.last_frame_hold == 4
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
def test_another_plugin_stopping_does_not_strand_a_hold(self, dm):
|
||||
dm.set_scrolling_state(True, frame_hold=3)
|
||||
dm.set_scrolling_state(False) # some other plugin finishes
|
||||
dm.set_scrolling_state(True) # a plugin that wants no hold
|
||||
try:
|
||||
dm.draw.rectangle([0, 0, 6, 6], fill=(200, 10, 10))
|
||||
with _SwapSpy(dm.matrix) as spy:
|
||||
dm.update_display()
|
||||
assert spy.last_frame_hold == 1
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
def test_default_keeps_previous_behaviour(self, dm):
|
||||
"""Callers that never heard of frame holds get one frame per refresh."""
|
||||
dm.set_scrolling_state(True)
|
||||
try:
|
||||
assert dm._frame_hold == 1
|
||||
finally:
|
||||
dm.set_scrolling_state(False)
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
"""The visual-test double must not drift from the real DisplayManager.
|
||||
|
||||
Both directions of drift are silent and both are damaging:
|
||||
|
||||
* the double accepts an argument production does not -- the call passes
|
||||
every harness run and raises TypeError on the panel, which is precisely
|
||||
the failure a safety harness exists to prevent;
|
||||
* the double lacks an argument production has -- every plugin that
|
||||
legitimately uses it fails every render, and the harness blames the
|
||||
plugin.
|
||||
|
||||
`set_scrolling_state` has been each of those in turn across two branches, so
|
||||
the parity is asserted rather than remembered.
|
||||
|
||||
Read with ast rather than imported: src/display_manager.py imports rgbmatrix
|
||||
at module scope, which is absent anywhere without the panel library, and this
|
||||
check should hold on a laptop and in CI as well as on a Pi.
|
||||
"""
|
||||
|
||||
import ast
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
REAL = ROOT / "src" / "display_manager.py"
|
||||
DOUBLE = ROOT / "src" / "plugin_system" / "testing" / "visual_display_manager.py"
|
||||
|
||||
#: Methods a plugin calls on whichever manager it is handed.
|
||||
SHARED_METHODS = ["set_scrolling_state", "is_currently_scrolling"]
|
||||
|
||||
|
||||
def _signature(path: Path, class_hint: str, method: str):
|
||||
"""(name, default-repr) pairs for `method`, or None if it is absent."""
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.ClassDef) or class_hint not in node.name:
|
||||
continue
|
||||
for item in node.body:
|
||||
if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)) and item.name == method:
|
||||
args = [a.arg for a in item.args.args if a.arg != "self"]
|
||||
pad = [None] * (len(args) - len(item.args.defaults))
|
||||
defaults = pad + [ast.unparse(d) for d in item.args.defaults]
|
||||
return list(zip(args, defaults))
|
||||
return None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("method", SHARED_METHODS)
|
||||
def test_the_double_matches_production(method):
|
||||
real = _signature(REAL, "DisplayManager", method)
|
||||
double = _signature(DOUBLE, "DisplayManager", method)
|
||||
|
||||
assert real is not None, f"DisplayManager lost {method}"
|
||||
assert double is not None, \
|
||||
f"the test double is missing {method}, so every plugin using it fails to render"
|
||||
assert double == real, (
|
||||
f"{method} has drifted: production takes {real}, the double takes {double}. "
|
||||
"A double that is more permissive hides a production TypeError; one that "
|
||||
"is less permissive fails plugins that are actually correct."
|
||||
)
|
||||
|
||||
|
||||
def test_frame_hold_is_accepted_by_both():
|
||||
"""The specific argument that has drifted twice."""
|
||||
for path, label in ((REAL, "DisplayManager"), (DOUBLE, "VisualTestDisplayManager")):
|
||||
params = dict(_signature(path, "DisplayManager", "set_scrolling_state") or [])
|
||||
assert "frame_hold" in params, f"{label} does not accept frame_hold"
|
||||
assert params["frame_hold"] == "1", \
|
||||
f"{label} must default frame_hold to 1 so existing callers are unaffected"
|
||||
@@ -0,0 +1,467 @@
|
||||
"""Tests for the shared scroll configuration resolver."""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from src.common.scroll_config import ( # noqa: E402
|
||||
DEFAULT_PIXELS_PER_SECOND,
|
||||
MAX_PIXELS_PER_FRAME,
|
||||
crisp_ladder,
|
||||
solve_crisp,
|
||||
MAX_PIXELS_PER_SECOND,
|
||||
MIN_PIXELS_PER_SECOND,
|
||||
ScrollSettings,
|
||||
configure,
|
||||
refresh_hz_from_config,
|
||||
resolve,
|
||||
)
|
||||
|
||||
|
||||
class FakeHelper:
|
||||
"""Records what configure() applied."""
|
||||
|
||||
def __init__(self, with_optional=True):
|
||||
self.speed = None
|
||||
self.frame_based = None
|
||||
self.target_fps = None
|
||||
if not with_optional:
|
||||
del FakeHelper.set_frame_based_scrolling
|
||||
del FakeHelper.set_target_fps
|
||||
|
||||
def set_scroll_speed(self, speed):
|
||||
self.speed = speed
|
||||
|
||||
def set_frame_based_scrolling(self, enabled):
|
||||
self.frame_based = enabled
|
||||
|
||||
def set_target_fps(self, fps):
|
||||
self.target_fps = fps
|
||||
|
||||
|
||||
class FakeDisplayManager:
|
||||
"""Records the frame hold configure() applies."""
|
||||
|
||||
def __init__(self):
|
||||
self.hold = None
|
||||
|
||||
def set_frame_hold(self, refreshes):
|
||||
self.hold = refreshes
|
||||
|
||||
|
||||
class MinimalHelper:
|
||||
"""An older helper exposing only set_scroll_speed."""
|
||||
|
||||
def __init__(self):
|
||||
self.speed = None
|
||||
|
||||
def set_scroll_speed(self, speed):
|
||||
self.speed = speed
|
||||
|
||||
|
||||
class TestPrecedence:
|
||||
def test_display_options_pair_wins(self):
|
||||
s = resolve({"display_options": {"scroll_speed": 1.0, "scroll_delay": 0.01}})
|
||||
assert s.pixels_per_second == 100.0
|
||||
assert s.source == "display_options.scroll_speed/delay"
|
||||
|
||||
def test_display_block_used_when_options_absent(self):
|
||||
s = resolve({"display": {"scroll_speed": 2.0, "scroll_delay": 0.01}})
|
||||
assert s.pixels_per_second == 200.0
|
||||
assert s.source == "display.scroll_speed/delay"
|
||||
|
||||
def test_root_pair_used_when_both_blocks_absent(self):
|
||||
s = resolve({"scroll_speed": 1.0, "scroll_delay": 0.02})
|
||||
assert s.pixels_per_second == 50.0
|
||||
|
||||
def test_pixels_per_second_used_when_no_pair_given(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 120.0}})
|
||||
assert s.pixels_per_second == 120.0
|
||||
assert s.source == "display_options.scroll_pixels_per_second"
|
||||
|
||||
def test_global_display_is_the_last_resort_before_default(self):
|
||||
s = resolve({}, {"display": {"scroll_speed": 1.0, "scroll_delay": 0.005}})
|
||||
assert s.pixels_per_second == 200.0
|
||||
|
||||
def test_default_when_nothing_configured(self):
|
||||
s = resolve({}, {})
|
||||
assert s.pixels_per_second == DEFAULT_PIXELS_PER_SECOND
|
||||
assert s.source == "default"
|
||||
|
||||
|
||||
class TestDeprecatedKeyCannotOverrideExplicitPair:
|
||||
"""Regression for ChuckBuilds/ledmatrix-plugins#408.
|
||||
|
||||
odds-ticker ranked scroll_pixels_per_second above the documented
|
||||
scroll_speed/scroll_delay pair. Because that key carries a schema default,
|
||||
the documented settings became unreachable for every user and the ticker
|
||||
silently ran at the default speed. The pair must win.
|
||||
"""
|
||||
|
||||
def test_pair_beats_pixels_per_second_in_the_same_block(self):
|
||||
s = resolve(
|
||||
{
|
||||
"display_options": {
|
||||
"scroll_speed": 1.0,
|
||||
"scroll_delay": 0.01,
|
||||
"scroll_pixels_per_second": 50.0, # schema default
|
||||
}
|
||||
}
|
||||
)
|
||||
assert s.pixels_per_second == 100.0
|
||||
assert "scroll_speed/delay" in s.source
|
||||
|
||||
def test_pair_beats_pixels_per_second_in_an_outer_block(self):
|
||||
s = resolve(
|
||||
{
|
||||
"display_options": {"scroll_speed": 1.0, "scroll_delay": 0.01},
|
||||
"scroll_pixels_per_second": 50.0,
|
||||
}
|
||||
)
|
||||
assert s.pixels_per_second == 100.0
|
||||
|
||||
|
||||
class TestMalformedValues:
|
||||
@pytest.mark.parametrize("bad", [None, "fast", "", {}, [], float("nan")])
|
||||
def test_unusable_speed_falls_through(self, bad):
|
||||
s = resolve({"display_options": {"scroll_speed": bad, "scroll_delay": 0.01}})
|
||||
assert s.pixels_per_second == DEFAULT_PIXELS_PER_SECOND
|
||||
|
||||
@pytest.mark.parametrize("bad", [0, -5, 0.0])
|
||||
def test_non_positive_values_fall_through(self, bad):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": bad}})
|
||||
assert s.pixels_per_second == DEFAULT_PIXELS_PER_SECOND
|
||||
|
||||
def test_booleans_are_not_treated_as_numbers(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": True}})
|
||||
assert s.pixels_per_second == DEFAULT_PIXELS_PER_SECOND
|
||||
|
||||
def test_zero_delay_does_not_divide_by_zero(self):
|
||||
s = resolve({"display_options": {"scroll_speed": 1.0, "scroll_delay": 0}})
|
||||
assert s.pixels_per_second == DEFAULT_PIXELS_PER_SECOND
|
||||
|
||||
def test_non_dict_blocks_are_ignored(self):
|
||||
s = resolve({"display_options": "nonsense", "display": 5})
|
||||
assert s.pixels_per_second == DEFAULT_PIXELS_PER_SECOND
|
||||
|
||||
|
||||
class TestClamping:
|
||||
def test_absurdly_fast_is_clamped(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 100000.0}})
|
||||
assert s.pixels_per_second == MAX_PIXELS_PER_SECOND
|
||||
assert s.warning and "clamped" in s.warning
|
||||
|
||||
def test_absurdly_slow_is_clamped(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 0.01}})
|
||||
assert s.pixels_per_second == MIN_PIXELS_PER_SECOND
|
||||
|
||||
|
||||
class TestWholePixelWarning:
|
||||
"""The display rule: whole pixels per refresh, or it judders."""
|
||||
|
||||
def test_no_warning_when_speed_divides_evenly(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 100.0}}, refresh_hz=100)
|
||||
assert s.warning is None
|
||||
assert s.pixels_per_frame == pytest.approx(1.0)
|
||||
|
||||
def test_no_warning_at_an_integer_multiple(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 200.0}}, refresh_hz=100)
|
||||
assert s.warning is None
|
||||
|
||||
def test_warns_on_a_half_pixel_per_frame(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 50.0}}, refresh_hz=100)
|
||||
assert s.warning is not None
|
||||
assert "judder" in s.warning
|
||||
assert "100 px/s" in s.warning
|
||||
|
||||
def test_warns_on_a_fractional_multiple(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 130.0}}, refresh_hz=100)
|
||||
assert s.warning is not None
|
||||
|
||||
def test_respects_a_non_default_refresh_rate(self):
|
||||
s = resolve({"display_options": {"scroll_pixels_per_second": 150.0}}, refresh_hz=150)
|
||||
assert s.warning is None
|
||||
assert s.pixels_per_frame == pytest.approx(1.0)
|
||||
|
||||
|
||||
class TestConfigure:
|
||||
def test_applies_time_based_mode_and_speed(self):
|
||||
helper = FakeHelper()
|
||||
s = configure(helper, {"display_options": {"scroll_speed": 1.0, "scroll_delay": 0.01}})
|
||||
assert helper.speed == 100.0
|
||||
assert helper.frame_based is False, "must not use the wall-clock step gate"
|
||||
assert helper.target_fps == 100.0
|
||||
assert s.pixels_per_second == 100.0
|
||||
|
||||
def test_works_against_a_helper_without_optional_methods(self):
|
||||
helper = MinimalHelper()
|
||||
configure(helper, {"display_options": {"scroll_pixels_per_second": 100.0}})
|
||||
assert helper.speed == 100.0
|
||||
|
||||
def test_snapping_removes_the_judder_warning(self, caplog):
|
||||
"""50px/s cannot be shown in whole pixels at 100Hz *per refresh*, but it
|
||||
can as 1px every 2nd refresh -- so once snapped there is nothing to warn
|
||||
about. resolve() alone still warns; configure() resolves it."""
|
||||
assert "judder" in resolve(
|
||||
{"display_options": {"scroll_pixels_per_second": 50.0}},
|
||||
refresh_hz=100).warning
|
||||
helper, dm = FakeHelper(), FakeDisplayManager()
|
||||
with caplog.at_level(logging.WARNING):
|
||||
settings = configure(
|
||||
helper, {"display_options": {"scroll_pixels_per_second": 50.0}},
|
||||
display_manager=dm)
|
||||
assert settings.warning is None
|
||||
assert not [r for r in caplog.records if "judder" in r.getMessage()]
|
||||
|
||||
def test_reports_the_hold_without_applying_it(self):
|
||||
"""The hold belongs to a scroll, not a plugin's lifetime -- one left set
|
||||
at construction is wiped as soon as any other plugin stops scrolling.
|
||||
configure() reports it; the caller passes it to set_scrolling_state."""
|
||||
dm = FakeDisplayManager()
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 25.0}},
|
||||
display_manager=dm)
|
||||
assert s.frame_hold == 4, "25px/s at 100Hz is 1px every 4th refresh"
|
||||
assert dm.hold is None, "must not apply the hold behind the caller's back"
|
||||
|
||||
def test_full_speed_needs_no_hold(self):
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 100.0}},
|
||||
display_manager=FakeDisplayManager())
|
||||
assert s.frame_hold == 1
|
||||
|
||||
def test_frame_hold_is_one_when_snapping_is_off(self):
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 50.0}},
|
||||
snap_to_crisp=False)
|
||||
assert s.frame_hold == 1
|
||||
|
||||
def test_snaps_to_the_nearest_crisp_speed_and_reports_both(self):
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 45.0}},
|
||||
display_manager=FakeDisplayManager())
|
||||
assert s.requested_pixels_per_second == 45.0
|
||||
assert s.pixels_per_second == 50.0
|
||||
assert s.crisp is not None and s.crisp.pixels_per_frame == 1
|
||||
|
||||
def test_snapping_can_be_turned_off(self):
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 45.0}},
|
||||
snap_to_crisp=False)
|
||||
assert s.pixels_per_second == 45.0
|
||||
assert s.crisp is None
|
||||
|
||||
def test_helper_target_fps_matches_the_presentation_rate(self):
|
||||
"""At a hold of 4 the panel still refreshes at 100Hz, but frames are
|
||||
presented at 25/s -- that is the rate the helper should pace to."""
|
||||
helper = FakeHelper()
|
||||
configure(helper, {"display_options": {"scroll_pixels_per_second": 25.0}},
|
||||
display_manager=FakeDisplayManager())
|
||||
assert helper.target_fps == pytest.approx(25.0)
|
||||
|
||||
def test_returns_settings_describing_the_source(self):
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_speed": 2.0, "scroll_delay": 0.01}})
|
||||
assert isinstance(s, ScrollSettings)
|
||||
assert "200.0 px/s" in s.describe()
|
||||
|
||||
|
||||
class TestRefreshFromConfig:
|
||||
def test_reads_the_hardware_limit(self):
|
||||
assert refresh_hz_from_config(
|
||||
{"display": {"hardware": {"limit_refresh_rate_hz": 150}}}
|
||||
) == 150.0
|
||||
|
||||
@pytest.mark.parametrize("cfg", [None, {}, {"display": {}}, {"display": {"hardware": {}}},
|
||||
"nonsense", {"display": {"hardware": {"limit_refresh_rate_hz": None}}}])
|
||||
def test_falls_back_to_the_default(self, cfg):
|
||||
assert refresh_hz_from_config(cfg) == 100.0
|
||||
|
||||
@pytest.mark.parametrize("cfg", [
|
||||
{"display": "nonsense"},
|
||||
{"display": ["nonsense"]},
|
||||
{"display": 60},
|
||||
{"display": {"hardware": "nonsense"}},
|
||||
{"display": {"hardware": ["nonsense"]}},
|
||||
{"display": {"hardware": 60}},
|
||||
])
|
||||
def test_malformed_nesting_falls_back_rather_than_raising(self, cfg):
|
||||
# Truthy-but-not-a-mapping at either level used to reach .get() on a
|
||||
# str/list and raise AttributeError out of a function whose whole
|
||||
# contract is "a refresh rate, or the default".
|
||||
assert refresh_hz_from_config(cfg) == 100.0
|
||||
|
||||
|
||||
class TestCrispLadder:
|
||||
"""Whole-pixel speeds available on a given panel."""
|
||||
|
||||
def test_every_entry_is_exactly_reachable(self):
|
||||
for c in crisp_ladder(100):
|
||||
assert c.pixels_per_second == pytest.approx(
|
||||
c.refresh_hz / c.frame_hold * c.pixels_per_frame)
|
||||
|
||||
def test_sorted_slowest_first(self):
|
||||
speeds = [c.pixels_per_second for c in crisp_ladder(100)]
|
||||
assert speeds == sorted(speeds)
|
||||
|
||||
def test_no_duplicate_speeds(self):
|
||||
speeds = [round(c.pixels_per_second, 3) for c in crisp_ladder(100)]
|
||||
assert len(speeds) == len(set(speeds))
|
||||
|
||||
def test_duplicates_resolve_to_the_smaller_step(self):
|
||||
"""100px/s is 1px every refresh or 2px every 2nd; prefer the former."""
|
||||
entry = next(c for c in crisp_ladder(100)
|
||||
if c.pixels_per_second == pytest.approx(100.0))
|
||||
assert entry.pixels_per_frame == 1
|
||||
assert entry.frame_hold == 1
|
||||
|
||||
def test_ladder_scales_with_the_panel(self):
|
||||
assert any(c.pixels_per_second == pytest.approx(60.0)
|
||||
for c in crisp_ladder(60))
|
||||
assert any(c.pixels_per_second == pytest.approx(30.0)
|
||||
for c in crisp_ladder(60))
|
||||
|
||||
def test_full_refresh_speed_is_present(self):
|
||||
for hz in (60, 75, 100, 120):
|
||||
assert any(c.pixels_per_second == pytest.approx(float(hz))
|
||||
for c in crisp_ladder(hz))
|
||||
|
||||
|
||||
class TestSolveCrisp:
|
||||
def test_exact_targets_are_matched_exactly(self):
|
||||
for target in (100, 50, 25, 20):
|
||||
assert solve_crisp(target, 100).pixels_per_second == pytest.approx(target)
|
||||
|
||||
def test_prefers_smooth_motion_over_raw_proximity(self):
|
||||
"""30 -> 33.3 (1px, 33fps), not 28.6 (2px at 14fps) which is nearer."""
|
||||
got = solve_crisp(30, 100)
|
||||
assert got.pixels_per_second == pytest.approx(33.333, abs=0.01)
|
||||
assert got.pixels_per_frame == 1
|
||||
|
||||
def test_does_not_chase_a_jumpy_exact_match(self):
|
||||
"""45 -> 50 (1px, smooth) beats 42.9 (3px at 14fps)."""
|
||||
assert solve_crisp(45, 100).pixels_per_frame == 1
|
||||
|
||||
def test_allows_a_two_pixel_step_when_it_is_clearly_closer(self):
|
||||
"""60 -> 66.7 (2px at 33fps) rather than 50 (1px) which is 17% slow."""
|
||||
got = solve_crisp(60, 100)
|
||||
assert got.pixels_per_second == pytest.approx(66.667, abs=0.01)
|
||||
assert got.pixels_per_frame == 2
|
||||
|
||||
def test_result_is_always_on_the_ladder(self):
|
||||
ladder = {round(c.pixels_per_second, 3) for c in crisp_ladder(100)}
|
||||
for target in range(5, 205, 5):
|
||||
assert round(solve_crisp(target, 100).pixels_per_second, 3) in ladder
|
||||
|
||||
def test_adapts_to_a_slower_panel(self):
|
||||
got = solve_crisp(30, 60)
|
||||
assert got.pixels_per_second == pytest.approx(30.0)
|
||||
assert got.pixels_per_frame == 1
|
||||
assert got.frame_hold == 2
|
||||
|
||||
def test_speeds_beyond_the_panel_clamp_to_the_fastest_available(self):
|
||||
got = solve_crisp(10_000, 100)
|
||||
assert got.frame_hold == 1
|
||||
assert got.pixels_per_frame == MAX_PIXELS_PER_FRAME
|
||||
|
||||
def test_steppiness_labels_are_sane(self):
|
||||
assert solve_crisp(100, 100).steppiness == "smooth"
|
||||
assert crisp_ladder(100)[0].steppiness == "stepped"
|
||||
|
||||
|
||||
class TestRefreshFromDisplayManager:
|
||||
"""A plugin sees only its own config section, so the display manager is the
|
||||
authoritative source for the panel's refresh rate."""
|
||||
|
||||
class DM:
|
||||
def __init__(self, hz):
|
||||
self.refresh_hz = hz
|
||||
self.hold = None
|
||||
|
||||
def set_frame_hold(self, refreshes):
|
||||
self.hold = refreshes
|
||||
|
||||
def test_uses_the_display_manager_rate(self):
|
||||
dm = self.DM(60.0)
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 30.0}},
|
||||
display_manager=dm)
|
||||
# 30px/s on a 60Hz panel is 1px every 2nd refresh, exactly.
|
||||
assert s.pixels_per_second == pytest.approx(30.0)
|
||||
assert s.frame_hold == 2
|
||||
|
||||
def test_explicit_refresh_hz_wins_over_the_display_manager(self):
|
||||
dm = self.DM(60.0)
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 25.0}},
|
||||
display_manager=dm, refresh_hz=100)
|
||||
assert s.frame_hold == 4, "25px/s at 100Hz is 1px every 4th refresh"
|
||||
|
||||
def test_missing_attribute_falls_back_to_the_default(self):
|
||||
class Bare:
|
||||
def set_frame_hold(self, refreshes):
|
||||
self.hold = refreshes
|
||||
|
||||
bare = Bare()
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 50.0}},
|
||||
display_manager=bare)
|
||||
assert s.frame_hold == 2, "assumed 100Hz"
|
||||
|
||||
def test_reported_settings_describe_the_display_manager_rate(self):
|
||||
"""Without snapping, nothing downstream corrects a wrong refresh rate.
|
||||
|
||||
The rate used to be read *after* resolve() had already filled in
|
||||
target_fps, pixels_per_frame and the judder warning from the 100Hz
|
||||
default -- so on a 60Hz panel every one of those described 100Hz, and
|
||||
set_target_fps() paced the helper to 100 FPS.
|
||||
"""
|
||||
dm = self.DM(60.0)
|
||||
helper = FakeHelper()
|
||||
s = configure(helper, {"display_options": {"scroll_pixels_per_second": 30.0}},
|
||||
display_manager=dm, snap_to_crisp=False)
|
||||
|
||||
assert s.target_fps == pytest.approx(60.0)
|
||||
assert s.pixels_per_frame == pytest.approx(0.5), "30px/s over 60 frames"
|
||||
assert helper.target_fps == pytest.approx(60.0)
|
||||
|
||||
def test_judder_warning_is_computed_at_the_real_refresh_rate(self):
|
||||
dm = self.DM(60.0)
|
||||
# 60px/s is exactly 1px per refresh at 60Hz -- crisp, no warning. At
|
||||
# the 100Hz default it is 0.6px per refresh and would be flagged.
|
||||
s = configure(FakeHelper(), {"display_options": {"scroll_pixels_per_second": 60.0}},
|
||||
display_manager=dm, snap_to_crisp=False)
|
||||
assert s.warning is None, s.warning
|
||||
|
||||
def test_global_config_supplies_the_rate_without_a_display_manager(self):
|
||||
s = configure(
|
||||
FakeHelper(),
|
||||
{"display_options": {"scroll_pixels_per_second": 30.0}},
|
||||
{"display": {"hardware": {"limit_refresh_rate_hz": 60}}},
|
||||
snap_to_crisp=False,
|
||||
)
|
||||
assert s.target_fps == pytest.approx(60.0)
|
||||
|
||||
|
||||
class TestFrameHoldIsReportedNotApplied:
|
||||
"""configure() reports the hold; the caller applies it when it scrolls.
|
||||
|
||||
The hold belongs to a scroll, not to a plugin's lifetime -- plugins share
|
||||
one display manager, so one set at construction is reset the moment any
|
||||
other plugin stops scrolling.
|
||||
"""
|
||||
|
||||
class RecordingDM:
|
||||
refresh_hz = 100.0
|
||||
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def set_scrolling_state(self, is_scrolling, frame_hold=1):
|
||||
self.calls.append((is_scrolling, frame_hold))
|
||||
|
||||
def set_frame_hold(self, refreshes):
|
||||
self.calls.append(("set_frame_hold", refreshes))
|
||||
|
||||
def test_configure_does_not_touch_the_display_manager(self):
|
||||
dm = self.RecordingDM()
|
||||
settings = configure(
|
||||
FakeHelper(), {"display_options": {"scroll_pixels_per_second": 25.0}},
|
||||
display_manager=dm)
|
||||
assert dm.calls == [], "configure() must not apply the hold itself"
|
||||
assert settings.frame_hold == 4, "but it must report what to apply"
|
||||
@@ -11,7 +11,11 @@ import time
|
||||
from unittest.mock import patch
|
||||
from PIL import Image
|
||||
|
||||
from src.common.scroll_helper import ScrollHelper
|
||||
from src.common.scroll_helper import (
|
||||
ScrollHelper,
|
||||
format_frame_stats,
|
||||
frame_stats,
|
||||
)
|
||||
|
||||
|
||||
DISPLAY_W = 64
|
||||
@@ -315,3 +319,77 @@ class TestGetScrollInfo:
|
||||
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_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 == []
|
||||
|
||||
Reference in New Issue
Block a user