mirror of
https://github.com/bec-project/ophyd_devices.git
synced 2025-06-23 19:27:59 +02:00
export from internal gitlab
This commit is contained in:
60
tests/test_galil.py
Normal file
60
tests/test_galil.py
Normal file
@ -0,0 +1,60 @@
|
||||
import pytest
|
||||
from ophyd_devices.galil.galil_ophyd import GalilMotor
|
||||
|
||||
from utils import SocketMock
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"pos,msg,sign",
|
||||
[
|
||||
(1, b" -12800\n\r", 1),
|
||||
(-1, b" -12800\n\r", -1),
|
||||
],
|
||||
)
|
||||
def test_axis_get(pos, msg, sign):
|
||||
leyey = GalilMotor(
|
||||
"H",
|
||||
name="leyey",
|
||||
host="mpc2680.psi.ch",
|
||||
port=8081,
|
||||
sign=sign,
|
||||
socket_cls=SocketMock,
|
||||
)
|
||||
leyey.controller.sock.flush_buffer()
|
||||
leyey.stage()
|
||||
leyey.controller.sock.buffer_recv = msg
|
||||
val = leyey.read()
|
||||
assert val["leyey"]["value"] == pos
|
||||
assert leyey.readback.get() == pos
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target_pos,socket_put_messages,socket_get_messages",
|
||||
[
|
||||
(
|
||||
0,
|
||||
[
|
||||
b"MG allaxref\r",
|
||||
b"MG_XQ0\r",
|
||||
b"naxis=7\r",
|
||||
b"ntarget=0.000\r",
|
||||
b"movereq=1\r",
|
||||
b"XQ#NEWPAR\r",
|
||||
],
|
||||
[
|
||||
b"1.00",
|
||||
b"-1",
|
||||
b":",
|
||||
b":",
|
||||
b":",
|
||||
b":",
|
||||
],
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_axis_put(target_pos, socket_put_messages, socket_get_messages):
|
||||
leyey = GalilMotor("H", name="leyey", host="mpc2680.psi.ch", port=8081, socket_cls=SocketMock)
|
||||
leyey.controller.sock.flush_buffer()
|
||||
leyey.controller.sock.buffer_recv = socket_get_messages
|
||||
leyey.user_setpoint.put(target_pos)
|
||||
assert leyey.controller.sock.buffer_put == socket_put_messages
|
||||
144
tests/test_npoint_piezo.py
Normal file
144
tests/test_npoint_piezo.py
Normal file
@ -0,0 +1,144 @@
|
||||
import pytest
|
||||
|
||||
from ophyd_devices.npoint import NPointController, NPointAxis
|
||||
|
||||
|
||||
class SocketMock:
|
||||
def __init__(self, sock=None):
|
||||
self.buffer_put = ""
|
||||
self.buffer_recv = ""
|
||||
self.is_open = False
|
||||
if sock is None:
|
||||
self.open()
|
||||
else:
|
||||
self.sock = sock
|
||||
|
||||
def connect(self, host, port):
|
||||
print(f"connecting to {host} port {port}")
|
||||
# self.sock.create_connection((host, port))
|
||||
# self.sock.connect((host, port))
|
||||
|
||||
def _put(self, msg_bytes):
|
||||
self.buffer_put = msg_bytes
|
||||
print(self.buffer_put)
|
||||
|
||||
def _recv(self, buffer_length=1024):
|
||||
print(self.buffer_recv)
|
||||
return self.buffer_recv
|
||||
|
||||
def _initialize_socket(self):
|
||||
pass
|
||||
|
||||
def put(self, msg):
|
||||
return self._put(msg)
|
||||
|
||||
def receive(self, buffer_length=1024):
|
||||
return self._recv(buffer_length=buffer_length)
|
||||
|
||||
def open(self):
|
||||
self._initialize_socket()
|
||||
self.is_open = True
|
||||
|
||||
def close(self):
|
||||
self.sock = None
|
||||
self.is_open = False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"pos,msg",
|
||||
[
|
||||
(5, b"\xa2\x18\x12\x83\x11\xcd\xcc\x00\x00U"),
|
||||
(0, b"\xa2\x18\x12\x83\x11\x00\x00\x00\x00U"),
|
||||
(-5, b"\xa2\x18\x12\x83\x1133\xff\xffU"),
|
||||
],
|
||||
)
|
||||
def test_axis_put(pos, msg):
|
||||
controller = NPointController(SocketMock())
|
||||
npointx = NPointAxis(controller, 0, "nx")
|
||||
controller.on()
|
||||
npointx.set(pos)
|
||||
assert npointx.controller.socket.buffer_put == msg
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"pos, msg_in, msg_out",
|
||||
[
|
||||
(5.0, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x11\xcd\xcc\x00\x00U"),
|
||||
(0, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x11\x00\x00\x00\x00U"),
|
||||
(-5, b"\xa04\x13\x83\x11U", b"\xa0\x34\x13\x83\x1133\xff\xffU"),
|
||||
],
|
||||
)
|
||||
def test_axis_get_out(pos, msg_in, msg_out):
|
||||
controller = NPointController(SocketMock())
|
||||
npointx = NPointAxis(controller, 0, "nx")
|
||||
controller.on()
|
||||
npointx.controller.socket.buffer_recv = msg_out
|
||||
assert pytest.approx(npointx.get(), rel=0.01) == pos
|
||||
# assert controller.socket.buffer_put == msg_in
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"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"),
|
||||
],
|
||||
)
|
||||
def test_axis_get_in(axis, msg_in, msg_out):
|
||||
controller = NPointController(SocketMock())
|
||||
npointx = NPointAxis(controller, 0, "nx")
|
||||
controller.on()
|
||||
controller.socket.buffer_recv = msg_out
|
||||
controller._get_current_pos(axis)
|
||||
assert controller.socket.buffer_put == msg_in
|
||||
|
||||
|
||||
def test_axis_out_of_range():
|
||||
controller = NPointController(SocketMock())
|
||||
with pytest.raises(ValueError):
|
||||
npointx = NPointAxis(controller, 3, "nx")
|
||||
|
||||
|
||||
def test_get_axis_out_of_range():
|
||||
controller = NPointController(SocketMock())
|
||||
with pytest.raises(ValueError):
|
||||
controller._get_current_pos(3)
|
||||
|
||||
|
||||
def test_set_axis_out_of_range():
|
||||
controller = NPointController(SocketMock())
|
||||
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):
|
||||
assert (
|
||||
NPointController._hex_list_to_int(in_buffer, byteorder=byteorder, signed=signed)
|
||||
== val
|
||||
)
|
||||
|
||||
|
||||
@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"),
|
||||
],
|
||||
)
|
||||
def test_get_range(axis, msg_in, msg_out):
|
||||
controller = NPointController(SocketMock())
|
||||
npointx = NPointAxis(controller, 0, "nx")
|
||||
controller.on()
|
||||
controller.socket.buffer_recv = msg_out
|
||||
val = controller._get_range(axis)
|
||||
assert controller.socket.buffer_put == msg_in and val == 100
|
||||
238
tests/test_smaract.py
Normal file
238
tests/test_smaract.py
Normal file
@ -0,0 +1,238 @@
|
||||
import pytest
|
||||
from ophyd_devices.smaract import SmaractController
|
||||
from ophyd_devices.smaract.smaract_controller import SmaractCommunicationMode
|
||||
from ophyd_devices.smaract.smaract_errors import (
|
||||
SmaractCommunicationError,
|
||||
SmaractErrorCode,
|
||||
)
|
||||
from ophyd_devices.smaract.smaract_ophyd import SmaractMotor
|
||||
|
||||
from utils import SocketMock
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"axis,position,get_message,return_msg",
|
||||
[
|
||||
(0, 50, b":GP0\n", b":P0,50000000"),
|
||||
(1, 0, b":GP1\n", b":P1,0"),
|
||||
(0, -50, b":GP0\n", b":P0,-50000000"),
|
||||
(0, -50.23, b":GP0\n", b":P0,-50230000"),
|
||||
],
|
||||
)
|
||||
def test_get_position(axis, position, get_message, return_msg):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
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":GPPK0\n", b":PPK0,1", None),
|
||||
(1, True, b":GPPK1\n", b":PPK1,1", None),
|
||||
(0, False, b":GPPK0\n", b":PPK0,0", None),
|
||||
(200, False, b":GPPK0\n", b":PPK0,0", ValueError),
|
||||
],
|
||||
)
|
||||
def test_axis_is_referenced(axis, is_referenced, get_message, return_msg, exception):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
if exception is not None:
|
||||
with pytest.raises(exception):
|
||||
val = 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(
|
||||
"return_msg,exception,raised",
|
||||
[
|
||||
(b"false", SmaractCommunicationError, False),
|
||||
(b":E0,1", SmaractErrorCode, True),
|
||||
(b":E,1", SmaractCommunicationError, True),
|
||||
(b":E,-1", SmaractCommunicationError, True),
|
||||
],
|
||||
)
|
||||
def test_socket_put_and_receive_raises_exception(return_msg, exception, raised):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
with pytest.raises(exception):
|
||||
controller.socket_put_and_receive(b"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")
|
||||
else:
|
||||
assert controller.socket_put_and_receive(b"test") == return_msg.decode()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"mode,get_message,return_msg",
|
||||
[
|
||||
(0, b":GCM\n", b":CM0"),
|
||||
(1, b":GCM\n", b":CM1"),
|
||||
],
|
||||
)
|
||||
def test_communication_mode(mode, get_message, return_msg):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
val = controller.get_communication_mode()
|
||||
assert controller.sock.buffer_put[0] == get_message
|
||||
assert val == SmaractCommunicationMode(mode)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"is_moving,get_message,return_msg",
|
||||
[
|
||||
(0, b":GS0\n", b":S0,0"),
|
||||
(1, b":GS0\n", b":S0,1"),
|
||||
(1, b":GS0\n", b":S0,2"),
|
||||
(0, b":GS0\n", b":S0,3"),
|
||||
(1, b":GS0\n", b":S0,4"),
|
||||
(0, b":GS0\n", b":S0,5"),
|
||||
(0, b":GS0\n", b":S0,6"),
|
||||
(1, b":GS0\n", b":S0,7"),
|
||||
(0, b":GS0\n", b":S0,9"),
|
||||
(0, [b":GS0\n", b":GS0\n"], [b":E0,0", b":S0,9"]),
|
||||
],
|
||||
)
|
||||
def test_axis_is_moving(is_moving, get_message, return_msg):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
val = controller.is_axis_moving(0)
|
||||
assert val == is_moving
|
||||
if isinstance(controller.sock.buffer_put, list) and len(controller.sock.buffer_put) == 1:
|
||||
controller.sock.buffer_put = controller.sock.buffer_put[0]
|
||||
assert controller.sock.buffer_put == get_message
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"sensor_id,axis,get_msg,return_msg",
|
||||
[
|
||||
(1, 0, b":GST0\n", b":ST0,1"),
|
||||
(6, 0, b":GST0\n", b":ST0,6"),
|
||||
(6, 1, b":GST1\n", b":ST1,6"),
|
||||
],
|
||||
)
|
||||
def test_get_sensor_definition(sensor_id, axis, get_msg, return_msg):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
sensor = controller.get_sensor_type(axis)
|
||||
assert sensor.type_code == sensor_id
|
||||
|
||||
|
||||
@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"),
|
||||
],
|
||||
)
|
||||
def test_set_move_speed(move_speed, axis, get_msg, return_msg):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
controller.set_closed_loop_move_speed(axis, move_speed)
|
||||
assert controller.sock.buffer_put[0] == get_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"),
|
||||
],
|
||||
)
|
||||
def test_move_axis_to_absolute_position(pos, axis, hold_time, get_msg, return_msg):
|
||||
controller = SmaractController(socket=SocketMock(host="dummy", port=123))
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
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)
|
||||
assert controller.sock.buffer_put[0] == get_msg
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"pos,get_msg,return_msg",
|
||||
[
|
||||
(
|
||||
50,
|
||||
[b":GPPK0\n", b":MPA0,50000000,1000\n", b":GS0\n", b":GP0\n"],
|
||||
[b":PPK0,1", b":E0,0", b":S0,0", b":P0,50000000\n"],
|
||||
),
|
||||
(
|
||||
0,
|
||||
[b":GPPK0\n", b":MPA0,0,1000\n", b":GS0\n", b":GP0\n"],
|
||||
[b":PPK0,1", b":E0,0", b":S0,0", b":P0,0000000\n"],
|
||||
),
|
||||
(
|
||||
20.23,
|
||||
[b":GPPK0\n", b":MPA0,20230000,1000\n", b":GS0\n", b":GP0\n"],
|
||||
[b":PPK0,1", b":E0,0", b":S0,0", b":P0,20230000\n"],
|
||||
),
|
||||
(
|
||||
20.23,
|
||||
[b":GPPK0\n", b":GPPK0\n", b":MPA0,20230000,1000\n", b":GS0\n", b":GP0\n"],
|
||||
[b":S0,0", b":PPK0,1", b":E0,0", b":S0,0", b":P0,20230000\n"],
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_move_axis(pos, get_msg, return_msg):
|
||||
lsmarA = SmaractMotor(
|
||||
"A",
|
||||
name="lsmarA",
|
||||
host="mpc2680.psi.ch",
|
||||
port=8085,
|
||||
sign=1,
|
||||
socket_cls=SocketMock,
|
||||
)
|
||||
lsmarA.stage()
|
||||
controller = lsmarA.controller
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
lsmarA.move(pos)
|
||||
assert controller.sock.buffer_put == get_msg
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"num_axes,get_msg,return_msg",
|
||||
[
|
||||
(
|
||||
1,
|
||||
[b":S0\n"],
|
||||
[b":E0,0"],
|
||||
)
|
||||
],
|
||||
)
|
||||
def test_stop_axis(num_axes, get_msg, return_msg):
|
||||
lsmarA = SmaractMotor(
|
||||
"A",
|
||||
name="lsmarA",
|
||||
host="mpc2680.psi.ch",
|
||||
port=8085,
|
||||
sign=1,
|
||||
socket_cls=SocketMock,
|
||||
)
|
||||
lsmarA.stage()
|
||||
controller = lsmarA.controller
|
||||
controller.sock.flush_buffer()
|
||||
controller.sock.buffer_recv = return_msg
|
||||
controller.stop_all_axes()
|
||||
assert controller.sock.buffer_put == get_msg
|
||||
69
tests/test_socket.py
Normal file
69
tests/test_socket.py
Normal file
@ -0,0 +1,69 @@
|
||||
import pytest
|
||||
import socket
|
||||
from ophyd_devices.utils.socket import SocketIO
|
||||
|
||||
|
||||
class DummySocket:
|
||||
AF_INET = 2
|
||||
SOCK_STREAM = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.address_family = None
|
||||
self.socket_kind = None
|
||||
self.timeout = None
|
||||
|
||||
def socket(self, address_family, socket_kind):
|
||||
self.address_family = address_family
|
||||
self.socket_kind = socket_kind
|
||||
return self
|
||||
|
||||
def settimeout(self, timeout):
|
||||
self.timeout = timeout
|
||||
|
||||
def send(self, msg, *args, **kwargs):
|
||||
self.send_buffer = msg
|
||||
|
||||
def connect(self, address):
|
||||
self.host = address[0]
|
||||
self.port = address[1]
|
||||
self.connected = True
|
||||
|
||||
def close(self):
|
||||
self.connected = False
|
||||
|
||||
|
||||
def test_socket_init():
|
||||
socketio = SocketIO("localhost", 8080)
|
||||
|
||||
assert socketio.host == "localhost"
|
||||
assert socketio.port == 8080
|
||||
|
||||
assert socketio.is_open == False
|
||||
|
||||
assert socketio.sock.family == socket.AF_INET
|
||||
assert socketio.sock.type == socket.SOCK_STREAM
|
||||
|
||||
|
||||
def test_socket_put():
|
||||
dsocket = DummySocket()
|
||||
socketio = SocketIO("localhost", 8080)
|
||||
socketio.sock = dsocket
|
||||
socketio.put(b"message")
|
||||
assert dsocket.send_buffer == b"message"
|
||||
|
||||
|
||||
def test_open():
|
||||
dsocket = DummySocket()
|
||||
socketio = SocketIO("localhost", 8080)
|
||||
socketio.sock = dsocket
|
||||
socketio.open()
|
||||
assert socketio.is_open == True
|
||||
assert socketio.sock.host == socketio.host
|
||||
assert socketio.sock.port == socketio.port
|
||||
|
||||
|
||||
def test_close():
|
||||
socketio = SocketIO("localhost", 8080)
|
||||
socketio.close()
|
||||
assert socketio.sock == None
|
||||
assert socketio.is_open == False
|
||||
46
tests/utils.py
Normal file
46
tests/utils.py
Normal file
@ -0,0 +1,46 @@
|
||||
class SocketMock:
|
||||
def __init__(self, host, port):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.buffer_put = []
|
||||
self.buffer_recv = [b""]
|
||||
self.is_open = False
|
||||
self.open()
|
||||
|
||||
def connect(self):
|
||||
print(f"connecting to {self.host} port {self.port}")
|
||||
|
||||
def _put(self, msg_bytes):
|
||||
self.buffer_put.append(msg_bytes)
|
||||
print(self.buffer_put)
|
||||
|
||||
def _recv(self, buffer_length=1024):
|
||||
print(self.buffer_recv)
|
||||
if isinstance(self.buffer_recv, list):
|
||||
if len(self.buffer_recv) > 0:
|
||||
ret_val = self.buffer_recv.pop(0)
|
||||
else:
|
||||
ret_val = b""
|
||||
return ret_val
|
||||
return self.buffer_recv
|
||||
|
||||
def _initialize_socket(self):
|
||||
pass
|
||||
|
||||
def put(self, msg):
|
||||
return self._put(msg)
|
||||
|
||||
def receive(self, buffer_length=1024):
|
||||
return self._recv(buffer_length=buffer_length)
|
||||
|
||||
def open(self):
|
||||
self._initialize_socket()
|
||||
self.is_open = True
|
||||
|
||||
def close(self):
|
||||
self.sock = None
|
||||
self.is_open = False
|
||||
|
||||
def flush_buffer(self):
|
||||
self.buffer_put = []
|
||||
self.buffer_recv = ""
|
||||
Reference in New Issue
Block a user