read the gamepad from dora-joy and publish twist and command

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.

tests/test_deadman.py

@@ -8,8 +8,8 @@ import pytest
 from dora_teleop import main as loop
 from dora_teleop.arbiter import GAMEPAD, WEB
 from dora_teleop.config import Config
-from dora_teleop.gamepad import Sample
-from tests.conftest import FakeGamepad, RecordingNode, input_event, stick
+from dora_teleop.joy import Sample
+from tests.conftest import RecordingNode, input_event, joy_message, stick
 
 STOP = [0.0] * 6
 PUSHED = [0.0, -1.0]
@@ -34,19 +34,32 @@ def clock(monkeypatch):
 def build():
     config = Config.from_env()
     config.expo = 0.0
-    return RecordingNode(), loop.Teleop(config), FakeGamepad()
+    return RecordingNode(), loop.Teleop(config)
 
 
-def jog_from(node, state, gamepad, count=1):
-    """Run `count` ticks and return every velocity that reached the arm."""
+def send_joy(node, state, sample):
+    """Deliver one reading the way `dora-joy` publishes it, decoding included."""
+    loop._handle(node, state, input_event("joy", joy_message(sample)))
+
+
+def ticks(node, state, count=1):
+    """Run `count` ticks with nothing new arriving, and return every velocity sent."""
     for _ in range(count):
-        loop._tick(node, state, gamepad)
+        loop._tick(node, state)
     return node.jogs()
 
 
-def web_jog(node, state, velocity):
-    """Deliver one console jog message."""
-    payload = json.dumps({"type": "jog", "velocity": velocity})
+def drive(node, state, sample, count=1):
+    """Hold one reading for `count` ticks, the way a real pad and joy node would."""
+    for _ in range(count):
+        send_joy(node, state, sample)
+        loop._tick(node, state)
+    return node.jogs()
+
+
+def web_jog(node, state, axes):
+    """Deliver one console jog message, in the normalised units the page speaks."""
+    payload = json.dumps({"type": "jog", "axes": axes})
     loop._accept_ui_command(node, state, input_event("ui_command", pa.array([payload])))
 
 
@@ -55,10 +68,9 @@ def web_jog(node, state, velocity):
 
 def test_a_pushed_stick_takes_the_arm_and_drives_it(clock):
     # Nobody should have to reach for a browser to use the pad in their hands.
-    node, state, gamepad = build()
-    gamepad.sample = stick(axes=PUSHED)
+    node, state = build()
 
-    jogs = jog_from(node, state, gamepad)
+    jogs = drive(node, state, stick(axes=PUSHED))
 
     assert state.arbiter.owner == GAMEPAD
     assert jogs[-1][0] > 0
@@ -67,25 +79,49 @@ def test_a_pushed_stick_takes_the_arm_and_drives_it(clock):
 def test_a_gamepad_that_is_unplugged_mid_jog_stops_the_arm(clock):
     # An unplugged pad is not a centred pad, and waiting out the TTL here is the
     # difference between a dropped controller costing 10 ms of travel and 400.
-    node, state, gamepad = build()
-    gamepad.sample = stick(axes=PUSHED)
-    jog_from(node, state, gamepad)
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
     node.clear()
 
-    gamepad.sample = Sample(present=False)
-    jogs = jog_from(node, state, gamepad)
+    jogs = drive(node, state, Sample(present=False))
 
     assert jogs == [STOP]
 
 
-def test_a_stick_returning_to_centre_stops_the_arm(clock):
-    node, state, gamepad = build()
-    gamepad.sample = stick(axes=PUSHED)
-    jog_from(node, state, gamepad)
+def test_the_joy_node_dying_stops_the_arm(clock):
+    # The pad is another process now, and that process dying looks like silence
+    # rather than like an unplugged pad. It has to mean the same thing: a
+    # reading nobody is refreshing is not evidence that the stick is still held.
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
     node.clear()
 
