add dora-joy node publishing gamepad axes and buttons

dora_joy/gamepad.py

@@ -0,0 +1,121 @@
+"""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