Source code for taurus.core.taurusconfiguration

#!/usr/bin/env python
# -*- coding: utf-8 -*-

#############################################################################
##
## This file is part of Taurus
## 
## http://taurus-scada.org
##
## Copyright 2011 CELLS / ALBA Synchrotron, Bellaterra, Spain
## 
## Taurus is free software: you can redistribute it and/or modify
## it under the terms of the GNU Lesser General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
## 
## Taurus is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU Lesser General Public License for more details.
## 
## You should have received a copy of the GNU Lesser General Public License
## along with Taurus.  If not, see <http://www.gnu.org/licenses/>.
##
#############################################################################

"""This module contains the base class for a taurus attribute configuration"""

__all__ = ["TaurusConfigurationProxy", "TaurusConfiguration"]

__docformat__ = "restructuredtext"

from .taurusbasetypes import AttrAccess, TaurusElementType
from .taurusmodel import TaurusModel


[docs]class TaurusConfigurationProxy(object): """ TaurusAttribute has a reference to TaurusConfiguration and it should also have a reference to TaurusAttribute. To solve this cyclic dependency, TaurusConfiguration has a weak reference to TaurusAttribute. But then we must be sure that no other references to TaurusConfiguration exist so that no one tries to use it after its TaurusAttribute has disappeared. That's why to the outside world we don't give access to it directly but to objects of this new TaurusConfigurationProxy class. """ def __init__(self, parent): self.__parent = parent def __getattr__(self, name): return getattr(self.__parent._getRealConfig(), name)
[docs] def getRealConfigClass(self): return self.__parent._getRealConfig().__class__
[docs]class TaurusConfiguration(TaurusModel): no_cfg_value = '-----' no_unit = 'No unit' no_standard_unit = 'No standard unit' no_display_unit = 'No display unit' no_description = 'No description' not_specified = 'Not specified' no_min_value = no_max_value = not_specified no_min_alarm = no_max_alarm = not_specified no_min_warning = no_max_warning = not_specified no_delta_t = no_delta_val = not_specified no_rel_change = no_abs_change = not_specified no_archive_rel_change = no_archive_abs_change = not_specified no_archive_period = not_specified DftTimeToLive = 10000 # 10s def __init__(self, name, parent, storeCallback = None): # the last configuration value self._attr_info = None # the last time the configuration was read self._attr_timestamp = 0 # the configuration event identifier self._cfg_evt_id = None self.call__init__(TaurusModel, name, parent) # Everything went ok so now we are sure we can store the object if not storeCallback is None: storeCallback(self.getFullName(),self) self._dev_hw_obj = self._getDev().getHWObj() self._subscribeEvents() def __str__(self): return self.getFullName() def _subscribeEvents(self): pass
[docs] def cleanUp(self): self.trace("[TaurusConfiguration] cleanUp") self._unsubscribeEvents() self._attr_info = None self._dev_hw_obj = None TaurusModel.cleanUp(self)
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~- # TaurusModel implementation #-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~- @classmethod
[docs] def getTaurusElementType(cls): return TaurusElementType.Configuration
@classmethod
[docs] def buildModelName(cls, parent_model, relative_name): """build an 'absolute' model name from the parent model and the 'relative' name. - If parent_model is a TaurusAttribute, the return is a composition of the database model name and is device name - If parent_model is a TaurusConfiguration, the relative name is ignored and the parent name is returned """ if parent_model is None: return relative_name parent_name = parent_model.getFullName() if not parent_name: return relative_name if isinstance(parent_model, cls): return parent_name return '%s%s' % (parent_name,relative_name)
@classmethod
[docs] def getNameValidator(cls): import taurusvalidator return taurusvalidator.ConfigurationNameValidator()
def _getDevName(self): params = self.getNameValidator().getParams(self.getFullName()) return params.get('devicename') or params.get('devalias') def _getFullAttrName(self): params = self.getNameValidator().getParams(self.getFullName()) ret = params.get('devicename') or params.get('devalias') ret += '/' + params.get('attributename') return ret def _getAttrName(self): return self.getNameValidator().getParams(self.getFullName()).get('attributename') def _getDev(self): dev = None attrObj = self.getParentObj() if attrObj is None or attrObj.getParent() is None: devname = self._getDevName() dev = self.factory().getDevice(devname) else: dev = attrObj.getParent() return dev def _getAttr(self, createAttr=False): attrObj = self.getParentObj() if attrObj is None: attrname = self._getFullAttrName() if createAttr: attrObj = self.factory().getAttribute(attrname) else: attrObj = self.factory().getExistingAttribute(attrname) return attrObj
[docs] def getValueObj(self, cache=True): """ Returns the current configuration for the attribute. if cache is set to True (default) and the the configuration has events active then it will return the local cached value. Otherwise it will read from the tango layer.""" raise RuntimeError("May not be called in abstract TaurusConfiguration")
[docs] def getDisplayValue(self,cache=True): confvalue = self.getValueObj(cache=cache) if not confvalue: return None return confvalue.label
[docs] def getDisplayDescription(self,cache=True): return self.getDescription(cache)
[docs] def getDisplayDescrObj(self,cache=True): attrObj = self.getParentObj() if attrObj is None: return [('name', self.getLabel(cache=cache))] return attrObj.getDisplayDescrObj(cache=cache)
# def isWritable(self): # return True #-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~- # API for listeners #-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~- s # def addListener(self, listener): # """ Add a TaurusListener object in the listeners list. # If the listener is already registered nothing happens.""" # ret = TaurusModel.addListener(self, listener) # if not ret: # return ret # #fire a first configuration event # self.fireEvent(TaurusEventType.Config, self._attr_info, listener) # return ret
[docs] def hasEvents(self): self.deprecated("Don't use this anymore. Use isUsingEvents instead") return self.isUsingEvents()
[docs] def isUsingEvents(self): return self._cfg_evt_id
[docs] def isWritable(self, cache=True): return not self.isReadOnly(cache)
[docs] def getWritable(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.writable return None
[docs] def getType(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.type return None
[docs] def getDataFormat(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.data_format return None
[docs] def getMaxDim(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.max_dim return None
[docs] def getMaxDimX(self, cache=True): '''.. warning: Deprecated. Use :meth:`getMaxDim` ''' self.info('Deprecation warning: TaurusConfiguration.getMaxDimX is deprecated. Use getMaxDim') c = self.getValueObj(cache=cache) if c: return c.max_dim[0] return None
[docs] def getMaxDimY(self, cache=True): '''.. warning: Deprecated. Use :meth:`getMaxDim` ''' self.info('Deprecation warning: TaurusConfiguration.getMaxDimX is deprecated. Use getMaxDim') c = self.getValueObj(cache=cache) if c: return c.max_dim[1] return None
[docs] def getShape(self, cache=True): # force a read if not using cache. Then use cache until the # end of the method if not cache: self.getDataFormat(cache=cache) if self.isScalar(): return () elif self.isSpectrum(): return (self.getMaxDimX(),) else: return (self.getMaxDimX(), self.getMaxDimY())
[docs] def getDescription(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.description return None
[docs] def getLabel(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.label return None
[docs] def getUnit(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.unit return None
[docs] def getStandardUnit(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.standard_unit return None
[docs] def getDisplayUnit(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.display_unit return None
[docs] def getFormat(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.format return None
[docs] def getMinValue(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.range[0] return None
[docs] def getMaxValue(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.range[1] return None
[docs] def getLimits(self, cache=True): '''.. warning: Deprecated. Use :meth:`getRange` ''' self.info('Deprecation warning: TaurusConfiguration.getLimits is deprecated. Use getRange') self.getRange(cache=cache)
[docs] def getRange(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.range return None
[docs] def getRanges(self, cache=True): '''.. warning: Deprecated. Use :meth:`getRange` ''' self.info('Deprecation warning: TaurusConfiguration.getRanges is deprecated. Use getRange') c = self.getValueObj(cache=cache) if c: return [c.range[0], c.alarm[0], c.warning[0], c.warning[1], c.alarm[1], c.range[1]] return None
[docs] def getMinAlarm(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.alarm[0] return None
[docs] def getMaxAlarm(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.alarm[1] return None
[docs] def getAlarms(self, cache=True): c = self.getValueObj(cache=cache) if c: return list(c.alarm) return None
[docs] def getMinWarning(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.warning[0] return None
[docs] def getMaxWarning(self, cache=True): c = self.getValueObj(cache=cache) if c: return c.warning[1] return None
[docs] def getWarnings(self, cache=True): c = self.getValueObj(cache=cache) if c: return list(c.warning) return None
[docs] def getParam(self, param_name): config = self.getValueObj() if config: try: return getattr(config, param_name) except: return None
[docs] def setParam(self, param_name, value): config = self.getValueObj() if config and self.getParam(param_name): setattr(config, param_name, value) self._applyConfig()
[docs] def setDescription(self,descr): config = self.getValueObj() if config: config.description = descr self._applyConfig()
[docs] def setLabel(self,lbl): config = self.getValueObj() if config: config.label = lbl self._applyConfig()
[docs] def setUnit(self,unit): config = self.getValueObj() if config: config.unit = unit self._applyConfig()
[docs] def setStandardUnit(self,standard_unit): config = self.getValueObj() if config: config.standard_unit = standard_unit self._applyConfig()
[docs] def setDisplayUnit(self,display_unit): config = self.getValueObj() if config: config.display_unit = display_unit self._applyConfig()
[docs] def setFormat(self,fmt): config = self.getValueObj() if config: config.format = fmt self._applyConfig()
[docs] def setLimits(self, low, high): '''.. warning: Deprecated. Use :meth:`setRange` ''' self.info('Deprecation warning: TaurusConfiguration.setLimits is deprecated. Use setRange') self.setRange(low, high)
[docs] def setRange(self,low, high): config = self.getValueObj() if config: config.range = [low, high] self._applyConfig()
[docs] def setWarnings(self,low, high): config = self.getValueObj() if config: config.warning = [low, high] self._applyConfig()
[docs] def setAlarms(self,low, high): config = self.getValueObj() if config: config.alarm = [low, high] self._applyConfig()
def _applyConfig(self): pass
[docs] def isBoolean(self, cache=True): attr = self.getParentObj() if attr is None: return False v = attr.read(cache=cache) return isinstance(v.value, bool)
[docs] def isScalar(self, cache=True): attr = self.getParentObj() if attr is None: return False v = attr.read(cache=cache) import numpy return numpy.isscalar(v.value)
[docs] def isSpectrum(self, cache=True): attr = self.getParentObj() if attr is None: return False v = attr.read(cache=cache) import numpy return not numpy.isscalar(v.value) and numpy.array(v.value).ndim == 1
[docs] def isImage(self, cache=True): attr = self.getParentObj() if attr is None: return False v = attr.read(cache=cache) import numpy return not numpy.isscalar(v.value) and numpy.array(v.value).ndim == 2
[docs] def isWrite(self, cache=True): return self.getWritable(cache) == AttrAccess.WRITE
[docs] def isReadOnly(self, cache=True): return self.getWritable(cache) == AttrAccess.READ
[docs] def isReadWrite(self, cache=True): return self.getWritable(cache) == AttrAccess.READ_WRITE
#del AttrAccess #del TaurusModel