-    gamepad.sample = stick()
-    jogs = jog_from(node, state, gamepad)
+    clock.advance(state.config.deadman.joy_ttl_s + 0.01)
+    jogs = ticks(node, state)
+
+    assert jogs == [STOP]
+
+
+def test_a_joy_node_that_keeps_publishing_keeps_the_arm_moving(clock):
+    # The mirror of the test above: time passing is not by itself a reason to
+    # stop, or a held stick would stutter.
+    node, state = build()
+
+    sent = []
+    for _ in range(4):
+        clock.advance(0.05)
+        sent = drive(node, state, stick(axes=PUSHED))
+
+    assert len(sent) == 4
+    assert all(jog[0] > 0 for jog in sent)
+
+
+def test_a_stick_returning_to_centre_stops_the_arm(clock):
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
+    node.clear()
+
+    jogs = drive(node, state, stick())
 
     assert jogs == [STOP]
 
@@ -93,13 +129,11 @@ def test_a_stick_returning_to_centre_stops_the_arm(clock):
 def test_a_stick_at_rest_sends_one_zero_not_a_stream(clock):
     # A stream of zeroes would keep the arm node believing a jog is live and
     # hold robot_state at `moving` forever.
-    node, state, gamepad = build()
-    gamepad.sample = stick(axes=PUSHED)
-    jog_from(node, state, gamepad)
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
     node.clear()
 
-    gamepad.sample = stick()
-    jogs = jog_from(node, state, gamepad, count=5)
+    jogs = drive(node, state, stick(), count=5)
 
     assert jogs == [STOP]
 
@@ -107,29 +141,39 @@ def test_a_stick_at_rest_sends_one_zero_not_a_stream(clock):
 def test_a_held_stick_is_resent_every_tick(clock):
     # The arm's controller expires a velocity on its own; the re-send is what
     # keeps that backstop from firing mid-motion.
-    node, state, gamepad = build()
-    gamepad.sample = stick(axes=PUSHED)
+    node, state = build()
 
-    jogs = jog_from(node, state, gamepad, count=4)
+    jogs = drive(node, state, stick(axes=PUSHED), count=4)
 
     assert len(jogs) == 4
     assert all(jog[0] > 0 for jog in jogs)
 
 
+def test_a_pad_that_was_never_wired_up_leaves_the_arm_alone(clock):
+    # `joy` is optional: a browser-only console wires no pad at all, and the
+    # node must not read that as a centred stick asking for the arm.
+    node, state = build()
+
+    jogs = ticks(node, state, count=3)
+
+    assert jogs == []
+    assert state.arbiter.owner is None
+
+
 # --- the browser ---
 
 
 def test_the_last_browser_leaving_stops_the_arm(clock):
     # Closing the tab has to stop the arm, and nothing is sent when it does.
-    node, state, gamepad = build()
+    node, state = build()
     state.clients = 1
     state.arbiter.take(WEB, clock.t)
-    web_jog(node, state, [10.0, 0, 0, 0, 0, 0])
-    jog_from(node, state, gamepad)
+    web_jog(node, state, [1.0, 0, 0, 0, 0, 0])
+    ticks(node, state)
     node.clear()
 
     loop._accept_clients(state, input_event("clients", pa.array([{"connected": 0, "event": "close"}])))
-    jogs = jog_from(node, state, gamepad)
+    jogs = ticks(node, state)
 
     assert jogs == [STOP]
     assert state.arbiter.owner is None
@@ -137,41 +181,41 @@ def test_the_last_browser_leaving_stops_the_arm(clock):
 
 def test_a_browser_that_goes_quiet_stops_the_arm(clock):
     # The console refreshes every 150 ms against a 400 ms deadline.
-    node, state, gamepad = build()
+    node, state = build()
     state.clients = 1
     state.arbiter.take(WEB, clock.t)
-    web_jog(node, state, [10.0, 0, 0, 0, 0, 0])
-    jog_from(node, state, gamepad)
+    web_jog(node, state, [1.0, 0, 0, 0, 0, 0])
+    ticks(node, state)
     node.clear()
 
     clock.advance(0.5)
-    jogs = jog_from(node, state, gamepad)
+    jogs = ticks(node, state)
 
     assert jogs == [STOP]
 
 
 def test_a_browser_that_keeps_refreshing_keeps_the_arm_moving(clock):
