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

fix: adjust controller-based devices to new base class

Browse Source
This commit is contained in:
wakonig_k committed 2026-09-15 12:03:20 +02:00
1 parent fe4cb2e661
commit fc17c3a52f
11 files changed
+148 -133
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

csaxs_bec/devices/mcs2/mcs2_controller.py
+7 -42
View File
@@ -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 <CR><LF> 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
csaxs_bec/devices/omny/galil/galil_ophyd.py
+8 -12
View File
@@ -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:
"""
csaxs_bec/devices/sim/sim_galil.py
+6 -1
View File
@@ -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"):
csaxs_bec/devices/sim/sim_socket.py
+1 -1
View File
@@ -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)
csaxs_bec/devices/smaract/smaract_controller.py
+9 -30
View File
@@ -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
tests/tests_devices/test_fupr_ophyd.py
+3 -3
View File
@@ -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"],
)
],
)
tests/tests_devices/test_galil.py
+32 -10
View File
@@ -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"]
tests/tests_devices/test_galil_flomni.py
+30 -8
View File
@@ -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(
tests/tests_devices/test_mcs2.py
+17 -5
View File
@@ -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
tests/tests_devices/test_rt_flomni.py
+21 -7
View File
@@ -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):
tests/tests_devices/test_smaract.py
+14 -14
View File
@@ -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