#!/usr/bin/env python
#############################################################################
##
## 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"""
__all__ = ["TaurusAttribute", "TaurusStateAttribute"]
__docformat__ = "restructuredtext"
import weakref
from taurus.core.taurusbasetypes import TaurusElementType
from .taurusmodel import TaurusModel
from .taurusconfiguration import TaurusConfigurationProxy
[docs]class TaurusAttribute(TaurusModel):
def __init__(self, name, parent, **kwargs):
self.call__init__(TaurusModel, name, parent)
self.__parentDevice = parent # just to keep it alive
# User enabled/disabled polling
self.__enable_polling = kwargs.get('enablePolling', True)
# attribute should be polled. The attribute is in fact polled only if the polling is also enabled
self.__activate_polling = False
# efectively tells if the attribute is being polled periodically
# in summary: polled = enable_polling and activate_polling
self.__polled = False
# current polling period
self.__polling_period = kwargs.get("pollingPeriod", 3000)
# stores if polling has been forced by user API
self.__forced_polling = False
# Everything went ok so now we are sure we can store the object
storeCallback = kwargs.get("storeCallback", None)
if not storeCallback is None:
storeCallback(self)
self._dev_hw_obj = parent.getHWObj()
[docs] def cleanUp(self):
self.trace("[TaurusAttribute] cleanUp")
self._unsubscribeEvents()
self._dev_hw_obj = None
TaurusModel.cleanUp(self)
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# TaurusModel implementation
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
@classmethod
[docs] def getTaurusElementType(cls):
return TaurusElementType.Attribute
@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 TaurusDevice, the return is a composition of
the database model name and is device name
- If parent_model is a TaurusAttribute, 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.AttributeNameValidator()
# received configuration events
[docs] def eventReceived(self, src, src_type, evt_value):
"""Method invoked by the configuration object when a configuration event
is received. Default implementation propagates the event to all
listeners."""
#self.fireEvent(src_type, evt_value)
pass
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# Necessary to overwrite in subclass
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def isNumeric(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.isNumeric")
[docs] def isBoolean(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.isBoolean")
[docs] def isState(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.isState")
[docs] def getDisplayValue(self,cache=True):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.getDisplayValue")
[docs] def encode(self, value):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.encode")
[docs] def decode(self, attr_value):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.decode")
[docs] def write(self, value, with_read=True):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.write")
[docs] def read(self, cache=True):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.read")
[docs] def poll(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.poll")
def _subscribeEvents(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute._subscribeEvents")
def _unsubscribeEvents(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute._unsubscribeEvents")
[docs] def isUsingEvents(self):
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute.isUsingEvents")
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# TaurusModel necessary overwrite
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def getValueObj(self, cache=True):
try:
return self.read(cache=cache)
except Exception:
return None
[docs] def getDisplayDescrObj(self,cache=True):
name = self.getLabel(cache=cache)
if name:
name += " (" + self.getNormalName().upper() + ")"
else:
name = self.getDisplayName(cache=cache)
obj = [('name', name)]
descr = self.getDescription(cache=cache)
if descr and descr != self.no_description:
obj.append(('description',descr.replace("<","<").replace(">",">")))
unit = self.getUnit(cache=cache)
if unit and unit != self.no_unit:
obj.append(('unit', unit))
limits = self.getRange(cache=cache)
if limits and (limits[0] != self.no_min_value or \
limits[1] != self.no_max_value):
if limits[0] == self.no_min_value: limits[0] = self.no_cfg_value
if limits[1] == self.no_max_value: limits[1] = self.no_cfg_value
obj.append(('limits', "[%s, %s]" % (limits[0],limits[1])))
alarms = self.getAlarms(cache=cache)
if alarms and (alarms[0] != self.no_min_alarm or \
alarms[1] != self.no_max_alarm):
if alarms[0] == self.no_min_alarm: alarms[0] = self.no_cfg_value
if alarms[1] == self.no_max_alarm: alarms[1] = self.no_cfg_value
obj.append(('alarms', "[%s, %s]" % (alarms[0],alarms[1])))
warnings = self.getWarnings(cache=cache)
if warnings and (warnings[0] != self.no_min_warning or \
warnings[1] != self.no_max_warning):
if warnings[0] == self.no_min_warning: warnings[0] = self.no_cfg_value
if warnings[1] == self.no_max_warning: warnings[1] = self.no_cfg_value
obj.append(('warnings', "[%s, %s]" % (warnings[0],warnings[1])))
return obj
[docs] def areStrValuesEqual(self,v1,v2):
try:
if "nan" == str(v1).lower() == str(v2).lower(): return True
return self.encode(v1) == self.encode(v2)
except:
return False
[docs] def getDisplayWriteValue(self,cache=True):
if not self.isWritable():
self.warning("requesting write value of a read-only attribute")
return None
attrvalue = self.getValueObj(cache=cache)
if not attrvalue:
return None
v = attrvalue.w_value
return self.displayValue(v)
[docs] def displayValue(self,value):
if value is None:
return None
ret = None
try:
if self.isScalar():
fmt = self.getFormat()
if self.isNumeric() and fmt is not None:
ret = fmt % value
else:
ret = str(value)
elif self.isSpectrum():
ret = str(value)
else:
ret = str(value)
except:
# if cannot calculate value based on the format just return the value
ret = str(value)
return ret
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# API for listeners
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def hasEvents(self):
self.deprecated("Don't use this anymore. Use isUsingEvents instead")
return self.isUsingEvents()
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# Polling (client side)
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def enablePolling(self, force=False):
'''Enable polling. See :meth:`isPollingEnabled` for clarification of
what enabled polling means.
:param force: (bool) True also activates polling
(see: :meth:`activatePolling`)
'''
self.__enable_polling = True
self.__forced_polling = force
if force:
self.__activate_polling = True
if self.__activate_polling:
self._activatePolling()
[docs] def disablePolling(self):
'''Disable polling and if polling is active also deactivate it.
See :meth:`isPollingEnabled` for clarification of
what enabled polling means.
'''
self.__enable_polling = False
self.__forced_polling = False
if self.__activate_polling:
self._deactivatePolling()
[docs] def isPollingEnabled(self):
'''Indicate whether polling was activated/deactivated by user.
Enabled polling does not mean that it is active - periodically poll
the attribute. By default the attribute creation enables polling.
:return: (bool) whether polling is enabled
:see: :meth:`enablePolling`, :meth:`disablePolling`
'''
return self.__enable_polling
def _activatePolling(self):
self.__activate_polling = True
if not self.isPollingEnabled():
return
self.factory().addAttributeToPolling(self, self.getPollingPeriod())
self.__polled = True
def _deactivatePolling(self):
self.__activate_polling = False
self.factory().removeAttributeFromPolling(self)
self.__polled = False
[docs] def isPollingActive(self):
'''Indicate whether polling is active. Active polling means that
a periodic timer poll the attribute. By default the attribute creation
does not activate polling.
:return: (bool) whether polling is active
:see: :meth:`activatePolling`, :meth:`disablePolling`
'''
return self.__polled
[docs] def isPollingForced(self):
return self.__forced_polling
[docs] def changePollingPeriod(self, period):
"""change polling period to period miliseconds """
if self.__polling_period == period and self.__activate_polling:
return
self.__polling_period = period
if self.__activate_polling:
self._deactivatePolling()
self._activatePolling()
[docs] def isPolled(self):
self.deprecated("use isPollingActive()")
return self.isPollingActive()
[docs] def getPollingPeriod(self):
"""returns the polling period """
return self.__polling_period
# The following are deprecated and maintained only for compatibility
[docs] def activatePolling(self, period, unsubscribe_evts=False, force=False):
"""activate polling for attribute.
:param period: polling period (in miliseconds)
:type period: int
"""
## REENABLED, used to solve problems with ID GUI's and other systems where event independency is needed.
#self.deprecated("use changePollingPeriod(). Not exactly the same functionality. Only activates polling if necessary")
self.changePollingPeriod(period)
self.enablePolling(force=force)
[docs] def deactivatePolling(self, maintain_enabled=False):
"""unregister attribute from polling"""
self.deprecated("use disablePolling()")
self.disablePolling()
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# API for attribute configuration
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
def _getRealConfig(self):
""" Returns the current configuration of the attribute."""
raise RuntimeError("Not allowed to call AbstractClass TaurusAttribute::_getRealConfig")
[docs] def getConfig(self):
""" Returns the current configuration of the attribute."""
try:
ob = None
ob = self.__weakFakeConfigObj()
except Exception:
pass
if ob is not None:
return ob
ob = TaurusConfigurationProxy(self)
self.__weakFakeConfigObj = weakref.ref(ob)
return ob
[docs] def isWritable(self, cache=True):
return not self._getRealConfig().isReadOnly(cache=cache)
[docs] def isWrite(self, cache=True):
return self._getRealConfig().isWrite(cache=cache)
[docs] def isReadOnly(self, cache=True):
return self._getRealConfig().isReadOnly(cache=cache)
[docs] def isReadWrite(self, cache=True):
return self._getRealConfig().isReadWrite(cache=cache)
[docs] def getWritable(self, cache=True):
return self._getRealConfig().getWritable(cache=cache)
[docs]class TaurusStateAttribute(TaurusAttribute):
""" """
def __init__(self, name, parent, **kwargs):
self.call__init__(TaurusAttribute, name, parent, **kwargs)
#del weakef
#del TaurusModel
#del TaurusConfigurationProxy