#!/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 device"""
__all__ = ["TaurusDevice"]
__docformat__ = "restructuredtext"
from .taurusbasetypes import TaurusSWDevState, TaurusEventType, \
TaurusLockInfo, TaurusElementType
from .taurusmodel import TaurusModel
DFT_DEVICE_DESCRIPTION = "A device"
[docs]class TaurusDevice(TaurusModel):
SHUTDOWNS = (TaurusSWDevState.Shutdown, TaurusSWDevState.Crash,
TaurusSWDevState.EventSystemShutdown)
"""A Device object representing an abstraction of the PyTango.DeviceProxy
object in the taurus.core layer"""
def __init__(self, name, **kw):
"""Object initialization."""
parent = kw.pop('parent', None)
storeCallback = kw.pop('storeCallback', None)
self.__dict__.update(kw)
self.call__init__(TaurusModel, name, parent)
self._deviceObj = self._createHWObject()
self._deviceStateObj = None
self._lock_info = TaurusLockInfo()
self._descr = None
self._deviceSwState = self.decode(TaurusSWDevState.Uninitialized)
if storeCallback:
storeCallback(self)
[docs] def cleanUp(self):
self.trace("[TaurusDevice] cleanUp")
self._deviceObj = None
self._descr = None
#self._deviceSwObj
if not self._deviceStateObj is None:
self._deviceStateObj.removeListener(self)
self._deviceStateObj = None
TaurusModel.cleanUp(self)
# Export the DeviceProxy interface into this object.
# This way we can call for example read_attribute on an object of this class
def __getattr__(self, name):
if '_deviceObj' in self.__dict__ and self._deviceObj is not None:
return getattr(self._deviceObj, name)
cls_name = self.__class__.__name__
raise AttributeError("'%s' has no attribute '%s'" % (cls_name, name))
#def __setattr__(self, name, value):
# if '_deviceObj' in self.__dict__ and self._deviceObj is not None:
# return setattr(self._deviceObj, name, value)
# super(TaurusDevice, self).__setattr__(name, value)
# Export the 'act like dictionary' feature of PyTango.DeviceProxy
def __getitem__(self, key):
attr = self.getAttribute(key)
return attr.read()
def __setitem__(self, key, value):
attr = self.getAttribute(key)
return attr.write(value)
def __contains__(self, key):
hw = self.getHWObj()
if hw is None:
return False
return hw.__contains__(key)
[docs] def getHWObj(self):
return self._deviceObj
[docs] def getStateObj(self):
if self._deviceStateObj is None:
self._deviceStateObj = self.factory().getAttribute("%s/state" % self.getFullName())
return self._deviceStateObj
[docs] def getState(self, cache=True):
stateAttrValue = self.getStateObj().getValueObj(cache=cache)
if not stateAttrValue is None:
return stateAttrValue.value
return None
[docs] def getSWState(self, cache=True):
return self.getValueObj(cache=cache).value
[docs] def getAttribute(self, attrname):
"""Returns the attribute object given its name"""
slashnb = attrname.count('/')
if slashnb == 0:
attrname = "%s/%s" % (self.getFullName(), attrname)
elif attrname[0] == '/':
attrname = "%s%s" % (self.getFullName(), attrname)
import taurusattribute
return self.factory().getObject(taurusattribute.TaurusAttribute,attrname)
[docs] def isValidDev(self):
"""returns True if the device is in "working conditions
The default implementation always returns True. Reimplement it in
subclasses if there are cases in which the device cannot be queried
(e.g. in Tango, the TangoDevice object may exist even if there is not a real
hardware device associated, in which case this method should return False)
"""
return True
[docs] def getLockInfo(self, cache=False):
return self._lock_info
[docs] def lock(self, force=False):
pass
[docs] def unlock(self, force=False):
pass
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# Mandatory implementation in sub class
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
def _createHWObject(self):
raise NotImplementedError
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# TaurusModel implementation
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
@classmethod
[docs] def getTaurusElementType(cls):
return TaurusElementType.Device
@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 TaurusDatabase, the return is a composition of
the database model name and is device name
- If parent_model is a TaurusDevice, 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.DeviceNameValidator()
[docs] def getValueObj(self, cache=True):
if not self.hasListeners() or not cache:
try:
v = self.getStateObj().read(cache=cache)
except Exception as e:
v = e
self._deviceSwState = self.decode(v)
return self._deviceSwState
[docs] def getDisplayValue(self,cache=True):
return TaurusSWDevState.whatis(self.getValueObj(cache).value)
[docs] def getDisplayDescrObj(self,cache=True):
obj = []
obj.append(('name', self.getDisplayName(cache=cache)))
descr = self.getDescription(cache=cache)
if descr.lower() != self._getDefaultDescription().lower():
obj.append(('description', descr))
obj.append(('device state', self.getStateObj().getDisplayValue()) or self.getNoneValue())
obj.append(('SW state', self.getDisplayValue()))
return obj
[docs] def getDescription(self,cache=True):
if self._descr is None or not cache:
try:
self._descr = self.description()
except:
self._descr = self._getDefaultDescription()
return self._descr
[docs] def removeListener(self, listener):
ret = TaurusModel.removeListener(self, listener)
if not ret or self.hasListeners():
return ret # False, None or True
return self.getStateObj().removeListener(self)
[docs] def addListener(self, listener):
weWereListening = self.hasListeners()
ret = TaurusModel.addListener(self, listener)
if not ret:
return ret
# We are only listening to State if someone is listening to us
if weWereListening:
# We were listening already, so we must fake an event to the new
# subscribed listener with the current value
self.fireEvent(TaurusEventType.Change, self.getValueObj(), hasattr(listener,'__iter__') and listener or [listener])
else:
# We were not listening to events, but now we have to
self.getStateObj().addListener(self)
return ret
[docs] def getChildObj(self, child_name):
if child_name is None or len(child_name) == 0:
return None
return self.factory().findObject("%s%s" % (self.getFullName(), child_name))
return self.getAttribute(child_name)
[docs] def eventReceived(self, event_src, event_type, event_value):
if event_type == TaurusEventType.Config:
return
value = self.decode(event_value)
if value.value != self._deviceSwState.value:
msg = "SW Device State changed %s -> %s" %\
(TaurusSWDevState.whatis(self._deviceSwState.value),
TaurusSWDevState.whatis(value.value))
self.debug(msg)
self._deviceSwState = value
self.fireEvent(TaurusEventType.Change, value)
def _getDefaultDescription(self):
return DFT_DEVICE_DESCRIPTION
[docs] def poll(self, attrs, asynch=False, req_id=None):
'''Polling certain attributes of the device. This default
implementation simply polls each attribute one by one'''
# asynchronous requests are not supported. If asked to do it,
# just return an ID of 1 and in the reply (req_id != None) we do a
# synchronous polling.
if asynch is True:
return 1
for attr in attrs.values():
attr.poll()