mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-25 10:13:43 +08:00
Compare commits
2 Commits
Scotty
...
Keyboard-Test
| Author | SHA1 | Date | |
|---|---|---|---|
| b32bcf0ce4 | |||
| 5de4787a57 |
+1
-1
Submodule opendbc_repo updated: a0e6fdde92...e074f31290
@@ -0,0 +1,194 @@
|
|||||||
|
import atexit
|
||||||
|
import os
|
||||||
|
import struct
|
||||||
|
import time
|
||||||
|
from collections import deque
|
||||||
|
from glob import glob
|
||||||
|
|
||||||
|
import pyray as rl
|
||||||
|
|
||||||
|
EVENT_FORMAT = "llHHi"
|
||||||
|
EVENT_SIZE = struct.calcsize(EVENT_FORMAT)
|
||||||
|
EV_KEY = 0x01
|
||||||
|
|
||||||
|
KEY_LEFTSHIFT = 42
|
||||||
|
KEY_RIGHTSHIFT = 54
|
||||||
|
KEY_CAPSLOCK = 58
|
||||||
|
|
||||||
|
LINUX_TO_RL_KEY = {
|
||||||
|
1: rl.KEY_ESCAPE,
|
||||||
|
14: rl.KEY_BACKSPACE,
|
||||||
|
28: rl.KEY_ENTER,
|
||||||
|
96: rl.KEY_ENTER, # keypad enter
|
||||||
|
102: rl.KEY_HOME,
|
||||||
|
103: rl.KEY_UP,
|
||||||
|
105: rl.KEY_LEFT,
|
||||||
|
106: rl.KEY_RIGHT,
|
||||||
|
107: rl.KEY_END,
|
||||||
|
108: rl.KEY_DOWN,
|
||||||
|
111: rl.KEY_DELETE,
|
||||||
|
}
|
||||||
|
|
||||||
|
LETTER_BY_CODE = {
|
||||||
|
16: "q", 17: "w", 18: "e", 19: "r", 20: "t", 21: "y", 22: "u", 23: "i", 24: "o", 25: "p",
|
||||||
|
30: "a", 31: "s", 32: "d", 33: "f", 34: "g", 35: "h", 36: "j", 37: "k", 38: "l",
|
||||||
|
44: "z", 45: "x", 46: "c", 47: "v", 48: "b", 49: "n", 50: "m",
|
||||||
|
}
|
||||||
|
|
||||||
|
UNSHIFTED_BY_CODE = {
|
||||||
|
2: "1", 3: "2", 4: "3", 5: "4", 6: "5", 7: "6", 8: "7", 9: "8", 10: "9", 11: "0",
|
||||||
|
12: "-", 13: "=",
|
||||||
|
26: "[", 27: "]",
|
||||||
|
39: ";", 40: "'", 41: "`", 43: "\\",
|
||||||
|
51: ",", 52: ".", 53: "/",
|
||||||
|
57: " ",
|
||||||
|
}
|
||||||
|
|
||||||
|
SHIFTED_BY_CODE = {
|
||||||
|
2: "!", 3: "@", 4: "#", 5: "$", 6: "%", 7: "^", 8: "&", 9: "*", 10: "(", 11: ")",
|
||||||
|
12: "_", 13: "+",
|
||||||
|
26: "{", 27: "}",
|
||||||
|
39: ":", 40: "\"", 41: "~", 43: "|",
|
||||||
|
51: "<", 52: ">", 53: "?",
|
||||||
|
57: " ",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class USBKeyboard:
|
||||||
|
"""Fallback keyboard reader for platforms where raylib keyboard input is unavailable."""
|
||||||
|
def __init__(self):
|
||||||
|
self._fds: dict[str, int] = {}
|
||||||
|
self._last_scan_t = 0.0
|
||||||
|
self._left_shift_down = False
|
||||||
|
self._right_shift_down = False
|
||||||
|
self._caps_lock_on = False
|
||||||
|
self._key_queue: deque[int] = deque(maxlen=128)
|
||||||
|
self._char_queue: deque[int] = deque(maxlen=256)
|
||||||
|
atexit.register(self.close)
|
||||||
|
|
||||||
|
def _device_paths(self) -> list[str]:
|
||||||
|
paths = []
|
||||||
|
paths.extend(glob("/dev/input/by-id/*-kbd"))
|
||||||
|
paths.extend(glob("/dev/input/by-path/*-kbd"))
|
||||||
|
# Keep deterministic order and de-duplicate paths if both globs resolve to same target.
|
||||||
|
return sorted(set(paths))
|
||||||
|
|
||||||
|
def _refresh_devices(self) -> None:
|
||||||
|
now = time.monotonic()
|
||||||
|
if (now - self._last_scan_t) < 1.0:
|
||||||
|
return
|
||||||
|
|
||||||
|
self._last_scan_t = now
|
||||||
|
current_paths = set(self._device_paths())
|
||||||
|
|
||||||
|
for path in list(self._fds):
|
||||||
|
if path not in current_paths:
|
||||||
|
try:
|
||||||
|
os.close(self._fds[path])
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
del self._fds[path]
|
||||||
|
|
||||||
|
for path in current_paths:
|
||||||
|
if path in self._fds:
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
fd = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
|
||||||
|
except OSError:
|
||||||
|
continue
|
||||||
|
self._fds[path] = fd
|
||||||
|
|
||||||
|
def _key_to_char(self, code: int) -> str | None:
|
||||||
|
shift = self._left_shift_down or self._right_shift_down
|
||||||
|
|
||||||
|
if code in LETTER_BY_CODE:
|
||||||
|
c = LETTER_BY_CODE[code]
|
||||||
|
return c.upper() if (self._caps_lock_on ^ shift) else c
|
||||||
|
|
||||||
|
if code in UNSHIFTED_BY_CODE:
|
||||||
|
if shift and code in SHIFTED_BY_CODE:
|
||||||
|
return SHIFTED_BY_CODE[code]
|
||||||
|
return UNSHIFTED_BY_CODE[code]
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
def _process_event(self, event_type: int, code: int, value: int) -> None:
|
||||||
|
if event_type != EV_KEY:
|
||||||
|
return
|
||||||
|
|
||||||
|
if code == KEY_LEFTSHIFT:
|
||||||
|
self._left_shift_down = value != 0
|
||||||
|
return
|
||||||
|
if code == KEY_RIGHTSHIFT:
|
||||||
|
self._right_shift_down = value != 0
|
||||||
|
return
|
||||||
|
if code == KEY_CAPSLOCK and value == 1:
|
||||||
|
self._caps_lock_on = not self._caps_lock_on
|
||||||
|
return
|
||||||
|
|
||||||
|
# Linux key events: 0=release, 1=press, 2=repeat
|
||||||
|
if value not in (1, 2):
|
||||||
|
return
|
||||||
|
|
||||||
|
key = LINUX_TO_RL_KEY.get(code)
|
||||||
|
if key is not None:
|
||||||
|
self._key_queue.append(key)
|
||||||
|
|
||||||
|
char = self._key_to_char(code)
|
||||||
|
if char is not None:
|
||||||
|
self._char_queue.append(ord(char))
|
||||||
|
|
||||||
|
def _poll(self) -> None:
|
||||||
|
self._refresh_devices()
|
||||||
|
|
||||||
|
for path, fd in list(self._fds.items()):
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
data = os.read(fd, EVENT_SIZE * 32)
|
||||||
|
except BlockingIOError:
|
||||||
|
break
|
||||||
|
except OSError:
|
||||||
|
try:
|
||||||
|
os.close(fd)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
del self._fds[path]
|
||||||
|
break
|
||||||
|
|
||||||
|
if not data:
|
||||||
|
break
|
||||||
|
|
||||||
|
valid_bytes = len(data) - (len(data) % EVENT_SIZE)
|
||||||
|
for i in range(0, valid_bytes, EVENT_SIZE):
|
||||||
|
_, _, event_type, code, value = struct.unpack_from(EVENT_FORMAT, data, i)
|
||||||
|
self._process_event(event_type, code, value)
|
||||||
|
|
||||||
|
if len(data) < (EVENT_SIZE * 32):
|
||||||
|
break
|
||||||
|
|
||||||
|
def get_key_pressed(self) -> int:
|
||||||
|
self._poll()
|
||||||
|
return self._key_queue.popleft() if self._key_queue else 0
|
||||||
|
|
||||||
|
def get_char_pressed(self) -> int:
|
||||||
|
self._poll()
|
||||||
|
return self._char_queue.popleft() if self._char_queue else 0
|
||||||
|
|
||||||
|
def close(self) -> None:
|
||||||
|
for fd in self._fds.values():
|
||||||
|
try:
|
||||||
|
os.close(fd)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
self._fds.clear()
|
||||||
|
|
||||||
|
|
||||||
|
_usb_keyboard = USBKeyboard()
|
||||||
|
|
||||||
|
|
||||||
|
def get_usb_key_pressed() -> int:
|
||||||
|
return _usb_keyboard.get_key_pressed()
|
||||||
|
|
||||||
|
|
||||||
|
def get_usb_char_pressed() -> int:
|
||||||
|
return _usb_keyboard.get_char_pressed()
|
||||||
@@ -2,6 +2,7 @@ import pyray as rl
|
|||||||
import time
|
import time
|
||||||
from openpilot.system.ui.lib.application import gui_app, MousePos, FONT_SCALE
|
from openpilot.system.ui.lib.application import gui_app, MousePos, FONT_SCALE
|
||||||
from openpilot.system.ui.lib.text_measure import measure_text_cached
|
from openpilot.system.ui.lib.text_measure import measure_text_cached
|
||||||
|
from openpilot.system.ui.lib.usb_keyboard import get_usb_char_pressed, get_usb_key_pressed
|
||||||
from openpilot.system.ui.widgets import Widget
|
from openpilot.system.ui.widgets import Widget
|
||||||
|
|
||||||
PASSWORD_MASK_CHAR = "•"
|
PASSWORD_MASK_CHAR = "•"
|
||||||
@@ -25,6 +26,7 @@ class InputBox(Widget):
|
|||||||
self._visible_width = 0
|
self._visible_width = 0
|
||||||
self._last_char_time = 0 # Track when last character was added
|
self._last_char_time = 0 # Track when last character was added
|
||||||
self._masked_length = 0 # How many characters are currently masked
|
self._masked_length = 0 # How many characters are currently masked
|
||||||
|
self._submit_requested = False
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def text(self):
|
def text(self):
|
||||||
@@ -43,6 +45,12 @@ class InputBox(Widget):
|
|||||||
self._input_text = ''
|
self._input_text = ''
|
||||||
self._cursor_position = 0
|
self._cursor_position = 0
|
||||||
self._text_offset = 0
|
self._text_offset = 0
|
||||||
|
self._submit_requested = False
|
||||||
|
|
||||||
|
def consume_submit_requested(self) -> bool:
|
||||||
|
requested = self._submit_requested
|
||||||
|
self._submit_requested = False
|
||||||
|
return requested
|
||||||
|
|
||||||
def set_cursor_position(self, position):
|
def set_cursor_position(self, position):
|
||||||
"""Set the cursor position and reset the blink counter."""
|
"""Set the cursor position and reset the blink counter."""
|
||||||
@@ -191,6 +199,8 @@ class InputBox(Widget):
|
|||||||
def _handle_keyboard_input(self):
|
def _handle_keyboard_input(self):
|
||||||
# Handle navigation keys
|
# Handle navigation keys
|
||||||
key = rl.get_key_pressed()
|
key = rl.get_key_pressed()
|
||||||
|
if key == 0:
|
||||||
|
key = get_usb_key_pressed()
|
||||||
if key != 0:
|
if key != 0:
|
||||||
self._process_key(key)
|
self._process_key(key)
|
||||||
if key in (rl.KEY_LEFT, rl.KEY_RIGHT, rl.KEY_BACKSPACE, rl.KEY_DELETE):
|
if key in (rl.KEY_LEFT, rl.KEY_RIGHT, rl.KEY_BACKSPACE, rl.KEY_DELETE):
|
||||||
@@ -208,6 +218,8 @@ class InputBox(Widget):
|
|||||||
|
|
||||||
# Handle text input
|
# Handle text input
|
||||||
char = rl.get_char_pressed()
|
char = rl.get_char_pressed()
|
||||||
|
if char == 0:
|
||||||
|
char = get_usb_char_pressed()
|
||||||
if char != 0 and char >= 32: # Filter out control characters
|
if char != 0 and char >= 32: # Filter out control characters
|
||||||
self.add_char_at_cursor(chr(char))
|
self.add_char_at_cursor(chr(char))
|
||||||
|
|
||||||
@@ -226,3 +238,5 @@ class InputBox(Widget):
|
|||||||
self.set_cursor_position(0)
|
self.set_cursor_position(0)
|
||||||
elif key == rl.KEY_END:
|
elif key == rl.KEY_END:
|
||||||
self.set_cursor_position(len(self._input_text))
|
self.set_cursor_position(len(self._input_text))
|
||||||
|
elif key == rl.KEY_ENTER:
|
||||||
|
self._submit_requested = True
|
||||||
|
|||||||
@@ -145,6 +145,8 @@ class Keyboard(Widget):
|
|||||||
input_margin = 25
|
input_margin = 25
|
||||||
input_box_rect = rl.Rectangle(rect.x + input_margin, rect.y + 160, rect.width - input_margin, 100)
|
input_box_rect = rl.Rectangle(rect.x + input_margin, rect.y + 160, rect.width - input_margin, 100)
|
||||||
self._render_input_area(input_box_rect)
|
self._render_input_area(input_box_rect)
|
||||||
|
if self._input_box.consume_submit_requested() and len(self._input_box.text) >= self._min_text_size:
|
||||||
|
self._render_return_status = 1
|
||||||
|
|
||||||
# Process backspace key repeat if it's held down
|
# Process backspace key repeat if it's held down
|
||||||
if not self._all_keys[BACKSPACE_KEY].is_pressed:
|
if not self._all_keys[BACKSPACE_KEY].is_pressed:
|
||||||
|
|||||||
Reference in New Issue
Block a user