wip initial wip coded codex integration of the M850 controller and motor class

csaxs_bec/device_configs/tmp.md

@@ -0,0 +1,57 @@
+# Falcon integration
+Discussion about the integration of the Falcon detector at different beamlines.
+
+## cSAXS
+All PVs are set up when the detector is being connected to BEC. At least the ones that were found relevant at the time of the integration. The detector is operated in MAPPING mode, with the pixel advance mode set to GATE. The number of pixels per buffer is set to 20 with a maximum queue size of 2000 on the hdf5 side. 
+
+During the beamtime in 2024, this IOC was running on a console and we were able to measure at roughly ~50Hz. Any faster acquisition meant that we loose data.
+
+cSAXS might like to operate the detector at much higher frequencies.
+
+
+``` python
+# Core Settings
+self.preset_mode.put(1)# 0 Normal, 1 Inverted
+self.input_logic_polarity.put(0)# 0 Manual 1 Auto
+self.auto_pixels_per_buffer.put(0)# Sets the number of pixels/spectra in the buffer
+self.pixels_per_buffer.put(20)
+
+
+## HDF5 settings
+self.hdf5.enable.put(1)
+self.hdf5.xml_file_name.put("layout.xml") # csaxs file layout
+self.hdf5.lazy_open.put(1)
+self.hdf5.temp_suffix.put("")
+self.hdf5.queue_size.put(2000)
+self.hdf5.file_template.put("%s%s")
+self.hdf5.file_write_mode.put(2)
+self.nd_array_mode.put(1) # Segment spectra in Epics =1, 0 no segmentation
+
+## Acquisition settings
+self.collect_mode.put(1) # SPECTRUM = 0 MAPPING = 1
+self.pixel_advance_mode.put(1) # USER = 0 GATE = 1 SYNC = 2
+self.ignore_gate.put(0) # 0 False, 1 True
+```
+
+## SuperXAS
+At the SuperXAS beamline, the Falcon detector is operated in SPECTRUM mode and MAPPING mode. However, so far we have only worked with the SPECTRUM mode. Different to cSAXS, settings of the detector IOC are loaded through clicking a configuration icon on a desktop. The only settings that are set are
+
+``` python
+self.collect_mode.set(0).wait()
+self.preset_real_time.set(0).wait()
+
+# The detector is also gated, I believe. And we then read out the following relevant channels
+"""Get dead time corrected counts base on signals from dxp and mca of Falcon"""
+dxp: EpicsDXPFalcon = getattr(self.parent, f"dxp{self._channel}")
+mca: EpicsMCARecord = getattr(self.parent, f"mca{self._channel}")
+
+icr = dxp.input_count_rate.get()
+ocr = dxp.output_count_rate.get()
+roi = mca.rois.roi0.count.get()
+ert = mca.elapsed_real_time.get()
+```
+
+The issue here is that the channels update continously, and we do not have a way of knowing that no new updates are coming in. How do we solve that?
+
+### Debye
+Fully HW mode, no integration to BEC yet.

csaxs_bec/devices/hexapod/__init__.py

@@ -0,0 +1,2 @@
+from .c887_controller import C887Controller
+from .m850_ophyd import M850Motor

csaxs_bec/devices/hexapod/c887_controller.py

