diff --git a/siriuspy/siriuspy/VERSION b/siriuspy/siriuspy/VERSION index 0044d6cb9..38f77a65b 100644 --- a/siriuspy/siriuspy/VERSION +++ b/siriuspy/siriuspy/VERSION @@ -1 +1 @@ -1.20.1 +2.0.1 diff --git a/siriuspy/siriuspy/csdev.py b/siriuspy/siriuspy/csdev.py index f3d793e66..e0da2e18f 100644 --- a/siriuspy/siriuspy/csdev.py +++ b/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) diff --git a/siriuspy/siriuspy/devices/sofb.py b/siriuspy/siriuspy/devices/sofb.py index 6e42e1c74..9556cdcac 100644 --- a/siriuspy/siriuspy/devices/sofb.py +++ b/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 - '' + 'Sum-Mon', - '' + 'OrbX-Mon', - '' + 'OrbY-Mon', - # properties used only for ring-type accelerators - '' + 'Idx' + 'OrbX-Mon', - '' + '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) - if self.data.isring or \ - ('IdxOrbX-Mon' not in propty and - 'IdxOrbY-Mon' not in propty): - properties.append(propty) - return orbtp, properties diff --git a/siriuspy/siriuspy/sofb/csdev.py b/siriuspy/siriuspy/sofb/csdev.py index bdee788ef..c5adaa21d 100644 --- a/siriuspy/siriuspy/sofb/csdev.py +++ b/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 diff --git a/siriuspy/siriuspy/util.py b/siriuspy/siriuspy/util.py index 60fd13a8e..790355237 100644 --- a/siriuspy/siriuspy/util.py +++ b/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: diff --git a/siriuspy/tests/test_util.py b/siriuspy/tests/test_util.py index c78e2f8ce..bb816585f 100755 --- a/siriuspy/tests/test_util.py +++ b/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'))