Merge branch 'master' into PR-fix-bugs

siriuspy/siriuspy/VERSION

@@ -1 +1 @@
-1.20.1
+2.0.1

siriuspy/siriuspy/csdev.py

@@ -1,12 +1,15 @@
 """Control system Device Util Module."""
 
 import copy as _copy
+
+from mathphys.functions import get_namedtuple as _get_namedtuple
+
 from . import util as _util
 from .clientweb import beaglebone_ip_list as _bbb_ip_list
 from .search import PSSearch as _PSSearch
 
 
-_DEV_2_IOC_IP_DICT = None
+_DEV_2_IOC_IP_DICT = dict()
 
 
 class ETypes:
@@ -28,15 +31,15 @@ class ETypes:
 class Const:
     """Const class defining power supply constants."""
 
-    DsblEnbl = _util.get_namedtuple('DsblEnbl', _et.DSBL_ENBL)
-    OffOn = _util.get_namedtuple('OffOn', _et.OFF_ON)
-    CloseOpen = _util.get_namedtuple('CloseOpen', _et.CLOSE_OPEN)
-    DisconnConn = _util.get_namedtuple('DisconnConn', _et.DISCONN_CONN)
+    DsblEnbl = _get_namedtuple('DsblEnbl', _et.DSBL_ENBL)
+    OffOn = _get_namedtuple('OffOn', _et.OFF_ON)
+    CloseOpen = _get_namedtuple('CloseOpen', _et.CLOSE_OPEN)
+    DisconnConn = _get_namedtuple('DisconnConn', _et.DISCONN_CONN)
 
     @staticmethod
     def register(name, field_names, values=None):
         """Register namedtuple."""
-        return _util.get_namedtuple(name, field_names, values)
+        return _get_namedtuple(name, field_names, values)
 
 
 def add_pvslist_cte(database, prefix=''):
@@ -55,30 +58,28 @@ def add_pvslist_cte(database, prefix=''):
 
 def get_device_2_ioc_ip(reload=False):
     """Return a dict of ioc IP numbers for csdevices."""
-    if _DEV_2_IOC_IP_DICT is None or reload is True:
+    if not _DEV_2_IOC_IP_DICT or reload:
         _reload_device_2_ioc_ip()
     return _copy.deepcopy(_DEV_2_IOC_IP_DICT)
 
 
 def _reload_device_2_ioc_ip():
-    global _DEV_2_IOC_IP_DICT
-    _DEV_2_IOC_IP_DICT = dict()
 
     # beaglebone IPs
     text, _ = _util.read_text_data(_bbb_ip_list())
     for item in text:
         if len(item) == 2:
-            bbbname, ip = item
-            _DEV_2_IOC_IP_DICT[bbbname] = ip
+            bbbname, ipm = item
+            _DEV_2_IOC_IP_DICT[bbbname] = ipm
 
     # power supplies
     dic = dict()
-    for bbbname, ip in _DEV_2_IOC_IP_DICT.items():
+    for bbbname, ipm in _DEV_2_IOC_IP_DICT.items():
         try:
             bsmps = _PSSearch.conv_bbbname_2_bsmps(bbbname)
             for bsmp in bsmps:
                 psname, _ = bsmp
-                dic[psname] = ip
+                dic[psname] = ipm
         except KeyError:
             pass
     _DEV_2_IOC_IP_DICT.update(dic)

siriuspy/siriuspy/devices/sofb.py

@@ -19,7 +19,7 @@ class DEVICES:
         ALL = (TB, BO, TS, SI)
 
     _propty_tmpl = (
-        'SlowOrbX-Mon', 'SlowOrbY-Mon',
+        'SOFBMode-Sel', 'SOFBMode-Sts',
         'KickCH-Mon', 'KickCV-Mon',
         'DeltaKickCH-Mon', 'DeltaKickCV-Mon',
         'DeltaKickCH-SP', 'DeltaKickCV-SP',
@@ -38,13 +38,11 @@ class DEVICES:
         'TrigNrSamplesPost-SP',
         'TrigNrSamplesPost-RB',
         'ClosedLoop-Sts', 'ClosedLoop-Sel',
-        # ring-type dependent properties
-        '<ORBTP>' + 'Sum-Mon',
-        '<ORBTP>' + 'OrbX-Mon',
-        '<ORBTP>' + 'OrbY-Mon',
-        # properties used only for ring-type accelerators
-        '<ORBTP>' + 'Idx' + 'OrbX-Mon',
-        '<ORBTP>' + 'Idx' + 'OrbY-Mon')
+        'SPassSum-Mon', 'SPassOrbX-Mon', 'SPassOrbY-Mon',
+        # properties used only for ring-type accelerators:
+        'SlowOrbX-Mon', 'SlowOrbY-Mon',
+        'MTurnSum-Mon', 'MTurnOrbX-Mon', 'MTurnOrbY-Mon',
+        'MTurnIdxOrbX-Mon', 'MTurnIdxOrbY-Mon', 'MTurnIdxSum-Mon')
 
     _default_timeout = 10  # [s]
     _off, _on = 0, 1