@@ -0,0 +1,278 @@
+import functools
+import logging
+import time
+
+from ophyd_devices.utils.controller import Controller, axis_checked, threadlocked
+from prettytable import PrettyTable
+from typeguard import typechecked
+
+from csaxs_bec.devices.hexapod.c887_errors import C887CommunicationError, C887ErrorCode
+
+logger = logging.getLogger("c887_controller")
+
+
+def retry_once(fcn):
+    """Retry once if a stale socket response caused a parsing/communication error."""
+
+    @functools.wraps(fcn)
+    def wrapper(self, *args, **kwargs):
+        try:
+            val = fcn(self, *args, **kwargs)
+        except C887CommunicationError:
+            val = fcn(self, *args, **kwargs)
+        return val
+
+    return wrapper
+
+
+class C887Controller(Controller):
+    """Socket controller for PI C-887 hexapod controllers using GCS 2.0."""
+
+    _axes_per_controller = 6
+    _initialized = False
+    AXES = ("X", "Y", "Z", "U", "V", "W")
+    USER_ACCESS = [
+        "socket_put_and_receive",
+        "describe",
+        "axis_is_referenced",
+        "all_axes_referenced",
+        "find_reference_mark",
+        "get_position",
+        "get_target_position",
+        "get_position_unit",
+        "get_position_limits",
+        "get_system_velocity",
+        "halt_axis",
+        "is_axis_moving",
+        "move_axis_to_absolute_position",
+        "move_axis_to_relative_position",
+        "set_servo_mode",
+        "set_system_velocity",
+        "stop_all_axes",
+    ]
+
+    def __init__(
+        self,
+        *,
+        name="C887Controller",
+        kind=None,
+        parent=None,
+        socket_cls=None,
+        socket_host=None,
+        socket_port=None,
+        device_manager=None,
+        attr_name="",
+        labels=None,
+    ):
+        if not self._initialized:
+            super().__init__(
+                name=name,
+                socket_cls=socket_cls,
+                socket_host=socket_host,
+                socket_port=socket_port,
+                device_manager=device_manager,
+                attr_name=attr_name,
+                parent=parent,
+                labels=labels,
+                kind=kind,
+            )
+
+    @threadlocked
+    def socket_put(self, val: str):
+        self.sock.put(f"{val}\n".encode())
+
+    @threadlocked
+    def socket_put_raw(self, val: bytes):
+        self.sock.put(val)
+
+    @threadlocked
+    def socket_put_and_receive(self, val: str, remove_trailing_chars=True) -> str:
+        self.socket_put(val)
+        return_val = self._receive_until_lf()
+        if remove_trailing_chars:
+            return_val = self._remove_trailing_characters(return_val)
+            logger.debug(f"Sending {val}; Returned {return_val}")
+        return return_val
+
+    @threadlocked
+    def socket_put_raw_and_receive(self, val: bytes, remove_trailing_chars=True) -> str:
+        self.socket_put_raw(val)
+        return_val = self._receive_until_lf()
+        if remove_trailing_chars:
+            return_val = self._remove_trailing_characters(return_val)
+            logger.debug(f"Sending {val}; Returned {return_val}")
+        return return_val
+
+    def _receive_until_lf(self) -> str:
+        return_val = ""
+        max_wait_time = 1
+        elapsed_time = 0
+        sleep_time = 0.01
+        while True:
+            ret = self.socket_get()
+            return_val += ret
+            if ret.endswith("\n"):
+                break
+            time.sleep(sleep_time)
+            elapsed_time += sleep_time
+            if elapsed_time > max_wait_time:
+                break
+        return return_val
+
+    @threadlocked
+    def socket_command(self, val: str, allowed_error_codes: tuple[int, ...] = ()) -> None:
+        """Send a command that has no response and validate the following ERR? code."""
+        self.socket_put(val)
+        self._check_error_code(self.get_error(), allowed_error_codes=allowed_error_codes)
+
+    @retry_once
+    def get_error(self) -> int:
+        return int(self.socket_put_and_receive("ERR?"))
+
+    @retry_once
+    @axis_checked
+    def get_position(self, axis_Id_numeric: int) -> float:
+        axis = self._axis_name(axis_Id_numeric)
+        return self._parse_axis_value(self.socket_put_and_receive(f"POS? {axis}"), axis, float)
+
+    @retry_once
+    @axis_checked
+    def get_target_position(self, axis_Id_numeric: int) -> float:
+        axis = self._axis_name(axis_Id_numeric)
+        return self._parse_axis_value(self.socket_put_and_receive(f"MOV? {axis}"), axis, float)
+
+    @retry_once
+    @axis_checked
+    @typechecked
+    def move_axis_to_absolute_position(self, axis_Id_numeric: int, target_val: float) -> None:
+        axis = self._axis_name(axis_Id_numeric)
+        self.socket_command(f"MOV {axis} {target_val:g}")
+
+    @retry_once
+    @axis_checked
+    @typechecked
+    def move_axis_to_relative_position(self, axis_Id_numeric: int, target_val: float) -> None:
+        axis = self._axis_name(axis_Id_numeric)
+        self.socket_command(f"MVR {axis} {target_val:g}")
+
+    @retry_once
+    @axis_checked
+    def axis_is_referenced(self, axis_Id_numeric: int) -> bool:
+        axis = self._axis_name(axis_Id_numeric)
+        return bool(self._parse_axis_value(self.socket_put_and_receive(f"FRF? {axis}"), axis, int))
+
+    def all_axes_referenced(self) -> bool:
+        return all(
+            self.axis_is_referenced(ax.axis_Id_numeric) for ax in self._axis if ax is not None
+        )
+
+    @retry_once
+    @axis_checked
+    def find_reference_mark(self, axis_Id_numeric: int) -> None:
+        axis = self._axis_name(axis_Id_numeric)
+        self.socket_command(f"FRF {axis}")
+
+    @retry_once
+    @axis_checked
+    def get_on_target(self, axis_Id_numeric: int) -> bool:
+        axis = self._axis_name(axis_Id_numeric)
+        return bool(self._parse_axis_value(self.socket_put_and_receive(f"ONT? {axis}"), axis, int))
+
+    @retry_once
+    @axis_checked
+    def is_axis_moving(self, axis_Id_numeric: int) -> bool:
+        return not self.get_on_target(axis_Id_numeric)
+
+    @retry_once
+    def get_motion_status(self) -> int:
+        """Return the bit-mapped C-887 motion status from single-character command #5."""
+        return int(self.socket_put_raw_and_receive(bytes([5])), 16)
+
+    @retry_once
+    def stop_all_axes(self):
+        self.socket_command("STP", allowed_error_codes=(10,))
+
+    @retry_once
+    @axis_checked
+    def halt_axis(self, axis_Id_numeric: int):
+        axis = self._axis_name(axis_Id_numeric)
+        self.socket_command(f"HLT {axis}", allowed_error_codes=(10,))
+
+    @retry_once
+    @axis_checked
+    def set_servo_mode(self, axis_Id_numeric: int, enabled: bool) -> None:
+        axis = self._axis_name(axis_Id_numeric)
+        self.socket_command(f"SVO {axis} {int(enabled)}")
+
+    @retry_once
+    @axis_checked
+    def get_servo_mode(self, axis_Id_numeric: int) -> bool:
+        axis = self._axis_name(axis_Id_numeric)
+        return bool(self._parse_axis_value(self.socket_put_and_receive(f"SVO? {axis}"), axis, int))
+
+    @retry_once
+    @typechecked
+    def set_system_velocity(self, velocity: float) -> None:
+        if velocity < 0:
+            raise ValueError("System velocity must be >= 0.")
+        self.socket_command(f"VLS {velocity:g}")
+
+    @retry_once
+    def get_system_velocity(self) -> float:
+        return float(self.socket_put_and_receive("VLS?"))
+
+    @retry_once
+    @axis_checked
+    def get_position_unit(self, axis_Id_numeric: int) -> str:
+        axis = self._axis_name(axis_Id_numeric)
+        return self._parse_axis_value(self.socket_put_and_receive(f"PUN? {axis}"), axis, str)
+
+    @retry_once
+    @axis_checked
+    def get_position_limits(self, axis_Id_numeric: int) -> tuple[float, float]:
+        axis = self._axis_name(axis_Id_numeric)
+        low = self._parse_axis_value(self.socket_put_and_receive(f"TMN? {axis}"), axis, float)
+        high = self._parse_axis_value(self.socket_put_and_receive(f"TMX? {axis}"), axis, float)
+        return (low, high)
+
+    def describe(self) -> None:
+        t = PrettyTable()
+        t.title = f"{self.__class__.__name__} on {self.sock.host}:{self.sock.port}"
+        t.field_names = ["Axis", "Name", "Connected", "Referenced", "Position", "Unit"]
+        for ax in range(self._axes_per_controller):
+            axis = self._axis[ax]
+            if axis is not None:
+                t.add_row(
+                    [
+                        axis.axis_Id,
+                        axis.name,
+                        axis.connected,
+                        self.axis_is_referenced(axis.axis_Id_numeric),
+                        axis.readback.read().get(axis.name).get("value"),
+                        self.get_position_unit(axis.axis_Id_numeric),
+                    ]
+                )
+            else:
+                t.add_row([None for _ in t.field_names])
+        print(t)
+
+    def _axis_name(self, axis_Id_numeric: int) -> str:
+        return self.AXES[axis_Id_numeric]
+
+    def _parse_axis_value(self, msg: str, axis: str, value_type):
+        prefix = f"{axis}="
+        if not msg.startswith(prefix):
+            raise C887CommunicationError(
+                f"Expected to receive a return message starting with {prefix} but instead"
+                f" received '{msg}'"
+            )
+        return value_type(msg.split("=", maxsplit=1)[1])
+
+    def _check_error_code(self, error_code: int, allowed_error_codes: tuple[int, ...] = ()) -> None:
+        if error_code != 0 and error_code not in allowed_error_codes:
+            raise C887ErrorCode(error_code)
+
+    def _remove_trailing_characters(self, var: str) -> str:
+        if len(var) > 1:
+            return var.split("\n")[0].strip()
+        return var

