add the gamepad reader and the node's settings

The reader owns the pad on a thread of its own, and nothing outside that file
may touch pygame: everything else sees a plain immutable Sample. That is what
lets the rest of the node be tested with no hardware, and what keeps SDL's
quirks out of the arm logic.

Sample.present is the field that matters. A pad that has been unplugged reads as
absent rather than as centred, so "the operator let go" and "the pad fell off
the desk" stay distinguishable — they call for the same stop, but only one of
them is worth logging.

Only the display and joystick subsystems are initialised. pygame.init() also
opens audio, which on a headless robot fills stderr with ALSA errors for a
device this node never uses; the display is needed despite there being no
window, because SDL refreshes joystick state during the event pump.

Settings are parsed once at startup so a deadzone that would silently make the
pad inert stops the node before it joins the dataflow.
This commit is contained in:
2026-09-15 15:42:48 -05:00
parent 253df68cd4
commit bde8582de3
10 changed files with 910 additions and 0 deletions
+96
View File
@@ -0,0 +1,96 @@
"""Shared fakes. Nothing here needs a gamepad, an arm, a browser or a daemon."""
import json
import pytest
from dora_teleop.config import Config
from dora_teleop.gamepad import Sample
class RecordingNode:
"""A dora node that records outputs *with* their metadata.
`dora.testing.MockNode` drops the metadata dict on `send_output`, and the
metadata is where `request_id` rides. So the tests record it themselves.
"""
def __init__(self):
self.sent = []
def send_output(self, output_id, data, metadata=None):
self.sent.append((output_id, data, dict(metadata or {})))
def outputs(self, output_id):
"""Return every message sent on ``output_id`` as (value, metadata) pairs."""
return [(data[0].as_py(), metadata) for name, data, metadata in self.sent if name == output_id]
def jogs(self):
"""Return every velocity sent to the arm, in order."""
return [row["velocity"] for row, _ in self.outputs("jog")]
def services(self):
"""Return the name of every service called, in order."""
return [row["service"] for row, _ in self.outputs("request")]
def ui(self):
"""Return every object handed to the bridge, parsed."""
return [json.loads(value) for value, _ in self.outputs("ui_state")]
def clear(self):
self.sent.clear()
class FakeGamepad:
"""Stands in for the reader thread: hands back whatever the test set."""
def __init__(self, sample=None):
self.sample = sample or Sample()
self.stopped = False
def read(self):
return self.sample
def shutdown(self):
self.stopped = True
def stick(axes=None, buttons=None, present=True):
"""Build one gamepad reading, padded to a full six axes."""
values = list(axes or [])
values += [0.0] * (6 - len(values))
# A trigger at rest reads -1, and a test that forgot to say so would look
# like both triggers held down.
for index in (2, 5):
if axes is None or index >= len(axes):
values[index] = -1.0
return Sample(axes=values, buttons=list(buttons or []), present=present)
def input_event(event_id, value, metadata=None):
"""Build the event dict dora hands a node for one input."""
return {"type": "INPUT", "id": event_id, "value": value, "metadata": dict(metadata or {})}
@pytest.fixture(autouse=True)
def clean_env(monkeypatch):
"""Run every test against a bare environment, whatever the shell set."""
for name in (
"SPEED_MM_S",
"SPEED_DEG_S",
"DEADZONE",
"EXPO",
"JOYSTICK_INDEX",
"JOG_TTL_S",
"GAMEPAD_IDLE_RELEASE_S",
"STATE_POLL_S",
):
monkeypatch.delenv(name, raising=False)
@pytest.fixture
def config():
"""Return the default configuration, with expo off so a test can do the arithmetic."""
built = Config.from_env()
built.expo = 0.0
return built