dora-teleop/tests/conftest.py

"""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
from dora_teleop.mapping import LAYOUT, TRIGGER_REST

# 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. Which indices those are comes from the
    # layout, so a test never has to be edited when a pad renumbers them.
    for index in (LAYOUT.left_trigger, LAYOUT.right_trigger):
        if axes is None or index >= len(axes):
            values[index] = -1.0
    return Sample(axes=values, buttons=list(buttons or []), present=present)


def axes_for(
    left_x=0.0,
    left_y=0.0,
    right_x=0.0,
    right_y=0.0,
    left_trigger=TRIGGER_REST,
    right_trigger=TRIGGER_REST,
):
    """Lay out a pad's axes by what each control is rather than by its index.

    A test that numbers its axes says nothing about which control it meant, and
    goes on passing while meaning something else once a pad renumbers them.
    """
    values = [0.0] * 6
    values[LAYOUT.left_x] = left_x
    values[LAYOUT.left_y] = left_y
    values[LAYOUT.right_x] = right_x
    values[LAYOUT.right_y] = right_y
    values[LAYOUT.left_trigger] = left_trigger
    values[LAYOUT.right_trigger] = right_trigger
    return values


def buttons_for(
    grasp=False, release=False, clear=False, freedrive=False, yaw_left=False, yaw_right=False, estop=False
):
    """Lay out a pad's buttons by what each one does rather than by its index.

    The same reason as :func:`axes_for`, and more pressing: an unpressed button
    reads identically on every pad, so a test that numbers them is the only
    place a wrong index could ever have been caught.
    """
    wanted = {
        LAYOUT.grasp: grasp,
        LAYOUT.release: release,
        LAYOUT.clear: clear,
        LAYOUT.freedrive: freedrive,
        LAYOUT.yaw_left: yaw_left,
        LAYOUT.yaw_right: yaw_right,
        LAYOUT.estop: estop,
    }
    return [wanted.get(index, False) for index in range(max(wanted) + 1)]


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",
        "JOG_KEEPALIVE_S",
        "JOY_TTL_S",
        "GAMEPAD_IDLE_RELEASE_S",
        "STATE_POLL_S",
        "WEB_HOST",
        "WEB_PORT",
    ):
        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