csaxs_bec/devices/hexapod/c887_errors.py

@@ -0,0 +1,59 @@
+C887_ERRORS = {
+    0: "No error",
+    1: "Parameter syntax error",
+    2: "Unknown command",
+    3: "Command length out of limits or command buffer overrun",
+    4: "Error while scanning",
+    5: "Unallowable move attempted on unreferenced axis, or move attempted with servo off",
+    6: "Parameter for SGA not valid",
+    7: "Position out of limits",
+    10: "Controller was stopped by command",
+    14: "Invalid axis identifier",
+    15: "Invalid stage name",
+    17: "Parameter out of range",
+    19: "Macro record error",
+    20: "Macro not found",
+    21: "Axis has no brake",
+    22: "Axis identifier specified more than once",
+    23: "Illegal axis",
+    24: "Incorrect number of parameters",
+    25: "Invalid floating point number",
+    26: "Parameter missing",
+    27: "Soft limit out of range",
+    28: "No manual pad found",
+    29: "No more step-response values",
+    30: "No step-response values recorded",
+    31: "Axis has no reference sensor",
+    32: "Axis has no limit switch",
+    33: "No relay card installed",
+    34: "Command not allowed for selected stage",
+    35: "No digital input installed",
+    36: "No digital output configured",
+    39: "No joystick configured",
+    45: "Referencing failed",
+    54: "Command not allowed in current motion mode",
+    55: "No parameter set loaded",
+    56: "Command not allowed in current servo mode",
+    73: "Motion still in progress",
+    100: "PI LabVIEW driver reports error",
+    200: "No stage connected",
+}
+
+
+class C887Error(Exception):
+    pass
+
+
+class C887CommunicationError(C887Error):
+    pass
+
+
+class C887ErrorCode(C887Error):
+    def __init__(self, error_code: int, message=""):
+        self.error_code = error_code
+        self.error_code_message = C887_ERRORS.get(error_code, "UNKNOWN ERROR")
+        self.message = message
+        super().__init__(self.message)
+
+    def __str__(self):
+        return f"{self.error_code} / {self.error_code_message}. {self.message}"

