Coverage for src/ph/gamepad.py: 84.8%
166 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-28 08:17 +0000
1"""Direct Linux joystick input for handheld controls."""
3import fcntl
4import os
5import struct
6from collections import deque
7from dataclasses import dataclass, field
8from enum import Enum
9from pathlib import Path
10from time import monotonic
13class InputAction(Enum):
14 """Logical actions understood by the terminal UI."""
16 UP = "up"
17 DOWN = "down"
18 LEFT = "left"
19 RIGHT = "right"
20 SELECT = "select"
21 BACK = "back"
22 BACKSPACE = "backspace"
23 SUBMIT_SEARCH = "submit_search"
24 PAGE_UP = "page_up"
25 PAGE_DOWN = "page_down"
26 START = "start"
29JS_EVENT_BUTTON = 0x01
30JS_EVENT_AXIS = 0x02
31JS_EVENT_INIT = 0x80
32AXIS_THRESHOLD = 16_000
33EVENT_STRUCT = struct.Struct("IhBB")
34REPEAT_DELAY_SECONDS = 0.35
35REPEAT_INTERVAL_SECONDS = 0.08
36REPEAT_ACTIONS = frozenset((InputAction.UP, InputAction.DOWN))
38# Semantic Linux input-event codes returned for joydev button indexes.
39KEY_UP = 103
40KEY_LEFT = 105
41KEY_RIGHT = 106
42KEY_DOWN = 108
43BTN_SOUTH = 0x130
44BTN_EAST = 0x131
45BTN_NORTH = 0x133
46BTN_WEST = 0x134
47BTN_TL = 0x136
48BTN_TR = 0x137
49BTN_SELECT = 0x13A
50BTN_START = 0x13B
51BTN_DPAD_UP = 0x220
52BTN_DPAD_DOWN = 0x221
53BTN_DPAD_LEFT = 0x222
54BTN_DPAD_RIGHT = 0x223
56# Values generated by Linux's _IOR macro for linux/joystick.h.
57JSIOCGBUTTONS = 0x80016A12
58JSIOCGBTNMAP = 0x84006A34
59BUTTON_MAP_BYTES = 1024
60BUTTON_CODE_ACTIONS: dict[int, InputAction] = {
61 KEY_UP: InputAction.UP,
62 KEY_DOWN: InputAction.DOWN,
63 KEY_LEFT: InputAction.LEFT,
64 KEY_RIGHT: InputAction.RIGHT,
65 BTN_SOUTH: InputAction.BACK,
66 BTN_EAST: InputAction.SELECT,
67 BTN_NORTH: InputAction.SUBMIT_SEARCH,
68 BTN_WEST: InputAction.BACKSPACE,
69 BTN_TL: InputAction.PAGE_UP,
70 BTN_TR: InputAction.PAGE_DOWN,
71 BTN_SELECT: InputAction.BACK,
72 BTN_START: InputAction.START,
73 BTN_DPAD_UP: InputAction.UP,
74 BTN_DPAD_DOWN: InputAction.DOWN,
75 BTN_DPAD_LEFT: InputAction.LEFT,
76 BTN_DPAD_RIGHT: InputAction.RIGHT,
77}
79BUTTON_ACTIONS: dict[int, InputAction] = {
80 0: InputAction.BACK,
81 1: InputAction.SELECT,
82 2: InputAction.SUBMIT_SEARCH,
83 3: InputAction.BACKSPACE,
84 4: InputAction.PAGE_UP,
85 5: InputAction.PAGE_DOWN,
86 # Common images expose either Select/Start pair; accept both joydev layouts.
87 6: InputAction.BACK,
88 7: InputAction.START,
89 8: InputAction.BACK,
90 9: InputAction.START,
91}
92STANDARD_DPAD_BUTTONS: dict[int, InputAction] = {
93 11: InputAction.UP,
94 12: InputAction.DOWN,
95 13: InputAction.LEFT,
96 14: InputAction.RIGHT,
97}
98ALTERNATE_DPAD_BUTTONS: dict[int, InputAction] = {
99 14: InputAction.UP,
100 15: InputAction.DOWN,
101 16: InputAction.LEFT,
102 17: InputAction.RIGHT,
103}
104# Linux joydev commonly exposes X/Y on 0/1, Z or RX on 2/3, RY on 4, and the
105# D-pad hat on 6/7. This also covers device-tree variants where the right
106# stick is reported as RX/RY rather than the older Z/RX pairing.
107HORIZONTAL_AXES = {0, 2, 6}
108VERTICAL_AXES = {1, 3, 4, 7}
111@dataclass(slots=True)
112class LinuxJoystick:
113 """Non-blocking reader for Linux's stable ``/dev/input/js*`` interface."""
115 path: Path
116 _file_descriptor: int
117 _axis_state: dict[int, int] = field(default_factory=dict)
118 _pending: deque[InputAction] = field(default_factory=deque)
119 _repeat_action: InputAction | None = None
120 _repeat_due_at: float = 0.0
121 _highest_initialized_button: int = -1
122 _button_codes: tuple[int, ...] = ()
124 @classmethod
125 def open_first(cls, input_directory: Path = Path("/dev/input")) -> LinuxJoystick | None:
126 """Open the first readable built-in or connected Linux joystick."""
128 for path in sorted(input_directory.glob("js*")):
129 try:
130 descriptor = os.open(path, os.O_RDONLY | os.O_NONBLOCK)
131 except OSError:
132 continue
133 button_codes = cls._read_button_codes(descriptor)
134 return cls(
135 path=path,
136 _file_descriptor=descriptor,
137 _button_codes=button_codes,
138 )
139 return None
141 @staticmethod
142 def _read_button_codes(descriptor: int) -> tuple[int, ...]:
143 """Read semantic Linux key codes for every joydev button index."""
145 try:
146 count_buffer = bytearray(1)
147 fcntl.ioctl(descriptor, JSIOCGBUTTONS, count_buffer, True)
148 count = count_buffer[0]
149 mapping_buffer = bytearray(BUTTON_MAP_BYTES)
150 fcntl.ioctl(descriptor, JSIOCGBTNMAP, mapping_buffer, True)
151 except OSError:
152 return ()
153 return struct.unpack_from(f"={count}H", mapping_buffer)
155 def poll(self) -> InputAction | None:
156 """Return the next available action without blocking the TUI."""
158 if self._pending: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 return self._pending.popleft()
160 try:
161 payload = os.read(self._file_descriptor, EVENT_STRUCT.size * 16)
162 except BlockingIOError:
163 return self._poll_repeat()
164 except OSError:
165 self.close()
166 return None
168 for offset in range(0, len(payload) - EVENT_STRUCT.size + 1, EVENT_STRUCT.size): 168 ↛ 169line 168 didn't jump to line 169 because the loop on line 168 never started
169 _timestamp, value, event_type, number = EVENT_STRUCT.unpack_from(payload, offset)
170 action = self.decode_event(value, event_type, number)
171 if action is not None:
172 self._pending.append(action)
173 return self._pending.popleft() if self._pending else self._poll_repeat()
175 def decode_event(self, value: int, event_type: int, number: int) -> InputAction | None:
176 """Translate one js event; public to allow device-independent tests."""
178 kind = event_type & ~JS_EVENT_INIT
179 if event_type & JS_EVENT_INIT:
180 if kind == JS_EVENT_BUTTON: 180 ↛ 182line 180 didn't jump to line 182 because the condition on line 180 was always true
181 self._highest_initialized_button = max(self._highest_initialized_button, number)
182 return None
183 if kind == JS_EVENT_BUTTON:
184 action = self._button_action(number)
185 if action not in REPEAT_ACTIONS:
186 return action if value else None
187 if value:
188 self._start_repeat(action)
189 return action
190 self._stop_repeat(action)
191 return None
192 if kind != JS_EVENT_AXIS: 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true
193 return None
195 direction = -1 if value < -AXIS_THRESHOLD else 1 if value > AXIS_THRESHOLD else 0
196 previous = self._axis_state.get(number, 0)
197 self._axis_state[number] = direction
198 previous_action = self._axis_action(number, previous)
199 if direction == 0:
200 self._stop_repeat(previous_action)
201 return None
202 if direction == previous:
203 return None
204 action = self._axis_action(number, direction)
205 self._stop_repeat(previous_action)
206 if action in REPEAT_ACTIONS:
207 self._start_repeat(action)
208 return action
210 def _button_action(self, number: int) -> InputAction | None:
211 if number < len(self._button_codes):
212 code_action = BUTTON_CODE_ACTIONS.get(self._button_codes[number])
213 if code_action is not None: 213 ↛ 215line 213 didn't jump to line 215 because the condition on line 213 was always true
214 return code_action
215 action = BUTTON_ACTIONS.get(number)
216 if action is not None:
217 return action
218 dpad = (
219 ALTERNATE_DPAD_BUTTONS
220 if self._highest_initialized_button < 11 or self._highest_initialized_button > 14
221 else STANDARD_DPAD_BUTTONS
222 )
223 return dpad.get(number)
225 @staticmethod
226 def _axis_action(number: int, direction: int) -> InputAction | None:
227 if not direction:
228 return None
229 if number in HORIZONTAL_AXES:
230 return InputAction.LEFT if direction < 0 else InputAction.RIGHT
231 if number in VERTICAL_AXES:
232 return InputAction.UP if direction < 0 else InputAction.DOWN
233 return None
235 def _start_repeat(self, action: InputAction) -> None:
236 self._repeat_action = action
237 self._repeat_due_at = monotonic() + REPEAT_DELAY_SECONDS
239 def _stop_repeat(self, action: InputAction | None) -> None:
240 if action is not None and action == self._repeat_action:
241 self._repeat_action = None
242 self._repeat_due_at = 0.0
244 def _poll_repeat(self) -> InputAction | None:
245 if self._repeat_action is None or monotonic() < self._repeat_due_at:
246 return None
247 self._repeat_due_at = monotonic() + REPEAT_INTERVAL_SECONDS
248 return self._repeat_action
250 def close(self) -> None:
251 """Release the device descriptor, tolerating repeated calls."""
253 if self._file_descriptor < 0: 253 ↛ 254line 253 didn't jump to line 254 because the condition on line 253 was never true
254 return
255 try:
256 os.close(self._file_descriptor)
257 finally:
258 self._file_descriptor = -1