#!/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/>.
##
#############################################################################
"""
The TaurusFactory model containning the abstract base class that any valid
Factory in Taurus must inherit.
Taurus object naming is URI based:
foo://username:password@example.com:8042/over/there/index.dtb;type=animal?name=ferret#nose
\ / \________________/\_________/ \__/\_________/ \___/ \_/ \_________/ \_________/ \__/
| | | | | | | | | |
scheme userinfo hostname port path filename extension parameter(s) query fragment
\________________________________/
authority
"""
__all__ = ["TaurusFactory"]
__docformat__ = "restructuredtext"
from .taurusbasetypes import OperationMode
from .taurusdatabase import TaurusDatabase
from .taurusdevice import TaurusDevice
from .taurusattribute import TaurusAttribute
from .taurusconfiguration import TaurusConfiguration, TaurusConfigurationProxy
[docs]class TaurusFactory(object):
"""The base class for valid Factories in Taurus."""
schemes = ()
DefaultPollingPeriod = 3000
def __init__(self):
self._polling_period = self.DefaultPollingPeriod
self.operation_mode = OperationMode.ONLINE
self.polling_timers = {}
self._polling_enabled = True
import taurusmanager
manager = taurusmanager.TaurusManager()
self._serialization_mode = manager.getSerializationMode()
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# API for serialization
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def setSerializationMode(self, mode):
"""Sets the serialization mode for the system.
:param mode: (TaurusSerializationMode) the new serialization mode"""
self._serialization_mode = mode
[docs] def getSerializationMode(self):
"""Gives the serialization operation mode.
:return: (TaurusSerializationMode) the current serialization mode"""
return self._serialization_mode
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# Methods that must be implemented by the specific Factory
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def findObjectClass(self, absolute_name):
"""findObjectClass(string absolute_name) -> taurus.core.taurusmodel.TaurusModel subclass
Obtain the class object corresponding to the given name.
@param[in] absolute_name the object absolute name string
@return a class object that should be a subclass of a taurus.core.taurusmodel.TaurusModel
@throws TaurusException if the given name is invalid.
"""
raise RuntimeError("findObjectClass cannot be called for abstract" \
" TaurusFactory")
[docs] def getDatabase(self, db_name=None):
"""getDatabase(string db_name) -> taurus.core.taurusdatabase.TaurusDatabase
Obtain the object corresponding to the given database name or the
default database if db_name is None.
If the corresponding database object already exists, the existing
instance is returned. Otherwise a new instance is stored and returned.
@param[in] db_name database name string. It should be formed like:
<scheme>://<authority>. If <scheme> is ommited then
it will use the default scheme. if db_name is None,
the default database is used
@return a taurus.core.taurusdatabase.TaurusDatabase object
@throws TaurusException if the given name is invalid.
"""
raise RuntimeError("getDatabase cannot be called for abstract" \
" TaurusFactory")
[docs] def getDevice(self, dev_name, **kw):
"""getDevice(string dev_name) -> taurus.core.taurusdevice.TaurusDevice
Obtain the object corresponding to the given device name. If the
corresponding device already exists, the existing instance is returned.
Otherwise a new instance is stored and returned.
@param[in] dev_name the device name string. It should be formed like:
<scheme>://<authority>/<device name>. If <scheme>
is ommited then it will use the default scheme.
If authority is ommited then it will use the
default authority for the scheme.
@return a taurus.core.taurusdevice.TaurusDevice object
@throws TaurusException if the given name is invalid.
"""
raise RuntimeError("getDevice cannot be called for abstract" \
" TaurusFactory")
[docs] def getAttribute(self, attr_name):
"""getAttribute(string attr_name) -> taurus.core.taurusattribute.TaurusAttribute
Obtain the object corresponding to the given attribute name.
If the corresponding attribute already exists, the existing instance
is returned. Otherwise a new instance is stored and returned.
@param[in] attr_name string attribute name
@return a taurus.core.taurusattribute.TaurusAttribute object
@throws TaurusException if the given name is invalid.
"""
raise RuntimeError("getAttribute cannot be called for abstract" \
" TaurusFactory")
[docs] def getConfiguration(self, param):
"""getConfiguration(param) -> taurus.core.taurusconfiguration.TaurusConfiguration
Obtain the object corresponding to the given attribute or full name.
If the corresponding configuration already exists, the existing instance
is returned. Otherwise a new instance is stored and returned.
@param[in] param taurus.core.taurusattribute.TaurusAttribute object or full configuration name
@return a taurus.core.taurusattribute.TaurusAttribute object
@throws TaurusException if the given name is invalid.
"""
raise RuntimeError("getConfiguration cannot be called for abstract" \
" TaurusFactory")
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# Factory extension API
# Override the following methods if you need to provide special classes for
# special object types
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def registerAttributeClass(self, attr_name, attr_klass):
pass
[docs] def unregisterAttributeClass(self, attr_name):
pass
[docs] def registerDeviceClass(self, dev_klass_name, dev_klass):
pass
[docs] def unregisterDeviceClass(self, dev_klass_name):
pass
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
# Generic methods
#-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-~-
[docs] def supportsScheme(self, scheme):
return scheme in self.shemes
[docs] def setOperationMode(self, mode):
""" setOperationMode(OperationMode mode) -> None
Sets the operation mode for the Tango system."""
self.operation_mode = mode
[docs] def getOperationMode(self):
return self.operation_mode
[docs] def findObject(self, absolute_name):
""" Must give an absolute name"""
if self.operation_mode == OperationMode.OFFLINE or not absolute_name:
return None
obj = None
cls = self.findObjectClass(absolute_name)
if cls:
obj = self.getObject(cls, absolute_name)
return obj
[docs] def getObject(self, cls, name):
if issubclass(cls, TaurusDatabase):
return self.getDatabase(name)
elif issubclass(cls, TaurusDevice):
return self.getDevice(name)
elif issubclass(cls, TaurusAttribute):
return self.getAttribute(name)
elif issubclass(cls, TaurusConfiguration):
return self.getConfiguration(name)
elif issubclass(cls, TaurusConfigurationProxy):
return self.getConfiguration(name)
else:
return None
[docs] def changeDefaultPollingPeriod(self, period):
if period > 0:
self._polling_period = period
[docs] def getDefaultPollingPeriod(self):
return self._polling_period
[docs] def isPollingEnabled(self):
"""Tells if the local tango polling is enabled
:return: (bool) wheter or not the polling is enabled
"""
return self._polling_enabled
[docs] def disablePolling(self):
"""Disable the application tango polling"""
if not self.isPollingEnabled():
return
self._polling_enabled = False
for period,timer in self.polling_timers.iteritems():
timer.stop()
[docs] def enablePolling(self):
"""Enable the application tango polling"""
if self.isPollingEnabled():
return
for period,timer in self.polling_timers.iteritems():
timer.start()
self._polling_enabled = True
[docs] def addAttributeToPolling(self, attribute, period, unsubscribe_evts = False):
"""Activates the polling (client side) for the given attribute with the
given period (seconds).
:param attribute: (taurus.core.tango.TangoAttribute) attribute name.
:param period: (float) polling period (in seconds)
:param unsubscribe_evts: (bool) whether or not to unsubscribe from events
"""
raise RuntimeError("addAttributeToPolling cannot be called for abstract" \
" TaurusFactory")
[docs] def removeAttributeFromPolling(self, attribute):
"""Deactivate the polling (client side) for the given attribute. If the
polling of the attribute was not previously enabled, nothing happens.
:param attribute: (str) attribute name.
"""
raise RuntimeError("removeAttributeFromPolling cannot be called for abstract" \
" TaurusFactory")
def __str__(self):
return '{0}()'.format(self.__class__.__name__)
def __repr__(self):
return '{0}(schemes={1})'.format(self.__class__.__name__, ", ".join(self.schemes))