csaxs_bec/devices/hexapod/m850_ophyd.py

@@ -0,0 +1,234 @@
+import logging
+import threading
+import time
+
+import numpy as np
+from bec_lib import bec_logger
+from ophyd import Component as Cpt
+from ophyd import Device, PositionerBase, Signal
+from ophyd.status import wait as status_wait
+from ophyd.utils import LimitError, ReadOnlyError
+from ophyd_devices.utils.controller import threadlocked
+from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected
+
+from csaxs_bec.devices.hexapod.c887_controller import C887Controller
+from csaxs_bec.devices.hexapod.c887_errors import C887Error
+
+logger = bec_logger.logger
+
+
+class M850SignalBase(SocketSignal):
+    def __init__(self, signal_name, **kwargs):
+        self.signal_name = signal_name
+        super().__init__(**kwargs)
+        self.controller = self.parent.controller
+        self.sock = self.parent.controller.sock
+
+
+class M850SignalRO(M850SignalBase):
+    def __init__(self, signal_name, **kwargs):
+        super().__init__(signal_name, **kwargs)
+        self._metadata["write_access"] = False
+
+    @threadlocked
+    def _socket_set(self, val):
+        raise ReadOnlyError("Read-only signals cannot be set")
+
+
+class M850ReadbackSignal(M850SignalRO):
+    @threadlocked
+    def _socket_get(self):
+        return self.controller.get_position(self.parent.axis_Id_numeric) * self.parent.sign
+
+
+class M850SetpointSignal(M850SignalBase):
+    def __init__(self, signal_name, **kwargs):
+        super().__init__(signal_name, **kwargs)
+        self.setpoint = 0
+
+    @threadlocked
+    def _socket_get(self):
+        return self.setpoint
+
+    @threadlocked
+    def _socket_set(self, val):
+        target_val = val * self.parent.sign
+        self.setpoint = target_val
+
+        if self.controller.axis_is_referenced(self.parent.axis_Id_numeric):
+            self.controller.move_axis_to_absolute_position(self.parent.axis_Id_numeric, target_val)
+        else:
+            raise C887Error(f"Axis {self.parent.axis_Id} is not referenced.")
+
+
+class M850MotorIsMoving(M850SignalRO):
+    @threadlocked
+    def _socket_get(self):
+        return self.controller.is_axis_moving(self.parent.axis_Id_numeric)
+
+
+class M850AxisReferenced(M850SignalRO):
+    @threadlocked
+    def _socket_get(self):
+        return self.parent.controller.axis_is_referenced(self.parent.axis_Id_numeric)
+
+
+class M850Motor(Device, PositionerBase):
+    USER_ACCESS = ["controller"]
+    readback = Cpt(M850ReadbackSignal, signal_name="readback", kind="hinted")
+    user_setpoint = Cpt(M850SetpointSignal, signal_name="setpoint")
+    motor_is_moving = Cpt(M850MotorIsMoving, signal_name="motor_is_moving", kind="normal")
+    axis_is_referenced = Cpt(M850AxisReferenced, signal_name="axis_is_referenced", kind="config")
+    high_limit_travel = Cpt(Signal, value=0, kind="omitted")
+    low_limit_travel = Cpt(Signal, value=0, kind="omitted")
+
+    SUB_READBACK = "readback"
+    SUB_CONNECTION_CHANGE = "connection_change"
+    _default_sub = SUB_READBACK
+
+    def __init__(
+        self,
+        axis_Id,
+        prefix="",
+        *,
+        name,
+        kind=None,
+        read_attrs=None,
+        configuration_attrs=None,
+        parent=None,
+        host="192.168.1.28",
+        port=50000,
+        limits=None,
+        sign=1,
+        socket_cls=SocketIO,
+        device_manager=None,
+        **kwargs,
+    ):
+        self.controller = C887Controller(
+            socket_cls=socket_cls, socket_host=host, socket_port=port, device_manager=device_manager
+        )
+        self.axis_Id = axis_Id
+        self.sign = sign
+        self.controller.set_axis(axis=self, axis_nr=self.axis_Id_numeric)
+        self.tolerance = kwargs.pop("tolerance", 0.001)
+
+        super().__init__(
+            prefix,
+            name=name,
+            kind=kind,
+            read_attrs=read_attrs,
+            configuration_attrs=configuration_attrs,
+            parent=parent,
+            **kwargs,
+        )
+        self.readback.name = self.name
+        self.controller.subscribe(
+            self._update_connection_state, event_type=self.SUB_CONNECTION_CHANGE
+        )
+        self._update_connection_state()
+        if limits is not None:
+            assert len(limits) == 2
+            self.low_limit_travel.put(limits[0])
+            self.high_limit_travel.put(limits[1])
+
+    def wait_for_connection(self, timeout: float = 30.0) -> bool:
+        self.controller.on(timeout=timeout)
+        self._update_setpoint_from_readback()
+
+    def _update_setpoint_from_readback(self):
+        self.user_setpoint.setpoint = self.readback.get()
+
+    def destroy(self):
+        self.controller.off(update_config=False)
+        return super().destroy()
+
+    @property
+    def limits(self):
+        return (self.low_limit_travel.get(), self.high_limit_travel.get())
+
+    @property
+    def low_limit(self):
+        return self.limits[0]
+
+    @property
+    def high_limit(self):
+        return self.limits[1]
+
+    def check_value(self, pos):
+        low_limit, high_limit = self.limits
+
+        if low_limit < high_limit and not (low_limit <= pos <= high_limit):
+            raise LimitError(f"position={pos} not within limits {self.limits}")
+
+    def _update_connection_state(self, **kwargs):
+        for walk in self.walk_signals():
+            walk.item._metadata["connected"] = self.controller.connected
+
+    @raise_if_disconnected
+    def move(self, position, wait=True, **kwargs):
+        self._started_moving = False
+        timeout = kwargs.pop("timeout", 10)
+        status = super().move(position, timeout=timeout, **kwargs)
+        self.user_setpoint.put(position, wait=False)
+
+        def move_and_finish():
+            while self.motor_is_moving.get():
+                val = self.readback.read()
+                self._run_subs(sub_type=self.SUB_READBACK, value=val, timestamp=time.time())
+                time.sleep(0.1)
+            val = self.readback.read()
+            success = np.isclose(val[self.name]["value"], position, atol=self.tolerance)
+            self._done_moving(success=success)
+
+        threading.Thread(target=move_and_finish, daemon=True).start()
+        try:
+            if wait:
+                status_wait(status)
+        except KeyboardInterrupt:
+            self.stop()
+            raise
+
+        return status
+
+    @property
+    def axis_Id(self):
+        return self._axis_Id_alpha
+
+    @axis_Id.setter
+    def axis_Id(self, val):
+        if isinstance(val, str):
+            val = val.upper()
+            if val not in C887Controller.AXES:
+                raise ValueError(f"Only C-887 hexapod axes {C887Controller.AXES} are supported.")
+            self._axis_Id_alpha = val
+            self._axis_Id_numeric = C887Controller.AXES.index(val)
+        else:
+            raise TypeError(f"Expected value of type str but received {type(val)}")
+
+    @property
+    def axis_Id_numeric(self):
+        return self._axis_Id_numeric
+
+    @axis_Id_numeric.setter
+    def axis_Id_numeric(self, val):
+        if isinstance(val, int):
+            if val >= len(C887Controller.AXES):
+                raise ValueError("Numeric value exceeds supported range.")
+            self._axis_Id_numeric = val
+            self._axis_Id_alpha = C887Controller.AXES[val]
+        else:
+            raise TypeError(f"Expected value of type int but received {type(val)}")
+
+    @property
+    def egu(self):
+        return "deg" if self.axis_Id in ("U", "V", "W") else "mm"
+
+    def stage(self) -> list[object]:
+        return super().stage()
+
+    def unstage(self) -> list[object]:
+        return super().unstage()
+
+    def stop(self, *, success=False):
+        self.controller.halt_axis(self.axis_Id_numeric)
+        return super().stop(success=success)

