Mirko HollerandClaude Sonnet 5
a65ab6cbdd
CI for csaxs_bec / test (push) Successful in 1m54s
Adds a new driver for the SmarAct MCS2 controller (MCS2-C-0008, 9 channels), parallel to the existing MCS1 (SCU) implementation. Uses the MCS2's raw ASCII/SCPI interface over TCP (port 55551), which differs from MCS1 in units (picometers), message termination (\r\n), and error handling (a polled error queue instead of inline echoes). - csaxs_bec/devices/mcs2: Mcs2Controller/Mcs2Motor and errors - csaxs_bec/devices/sim/sim_mcs2.py: simulated backend for testing - tests/tests_devices/test_mcs2.py: unit tests + sim end-to-end test - device_configs: commented example stage in bl_optics_hutch.yaml, and a standalone single-axis mcs2_config_test.yaml Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
318 lines
11 KiB
Python
318 lines
11 KiB
Python
from unittest import mock
|
|
|
|
import pytest
|
|
from ophyd_devices.tests.utils import SocketMock
|
|
|
|
from csaxs_bec.devices.mcs2 import Mcs2Controller
|
|
from csaxs_bec.devices.mcs2.mcs2_controller import Mcs2ChannelType
|
|
from csaxs_bec.devices.mcs2.mcs2_errors import Mcs2CommunicationError, Mcs2ErrorCode
|
|
from csaxs_bec.devices.mcs2.mcs2_ophyd import Mcs2Motor
|
|
from csaxs_bec.devices.sim.sim_mcs2 import SimMcs2Motor
|
|
from csaxs_bec.devices.sim.sim_socket import SimStateRegistry
|
|
|
|
NO_ERROR = b'0,"No Error"\r\n'
|
|
|
|
|
|
@pytest.fixture
|
|
def controller(dm_with_devices):
|
|
Mcs2Controller._reset_controller()
|
|
controller = Mcs2Controller(
|
|
socket_cls=SocketMock, socket_host="dummy", socket_port=55551, device_manager=dm_with_devices
|
|
)
|
|
controller.on()
|
|
controller.sock.flush_buffer()
|
|
yield controller
|
|
|
|
|
|
@pytest.fixture
|
|
def lmcs2A(dm_with_devices):
|
|
Mcs2Controller._reset_controller()
|
|
motor_a = Mcs2Motor(
|
|
"A",
|
|
name="lmcs2A",
|
|
host="dummy",
|
|
port=55551,
|
|
sign=1,
|
|
socket_cls=SocketMock,
|
|
device_manager=dm_with_devices,
|
|
)
|
|
motor_a.controller.on()
|
|
motor_a.controller.sock.flush_buffer()
|
|
motor_a.stage()
|
|
yield motor_a
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis,position,get_message,return_msg",
|
|
[
|
|
(0, 50, b":CHAN0:POS?\r\n", b"50000000000\r\n"),
|
|
(1, 0, b":CHAN1:POS?\r\n", b"0\r\n"),
|
|
(0, -50, b":CHAN0:POS?\r\n", b"-50000000000\r\n"),
|
|
(0, -50.23, b":CHAN0:POS?\r\n", b"-50230000000\r\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
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"axis,is_referenced,get_message,return_msg,exception",
|
|
[
|
|
(0, True, b":CHAN0:STAT?\r\n", b"128\r\n", None),
|
|
(1, True, b":CHAN1:STAT?\r\n", b"129\r\n", None),
|
|
(0, False, b":CHAN0:STAT?\r\n", b"0\r\n", None),
|
|
(200, False, b":CHAN0:STAT?\r\n", b"0\r\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(
|
|
"is_moving,get_message,return_msg",
|
|
[
|
|
(False, b":CHAN0:STAT?\r\n", b"0\r\n"),
|
|
(True, b":CHAN0:STAT?\r\n", b"1\r\n"), # ACTIVELY_MOVING
|
|
(True, b":CHAN0:STAT?\r\n", b"8\r\n"), # REFERENCING
|
|
(True, b":CHAN0:STAT?\r\n", b"4\r\n"), # CALIBRATING
|
|
(True, b":CHAN0:STAT?\r\n", b"16\r\n"), # MOVE_DELAYED
|
|
(False, b":CHAN0:STAT?\r\n", b"32\r\n"), # SENSOR_PRESENT only
|
|
(False, b":CHAN0:STAT?\r\n", b"128\r\n"), # IS_REFERENCED only
|
|
(True, b":CHAN0:STAT?\r\n", b"129\r\n"), # ACTIVELY_MOVING | IS_REFERENCED
|
|
],
|
|
)
|
|
def test_is_axis_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_command_raises_scpi_error_code(controller):
|
|
controller.sock.buffer_recv = b'-101,"Invalid character"\r\n'
|
|
with pytest.raises(Mcs2ErrorCode) as exc_info:
|
|
controller.command(":BOGUS")
|
|
assert exc_info.value.error_code == -101
|
|
|
|
|
|
def test_command_raises_mcs2_error_code(controller):
|
|
controller.sock.buffer_recv = b'514,"Range limit reached"\r\n'
|
|
with pytest.raises(Mcs2ErrorCode) as exc_info:
|
|
controller.command(":MOVE0 1000000000000")
|
|
assert exc_info.value.error_code == 514
|
|
|
|
|
|
def test_command_no_error_does_not_raise(controller):
|
|
controller.sock.buffer_recv = NO_ERROR
|
|
controller.command(":STOP0")
|
|
assert controller.sock.buffer_put == [b":STOP0\r\n", b":SYST:ERR:NEXT?\r\n"]
|
|
|
|
|
|
def test_query_timeout_without_error_raises_communication_error(controller):
|
|
# _send_and_read is patched directly (rather than driving the real 1s socket
|
|
# timeout loop via buffer_recv) to keep this test fast and deterministic: the
|
|
# first call simulates the original query timing out (empty response), the
|
|
# second simulates the subsequent error-queue poll finding nothing queued.
|
|
with mock.patch.object(controller, "_send_and_read", side_effect=["", '0,"No Error"']):
|
|
with pytest.raises(Mcs2CommunicationError):
|
|
controller.query(":CHAN0:POS?")
|
|
|
|
|
|
def test_query_timeout_with_queued_error_raises_error_code(controller):
|
|
with mock.patch.object(
|
|
controller, "_send_and_read", side_effect=["", '-113,"Undefined header"']
|
|
):
|
|
with pytest.raises(Mcs2ErrorCode) as exc_info:
|
|
controller.query(":CHAN0:BOGUS?")
|
|
assert exc_info.value.error_code == -113
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"pos,axis,hold_time,get_msg",
|
|
[
|
|
(50, 0, None, [b":CHAN0:MMOD 0\r\n", b":CHAN0:HOLD 1000\r\n", b":MOVE0 50000000000\r\n"]),
|
|
(0, 0, 800, [b":CHAN0:MMOD 0\r\n", b":CHAN0:HOLD 800\r\n", b":MOVE0 0\r\n"]),
|
|
(20.23, 1, None, [b":CHAN1:MMOD 0\r\n", b":CHAN1:HOLD 1000\r\n", b":MOVE1 20230000000\r\n"]),
|
|
],
|
|
)
|
|
def test_move_axis_to_absolute_position(controller, pos, axis, hold_time, get_msg):
|
|
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR, NO_ERROR]
|
|
if hold_time is not None:
|
|
controller.move_axis_to_absolute_position(axis, pos, hold_time=hold_time)
|
|
else:
|
|
controller.move_axis_to_absolute_position(axis, pos)
|
|
expected = []
|
|
for msg in get_msg:
|
|
expected.append(msg)
|
|
expected.append(b":SYST:ERR:NEXT?\r\n")
|
|
assert controller.sock.buffer_put == expected
|
|
|
|
|
|
def test_move_open_loop_steps(controller):
|
|
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR, NO_ERROR, NO_ERROR]
|
|
controller.move_open_loop_steps(0, 500, amplitude=65535, frequency=1000)
|
|
assert controller.sock.buffer_put == [
|
|
b":CHAN0:STEP:FREQ 1000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:STEP:AMPL 65535\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:MMOD 4\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":MOVE0 500\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
]
|
|
|
|
|
|
def test_stop_all_axes(lmcs2A):
|
|
controller = lmcs2A.controller
|
|
controller.sock.buffer_recv = NO_ERROR
|
|
controller.stop_all_axes()
|
|
assert controller.sock.buffer_put == [b":STOP0\r\n", b":SYST:ERR:NEXT?\r\n"]
|
|
|
|
|
|
def test_find_reference_mark(controller):
|
|
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR, NO_ERROR, NO_ERROR]
|
|
controller.find_reference_mark(0, direction=1, autoZero=1, velocity=1.0, acceleration=10.0)
|
|
assert controller.sock.buffer_put == [
|
|
b":CHAN0:REF:OPT 6\r\n", # REVERSE_DIR (0x2) | AUTO_ZERO (0x4)
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:VEL 1000000000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:ACC 10000000000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":REF0\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"move_speed,axis,get_msg",
|
|
[(50, 0, b":CHAN0:VEL 50000000000\r\n"), (0, 0, b":CHAN0:VEL 0\r\n"), (20.23, 1, b":CHAN1:VEL 20230000000\r\n")],
|
|
)
|
|
def test_set_closed_loop_move_speed(controller, move_speed, axis, get_msg):
|
|
controller.sock.buffer_recv = NO_ERROR
|
|
controller.set_closed_loop_move_speed(axis, move_speed)
|
|
assert controller.sock.buffer_put == [get_msg, b":SYST:ERR:NEXT?\r\n"]
|
|
|
|
|
|
def test_get_closed_loop_move_speed(controller):
|
|
controller.sock.buffer_recv = b"1000000000\r\n"
|
|
val = controller.get_closed_loop_move_speed(0)
|
|
assert val == 1.0
|
|
assert controller.sock.buffer_put[0] == b":CHAN0:VEL?\r\n"
|
|
|
|
|
|
def test_get_position_limits(controller):
|
|
controller.sock.buffer_recv = [b"-1000000000\r\n", b"1000000000\r\n"]
|
|
val = controller.get_position_limits(0)
|
|
assert val == [-1.0, 1.0]
|
|
assert controller.sock.buffer_put == [b":CHAN0:RLIM:MIN?\r\n", b":CHAN0:RLIM:MAX?\r\n"]
|
|
|
|
|
|
def test_set_position_limits(controller):
|
|
controller.sock.buffer_recv = [NO_ERROR, NO_ERROR]
|
|
controller.set_position_limits(0, -1.0, 1.0)
|
|
assert controller.sock.buffer_put == [
|
|
b":CHAN0:RLIM:MIN -1000000000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:RLIM:MAX 1000000000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
]
|
|
|
|
|
|
def test_get_positioner_type(controller):
|
|
controller.sock.buffer_recv = b'"SLC-24180-S"\r\n'
|
|
val = controller.get_positioner_type(0)
|
|
assert val == "SLC-24180-S"
|
|
assert controller.sock.buffer_put[0] == b":CHAN0:PTYP:NAME?\r\n"
|
|
|
|
|
|
def test_get_channel_type(controller):
|
|
controller.sock.buffer_recv = b"1\r\n"
|
|
val = controller.get_channel_type(0)
|
|
assert val == Mcs2ChannelType.STICK_SLIP_PIEZO_DRIVER
|
|
|
|
|
|
def test_get_number_of_channels(controller):
|
|
controller.sock.buffer_recv = b"9\r\n"
|
|
assert controller.get_number_of_channels() == 9
|
|
assert controller.sock.buffer_put[0] == b":DEV:NOCH?\r\n"
|
|
|
|
|
|
def test_get_idn(controller):
|
|
controller.sock.buffer_recv = b'"SmarAct;MCS2-00001234;TestDevice;09/22/23"\r\n'
|
|
assert controller.get_idn() == "SmarAct;MCS2-00001234;TestDevice;09/22/23"
|
|
|
|
|
|
def test_all_axes_referenced(lmcs2A):
|
|
controller = lmcs2A.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_move_axis(lmcs2A):
|
|
controller = lmcs2A.controller
|
|
controller.sock.buffer_recv = [
|
|
b"128\r\n", # axis_is_referenced() -> STAT? (IS_REFERENCED only, not moving)
|
|
NO_ERROR, # MMOD command error-check
|
|
NO_ERROR, # HOLD command error-check
|
|
NO_ERROR, # MOVE command error-check
|
|
b"128\r\n", # background thread's is_axis_moving() -> STAT? (not moving)
|
|
b"50000000000\r\n", # final readback.read() -> POS?
|
|
]
|
|
lmcs2A.move(50)
|
|
assert controller.sock.buffer_put == [
|
|
b":CHAN0:STAT?\r\n",
|
|
b":CHAN0:MMOD 0\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:HOLD 1000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":MOVE0 50000000000\r\n",
|
|
b":SYST:ERR:NEXT?\r\n",
|
|
b":CHAN0:STAT?\r\n",
|
|
b":CHAN0:POS?\r\n",
|
|
]
|
|
|
|
|
|
def test_sim_mcs2_end_to_end(dm_with_devices):
|
|
"""Exercises the simulated MCS2 backend (SimMcs2Socket/SimMcs2State) end-to-end
|
|
through the real Mcs2Motor/Mcs2Controller classes, as a sanity check independent of
|
|
the exact-byte-sequence unit tests above."""
|
|
Mcs2Controller._reset_controller()
|
|
SimStateRegistry.reset()
|
|
mot = SimMcs2Motor(
|
|
"A",
|
|
name="simmcs2a",
|
|
host="sim-mcs2",
|
|
port=55551,
|
|
sign=1,
|
|
device_manager=dm_with_devices,
|
|
sim_initial_position=1.0,
|
|
sim_velocity=5.0,
|
|
sim_referenced=True,
|
|
)
|
|
mot.controller.on()
|
|
mot.stage()
|
|
try:
|
|
assert mot.controller.axis_is_referenced(0) is True
|
|
assert abs(mot.read()[mot.name]["value"] - 1.0) < 1e-6
|
|
status = mot.move(2.0, wait=True)
|
|
assert status.done
|
|
assert abs(mot.read()[mot.name]["value"] - 2.0) < 0.5
|
|
mot.stop()
|
|
finally:
|
|
mot.unstage()
|
|
mot.controller.off()
|