diff --git a/csaxs_bec/devices/npoint/npoint.py b/csaxs_bec/devices/npoint/npoint.py index 0ab35893..e570d358 100644 --- a/csaxs_bec/devices/npoint/npoint.py +++ b/csaxs_bec/devices/npoint/npoint.py @@ -1,4 +1,3 @@ -import functools import threading import time @@ -8,25 +7,13 @@ from ophyd import Component as Cpt from ophyd import Device, PositionerBase, Signal, SignalRO from ophyd.status import wait as status_wait from ophyd.utils import LimitError, ReadOnlyError -from ophyd_devices.utils.controller import Controller, threadlocked +from ophyd_devices.utils.controller import Controller, ControllerCommunicationError, threadlocked from ophyd_devices.utils.socket import SocketIO, SocketSignal, raise_if_disconnected from prettytable import PrettyTable logger = bec_logger.logger -def channel_checked(fcn): - """Decorator to catch attempted access to channels that are not available.""" - - @functools.wraps(fcn) - def wrapper(self, *args, **kwargs): - # pylint: disable=protected-access - self._check_channel(args[0]) - return fcn(self, *args, **kwargs) - - return wrapper - - class NpointError(Exception): """ Base class for Npoint errors. @@ -40,11 +27,17 @@ class NPointController(Controller): """ _axes_per_controller = 3 - _read_single_loc_bit = "A0" - _write_single_loc_bit = "A2" - _trailing_bit = "55" - _range_offset = "78" - _channel_base = ["11", "83"] + READ = b"\xa0" + WRITE = b"\xa2" + END = b"\x55" + + CHANNEL_STRIDE = 0x1000 + RANGE = 0x11831078 + POSITION = 0x11831334 + TARGET = 0x11831218 + SERVO = 0x11831084 + + COUNTS_PER_UM = 1_048_574 / 100 def show_all(self) -> None: """Display current status of all channels @@ -62,244 +55,70 @@ class NPointController(Controller): t.add_row([ii, self._get_range(ii), self.get_current_pos(ii), self.get_target_pos(ii)]) print(t) - @channel_checked - def _get_range(self, channel: int) -> int: - """Get the range of the specified channel axis. - - Args: - channel (int): Channel for which the range should be requested. - - Raises: - RuntimeError: Raised if the received message doesn't have the expected number of bytes (10). - - Returns: - int: Range - """ - - # for first channel: 0x11 83 10 78 - addr = self._channel_base.copy() - addr.extend([f"{16 + 16 * channel:x}", self._range_offset]) - send_buffer = self.__read_single_location_buffer(addr) - - recvd = self._put_and_receive(send_buffer) - if len(recvd) != 10: - raise RuntimeError( - f"Received buffer is corrupted. Expected 10 bytes and instead got {len(recvd)}" - ) - device_range = self._hex_list_to_int(recvd[5:-1], signed=False) - return device_range - - @channel_checked def get_current_pos(self, channel: int) -> float: - # for first channel: 0x11 83 13 34 - addr = self._channel_base.copy() - addr.extend([f"{19 + 16 * channel:x}", "34"]) - send_buffer = self.__read_single_location_buffer(addr) + """ + Return the current channel position in micrometres. - recvd = self._put_and_receive(send_buffer) + Args: + channel (int): The channel number (0, 1, or 2). + """ + return self._read_register(self.POSITION, channel) / self.COUNTS_PER_UM - pos_buffer = recvd[5:-1] - pos = self._hex_list_to_int(pos_buffer) / 1048574 * 100 - return pos - - @channel_checked - def set_target_pos(self, channel: int, pos: float) -> None: - # for first channel: 0x11 83 12 18 00 00 00 00 - addr = self._channel_base.copy() - addr.extend([f"{18 + channel * 16:x}", "18"]) - - target = int(round(1048574 / 100 * pos)) - data = [f"{m:02x}" for m in target.to_bytes(4, byteorder="big", signed=True)] - - send_buffer = self.__write_single_location_buffer(addr, data) - self._put(send_buffer) - - @channel_checked def get_target_pos(self, channel: int) -> float: - # for first channel: 0x11 83 12 18 - addr = self._channel_base.copy() - addr.extend([f"{18 + channel * 16:x}", "18"]) - send_buffer = self.__read_single_location_buffer(addr) - - recvd = self._put_and_receive(send_buffer) - pos_buffer = recvd[5:-1] - pos = self._hex_list_to_int(pos_buffer) / 1048574 * 100 - return pos - - @channel_checked - def _set_servo(self, channel: int, enable: bool) -> None: - print("Not tested") - return - # # for first channel: 0x11 83 10 84 00 00 00 00 - # addr = self._channel_base.copy() - # addr.extend([f"{16 + channel * 16:x}", "84"]) - - # if enable: - # data = ["00"] * 3 + ["01"] - # else: - # data = ["00"] * 4 - # send_buffer = self.__write_single_location_buffer(addr, data) - - # self._put(send_buffer) - - @channel_checked - def _get_servo(self, channel: int) -> int: - # for first channel: 0x11 83 10 84 00 00 00 00 - addr = self._channel_base.copy() - addr.extend([f"{16 + channel * 16:x}", "84"]) - send_buffer = self.__read_single_location_buffer(addr) - - recvd = self._put_and_receive(send_buffer) - buffer = recvd[5:-1] - status = self._hex_list_to_int(buffer) - return status - - @threadlocked - def _put(self, buffer: list) -> None: - """Translates a list of hex values to bytes and sends them to the socket. + """ + Return the target channel position in micrometres. Args: - buffer (list): List of hex values without leading 0x - - Returns: - None + channel (int): The channel number (0, 1, or 2). """ + return self._read_register(self.TARGET, channel) / self.COUNTS_PER_UM - buffer = b"".join([bytes.fromhex(m) for m in buffer]) - self.sock.put(buffer) - - @threadlocked - def _put_and_receive(self, msg_hex_list: list) -> list: - """Send msg to socket and wait for a reply. + def set_target_pos(self, channel: int, pos: float) -> None: + """ + Set the target channel position in micrometres. Args: - msg_hex_list (list): List of hex values without leading 0x. - - Returns: - list: Received message as a list of hex values + channel (int): The channel number (0, 1, or 2). + pos (float): The target position in micrometres. """ - - buffer = b"".join([bytes.fromhex(m) for m in msg_hex_list]) - self.sock.put(buffer) - recv_msg = self.sock.receive() - recv_hex_list = [hex(m) for m in recv_msg] - self._verify_received_msg(msg_hex_list, recv_hex_list) - return recv_hex_list - - def _verify_received_msg(self, in_list: list, out_list: list) -> None: - """Ensure that the first address bits of sent and received messages are the same. - - Args: - in_list (list): list containing the sent message - out_list (list): list containing the received message - - Raises: - RuntimeError: Raised if first two address bits of 'in' and 'out' are not identical - - Returns: - None - """ - - # first, translate hex (str) values to int - in_list_int = [int(val, 16) for val in in_list] - out_list_int = [int(val, 16) for val in out_list] - - # first ints of the reply should be the same. Otherwise something went wrong - if not in_list_int[:2] == out_list_int[:2]: - raise RuntimeError("Connection failure. Please restart the controller.") + self._write_register(self.TARGET, channel, round(pos * self.COUNTS_PER_UM)) def _check_channel(self, channel: int) -> None: - if channel >= self._axes_per_controller: - raise ValueError( - f"Channel {channel+1} exceeds the available number of channels ({self._axes_per_controller})" + if not 0 <= channel < self._axes_per_controller: + raise ValueError(f"Invalid channel: {channel}") + + def _register_address(self, register: int, channel: int) -> bytes: + self._check_channel(channel) + return (register + channel * self.CHANNEL_STRIDE).to_bytes(4, "little") + + def _read_register(self, register: int, channel: int, *, signed: bool = True) -> int: + """Read a 32-bit register, validating the opcode, address and end marker.""" + header = self.READ + self._register_address(register, channel) + response = self.socket_put_and_receive(header + self.END, response_length=10) + if response[:5] != header or response[-1:] != self.END: + raise NpointError( + f"Invalid response for {header.hex(' ')}: {response.hex(' ')}. " + "Connection failure. Please restart the controller." ) + return int.from_bytes(response[5:9], "little", signed=signed) - @staticmethod - def _hex_list_to_int(in_buffer: list, byteorder="little", signed=True) -> int: - """Translate hex list to int. + def _write_register(self, register: int, channel: int, value: int) -> None: + """Write a signed 32-bit register. The controller sends no reply.""" + address = self._register_address(register, channel) + payload = value.to_bytes(4, "little", signed=True) + self.socket_put(self.WRITE + address + payload + self.END) - Args: - in_buffer (list): Input buffer; received as list of hex values - byteorder (str, optional): Byteorder of in_buffer. Defaults to "little". - signed (bool, optional): Whether the hex list represents a signed int. Defaults to True. + def _get_range(self, channel: int) -> int: + """Return the channel range as an unsigned register value.""" + return self._read_register(self.RANGE, channel, signed=False) - Returns: - int: Translated integer. - """ - if byteorder == "little": - in_buffer.reverse() + def _set_servo(self, channel: int, enable: bool) -> None: + """Servo writes remain disabled until the command has been verified.""" + self._check_channel(channel) + print("Not tested") - # make sure that all hex strings have the same format ("FF") - val_hex = [f"{int(m, 16):02x}" for m in in_buffer] - - val_bytes = [bytes.fromhex(m) for m in val_hex] - val = int.from_bytes(b"".join(val_bytes), byteorder="big", signed=signed) - return val - - @staticmethod - def __read_single_location_buffer(addr) -> list: - """Prepare buffer for reading from a single memory location (hex address). - Number of bytes: 6 - Format: 0xA0 [addr] 0x55 - Return Value: 0xA0 [addr] [data] 0x55 - Sample Hex Transmission from PC to LC.400: A0 18 12 83 11 55 - Sample Hex Return Transmission from LC.400 to PC: A0 18 12 83 11 64 00 00 00 55 - - Args: - addr (list): Hex address to read from - - Returns: - list: List of hex values representing the read instruction. - """ - buffer = [] - buffer.append(NPointController._read_single_loc_bit) - if isinstance(addr, list): - addr.reverse() - buffer.extend(addr) - else: - buffer.append(addr) - buffer.append(NPointController._trailing_bit) - - return buffer - - @staticmethod - def __write_single_location_buffer(addr: list, data: list) -> list: - """Prepare buffer for writing to a single memory location (hex address). - Number of bytes: 10 - Format: 0xA2 [addr] [data] 0x55 - Return Value: none - Sample Hex Transmission from PC to C.400: A2 18 12 83 11 E8 03 00 00 55 - - Args: - addr (list): List of hex values representing the address to write to. - data (list): List of hex values representing the data that should be written. - - Returns: - list: List of hex values representing the write instruction. - """ - buffer = [] - buffer.append(NPointController._write_single_loc_bit) - if isinstance(addr, list): - addr.reverse() - buffer.extend(addr) - else: - buffer.append(addr) - - if isinstance(data, list): - data.reverse() - buffer.extend(data) - else: - buffer.append(data) - buffer.append(NPointController._trailing_bit) - return buffer - - @staticmethod - def __read_array(): - raise NotImplementedError - - @staticmethod - def __write_next_command(): - raise NotImplementedError + def _get_servo(self, channel: int) -> int: + return self._read_register(self.SERVO, channel) def __del__(self): if self.connected: @@ -452,7 +271,7 @@ class NPointAxis(Device, PositionerBase): try: self.controller.on(timeout=timeout) self._update_setpoint_from_readback() - except TimeoutError: + except (ControllerCommunicationError, TimeoutError): self.controller.off(update_config=False) time.sleep(1) else: @@ -463,7 +282,7 @@ class NPointAxis(Device, PositionerBase): f"Try to reload the config and if the problem persists, check the connection to the nPoint controller " f"and ensure that it is powered on and accessible at {self.controller._socket_host}:{self.controller._socket_port}." ) - + def _update_setpoint_from_readback(self): """ The setpoint is only stored locally. After a restart, diff --git a/tests/tests_devices/test_npoint_piezo.py b/tests/tests_devices/test_npoint_piezo.py index 85e6e545..f9682d8c 100644 --- a/tests/tests_devices/test_npoint_piezo.py +++ b/tests/tests_devices/test_npoint_piezo.py @@ -1,9 +1,9 @@ -import copy from unittest import mock import pytest from csaxs_bec.devices.npoint import NPointAxis, NPointController +from csaxs_bec.devices.npoint.npoint import NpointError # pylint: disable=protected-access # pylint: disable=redefined-outer-name @@ -101,8 +101,8 @@ def test_axis_get_out(npointx, pos, msg_in, msg_out): "axis, msg_in, msg_out", [ (0, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x11\xcd\xcc\x00\x00U"), - (1, b"\xa04#\x83\x11U", b"\xa0\x34\x13\x83\x11\x00\x00\x00\x00U"), - (2, b"\xa043\x83\x11U", b"\xa0\x34\x13\x83\x1133\xff\xffU"), + (1, b"\xa04#\x83\x11U", b"\xa0\x34\x23\x83\x11\x00\x00\x00\x00U"), + (2, b"\xa043\x83\x11U", b"\xa0\x34\x33\x83\x1133\xff\xffU"), ], ) def test_axis_get_in(npointx, axis, msg_in, msg_out): @@ -130,48 +130,32 @@ def test_axis_out_of_range(dm_with_devices): ) -def test_get_axis_out_of_range(controller): +@pytest.mark.parametrize("channel", [-1, 3]) +def test_get_axis_out_of_range(controller, channel): """ Test that an error is raised when trying to get the current position of an invalid axis. """ with pytest.raises(ValueError): - controller.get_current_pos(3) + controller.get_current_pos(channel=channel) + controller.sock.put.assert_not_called() -def test_set_axis_out_of_range(controller): +@pytest.mark.parametrize("channel", [-1, 3]) +def test_set_axis_out_of_range(controller, channel): """ Test that an error is raised when trying to set the target position of an invalid axis. """ with pytest.raises(ValueError): - controller.set_target_pos(3, 5) - - -@pytest.mark.parametrize( - "in_buffer, byteorder, signed, val", - [ - (["0x0", "0x0", "0xcc", "0xcd"], "big", True, 52429), - (["0xcd", "0xcc", "0x0", "0x0"], "little", True, 52429), - (["cd", "cc", "00", "00"], "little", True, 52429), - ], -) -def test_hex_list_to_int(in_buffer, byteorder, signed, val): - """ - Test that the hex list is correctly converted to an integer - """ - assert ( - NPointController._hex_list_to_int( - copy.deepcopy(in_buffer), byteorder=byteorder, signed=signed - ) - == val - ) + controller.set_target_pos(channel=channel, pos=5) + controller.sock.put.assert_not_called() @pytest.mark.parametrize( "axis, msg_in, msg_out", [ - (0, b"\xa0x\x10\x83\x11U", b"\xa0\x78\x13\x83\x11\x64\x00\x00\x00U"), - (1, b"\xa0x \x83\x11U", b"\xa0\x78\x13\x83\x11\x64\x00\x00\x00U"), - (2, b"\xa0x0\x83\x11U", b"\xa0\x78\x13\x83\x11\x64\x00\x00\x00U"), + (0, b"\xa0x\x10\x83\x11U", b"\xa0\x78\x10\x83\x11\x64\x00\x00\x00U"), + (1, b"\xa0x \x83\x11U", b"\xa0\x78\x20\x83\x11\x64\x00\x00\x00U"), + (2, b"\xa0x0\x83\x11U", b"\xa0\x78\x30\x83\x11\x64\x00\x00\x00U"), ], ) def test_get_range(npointx, axis, msg_in, msg_out): @@ -183,3 +167,58 @@ def test_get_range(npointx, axis, msg_in, msg_out): val = npointx.controller._get_range(axis) npointx.controller.sock.put.assert_called_once_with(msg_in) assert val == 100 + + +@pytest.mark.parametrize( + "response", + [ + b"\xa2\x34\x13\x83\x11\x00\x00\x00\x00U", # Wrong opcode + b"\xa0\x34\x23\x83\x11\x00\x00\x00\x00U", # Wrong channel + b"\xa0\x34\x13\x82\x11\x00\x00\x00\x00U", # Wrong address + b"\xa0\x34\x13\x83\x11\x00\x00\x00\x00X", # Wrong end marker + ], +) +def test_read_rejects_invalid_frame(controller, response): + controller.sock.receive.side_effect = [response] + with pytest.raises(NpointError, match="Invalid response"): + controller.get_current_pos(0) + + +def test_read_fragmented_payload_with_end_marker(controller): + controller.sock.receive.side_effect = [b"\xa0\x34\x13\x83\x11U", b"\x00\x00", b"\x00U"] + assert controller.get_current_pos(0) == pytest.approx(85 / 1048574 * 100) + assert [call.kwargs["buffer_length"] for call in controller.sock.receive.call_args_list] == [ + 10, + 4, + 2, + ] + + +@pytest.mark.parametrize( + "method,command,response,expected", + [ + ( + "get_target_pos", + b"\xa0\x18\x22\x83\x11U", + b"\xa0\x18\x22\x83\x1133\xff\xffU", + -52429 / 1048574 * 100, + ), + ("_get_servo", b"\xa0\x84\x20\x83\x11U", b"\xa0\x84\x20\x83\x11\x01\x00\x00\x00U", 1), + ( + "_get_range", + b"\xa0\x78\x20\x83\x11U", + b"\xa0\x78\x20\x83\x11\xff\xff\xff\xffU", + 4294967295, + ), + ], +) +def test_register_read_types(controller, method, command, response, expected): + controller.sock.receive.side_effect = [response] + assert getattr(controller, method)(channel=1) == pytest.approx(expected) + controller.sock.put.assert_called_once_with(command) + + +def test_target_write_channel_offset(controller): + controller.set_target_pos(channel=2, pos=-5) + controller.sock.put.assert_called_once_with(b"\xa2\x18\x32\x83\x1133\xff\xffU") + controller.sock.receive.assert_not_called()