122 lines
4.4 KiB
Python
122 lines
4.4 KiB
Python
"""The gamepad, and the thread that reads it.
|
|
|
|
**Nothing outside this file may touch pygame.** The rest of the node sees only
|
|
:class:`Sample`, which is a plain immutable reading — that is what lets every
|
|
other module be tested without a pad plugged in, and what keeps SDL's quirks
|
|
from leaking past this node's edge.
|
|
|
|
The reader is a thread because SDL wants to be pumped at its own cadence and
|
|
because a pad being unplugged should be noticed within milliseconds rather than
|
|
on whenever the dataflow next ticks. It never publishes anything; it fills a
|
|
slot, and the event loop drains it.
|
|
"""
|
|
|
|
import os
|
|
import threading
|
|
from dataclasses import dataclass, field
|
|
|
|
from dora_joy.log import log
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Sample:
|
|
"""One reading of the pad.
|
|
|
|
``present`` is the one that matters for safety: a pad that has been
|
|
unplugged reads as absent rather than as centred, so the difference between
|
|
"the operator let go" and "the operator's pad fell off the desk" is not lost.
|
|
|
|
Nothing here is shaped, scaled or filtered. A deadzone belongs to whoever is
|
|
deciding what the pad *means*, which is not this node.
|
|
"""
|
|
|
|
axes: list[float] = field(default_factory=list)
|
|
buttons: list[bool] = field(default_factory=list)
|
|
present: bool = False
|
|
|
|
|
|
ABSENT = Sample()
|
|
|
|
|
|
class Gamepad:
|
|
"""Reads one pad on a thread and keeps the latest reading available."""
|
|
|
|
def __init__(self, index: int, poll_interval_s: float = 0.01):
|
|
self.index = index
|
|
self.poll_interval_s = poll_interval_s
|
|
self._lock = threading.Lock()
|
|
self._sample = ABSENT
|
|
self._stopping = threading.Event()
|
|
self._thread = threading.Thread(target=self._run, name="dora-joy", daemon=True)
|
|
|
|
def start(self) -> None:
|
|
"""Start reading. Never blocks, and never fails for want of a pad."""
|
|
self._thread.start()
|
|
|
|
def read(self) -> Sample:
|
|
"""Return the most recent reading."""
|
|
with self._lock:
|
|
return self._sample
|
|
|
|
def shutdown(self) -> None:
|
|
"""Stop reading and let the thread finish."""
|
|
self._stopping.set()
|
|
self._thread.join(timeout=1.0)
|
|
|
|
def _run(self) -> None:
|
|
"""Open the pad, read it, and keep trying if it is not there.
|
|
|
|
A missing pad is an ordinary state, not an error. This node publishes
|
|
`present: false` and keeps looking, so a consumer learns that the pad
|
|
went away rather than that the dataflow did.
|
|
"""
|
|
# Two subsystems, not all of them. `pygame.init()` also opens audio,
|
|
# which on a headless robot means a screenful of ALSA errors in the
|
|
# dataflow's stderr for a device this node never uses. The display is
|
|
# needed despite there being no window: SDL refreshes joystick state
|
|
# during the event pump, and the pump needs the video subsystem.
|
|
os.environ.setdefault("SDL_VIDEODRIVER", "dummy")
|
|
os.environ.setdefault("SDL_AUDIODRIVER", "dummy")
|
|
import pygame
|
|
|
|
pygame.display.init()
|
|
pygame.joystick.init()
|
|
joystick = None
|
|
|
|
while not self._stopping.is_set():
|
|
pygame.event.pump()
|
|
if joystick is None:
|
|
joystick = self._open(pygame)
|
|
if joystick is None:
|
|
self._store(ABSENT)
|
|
self._stopping.wait(1.0)
|
|
continue
|
|
try:
|
|
sample = Sample(
|
|
axes=[joystick.get_axis(i) for i in range(joystick.get_numaxes())],
|
|
buttons=[bool(joystick.get_button(i)) for i in range(joystick.get_numbuttons())],
|
|
present=True,
|
|
)
|
|
except Exception as error: # noqa: BLE001 — an unplugged pad raises from inside SDL
|
|
log(f"the gamepad went away ({error})")
|
|
joystick = None
|
|
self._store(ABSENT)
|
|
continue
|
|
self._store(sample)
|
|
self._stopping.wait(self.poll_interval_s)
|
|
|
|
def _open(self, pygame):
|
|
"""Open the configured pad, or return None if it is not there."""
|
|
pygame.joystick.quit()
|
|
pygame.joystick.init()
|
|
if pygame.joystick.get_count() <= self.index:
|
|
return None
|
|
joystick = pygame.joystick.Joystick(self.index)
|
|
joystick.init()
|
|
log(f"gamepad {self.index}: {joystick.get_name()}")
|
|
return joystick
|
|
|
|
def _store(self, sample: Sample) -> None:
|
|
with self._lock:
|
|
self._sample = sample
|