From 735968137cda54bf021673d8e06903a75866d43b Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Wed, 7 Oct 2026 12:45:07 -0500 Subject: [PATCH 1/3] displaydev: frames per second and present time, off by default DisplayDriver.measure_fps() wraps the instance's show and flush_rect with timers (restored when switched off); fps() reads frames, the last second's and the average rate, present time per frame and the busy share; fps_reset() and fps_print() on a multimer timer. PYDEVICES_FPS=1 and PYDEVICES_FPS_PRINT=N switch it on at driver start. frame_done() is the hook a GUI calls when it presents with flush_rect (LVGL direct panels). appdev's refresh timer now looks display.show up on each tick, so it calls a wrapper installed after the App started. --- docs/displaydev.md | 37 ++++ lib/appdev/app.py | 9 +- lib/displaydev/__init__.py | 290 ++++++++++++++++++++++++++++++ tests/README.md | 1 + tests/test_displaydev_fps.py | 330 +++++++++++++++++++++++++++++++++++ 5 files changed, 663 insertions(+), 4 deletions(-) create mode 100644 tests/test_displaydev_fps.py diff --git a/docs/displaydev.md b/docs/displaydev.md index b2e6f93a..3f2e0104 100644 --- a/docs/displaydev.md +++ b/docs/displaydev.md @@ -171,6 +171,43 @@ pydevices-examples does not include a task scheduler. Options: - **`asyncio`** — works on CPython, MicroPython, and PyScript (required there) - **[multimer](multimer.md)** — one `Timer` on every host; it rides the page's loop on PyScript +## Frame rate + +To see how fast an app draws, switch on the display's meter and read it: + +```python +display_drv.measure_fps(True) # or PYDEVICES_FPS=1 in the environment +... +display_drv.fps() +# {"fps": 29.8, "avg_fps": 29.5, "frames": 1180, "seconds": 40.0, +# "present_ms": 4.2, "present_ms_max": 12.0, "busy": 0.13, +# "flushes": 0, "pixels": 0} +display_drv.fps_reset() # start a fresh measurement +display_drv.fps_print(5) # one line every 5 s; 0 stops it +``` + +`PYDEVICES_FPS=1` measures any example without editing it, and +`PYDEVICES_FPS_PRINT=5` also prints a line every five seconds, for tailing a +log: + +```text +fps 29.8 avg 29.5 frames 1180 present 4.2 ms (max 12.0) busy 13% +``` + +`fps` is the last whole second and `avg_fps` everything since it was switched +on or reset. `present_ms` is the time spent inside the present call per frame, +and `busy` is the share of wall time spent there: a slow frame rate with a high +`busy` means the display is the bottleneck, a low one means the app is. + +A frame is one `show()`. On a panel LVGL draws into directly, LVGL presents +with `flush_rect()` once per dirty area, so `display_driver` calls +`frame_done()` when a frame's last area is out; those `flush_rect()` calls are +counted as `flushes` and `pixels`, not frames. + +It is off by default and costs nothing then: `show` and `flush_rect` are the +class's own methods until you switch it on, and again after +`measure_fps(False)`. + ## Vertical scrolling Many drivers expose **ILI9341-style** vertical scroll: a top fixed band (TFA), a scrollable middle (VSA), and a bottom fixed band (BFA). You define regions with `set_vscroll(tfa, bfa)` or `vscrdef`, then move content with the `vscroll` property (wrapper around `vscsad`). diff --git a/lib/appdev/app.py b/lib/appdev/app.py index 7872047f..1476f44a 100644 --- a/lib/appdev/app.py +++ b/lib/appdev/app.py @@ -445,14 +445,15 @@ def _wire_one_refresh(self, display, refresh_period): period = int(refresh_period) if period <= 0: return - show = getattr(display, "show", None) - if not callable(show): + if not callable(getattr(display, "show", None)): return - def _show(timer_obj, _display=display, _show=show): + # Look show up on every tick: a wrapper installed after the App started + # (displaydev's measure_fps) must be the one that is called. + def _show(timer_obj, _display=display): if self._refresh_paused: return - _show(timer_obj) + _display.show(timer_obj) name = "refresh:%s" % (getattr(display, "__class__", type(display)).__name__,) tim = multimer.every(period, _show, name=name) diff --git a/lib/displaydev/__init__.py b/lib/displaydev/__init__.py index 87365c63..b851a6bd 100644 --- a/lib/displaydev/__init__.py +++ b/lib/displaydev/__init__.py @@ -493,6 +493,156 @@ def _tick(self, _timer=None): fn() +_fps_clock_cache = None + + +def _fps_clock(): + """``(now, diff)`` for timing presents: ``diff(later, earlier)`` is in µs. + + ``time.ticks_us`` on MicroPython, ``perf_counter_ns`` on CPython, + ``monotonic_ns`` on CircuitPython, and ``supervisor.ticks_ms`` (ms steps) + on a CircuitPython build without long ints. Chosen on first use, so a + driver that never measures never pays for it. + """ + global _fps_clock_cache + if _fps_clock_cache is None: + import time + + ticks_us = getattr(time, "ticks_us", None) + ns = getattr(time, "perf_counter_ns", None) or getattr(time, "monotonic_ns", None) + if ticks_us is not None: + _fps_clock_cache = (ticks_us, time.ticks_diff) + elif ns is not None: + + def now(): + return ns() // 1000 + + def diff(a, b): + return a - b + + _fps_clock_cache = (now, diff) + else: + from supervisor import ticks_ms + + def diff_ms(a, b): + return (((a - b) + 0x10000000) & 0x1FFFFFFF) - 0x10000000 + + def diff(a, b): + return diff_ms(a, b) * 1000 + + _fps_clock_cache = (ticks_ms, diff) + return _fps_clock_cache + + +class _FpsMeter: + """Frame and present-time counters for one display (see ``measure_fps``). + + Counts go into a window that is folded into the totals once a second, so + the per-frame arithmetic stays on small ints (no heap on a board) and the + last whole second's rate falls out of the fold. Wall time is added up + between events rather than read against a start time, so a 32-bit + board's wrapping ``ticks_us`` is fine as long as something happens every + few minutes (a frame, ``fps()``, the ``fps_print`` line). + """ + + def __init__(self, now, diff): + self.now = now + self.diff = diff + self.reset() + + def reset(self): + self.mark = self.now() + self.win_us = 0 + self.win_frames = 0 + self.win_present = 0 + self.win_flushes = 0 + self.win_pixels = 0 + self.frame_us = 0 + self.frame_max = 0 + self.last_fps = None + self.seconds = 0.0 + self.frames = 0 + self.present_us = 0.0 + self.flushes = 0 + self.pixels = 0 + + def _advance(self, t): + self.win_us += self.diff(t, self.mark) + self.mark = t + if self.win_us >= 1000000: + self._fold() + + def _fold(self): + w = self.win_us + if w > 0: + self.last_fps = self.win_frames * 1000000 / w + self.seconds += w / 1000000 + self.frames += self.win_frames + self.present_us += self.win_present + self.flushes += self.win_flushes + self.pixels += self.win_pixels + self.win_us = 0 + self.win_frames = 0 + self.win_present = 0 + self.win_flushes = 0 + self.win_pixels = 0 + + def frame(self, t, dt): + """A present ended at ``t`` after ``dt`` µs; it finished a frame.""" + f = self.frame_us + dt + self.frame_us = 0 + if f > self.frame_max: + self.frame_max = f + self.win_present += dt + self.win_frames += 1 + self._advance(t) + + def flush(self, t, dt, pixels): + """A ``flush_rect`` ended at ``t`` after ``dt`` µs: part of a frame.""" + self.frame_us += dt + self.win_present += dt + self.win_flushes += 1 + self.win_pixels += pixels + self._advance(t) + + def snapshot(self): + self._advance(self.now()) + seconds = self.seconds + self.win_us / 1000000 + frames = self.frames + self.win_frames + present = self.present_us + self.win_present + if self.last_fps is not None: + fps = self.last_fps + elif self.win_us > 0: + fps = self.win_frames * 1000000 / self.win_us + else: + fps = 0.0 + return { + "fps": round(fps, 1), + "avg_fps": round(frames / seconds, 1) if seconds > 0 else 0.0, + "frames": frames, + "seconds": round(seconds, 1), + "present_ms": round(present / frames / 1000, 2) if frames else 0.0, + "present_ms_max": round(self.frame_max / 1000, 2), + "busy": round(present / (seconds * 1000000), 3) if seconds > 0 else 0.0, + "flushes": self.flushes + self.win_flushes, + "pixels": self.pixels + self.win_pixels, + } + + +def _fps_line(s): + line = "fps %.1f avg %.1f frames %d present %.1f ms (max %.1f) busy %d%%" % ( + s["fps"], + s["avg_fps"], + s["frames"], + s["present_ms"], + s["present_ms_max"], + round(s["busy"] * 100), + ) + if s["flushes"]: + line += " flushes %d" % s["flushes"] + return line + + class DisplayDriver: """ Base class for all display backends (BusDisplay, SDLDisplay, PGDisplay, FBDisplay, etc.). @@ -509,9 +659,16 @@ class DisplayDriver: buffer(s) from :meth:`framebuffers` for direct rendering (e.g. LVGL ``DISPLAY_RENDER_MODE.DIRECT``). Default False — desktop/bus drivers keep their own draw buffers. + + Frame rate: :meth:`measure_fps` (or ``PYDEVICES_FPS=1``) counts frames and + times presents; :meth:`fps` reads them. Off by default, and nothing is + wrapped or timed while it is off. """ needs_refresh = False + # The frame-rate meter while measuring (measure_fps); None means off. + _fps = None + _fps_timer = None # The display's own frame period: what ``appdev.App`` presents at and what a # GUI's refresh timer is set to. A backend that can measure its host's # refresh (vsync, requestAnimationFrame) sets it; 33 ms otherwise. @@ -579,6 +736,11 @@ def __init__(self, *, quiet=False): f"{self.__class__.__name__}: warning: slow byteswap fallback; " "install utils/byteswap (GitHub MIP) for viper/numpy swap" ) + every = env_float("PYDEVICES_FPS_PRINT", 0) + if every > 0: + self.fps_print(every) + elif env_bool("PYDEVICES_FPS"): + self.measure_fps(True) def __del__(self): self.deinit() @@ -867,6 +1029,132 @@ def bfa_area(self): """ return (0, self.tfa + self.vsa, self.width, self.bfa) + ############### Frame rate ################ + + def measure_fps(self, on=True): + """Count frames and time presents (``on=True``), or stop (``on=False``). + + Switched on, the instance's ``show`` and ``flush_rect`` are wrapped + with timers; switched off they are the class's own methods again. A + frame is one ``show()``, or one :meth:`frame_done` from a GUI that + presents with ``flush_rect`` (LVGL on a direct-framebuffer panel). + ``flush_rect`` calls are counted as ``flushes`` and ``pixels``, not + frames. ``PYDEVICES_FPS=1`` switches it on when the driver starts. + + Returns: + bool: Whether it is measuring now. + """ + if on: + if self._fps is None: + now, diff = _fps_clock() + self._fps = _FpsMeter(now, diff) + self._fps_wrapped = [] + self._fps_wrap("show") + self._fps_wrap("flush_rect") + elif self._fps is not None: + self.fps_print(0) + for name, wrapper, prev in self._fps_wrapped: + if getattr(self, name, None) is not wrapper: + continue # someone wrapped ours since: leave theirs alone + if prev is None: + delattr(self, name) + else: + setattr(self, name, prev) + self._fps_wrapped = [] + self._fps = None + return self._fps is not None + + def _fps_wrap(self, name): + orig = getattr(self, name, None) + if orig is None: + return + try: + own = name in self.__dict__ + except AttributeError: + own = False + meter = self._fps + now = meter.now + diff = meter.diff + if name == "flush_rect": + + def wrapper(x, y, w, h, *args, **kwargs): + t = now() + try: + return orig(x, y, w, h, *args, **kwargs) + finally: + t1 = now() + meter.flush(t1, diff(t1, t), w * h) + + else: + + def wrapper(*args, **kwargs): + t = now() + try: + return orig(*args, **kwargs) + finally: + t1 = now() + meter.frame(t1, diff(t1, t)) + + self._fps_wrapped.append((name, wrapper, orig if own else None)) + setattr(self, name, wrapper) + + def frame_done(self): + """A GUI that presents with ``flush_rect`` says a frame's last area is done. + + ``display_driver`` calls this on a direct-framebuffer panel when the + last area was presented by ``flush_rect`` and no ``show()`` followed. + Does nothing unless :meth:`measure_fps` is on. + """ + meter = self._fps + if meter is not None: + meter.frame(meter.now(), 0) + + def fps(self): + """The frame rate and present times, or ``None`` when not measuring. + + Returns a dict: ``fps`` (the last whole second), ``avg_fps`` (since + :meth:`measure_fps` or :meth:`fps_reset`), ``frames``, ``seconds``, + ``present_ms`` / ``present_ms_max`` (mean and worst time presenting a + frame, its ``flush_rect`` calls included), ``busy`` (the share of + wall time spent presenting: high means the display is the + bottleneck, low means the app is), ``flushes`` and ``pixels``. + """ + meter = self._fps + return None if meter is None else meter.snapshot() + + def fps_reset(self): + """Start a fresh measurement (does nothing when not measuring).""" + if self._fps is not None: + self._fps.reset() + + def fps_print(self, seconds=5): + """Print a one-line summary every ``seconds``; ``0`` stops it. + + Switches :meth:`measure_fps` on. The line is short, for agents + tailing a log:: + + fps 29.8 avg 29.5 frames 1180 present 4.2 ms (max 12.0) busy 13% + + ``PYDEVICES_FPS_PRINT=N`` starts it when the driver starts. + """ + timer = self._fps_timer + self._fps_timer = None + if timer is not None: + timer.deinit() + if not seconds or seconds <= 0: + return + self.measure_fps(True) + import multimer + + def _print(_timer=None): + meter = self._fps + if meter is not None: + print(_fps_line(meter.snapshot())) + + self._fps_timer = multimer.every( + max(1, int(seconds * 1000)), _print, name="fps:%s" % (self.__class__.__name__,) + ) + ############### Common API Methods, sometimes overridden ################ def vscrdef(self, tfa: int, vsa: int, bfa: int) -> None: @@ -992,6 +1280,8 @@ def deinit(self) -> None: if getattr(self, "_deinitialized", False): return self._deinitialized = True + if self._fps_timer is not None: + self.fps_print(0) # the counts stay readable after quit self._deinit() def _deinit(self) -> None: diff --git a/tests/README.md b/tests/README.md index 151e388f..37792872 100644 --- a/tests/README.md +++ b/tests/README.md @@ -25,6 +25,7 @@ Do not run audiodev tests while another process is playing audio — see | `test_autodisplay.py` | `displaydev.auto` host selection | | `test_displaydev_auto.py` | backends must not import `displaydev.auto` | | `test_displaydev_lifecycle.py` / `test_displaydev_capabilities.py` | quit lifecycle and `capabilities()` | +| `test_displaydev_fps.py` | frame-rate meter: nothing wrapped while off, frames/present/busy counted while on, `flush_rect` as flushes not frames, the appdev refresh calling a wrapper installed late | | `test_env.py` / `test_color.py` / `test_byteswap.py` / `test_desktop_scale.py` | displaydev helpers | | `test_needs_refresh.py` / `test_backend_isolation.py` | backend flags and imports | | `test_jndisplay_scroll.py` | JNDisplay scroll (needs IPython + Pillow) | diff --git a/tests/test_displaydev_fps.py b/tests/test_displaydev_fps.py new file mode 100644 index 00000000..be3fa5a4 --- /dev/null +++ b/tests/test_displaydev_fps.py @@ -0,0 +1,330 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""DisplayDriver frame-rate measurement: measure_fps, fps, fps_reset, fps_print.""" + +import io +import os +import time +import unittest +from contextlib import redirect_stdout +from unittest import mock + +import _env # noqa: F401 +from _support import quiet + +import displaydev +import multimer +from appdev import App +from displaydev import DisplayDriver + + +class _Clock: + """A virtual microsecond clock: ``work(us)`` is time spent inside a call.""" + + def __init__(self): + self.t = 0 + + def now(self): + return self.t + + @staticmethod + def diff(a, b): + return a - b + + +class Panel(DisplayDriver): + def __init__(self, clock=None, show_us=0, flush_us=0): + self._width = 8 + self._height = 4 + self._rotation = 0 + self._requires_byteswap = False + self.clock = clock + self.show_us = show_us + self.flush_us = flush_us + self.shows = 0 + self.flushed = [] + super().__init__(quiet=True) + + def init(self): + pass + + def fill_rect(self, x, y, w, h, c): + return (x, y, w, h) + + def blit_rect(self, buf, x, y, w, h): + return (x, y, w, h) + + def show(self, _timer=None): + self.shows += 1 + if self.clock is not None: + self.clock.t += self.show_us + + def flush_rect(self, x, y, w, h): + self.flushed.append((x, y, w, h)) + if self.clock is not None: + self.clock.t += self.flush_us + return True + + +class _FakeClockMixin: + def setUp(self): + self.clock = _Clock() + self._saved = displaydev._fps_clock_cache + displaydev._fps_clock_cache = (self.clock.now, self.clock.diff) + + def tearDown(self): + displaydev._fps_clock_cache = self._saved + + +class TestSwitchedOff(unittest.TestCase): + def test_nothing_is_wrapped(self): + d = Panel() + self.assertNotIn("show", d.__dict__) + self.assertNotIn("flush_rect", d.__dict__) + self.assertIs(d.show.__func__, Panel.show) + self.assertIs(d.flush_rect.__func__, Panel.flush_rect) + self.assertIsNone(d.fps()) + d.frame_done() # a no-op + d.fps_reset() # a no-op + self.assertIsNone(d._fps) + + def test_switching_off_restores_the_class_methods(self): + d = Panel() + self.assertTrue(d.measure_fps(True)) + self.assertIn("show", d.__dict__) + self.assertFalse(d.measure_fps(False)) + self.assertNotIn("show", d.__dict__) + self.assertNotIn("flush_rect", d.__dict__) + self.assertIs(d.show.__func__, Panel.show) + self.assertIsNone(d.fps()) + + def test_an_instance_show_of_its_own_comes_back(self): + d = Panel() + own = lambda *a: None # noqa: E731 + d.show = own + d.measure_fps(True) + self.assertIsNot(d.show, own) + d.measure_fps(False) + self.assertIs(d.show, own) + + def test_a_wrapper_put_over_ours_is_left_alone(self): + d = Panel() + d.measure_fps(True) + ours = d.show + theirs = lambda *a: ours(*a) # noqa: E731 + d.show = theirs + d.measure_fps(False) + self.assertIs(d.show, theirs) + + def test_switching_on_twice_wraps_once(self): + d = Panel() + d.measure_fps(True) + first = d.show + d.measure_fps(True) + self.assertIs(d.show, first) + d.show() + self.assertEqual(1, d.fps()["frames"]) + + +class TestCounting(_FakeClockMixin, unittest.TestCase): + def test_frames_present_and_busy(self): + d = Panel(self.clock, show_us=4000) + d.measure_fps(True) + for _ in range(90): # 30 fps for three seconds: 4 ms showing, 29.3 ms app + d.show() + self.clock.t += 29333 + s = d.fps() + self.assertEqual(90, s["frames"]) + self.assertEqual(90, d.shows) + self.assertAlmostEqual(4.0, s["present_ms"]) + self.assertAlmostEqual(4.0, s["present_ms_max"]) + self.assertAlmostEqual(30.0, s["fps"], delta=0.6) + self.assertAlmostEqual(30.0, s["avg_fps"], delta=0.6) + self.assertAlmostEqual(0.12, s["busy"], delta=0.01) + self.assertEqual(0, s["flushes"]) + self.assertEqual(0, s["pixels"]) + + def test_worst_present_is_kept(self): + d = Panel(self.clock, show_us=1000) + d.measure_fps(True) + d.show() + d.show_us = 12000 + d.show() + d.show_us = 2000 + d.show() + s = d.fps() + self.assertAlmostEqual(12.0, s["present_ms_max"]) + self.assertAlmostEqual(5.0, s["present_ms"]) + + def test_fps_is_the_last_whole_second(self): + d = Panel(self.clock) + d.measure_fps(True) + for _ in range(60): # a fast first second: 60 fps + self.clock.t += 16667 + d.show() + for _ in range(10): # then a slow one: 10 fps + self.clock.t += 100000 + d.show() + s = d.fps() + self.assertAlmostEqual(10.0, s["fps"], delta=0.5) + self.assertAlmostEqual(35.0, s["avg_fps"], delta=0.5) + + def test_a_stalled_display_reads_zero(self): + d = Panel(self.clock) + d.measure_fps(True) + for _ in range(40): + self.clock.t += 25000 + d.show() + self.clock.t += 3000000 # three seconds with nothing presented + self.assertEqual(0.0, d.fps()["fps"]) + + def test_flush_rect_counts_flushes_and_pixels_not_frames(self): + d = Panel(self.clock, flush_us=500) + d.measure_fps(True) + for _ in range(3): # three frames, each two dirty areas + self.assertTrue(d.flush_rect(0, 0, 8, 2)) # the return value passes through + d.flush_rect(0, 2, 4, 2) + d.frame_done() + self.clock.t += 10000 + s = d.fps() + self.assertEqual(3, s["frames"]) + self.assertEqual(6, s["flushes"]) + self.assertEqual(3 * (16 + 8), s["pixels"]) + self.assertAlmostEqual(1.0, s["present_ms"]) # both flushes of a frame + self.assertAlmostEqual(1.0, s["present_ms_max"]) + self.assertEqual(6, len(d.flushed)) + + def test_flushes_alone_are_not_frames(self): + d = Panel(self.clock, flush_us=100) + d.measure_fps(True) + for _ in range(5): + d.flush_rect(0, 0, 2, 2) + self.clock.t += 1000 + s = d.fps() + self.assertEqual(0, s["frames"]) + self.assertEqual(5, s["flushes"]) + self.assertGreater(s["busy"], 0) + + def test_fps_reset_starts_fresh(self): + d = Panel(self.clock, show_us=9000) + d.measure_fps(True) + for _ in range(5): + d.show() + self.clock.t += 1000 + d.fps_reset() + d.show_us = 1000 + d.show() + self.clock.t += 199000 + s = d.fps() + self.assertEqual(1, s["frames"]) + self.assertAlmostEqual(1.0, s["present_ms_max"]) + self.assertAlmostEqual(0.2, s["seconds"]) + + def test_a_show_that_raises_is_still_timed(self): + d = Panel(self.clock) + + def broken(_timer=None): + raise RuntimeError("panel gone") + + d.show = broken + d.measure_fps(True) + with self.assertRaises(RuntimeError): + d.show() + self.assertEqual(1, d.fps()["frames"]) + + +class TestEnvironmentAndPrint(unittest.TestCase): + def tearDown(self): + multimer.stop_all() + + def _clean_env(self, **env): + for k in ("PYDEVICES_FPS", "PYDEVICES_FPS_PRINT"): + displaydev._overrides.pop(k, None) + patched = {k: v for k, v in os.environ.items() if not k.startswith("PYDEVICES_FPS")} + patched.update(env) + return mock.patch.dict(os.environ, patched, clear=True) + + def test_pydevices_fps_switches_it_on_at_start(self): + with self._clean_env(PYDEVICES_FPS="1"): + d = Panel() + self.assertIn("show", d.__dict__) + self.assertIsNotNone(d.fps()) + self.assertIsNone(d._fps_timer) + + def test_off_unless_asked(self): + with self._clean_env(): + d = Panel() + self.assertNotIn("show", d.__dict__) + + def test_pydevices_fps_print_starts_the_line(self): + with self._clean_env(PYDEVICES_FPS_PRINT="0.05"): + d = Panel() + self.assertIsNotNone(d.fps()) + self.assertIsNotNone(d._fps_timer) + self.assertEqual(50, d._fps_timer.period) + out = io.StringIO() + with redirect_stdout(out): + deadline = time.monotonic() + 2 + while "fps " not in out.getvalue() and time.monotonic() < deadline: + d.show() + multimer.sleep_ms(5) + self.assertRegex( + out.getvalue(), + r"fps [\d.]+ avg [\d.]+ frames \d+ present [\d.]+ ms \(max [\d.]+\) busy \d+%", + ) + d.deinit() # stops the line; the counts stay readable + self.assertIsNone(d._fps_timer) + self.assertIsNotNone(d.fps()) + + def test_fps_print_zero_stops_it(self): + d = Panel() + d.fps_print(1) + self.assertTrue(d._fps_timer.running) + timer = d._fps_timer + d.fps_print(0) + self.assertIsNone(d._fps_timer) + self.assertFalse(timer.running) + self.assertIsNotNone(d.fps()) # still measuring + + def test_the_line(self): + line = displaydev._fps_line( + { + "fps": 29.8, + "avg_fps": 29.5, + "frames": 1180, + "seconds": 40.0, + "present_ms": 4.2, + "present_ms_max": 12.0, + "busy": 0.13, + "flushes": 0, + "pixels": 0, + } + ) + self.assertEqual("fps 29.8 avg 29.5 frames 1180 present 4.2 ms (max 12.0) busy 13%", line) + + +class TestAppRefreshSeesTheWrapper(unittest.TestCase): + def tearDown(self): + app = App.current() + if app is not None: + app._perform_teardown() + multimer.stop_all() + multimer.keepalive(False) + + def test_refresh_timer_calls_a_wrapper_installed_after_the_app_started(self): + with quiet(): + d = Panel() + d.needs_refresh = True + d.refresh_period_ms = 10 + App(displays=[d]) + d.measure_fps(True) # after the App wired its refresh timer + deadline = time.monotonic() + 2 + while d.fps()["frames"] < 3 and time.monotonic() < deadline: + multimer.sleep_ms(5) + self.assertGreaterEqual(d.fps()["frames"], 3) + self.assertEqual(d.fps()["frames"], d.shows) + + +if __name__ == "__main__": + unittest.main() From 87088efa5c75284a393d6aba4d549e3e83301f46 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Wed, 7 Oct 2026 14:58:38 -0500 Subject: [PATCH 2/3] display_driver moves here from lvgl-bindings, with the direct-panel frame hook display_driver is PyDevices plumbing: it imports appdev, events, keys and multimer and builds its App from board_config, so it is worthless without them, while the bindings are useful to anyone with plumbing of their own. Here it ships in the pydevices mip package, the pydevices wheel and the frozen module with the things it needs, and a change to it no longer needs an LVGL release. The file is lvgl-bindings#24's: a direct-framebuffer flush calls the panel's frame_done() on a frame's last area and presents counts it, so measure_fps counts frames, not flushes. lvgl-bindings#24 and lvgl-micropython#23 close in favour of this. pydevices-desktop.toml is regenerated with the validate workflow's mappings. --- docs/board-bringup.md | 21 +- docs/timing-hardware-tests.md | 2 +- lib/display_driver.py | 1701 +++++++++++++++++++++++++++++++++ pydevices-desktop.toml | 1 + 4 files changed, 1715 insertions(+), 10 deletions(-) create mode 100644 lib/display_driver.py diff --git a/docs/board-bringup.md b/docs/board-bringup.md index 1ad10066..bc6df230 100644 --- a/docs/board-bringup.md +++ b/docs/board-bringup.md @@ -23,7 +23,7 @@ modules (micropython-pydevices' `build_mp.py --modules displayif,lvgl-micropython,usbif,pygraphics,audiodsp,ulab`, say) the following are **frozen or built in** — installing them is wasted effort: -- `lvgl` and `display_driver` (from lvgl-bindings) +- `lvgl` (from lvgl-micropython) - `dotclockframebuffer` (from displayif) — the RGB panel interface - `_usbif` and the `usbif` Python package: usbif's manifest freezes its Python half beside its C half, because the two version together @@ -32,7 +32,8 @@ are **frozen or built in** — installing them is wasted effort: and the following are **not**, and must be installed: - `displaydev`, `appdev`, `multimer`, `audiodev`, `boarddev`, `events`, `keys` - (pydevices). An image built with `--modules all` freezes these too, and + and `display_driver`, LVGL's coordinator (pydevices). An image built with + `--modules all` freezes these too, and `.frozen` comes before `/lib` on `sys.path`, so there a mip-installed copy never runs: rebuild to update them. - `board_config` and `board_peripherals` for the specific board @@ -50,13 +51,15 @@ for m in ("lvgl", "display_driver", "dotclockframebuffer", "mipidsi", (`mipidsi` is the panel interface on the ESP32-P4 boards, `dotclockframebuffer` on the RGB-panel S3 boards; whichever is not yours will simply be missing.) -**Trap.** `import display_driver` fails with `ImportError` on a bare board even -though it is frozen — because `display_driver` imports `appdev`, and it is -*that* import failing (`no module named 'appdev'`). The same shape hides -behind `usbif`, which trips on `events`. The probe prints the exception text -so you can see which module is really absent; a frozen module that reports -`MISS` with somebody else's name in the message is present and fine. Install -`pydevices` (which carries `appdev` and `events`), then re-probe. +**Trap.** `import usbif` fails with `ImportError` on a bare board even +though it is frozen — because `usbif` imports `events`, and it is *that* +import failing (`no module named 'events'`). The probe prints the exception +text so you can see which module is really absent; a frozen module that +reports `MISS` with somebody else's name in the message is present and fine. +Install `pydevices` (which carries `events`, `appdev` and `display_driver`), +then re-probe. Firmware built before display_driver moved to pydevices +(2026-10) froze it from lvgl-micropython, where it showed the same trap with +`appdev`. ## 2. Installing, over Wi-Fi diff --git a/docs/timing-hardware-tests.md b/docs/timing-hardware-tests.md index 6b059508..ce9b999e 100644 --- a/docs/timing-hardware-tests.md +++ b/docs/timing-hardware-tests.md @@ -96,7 +96,7 @@ re-armed from its own (scheduled) callback. Nothing in the interpreter changes. 1. Flash the kitchen-sink image; `mpftp put` the changed `pydevices/lib` - (multimer, appdev, displaydev) and `lvgl-bindings/python/display_driver.py` + (multimer, appdev, displaydev, display_driver.py) to `/lib`, which beats the frozen copies. 2. **Function check, no display:** `mpftp exec` the body of `demo_timers.py` (or put it as `/demo_timers.py` and `import demo_timers`). diff --git a/lib/display_driver.py b/lib/display_driver.py new file mode 100644 index 00000000..6f70cdd1 --- /dev/null +++ b/lib/display_driver.py @@ -0,0 +1,1701 @@ +# SPDX-FileCopyrightText: 2024 Brad Barnett +# SPDX-FileCopyrightText: 2021 Amir Gonnen (event_loop; MIT) +# +# SPDX-License-Identifier: MIT + +""" +display_driver.py - LVGL displaydev/input wiring and event loop for PyDevices. + +It lives in PyDevices/pydevices (``lib/display_driver.py``) because it is +PyDevices plumbing: it needs ``appdev``, ``displaydev``, ``events``, ``keys``, +``multimer`` and the ``board_config`` idiom, and ships with them in the +``pydevices`` mip package, the ``pydevices`` wheel and the frozen +``pydevices`` module. The LVGL bindings (lvgl-bindings and the lvgl-micropython, +lvgl-circuitpython and lvgl-python consumers) carry only ``lvgl`` itself and +``fs_driver``, so anyone can use them with plumbing of their own. + +Importing this module uses the active :class:`appdev.App` when the application +has already constructed one explicitly. For compatibility, it falls back to +creating an app from ``board_config`` when no active app exists. The selected +coordinator registers display flush and input devices; the LVGL loop is a +``multimer`` timer. + +``event_loop`` was adapted from upstream lv_utils (Amir Gonnen). Integration +changes: + +* One ``multimer`` ONE_SHOT timer runs ``lv.timer_handler()`` and re-arms + itself to what LVGL returns (the ms until LVGL's next timer is due), so an + idle screen wakes when LVGL wants it, not on a 10 ms poll. After a pass + longer than its slot the loop yields ``min(pass, max_yield_ms)`` before the + next one (lvgl-bindings#15 and #19). +* ``lv.tick_inc`` is fed from ``multimer.ticks_ms`` before each pass. +* PARTIAL panels present from the display's own frame clock + (``display.frame_clock``), and only when a flush happened since the last + present; DIRECT / shared-framebuffer panels present from ``flush_is_last``. +* Application lifecycle is ``appdev.App`` on ``multimer``: the app keeps + itself alive past the end of the script body, so a trailing ``app.run()`` + is optional, and callbacks run on the main thread at safe points on every + host, so there is no separate async mode. +""" + +import gc +import sys + +import lvgl as lv + +# The binding-internal callback re-entrancy counter. MicroPython and +# CircuitPython export it (an audited exception to the canonical model) and +# the task-handler gates on it. CPython deliberately does not export it +# - re-entrancy is guarded in C with a ContextVar-scoped counter - so there +# the Python-side gate simply stands down. +_LV_NESTING = getattr(lv, "_nesting", None) + +import appdev +import events +import keys +import multimer +from multimer import ticks_diff, ticks_ms + +app = appdev.App.current() +if app is None: + import board_config + + app = appdev.App(board_config) + +display_drv = app.primary +if display_drv is None: + raise RuntimeError("display_driver requires an appdev.App with a display") + +# The most the loop waits before asking LVGL again, whatever it returned: a +# bound on idle wake-ups, and the input read cadence LVGL is set to. +LVGL_PERIOD_MS = 10 +# Fallback present cadence when a display has no frame period of its own. +LVGL_REFR_PERIOD_MS = 33 +_driver_ref = None # primary DisplayDriver (compat) +_drivers = [] # all DisplayDriver instances +_host_pump_timer = None +# LVGL owns presentation and input while it runs: the App's own refresh timer +# and service tick stand down (app.pause_refresh, app.pause_polling), so a +# PARTIAL panel is presented once per frame from the display's frame clock, +# and the host events reach LVGL's indevs instead of being consumed first. +_refresh_claim = None +_polling_claim = None + +HOST = appdev.HOST +POINTER = appdev.POINTER +ENCODER = appdev.ENCODER +KEYPAD = appdev.KEYPAD +JOYSTICK = appdev.JOYSTICK + + +class InputDevice: + """Small input adapter used only by the LVGL bridge.""" + + type = -1 + responses = events.filter + + def __init__(self, read=None, data=None, read2=None, data2=None): + self._read = read if read is not None else lambda: None + self._data = data + self._read2 = read2 if read2 is not None else lambda: None + self._data2 = data2 + self._state = None + self._app = None + self._user_data = None + self._callbacks = [] + + @property + def app(self): + return self._app + + @app.setter + def app(self, value): + self._app = value + + @property + def user_data(self): + return self._user_data + + @user_data.setter + def user_data(self, value): + self._user_data = value + + def subscribe(self, callback, event_types=None): + if not callable(callback): + raise ValueError("callback is not callable") + item = (callback, event_types) + if item not in self._callbacks: + self._callbacks.append(item) + + def unsubscribe(self, callback, event_types=None): + self._callbacks = [item for item in self._callbacks if item[0] is not callback] + + def poll(self, *args): + raw = self._poll() + if raw is None: + return [] + result = raw if isinstance(raw, list) else [raw] + result = [event for event in result if event.type in events.filter] + for event in result: + for callback, event_types in tuple(self._callbacks): + if event_types is None or event.type in event_types: + callback(event, *args) + return result + + +class HostInput(InputDevice): + """Adapt a host display's ``get_events`` callback for LVGL.""" + + type = HOST + + def __init__(self, host_read, display=None, event_filter=None): + super().__init__(read=host_read, data=display, data2=event_filter or events.filter) + self.scale = getattr(display, "touch_scale", 1) if display is not None else 1 + self._quit_chord_ok = hasattr(display, "quit_chord") + + def _touch_scale_for(self, window_id): + panel = self._data + if window_id is not None and self._app is not None: + for candidate in self._app.displays: + if getattr(candidate, "_window_id", None) == window_id: + panel = candidate + break + scale = getattr(panel, "touch_scale", None) if panel is not None else None + if scale is None: + return self.scale + self.scale = scale + return scale + + def _poll(self): + incoming = self._read() + if incoming is None: + return None + result = [] + quit_chord = self._data.quit_chord if self._quit_chord_ok else None + chord_key = quit_chord[0] if quit_chord else None + for event in incoming: + if event.type == events.KEYDOWN and keys.chord_matches( + quit_chord, event.key, event.mod + ): + event = events.Quit(events.QUIT) + elif event.type == events.KEYUP and quit_chord and event.key == chord_key: + continue + if event.type not in self._data2: + continue + if event.type in ( + events.MOUSEMOTION, + events.MOUSEBUTTONDOWN, + events.MOUSEBUTTONUP, + ): + scale = self._touch_scale_for(getattr(event, "window", None)) + if scale and scale != 1: + pos = (int(event.pos[0] // scale), int(event.pos[1] // scale)) + if event.type == events.MOUSEMOTION: + event = events.Motion( + event.type, + pos, + (event.rel[0] // scale, event.rel[1] // scale), + event.buttons, + event.touch, + event.window, + ) + else: + event = events.Button( + event.type, pos, event.button, event.touch, event.window + ) + result.append(event) + return result or None + + +_DEFAULT_TOUCH_ROTATION_TABLE = (0b000, 0b101, 0b110, 0b011) +_SWAP_XY = 0b001 +_REVERSE_X = 0b010 +_REVERSE_Y = 0b100 + + +def _normalize_points(sample): + if not sample: + return () + if isinstance(sample[0], int): + return (tuple(sample),) + return tuple(tuple(point) for point in sample) + + +class TouchInput(InputDevice): + """Adapt a board touch callable to pointer events for LVGL.""" + + type = POINTER + responses = (events.MOUSEMOTION, events.MOUSEBUTTONDOWN, events.MOUSEBUTTONUP) + + def __init__(self, read, display, rotation_table=None): + super().__init__(read=read, data=display, data2=rotation_table) + self._data2 = self._data2 or _DEFAULT_TOUCH_ROTATION_TABLE + self.rotation = display.rotation + try: + display.touch_device = self + except Exception: + pass + self.points = () + + @property + def rotation(self): + return self._rotation + + @rotation.setter + def rotation(self, value): + self._rotation = value % 360 + self._mask = self._data2[self._rotation // 90] + + def _map_point(self, point): + x, y = int(point[0]), int(point[1]) + if self._mask & _SWAP_XY: + x, y = y, x + if self._mask & _REVERSE_X: + x = self._data.width - x - 1 + if self._mask & _REVERSE_Y: + y = self._data.height - y - 1 + return (x, y) + tuple(point[2:]) if len(point) > 2 else (x, y) + + def _poll(self): + try: + mapped = tuple(self._map_point(point) for point in _normalize_points(self._read())) + except OSError: + return None + self.points = mapped + if mapped: + x, y = int(mapped[0][0]), int(mapped[0][1]) + previous = self._state + self._state = (x, y) + if previous is None: + return events.Button(events.MOUSEBUTTONDOWN, self._state, 1, False, None) + return events.Motion( + events.MOUSEMOTION, + self._state, + (x - previous[0], y - previous[1]), + (1, 0, 0), + False, + None, + ) + if self._state is not None: + previous = self._state + self._state = None + return events.Button(events.MOUSEBUTTONUP, previous, 1, False, None) + return None + + +class KeypadInput(InputDevice): + """Adapt a pressed-key collection to KEYDOWN/KEYUP events.""" + + type = KEYPAD + responses = (events.KEYDOWN, events.KEYUP) + + def __init__(self, read): + super().__init__(read=read) + self._state = set() + + @staticmethod + def _name(key): + name = keys.keyname(key) + if name != "Unknown": + return name + if isinstance(key, int) and 32 <= key <= 126: + return chr(key) + return "0x%x" % key if isinstance(key, int) else str(key) + + def _poll(self): + current = set(self._read()) + released = self._state - current + if released: + key = released.pop() + self._state.remove(key) + return events.Key(events.KEYUP, self._name(key), key, 0, 0, None) + pressed = current - self._state + if pressed: + key = pressed.pop() + self._state.add(key) + return events.Key(events.KEYDOWN, self._name(key), key, 0, 0, None) + return None + + +class EncoderInput(InputDevice): + """Adapt an encoder position and optional button to LVGL events.""" + + type = ENCODER + responses = (events.MOUSEWHEEL, events.MOUSEBUTTONDOWN, events.MOUSEBUTTONUP) + + def __init__(self, read, button_read=None, button=2): + super().__init__(read=read, read2=button_read, data=button) + self._state = (0, False) + + def _poll(self): + last_pos, last_pressed = self._state + pressed = self._read2() + if pressed != last_pressed: + self._state = (last_pos, pressed) + return events.Button( + events.MOUSEBUTTONDOWN if pressed else events.MOUSEBUTTONUP, + (0, 0), + self._data, + False, + None, + ) + pos = self._read() + if pos != last_pos: + steps = pos - last_pos + self._state = (pos, last_pressed) + if self._data % 2 == 0: + return events.Wheel(events.MOUSEWHEEL, False, 0, steps, 0, steps, False, None) + return events.Wheel(events.MOUSEWHEEL, False, steps, 0, steps, 0, False, None) + return None + + +_virtual_peers = {} +_virtual_pending = {} + + +class VirtualDevices: + """Fan one host input into LVGL pointer, encoder, and keypad inputs.""" + + class VirtualDevice: + def __init__(self, owner, device_type): + self._owner = owner + self.type = device_type + self.user_data = None + self._fifo = [] + self._callbacks = [] + self._active_key_event = None + self.points = () + self._fingers = {} + + def subscribe(self, callback, event_types=None): + if callback not in self._callbacks: + self._callbacks.append(callback) + + def unsubscribe(self, callback, event_types=None): + if callback in self._callbacks: + self._callbacks.remove(callback) + + @property + def has_pending(self): + return bool(self._fifo) + + def poll(self, *args): + self._owner.poll_host_device() + event = self._fifo.pop(0) if self._fifo else None + for callback in tuple(self._callbacks): + callback(event, *args) + + def add_event(self, event): + if ( + event.type == events.MOUSEWHEEL + and self._fifo + and self._fifo[-1].type == events.MOUSEWHEEL + ): + # Sum, don't replace: unlike a motion sample (only the latest + # position matters), every wheel delta has to count toward + # the total scroll distance. Some backends (WinDisplay, + # confirmed) emit far more wheel messages per physical + # gesture than others (SDL) - without this, a fast scroll + # queues faster than the "keep calling me until empty" + # read-cycle mechanism can drain one at a time, and the + # window stops responding for as long as scrolling continues. + last = self._fifo[-1] + self._fifo[-1] = events.Wheel( + event.type, + event.flipped, + last.x + event.x, + last.y + event.y, + last.precise_x + event.precise_x, + last.precise_y + event.precise_y, + event.touch, + event.window, + ) + return + if ( + event.type == events.MOUSEMOTION + and self._fifo + and self._fifo[-1].type == events.MOUSEMOTION + ): + self._fifo[-1] = event + return + if ( + event.type == events.KEYDOWN + and self._fifo + and self._fifo[-1].type == events.KEYDOWN + and getattr(self._fifo[-1], "key", None) == getattr(event, "key", None) + ): + self._fifo[-1] = event + return + if self.type == KEYPAD: + key = getattr(event, "key", None) + active = self._active_key_event + active_key = getattr(active, "key", None) + if event.type == events.KEYDOWN: + if active is not None and active_key != key: + self._fifo.append( + events.Key( + events.KEYUP, + active.name, + active.key, + active.mod, + active.scancode, + active.window, + ) + ) + self._active_key_event = event + elif event.type == events.KEYUP: + if active is not None and active_key != key: + return + self._active_key_event = None + self._fifo.append(event) + + def _set_finger(self, finger_id, point): + if point is None: + self._fingers.pop(finger_id, None) + else: + self._fingers[finger_id] = point + self.points = tuple( + (pos[0], pos[1], fid) for fid, pos in self._fingers.items() + ) + + def __init__(self, host_device, window_id=None): + self._host_device = host_device + self._window_id = window_id + # For QUIT below: the app that owns this HOST device (App.add_device + # sets dev.app = self on every device it registers), so a window + # close reaches the same app.request_quit() a script calls by hand. + self._app = getattr(host_device, "app", None) + self._vd_pointer = self.VirtualDevice(self, POINTER) + self._vd_encoder = self.VirtualDevice(self, ENCODER) + self._vd_keypad = self.VirtualDevice(self, KEYPAD) + self.devices = [self._vd_pointer, self._vd_encoder, self._vd_keypad] + peers = _virtual_peers.setdefault(id(host_device), []) + peers.append(self) + self._peers = peers + + def _accepts_window(self, event): + if self._window_id is None: + return True + window = getattr(event, "window", None) + return window is None or window == self._window_id + + def poll_host_device(self): + if self._peers and self._peers[0] is not self: + return + pending = _virtual_pending.setdefault(id(self._host_device), []) + if not pending: + batch = self._host_device.poll() + if batch: + pending.extend(batch) + while pending: + event = pending.pop(0) + for peer in self._peers: + peer._route(event) + if event.type in (events.FINGERDOWN, events.FINGERUP, events.FINGERMOTION): + return + + def _route(self, event): + if not self._accepts_window(event): + return + if event.type in (events.FINGERDOWN, events.FINGERMOTION): + pointer = self._vd_pointer + pointer._set_finger(event.finger_id, event.pos) + if pointer._fingers: + primary_id = min(pointer._fingers) + x, y = pointer._fingers[primary_id] + if event.finger_id == primary_id: + if event.type == events.FINGERDOWN: + pointer.add_event( + events.Button( + events.MOUSEBUTTONDOWN, (x, y), 1, True, event.window + ) + ) + else: + pointer.add_event( + events.Motion( + events.MOUSEMOTION, + (x, y), + (0, 0), + (1, 0, 0), + True, + event.window, + ) + ) + elif event.type == events.FINGERUP: + pointer = self._vd_pointer + was_primary = pointer._fingers and event.finger_id == min(pointer._fingers) + last = pointer._fingers.get(event.finger_id, event.pos) + pointer._set_finger(event.finger_id, None) + if was_primary: + pointer.add_event( + events.Button(events.MOUSEBUTTONUP, last, 1, True, event.window) + ) + elif event.type in (events.MOUSEBUTTONDOWN, events.MOUSEBUTTONUP) or ( + event.type == events.MOUSEMOTION and event.buttons[0] + ): + if not (getattr(event, "touch", False) and self._vd_pointer._fingers): + self._vd_pointer.add_event(event) + elif event.type == events.MOUSEWHEEL: + self._vd_encoder.add_event(event) + _wheel_navigate_cb(event) + elif event.type in (events.KEYDOWN, events.KEYUP): + self._vd_keypad.add_event(event) + elif event.type == events.QUIT: + # Window-close reaches here (e.g. windisplay.WinDisplay's + # WM_CLOSE handler queues one) but nothing previously acted on + # it: every LVGL indev is polled straight from LVGL's own read + # timer, not from App.poll()'s loop, so App's own QUIT handling + # never saw it — the event was silently dropped and the window + # could not be closed at all. + if self._app is not None: + self._app.request_quit() + + + +class event_loop: + """The LVGL loop: one ``multimer`` timer that runs ``lv.timer_handler``. + + One instance may be active at a time. Prefer ``import display_driver`` + (module ``main()``) over constructing this by hand unless you need custom + ``period_ms`` / ``max_yield_ms`` settings. + """ + + _current_instance = None + + # The most the loop holds LVGL back after a slow pass (see _next_delay). + max_yield_ms = 100 + + def __init__( + self, + freq=None, + max_scheduled=2, + refresh_cb=None, + asynchronous=False, + exception_sink=None, + period_ms=None, + max_yield_ms=None, + ): + """Create and register the LVGL loop. + + Args: + freq: Desired Hz when ``period_ms`` is omitted (period = ``1000 // freq``). + max_scheduled: Kept for lv_utils API parity (unused). + refresh_cb: Optional zero-arg callable after each ``lv.timer_handler()``. + asynchronous: Kept for API parity and ignored: ``multimer`` + delivers on the host's own loop where the host owns one. + exception_sink: Callable receiving exceptions from task handling; + defaults to :meth:`default_exception_sink`. + period_ms: The most the loop waits between passes, in ms + (default ``LVGL_PERIOD_MS``). LVGL asks for less when a timer + is due sooner. + max_yield_ms: Most the loop stays off after a slow pass, leaving + the thread to the application (default 100; see :meth:`_next_delay`). + + Raises: + RuntimeError: Another loop is already running. + """ + if self.is_running(): + raise RuntimeError("Event loop is already running!") + + if not lv.is_initialized(): + lv.init() + + event_loop._current_instance = self + + if period_ms is not None: + self.delay = max(1, int(period_ms)) + elif freq is not None: + self.delay = max(1, 1000 // int(freq)) + else: + self.delay = LVGL_PERIOD_MS + if max_yield_ms is not None: + self.max_yield_ms = int(max_yield_ms) + + self.refresh_cb = refresh_cb + self.exception_sink = exception_sink if exception_sink else self.default_exception_sink + # Start paused; ``enable()`` arms the timer. Nothing may run LVGL + # before the draw buffers exist (on an ESP32-P4 even a no-op callback + # interrupting SPIRAM ``draw_buf_create`` corrupts LVGL handlers). + self._pause = 1 + self._in_task = False + self._last_tick_ms = None + self._timer = multimer.Timer(-1) + self._timer.name = "lvgl" + self.passes = 0 + self.slow_passes = 0 + self.last_pass_ms = 0 + + @property + def timer(self): + """The ``multimer.Timer`` driving LVGL (for ``multimer.report()``).""" + return self._timer + + def _arm(self): + if self._timer.running: + return + self._timer.init(mode=multimer.Timer.ONE_SHOT, period=self.delay, callback=self._on_timer) + + def deinit(self): + """Stop the timer and clear the singleton.""" + self._timer.deinit() + event_loop._current_instance = None + + def disable(self): + """Pause LVGL task handling (re-entrant; pair with :meth:`enable`).""" + self._pause += 1 + + def enable(self): + """Resume LVGL task handling after :meth:`disable`; arms the timer.""" + if self._pause > 0: + self._pause -= 1 + if self._pause == 0: + self._arm() + + @staticmethod + def is_running(): + """True when an :class:`event_loop` instance is currently registered.""" + return event_loop._current_instance is not None + + @staticmethod + def current_instance(): + """Return the active :class:`event_loop`, or ``None``.""" + return event_loop._current_instance + + def task_handler(self, _=None): + """Run ``lv.timer_handler()`` once when not paused and not nested. + + Returns the ms until LVGL's next timer is due, or None when the pass + did not run. + """ + if self._in_task or self._pause > 0: + return None + self._in_task = True + try: + if _LV_NESTING is None or _LV_NESTING.value == 0: + nxt = lv.timer_handler() + if self.refresh_cb: + self.refresh_cb() + return nxt + except Exception as e: + if self.exception_sink: + self.exception_sink(e) + finally: + self._in_task = False + return None + + def tick(self): + """Run one pass by hand (same path as the timer).""" + self._on_timer(self._timer) + + def _advance_lvgl_clock(self): + now = ticks_ms() + if self._last_tick_ms is None: + self._last_tick_ms = now + elapsed = ticks_diff(now, self._last_tick_ms) + if elapsed > 0: + lv.tick_inc(elapsed) + self._last_tick_ms = now + return now + + def _next_delay(self, wanted, work_ms): + """When to run again: what LVGL asked for, bounded, and the yield rule. + + ``wanted`` is what ``lv.timer_handler`` returned (ms until its next + timer), capped at ``self.delay`` so input is read at least that + often. A pass that took longer than that -- LVGL asking to run + "again now" after any real work counts -- is followed by at least a + period off, and by ``min(work, max_yield_ms)`` when the pass was + longer than a period: on MicroPython the pass runs between the + application's own bytecodes, and resuming the instant it ended took + ~87 % of the thread on an ESP32-P4 (lvgl-bindings#15); an uncapped + hold doubled every UI stall on an ESP32-S3 (lvgl-bindings#19). + """ + if wanted is None or wanted < 0: + wanted = self.delay + delay = min(int(wanted), self.delay) + if work_ms > delay: + # The pass outran what LVGL asked for (including "again now"): + # at least a period off, longer for a slow pass, capped. + self.slow_passes += 1 + delay = max(self.delay, min(work_ms, self.max_yield_ms)) + return max(1, delay) + + def _on_timer(self, timer): + start = self._advance_lvgl_clock() + if self._pause > 0: + timer.reschedule(self.delay) + return + wanted = self.task_handler() + end = ticks_ms() + work = ticks_diff(end, start) + self.passes += 1 + self.last_pass_ms = work + if not timer.running: + return # deinit() during the pass + timer.reschedule(self._next_delay(wanted, work)) + + def default_exception_sink(self, e): + """Print ``e`` with traceback to stderr (default :attr:`exception_sink`).""" + if hasattr(sys, "print_exception"): # MicroPython / CircuitPython + sys.print_exception(e) + else: # CPython has no sys.print_exception + import traceback + + traceback.print_exception(type(e), e, e.__traceback__) + + +def main(): + """Initialize LVGL, wire :class:`DisplayDriver`, and enable the event loop. + + Called automatically on ``import display_driver`` using the active + :class:`appdev.App`, or the legacy ``board_config`` fallback. + """ + global _driver_ref, _drivers, _refresh_claim, _polling_claim + gc.collect() + if not lv.is_initialized(): + lv.init() + if _refresh_claim is None and getattr(app, "_refresh_claim", None) is None: + _refresh_claim = app.pause_refresh() + if _polling_claim is None and getattr(app, "_polling_claim", None) is None: + _polling_claim = app.pause_polling() + loop_inst = event_loop.current_instance() + if loop_inst is not None: + # Already-running loop: pause around driver (re)construction. + loop_inst.disable() + try: + # No timer callback of any kind runs while the draw buffers are + # created (a soft-timer callback during draw_buf_create corrupts LVGL + # handlers on an ESP32-P4); the hold releases what came due after. + with multimer.hold(): + if lv.group_get_default() is None: + lv.group_create().set_default() + _driver_ref = DisplayDriver(display_drv, app.devices) + _drivers = [_driver_ref] + if loop_inst is None: + loop_inst = event_loop(period_ms=LVGL_PERIOD_MS) + _ensure_host_pump() + finally: + if loop_inst is not None: + loop_inst.enable() + + def _lvgl_shutdown_before_quit(): + # Stop the bridge before releasing the display so no callback can touch + # LVGL state during interpreter finalization. + global _host_pump_timer, _refresh_claim, _polling_claim + if _host_pump_timer is not None: + try: + _host_pump_timer.deinit() + except Exception: + pass + _host_pump_timer = None + for drv in _drivers: + drv.release_frame_clock() + if _refresh_claim is not None: + try: + _refresh_claim.release() + except Exception: + pass + _refresh_claim = None + if _polling_claim is not None: + try: + _polling_claim.release() + except Exception: + pass + _polling_claim = None + inst = event_loop.current_instance() + if inst is not None: + inst.deinit() + try: + if lv.is_initialized(): + lv.deinit() + except Exception: + pass + + app.before_quit = _lvgl_shutdown_before_quit + + +def _ensure_host_pump(): + """Keep HOST/SDL draining on a 10 ms timer for all drivers. + + LVGL reads its indevs from its own timers, and they stop while the loop + is paused; the window must still answer the OS, so the host queue is + drained here regardless. + """ + global _host_pump_timer + if _host_pump_timer is not None and _host_pump_timer.running: + return + if not any(getattr(drv, "virtual_devices", ()) for drv in _drivers): + # A board with no host window has nothing to drain: no timer, so a + # panel is not woken 100 times a second for an empty loop. + return + + def _host_pump(_t): + for drv in _drivers: + for vd in getattr(drv, "virtual_devices", ()): + vd.poll_host_device() + + _host_pump_timer = multimer.every(LVGL_PERIOD_MS, _host_pump, name="lvgl.host_pump") + + +def attach(display, devices=None, *, color_format=None, blocking=True): + """Attach an additional displaydev panel as an LVGL display. + + Call after ``import display_driver`` (primary already wired). The display + is also registered with this module's LVGL app. + + Args: + display: Secondary displaydev driver. + devices: Optional LVGL input devices to bind as indevs on this display. + When omitted and ``app.host_dev`` exists, that host device is + reused (window-filtered) so the secondary panel receives pointer + input. + color_format: LVGL color format; default RGB565. + blocking: Passed to :class:`DisplayDriver`. + + Returns: + DisplayDriver: The new bridge instance. + """ + global _drivers + if not lv.is_initialized(): + raise RuntimeError("import display_driver before attach()") + prev_default = lv.display_get_default() if hasattr(lv, "display_get_default") else None + if devices is None: + devices = [] + if getattr(app, "host_dev", None) is not None: + devices = [app.host_dev] + app.add_display(display) + kwargs = {"devs": devices, "blocking": blocking} + if color_format is not None: + kwargs["color_format"] = color_format + with multimer.hold(): + drv = DisplayDriver(display, **kwargs) + _drivers.append(drv) + _ensure_host_pump() + if prev_default is not None and hasattr(prev_default, "set_default"): + prev_default.set_default() + return drv + + +def attach_devices(devs, lv_display=None): + """Register LVGL input devices as LVGL indevs without creating a display. + + Args: + devs: Iterable of LVGL input devices (encoder, keypad, pointer, …). + lv_display: Target ``lv.display``; default is the primary LVGL display. + + Returns: + list: Virtual devices accumulated by :func:`create_devices`. + """ + if lv_display is None: + if not _drivers: + raise RuntimeError("no LVGL display; import display_driver first") + lv_display = _drivers[0].lv_display + return create_devices(devs, lv_display) + + +def _touch_state_for(device): + """Per-pointer touch state (must not be module-global — multi-display).""" + st = getattr(device, "_lv_touch", None) + if st is None: + st = {"x": 0, "y": 0, "pressed": False} + device._lv_touch = st + return st + + +def _make_touch_cb(device): + """Build a pointer event_cb that updates only ``device``'s touch state.""" + + def _touch_cb(event, indev, data): + st = _touch_state_for(device) + if event is not None: + if event.type == events.MOUSEBUTTONDOWN and event.button == 1: + st["x"], st["y"] = event.pos + st["pressed"] = True + elif event.type == events.MOUSEMOTION and event.buttons[0]: + st["x"], st["y"] = event.pos + elif event.type == events.MOUSEBUTTONUP and event.button == 1: + st["x"], st["y"] = event.pos + st["pressed"] = False + data.point = lv.point_t({"x": st["x"], "y": st["y"]}) + data.state = lv.INDEV_STATE.PRESSED if st["pressed"] else lv.INDEV_STATE.RELEASED + + return _touch_cb + + +# CPython: module-level lv.indev_gesture_recognizers_*; MP/CP: indev methods. +_GESTURE_UPDATE = hasattr(lv, "indev_touch_data_t") +# LVGL ``LV_GESTURE_MAX_POINTS`` is 2; finger id is stored as int8_t (-1 = free). +_MAX_GESTURE_TOUCHES = 2 +# Windows/pygame often flickers or renumbers finger_id mid-pinch. Track by +# position → stable LVGL slots 0/1, and hold a slot briefly after the OS drops it +# so LVGL does not cancel ONGOING pinch (requires finger_cnt == 2). +_GESTURE_STICKY_MS = 250 +_gesture_touches = None +# id(device) -> {slot: (x, y, last_ms)} +_gesture_slots = {} + + +def _gesture_tick_ms(): + try: + return int(lv.tick_get()) + except Exception: + return 0 + + +def _gesture_dist2(a, b): + dx = a[0] - b[0] + dy = a[1] - b[1] + return dx * dx + dy * dy + + +def _gesture_track_slots(dev_key, points, now): + """Map live contacts to stable slots 0..1 by nearest prior position. + + Returns (pressed dict slot→(x,y), released slot list). + """ + live = [(int(pt[0]), int(pt[1])) for pt in points] + prev = _gesture_slots.get(dev_key) or {} + new_slots = {} + assigned_live = set() + + # Match against last-known positions (ignore OS finger_id churn). + if live and prev: + slot_ids = list(prev.keys()) + if len(live) == 2 and len(slot_ids) == 2: + s0, s1 = slot_ids[0], slot_ids[1] + d_same = _gesture_dist2(live[0], prev[s0][:2]) + _gesture_dist2(live[1], prev[s1][:2]) + d_swap = _gesture_dist2(live[0], prev[s1][:2]) + _gesture_dist2(live[1], prev[s0][:2]) + if d_same <= d_swap: + new_slots[s0] = (live[0][0], live[0][1], now) + new_slots[s1] = (live[1][0], live[1][1], now) + else: + new_slots[s1] = (live[0][0], live[0][1], now) + new_slots[s0] = (live[1][0], live[1][1], now) + assigned_live = {0, 1} + else: + pairs = [] + for li, xy in enumerate(live): + for s, (sx, sy, _) in prev.items(): + pairs.append((_gesture_dist2(xy, (sx, sy)), li, s)) + pairs.sort() + used_s = set() + for _, li, s in pairs: + if li in assigned_live or s in used_s: + continue + assigned_live.add(li) + used_s.add(s) + x, y = live[li] + new_slots[s] = (x, y, now) + + for li, xy in enumerate(live): + if li in assigned_live: + continue + for s in range(_MAX_GESTURE_TOUCHES): + if s not in new_slots: + new_slots[s] = (xy[0], xy[1], now) + assigned_live.add(li) + break + + # Hold dropped contacts briefly only while another contact is still live + # (pinch 2→1 flicker). On a full lift, clear immediately so the pointer + # RELEASED / SHORT_CLICKED path is not blocked by a sticky PRESSED slot. + if live: + for s, (x, y, t) in prev.items(): + if s in new_slots: + continue + age = (now - t) & 0xFFFFFFFF + if age <= _GESTURE_STICKY_MS and len(new_slots) < _MAX_GESTURE_TOUCHES: + new_slots[s] = (x, y, t) + + released = [s for s in prev if s not in new_slots] + _gesture_slots[dev_key] = new_slots + pressed = {s: (xy[0], xy[1]) for s, xy in new_slots.items()} + return pressed, released + + +def _gesture_recognizers_update(indev, touches, touch_cnt): + fn = getattr(lv, "indev_gesture_recognizers_update", None) + if fn is not None: + fn(indev, touches, touch_cnt) + else: + indev.gesture_recognizers_update(touches, touch_cnt) + + +def _gesture_recognizers_set_data(indev, data): + fn = getattr(lv, "indev_gesture_recognizers_set_data", None) + if fn is not None: + fn(indev, data) + else: + indev.gesture_recognizers_set_data(data) + + +def _configure_gesture_recognizers(indev): + """Tune LVGL multitouch recognizers so pinch is not stolen. + + Upstream ``lv_indev_gesture_detect_rotation`` zero-inits its config; with + ``rotation_angle_rad_threshold == 0``, any tiny twist becomes RECOGNIZED + and ``recognizers_update`` resets the still-ONGOING pinch. Two-finger + swipe can steal the same way once the contact center moves + ``gesture_min_distance`` pixels. + """ + if not _GESTURE_UPDATE: + return + + set_rot = getattr(lv, "indev_set_rotation_rad_threshold", None) + if set_rot is not None: + set_rot(indev, 3.5) + elif hasattr(indev, "set_rotation_rad_threshold"): + indev.set_rotation_rad_threshold(3.5) + + set_md = getattr(lv, "indev_set_gesture_min_distance", None) + if set_md is not None: + set_md(indev, 255) + elif hasattr(indev, "set_gesture_min_distance"): + indev.set_gesture_min_distance(255) + + # Laptop touchscreens rarely hit the stock 0.75 / 1.5 pinch gates cleanly. + set_down = getattr(lv, "indev_set_pinch_down_threshold", None) + set_up = getattr(lv, "indev_set_pinch_up_threshold", None) + if set_down is not None: + set_down(indev, 0.92) + elif hasattr(indev, "set_pinch_down_threshold"): + indev.set_pinch_down_threshold(0.92) + if set_up is not None: + set_up(indev, 1.12) + elif hasattr(indev, "set_pinch_up_threshold"): + indev.set_pinch_up_threshold(1.12) + + +def _gesture_feed(indev, data, device): + """Feed multipoint contacts into LVGL gesture recognizers when available.""" + global _gesture_touches + if not _GESTURE_UPDATE: + return + + points = getattr(device, "points", None) + if not points: + st = _touch_state_for(device) + points = ((st["x"], st["y"]),) if st["pressed"] else () + + dev_key = id(device) + now = _gesture_tick_ms() + pressed, released = _gesture_track_slots(dev_key, points or (), now) + + count = len(pressed) + len(released) + if _gesture_touches is None: + _gesture_touches = lv.indev_touch_data_t(_MAX_GESTURE_TOUCHES) + + if count == 0: + _gesture_slots[dev_key] = {} + _gesture_recognizers_update(indev, _gesture_touches, 0) + _gesture_recognizers_set_data(indev, data) + return + + n = count if count <= _MAX_GESTURE_TOUCHES else _MAX_GESTURE_TOUCHES + ts = now + idx = 0 + for contact_id, (x, y) in pressed.items(): + if idx >= n: + break + t = _gesture_touches[idx] + t.point = lv.point_t({"x": x, "y": y}) + t.state = lv.INDEV_STATE.PRESSED + t.id = contact_id + t.timestamp = ts + idx += 1 + for contact_id in released: + if idx >= n: + break + t = _gesture_touches[idx] + t.point = lv.point_t({"x": 0, "y": 0}) + t.state = lv.INDEV_STATE.RELEASED + t.id = contact_id + t.timestamp = ts + idx += 1 + + _gesture_recognizers_update(indev, _gesture_touches, idx) + _gesture_recognizers_set_data(indev, data) + st = _touch_state_for(device) + data.point = lv.point_t({"x": st["x"], "y": st["y"]}) + + +# A wheel event carries a legacy integer x/y pair and a float +# precise_x/precise_y pair, and which pair holds real data is a per-build +# fact about usdl2 (verified empirically 2026-08-27, WSLg): the desktop +# MicroPython build labels both axes correctly only in precise_* (a pure +# vertical swipe *also* sets a spurious legacy x), while the CPython wheel +# build has usable legacy fields but garbles precise_* into float32 +# reinterpretations of small ints — denormals around 1e-45. The epsilon +# guard below rejects that garbage, so one rule serves both: use precise +# when it carries sane data, else fall back to legacy, and never mix the +# pairs within one event (per-channel mixing double-counts the mislabeled +# builds). On the legacy path the primary scroll arrives on x — the field +# this callback has always read. +_WHEEL_EPSILON = 1e-3 + +# App-configurable mapping, see set_wheel_mapping(). Defaults preserve the +# historical behavior exactly: the primary axis adjusts, nothing navigates. +_wheel_adjust_axis = "v" +_wheel_adjust_sign = 1 +_wheel_navigate_sign = 1 +_wheel_navigates = False +_wheel_adjust_accum = 0.0 +_wheel_navigate_accum = 0.0 + + +def set_wheel_mapping(adjust_axis=None, adjust_sign=None, navigate=None, navigate_sign=None): + """Configure how the two wheel/swipe axes map onto LVGL. + + ``adjust_axis``: "v" (default) or "h" — which axis drives the encoder + indev, i.e. adjusts the focused control's value. Pick the axis running + parallel to the control's orientation: horizontal sliders read best + with ``"h"``, vertical sliders and knobs with ``"v"``. + ``adjust_sign``: 1 or -1 to flip the adjust direction. + ``navigate``: when True, the *other* axis moves group focus between + controls (``lv.group_t.focus_next``/``focus_prev`` on the default + group), giving wheel-only browse-and-tweak. When False (default) the + secondary axis is used only as a fallback when the primary is silent, + which keeps single-axis sources such as hardware encoders working + regardless of which field they populate. + ``navigate_sign``: 1 or -1 to flip which way focus travels. Needed + independently of ``adjust_sign`` because the two axes come from + different sources with their own conventions -- SDL and Win32 disagree + about the sign of vertical scroll, and "swipe down goes to the next + control" is a claim about that axis alone. + """ + global _wheel_adjust_axis, _wheel_adjust_sign, _wheel_navigates + global _wheel_navigate_sign + if adjust_axis is not None: + if adjust_axis not in ("v", "h"): + raise ValueError("adjust_axis must be 'v' or 'h'") + _wheel_adjust_axis = adjust_axis + if adjust_sign is not None: + _wheel_adjust_sign = 1 if adjust_sign >= 0 else -1 + if navigate is not None: + _wheel_navigates = bool(navigate) + if navigate_sign is not None: + _wheel_navigate_sign = 1 if navigate_sign >= 0 else -1 + + +def _wheel_axes(event): + """Resolve one MOUSEWHEEL event to (horizontal, vertical) deltas.""" + px, py = event.precise_x, event.precise_y + if -_WHEEL_EPSILON < px < _WHEEL_EPSILON: + px = 0.0 + if -_WHEEL_EPSILON < py < _WHEEL_EPSILON: + py = 0.0 + if px or py: + h, v = px, py + else: + v, h = event.x, event.y + if event.flipped: + h, v = -h, -v + return h, v + + +def _wheel_split(event): + """Return (adjust_delta, navigate_delta) per the configured mapping.""" + h, v = _wheel_axes(event) + adjust, other = (v, h) if _wheel_adjust_axis == "v" else (h, v) + if not _wheel_navigates and adjust == 0: + adjust, other = other, 0.0 + if not _wheel_navigates: + return adjust * _wheel_adjust_sign, 0.0 + return adjust * _wheel_adjust_sign, other * _wheel_navigate_sign + + +def _encoder_cb(event, indev=None, data=None): + global _wheel_adjust_accum + if event is None or data is None: + return + if event.type == events.MOUSEWHEEL: + adjust, _ = _wheel_split(event) + _wheel_adjust_accum += adjust + steps = int(_wheel_adjust_accum) + _wheel_adjust_accum -= steps + data.enc_diff = steps + elif event.type == events.MOUSEBUTTONDOWN and event.button == 3: + data.state = lv.INDEV_STATE.PRESSED + elif event.type == events.MOUSEBUTTONUP and event.button == 3: + data.state = lv.INDEV_STATE.RELEASED + + +def _wheel_navigate_cb(event): + """Move default-group focus with the non-adjust axis (opt-in). + + Calls the group API directly rather than adding a second encoder + indev: ``focus_next``/``focus_prev`` send FOCUSED/DEFOCUSED but never + touch the group's editing flag, so the control landed on keeps + whatever edit state its own FOCUSED handler establishes and the + adjust axis acts on it immediately. + """ + global _wheel_navigate_accum + if not _wheel_navigates: + return + _, navigate = _wheel_split(event) + _wheel_navigate_accum += navigate + steps = int(_wheel_navigate_accum) + _wheel_navigate_accum -= steps + if not steps: + return + g = lv.group_get_default() + if g is None: + return + for _ in range(abs(steps)): + if steps > 0: + g.focus_next() + else: + g.focus_prev() + + +# US QWERTY unshifted → shifted printable (SDL often reports base key + KMOD_SHIFT). +_SHIFT_MAP = { + ord("`"): ord("~"), + ord("1"): ord("!"), + ord("2"): ord("@"), + ord("3"): ord("#"), + ord("4"): ord("$"), + ord("5"): ord("%"), + ord("6"): ord("^"), + ord("7"): ord("&"), + ord("8"): ord("*"), + ord("9"): ord("("), + ord("0"): ord(")"), + ord("-"): ord("_"), + ord("="): ord("+"), + ord("["): ord("{"), + ord("]"): ord("}"), + ord("\\"): ord("|"), + ord(";"): ord(":"), + ord("'"): ord('"'), + ord(","): ord("<"), + ord("."): ord(">"), + ord("/"): ord("?"), +} + + +def _modifier_bit(event): + """Return ``KMOD_*`` bit for a modifier key event, or 0.""" + k = event.key + name = getattr(event, "name", None) or "" + by_key = { + keys.K_LSHIFT: keys.KMOD_LSHIFT, + keys.K_RSHIFT: keys.KMOD_RSHIFT, + keys.K_LCTRL: keys.KMOD_LCTRL, + keys.K_RCTRL: keys.KMOD_RCTRL, + keys.K_LALT: keys.KMOD_LALT, + keys.K_RALT: keys.KMOD_RALT, + keys.K_LGUI: keys.KMOD_LGUI, + keys.K_RGUI: keys.KMOD_RGUI, + } + bit = by_key.get(k) + if bit: + return bit + by_name = { + "Left Shift": keys.KMOD_LSHIFT, + "Right Shift": keys.KMOD_RSHIFT, + "Left Ctrl": keys.KMOD_LCTRL, + "Right Ctrl": keys.KMOD_RCTRL, + "Left Alt": keys.KMOD_LALT, + "Right Alt": keys.KMOD_RALT, + "Left GUI": keys.KMOD_LGUI, + "Right GUI": keys.KMOD_RGUI, + } + return by_name.get(name, 0) + + +def _apply_mods(k, mod): + """Apply Shift/Caps to a printable ASCII codepoint.""" + shift = bool(mod & keys.KMOD_SHIFT) + caps = bool(mod & keys.KMOD_CAPS) + if 97 <= k <= 122: # a-z + if shift ^ caps: + return k - 32 + return k + if 65 <= k <= 90: # A-Z + if shift ^ caps: + return k + return k + 32 + if shift and k in _SHIFT_MAP: + return _SHIFT_MAP[k] + return k + + +def _lv_key_from_event(event, tracked_mods=0): + """Map shared SDL-style key codes to ``lv.KEY_*`` / Unicode for LVGL. + + Arrows become caret keys (``lv.KEY.LEFT``/…). Tab still moves group focus + (``NEXT`` / ``PREV``). Modifier keys are not returned — they corrupt text + widgets if inserted as huge SDLK values. Printable ASCII gets Shift/Caps + via ``event.mod`` and optional ``tracked_mods``. + + Returns ``None`` for keys that must not update the LVGL keypad. + """ + k = event.key + name = getattr(event, "name", None) or "" + mod = (getattr(event, "mod", 0) or 0) | (tracked_mods or 0) + + if _modifier_bit(event): + return None + + # Scancode-derived SDLK → character / control (if a host skipped sdldisplay normalize). + if isinstance(k, int) and (k & 0x40000000) and name: + if len(name) == 1: + k = ord(name.lower()) + elif name == "Space": + k = 32 + elif name == "Return": + k = keys.K_RETURN + elif name == "Backspace": + k = keys.K_BACKSPACE + elif name == "Escape": + k = keys.K_ESCAPE + elif name == "Tab": + k = keys.K_TAB + elif name == "Delete": + k = keys.K_DELETE + + if k == keys.K_TAB or name == "Tab": + if mod & keys.KMOD_SHIFT: + return lv.KEY.PREV + return lv.KEY.NEXT + if k == keys.K_RIGHT or name == "Right": + return lv.KEY.RIGHT + if k == keys.K_LEFT or name == "Left": + return lv.KEY.LEFT + if k == keys.K_DOWN or name == "Down": + return lv.KEY.DOWN + if k == keys.K_UP or name == "Up": + return lv.KEY.UP + if k in (keys.K_RETURN, keys.K_KP_ENTER) or name == "Return": + return lv.KEY.ENTER + if k == keys.K_ESCAPE or name == "Escape": + return lv.KEY.ESC + if k == keys.K_BACKSPACE or name == "Backspace": + return lv.KEY.BACKSPACE + if k == keys.K_DELETE or name == "Delete": + return lv.KEY.DEL + if k == keys.K_HOME or name == "Home": + return lv.KEY.HOME + if k == keys.K_END or name == "End": + return lv.KEY.END + if not isinstance(k, int) or not (32 <= k <= 126): + return None + return _apply_mods(k, mod) + + +def _make_keypad_cb(device): + """Build a keypad event_cb that always writes press state (idle-safe).""" + st = getattr(device, "_lv_key", None) + if st is None: + st = {"key": 0, "pressed": False, "mods": 0} + device._lv_key = st + else: + st.setdefault("mods", 0) + + def _keypad_cb(event, indev=None, data=None): + if data is None: + return + if event is not None: + bit = _modifier_bit(event) + if bit: + if event.type == events.KEYDOWN: + st["mods"] |= bit + elif event.type == events.KEYUP: + st["mods"] &= ~bit + else: + key = _lv_key_from_event(event, st["mods"]) + if key is not None: + if event.type == events.KEYDOWN: + st["pressed"] = True + st["key"] = key + elif event.type == events.KEYUP: + st["pressed"] = False + st["key"] = key + data.state = lv.INDEV_STATE.PRESSED if st["pressed"] else lv.INDEV_STATE.RELEASED + data.key = st["key"] + + return _keypad_cb + + +def create_devices(devs, lv_display, virtual_devices=None, window_id=None): + """Register LVGL input devices as LVGL indevs (pointer / encoder / keypad). + + Args: + devs: Iterable of LVGL input devices from ``app.devices``. + lv_display: LVGL display object to attach indevs to. + virtual_devices: Optional list mutated when expanding :class:`HostEventsDevice` + into virtual pointer/keypad devices. + window_id: OS window id for host fan-out filtering (multi-display). + + Returns: + list: Accumulated virtual devices (for host expansion). + """ + if virtual_devices is None: + virtual_devices = [] + for device in devs: + if device.type in (POINTER, ENCODER, KEYPAD): + indev = lv.indev_create() + indev.set_display(lv_display) + device.user_data = indev + if device.type == POINTER: + event_cb = _make_touch_cb(device) + device.subscribe(event_cb) + indev.set_type(lv.INDEV_TYPE.POINTER) + _configure_gesture_recognizers(indev) + elif device.type == ENCODER: + event_cb = _encoder_cb + device.subscribe(event_cb) + indev.set_type(lv.INDEV_TYPE.ENCODER) + elif device.type == KEYPAD: + event_cb = _make_keypad_cb(device) + device.subscribe(event_cb) + indev.set_type(lv.INDEV_TYPE.KEYPAD) + + # LVGL calls read_cb every period with (indev, data). device.poll + # only invokes subscribers when there is a new event, so idle + # reads never wrote data.state/point — taps were invisible. + def _read_cb(indev_obj, data, _dev=device, _cb=event_cb): + _dev.poll(indev_obj, data) + _cb(None, indev_obj, data) + # Host backends drain native input in batches. Ask LVGL to call + # us again in this read cycle until the virtual-device FIFO is + # empty, preserving fast KEYDOWN/KEYUP sequences without adding + # one LVGL refresh period of latency per transition. + data.continue_reading = bool(getattr(_dev, "has_pending", False)) + if _dev.type == POINTER: + _gesture_feed(indev_obj, data, _dev) + + indev.set_group(lv.group_get_default()) + indev.set_read_cb(_read_cb) + # Default indev timer uses LV_DEF_REFR_PERIOD (~33 ms); match task_handler. + read_timer = indev.get_read_timer() + if read_timer is not None: + read_timer.set_period(LVGL_PERIOD_MS) + elif device.type == HOST: + wid = window_id + if wid is None: + host_disp = getattr(device, "_data", None) + if host_disp is not None: + wid = getattr(host_disp, "_window_id", None) + vd = VirtualDevices(device, window_id=wid) + virtual_devices.append(vd) + create_devices(vd.devices, lv_display, virtual_devices, window_id=wid) + return virtual_devices + + + +class DisplayDriver: + """Bridge a displaydev driver to an LVGL display + input devices. + + Creates the LVGL display, chooses DIRECT (shared framebuffer) or PARTIAL + render mode, installs flush callbacks, and wires LVGL input devices via + :func:`create_devices`. A PARTIAL panel is presented from the display's + own frame clock, only after a flush; a DIRECT panel presents from + ``flush_is_last``. LVGL's refresh timer is set to the display's frame + period, so rendering and presenting share one cadence. + """ + + def __init__( + self, + display_drv, + devs=None, + color_format=lv.COLOR_FORMAT.RGB565, + blocking=True, + ): + """Create LVGL display buffers and register input devices. + + Args: + display_drv: displaydev driver (BusDisplay, SDLDisplay, FBDisplay, …). + devs: Iterable of LVGL input devices to register as LVGL indevs. + color_format: LVGL color format (default RGB565). + blocking: When False, register a bus flush-ready callback for async blit. + """ + if devs is None: + devs = [] + gc.collect() + self.display_drv = display_drv + if display_drv.requires_byteswap: + self._needs_swap = display_drv.disable_auto_byteswap(True) + else: + self._needs_swap = False + self._color_size = lv.color_format_get_size(color_format) + self._blocking = blocking + self._share_fb = False + self._draw_buf1 = None + self._draw_buf2 = None + # Keep Python refs alive for set_buffers panel views (GC must not free). + self._fb_share = None + self._dirty = False + self._frame_clock = None + self.presents = 0 + + self.lv_display = lv.display_create(display_drv.width, display_drv.height) + self.lv_display.set_color_format(color_format) + + share = bool(getattr(display_drv, "share_framebuffer", False)) + # Byteswap + shared FB not supported yet — keep PARTIAL blit path. + fbs = None + if share and not self._needs_swap: + try: + fbs = display_drv.framebuffers() + except Exception: + fbs = None + + if fbs is not None: + buf1, buf2, nbytes, stride = fbs + packed = int(display_drv.width) * self._color_size + self._fb_share = (buf1, buf2) + self._share_fb = True + self.lv_display.set_flush_cb(self._flush_cb_direct) + if ( + stride + and int(stride) != packed + and hasattr(self.lv_display, "set_buffers_with_stride") + ): + self.lv_display.set_buffers_with_stride( + buf1, buf2, int(nbytes), int(stride), lv.DISPLAY_RENDER_MODE.DIRECT + ) + else: + self.lv_display.set_buffers(buf1, buf2, int(nbytes), lv.DISPLAY_RENDER_MODE.DIRECT) + else: + self._draw_buf1 = lv.draw_buf_create( + display_drv.width, display_drv.height // 10, color_format, 0 + ) + self._draw_buf2 = lv.draw_buf_create( + display_drv.width, display_drv.height // 10, color_format, 0 + ) + self.lv_display.set_flush_cb(self._flush_cb) + if not self._blocking: + display_drv.display_bus.register_callback(self.lv_display.flush_ready) + self.lv_display.set_draw_buffers(self._draw_buf1, self._draw_buf2) + self.lv_display.set_render_mode(lv.DISPLAY_RENDER_MODE.PARTIAL) + + self._match_refresh_period() + if not self._share_fb and callable(getattr(display_drv, "show", None)): + self._subscribe_frame_clock() + + self.virtual_devices = create_devices( + devs, + self.lv_display, + window_id=getattr(display_drv, "_window_id", None), + ) + + # -- frame pacing ------------------------------------------------------ + + @property + def frame_period_ms(self): + """The display's frame period, from its frame clock or the fallback.""" + period = getattr(self.display_drv, "refresh_period_ms", None) + try: + period = int(period) + except (TypeError, ValueError): + period = 0 + return period if period > 0 else LVGL_REFR_PERIOD_MS + + def _match_refresh_period(self): + """Set LVGL's refresh timer to the display's frame period.""" + try: + timer = self.lv_display.get_refr_timer() + except Exception: + return + if timer is not None: + try: + timer.set_period(self.frame_period_ms) + except Exception: + pass + + def _subscribe_frame_clock(self): + fc = getattr(self.display_drv, "frame_clock", None) + if fc is None: + return + self._frame_clock = fc + fc.subscribe(self._present) + + def release_frame_clock(self): + fc = self._frame_clock + self._frame_clock = None + if fc is not None: + try: + fc.unsubscribe(self._present) + except Exception: + pass + + def _present(self): + """Frame-clock callback: present a PARTIAL panel that was flushed to.""" + if not self._dirty: + return + self._dirty = False + panel = self.display_drv + if hasattr(panel, "_sdl_active") and not panel._sdl_active(): + return + try: + panel.show() + self.presents += 1 + except Exception: + pass + + # -- flush ------------------------------------------------------------- + + def _flush_cb_direct(self, disp_drv, area, color_p): + """DIRECT: LVGL already painted the panel FB; present on last area.""" + panel = self.display_drv + if hasattr(panel, "_sdl_active") and not panel._sdl_active(): + self.lv_display.flush_ready() + return + try: + last = self.lv_display.flush_is_last() + except Exception: + last = True + synced = False + flush_rect = getattr(panel, "flush_rect", None) + if flush_rect is not None: + try: + synced = bool( + flush_rect( + area.x1, + area.y1, + area.x2 - area.x1 + 1, + area.y2 - area.y1 + 1, + ) + ) + except Exception: + synced = False + if last: + # The frame is out: by show() when flush_rect could not sync it, + # otherwise flush_rect already did and frame_done() says so (a + # no-op unless the panel's measure_fps is on). Either way it is + # one present, however many areas it took. + try: + if synced: + frame_done = getattr(panel, "frame_done", None) + if frame_done is not None: + frame_done() + else: + panel.show() + self.presents += 1 + except Exception: + pass + if self._blocking: + self.lv_display.flush_ready() + + def _flush_cb(self, disp_drv, area, color_p): + panel = self.display_drv + if hasattr(panel, "_sdl_active") and not panel._sdl_active(): + self.lv_display.flush_ready() + return + width = area.x2 - area.x1 + 1 + height = area.y2 - area.y1 + 1 + + if self._needs_swap: + lv.draw_sw_rgb565_swap(color_p, width * height) + + data = color_p.__dereference__(width * height * self._color_size) + panel.blit_rect(data, area.x1, area.y1, width, height) + self._dirty = True + if self._blocking: + self.lv_display.flush_ready() + + +# Import-time bootstrap (same as before the probe split). +main() diff --git a/pydevices-desktop.toml b/pydevices-desktop.toml index 0cfe3e3e..6cd28bde 100644 --- a/pydevices-desktop.toml +++ b/pydevices-desktop.toml @@ -38,6 +38,7 @@ "https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_config.py" = "/lib/board_config.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/board_configs/desktop/board_peripherals.py" = "/lib/board_peripherals.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/boarddev.py" = "/lib/boarddev.py" +"https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/display_driver.py" = "/lib/display_driver.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/displaydev/__init__.py" = "/lib/displaydev/__init__.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/displaydev/_desktop.py" = "/lib/displaydev/_desktop.py" "https://raw.githubusercontent.com/PyDevices/pydevices/main/lib/displaydev/_domkeys.py" = "/lib/displaydev/_domkeys.py" From d95ac7b3ef4accdf8428e353e343ac06036ca2c3 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Wed, 7 Oct 2026 14:58:49 -0500 Subject: [PATCH 3/3] PYDEVICES_REFRESH_MS runs any app at another frame period The display's refresh_period_ms takes the override when the driver starts, after the backend's own init(). appdev's refresh timer, the frame clock and display_driver's LVGL refresh timer all read that attribute, so LVGL and non-LVGL apps both follow it with no change to the app or the board config. env_set feeds it on hosts without an environment. An app that passes refresh_period= in code keeps its own rate, and 0 still means the app presents. tests/test_lvgl_display_driver.py brings display_driver's tests from lvgl-bindings (pacing, the nesting gate, direct-panel frames) as unittest cases, plus the LVGL refresh period; tests/test_refresh_override.py covers displaydev and appdev. --- docs/displaydev.md | 21 ++ lib/displaydev/__init__.py | 20 ++ tests/test_lvgl_display_driver.py | 407 ++++++++++++++++++++++++++++++ tests/test_refresh_override.py | 146 +++++++++++ 4 files changed, 594 insertions(+) create mode 100644 tests/test_lvgl_display_driver.py create mode 100644 tests/test_refresh_override.py diff --git a/docs/displaydev.md b/docs/displaydev.md index 3f2e0104..f85470d6 100644 --- a/docs/displaydev.md +++ b/docs/displaydev.md @@ -208,6 +208,27 @@ It is off by default and costs nothing then: `show` and `flush_rect` are the class's own methods until you switch it on, and again after `measure_fps(False)`. +### Refresh rate + +To run any app at another frame rate without touching its code or its board +config, set `PYDEVICES_REFRESH_MS`: + +```bash +PYDEVICES_REFRESH_MS=16 PYDEVICES_FPS_PRINT=5 micropython lv_test_timer.py # about 60 fps +PYDEVICES_REFRESH_MS=50 PYDEVICES_FPS_PRINT=5 micropython lv_test_timer.py # about 20 fps +``` + +It replaces the display's `refresh_period_ms` (33 ms, or 16 in the browser) +when the driver starts. Everything that paces frames reads that one attribute: +`appdev.App`'s refresh timer, the display's frame clock, and +`display_driver`'s LVGL refresh timer, so LVGL and non-LVGL apps both follow +it. A board config can set it with `displaydev.env_set("PYDEVICES_REFRESH_MS", +16)` before it builds the display, on hosts where there is no environment. + +Unset, zero or not a number leaves the display's own period. An app that +passes `appdev.App(..., refresh_period=N)` has chosen its rate in code and keeps +it; `refresh_period=0` still means the app presents frames itself. + ## Vertical scrolling Many drivers expose **ILI9341-style** vertical scroll: a top fixed band (TFA), a scrollable middle (VSA), and a bottom fixed band (BFA). You define regions with `set_vscroll(tfa, bfa)` or `vscrdef`, then move content with the `vscroll` property (wrapper around `vscsad`). diff --git a/lib/displaydev/__init__.py b/lib/displaydev/__init__.py index b851a6bd..0497dfea 100644 --- a/lib/displaydev/__init__.py +++ b/lib/displaydev/__init__.py @@ -672,6 +672,7 @@ class DisplayDriver: # The display's own frame period: what ``appdev.App`` presents at and what a # GUI's refresh timer is set to. A backend that can measure its host's # refresh (vsync, requestAnimationFrame) sets it; 33 ms otherwise. + # ``PYDEVICES_REFRESH_MS`` overrides it per instance when the driver starts. refresh_period_ms = 33 share_framebuffer = False # HostEventsDevice reads this ``(key, mod)`` tuple; None disables keyboard quit. @@ -724,6 +725,7 @@ def __init__(self, *, quiet=False): self.init() gc.collect() self._deinitialized = False + self._apply_refresh_override() if not self._quiet: print(f"{self.__class__.__name__}: initialized.") if self.requires_byteswap: @@ -742,6 +744,24 @@ def __init__(self, *, quiet=False): elif env_bool("PYDEVICES_FPS"): self.measure_fps(True) + def _apply_refresh_override(self): + """``PYDEVICES_REFRESH_MS=N`` replaces this display's frame period. + + Read once, when the driver starts, after the backend's own ``init()`` + has had its say, so it wins over a backend's default (33 ms, or the + browser's 16). ``appdev.App``'s refresh timer, the frame clock and + ``display_driver``'s LVGL refresh timer all read + :attr:`refresh_period_ms`, so LVGL and non-LVGL apps both follow it. + Unset, empty, zero or not a number leaves the backend's period alone. + """ + period = env_int("PYDEVICES_REFRESH_MS", 0) + if period <= 0: + return + self.refresh_period_ms = period + fc = getattr(self, "_frame_clock", None) + if fc is not None: + fc.period_ms = period + def __del__(self): self.deinit() diff --git a/tests/test_lvgl_display_driver.py b/tests/test_lvgl_display_driver.py new file mode 100644 index 00000000..e6f219d6 --- /dev/null +++ b/tests/test_lvgl_display_driver.py @@ -0,0 +1,407 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""``lib/display_driver.py``, the LVGL coordinator, without LVGL. + +Importing the module needs ``lvgl`` and a live ``appdev.App``, so these tests +extract one class from the real source and execute it against stand-ins, the +way they did in lvgl-bindings before the file moved here. Four things are held: + +* **Pacing.** ``event_loop`` runs ``lv.timer_handler()`` from one ``multimer`` + ONE_SHOT timer and re-arms it to what LVGL asked for. After a pass longer + than the tick it holds off for ``min(pass, max_yield_ms)``, so a slow frame + lowers the frame rate instead of taking the application's thread. Measured + on an ESP32-P4: without the hold LVGL took ~87 % of the thread + (lvgl-bindings#15); an uncapped hold doubled every UI stall on an ESP32-S3 + (lvgl-bindings#19). +* **The nesting gate.** On MicroPython and CircuitPython the binding exports + ``lv._nesting`` and ``task_handler`` gates re-entrant passes on it. CPython + exports nothing (a C ContextVar guards re-entrancy), and the gate must stand + down rather than raise. lvgl-bindings keeps the other half of this contract: + its tests fail if the generated C stops exporting ``_nesting``. +* **Direct panels.** On a DIRECT panel LVGL calls the flush once per dirty + area; the frame's last area is one present, by ``show()`` when + ``flush_rect`` could not sync the area or ``frame_done()`` when it could. +* **The frame period.** LVGL's refresh timer runs at the display's + ``refresh_period_ms``, which ``PYDEVICES_REFRESH_MS`` overrides. +""" + +import ast +import os +import sys +import types +import unittest +from unittest import mock + +import _env +from _support import quiet + +import displaydev +from displaydev import DisplayDriver + +DISPLAY_DRIVER = _env.ROOT / "lib" / "display_driver.py" + + +def _class_source(name): + source = DISPLAY_DRIVER.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(DISPLAY_DRIVER)) + node = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == name) + text = ast.get_source_segment(source, node) + assert text, "could not extract %s" % name + return text + + +def _exec_class(name, namespace): + exec(compile(_class_source(name), str(DISPLAY_DRIVER), "exec"), namespace) + return namespace[name] + + +# -- pacing ------------------------------------------------------------------- + + +class _Clock: + """A virtual ``ticks_ms`` the test advances by hand.""" + + def __init__(self): + self.now = 0 + + def ticks_ms(self): + return self.now + + @staticmethod + def ticks_diff(a, b): + return a - b + + +class _FakeTimer: + """The slice of ``multimer.Timer`` the loop uses, on the virtual clock.""" + + ONE_SHOT = 0 + PERIODIC = 1 + + def __init__(self, clock): + self._clock = clock + self.name = None + self.running = False + self.due = None + self.callback = None + + def init(self, *, mode=PERIODIC, freq=-1, period=-1, callback=None, hard=False): + self.running = True + self.callback = callback + self.due = self._clock.now + period + + def deinit(self): + self.running = False + self.due = None + + def reschedule(self, delay_ms): + self.due = self._clock.now + max(0, int(delay_ms)) + + +def _fake_multimer(clock): + mod = types.SimpleNamespace() + mod.Timer = lambda _id=-1: _FakeTimer(clock) + mod.Timer.ONE_SHOT = 0 + mod.Timer.PERIODIC = 1 + return mod + + +def _event_loop_class(lv_module, clock, nesting=None): + return _exec_class( + "event_loop", + { + "lv": lv_module, + "sys": sys, + "multimer": _fake_multimer(clock), + "ticks_ms": clock.ticks_ms, + "ticks_diff": clock.ticks_diff, + "LVGL_PERIOD_MS": 10, + "LVGL_REFR_PERIOD_MS": 33, + # The module resolves this once at import: the binding's counter + # on MicroPython / CircuitPython, None on CPython. + "_LV_NESTING": nesting, + }, + ) + + +def _mock_lv(clock, work_ms, wanted=0): + """An ``lv`` whose ``timer_handler`` costs ``work_ms`` and says a timer is + due again in ``wanted`` ms (0: at once, the animating case).""" + lv = types.SimpleNamespace() + calls = {"timer_handler": 0} + + def timer_handler(): + calls["timer_handler"] += 1 + clock.now += work_ms + return wanted + + lv.timer_handler = timer_handler + lv.tick_inc = lambda ms: None + lv.is_initialized = lambda: True + lv.init = lambda: None + return lv, calls + + +def _run(period_ms, work_ms, horizon_ms, **kwargs): + """Drive the real loop over ``horizon_ms`` of virtual wall time. + + The timer's callback is delivered on the application's thread the instant + it is due; every millisecond it is not due is one the application got. + """ + clock = _Clock() + lv, calls = _mock_lv(clock, work_ms) + loop = _event_loop_class(lv, clock)(period_ms=period_ms, **kwargs) + loop.enable() # clear the constructor's initial pause: arms the timer + timer = loop.timer + app_ms = 0 + while clock.now < horizon_ms: + if timer.running and clock.now >= timer.due: + loop._on_timer(timer) + else: + clock.now += 1 + app_ms += 1 + return app_ms, calls["timer_handler"], clock.now + + +def _hold_after_one_pass(work_ms, **kwargs): + """How long the loop stays off after one overrunning pass.""" + clock = _Clock() + lv, _calls = _mock_lv(clock, work_ms) + loop = _event_loop_class(lv, clock)(period_ms=10, **kwargs) + loop.enable() + timer = loop.timer + clock.now = timer.due + loop._on_timer(timer) # this pass overruns by work_ms + return timer.due - clock.now + + +class TestPacing(unittest.TestCase): + def test_slow_pass_leaves_the_application_a_share_of_the_thread(self): + app_ms, passes, elapsed = _run(10, 60, 1000) + self.assertGreater(passes, 0, "the loop never let LVGL run at all") + self.assertGreaterEqual(app_ms / elapsed, 0.4) + + def test_the_model_reproduces_the_board_without_the_hold(self): + """Calibration: with the yield removed LVGL takes ~87 % of the thread, + the figure lvgl-bindings#15 measured. A model that cannot fail proves + nothing.""" + app_ms, _passes, elapsed = _run(10, 68, 1000, max_yield_ms=0) + self.assertLess(app_ms / elapsed, 0.2) + + def test_fast_passes_keep_the_full_tick_cadence(self): + """The control: a fix that just throttled LVGL would pass the first test.""" + app_ms, passes, elapsed = _run(10, 1, 1000) + self.assertGreaterEqual(passes, 0.9 * (1000 // 10)) + self.assertGreaterEqual(app_ms / elapsed, 0.8) + + def test_the_loop_runs_when_lvgl_asks_not_on_a_poll(self): + """LVGL saying "nothing for 33 ms" means one pass per 33 ms, not per 10.""" + clock = _Clock() + lv, calls = _mock_lv(clock, 1, wanted=33) + loop = _event_loop_class(lv, clock)(period_ms=100) + loop.enable() + timer = loop.timer + while clock.now < 1000: + if timer.running and clock.now >= timer.due: + loop._on_timer(timer) + else: + clock.now += 1 + self.assertTrue(27 <= calls["timer_handler"] <= 31, calls) + + def test_long_pass_holds_no_longer_than_max_yield(self): + self.assertEqual(60, _hold_after_one_pass(60)) # lvgl-bindings#15's case + self.assertEqual(100, _hold_after_one_pass(500)) + self.assertEqual(30, _hold_after_one_pass(500, max_yield_ms=30)) + self.assertEqual(10, _hold_after_one_pass(500, max_yield_ms=0)) # never under a period + + +# -- the nesting gate --------------------------------------------------------- + + +class _NestingCounter: + """The shape of the binding's ``lv._nesting`` blob wrapper.""" + + def __init__(self, value=0): + self.value = value + + +class TestNestingGate(unittest.TestCase): + def _task_handler(self, nesting): + clock = _Clock() + lv, calls = _mock_lv(clock, 1, wanted=10) + loop = _event_loop_class(lv, clock, nesting=nesting)(period_ms=10) + loop._pause = 0 # task_handler called directly: no App enable() + errors = [] + loop.exception_sink = errors.append + loop.task_handler() + return calls["timer_handler"], errors + + def test_runs_with_the_counter_at_rest(self): + passes, errors = self._task_handler(_NestingCounter(0)) + self.assertEqual([], errors) + self.assertEqual(1, passes) + + def test_stands_down_while_a_callback_is_running(self): + passes, errors = self._task_handler(_NestingCounter(1)) + self.assertEqual([], errors) + self.assertEqual(0, passes) + + def test_runs_when_the_counter_is_absent_cpython_shape(self): + passes, errors = self._task_handler(None) + self.assertEqual([], errors) + self.assertEqual(1, passes) + + +# -- direct panels ------------------------------------------------------------ + + +def _display_driver_class(): + lv = types.SimpleNamespace(COLOR_FORMAT=types.SimpleNamespace(RGB565=0)) + return _exec_class("DisplayDriver", {"lv": lv, "sys": sys, "LVGL_REFR_PERIOD_MS": 33}) + + +class _LvDisplay: + """The slice of ``lv.display`` the direct flush and the frame period use.""" + + def __init__(self): + self.last = False + self.ready = 0 + self.refr_timer = _RefrTimer() + + def flush_is_last(self): + return self.last + + def flush_ready(self): + self.ready += 1 + + def get_refr_timer(self): + return self.refr_timer + + +class _RefrTimer: + period = None + + def set_period(self, ms): + self.period = ms + + +class _Area: + def __init__(self, x1, y1, x2, y2): + self.x1, self.y1, self.x2, self.y2 = x1, y1, x2, y2 + + +class _Panel: + def __init__(self, synced, has_frame_done=True): + self.synced = synced + self.flushes = 0 + self.shows = 0 + self.frames_done = 0 + if not has_frame_done: + self.frame_done = None + + def flush_rect(self, x, y, w, h): + self.flushes += 1 + return self.synced + + def show(self): + self.shows += 1 + + def frame_done(self): + self.frames_done += 1 + + +def _drive(panel, frames=3, areas=4): + cls = _display_driver_class() + drv = cls.__new__(cls) + drv.display_drv = panel + drv.lv_display = _LvDisplay() + drv._blocking = True + drv.presents = 0 + for _ in range(frames): + for i in range(areas): + drv.lv_display.last = i == areas - 1 + drv._flush_cb_direct(None, _Area(0, i * 10, 99, i * 10 + 9), None) + return drv + + +class TestDirectPanel(unittest.TestCase): + def test_a_synced_panel_reports_each_frame_once(self): + panel = _Panel(synced=True) + drv = _drive(panel) + self.assertEqual(12, panel.flushes) + self.assertEqual(0, panel.shows) + self.assertEqual(3, panel.frames_done) + self.assertEqual(3, drv.presents) + self.assertEqual(12, drv.lv_display.ready) + + def test_an_unsynced_panel_is_shown_once_a_frame(self): + panel = _Panel(synced=False) + drv = _drive(panel) + self.assertEqual(12, panel.flushes) + self.assertEqual(3, panel.shows) + self.assertEqual(0, panel.frames_done) + self.assertEqual(3, drv.presents) + + def test_a_panel_without_frame_done_still_presents(self): + """An older displaydev has no frame_done: nothing breaks.""" + panel = _Panel(synced=True, has_frame_done=False) + drv = _drive(panel) + self.assertEqual(0, panel.shows) + self.assertEqual(3, drv.presents) + + +# -- the frame period --------------------------------------------------------- + + +class _Display(DisplayDriver): + def __init__(self): + self._width = 8 + self._height = 4 + self._rotation = 0 + self._requires_byteswap = False + super().__init__(quiet=True) + + def init(self): + pass + + def fill_rect(self, x, y, w, h, c): + pass + + def blit_rect(self, buf, x, y, w, h): + pass + + def show(self, _timer=None): + pass + + +class TestFramePeriod(unittest.TestCase): + def _lvgl_period(self, **env): + displaydev._overrides.pop("PYDEVICES_REFRESH_MS", None) + clean = {k: v for k, v in os.environ.items() if not k.startswith("PYDEVICES_")} + clean.update(env) + with mock.patch.dict(os.environ, clean, clear=True), quiet(): + panel = _Display() + cls = _display_driver_class() + drv = cls.__new__(cls) + drv.display_drv = panel + drv.lv_display = _LvDisplay() + drv._match_refresh_period() + return drv.frame_period_ms, drv.lv_display.refr_timer.period + + def test_lvgl_refreshes_at_the_display_period(self): + self.assertEqual((33, 33), self._lvgl_period()) + + def test_pydevices_refresh_ms_reaches_lvgl(self): + self.assertEqual((16, 16), self._lvgl_period(PYDEVICES_REFRESH_MS="16")) + self.assertEqual((50, 50), self._lvgl_period(PYDEVICES_REFRESH_MS="50")) + + def test_a_display_without_a_period_uses_the_fallback(self): + cls = _display_driver_class() + drv = cls.__new__(cls) + drv.display_drv = types.SimpleNamespace() + self.assertEqual(33, drv.frame_period_ms) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_refresh_override.py b/tests/test_refresh_override.py new file mode 100644 index 00000000..3aa7c581 --- /dev/null +++ b/tests/test_refresh_override.py @@ -0,0 +1,146 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""``PYDEVICES_REFRESH_MS``: run any app at another frame period. + +The override lands on the display's ``refresh_period_ms`` when the driver +starts. ``appdev.App``'s refresh timer and the display's frame clock read that +attribute, and so does ``display_driver``'s LVGL refresh timer +(``test_lvgl_display_driver``), so one number moves every app. +""" + +import os +import unittest +from unittest import mock + +import _env # noqa: F401 +from _support import quiet + +import displaydev +import multimer +from appdev import App +from displaydev import DisplayDriver + + +class Panel(DisplayDriver): + def __init__(self): + self._width = 8 + self._height = 4 + self._rotation = 0 + self._requires_byteswap = False + self.shows = 0 + super().__init__(quiet=True) + + def init(self): + pass + + def fill_rect(self, x, y, w, h, c): + pass + + def blit_rect(self, buf, x, y, w, h): + pass + + def show(self, _timer=None): + self.shows += 1 + + +class BrowserPanel(Panel): + refresh_period_ms = 16 + + +class MeasuringPanel(Panel): + """A backend that learns its host's rate in init(), as a vsync one would.""" + + def init(self): + self.refresh_period_ms = 25 + + +class _Env: + def _env(self, **env): + displaydev._overrides.pop("PYDEVICES_REFRESH_MS", None) + clean = {k: v for k, v in os.environ.items() if not k.startswith("PYDEVICES_")} + clean.update(env) + return mock.patch.dict(os.environ, clean, clear=True) + + def tearDown(self): + displaydev._overrides.pop("PYDEVICES_REFRESH_MS", None) + app = App.current() + if app is not None: + app._perform_teardown() + multimer.stop_all() + multimer.keepalive(False) + + +class TestTheDisplayPeriod(_Env, unittest.TestCase): + def test_unset_keeps_the_backend_default(self): + with self._env(): + self.assertEqual(33, Panel().refresh_period_ms) + self.assertEqual(16, BrowserPanel().refresh_period_ms) + + def test_the_environment_replaces_it(self): + with self._env(PYDEVICES_REFRESH_MS="16"): + d = Panel() + self.assertEqual(16, d.refresh_period_ms) + self.assertEqual(33, Panel.refresh_period_ms, "the class default is untouched") + + def test_it_wins_over_the_browser_and_a_measured_period(self): + with self._env(PYDEVICES_REFRESH_MS="50"): + self.assertEqual(50, BrowserPanel().refresh_period_ms) + self.assertEqual(50, MeasuringPanel().refresh_period_ms) + with self._env(): + self.assertEqual(25, MeasuringPanel().refresh_period_ms) + + def test_nonsense_leaves_the_period_alone(self): + for value in ("", "0", "-5", "fast", " "): + with self.subTest(value=value), self._env(PYDEVICES_REFRESH_MS=value): + self.assertEqual(33, Panel().refresh_period_ms) + + def test_env_set_feeds_it_without_an_environment(self): + """A board config on a host with no os.environ uses env_set.""" + with self._env(): + displaydev.env_set("PYDEVICES_REFRESH_MS", 40) + self.assertEqual(40, Panel().refresh_period_ms) + + def test_the_frame_clock_follows(self): + with self._env(PYDEVICES_REFRESH_MS="20"): + d = Panel() + self.assertEqual(20, d.frame_clock.period_ms) + + +class TestTheAppRefreshTimer(_Env, unittest.TestCase): + def _app_period(self, panel_cls=Panel, **kwargs): + with quiet(): + d = panel_cls() + d.needs_refresh = True + app = App(displays=[d], **kwargs) + return [t.period for t in app._refresh_timers] + + def test_appdev_presents_at_the_override(self): + with self._env(PYDEVICES_REFRESH_MS="16"): + self.assertEqual([16], self._app_period()) + + def test_appdev_presents_at_the_default_without_it(self): + with self._env(): + self.assertEqual([33], self._app_period()) + + def test_an_app_that_chose_a_period_in_code_keeps_it(self): + with self._env(PYDEVICES_REFRESH_MS="16"): + self.assertEqual([50], self._app_period(refresh_period=50)) + + def test_refresh_period_zero_still_means_the_app_presents(self): + with self._env(PYDEVICES_REFRESH_MS="16"): + self.assertEqual([], self._app_period(refresh_period=0)) + + def test_the_override_sets_the_rate_frames_arrive_at(self): + """Count real presents: 10 ms apart is about 3x the 33 ms default.""" + with self._env(PYDEVICES_REFRESH_MS="10"): + with quiet(): + d = Panel() + d.needs_refresh = True + App(displays=[d]) + multimer.sleep_ms(330) + self.assertGreaterEqual(d.shows, 20, d.shows) + + +if __name__ == "__main__": + unittest.main()