Jean-Pierre MartineauandClaude Opus 5.5
3e251a9707
The same rewrite as for the core package. The physical screen of a handheld is now the display, and the computer is the handheld. The comment on FONT_ENV now says that it is the environment variable. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
293 lines
11 KiB
Python
293 lines
11 KiB
Python
# SPDX-License-Identifier: Apache-2.0
|
||
"""On the Cardputer Zero: pixels to the framebuffer, key events from the keyboard's device file.
|
||
|
||
The launcher starts an app with the screen for that app only, and tells it where things
|
||
are. ``APPLAUNCH_LINUX_FBDEV_DEVICE`` names the framebuffer of the display, and
|
||
``APPLAUNCH_LINUX_KEYBOARD_DEVICE`` names its keyboard. The launcher shares the keyboard
|
||
and does not grab it. Thus the app reads the key events itself, while the launcher
|
||
watches the same stream for a held Esc key (3 s, then SIGTERM to the process group of the
|
||
app, and SIGKILL 3 s after that).
|
||
|
||
This front end obeys that contract. The Esc key goes through
|
||
:mod:`~meshterm.services.hold_to_quit`, which shows the hold on the display, where the
|
||
launcher does not draw any more. The front end takes the SIGTERM as a quit, because that is
|
||
what it is (:func:`~meshterm.services.hold_to_quit.leave_on_sigterm`).
|
||
|
||
**Written before the hardware was available** (2026-09-30), from the published sources of
|
||
M5:
|
||
|
||
* the framebuffer and keyboard paths, and the Esc policy, from the launcher,
|
||
* the key codes, from the keymaps of the keyboard driver, and
|
||
* the characters of the Sym layer, from the console keymap of M5
|
||
(``tca8418_keypad_m5stack_keymap.map``).
|
||
|
||
The key mapping has unit tests. **Run on the handheld since 2026-10-05.** There, the key
|
||
codes and the Sym layer agreed with the keyboard and with the installed keymap of M5. We
|
||
also found there that the display must get its pixels through a memory mapping
|
||
(:class:`Framebuffer`).
|
||
|
||
The keyboard driver does the Fn layer in the kernel. Thus Fn+4 arrives as a plain
|
||
``KEY_F4``, and the arrows arrive as real arrow keys. Shift is a real ``KEY_LEFTSHIFT``
|
||
that stays down while the sticky Shift is armed. Sym is the exception. Its layer sends
|
||
placeholder key codes (``KEY_LEFTBRACE`` for ``!``, …), and the keymap of M5 gives them
|
||
their symbols. This module gives them the same symbols, and it is the only place in
|
||
MeshTerm that knows them.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import mmap
|
||
import os
|
||
import struct
|
||
import threading
|
||
from collections.abc import Callable
|
||
from pathlib import Path
|
||
|
||
from ..services import hold_to_quit, modifier_watch
|
||
from .font import Font
|
||
from .keys import Key, encode
|
||
from .raster import Raster, rgb565
|
||
from .run import DEFAULT_BG, DEFAULT_FG
|
||
from .vt import Terminal
|
||
|
||
FB_ENV = "APPLAUNCH_LINUX_FBDEV_DEVICE"
|
||
KEYBOARD_ENV = "APPLAUNCH_LINUX_KEYBOARD_DEVICE"
|
||
#: The keyboard device file in the launcher's own default, when no variable names one.
|
||
DEFAULT_KEYBOARD = "/dev/input/by-path/platform-3f804000.i2c-event"
|
||
|
||
#: ``struct input_event`` on a 64-bit kernel: timeval, then type, code, and value.
|
||
_EVENT = struct.Struct("llHHi")
|
||
_EV_KEY = 0x01
|
||
|
||
#: ``KEY_ESC``. When it is held, it is the launcher's way out (refer to
|
||
#: :mod:`~meshterm.services.hold_to_quit`).
|
||
_ESC = 1
|
||
|
||
# Modifier key codes (linux/input-event-codes.h).
|
||
_SHIFT = {42, 54}
|
||
_CTRL = {29, 97}
|
||
_ALT = {56, 100}
|
||
|
||
#: Key codes that are not text, mapped to the names that :mod:`.keys` spells them with.
|
||
_NAMED: dict[int, str] = {
|
||
1: "escape",
|
||
14: "backspace",
|
||
15: "tab",
|
||
28: "enter",
|
||
96: "enter",
|
||
102: "home",
|
||
103: "up",
|
||
104: "pageup",
|
||
105: "left",
|
||
106: "right",
|
||
107: "end",
|
||
108: "down",
|
||
109: "pagedown",
|
||
110: "insert",
|
||
111: "delete",
|
||
**{59 + n: f"f{n + 1}" for n in range(10)}, # KEY_F1..KEY_F10
|
||
87: "f11",
|
||
88: "f12",
|
||
}
|
||
|
||
#: The characters of the base layer.
|
||
_TEXT: dict[int, str] = {
|
||
**{2 + n: "1234567890"[n] for n in range(10)},
|
||
**{16 + n: "qwertyuiop"[n] for n in range(10)},
|
||
**{30 + n: "asdfghjkl"[n] for n in range(9)},
|
||
**{44 + n: "zxcvbnm"[n] for n in range(7)},
|
||
57: " ",
|
||
}
|
||
|
||
#: The placeholder codes of the Sym layer, and the characters that the console keymap of
|
||
#: M5 gives them.
|
||
_SYM: dict[int, str] = {
|
||
26: "!", 27: "@", 39: "#", 40: "$", 41: "%", 43: "^", 51: "&", 52: "*",
|
||
53: "(", 94: ")", 55: "~", 69: "`", 70: "_", 71: "-", 72: "+", 73: "=",
|
||
74: "[", 75: "]", 76: "{", 77: "}", 79: ";", 80: ":", 81: "'", 82: '"',
|
||
83: "<", 85: ">", 86: "\\", 89: "|", 90: ",", 91: ".", 92: "/", 93: "?",
|
||
} # fmt: skip
|
||
|
||
|
||
class KeyState:
|
||
"""Changes keyboard events into :class:`~.keys.Key` presses, and tracks the modifiers."""
|
||
|
||
def __init__(self) -> None:
|
||
"""Start with no modifier held."""
|
||
self._held: set[int] = set()
|
||
|
||
@property
|
||
def shift(self) -> bool:
|
||
"""Whether a Shift key is held or armed. The driver reports the two in the same way."""
|
||
return bool(self._held & _SHIFT)
|
||
|
||
def event(self, code: int, value: int) -> Key | None:
|
||
"""The key press that one ``EV_KEY`` event makes (``value`` 1 down, 2 repeat, 0 up)."""
|
||
if code in _SHIFT | _CTRL | _ALT:
|
||
if value:
|
||
self._held.add(code)
|
||
else:
|
||
self._held.discard(code)
|
||
return None
|
||
if value == 0:
|
||
return None
|
||
shift = self.shift
|
||
ctrl = bool(self._held & _CTRL)
|
||
alt = bool(self._held & _ALT)
|
||
name = _NAMED.get(code)
|
||
if name is not None:
|
||
return Key(name=name, shift=shift, ctrl=ctrl, alt=alt)
|
||
text = _SYM.get(code) or _TEXT.get(code)
|
||
if text is None:
|
||
return None # Fn and Sym themselves, media keys, brightness: nothing to type
|
||
if shift and text.isalpha():
|
||
text = text.upper()
|
||
return Key(text=text, ctrl=ctrl, alt=alt)
|
||
|
||
|
||
class Framebuffer:
|
||
"""The display's framebuffer, written a band of pixel rows at a time through a memory mapping.
|
||
|
||
**Through a memory mapping, never ``write()``.** The display has a DRM driver
|
||
(``panel-mipi-dbi``), and its ``/dev/fb0`` is the fbdev emulation of the kernel. That
|
||
is a buffer that the kernel copies to the display only where it knows that something
|
||
changed. What the kernel watches is the memory mapping. A write through the mapping
|
||
causes a page fault, and the kernel sends that page at the next update of the display.
|
||
|
||
Bytes from ``pwrite`` go into the same buffer, and a read of ``/dev/fb0`` shows them.
|
||
But they never get to the display. On the handheld, MeshTerm ran while the loading
|
||
screen of the launcher stayed visible over it, and the user quit a splash that they
|
||
could not see. The apps of M5 also draw through the memory mapping (LVGL's
|
||
``lv_linux_fbdev``). We found this on the hardware, 2026-10-05.
|
||
"""
|
||
|
||
def __init__(self, path: str, *, width: int, height: int) -> None:
|
||
"""Open and map the framebuffer at ``path``, for a ``width`` × ``height`` display."""
|
||
self.path = path
|
||
self._fd = os.open(path, os.O_RDWR)
|
||
stride = self._sysfs("stride")
|
||
self.stride = int(stride) if stride else width * 2
|
||
self.width = width
|
||
self.height = height
|
||
self._map = mmap.mmap(
|
||
self._fd,
|
||
self.stride * height,
|
||
mmap.MAP_SHARED, # type: ignore[attr-defined] # Linux only, as is the display
|
||
mmap.PROT_READ | mmap.PROT_WRITE, # type: ignore[attr-defined]
|
||
)
|
||
|
||
def _sysfs(self, name: str) -> str | None:
|
||
node = Path("/sys/class/graphics") / Path(self.path).name / name
|
||
try:
|
||
return node.read_text().strip()
|
||
except OSError:
|
||
return None
|
||
|
||
def write(self, pixels: bytes | bytearray, row_bytes: int, first: int, last: int) -> None:
|
||
"""Write pixel rows ``first`` to ``last`` (exclusive) of a ``row_bytes``-wide image."""
|
||
view = memoryview(pixels)
|
||
for y in range(first, min(last, self.height)):
|
||
start = y * self.stride
|
||
self._map[start : start + row_bytes] = view[y * row_bytes : (y + 1) * row_bytes]
|
||
|
||
def close(self) -> None:
|
||
"""Unmap and close the framebuffer."""
|
||
self._map.close()
|
||
os.close(self._fd)
|
||
|
||
|
||
class Device:
|
||
"""The front end on the handheld: a framebuffer writer and a keyboard reader."""
|
||
|
||
def __init__(
|
||
self,
|
||
terminal: Terminal,
|
||
lock: threading.Lock,
|
||
type_text: Callable[[str], None],
|
||
*,
|
||
font: Font,
|
||
framebuffer: str,
|
||
keyboard: str,
|
||
) -> None:
|
||
"""Open the framebuffer, then start two threads: one draws, one reads the keyboard."""
|
||
self._lock = lock
|
||
self._type = type_text
|
||
self._raster = Raster(
|
||
terminal, font, pack=rgb565, default_fg=DEFAULT_FG, default_bg=DEFAULT_BG
|
||
)
|
||
self._fb = Framebuffer(framebuffer, width=self._raster.width, height=self._raster.height)
|
||
self._keyboard = keyboard
|
||
self._ready = threading.Event()
|
||
self._closing = False
|
||
threading.Thread(target=self._draw, name="cardputer-fb", daemon=True).start()
|
||
threading.Thread(target=self._read_keys, name="cardputer-keys", daemon=True).start()
|
||
|
||
def frame_ready(self) -> None:
|
||
"""Wake the thread that draws: a frame is complete."""
|
||
self._ready.set()
|
||
|
||
def close(self) -> None:
|
||
"""Stop both threads."""
|
||
self._closing = True
|
||
self._ready.set()
|
||
|
||
def _draw(self) -> None:
|
||
while not self._closing:
|
||
self._ready.wait()
|
||
self._ready.clear()
|
||
if self._closing:
|
||
break
|
||
with self._lock:
|
||
bands = self._raster.update()
|
||
pixels = bytes(self._raster.pixels)
|
||
for first, last in bands:
|
||
self._fb.write(pixels, self._raster.stride, first, last)
|
||
self._fb.close()
|
||
|
||
def _read_keys(self) -> None:
|
||
state = KeyState()
|
||
esc = hold_to_quit.EscKey(self._type)
|
||
with open(self._keyboard, "rb", buffering=0) as stream:
|
||
while not self._closing:
|
||
data = stream.read(_EVENT.size)
|
||
if len(data) < _EVENT.size:
|
||
break
|
||
_, _, kind, code, value = _EVENT.unpack(data)
|
||
if kind != _EV_KEY:
|
||
continue
|
||
if code == _ESC:
|
||
# Down and its repeats are one hold. A tap is typed when it comes up.
|
||
if value:
|
||
esc.down()
|
||
else:
|
||
esc.up()
|
||
continue
|
||
was_shifted = state.shift
|
||
key = state.event(code, value)
|
||
if state.shift != was_shifted:
|
||
modifier_watch.report_shift(state.shift)
|
||
if key is not None:
|
||
sequence = encode(key)
|
||
if sequence:
|
||
esc.before()
|
||
self._type(sequence)
|
||
|
||
|
||
def front_end():
|
||
"""The handheld as a :data:`~.run.FrontEndFactory`, from the launcher's environment."""
|
||
from .font import find_font
|
||
|
||
font = find_font()
|
||
framebuffer = os.environ.get(FB_ENV) or "/dev/fb0"
|
||
keyboard = os.environ.get(KEYBOARD_ENV) or DEFAULT_KEYBOARD
|
||
# From here, not only from the CLI: the launcher's SIGTERM can arrive while MeshTerm
|
||
# still imports its modules. Then MeshTerm must stop as an interrupt, not unhandled.
|
||
hold_to_quit.leave_on_sigterm()
|
||
|
||
def build(terminal: Terminal, lock: threading.Lock, type_text: Callable[[str], None]):
|
||
return Device(
|
||
terminal, lock, type_text, font=font, framebuffer=framebuffer, keyboard=keyboard
|
||
)
|
||
|
||
return build
|