-    node, state, gamepad = build()
+    node, state = build()
     state.clients = 1
     state.arbiter.take(WEB, clock.t)
 
     sent = []
     for _ in range(4):
         clock.advance(0.15)
-        web_jog(node, state, [10.0, 0, 0, 0, 0, 0])
-        loop._tick(node, state, gamepad)
+        web_jog(node, state, [1.0, 0, 0, 0, 0, 0])
+        loop._tick(node, state)
         sent = node.jogs()
 
-    assert all(jog[0] == pytest.approx(10.0) for jog in sent)
+    assert all(jog[0] == pytest.approx(state.config.speed_mm_s) for jog in sent)
 
 
 def test_a_non_owner_jog_is_ignored(clock):
-    node, state, gamepad = build()
+    node, state = build()
     state.clients = 1
     state.arbiter.take(GAMEPAD, clock.t)
 
-    web_jog(node, state, [10.0, 0, 0, 0, 0, 0])
-    loop._tick(node, state, gamepad)
+    web_jog(node, state, [1.0, 0, 0, 0, 0, 0])
+    loop._tick(node, state)
 
     assert node.jogs() == []
     assert any(message.get("type") == "error" for message in node.ui())
@@ -183,7 +227,7 @@ def test_a_non_owner_jog_is_ignored(clock):
 def test_estop_is_never_gated_by_the_lease(clock):
     # Somebody reaching for the stop is the one moment when asking who owns the
     # arm is the wrong question.
-    node, state, _ = build()
+    node, state = build()
     state.clients = 1
     state.arbiter.take(GAMEPAD, clock.t)
 
@@ -194,18 +238,87 @@ def test_estop_is_never_gated_by_the_lease(clock):
     assert "stop" in node.services()
 
 
-def test_the_estop_button_works_without_the_pad_holding_the_arm(clock):
-    node, state, gamepad = build()
-    state.arbiter.take(WEB, clock.t)
-    gamepad.sample = stick(buttons=[False] * 6 + [True])
+def test_a_latched_estop_keeps_the_arm_still_under_a_pushed_stick(clock):
+    # A stop somebody had to reach for is not undone by the next thumb
+    # movement. Without this the pad stops the arm for one tick and drives it
+    # again on the next, which is not a stop at all.
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
+    node.clear()
 
-    loop._tick(node, state, gamepad)
+    drive(node, state, stick(axes=PUSHED, buttons=[False] * 6 + [True]))
+    node.clear()
+    jogs = drive(node, state, stick(axes=PUSHED), count=4)
+
+    assert jogs == []
+
+
+def test_the_pad_can_release_the_latch_it_set(clock):
+    # A pad-only dataflow has no console RESET, so a latch nothing on the pad
+    # can release would strand the arm until the process was restarted.
+    node, state = build()
+    drive(node, state, stick(buttons=[False] * 6 + [True]))
+    assert state.view.estop is True
+    node.clear()
+
+    drive(node, state, stick(buttons=[False, False, True]))
+
+    assert state.view.estop is False
+    assert "clear" in node.services()
+
+
+def test_the_latch_is_released_without_holding_the_arm(clock):
+    # An idle pad hands the lease back two seconds after the stop it just made,
+    # so gating the release on the lease would be a trap.
+    node, state = build()
+    drive(node, state, stick(buttons=[False] * 6 + [True]))
+    state.arbiter.release(GAMEPAD)
+    state.arbiter.take(WEB, clock.t)
+    node.clear()
+
+    drive(node, state, stick(buttons=[False, False, True]))
+
+    assert state.view.estop is False
+
+
+def test_the_arm_drives_again_once_the_latch_is_released(clock):
+    node, state = build()
+    drive(node, state, stick(buttons=[False] * 6 + [True]))
+    drive(node, state, stick(buttons=[False, False, True]))
+    node.clear()
+
+    jogs = drive(node, state, stick(axes=PUSHED))
+
+    assert jogs[-1][0] > 0
+
+
+def test_a_latched_estop_stops_the_browser_too(clock):
+    # The latch is the node's, not the pad's. A console jog must not walk
+    # through a stop somebody made on the gamepad.
+    node, state = build()
+    state.clients = 1
+    drive(node, state, stick(buttons=[False] * 6 + [True]))
+    state.arbiter.release(GAMEPAD)
+    state.arbiter.take(WEB, clock.t)
+    node.clear()
+
+    web_jog(node, state, [1.0, 0, 0, 0, 0, 0])
+    jogs = ticks(node, state)
+
+    assert jogs == []
+
+
+def test_the_estop_button_works_without_the_pad_holding_the_arm(clock):
+    node, state = build()
+    state.arbiter.take(WEB, clock.t)
+
+    drive(node, state, stick(buttons=[False] * 6 + [True]))
 
     assert "stop" in node.services()
 
 
 def test_a_latched_estop_is_cleared_by_reset(clock):