tests/tests_devices/test_m850_hexapod.py

@@ -0,0 +1,194 @@
+from unittest import mock
+
+import pytest
+from ophyd_devices.tests.utils import SocketMock
+
+from csaxs_bec.devices.hexapod import C887Controller, M850Motor
+from csaxs_bec.devices.hexapod.c887_errors import C887CommunicationError, C887ErrorCode
+
+
+@pytest.fixture
+def controller(dm_with_devices):
+    C887Controller._reset_controller()
+    controller = C887Controller(
+        socket_cls=SocketMock,
+        socket_host="dummy",
+        socket_port=50000,
+        device_manager=dm_with_devices,
+    )
+    controller.on()
+    controller.sock.flush_buffer()
+    yield controller
+
+
+@pytest.fixture
+def hex_x(dm_with_devices):
+    C887Controller._reset_controller()
+    motor = M850Motor(
+        "X",
+        name="hex_x",
+        host="dummy",
+        port=50000,
+        socket_cls=SocketMock,
+        device_manager=dm_with_devices,
+    )
+    motor.controller.on()
+    motor.controller.sock.flush_buffer()
+    motor.stage()
+    yield motor
+
+
+@pytest.mark.parametrize(
+    "axis,position,get_message,return_msg",
+    [
+        (0, 50, b"POS? X\n", b"X=50.000000\n"),
+        (1, 0, b"POS? Y\n", b"Y=0\n"),
+        (2, -25.23, b"POS? Z\n", b"Z=-25.230000\n"),
+        (3, 1.5, b"POS? U\n", b"U=1.500000\n"),
+    ],
+)
+def test_get_position(controller, axis, position, get_message, return_msg):
+    controller.sock.buffer_recv = return_msg
+    val = controller.get_position(axis)
+    assert val == position
+    assert controller.sock.buffer_put[0] == get_message
+
+
+def test_get_position_raises_on_unexpected_axis_response(controller):
+    controller.sock.buffer_recv = b"Y=1.0\n"
+    with pytest.raises(C887CommunicationError):
+        controller.get_position(0)
+
+
+@pytest.mark.parametrize(
+    "axis,is_referenced,get_message,return_msg,exception",
+    [
+        (0, True, b"FRF? X\n", b"X=1\n", None),
+        (1, False, b"FRF? Y\n", b"Y=0\n", None),
+        (6, False, b"FRF? X\n", b"X=0\n", ValueError),
+    ],
+)
+def test_axis_is_referenced(controller, axis, is_referenced, get_message, return_msg, exception):
+    controller.sock.buffer_recv = return_msg
+    if exception is not None:
+        with pytest.raises(exception):
+            controller.axis_is_referenced(axis)
+    else:
+        val = controller.axis_is_referenced(axis)
+        assert val == is_referenced
+        assert controller.sock.buffer_put[0] == get_message
+
+
+@pytest.mark.parametrize(
+    "axis,pos,get_msg,return_msg",
+    [
+        (0, 50, [b"MOV X 50\n", b"ERR?\n"], [b"0\n"]),
+        (1, 0, [b"MOV Y 0\n", b"ERR?\n"], [b"0\n"]),
+        (3, -2.3, [b"MOV U -2.3\n", b"ERR?\n"], [b"0\n"]),
+    ],
+)
+def test_move_axis_to_absolute_position(controller, axis, pos, get_msg, return_msg):
+    controller.sock.buffer_recv = return_msg
+    controller.move_axis_to_absolute_position(axis, pos)
+    assert controller.sock.buffer_put == get_msg
+
+
+def test_move_axis_raises_controller_error(controller):
+    controller.sock.buffer_recv = [b"7\n"]
+    with pytest.raises(C887ErrorCode):
+        controller.move_axis_to_absolute_position(0, 200)
+
+
+@pytest.mark.parametrize(
+    "is_moving,get_message,return_msg",
+    [(False, b"ONT? X\n", b"X=1\n"), (True, b"ONT? X\n", b"X=0\n")],
+)
+def test_axis_is_moving(controller, is_moving, get_message, return_msg):
+    controller.sock.buffer_recv = return_msg
+    val = controller.is_axis_moving(0)
+    assert val == is_moving
+    assert controller.sock.buffer_put[0] == get_message
+
+
+def test_motion_status_uses_single_character_command(controller):
+    controller.sock.buffer_recv = b"3\n"
+    assert controller.get_motion_status() == 3
+    assert controller.sock.buffer_put[0] == bytes([5])
+
+
+def test_set_and_get_system_velocity(controller):
+    controller.sock.buffer_recv = [b"0\n", b"2.5\n"]
+    controller.set_system_velocity(2.5)
+    velocity = controller.get_system_velocity()
+    assert velocity == 2.5
+    assert controller.sock.buffer_put == [b"VLS 2.5\n", b"ERR?\n", b"VLS?\n"]
+
+
+def test_get_position_limits(controller):
+    controller.sock.buffer_recv = [b"X=-50\n", b"X=50\n"]
+    assert controller.get_position_limits(0) == (-50, 50)
+    assert controller.sock.buffer_put == [b"TMN? X\n", b"TMX? X\n"]
+
+
+@pytest.mark.parametrize(
+    "pos,get_msg,return_msg",
+    [
+        (
+            20.23,
+            [b"FRF? X\n", b"MOV X 20.23\n", b"ERR?\n", b"ONT? X\n", b"POS? X\n"],
+            [b"X=1\n", b"0\n", b"X=1\n", b"X=20.230000\n"],
+        ),
+        (
+            0,
+            [b"FRF? X\n", b"MOV X 0\n", b"ERR?\n", b"ONT? X\n", b"POS? X\n"],
+            [b"X=1\n", b"0\n", b"X=1\n", b"X=0\n"],
+        ),
+    ],
+)
+def test_move_axis(hex_x, pos, get_msg, return_msg):
+    controller = hex_x.controller
+    controller.sock.buffer_recv = return_msg
+    hex_x.move(pos)
+    assert controller.sock.buffer_put == get_msg
+
+
+def test_stop_axis(hex_x):
+    controller = hex_x.controller
+    controller.sock.buffer_recv = [b"10\n"]
+    hex_x.stop()
+    assert controller.sock.buffer_put == [b"HLT X\n", b"ERR?\n"]
+
+
+def test_all_axes_referenced(hex_x):
+    controller = hex_x.controller
+    with mock.patch.object(controller, "axis_is_referenced", return_value=True) as mock_is_ref:
+        val = controller.all_axes_referenced()
+        assert val
+        mock_is_ref.assert_called_once_with(0)
+
+
+def test_axis_units(dm_with_devices):
+    C887Controller._reset_controller()
+    x_motor = M850Motor(
+        "X",
+        name="hex_x",
+        host="dummy",
+        port=50000,
+        socket_cls=SocketMock,
+        device_manager=dm_with_devices,
+    )
+    assert x_motor.egu == "mm"
+    assert x_motor.axis_Id_numeric == 0
+    assert x_motor.axis_Id == "X"
+
+    C887Controller._reset_controller()
+    u_motor = M850Motor(
+        "U",
+        name="hex_u",
+        host="dummy",
+        port=50000,
+        socket_cls=SocketMock,
+        device_manager=dm_with_devices,
+    )
+    assert u_motor.egu == "deg"
+    assert u_motor.axis_Id_numeric == 3