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/displaydev.md b/docs/displaydev.md index b2e6f93a..f85470d6 100644 --- a/docs/displaydev.md +++ b/docs/displaydev.md @@ -171,6 +171,64 @@ 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)`. + +### 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/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/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/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/lib/displaydev/__init__.py b/lib/displaydev/__init__.py index 87365c63..0497dfea 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,12 +659,20 @@ 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. + # ``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. @@ -567,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: @@ -579,6 +738,29 @@ 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 _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() @@ -867,6 +1049,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 +1300,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/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" 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() 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()