cristhian
d336f8d499
The pad moves to dora-joy, so this node keeps no hardware dependency and the pad becomes reusable. It arrives on `joy` as a plain reading; `JoyInput.read` treats silence past JOY_TTL_S as absence, because that node dying looks like silence rather than like an unplugged pad and the arm cannot tell them apart. Both control sources are now normalised to a [-1, 1] six-vector before anything is scaled, and the scaling happens in one function. The web path used to pass a velocity straight through without going near mapping, which is why the console had to speak the arm's units. It no longer does. `twist` (SI) and `command` (robot-agnostic verbs) are the standard ports. `jog` and `request` still carry the same intent in dora-ulite6's units and service names so the two repositories can move independently; LEGACY_REQUEST is the only place that knows what that node calls a verb. Wire one or the other, never both. `joint_jog` is specified in the manifest and deliberately not built. Schemas are declared rather than inferred: pyarrow reads a null `value` as its own type, so an inferred `command` would change its port's type between one verb and the next. Three bugs found while moving this code, each with a test: - `Sample.active` compared raw axes against a threshold, but SDL rests a trigger at -1.0, so every connected pad read as in use for ever: the pad took the lease on plug-in and GAMEPAD_IDLE_RELEASE_S never fired. - The e-stop latch was never consulted when deciding a velocity, so Back stopped the arm for one tick and the next stick nudge drove it again. Nothing on the pad could clear it either. It is now checked above the lease, and X releases it and clears the controller's error, as the console's RESET does. - Buttons were read before the arbiter granted the lease, so the first press after the pad had been idle was swallowed: not owned yet when read, and no longer a rising edge by the next tick.
117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""Shared fakes. Nothing here needs a gamepad, an arm, a browser or a daemon."""
|
|
|
|
import json
|
|
|
|
import pyarrow as pa
|
|
import pytest
|
|
|
|
from dora_teleop.config import Config
|
|
from dora_teleop.joy import Sample
|
|
|
|
# The shape `dora-joy` publishes. Declared here rather than inferred for the same
|
|
# reason it is declared there: pyarrow reads an empty `axes` list as `list<null>`,
|
|
# so a test for an absent pad would otherwise build a message no consumer sees.
|
|
JOY = pa.struct(
|
|
[
|
|
("axes", pa.list_(pa.float32())),
|
|
("buttons", pa.list_(pa.bool_())),
|
|
("present", pa.bool_()),
|
|
]
|
|
)
|
|
|
|
|
|
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 on the deprecated `jog` port, in order."""
|
|
return [row["velocity"] for row, _ in self.outputs("jog")]
|
|
|
|
def twists(self):
|
|
"""Return every Twist sent on the standard port, in order."""
|
|
return [row for row, _ in self.outputs("twist")]
|
|
|
|
def actions(self):
|
|
"""Return the verb of every standard `command`, in order."""
|
|
return [row["action"] for row, _ in self.outputs("command")]
|
|
|
|
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()
|
|
|
|
|
|
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 joy_message(sample):
|
|
"""Build the Arrow row `dora-joy` would publish for one reading."""
|
|
return pa.array(
|
|
[
|
|
{
|
|
"axes": [float(value) for value in sample.axes],
|
|
"buttons": [bool(down) for down in sample.buttons],
|
|
"present": sample.present,
|
|
}
|
|
],
|
|
type=JOY,
|
|
)
|
|
|
|
|
|
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",
|
|
"JOG_TTL_S",
|
|
"JOY_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
|