EpicsMotorIOC

.travis.yml

@@ -18,6 +18,7 @@ env:
     - secure: "FhNkkbod0Wc/zUf9cTvwziAYHcjfte2POf+hoVSmC+v/RcYKCNCo+mGGMhF9F4KyC2nzvulfzow7YXoswZqav4+TEEu+mpuPaGlf9aqp8V61eij8MVTwonzQEYmHAy3KatwXxyvvhQpfj3gOuDVolfOg2MtNZi6QERES4E1sjOn714fx2HkVxqH2Y8/PF/FzzGeJaRlVaVci0EdIJ5Ss5c5SjO6JGgxj4hzhTPHjTaLjdLHlVhuB9Yatl80zbhGriljLcDQTHmoSODwBpAh5YLDUZq6B9vomaNB9Hb3e0D5gItjOdj53v6AsHU8LkncZMvsgJgh2sZZqMO6nkpHcYPwJgbPbKd3RtVlk6Kg/tvKQk0rMcxl5fFFeD2i9POnANg/xJsKN6yAEY3kaRwQtajQmlcicSa/wdwv9NhUTtBmA/mnyzxHbQXrB0bEc2P2QVu7U8en6dWaOAqc1VCMrWIhp2ADNWb7JZhYj70TgmExIU3UH8qlMb6dyx50SJUE9waJj3fiiZVkjh+E568ZRSMvL9n+bLlFt4uDT4AysSby6cj+zjfNViKFstTAqjyd5VJEvCoUu73vNzWEiWFtEvKKVL1P3pbLN/G3aSSJMa5fc1o+2lRUwdwNNOOdH6iKBDZGNpE8nGDlTP2b2dhFyEt8nICKJhbgU208jhyyH8Vk="
 
 script:
+  - export OPHYD_CONTROL_LAYER=caproto
   - coverage run -m pytest  # Run the tests and check for test coverage.
   - coverage report -m  # Generate test coverage report.
   - codecov  # Upload the report to codecov.

nslsii/iocs/epics_motor_ioc_sim.py

