MotorGroupIOC

.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_record.py

@@ -0,0 +1,166 @@
+from caproto.server import pvproperty, PVGroup
+from caproto import ChannelType
+
+from threading import Lock
+
+
+class EpicsMotorRecord(PVGroup):
+    """
+    Simulates EPICS motor record.
+    """
+
+    def __init__(self, prefix, *, ioc, **kwargs):
+        super().__init__(prefix, **kwargs)
+        self.ioc = ioc
+
+    _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

nslsii/iocs/motor_group_ioc_sim.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+from caproto.server import PVGroup, template_arg_parser, run
+
+from nslsii.iocs.epics_motor_record import EpicsMotorRecord
+
+
+class MotorGroupIOC(PVGroup):
+    """
+    Simulates a group of EPICS motor records.
+    """
+
+    def __init__(self, prefix, *, groups, **kwargs):
+        super().__init__(prefix, **kwargs)
+        self.groups = groups
+
+
+def create_ioc(prefix, axes, **ioc_options):
+
+    groups = {}
+
+    mg_prefix = prefix.replace('{', '{'*2, 1)
+    ioc = MotorGroupIOC(prefix=mg_prefix, groups=groups, **ioc_options)
+
+    rec_mg_prefix = prefix.replace('{', '{'*4, 1)
+
+    for group_prefix in axes:
+        rec_group_prefix = group_prefix.replace('}', '}'*4, 1)
+        record_prefix = rec_mg_prefix + rec_group_prefix
+        groups[rec_group_prefix] = EpicsMotorRecord(record_prefix,
+                                                    ioc=ioc)
+
+    for prefix, group in groups.items():
+        ioc.pvdb.update(**group.pvdb)
+
+    return ioc
+
+
+if __name__ == '__main__':
+
+    parser, split_args = template_arg_parser(
+        default_prefix='test{tst-Ax:',
+        desc=MotorGroupIOC.__doc__,
+    )
+
+    axes_help = 'Comma-separated list of axes'
+
+    parser.add_argument('--axes', help=axes_help,
+                        required=True, type=str)
+
+    args = parser.parse_args()
+    ioc_options, run_options = split_args(args)
+
+    axes = [x.strip() for x in args.axes.split(',')]
+
+    ioc = create_ioc(axes=axes, **ioc_options)
+    run(ioc.pvdb, **run_options)

nslsii/tests/temperature_controllers_test.py

@@ -6,6 +6,7 @@
 import os
 import sys
 import pytest
+import time
 
 
 @pytest.fixture
@@ -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/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)

nslsii/tests/test_motorgroup_ioc.py

@@ -0,0 +1,122 @@
+import os
+import pytest
+import subprocess
+import sys
+import time
+
+from ophyd import Device, EpicsMotor
+from ophyd import Component
+# from ophyd.device import create_device_from_components
+
+from caproto.sync.client import read
+
+from collections import OrderedDict
+
+
+def create_device_from_components(name, *, docstring=None,
+                                  default_read_attrs=None,
+                                  default_configuration_attrs=None,
+                                  base_class=Device, class_kwargs=None,
+                                  **components):
+
+    if docstring is None:
+        docstring = f'{name} Device'
+
+    if not isinstance(base_class, tuple):
+        base_class = (base_class, )
+
+    if class_kwargs is None:
+        class_kwargs = {}
+
+    clsdict = OrderedDict(
+        __doc__=docstring,
+        _default_read_attrs=default_read_attrs,
+        _default_configuration_attrs=default_configuration_attrs
+    )
+
+    for attr, component in components.items():
+        if not isinstance(component, Component):
+            raise ValueError(f'Attribute {attr} is not a Component. '
+                             f'It is of type {type(component).__name__}')
+
+        clsdict[attr] = component
+
+    return type(name, base_class, clsdict, **class_kwargs)
+
+
+def slit(name, axes=None, *, docstring=None, default_read_attrs=None,
+         default_configuration_attrs=None):
+
+    components = {}
+    for name, PV_suffix in axes.items():
+        components[name] = Component(EpicsMotor, PV_suffix, name=name)
+
+    new_class = create_device_from_components(
+        name, docstring=docstring, default_read_attrs=default_read_attrs,
+        default_configuration_attrs=default_configuration_attrs,
+        base_class=Device, **components)
+
+    return new_class
+
+
+prefix = 'test{tst-Ax:'
+axes = {'hg': 'HG}Mtr', 'hc': 'HC}Mtr', 'vg': 'VG}Mtr', 'vc': 'VC}Mtr',
+        'inb': 'I}Mtr', 'out': 'O}Mtr', 'top': 'T}Mtr', 'bot': 'B}Mtr'}
+
+
+@pytest.fixture(scope='class')
+def ioc_sim(request):
+
+    # setup code
+
+    stdout = subprocess.PIPE
+    stdin = None
+
+    axes_str = ''
+    for name, PV_suffix in axes.items():
+        axes_str += PV_suffix + ','
+
+    ioc_process = subprocess.Popen([sys.executable,
+                                    '-m', 'nslsii.iocs.motor_group_ioc_sim',
+                                    '--axes', axes_str],
+                                   stdout=stdout, stdin=stdin,
+                                   env=os.environ)
+
+    print(f'nslsii.iocs.motor_group_ioc_sim is now running')
+
+    time.sleep(5)
+
+    FourBladeSlits = slit(name='FourBladeSlits',
+                          axes=axes,
+                          docstring='Four Blades Slits')
+
+    slits = FourBladeSlits(prefix, name='slits')
+
+    time.sleep(5)
+
+    request.cls.slits = slits
+
+    yield
+
+    # teardown code
+
+    ioc_process.terminate()
+
+
+@pytest.mark.usefixtures('ioc_sim')
+class TestIOC:
+
+    def test_caproto_level(self):
+
+        for name, PV_suffix in axes.items():
+            pvname = prefix + PV_suffix + '.VELO'
+            res = read(pvname)
+            velocity_val = res.data[0]
+            assert velocity_val == 1.0
+
+    def test_device_level(self):
+
+        assert(hasattr(self.slits, 'hg'))
+
+        velocity_val = self.slits.hg.velocity.get()
+        assert velocity_val == 1