dora_teleop/ui.py

@@ -0,0 +1,170 @@
+"""Translation between the web console and the dataflow.
+
+The console was written against the monolithic repo's REST API and reads a
+telemetry snapshot by key. Those keys are the contract here: `app.js` does
+`t.owner === "web"` and `t.estop` directly, so renaming one breaks the page
+silently. They are reproduced rather than improved for that reason.
+
+The two vocabularies do not line up, which is most of what this file is for. The
+arm node reports `connecting|ready|moving|recovering|unavailable`; the console
+draws `idle|moving|jogging|freedrive|estop|stopped|disconnected`. Neither is
+wrong — one describes a driver's lifecycle, the other an operator's screen — and
+:func:`arm_state` is where the two are reconciled.
+"""
+
+from dataclasses import dataclass, field
+
+# What the arm node publishes on `robot_state`.
+CONNECTING = "connecting"
+READY = "ready"
+MOVING = "moving"
+RECOVERING = "recovering"
+UNAVAILABLE = "unavailable"
+
+# What the console draws.
+UI_IDLE = "idle"
+UI_MOVING = "moving"
+UI_JOGGING = "jogging"
+UI_FREEDRIVE = "freedrive"
+UI_ESTOP = "estop"
+UI_STOPPED = "stopped"
+UI_DISCONNECTED = "disconnected"
+
+# An arm the node cannot speak for is `disconnected` whatever the reason, because
+# that is the only thing an operator can act on. `recovering` is its own case:
+# the arm is there and answering, it just will not move yet.
+ARM_STATE = {
+    CONNECTING: UI_DISCONNECTED,
+    UNAVAILABLE: UI_DISCONNECTED,
+    RECOVERING: UI_STOPPED,
+    READY: UI_IDLE,
+    MOVING: UI_MOVING,
+}
+
+POSE_KEYS = ("x", "y", "z", "roll", "pitch", "yaw")
+
+
+@dataclass
+class Command:
+    """One thing the console asked for."""
+
+    type: str
+    velocity: list[float] = field(default_factory=list)
+    action: str = ""
+    preempt: bool = False
+    enabled: bool = False
+    row: dict = field(default_factory=dict)
+
+
+def command_of(payload: dict) -> Command | None:
+    """Read one console message, or None if it is not one we serve.
+
+    A message we do not recognise is dropped rather than refused. The console
+    and this node version independently, and a page from a newer build saying
+    something we have not learned yet must not take the arm down.
+    """
+    if not isinstance(payload, dict):
+        return None
+    kind = payload.get("type")
+    if not isinstance(kind, str):
+        return None
+
+    if kind == "jog":
+        velocity = payload.get("velocity") or []
+        if len(velocity) != 6 or not all(isinstance(v, (int, float)) for v in velocity):
+            return None
+        return Command(type="jog", velocity=[float(v) for v in velocity])
+    if kind in ("jog_stop", "estop", "reset", "home"):
+        return Command(type=kind)
+    if kind == "control":
+        return Command(
+            type="control",
+            action=str(payload.get("action", "")),
+            preempt=bool(payload.get("preempt", False)),
+        )
+    if kind == "gripper":
+        return Command(type="gripper", enabled=bool(payload.get("vacuum", False)))
+    if kind == "freedrive":
+        return Command(type="freedrive", enabled=bool(payload.get("enabled", False)))
+    if kind == "move":
+        row = {key: payload[key] for key in POSE_KEYS if isinstance(payload.get(key), (int, float))}
+        if not all(key in row for key in ("x", "y", "z")):
+            return None
+        if isinstance(payload.get("speed"), (int, float)):
+            row["speed"] = float(payload["speed"])
+        return Command(type="move", row=row)
+    return None
+
+
+@dataclass
+class ArmView:
+    """What this node currently believes about the arm.
+
+    Assembled from four places — `robot_state`, `tcp_pose`, `joint_pose` and the
+    answers to `get_state` — because no single one of them carries everything
+    the console draws. Whatever has not arrived yet stays at its default, so a
+    page that loads before the arm is reachable renders as disconnected rather
+    than as a stale pose.
+    """
+
+    state: str = CONNECTING
+    pose: list[float] = field(default_factory=list)
+    joints: list[float] = field(default_factory=list)
+    connected: bool = False
+    ready: bool = False
+    tool_engaged: bool = False
+    freedrive_on: bool = False
+    estop: bool = False
+
+    def take_get_state(self, row: dict) -> None:
+        """Absorb a `get_state` answer, which carries what telemetry does not."""
+        self.connected = bool(row.get("connected", False))
+        self.ready = bool(row.get("ready", False))
+        self.tool_engaged = bool(row.get("tool_engaged", False))
+        self.freedrive_on = bool(row.get("freedrive", False))
+
+
+def arm_state(view: ArmView, jogging: bool) -> str:
+    """Return the state the console should draw.
+
+    Order matters, and it is the order of what an operator most needs to know.
+    A latched e-stop wins over everything: the arm may well read `ready`, and
+    saying so would invite somebody to push it. Freedrive outranks idle because
+    an arm that can be pushed by hand is not the same as one holding station.
+    """
+    if view.estop:
+        return UI_ESTOP
+    if view.state in (CONNECTING, UNAVAILABLE):
+        return ARM_STATE[view.state]
+    if view.freedrive_on:
+        return UI_FREEDRIVE
+    if jogging and view.state == MOVING:
+        return UI_JOGGING
+    return ARM_STATE.get(view.state, UI_DISCONNECTED)
+
+
+def snapshot(view: ArmView, owner: str | None, jogging: bool, timestamp: float) -> dict:
+    """Build the telemetry object the console reads.
+
+    Every key here is read by name in `app.js`. `vacuum` is the console's name
+    for the mounted tool whatever it actually is, kept because renaming it would
+    mean editing the page for no gain.
+    """
+    return {
+        "type": "telemetry",
+        "ts": timestamp,
+        "connected": view.connected,
+        "ready": view.ready,
+        "arm_state": arm_state(view, jogging),
+        "owner": owner or "none",
+        "estop": view.estop,
+        "vacuum": view.tool_engaged,
+        "freedrive": view.freedrive_on,
+        "position": dict(zip(POSE_KEYS, view.pose, strict=False)) if view.pose else None,
+        "joints": list(view.joints),
+    }
+
+
+def error(message: str) -> dict:
+    """Build the message the console shows as a toast."""
+    return {"type": "error", "detail": message}