@@ -58,61 +56,107 @@ def __init__(self, devname):
         # SOFB object
         self.data = SOFBFactory.create(devname[:2])
 
-        # define device properties
-        self._orbtp, properties = \
-            self._set_attributes_properties()
+        propts = SOFB._propty_tmpl
+        if not self.data.isring:
+            propts = [p for p in propts if not p.startswith(('MTurn', 'Slow'))]
 
         # call base class constructor
-        super().__init__(devname, properties=properties)
+        super().__init__(devname, properties=propts)
+
+    @property
+    def opmode(self):
+        """."""
+        return self['SOFBMode-Sts']
 
-        # shortcut attributes to property names
-        self._trajx = self._orbtp + 'OrbX-Mon'
-        self._trajy = self._orbtp + 'OrbY-Mon'
-        self._sum = self._orbtp + 'Sum-Mon'
-        self._trajx_idx = self._orbtp + 'Idx' + 'OrbX-Mon'
-        self._trajy_idx = self._orbtp + 'Idx' + 'OrbY-Mon'
+    @opmode.setter
+    def opmode(self, value):
+        if value is None:
+            return
+        if isinstance(value, str) and value in self.data.SOFBMode._fields:
+            self['SOFBMode-Sel'] = self.data.SOFBMode._fields.index(value)
+        elif int(value) in self.data.SOFBMode:
+            self['SOFBMode-Sel'] = int(value)
 
     @property
-    def orbit_type(self):
+    def opmode_str(self):
         """."""
-        return self._orbtp
+        return self.data.SOFBMode._fields[self['SOFBMode-Sts']]
 
     @property
-    def trajx(self):
+    def sp_trajx(self):
         """."""
-        return self[self._trajx]
+        return self['SPassOrbX-Mon']
 
     @property
-    def trajy(self):
+    def sp_trajy(self):
         """."""
-        return self[self._trajy]
+        return self['SPassOrbY-Mon']
 
     @property
-    def sum(self):
+    def sp_sum(self):
+        """."""
+        return self['SPassSum-Mon']
+
+    @property
+    def mt_trajx(self):
+        """."""
+        return self['MTurnOrbX-Mon'] if self.data.isring else None
+
+    @property
+    def mt_trajy(self):
+        """."""
+        return self['MTurnOrbY-Mon'] if self.data.isring else None
+
+    @property
+    def mt_sum(self):
         """."""
-        return self[self._sum]
+        return self['MTurnSum-Mon'] if self.data.isring else None
 
     @property
-    def trajx_idx(self):
+    def mt_trajx_idx(self):
         """."""
-        return self[self._trajx_idx] if self.data.isring \
-            else self.trajx
+        return self['MTurnIdxOrbX-Mon'] if self.data.isring else None
 
     @property
-    def trajy_idx(self):
+    def mt_trajy_idx(self):
         """."""
-        return self[self._trajy_idx] if self.data.isring \
-            else self.trajy
+        return self['MTurnIdxOrbY-Mon'] if self.data.isring else None
+
+    @property
+    def mt_sum_idx(self):
+        """."""
+        return self['MTurnIdxSum-Mon'] if self.data.isring else None
+
+    @property
+    def trajx(self):
+        """."""
+        if self.data.isring and self.opmode == self.data.SOFBMode.MultiTurn:
+            return self.mt_trajx_idx
+        return self.sp_trajx
+
+    @property
+    def trajy(self):
+        """."""
+        if self.data.isring and self.opmode == self.data.SOFBMode.MultiTurn:
+            return self.mt_trajy_idx
+        return self.sp_trajy
+
+    @property
+    def sum(self):
+        """."""
+        if self.data.isring and self.opmode == self.data.SOFBMode.MultiTurn:
+            return self.mt_sum_idx
+        return self.sp_sum
 
     @property
     def orbx(self):
         """."""
-        return self['SlowOrbX-Mon']
+        return self['SlowOrbX-Mon'] if self.data.isring else None
 
     @property
     def orby(self):
         """."""
-        return self['SlowOrbY-Mon']
+        return self['SlowOrbY-Mon'] if self.data.isring else None
 
     @property
     def kickch(self):
@@ -227,12 +271,14 @@ def cvenbl(self, value):
     @property
     def rfenbl(self):
         """."""
-        return self['RFEnbl-Sts']
+        dta = self.data
+        return self['RFEnbl-Sts'] if dta.acc_idx == dta.Rings.SI else None
 
     @rfenbl.setter
     def rfenbl(self, value):
         """."""
-        self['RFEnbl-Sel'] = value
+        if self.data.acc_idx == self.data.Rings.SI:
+            self['RFEnbl-Sel'] = value
 
     @property
     def buffer_count(self):
@@ -304,16 +350,3 @@ def wait_buffer(self, timeout=None):
             _time.sleep(interval)
         else:
             print('WARN: Timed out waiting orbit.')