-    node, state, _ = build()
+    node, state = build()
     state.view.estop = True
 
     payload = json.dumps({"type": "reset"})
@@ -213,3 +326,82 @@ def test_a_latched_estop_is_cleared_by_reset(clock):
 
     assert state.view.estop is False
     assert "clear" in node.services()
+
+
+# --- the standard ports ---
+
+
+def test_the_same_deflection_from_either_source_asks_for_the_same_motion(clock):
+    # The whole reason both sources are normalised before they are scaled. If
+    # these ever differ, one path has grown a conversion the other has not —
+    # which is how a console ends up in the robot's units.
+    node, state = build()
+    drive(node, state, stick(axes=[0.0, -1.0]))
+    from_pad = node.twists()[-1]
+
+    node, state = build()
+    state.clients = 1
+    state.arbiter.take(WEB, clock.t)
+    web_jog(node, state, [1.0, 0, 0, 0, 0, 0])
+    ticks(node, state)
+    from_web = node.twists()[-1]
+
+    assert from_pad == from_web
+
+
+def test_the_twist_is_si_and_the_legacy_jog_is_not(clock):
+    # 60 mm/s is 0.06 m/s. A consumer reading the wrong port at the wrong scale
+    # is a thousandfold error, so this is worth pinning.
+    node, state = build()
+    drive(node, state, stick(axes=[0.0, -1.0]))
+
+    twist, jog = node.twists()[-1], node.jogs()[-1]
+
+    assert twist["linear"][0] == pytest.approx(state.config.speed_mm_s / 1000.0)
+    assert jog[0] == pytest.approx(state.config.speed_mm_s)
+    assert twist["frame"] == "base"
+
+
+def test_a_stop_reaches_both_ports(clock):
+    node, state = build()
+    drive(node, state, stick(axes=PUSHED))
+    node.clear()
+
+    drive(node, state, stick())
+
+    assert node.jogs() == [STOP]
+    assert node.twists() == [{"linear": [0.0] * 3, "angular": [0.0] * 3, "frame": "base"}]
+
+
+def test_a_button_asks_in_verbs_as_well_as_in_services(clock):
+    # The verb is what a robot speaking the standard reads; the service name is
+    # what dora-ulite6 reads until it does.
+    node, state = build()
+
+    drive(node, state, stick(buttons=[True]))
+
+    assert node.actions() == ["gripper_close"]
+    assert "grasp" in node.services()
+
+
+def test_every_verb_the_pad_can_ask_for_is_robot_agnostic(clock):
+    node, state = build()
+    drive(node, state, stick(buttons=[False, True]))
+    drive(node, state, stick(buttons=[False] * 3 + [True]))
+    drive(node, state, stick(buttons=[False] * 6 + [True]))
+    drive(node, state, stick(buttons=[False, False, True]))
+
+    assert node.actions() == ["gripper_open", "freedrive", "stop", "clear"]
+
+
+def test_the_first_press_after_the_pad_has_been_idle_is_not_dropped(clock):
+    # The lease has to be settled before the buttons are read. Read them first
+    # and the press arrives while nobody owns the arm, and by the next tick the
+    # button is held rather than newly down — so the edge is gone for good and
+    # the operator presses A and watches nothing happen.
+    node, state = build()
+    assert state.arbiter.owner is None
+
+    drive(node, state, stick(buttons=[True]), count=3)
+
+    assert node.actions() == ["gripper_close"]