From fc17c3a52f5ac3d361cc62966de1bfc164bd5ed6 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Tue, 15 Sep 2026 12:03:20 +0200 Subject: [PATCH] fix: adjust controller-based devices to new base class --- csaxs_bec/devices/mcs2/mcs2_controller.py | 49 +++---------------- csaxs_bec/devices/omny/galil/galil_ophyd.py | 20 +++----- csaxs_bec/devices/sim/sim_galil.py | 7 ++- csaxs_bec/devices/sim/sim_socket.py | 2 +- .../devices/smaract/smaract_controller.py | 39 ++++----------- tests/tests_devices/test_fupr_ophyd.py | 6 +-- tests/tests_devices/test_galil.py | 42 ++++++++++++---- tests/tests_devices/test_galil_flomni.py | 38 +++++++++++--- tests/tests_devices/test_mcs2.py | 22 +++++++-- tests/tests_devices/test_rt_flomni.py | 28 ++++++++--- tests/tests_devices/test_smaract.py | 28 +++++------ 11 files changed, 148 insertions(+), 133 deletions(-) diff --git a/csaxs_bec/devices/mcs2/mcs2_controller.py b/csaxs_bec/devices/mcs2/mcs2_controller.py index 1523eaf8..f66e1cf0 100644 --- a/csaxs_bec/devices/mcs2/mcs2_controller.py +++ b/csaxs_bec/devices/mcs2/mcs2_controller.py @@ -87,6 +87,9 @@ class Mcs2Controller(Controller): _axes_per_controller = 9 _initialized = False PM_PER_MM = 1e9 + put_trail_char = "\r\n" + put_lead_char = "" + get_trail_char = "\r\n" USER_ACCESS = [ "query", @@ -132,50 +135,14 @@ class Mcs2Controller(Controller): ) self._initialized = True - @threadlocked - def socket_put(self, cmd: str) -> None: - """Send a raw command. ``cmd`` must already include its own leading ':' or '*'.""" - self.command_history.append(f"[PUT]: {cmd}") - self.sock.put(f"{cmd}\r\n".encode()) - - @threadlocked - def _read_response(self) -> str: - """Read a single response, waiting for the terminator. Returns an - empty string if no response arrived within the timeout (this is expected for - set/move commands, and can also happen for malformed queries).""" - 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("\r\n"): - break - time.sleep(sleep_time) - elapsed_time += sleep_time - if elapsed_time > max_wait_time: - break - return self._remove_trailing_characters(return_val) - - def _remove_trailing_characters(self, var: str) -> str: - if len(var) > 1: - return var.split("\r\n")[0] - return var - - @threadlocked - def _send_and_read(self, cmd: str) -> str: - self.socket_put(cmd) - return self._read_response() - def get_error_count(self) -> int: """Return the number of errors currently in the device's error queue.""" - return int(self._send_and_read(":SYST:ERR:COUN?")) + return int(self.socket_put_and_receive(":SYST:ERR:COUN?")) def get_next_error(self) -> tuple[int, str]: """Pop and return the next (oldest) error from the device's error queue as (code, message). code == 0 means the queue was empty ("No Error").""" - raw = self._send_and_read(":SYST:ERR:NEXT?") + raw = self.socket_put_and_receive(":SYST:ERR:NEXT?") code_str, _, message = raw.partition(",") return int(code_str), message.strip().strip('"') @@ -198,7 +165,7 @@ class Mcs2Controller(Controller): """Send a query (``cmd`` must end with '?') and return the bare value as a string (quotes stripped for string properties). Numeric callers are expected to convert the result themselves, e.g. ``int(controller.query(...))``.""" - raw = self._send_and_read(cmd) + raw = self.socket_put_and_receive(cmd) if raw == "": code, message = self.get_next_error() if code != 0: @@ -283,9 +250,7 @@ class Mcs2Controller(Controller): @retry_once def stop_all_axes(self): - return [ - self.command(f":STOP{ax.axis_Id_numeric}") for ax in self._axis if ax is not None - ] + return [self.command(f":STOP{ax.axis_Id_numeric}") for ax in self._axis if ax is not None] @retry_once @axis_checked diff --git a/csaxs_bec/devices/omny/galil/galil_ophyd.py b/csaxs_bec/devices/omny/galil/galil_ophyd.py index de0ce5ff..d3eda488 100644 --- a/csaxs_bec/devices/omny/galil/galil_ophyd.py +++ b/csaxs_bec/devices/omny/galil/galil_ophyd.py @@ -2,9 +2,9 @@ This module contains the base class for Galil controllers as well as the signals used for Galil devices. """ -import functools +from __future__ import annotations + import time -from typing import Any from bec_lib import bec_logger from ophyd.utils import ReadOnlyError @@ -56,10 +56,9 @@ class GalilController(Controller): FAIL = "\033[91m" ENDC = "\033[0m" - @threadlocked - def socket_put(self, val: str) -> None: - self.command_history.append(f"[PUT]: {val}") - self.sock.put(f"{val}\r".encode()) + get_trail_char = "\r" + put_lead_char = "" + put_trail_char = "\r" @retry_once def socket_put_confirmed(self, val: str) -> None: @@ -114,8 +113,7 @@ class GalilController(Controller): def stop_all_axes(self) -> str: if not self.is_thread_active(1): return self.socket_put_and_receive("XQ#STOP,1") - else: - return ":" + return ":" def hard_abort_and_restore_positioning_mode( self, transfer_thread_id: int = 3, timeout: float = 5.0 @@ -161,8 +159,7 @@ class GalilController(Controller): # this is only valid for omny. consider moving to ogalil voltage = float(self.socket_put_and_receive(f"MG @AN[{axis_Id_numeric+1}]").strip()) voltage2 = float(self.socket_put_and_receive(f"MG @AN[{axis_Id_numeric+1}]").strip()) - if voltage2 < voltage: - voltage = voltage2 + voltage = min(voltage, voltage2) # convert from [-10,10]V to [0,300]degC temperature_degC = round((voltage + 10.0) / 20.0 * 300.0, 1) @@ -239,8 +236,7 @@ class GalilController(Controller): if not limit: raise GalilError(f"Failed to drive axis {axis_Id}/{axis_Id_numeric} to limit.") - else: - print("Limit reached.") + print("Limit reached.") def find_reference(self, axis_Id_numeric: int, verbose=0, raise_error=1) -> None: """ diff --git a/csaxs_bec/devices/sim/sim_galil.py b/csaxs_bec/devices/sim/sim_galil.py index f80e2a6b..d608650d 100644 --- a/csaxs_bec/devices/sim/sim_galil.py +++ b/csaxs_bec/devices/sim/sim_galil.py @@ -327,7 +327,12 @@ class SimGalilSocket(SimSocketBase): state_cls = SimGalilState - def handle_command(self, line: str): # noqa: C901 + def handle_command(self, line: str): + """Terminate each reply for the Galil controller's receive loop.""" + reply = self._handle_command(line) + return reply + "\r" if reply is not None else None + + def _handle_command(self, line: str): # noqa: C901 state: SimGalilState = self.state if line.startswith("MG"): diff --git a/csaxs_bec/devices/sim/sim_socket.py b/csaxs_bec/devices/sim/sim_socket.py index a52adaa3..3a747b68 100644 --- a/csaxs_bec/devices/sim/sim_socket.py +++ b/csaxs_bec/devices/sim/sim_socket.py @@ -99,7 +99,7 @@ class SimSocketBase: if reply is not None: self._recv_buffer.append(reply.encode()) - def receive(self, buffer_length=1024): + def receive(self, buffer_length=1024, timeout: int = 2): with self._lock: if self._recv_buffer: return self._recv_buffer.pop(0) diff --git a/csaxs_bec/devices/smaract/smaract_controller.py b/csaxs_bec/devices/smaract/smaract_controller.py index 5eca450c..b810873b 100644 --- a/csaxs_bec/devices/smaract/smaract_controller.py +++ b/csaxs_bec/devices/smaract/smaract_controller.py @@ -70,6 +70,9 @@ class SmaractController(Controller): "is_axis_moving", "print_command_history", ] + get_trail_char = "\n" + put_lead_char = ":" + put_trail_char = "\n" def __init__( self, @@ -98,34 +101,15 @@ class SmaractController(Controller): ) self._sensors = SmaractSensors() - @threadlocked - def socket_put(self, val: str): - self.command_history.append(f"[PUT]: {val}") - self.sock.put(f":{val}\n".encode()) - @threadlocked def socket_put_and_receive( - self, val: str, remove_trailing_chars=True, check_for_errors=True, raise_if_not_status=False + self, val: str, remove_trailing_chars=True, timeout: int = 2, raise_if_not_status=False ) -> str: - self.socket_put(val) - 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 - if remove_trailing_chars: - return_val = self._remove_trailing_characters(return_val) - logger.debug(f"Sending {val}; Returned {return_val}") - if check_for_errors: - self._check_for_error(return_val, raise_if_not_status=raise_if_not_status) + + return_val = super().socket_put_and_receive( + val, remove_trailing_chars=remove_trailing_chars, timeout=timeout + ) + self._check_for_error(return_val, raise_if_not_status=raise_if_not_status) return return_val @retry_once @@ -492,11 +476,6 @@ class SmaractController(Controller): "Expected error / status message but failed to parse it." ) - def _remove_trailing_characters(self, var: str) -> str: - if len(var) > 1: - return var.split("\n")[0] - return var - def _message_starts_with(self, msg: str, leading_chars: str) -> bool: if msg.startswith(leading_chars): return True diff --git a/tests/tests_devices/test_fupr_ophyd.py b/tests/tests_devices/test_fupr_ophyd.py index 9e568563..b96b0ec3 100644 --- a/tests/tests_devices/test_fupr_ophyd.py +++ b/tests/tests_devices/test_fupr_ophyd.py @@ -33,8 +33,8 @@ def fsamroy(dm_with_devices): @pytest.mark.parametrize( "pos,msg_received,msg_put,sign", [ - (-0.5, b" -12800\n\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], 1), - (-0.5, b" 12800\n\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], -1), + (-0.5, b" -12800\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], 1), + (-0.5, b" 12800\r", [b"TPA\r", b"MG_BGA\r", b"TPA\r"], -1), ], ) def test_axis_get(fsamroy, pos, msg_received, msg_put, sign): @@ -54,7 +54,7 @@ def test_axis_get(fsamroy, pos, msg_received, msg_put, sign): ( 0, [b"MG axisref\r", b"PAA=0\r", b"PAA=0\r", b"BGA\r"], - [b"1.00", b"-1", b":", b":", b":", b":", b"-1"], + [b"1.00\r", b"-1\r", b":\r", b":\r", b":\r", b":\r", b"-1\r"], ) ], ) diff --git a/tests/tests_devices/test_galil.py b/tests/tests_devices/test_galil.py index ac49678e..5732dd31 100644 --- a/tests/tests_devices/test_galil.py +++ b/tests/tests_devices/test_galil.py @@ -57,7 +57,7 @@ def leyex(dm_with_devices): leyex_motor.controller._reset_controller() -@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\n\r", 1), (-1, b" -12800\n\r", -1)]) +@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\r", 1), (-1, b" -12800\r", -1)]) def test_axis_get(leyey, pos, msg, sign): leyey.sign = sign leyey.controller.sock.flush_buffer() @@ -81,7 +81,7 @@ def test_axis_get(leyey, pos, msg, sign): b"XQ#NEWPAR\r", b"MG_XQ0\r", ], - [b"1.00", b"-1", b":", b":", b":", b":", b"-1"], + [b"1.00\r", b"-1\r", b":\r", b":\r", b":\r", b":\r", b"-1\r"], ) ], ) @@ -110,7 +110,18 @@ def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages): b"MG_XQ2\r", b"MG _LRA, _LFA\r", ], - [b":", b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"1.000 0.000"], + [ + b":\r", + b":\r", + b":\r", + b"-1\r", + b":\r", + b"0\r", + b"0\r", + b"-1\r", + b"-1\r", + b"1.000 0.000\r", + ], ), ( 1, @@ -127,7 +138,18 @@ def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages): b"MG_XQ2\r", b"MG _LRB, _LFB\r", ], - [b":", b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"0.000 1.000"], + [ + b":\r", + b":\r", + b":\r", + b"-1\r", + b":\r", + b"0\r", + b"0\r", + b"-1\r", + b"-1\r", + b"0.000 1.000\r", + ], ), ], ) @@ -154,7 +176,7 @@ def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, soc b"MG_XQ2\r", b"MG axisref[0]\r", ], - [b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"1.00"], + [b":\r", b":\r", b"-1\r", b":\r", b"0\r", b"0\r", b"-1\r", b"-1\r", b"1.00\r"], ), ( 1, @@ -169,7 +191,7 @@ def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, soc b"MG_XQ2\r", b"MG axisref[1]\r", ], - [b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"1.00"], + [b":\r", b":\r", b"-1\r", b":\r", b"0\r", b"0\r", b"-1\r", b"-1\r", b"1.00\r"], ), ], ) @@ -328,7 +350,7 @@ def test_galil_rio_signal_read(galil_rio): value_callback_buffer.append(readback) galil_rio.analog_in.ch0.subscribe(value_callback, run=False) - galil_rio.controller.sock.buffer_recv = [b" 2.5 2.6 2.7 2.8 2.9 3.0 3.1 3.2"] + galil_rio.controller.sock.buffer_recv = [b" 2.5 2.6 2.7 2.8 2.9 3.0 3.1 3.2\r\n"] expected_values = [2.5, 2.6, 2.7, 2.8, 2.9, 3.0, 3.1, 3.2] ################## @@ -381,7 +403,7 @@ def test_galil_rio_digital_out_signal(galil_rio): for ii in range(galil_rio.digital_out.ch0._NUM_DIGITAL_OUTPUT_CHANNELS): cmd = f"MG@OUT[{ii}]\r".encode() excepted_put_buffer.append(cmd) - recv = " 1.000".encode() + recv = b" 1.000\r\n" buffer_receive.append(recv) galil_rio.controller.sock.buffer_recv = buffer_receive # Mock response for readback @@ -395,7 +417,7 @@ def test_galil_rio_digital_out_signal(galil_rio): # Test writing to digital output channels galil_rio.controller.sock.buffer_put = [] # Clear buffer put - galil_rio.controller.sock.buffer_recv = [b":"] # Mock response for readback + galil_rio.controller.sock.buffer_recv = [b":\r\n"] # Mock response for readback # Set digital output channel 0 to high galil_rio.digital_out.ch0.put(1) @@ -403,6 +425,6 @@ def test_galil_rio_digital_out_signal(galil_rio): # Set digital output channel 0 to low galil_rio.controller.sock.buffer_put = [] # Clear buffer put - galil_rio.controller.sock.buffer_recv = [b":"] # Mock response for readback + galil_rio.controller.sock.buffer_recv = [b":\r\n"] # Mock response for readback galil_rio.digital_out.ch0.put(0) assert galil_rio.controller.sock.buffer_put == [b"CB0\r"] diff --git a/tests/tests_devices/test_galil_flomni.py b/tests/tests_devices/test_galil_flomni.py index 64b09cb2..d692aab7 100644 --- a/tests/tests_devices/test_galil_flomni.py +++ b/tests/tests_devices/test_galil_flomni.py @@ -40,7 +40,7 @@ def leyex(dm_with_devices): leyex_motor.controller._reset_controller() -@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\n\r", 1), (-1, b" -12800\n\r", -1)]) +@pytest.mark.parametrize("pos,msg,sign", [(1, b" -12800\r", 1), (-1, b" -12800\r", -1)]) def test_axis_get(leyey, pos, msg, sign): leyey.sign = sign leyey.controller.sock.flush_buffer() @@ -64,7 +64,7 @@ def test_axis_get(leyey, pos, msg, sign): b"XQ#NEWPAR\r", b"MG_XQ0\r", ], - [b"1.00", b"-1", b":", b":", b":", b":", b"-1"], + [b"1.00\r", b"-1\r", b":\r", b":\r", b":\r", b":\r", b"-1\r"], ) ], ) @@ -93,7 +93,18 @@ def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages): b"MG _MOA\r", b"MG _LRA, _LFA\r", ], - [b":", b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"1.000 0.000"], + [ + b":\r", + b":\r", + b":\r", + b"-1\r", + b":\r", + b"0\r", + b"0\r", + b"-1\r", + b"-1\r", + b"1.000 0.000\r", + ], ), ( 1, @@ -110,7 +121,18 @@ def test_axis_put(leyey, target_pos, socket_put_messages, socket_get_messages): b"MG _MOB\r", b"MG _LRB, _LFB\r", ], - [b":", b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"0.000 1.000"], + [ + b":\r", + b":\r", + b":\r", + b"-1\r", + b":\r", + b"0\r", + b"0\r", + b"-1\r", + b"-1\r", + b"0.000 1.000\r", + ], ), ], ) @@ -137,7 +159,7 @@ def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, soc b"MG _MOA\r", b"MG axisref[0]\r", ], - [b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"1.00"], + [b":\r", b":\r", b"-1\r", b":\r", b"0\r", b"0\r", b"-1\r", b"-1\r", b"1.00\r"], ), ( 1, @@ -152,7 +174,7 @@ def test_drive_axis_to_limit(leyex, axis_nr, direction, socket_put_messages, soc b"MG _MOB\r", b"MG axisref[1]\r", ], - [b":", b":", b"-1", b":", b"0", b"0", b"-1", b"-1", b"1.00"], + [b":\r", b":\r", b"-1\r", b":\r", b"0\r", b"0\r", b"-1\r", b"-1\r", b"1.00\r"], ), ], ) @@ -166,8 +188,8 @@ def test_find_reference(leyex, axis_nr, socket_put_messages, socket_get_messages @pytest.mark.parametrize( "axis_Id,socket_put_messages,socket_get_messages,triggered", [ - ("A", [b"MG @IN[14]\r"], [b" 1.0000\n"], True), - ("B", [b"MG @IN[14]\r"], [b" 0.0000\n"], False), + ("A", [b"MG @IN[14]\r"], [b" 1.0000\r"], True), + ("B", [b"MG @IN[14]\r"], [b" 0.0000\r"], False), ], ) def test_fosaz_light_curtain_is_triggered( diff --git a/tests/tests_devices/test_mcs2.py b/tests/tests_devices/test_mcs2.py index 8f8ac461..4c51ed11 100644 --- a/tests/tests_devices/test_mcs2.py +++ b/tests/tests_devices/test_mcs2.py @@ -17,7 +17,10 @@ NO_ERROR = b'0,"No Error"\r\n' def controller(dm_with_devices): Mcs2Controller._reset_controller() controller = Mcs2Controller( - socket_cls=SocketMock, socket_host="dummy", socket_port=55551, device_manager=dm_with_devices + socket_cls=SocketMock, + socket_host="dummy", + socket_port=55551, + device_manager=dm_with_devices, ) controller.on() controller.sock.flush_buffer() @@ -123,14 +126,14 @@ def test_query_timeout_without_error_raises_communication_error(controller): # 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 mock.patch.object(controller, "socket_put_and_receive", 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"'] + controller, "socket_put_and_receive", side_effect=["", '-113,"Undefined header"'] ): with pytest.raises(Mcs2ErrorCode) as exc_info: controller.query(":CHAN0:BOGUS?") @@ -142,7 +145,12 @@ def test_query_timeout_with_queued_error_raises_error_code(controller): [ (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"]), + ( + 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): @@ -197,7 +205,11 @@ def test_find_reference_mark(controller): @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")], + [ + (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 diff --git a/tests/tests_devices/test_rt_flomni.py b/tests/tests_devices/test_rt_flomni.py index 811d6571..01a0b307 100644 --- a/tests/tests_devices/test_rt_flomni.py +++ b/tests/tests_devices/test_rt_flomni.py @@ -17,6 +17,7 @@ def rt_flomni(): socket_port=8081, device_manager=mock.MagicMock(), ) + rt_flomni.on() with mock.patch.object(rt_flomni, "sock"): rtx = mock.MagicMock(spec=RtFlomniMotor) rtx.name = "rtx" @@ -34,6 +35,7 @@ def rt_flomni(): rt_flomni.set_axis(axis=rty, axis_nr=1) rt_flomni.set_axis(axis=rtz, axis_nr=2) yield rt_flomni + rt_flomni.off(update_config=False) RtFlomniController._reset_controller() @@ -46,7 +48,7 @@ def test_rt_flomni_move_to_zero(rt_flomni): ] -@pytest.mark.parametrize("return_value,is_running", [(b"1.00\n", False), (b"0.00\n", True)]) +@pytest.mark.parametrize("return_value,is_running", [(b"1.00\r\n", False), (b"0.00\r\n", True)]) def test_rt_flomni_feedback_is_running(rt_flomni, return_value, is_running): rt_flomni.sock.receive.return_value = return_value assert rt_flomni.feedback_is_running() == is_running @@ -54,7 +56,8 @@ def test_rt_flomni_feedback_is_running(rt_flomni, return_value, is_running): def test_feedback_enable_with_reset(rt_flomni): - + # Cyclic error compensation status (w0, w1), followed by PID voltage (g). + rt_flomni.sock.receive.side_effect = [b"32\r\n", b"32\r\n", b"0.05\r\n"] device_manager = rt_flomni.device_manager device_manager.devices.fsamx.user_parameter.get.return_value = 0.05 device_manager.devices.fsamx.obj.readback.get.return_value = 0.05 @@ -69,13 +72,24 @@ def test_feedback_enable_with_reset(rt_flomni): rt_flomni.feedback_enable_with_reset() laser_tracker_on.assert_called_once() + assert rt_flomni.rt_pid_voltage == 0.05 + device_manager.devices.rtx.update_user_parameter.assert_called_once_with( + {"rt_pid_voltage": 0.05} + ) + assert rt_flomni.sock.put.call_args_list[-3:] == [ + mock.call(b"w0\n"), + mock.call(b"w1\n"), + mock.call(b"g\n"), + ] + def test_move_samx_to_scan_region(rt_flomni): - device_manager = rt_flomni.device_manager - device_manager.devices.rtx.user_parameter.get.return_value = 1 - rt_flomni.move_samx_to_scan_region(20, 2) - assert mock.call(b"v0\n") not in rt_flomni.sock.put.mock_calls - assert mock.call(b"v1\n") in rt_flomni.sock.put.mock_calls + with mock.patch.object(rt_flomni, "get_pid_x", return_value=1): + device_manager = rt_flomni.device_manager + device_manager.devices.rtx.user_parameter.get.return_value = 1 + rt_flomni.move_samx_to_scan_region(20, 2) + assert mock.call(b"v0\n") not in rt_flomni.sock.put.mock_calls + assert mock.call(b"v1\n") in rt_flomni.sock.put.mock_calls def test_feedback_enable_without_reset(rt_flomni): diff --git a/tests/tests_devices/test_smaract.py b/tests/tests_devices/test_smaract.py index 3e77ca03..c77a1e62 100644 --- a/tests/tests_devices/test_smaract.py +++ b/tests/tests_devices/test_smaract.py @@ -78,24 +78,24 @@ def test_axis_is_referenced(controller, axis, is_referenced, get_message, return "return_msg,exception,raised", [ (b"false\n", SmaractCommunicationError, False), - (b":E0,1", SmaractErrorCode, True), - (b":E,1", SmaractCommunicationError, True), - (b":E,-1", SmaractCommunicationError, True), + (b":E0,1\n", SmaractErrorCode, True), + (b":E,1\n", SmaractCommunicationError, True), + (b":E,-1\n", SmaractCommunicationError, True), ], ) def test_socket_put_and_receive_raises_exception(controller, return_msg, exception, raised): controller.sock.buffer_recv = return_msg with pytest.raises(exception): - controller.socket_put_and_receive(b"test", raise_if_not_status=True) + controller.socket_put_and_receive("test", raise_if_not_status=True) controller.sock.flush_buffer() controller.sock.buffer_recv = return_msg if raised: with pytest.raises(exception): - controller.socket_put_and_receive(b"test") + controller.socket_put_and_receive("test") else: - assert controller.socket_put_and_receive(b"test") == return_msg.split(b"\n")[0].decode() + assert controller.socket_put_and_receive("test") == return_msg.split(b"\n")[0].decode() @pytest.mark.parametrize( @@ -120,7 +120,7 @@ def test_communication_mode(controller, mode, get_message, return_msg): (0, b":GS0\n", b":S0,6\n"), (1, b":GS0\n", b":S0,7\n"), (0, b":GS0\n", b":S0,9\n"), - (0, [b":GS0\n", b":GS0\n"], [b":E0,0\n", b":S0,9"]), + (0, [b":GS0\n", b":GS0\n"], [b":E0,0\n", b":S0,9\n"]), ], ) def test_axis_is_moving(controller, is_moving, get_message, return_msg): @@ -149,9 +149,9 @@ def test_get_sensor_definition(controller, sensor_id, axis, get_msg, return_msg) @pytest.mark.parametrize( "move_speed,axis,get_msg,return_msg", [ - (50, 0, b":SCLS0,50000000\n", b":E-1,0"), - (0, 0, b":SCLS0,0\n", b":E-1,0"), - (20.23, 1, b":SCLS1,20230000\n", b":E-1,0"), + (50, 0, b":SCLS0,50000000\n", b":E-1,0\n"), + (0, 0, b":SCLS0,0\n", b":E-1,0\n"), + (20.23, 1, b":SCLS1,20230000\n", b":E-1,0\n"), ], ) def test_set_move_speed(controller, move_speed, axis, get_msg, return_msg): @@ -163,9 +163,9 @@ def test_set_move_speed(controller, move_speed, axis, get_msg, return_msg): @pytest.mark.parametrize( "pos,axis,hold_time,get_msg,return_msg", [ - (50, 0, None, b":MPA0,50000000,1000\n", b":E0,0"), - (0, 0, 800, b":MPA0,0,800\n", b":E0,0"), - (20.23, 1, None, b":MPA1,20230000,1000\n", b":E0,0"), + (50, 0, None, b":MPA0,50000000,1000\n", b":E0,0\n"), + (0, 0, 800, b":MPA0,0,800\n", b":E0,0\n"), + (20.23, 1, None, b":MPA1,20230000,1000\n", b":E0,0\n"), ], ) def test_move_axis_to_absolute_position(controller, pos, axis, hold_time, get_msg, return_msg): @@ -209,7 +209,7 @@ def test_move_axis(lsmarA, pos, get_msg, return_msg): assert controller.sock.buffer_put == get_msg -@pytest.mark.parametrize("num_axes,get_msg,return_msg", [(1, [b":S0\n"], [b":E0,0"])]) +@pytest.mark.parametrize("num_axes,get_msg,return_msg", [(1, [b":S0\n"], [b":E0,0\n"])]) def test_stop_axis(lsmarA, num_axes, get_msg, return_msg): controller = lsmarA.controller controller.sock.buffer_recv = return_msg