bec/csaxs_bec
0
0
Fork 3
Code Issues 17 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

fix(npoint): cleanup and fix for new controller base class
CI for csaxs_bec / test (push) Failing after 2m5s
Details
CI for csaxs_bec / test (pull_request) Failing after 2m7s
Details

Browse Source
This commit is contained in:
wakonig_k
2026-09-15 15:21:32 +02:00
parent 68dafcf064
commit 07eb40b589
2 changed files with 129 additions and 271 deletions
Download Patch File Download Diff File Expand all files Collapse all files
csaxs_bec/devices/npoint/npoint.py
+60 -241
View File
@@ -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,
tests/tests_devices/test_npoint_piezo.py
+69 -30
View File
@@ -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()