@@ -0,0 +1,177 @@
+#!/usr/bin/env python3
+from caproto.server import pvproperty, PVGroup
+from caproto.server import ioc_arg_parser, run
+from caproto import ChannelType
+
+from threading import Lock
+
+
+class EpicsMotorIOC(PVGroup):
+    """
+    Simulates ophyd.EpicsMotor.
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    _dir_states = ['neg', 'pos']
+    _false_true_states = ['False', 'True']
+
+    _step_size = 0.1
+
+    # position
+
+    _upper_alarm_limit = 10.0
+    _lower_alarm_limit = -10.0
+
+    _upper_warning_limit = 9.0
+    _lower_warning_limit = -9.0
+
+    _upper_ctrl_limit = 11.0
+    _lower_ctrl_limit = -11.0
+
+    _egu = 'mm'
+
+    _precision = 3
+
+    user_readback = pvproperty(value=0.0, read_only=True,
+                               dtype=ChannelType.DOUBLE,
+                               upper_alarm_limit=_upper_alarm_limit,
+                               lower_alarm_limit=_lower_alarm_limit,
+                               upper_warning_limit=_upper_warning_limit,
+                               lower_warning_limit=_lower_warning_limit,
+                               upper_ctrl_limit=_upper_ctrl_limit,
+                               lower_ctrl_limit=_lower_ctrl_limit,
+                               units=_egu,
+                               precision=_precision,
+                               name='.RBV')
+    user_setpoint = pvproperty(value=0.0,
+                               dtype=ChannelType.DOUBLE,
+                               upper_alarm_limit=_upper_alarm_limit,
+                               lower_alarm_limit=_lower_alarm_limit,
+                               upper_warning_limit=_upper_warning_limit,
+                               lower_warning_limit=_lower_warning_limit,
+                               upper_ctrl_limit=_upper_ctrl_limit,
+                               lower_ctrl_limit=_lower_ctrl_limit,
+                               units=_egu,
+                               precision=_precision,
+                               name='.VAL')
+
+    putter_lock = Lock()
+
+    # calibration dial <--> user
+
+    user_offset = pvproperty(value=0.0, read_only=True,
+                             dtype=ChannelType.DOUBLE,
+                             name='.OFF')
+
+    user_offset_dir = pvproperty(value=_dir_states[1],
+                                 enum_strings=_dir_states,
+                                 dtype=ChannelType.ENUM,
+                                 name='.DIR')
+
+    offset_freeze_switch = pvproperty(value=_false_true_states[0],
+                                      enum_strings=_false_true_states,
+                                      dtype=ChannelType.ENUM,
+                                      name='.FOFF')
+    set_use_switch = pvproperty(value=_false_true_states[0],
+                                enum_strings=_false_true_states,
+                                dtype=ChannelType.ENUM,
+                                name='.SET')
+
+    # configuration
+
+    _velocity = 1.
+    _acceleration = 3.
+
+    velocity = pvproperty(value=_velocity, read_only=True,
+                          dtype=ChannelType.DOUBLE,
+                          name='.VELO')
+    acceleration = pvproperty(value=_acceleration, read_only=True,
+                              dtype=ChannelType.DOUBLE,
+                              name='.ACCL')
+    motor_egu = pvproperty(value=_egu, read_only=True,
+                           dtype=ChannelType.STRING,
+                           name='.EGU')
+
+    # motor status
+
+    motor_is_moving = pvproperty(value='False', read_only=True,
+                                 enum_strings=_false_true_states,
+                                 dtype=ChannelType.ENUM,
+                                 name='.MOVN')
+    motor_done_move = pvproperty(value='False', read_only=False,
+                                 enum_strings=_false_true_states,
+                                 dtype=ChannelType.ENUM,
+                                 name='.DMOV')
+
+    high_limit_switch = pvproperty(value=0, read_only=True,
+                                   dtype=ChannelType.INT,
+                                   name='.HLS')
+    low_limit_switch = pvproperty(value=0, read_only=True,
+                                  dtype=ChannelType.INT,
+                                  name='.LLS')
+
+    direction_of_travel = pvproperty(value=_dir_states[1],
+                                     enum_strings=_dir_states,
+                                     dtype=ChannelType.ENUM,
+                                     name='.TDIR')
+
+    # commands
+
+    _cmd_states = ['False', 'True']
+
+    motor_stop = pvproperty(value=_cmd_states[0],
+                            enum_strings=_cmd_states,
+                            dtype=ChannelType.ENUM,
+                            name='.STOP')
+    home_forward = pvproperty(value=_cmd_states[0],
+                              enum_strings=_cmd_states,
+                              dtype=ChannelType.ENUM,
+                              name='.HOMF')
+    home_reverse = pvproperty(value=_cmd_states[0],
+                              enum_strings=_cmd_states,
+                              dtype=ChannelType.ENUM,
+                              name='.HOMR')
+
+    # Methods
+
+    @user_setpoint.startup
+    async def user_setpoint(self, instance, async_lib):
+        instance.ev = async_lib.library.Event()
+        instance.async_lib = async_lib
+
+    @user_setpoint.putter
+    async def user_setpoint(self, instance, value):
+
+        if self.putter_lock.locked() is True:
+            return instance.value
+        else:
+            self.putter_lock.acquire()
+
+        p0 = instance.value
+        dwell = self._step_size/self._velocity
+        N = max(1, int((value - p0) / self._step_size))
+
+        await self.motor_done_move.write(value='False')
+
+        for j in range(N):
+            new_value = p0 + self._step_size*(j+1)
+            await instance.async_lib.library.sleep(dwell)
+            await self.user_readback.write(value=new_value)
+
+        await self.motor_done_move.write(value='True')
+
+        self.putter_lock.release()
+
+        return value
+
+
+if __name__ == '__main__':
+
+    ioc_options, run_options = ioc_arg_parser(
+        default_prefix='mtr:',
+        desc='EpicsMotor IOC.')
+
+    ioc = EpicsMotorIOC(**ioc_options)
+    run(ioc.pvdb, **run_options)

nslsii/tests/temperature_controllers_test.py

@@ -5,6 +5,7 @@
 import subprocess
 import os
 import sys
+import time
 import pytest
 
 
@@ -27,13 +28,14 @@ def test_Eurotherm(RE):
 
     # Start up an IOC based on the thermo_sim device in caproto.ioc_examples
     ioc_process = subprocess.Popen([sys.executable, '-m',
-                                    'caproto.tests.example_runner',
                                     'caproto.ioc_examples.thermo_sim'],
                                    stdout=stdout, stdin=stdin,
                                    env=os.environ)
 
     print(f'caproto.ioc_examples.thermo_sim is now running')
 
+    time.sleep(5)
+
     # Wrap the rest in a try-except to ensure the ioc is killed before exiting
     try:
         euro = Eurotherm('thermo:', name='euro')

nslsii/tests/test_epicsmotor_ioc.py

@@ -0,0 +1,87 @@
+import os
+import pytest
+import subprocess
+import sys
+import time
+
+from ophyd.epics_motor import EpicsMotor
+
+
+@pytest.fixture(scope='class')
+def ioc_sim(request):
+
+    # setup code
+
+    stdout = subprocess.PIPE
+    stdin = None
+
+    ioc_process = subprocess.Popen([sys.executable, '-m',
+                                    'nslsii.iocs.epics_motor_ioc_sim'],
+                                   stdout=stdout, stdin=stdin,
+                                   env=os.environ)
+
+    print(f'nslsii.iocs.epics_motor_ioc_sim is now running')
+
+    time.sleep(5)
+
+    mtr = EpicsMotor(prefix='mtr:', name='mtr')
+
+    time.sleep(5)
+
+    request.cls.mtr = mtr
+
+    yield
+
+    # teardown code
+
+    ioc_process.terminate()
+
+
+@pytest.mark.usefixtures('ioc_sim')
+class TestIOC:
+
+    def test_initial_values(self):
+
+        assert self.mtr.egu == 'mm'
+
+        velocity_val = self.mtr.velocity.get()
+        assert velocity_val == 1
+
+        assert self.mtr.low_limit == -11.0
+        assert self.mtr.high_limit == 11.0
+
+    def test_set_current_position(self):
+
+        target_val = 5
+        readback_val = self.mtr.user_readback.get()
+        velocity_val = self.mtr.velocity.get()
+        mvtime = (target_val - readback_val)/velocity_val
+
+        self.mtr.set_current_position(target_val)
+
+        time.sleep(mvtime)
+
+        setpoint_val = self.mtr.user_setpoint.get()
+        readback_val = self.mtr.user_readback.get()
+        assert round(setpoint_val, 3) == target_val
+        assert round(readback_val, 3) == target_val
+
+    def test_move_with_timeout_gt_moving_time(self):
+
+        target_val = 7
+        readback_val = self.mtr.user_readback.get()
+        velocity_val = self.mtr.velocity.get()
+        mvtime = (target_val - readback_val)/velocity_val
+
+        move_status = self.mtr.move(target_val, timeout=mvtime+1)
+        assert move_status.success is True
+
+    def test_move_with_timeout_lt_moving_time(self):
+
+        target_val = 9
+        readback_val = self.mtr.user_readback.get()
+        velocity_val = self.mtr.velocity.get()
+        mvtime = (target_val - readback_val)/velocity_val
+
+        with pytest.raises(RuntimeError):
+            self.mtr.move(target_val, timeout=mvtime-1)

nslsii/iocs/tests/test_epstwostate_ioc.py → nslsii/tests/test_epstwostate_ioc.py

@@ -44,7 +44,6 @@ def test_epstwostate_ioc():
     stdin = None
 
     ioc_process = subprocess.Popen([sys.executable, '-m',
-                                    'caproto.tests.example_runner',
                                     'nslsii.iocs.eps_two_state_ioc_sim'],
                                    stdout=stdout, stdin=stdin,
                                    env=os.environ)