Tests

.github/workflows/workflows.yml

@@ -13,11 +13,12 @@ jobs:
     - uses: actions/setup-python@v3
     - uses: pre-commit/action@v3.0.1
 
-  installation:
+  pytest:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v3
-      - name: Install dependencies
+      - name: Tests
         run: |
-          pip install -e .
+          pip install -e .[test]
+          pytest -sv

motor_stage_ui/motor_stage_gui.py

@@ -1,4 +1,4 @@
-from motor_stage_ui.PIStagesInterface import PIStagesInterface
+from motor_stage_ui.pi_stages_interface import PIStagesInterface
 from motor_stage_ui import logger
 import motor_stage_ui
 

motor_stage_ui/motor_stage_terminal.py

@@ -1,4 +1,4 @@
-from motor_stage_ui.PIStagesInterface import PIStagesInterface as MC
+from motor_stage_ui.pi_stages_interface import PIStagesInterface as MC
 import motor_stage_ui
 
 import os

motor_stage_ui/PIStagesInterface.py → motor_stage_ui/pi_stages_interface.py

@@ -8,7 +8,7 @@
 # sudo chown :usr /dev/ttyUSB0
 
 
-class PIStagesInterface:
+class SerialInterface:
     def __init__(
         self,
         port: str,
@@ -26,7 +26,6 @@ def __init__(
             stopbits=stopbits,
         )
         self.log = logger.setup_main_logger(__class__.__name__, logging.WARNING)
-        self.ureg = UnitRegistry()
         self._terminator = terminator
         self._lock = Lock()
 
@@ -45,19 +44,6 @@ def _write(self, command: str):
             self.log.debug(msg)
             self._serial.write(msg)
 
-    def _write_command(self, command: str, address: int = None):
-        """Encodes the command for the PI motor stages.
-        This includes a header '01' and an address to select the specific stage and deselect the others and the command.
-
-        Args:
-            command (str): Command for the stage
-            address (int, optional): Address of the specific motor stage. Defaults to None.
-        """
-        if address:
-            self._write(("\x01%d" % (address - 1)) + command)
-        else:
-            self.log.error("Commands needs motor address")
-
     def _read(self):
         """Read message from serial port.
 
@@ -70,10 +56,47 @@ def _read(self):
             .strip(self._terminator)
         )
         if msg == "":
-            self.log.error("No responds from motor controller.")
+            self.log.error("No responds from serial interface.")
             raise ValueError
         return msg
 
+
+class PIStagesInterface:
+    def __init__(
+        self,
+        port: str,
+        baud_rate: int = 9600,
+        parity: str = "N",
+        terminator: str = "\r",
+        timeout: float = 2,
+        stopbits: float = 2,
+        interface: type[SerialInterface] = SerialInterface,
+    ):
+        self.serial_interface = interface(
+            port=port,
+            baud_rate=baud_rate,
+            parity=parity,
+            terminator=terminator,
+            timeout=timeout,
+            stopbits=stopbits,
+        )
+
+        self.log = logger.setup_main_logger(__class__.__name__, logging.WARNING)
+        self.ureg = UnitRegistry()
+
+    def _write_command(self, command: str, address: int = None):
+        """Encodes the command for the PI motor stages.
+        This includes a header '01' and an address to select the specific stage and deselect the others and the command.
+
+        Args:
+            command (str): Command for the stage
+            address (int, optional): Address of the specific motor stage. Defaults to None.
+        """
+        if address:
+            self.serial_interface._write(("\x01%d" % (address - 1)) + command)
+        else:
+            self.log.error("Commands needs motor address")
+
     def _write_read(self, command: str, address: int = None):
         """Write command to port and read back answer.
 
@@ -85,10 +108,10 @@ def _write_read(self, command: str, address: int = None):
             _type_: Answer message
         """
         if address:
-            self._write(("\x01%d" % (address - 1)) + command)
+            self.serial_interface._write(("\x01%d" % (address - 1)) + command)
         else:
             self.log.error("Commands needs motor address")
-        return self._read()
+        return self.serial_interface._read()
 
     # Motor stage commands
 
@@ -136,15 +159,19 @@ def set_velocity(self, address: int, velocity: int):
         velocity = "SV" + str(velocity)
         self._write_command(velocity, address=address)
 
-    def find_edge(self, address: int, edge: int = 0) -> None:
+    def find_edge(
+        self, address: int, unit: str, stage: str, step_size: float, edge: int = 0
+    ) -> None:
         """Tries to move the motorstage to a edge.
 
         Args:
             address (int): Address of the motorstage
             edge (int, optional): edge of the stage set 0 or 1. Defaults to 0.
         """
         self._write_command("FE%d" % edge, address)
-        pos = self._wait(address)
+        pos = self.get_position(
+            address=address, unit=unit, stage=stage, step_size=step_size
+        )
         self.log.warning("Edge found at position: {pos}".format(pos=pos))
 
     def set_home(self, address: int) -> None:
@@ -258,7 +285,7 @@ def get_position(
             self.log.warning("Invalid stage type")
 
     def _move_to_position(self, address: int, value: int) -> None:
-        """Helper function for basil
+        """Helper function for pyserial
 
         Args:
             address (int): Address of the motorstage
@@ -270,7 +297,7 @@ def _move_to_position(self, address: int, value: int) -> None:
         )
 
     def _move_relative(self, address: int, value: int = 1000000) -> None:
-        """Helper function for basil
+        """Helper function for pyserial
 
         Args:
             address (int): Address of the motorstage
@@ -282,7 +309,7 @@ def _move_relative(self, address: int, value: int = 1000000) -> None:
         )
 
     def _get_position(self, address: int) -> None:
-        """Helper function for basil
+        """Helper function for pyserial
 
         Args:
             address (int): Address of the motorstage

motor_stage_ui/test/test_configuration.yaml

@@ -0,0 +1,28 @@
+
+# Define the connected stages here:
+#
+# name:
+#   stage:      define stage typus 'rotation', 'translation'
+#   address:    set with dip switches
+#   step_size:  depends on the design specifics of the hardware (stage and powering), check 'Design resolution' in e.g.:
+#               https://www.pi-usa.us/fileadmin/user_upload/physik_instrumente/files/CAT/PI-CAT132E-Precision-Positioning-and-Motion-Control-Web.pdf
+#               set in um for translation stage and in deg for rotational stage
+#               M-038.DG rotational stage:   step_size=0.59 urad / 34*10**(-6) deg
+#               M-403.6DG translation stage: step_size=0.018 um
+#   unit:       set output unit and default input unit
+
+x_axis:
+  stage_type: translation
+  address: 1
+  step_size: 0.018
+  unit: mm
+  port: '/dev/ttyUSB0'
+  baud_rate: 9600
+
+rot:
+  stage_type: rotation
+  address: 3
+  step_size: 34e-06
+  unit: deg
+  port: '/dev/ttyUSB0'
+  baud_rate: 9600

motor_stage_ui/test/test_pi_stage_interface.py

@@ -0,0 +1,192 @@
+from pathlib import Path
+import os
+import pytest
+import yaml
+from motor_stage_ui.pi_stages_interface import PIStagesInterface
+from motor_stage_ui.test.utils import SerialInterfaceMock
+
+
+FILEPATH = Path(__file__).parent
+CONFIG_FILE = FILEPATH / "test_configuration.yaml"
+
+with open(CONFIG_FILE) as yaml_file:
+    TESTCONFIG = yaml.safe_load(yaml_file)
+
+
+INTERFACE = SerialInterfaceMock
+
+PISTAGES = PIStagesInterface(
+    port=TESTCONFIG["x_axis"]["port"],
+    baud_rate=TESTCONFIG["x_axis"]["baud_rate"],
+    interface=INTERFACE,
+)
+
+
+def test_init_motor():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    PISTAGES.init_motor(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-3:] == [
+        b"\x010MN\r",
+        b"\x010RT\r",
+        b"\x010SV200000\r",
+    ]
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    PISTAGES.init_motor(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-3:] == [
+        b"\x012MN\r",
+        b"\x012RT\r",
+        b"\x012SV200000\r",
+    ]
+
+
+def test_find_edge():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    UNIT = TESTCONFIG["x_axis"]["unit"]
+    STEPSIZE = TESTCONFIG["x_axis"]["step_size"]
+    STAGE = TESTCONFIG["x_axis"]["stage_type"]
+
+    def __get_position(self, address):
+        # Blank get Position function
+        return 0
+
+    func_type = type(PISTAGES._get_position)
+    PISTAGES._get_position = func_type(__get_position, PISTAGES)
+    PISTAGES.find_edge(address=ADDRESS, unit=UNIT, stage=STAGE, step_size=STEPSIZE)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010FE0\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    UNIT = TESTCONFIG["rot"]["unit"]
+    STEPSIZE = float(TESTCONFIG["rot"]["step_size"])
+    STAGE = TESTCONFIG["rot"]["stage_type"]
+
+    def __get_position(self, address):
+        # Blank get Position function
+        return 0
+
+    func_type = type(PISTAGES._get_position)
+    PISTAGES._get_position = func_type(__get_position, PISTAGES)
+    PISTAGES.find_edge(address=ADDRESS, unit=UNIT, stage=STAGE, step_size=STEPSIZE)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012FE0\r"
+
+
+def test_set_home() -> None:
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    PISTAGES.set_home(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010DH\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    PISTAGES.set_home(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012DH\r"
+
+
+def test_go_home() -> None:
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    PISTAGES.go_home(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010GH\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    PISTAGES.go_home(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012GH\r"
+
+
+def test_abort():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    PISTAGES.abort(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010AB\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    PISTAGES.abort(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012AB\r"
+
+
+def test_move_to_position():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    UNIT = TESTCONFIG["x_axis"]["unit"]
+    STEPSIZE = TESTCONFIG["x_axis"]["step_size"]
+    STAGE = TESTCONFIG["x_axis"]["stage_type"]
+    PISTAGES.move_to_position(
+        address=ADDRESS, amount="1mm", unit=UNIT, stage=STAGE, step_size=STEPSIZE
+    )
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010MA55555\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    UNIT = TESTCONFIG["rot"]["unit"]
+    STEPSIZE = float(TESTCONFIG["rot"]["step_size"])
+    STAGE = TESTCONFIG["rot"]["stage_type"]
+    PISTAGES.move_to_position(
+        address=ADDRESS, amount="1deg", unit=UNIT, stage=STAGE, step_size=STEPSIZE
+    )
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012MA29411\r"
+
+
+def test_move_relative():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    UNIT = TESTCONFIG["x_axis"]["unit"]
+    STEPSIZE = TESTCONFIG["x_axis"]["step_size"]
+    STAGE = TESTCONFIG["x_axis"]["stage_type"]
+    PISTAGES.move_relative(
+        address=ADDRESS, amount="-1cm", unit=UNIT, stage=STAGE, step_size=STEPSIZE
+    )
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010MR-555555\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    UNIT = TESTCONFIG["rot"]["unit"]
+    STEPSIZE = float(TESTCONFIG["rot"]["step_size"])
+    STAGE = TESTCONFIG["rot"]["stage_type"]
+    PISTAGES.move_relative(
+        address=ADDRESS, amount="-1rad", unit=UNIT, stage=STAGE, step_size=STEPSIZE
+    )
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012MR-1685169\r"
+
+
+def test_get_position():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    UNIT = TESTCONFIG["x_axis"]["unit"]
+    STEPSIZE = TESTCONFIG["x_axis"]["step_size"]
+    STAGE = TESTCONFIG["x_axis"]["stage_type"]
+
+    def __get_position(self, address):
+        # Blank get Position function
+        return 0
+
+    func_type = type(PISTAGES._get_position)
+    PISTAGES._get_position = func_type(__get_position, PISTAGES)
+    assert (
+        PISTAGES.get_position(
+            address=ADDRESS, unit=UNIT, stage=STAGE, step_size=STEPSIZE
+        )
+        == "0.000"
+    )
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    UNIT = TESTCONFIG["rot"]["unit"]
+    STEPSIZE = float(TESTCONFIG["rot"]["step_size"])
+    STAGE = TESTCONFIG["rot"]["stage_type"]
+
+    def __get_position(self, address):
+        # Blank get Position function
+        return 0
+
+    func_type = type(PISTAGES._get_position)
+    PISTAGES._get_position = func_type(__get_position, PISTAGES)
+    assert (
+        PISTAGES.get_position(
+            address=ADDRESS, unit=UNIT, stage=STAGE, step_size=STEPSIZE
+        )
+        == "0.000"
+    )
+
+
+def test_get_stat():
+    ADDRESS = TESTCONFIG["x_axis"]["address"]
+    PISTAGES.get_stat(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x010TS\r"
+
+    ADDRESS = TESTCONFIG["rot"]["address"]
+    PISTAGES.get_stat(address=ADDRESS)
+    assert PISTAGES.serial_interface._serial_commands[-1] == b"\x012TS\r"
+
+
+if __name__ == "__main__":
+    pytest.main()