tests/test_ui.py

@@ -0,0 +1,138 @@
+"""Translation both ways: what the console asks for, and what it is shown."""
+
+from dora_teleop import ui
+from dora_teleop.arbiter import GAMEPAD, WEB
+
+# --- what the console asks for ---
+
+
+def test_a_jog_carries_its_six_numbers():
+    command = ui.command_of({"type": "jog", "velocity": [1, 2, 3, 4, 5, 6]})
+
+    assert command.type == "jog"
+    assert command.velocity == [1.0, 2.0, 3.0, 4.0, 5.0, 6.0]
+
+
+def test_a_jog_of_the_wrong_length_is_not_a_jog():
+    assert ui.command_of({"type": "jog", "velocity": [1, 2, 3]}) is None
+
+
+def test_a_jog_carrying_something_that_is_not_a_number_is_not_a_jog():
+    assert ui.command_of({"type": "jog", "velocity": [1, 2, 3, 4, 5, "fast"]}) is None
+
+
+def test_a_message_we_have_never_heard_of_is_dropped_not_refused():
+    # The console and this node version independently; a page from a newer build
+    # must not take the arm down by saying something we have not learned yet.
+    assert ui.command_of({"type": "hologram"}) is None
+    assert ui.command_of({}) is None
+    assert ui.command_of("not even an object") is None
+
+
+def test_taking_control_carries_whether_it_may_preempt():
+    command = ui.command_of({"type": "control", "action": "take", "preempt": True})
+
+    assert (command.action, command.preempt) == ("take", True)
+
+
+def test_the_gripper_button_says_which_way():
+    assert ui.command_of({"type": "gripper", "vacuum": True}).enabled is True
+    assert ui.command_of({"type": "gripper", "vacuum": False}).enabled is False
+
+
+def test_a_move_needs_a_position():
+    assert ui.command_of({"type": "move", "x": 1.0, "y": 2.0}) is None
+
+    command = ui.command_of({"type": "move", "x": 250.0, "y": 0.0, "z": 200.0, "speed": 40})
+
+    assert command.row == {"x": 250.0, "y": 0.0, "z": 200.0, "speed": 40.0}
+
+
+# --- what the console is shown ---
+
+
+def test_an_arm_that_cannot_be_reached_reads_as_disconnected():
+    view = ui.ArmView(state=ui.UNAVAILABLE)
+
+    assert ui.arm_state(view, jogging=False) == ui.UI_DISCONNECTED
+
+
+def test_an_arm_still_starting_up_reads_as_disconnected():
+    view = ui.ArmView(state=ui.CONNECTING)
+
+    assert ui.arm_state(view, jogging=False) == ui.UI_DISCONNECTED
+
+
+def test_a_latched_estop_outranks_everything_else():
+    # The arm may well read ready, and saying so would invite somebody to push it.
+    view = ui.ArmView(state=ui.READY, estop=True, freedrive_on=True)
+
+    assert ui.arm_state(view, jogging=True) == ui.UI_ESTOP
+
+
+def test_freedrive_outranks_idle():
+    # An arm that can be pushed by hand is not one holding station.
+    view = ui.ArmView(state=ui.READY, freedrive_on=True)
+
+    assert ui.arm_state(view, jogging=False) == ui.UI_FREEDRIVE
+
+
+def test_a_moving_arm_we_are_jogging_reads_as_jogging():
+    view = ui.ArmView(state=ui.MOVING)
+
+    assert ui.arm_state(view, jogging=True) == ui.UI_JOGGING
+    assert ui.arm_state(view, jogging=False) == ui.UI_MOVING
+
+
+def test_a_recovering_arm_is_there_but_will_not_move():
+    view = ui.ArmView(state=ui.RECOVERING)
+
+    assert ui.arm_state(view, jogging=False) == ui.UI_STOPPED
+
+
+def test_a_state_we_do_not_know_reads_as_disconnected_rather_than_as_ready():
+    view = ui.ArmView(state="dancing")
+
+    assert ui.arm_state(view, jogging=False) == ui.UI_DISCONNECTED
+
+
+def test_the_snapshot_carries_every_key_the_page_reads_by_name():
+    view = ui.ArmView(state=ui.READY, pose=[1, 2, 3, 4, 5, 6], joints=[0] * 6, connected=True)
+
+    snapshot = ui.snapshot(view, WEB, jogging=False, timestamp=123.0)
+
+    for key in (
+        "type",
+        "ts",
+        "connected",
+        "ready",
+        "arm_state",
+        "owner",
+        "estop",
+        "vacuum",
+        "freedrive",
+        "position",
+        "joints",
+    ):
+        assert key in snapshot
+    assert snapshot["position"] == {"x": 1, "y": 2, "z": 3, "roll": 4, "pitch": 5, "yaw": 6}
+
+
+def test_an_arm_with_no_pose_yet_reports_none_rather_than_a_stale_one():
+    snapshot = ui.snapshot(ui.ArmView(), None, jogging=False, timestamp=0.0)
+
+    assert snapshot["position"] is None
+    assert snapshot["owner"] == "none"
+
+
+def test_the_owner_is_named_so_the_page_can_gate_its_jog_pad():
+    assert ui.snapshot(ui.ArmView(), GAMEPAD, False, 0.0)["owner"] == "gamepad"
+    assert ui.snapshot(ui.ArmView(), WEB, False, 0.0)["owner"] == "web"
+
+
+def test_get_state_fills_in_what_telemetry_does_not_carry():
+    view = ui.ArmView()
+
+    view.take_get_state({"connected": True, "ready": True, "tool_engaged": True, "freedrive": True})
+
+    assert (view.connected, view.ready, view.tool_engaged, view.freedrive_on) == (True, True, True, True)