#!/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"]
__docformat__ = "restructuredtext"
import numpy
from taurus.core.taurusbasetypes import DataType, TaurusElementType
from taurus.core.units import Quantity
from taurus.core.util.log import deprecation_decorator
from .taurusmodel import TaurusModel
[docs]
class TaurusAttribute(TaurusModel):
DftTimeToLive = 10000 # 10s
_description = "A Taurus Attribute"
defaultFragmentName = "rvalue" # fragment to be used if none is specified
def __init__(self, name="", parent=None, **kwargs):
self.call__init__(TaurusModel, name, parent)
# User enabled/disabled polling
self.__enable_polling = kwargs.get("enablePolling", True)
# attribute should be polled.
# The attribute is polled only if the polling is also enabled
self.__activate_polling = False
# Indicates 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
# If everything went well, the object is stored
storeCallback = kwargs.get("storeCallback", None)
if storeCallback is not None:
storeCallback(self)
self.writable = None
self.data_format = None
self._label = self.getSimpleName()
self.type = None
self._range = [None, None]
self._alarm = [None, None]
self._warning = [None, None]
self.precision = None
self.enum_labels = []
[docs]
def cleanUp(self):
self.trace("[TaurusAttribute] cleanUp")
if hasattr(self, "_unsuscribeEvents"):
self.deprecated(
dep="TaurusAttribute._unsuscribeEvents API",
alt="If you need it called in cleanUp, re-implement cleanUp",
)
self._unsuscribeEvents()
TaurusModel.cleanUp(self)
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
# TaurusModel implementation
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
[docs]
@classmethod
def getTaurusElementType(cls):
return TaurusElementType.Attribute
[docs]
@classmethod
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 its device name
- If parent_model is a TaurusAttribute, the relative name is ignored
and the parent name is returned
Note: This is a basic implementation. You may need to reimplement this
for a specific scheme if it supports "useParentModel".
"""
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)
[docs]
@classmethod
def getNameValidator(cls):
return cls.factory().getAttributeNameValidator()
[docs]
def isNumeric(self):
return self.type in [DataType.Float, DataType.Integer]
@deprecation_decorator(rel=">4.0.1", alt=".type==DataType.DevState")
def isState(self):
return False
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
# Necessary to overwrite in subclass
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
[docs]
def encode(self, value):
raise NotImplementedError(
"Not allowed to call AbstractClass" + " TaurusAttribute.encode"
)
[docs]
def decode(self, attr_value):
raise NotImplementedError(
"Not allowed to call AbstractClass" + " TaurusAttribute.decode"
)
[docs]
def write(self, value, with_read=True):
raise NotImplementedError(
"Not allowed to call AbstractClass" + " TaurusAttribute.write"
)
[docs]
def read(self, cache=True):
raise NotImplementedError(
"Not allowed to call AbstractClass" + " TaurusAttribute.read"
)
[docs]
def poll(self):
raise NotImplementedError(
"Not allowed to call AbstractClass" + " TaurusAttribute.poll"
)
[docs]
def isUsingEvents(self):
raise NotImplementedError(
"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 areStrValuesEqual(self, v1, v2):
try:
if "nan" == str(v1).lower() == str(v2).lower():
return True
e1, e2 = self.encode(v1), self.encode(v2)
return e1 == e2 or numpy.allclose(e1, e2, rtol=1e-15, atol=1e-15)
except Exception:
return False
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
# API for listeners
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
[docs]
def hasEvents(self):
self.deprecated("Don't use this anymore. Use isUsingEvents instead")
return self.isUsingEvents()
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
# First time read (client side)
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
def _readFirstTime(self):
"""Request a first-time read for this attribute (client-side).
This private helper delegates to
:meth:`~taurus.core.taurusfactory.TaurusFactory.readAttributeFirstTime`
so that the attribute value is fetched once during initialization,
ensuring that its cache is primed before normal event subscriptions
or polling begin.
It is typically invoked automatically as part of the client-side
setup sequence when connecting to a Tango device or another data source.
:return: None
:rtype: None
"""
self.factory().readAttributeFirstTime(attribute=self)
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
# Polling (client side)
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
[docs]
def enablePolling(self, force=False):
"""Enable polling. See :meth:`isPollingEnabled` for clarification of
what enabled polling means.
:param force: True also activates polling (see:
:meth:`activatePolling`)
:type force: bool
"""
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: whether polling is enabled
:rtype: bool
: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: whether polling is active
:rtype: bool
: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
[docs]
def activatePolling(self, period, unsubscribe_evts=False, force=False):
"""activate polling for attribute.
:param period: polling period (in miliseconds)
:type period: int
"""
self.changePollingPeriod(period)
self.enablePolling(force=force)
[docs]
def deactivatePolling(self, maintain_enabled=False):
"""unregister attribute from polling"""
self.deprecated("use disablePolling()")
self.disablePolling()
def __str__(self):
return self.getFullName()
[docs]
def getDisplayDescription(self, cache=True):
return self.description
[docs]
def getDisplayDescrObj(self, cache=True):
name = self.getLabel(cache=cache)
obj = [("name", name), ("model", self.getFullName() or "")]
descr = self.description
if descr:
_descr = descr.replace("<", "<").replace(">", ">")
obj.append(("description", _descr))
if isinstance(self.rvalue, Quantity):
_unitless = self.rvalue.unitless
range = self._range
alarm = self._alarm
warning = self._warning
if range != [None, None]:
if not _unitless:
low = range[0]
high = range[1]
else:
low = range[0].magnitude
high = range[1].magnitude
obj.append(("range", "[%s, %s]" % (low, high)))
if alarm != [None, None]:
if not _unitless:
low = alarm[0]
high = alarm[1]
else:
low = alarm[0].magnitude
high = alarm[1].magnitude
obj.append(("alarm", "[%s, %s]" % (low, high)))
if warning != [None, None]:
if not _unitless:
low = warning[0]
high = warning[1]
else:
low = warning[0].magnitude
high = warning[1].magnitude
obj.append(("warning", "[%s, %s]" % (low, high)))
return obj
[docs]
def isWritable(self, cache=True):
return self.writable
[docs]
def getType(self, cache=True):
return self.type
[docs]
def getLabel(self, cache=True):
return self._label
[docs]
def getMinRange(self, cache=True):
return self._range[0]
[docs]
def getMaxRange(self, cache=True):
return self._range[1]
[docs]
def getRange(self, cache=True):
return self._range
[docs]
def getMinAlarm(self, cache=True):
return self._alarm[0]
[docs]
def getMaxAlarm(self, cache=True):
return self._alarm[1]
[docs]
def getAlarms(self, cache=True):
return self._alarm
[docs]
def getMinWarning(self, cache=True):
return self._warning[0]
[docs]
def getMaxWarning(self, cache=True):
return self._warning[1]
[docs]
def getWarnings(self, cache=True):
return self._warning
[docs]
def setLabel(self, lbl):
self._label = lbl
def __assertsValidLimits(self, limits):
assert len(limits) == 2, "The limits must be two values, low and high"
low, high = limits
assert isinstance(self.rvalue, Quantity), "rvalue is not a Quantity"
assert isinstance(low, Quantity), "low is not a Quantity"
assert isinstance(high, Quantity), "high is not a Quantity"
assert self.rvalue.dimensionality == low.dimensionality, (
"low and rvalue have different dimensionality"
)
assert self.rvalue.dimensionality == high.dimensionality, (
"high and rvalue have different dimensionality"
)
[docs]
def setRange(self, *limits):
if isinstance(limits[0], list):
limits = limits[0]
self.__assertsValidLimits(limits)
self._range = limits
[docs]
def setWarnings(self, *limits):
if isinstance(limits[0], list):
limits = limits[0]
self.__assertsValidLimits(limits)
self._warning = limits
[docs]
def setAlarms(self, *limits):
if isinstance(limits[0], list):
limits = limits[0]
self.__assertsValidLimits(limits)
self._alarm = limits
[docs]
def isBoolean(self, cache=True):
v = self.read(cache)
return isinstance(v.rvalue, bool)
@property
def description(self):
return self._description
@description.setter
def description(self, descr):
self._description = descr
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
# API to provide access to the TaurusAttributeValue-related fragments
# -~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~
@property
def rvalue(self, cache=True):
valueObj = self.getValueObj(cache=cache)
if valueObj is None:
return None
else:
return valueObj.rvalue
@property
def wvalue(self, cache=True):
valueObj = self.getValueObj()
if valueObj is None:
return None
else:
return valueObj.wvalue
@property
def time(self, cache=True):
valueObj = self.getValueObj()
if valueObj is None:
return None
else:
return valueObj.time
@property
def quality(self, cache=True):
valueObj = self.getValueObj()
if valueObj is None:
return None
else:
return valueObj.quality
label = property(getLabel, setLabel)
range = property(getRange, setRange)
warnings = property(getWarnings, setWarnings)
alarms = property(getAlarms, setAlarms)