diff --git a/dora_teleop/ui.py b/dora_teleop/ui.py new file mode 100644 index 0000000..c780779 --- /dev/null +++ b/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} diff --git a/tests/test_ui.py b/tests/test_ui.py new file mode 100644 index 0000000..ee6c5c8 --- /dev/null +++ b/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)