diff --git a/README.md b/README.md index 955a229..12158de 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,9 @@ This repo is a consumer/build repo for the LVGL stack: it consumes generated bin Requires a sibling clone of [lvgl-bindings](https://github.com/PyDevices/lvgl-bindings) whose generated binding inputs match the exact commit recorded in `LVGL_BINDINGS_COMMIT`. The Make and CMake integrations reject a mismatched source, LVGL pin, or configuration. -**Synced from lvgl-bindings:** `lib/display_driver.py` and `lib/fs_driver.py` are synced from [lvgl-bindings](https://github.com/PyDevices/lvgl-bindings) at the commit pinned in `LVGL_BINDINGS_COMMIT`, along with the generated bindings. Do not edit them here — change them in lvgl-bindings and re-sync. +**Synced from lvgl-bindings:** `lib/fs_driver.py` is synced from [lvgl-bindings](https://github.com/PyDevices/lvgl-bindings) at the commit pinned in `LVGL_BINDINGS_COMMIT`, along with the generated bindings. Do not edit it here — change it in lvgl-bindings and re-sync. + +**`display_driver` is not here.** LVGL's PyDevices coordinator lives in [pydevices `lib/`](https://github.com/PyDevices/pydevices/blob/main/lib/display_driver.py) and comes with `pydevices` (`mip.install("pydevices", index="https://PyDevices.github.io/mip")`, or frozen by micropython-pydevices' `--modules pydevices`), beside the `appdev`, `events`, `keys` and `multimer` it needs. ## Documentation @@ -59,7 +61,7 @@ include("/path/to/lvgl-micropython/manifest.py") On unix that manifest is `ports/unix/variants/standard/manifest.py`; on esp32 and rp2 it is usually `ports//boards/manifest.py`, unless your board brings its own. Then build as usual. The include brings in the `lvgl` C module -and freezes two helpers, `display_driver` and `fs_driver`. If lvgl-bindings is +and freezes `fs_driver`. If lvgl-bindings is not a sibling, pass `BINDINGS_DIR=/path/to/lvgl-bindings` to `make` on a Make port. The build stops if that checkout does not match the pin. Tested on the unix port against MicroPython v1.29.0, where `import lvgl` reports 9.5. @@ -129,7 +131,7 @@ through. ## App Usage & Timer Model -In MicroPython, `display_driver` uses `machine.Timer` (hardware interrupts): +With pydevices installed, pydevices' `display_driver` drives LVGL from `machine.Timer` (hardware interrupts): - **Interactive REPL (`micropython -i` or on-board prompt)**: Simply create widgets and drop out to the prompt. Hardware timer interrupts keep LVGL animations, timers, and touch input running continuously in the background while you inspect variables or test code interactively. - **Standalone Scripts**: Use `app.run()` if you need an explicit loop for non-interactive desktop scripts. @@ -161,8 +163,8 @@ The smoke suite belongs to the exact pinned `lvgl-bindings` source; this repo do | `micropython.mk` | Make ports (pre-1.29: `USER_C_MODULES` = parent directory) | | `micropython.cmake` | CMake ports (pre-1.29: `USER_C_MODULES` = this repo) | | `src/lv_mem_core_micropython.c` | GC-aware LVGL allocator | -| `manifest.py` | Names the C module and freezes `lib/display_driver.py`, `lib/fs_driver.py` (synced from lvgl-bindings) | -| `lib/display_driver.py` | Vendored PyDevices LVGL glue (`import display_driver`) | +| `manifest.py` | Names the C module and freezes `lib/fs_driver.py` (synced from lvgl-bindings) | +| `lib/fs_driver.py` | Python-backed LVGL filesystem driver (`import fs_driver`) | | `LVGL_BINDINGS_COMMIT` | Exact generator/artifact source consumed by builds | | `scripts/sync_from_lvgl_bindings.sh` | Refresh helpers and record an exact commit/tag | diff --git a/docs/newcomers.md b/docs/newcomers.md index cd91fc1..3a06e35 100644 --- a/docs/newcomers.md +++ b/docs/newcomers.md @@ -12,8 +12,9 @@ consumes an exact sibling checkout of ## Start by using the firmware entrypoint -On firmware built with this module, an application starts with the synced -display helper: +On firmware built with this module and with +[pydevices](https://github.com/PyDevices/pydevices) installed, an application +starts with pydevices' display helper: ```python import display_driver @@ -26,9 +27,9 @@ label = lv.label(button) label.set_text("Hello MicroPython LVGL!") ``` -`display_driver` wires the display, input, and timer integration. It needs -pydevices' `appdev`, `events`, and `keys` on the device, plus a `board_config` -unless your code has already created an `appdev.App`. At an interactive +`display_driver` wires the display, input, and timer integration. It comes +with pydevices, beside the `appdev`, `events`, and `keys` it needs, and wants a +`board_config` unless your code has already created an `appdev.App`. At an interactive MicroPython prompt, hardware timer callbacks keep LVGL active after the script reaches the prompt. @@ -44,13 +45,13 @@ lvgl-micropython build glue + GC-aware allocator MicroPython firmware module: import lvgl | v -synced display_driver helper -> display/input/timers +pydevices' display_driver -> display/input/timers ``` The Make and CMake integrations reject a mismatched binding source, LVGL pin, or generated configuration. Update `lvgl-bindings` first, then re-sync here; do not hand-edit the generated binding inputs or the synced -`display_driver.py`/`fs_driver.py` copies. +`fs_driver.py` copy. ## Repository map @@ -60,7 +61,6 @@ hand-edit the generated binding inputs or the synced | `micropython.cmake` | CMake-port user-module entrypoint. | | `src/lv_mem_core_micropython.c` | GC-aware LVGL allocator. | | `src/lvgl_micropython_build.c` | `lvgl_micropython.__revision__` build stamp. | -| `lib/display_driver.py` | Synced display, input, and timer helper. | | `lib/fs_driver.py` | Synced LVGL filesystem helper. | | `LVGL_BINDINGS_COMMIT` | Exact generated-binding source commit. | | `scripts/sync_from_lvgl_bindings.sh` | Refresh synced helpers and recorded source. | diff --git a/lib/display_driver.py b/lib/display_driver.py deleted file mode 100644 index 72f23c8..0000000 --- a/lib/display_driver.py +++ /dev/null @@ -1,1691 +0,0 @@ -# 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. - -Canonical copy lives in PyDevices/lvgl-bindings (``python/display_driver.py``). -Consumer repos (lvgl-micropython, lvgl-circuitpython, lvgl-python) -vendor a synced copy; do not edit those copies directly. - -Changes here are released only through the explicit bindings release workflow. -Consumers sync an exact bindings commit or immutable release tag. - -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 and not synced: - try: - 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/manifest.py b/manifest.py index bec9815..8127b26 100644 --- a/manifest.py +++ b/manifest.py @@ -1,12 +1,12 @@ # Frozen Python helpers that ship with the LVGL MicroPython usermod. -# Source of truth: PyDevices/lvgl-bindings python/ (display_driver.py, fs_driver.py) +# Source of truth: PyDevices/lvgl-bindings python/ (fs_driver.py). LVGL's +# PyDevices coordinator, display_driver.py, ships with pydevices (lib/). # Sync: ./scripts/sync_from_lvgl_bindings.sh # # This file only freezes usermod helpers and names the C module. Upstream # port/board/variant frozen modules come from your own manifest, which pulls # this one in with include("/manifest.py"). -module("display_driver.py", base_path="./lib", opt=3) module("fs_driver.py", base_path="./lib", opt=3) # MicroPython 1.29: the manifest names its own C module. diff --git a/scripts/sync_from_lvgl_bindings.sh b/scripts/sync_from_lvgl_bindings.sh index c5c53a0..3ca1389 100755 --- a/scripts/sync_from_lvgl_bindings.sh +++ b/scripts/sync_from_lvgl_bindings.sh @@ -1,5 +1,5 @@ #!/usr/bin/env bash -# Sync the hand-written Python helpers (display_driver.py, fs_driver.py) +# Sync the hand-written Python helper (fs_driver.py) # from PyDevices/lvgl-bindings on GitHub (not the local workspace). # # Usage: @@ -45,7 +45,7 @@ TMP=$(mktemp -d) cleanup() { rm -rf "$TMP"; } trap cleanup EXIT -HELPERS=(display_driver.py fs_driver.py) +HELPERS=(fs_driver.py) echo "Fetching ${LV_BINDINGS_REPO} @ ${REF}..." git clone --filter=blob:none --no-checkout "${LV_BINDINGS_REPO}" "${TMP}/lvgl-bindings"