-
-    # --- private methods ---
-
-    def _set_attributes_properties(self):
-        orbtp = 'MTurn' if self.data.isring else 'SPass'
-        properties = []
-        for propty in SOFB._propty_tmpl:
-            propty = propty.replace('<ORBTP>', orbtp)
-            if self.data.isring or \
-                    ('IdxOrbX-Mon' not in propty and
-                     'IdxOrbY-Mon' not in propty):
-                properties.append(propty)
-        return orbtp, properties

siriuspy/siriuspy/sofb/csdev.py

@@ -4,7 +4,6 @@
 
 from .. import csdev as _csdev
 from ..namesys import SiriusPVName as _PVName
-from ..util import get_namedtuple as _get_namedtuple
 from ..search import MASearch as _MASearch, BPMSearch as _BPMSearch, \
     LLTimeSearch as _TISearch, HLTimeSearch as _HLTISearch, \
     PSSearch as _PSSearch
@@ -130,9 +129,12 @@ def __init__(self, acc):
         self.cv_nicknames = _PSSearch.get_psnicknames(self.cv_names)
         self.bpm_pos = _BPMSearch.get_positions(self.bpm_names)
 
-        func = lambda x: x.substitute(dis='MA' if x.dis == 'PS' else 'PM')
-        self.ch_pos = _MASearch.get_mapositions(map(func, self.ch_names))
-        self.cv_pos = _MASearch.get_mapositions(map(func, self.cv_names))
+        self.ch_pos = _MASearch.get_mapositions(map(
+            lambda x: x.substitute(dis='MA' if x.dis == 'PS' else 'PM'),
+            self.ch_names))
+        self.cv_pos = _MASearch.get_mapositions(map(
+            lambda x: x.substitute(dis='MA' if x.dis == 'PS' else 'PM'),
+            self.cv_names))
         self.nr_ch = len(self.ch_names)
         self.nr_cv = len(self.cv_names)
         self.nr_chcv = self.nr_ch + self.nr_cv
@@ -303,7 +305,6 @@ def get_corrs_database(self, prefix=''):
     def get_orbit_database(self, prefix=''):
         """Return Orbit database."""
         nbpm = self.nr_bpms
-        evt = self.evt_acq_name
         pvs = [
             'RefOrbX-SP', 'RefOrbX-RB',
             'RefOrbY-SP', 'RefOrbY-RB',
@@ -667,7 +668,7 @@ def __init__(self, acc):
         evts = _HLTISearch.get_hl_trigger_allowed_evts(self.trigger_cor_name)
         vals = _cstiming.get_hl_trigger_database(self.trigger_cor_name)
         vals = tuple([vals['Src-Sel']['enums'].index(evt) for evt in evts])
-        self.CorrExtEvtSrc = _get_namedtuple('CorrExtEvtSrc', evts, vals)
+        self.CorrExtEvtSrc = self.register('CorrExtEvtSrc', evts, vals)
         self.circum = 518.396  # in meter
         self.rev_per = self.circum / 299792458  # in seconds
 

siriuspy/siriuspy/util.py

@@ -255,24 +255,6 @@ def check_public_interface_namespace(namespace, valid_interface,
     return True
 
 
-def get_namedtuple(name, field_names, values=None):
-    """Return an instance of a namedtuple Class.
-
-    Inputs:
-        - name:  Defines the name of the Class (str).
-        - field_names:  Defines the field names of the Class (iterable).
-        - values (optional): Defines field values . If not given, the value of
-            each field will be its index in 'field_names' (iterable).
-
-    Raises ValueError if at least one of the field names are invalid.
-    Raises TypeError when len(values) != len(field_names)
-    """
-    if values is None:
-        values = range(len(field_names))
-    field_names = [f.replace(' ', '_') for f in field_names]
-    return _namedtuple(name, field_names)(*values)
-
-
 # This solution was copied from:
 # https://stackoverflow.com/questions/5189699/how-to-make-a-class-property
 class ClassProperty:

siriuspy/tests/test_util.py

@@ -25,7 +25,6 @@
     'get_bit',
     'mode',
     'check_public_interface_namespace',
-    'get_namedtuple',
     'ClassProperty',
     )
 
@@ -281,18 +280,3 @@ class NameSpace:
         valid = util.check_public_interface_namespace(
             NameSpace, public_interface, print_flag=False)
         self.assertFalse(valid)
-
-    def test_get_namedtuple(self):
-        """Test get_namedtuple."""
-        fields = ('a', 'b', 'c')
-        values = (1, 4, 5)
-        ntp = util.get_namedtuple('A', fields, values)
-        self.assertTrue(ntp.__class__.__name__ == 'A')
-        self.assertTrue(ntp._fields == fields)
-        self.assertTrue(ntp.a == 1 and ntp.b == 4 and ntp.c == 5)
-        fields = ('a', 'b', 'c')
-        ntp = util.get_namedtuple('A', fields)
-        self.assertTrue(ntp.a == 0 and ntp.b == 1 and ntp.c == 2)
-        fields = ('a a', 'b b', 'c')
-        ntp = util.get_namedtuple('A', fields)
-        self.assertTrue(ntp._fields == ('a_a', 'b_b', 'c'))