#!/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/>.
#
# ###########################################################################
try:
pass
except ImportError:
# note that if PyTango is not installed the factory will not be available
from taurus.core.util.log import debug
msg = (
"cannot import PyTango module. "
+ 'Taurus will not support the "tango" scheme'
)
debug(msg)
raise
import PyTango
from taurus import tauruscustomsettings
from taurus.core.taurusbasetypes import (
TaurusElementType,
TaurusSerializationMode,
)
from taurus.core.taurusfactory import TaurusFactory
from taurus.core.taurusbasetypes import OperationMode
from taurus.core.taurusexception import TaurusException, DoubleRegistration
from taurus.core.util.log import Logger, taurus4_deprecation
from taurus.core.util.singleton import Singleton
from taurus.core.util.containers import CaselessWeakValueDict, CaselessDict
from .tangodatabase import TangoAuthority
from .tangoattribute import TangoAttribute
from .tangodevice import TangoDevice
_Authority = TangoAuthority
_Attribute = TangoAttribute
_Device = TangoDevice
__docformat__ = "restructuredtext"
[docs]
class TangoFactory(Singleton, TaurusFactory, Logger):
r"""A :class:`TaurusFactory` singleton class to provide Tango-specific
Taurus Element objects (TangoAuthority, TangoDevice, TangoAttribute)
Tango model names are URI based See https://tools.ietf.org/html/rfc3986.
For example, a TangoAttribute would be::
tango://foo.org:1234/a/b/c/d#label
\___/ \_____/ \__/ \_____/ \___/
| | | | |
| hostname port attr |
| \____________/\______/ |
| | | |
scheme authority path fragment
For Tango Elements:
- The 'scheme' must be the string "tango" (lowercase mandatory)
- The 'authority' identifies the Tango database (<hostname> and <port>
are mandatory if authority is given)
- The 'path' identifies Tango Device and Attributes.
For devices it must have the format _/_/_ or alias
For attributes it must have the format _/_/_/_ or devalias/_
- The 'fragment' is optional and it refers to a member of the model
object, thus not being part of the model name itself
"""
#: the list of schemes that this factory supports. This factory
# supports 'tango' and 'tango-nodb' schemes
schemes = ("tango", "tango-nodb")
caseSensitive = False
elementTypesMap = {
TaurusElementType.Authority: TangoAuthority,
TaurusElementType.Device: TangoDevice,
TaurusElementType.Attribute: TangoAttribute,
}
def __init__(self):
"""Initialization. Nothing to be done here for now."""
pass
[docs]
def init(self, *args, **kwargs):
"""Singleton instance initialization.
**For internal usage only**
"""
name = self.__class__.__name__
self.call__init__(Logger, name)
self.call__init__(TaurusFactory)
self._polling_enabled = True
self.reInit()
self.scheme = "tango"
self._serialization_mode = TaurusSerializationMode.get(
getattr(
tauruscustomsettings, "TANGO_SERIALIZATION_MODE", "TangoSerial"
)
)
[docs]
def reInit(self):
"""Reinitialize the singleton"""
self._default_tango_host = None
self._tango_subscribe_enabled = True
self.dft_db = None
self.tango_db = CaselessWeakValueDict()
self.tango_db_queries = CaselessWeakValueDict()
self.tango_attrs = CaselessWeakValueDict()
self.tango_devs = CaselessWeakValueDict()
self.tango_dev_queries = CaselessWeakValueDict()
self.tango_alias_devs = CaselessWeakValueDict()
self.polling_timers = {}
# Plugin device classes
self.tango_dev_klasses = {}
# Plugin attribute classes
self.tango_attr_klasses = CaselessDict()
[docs]
def cleanUp(self):
"""Cleanup the singleton instance"""
self.trace("[TangoFactory] cleanUp")
for k, v in self.tango_attrs.items():
v.cleanUp()
for k, v in self.tango_dev_queries.items():
v.cleanUp()
for k, v in self.tango_devs.items():
v.cleanUp()
self.dft_db = None
for k, v in self.tango_db_queries.items():
v.cleanUp()
for k, v in self.tango_db.items():
v.cleanUp()
try:
PyTango.Util.instance(False)
except PyTango.DevFailed:
try:
PyTango.ApiUtil.cleanup()
except AttributeError:
pass
self.reInit()
[docs]
def getExistingAttributes(self):
"""Returns a new dictionary will all registered attributes on this
factory
:return: dictionary will all registered attributes on this factory
:rtype: dict
"""
return dict(self.tango_attrs)
[docs]
def getExistingDevices(self):
"""Returns a new dictionary will all registered devices on this factory
:return: dictionary will all registered devices on this factory
:rtype: dict
"""
return dict(self.tango_devs)
[docs]
def getExistingDatabases(self):
"""Returns a new dictionary will all registered databases on this
factory
:return: dictionary will all registered databases on this factory
:rtype: dict
"""
return dict(self.tango_db)
[docs]
def set_default_tango_host(self, tango_host):
"""
Sets the new default tango host. The method will transform the given
name to an Authority URI.
.. note:: Calling this method also clears the device alias cache.
:param tango_host: the new tango host. It accepts any valid Tango
authority name or None to use the defined by $TANGO_HOST env. var.
:type tango_host: str
"""
# Translate to Authority URI
if tango_host and "//" not in tango_host:
tango_host = "//{0}".format(tango_host)
v = self.getAuthorityNameValidator()
self._default_tango_host = v.getUriGroups(tango_host)["authority"]
self.tango_alias_devs.clear()
self.dft_db = None
[docs]
def get_default_tango_host(self):
"""Retruns the current default tango host"""
return self._default_tango_host
[docs]
def set_tango_subscribe_enabled(self, value):
"""If True, enables event subscribing on TangoAttribute objects
.. warning:: This method belongs to a "Delayed Event Subscription" API
added in v.4.2.1-alpha as an *experimental* feature. This API
may not be stable and/or it may be removed in a future release
(even on a minor version change)
"""
self._tango_subscribe_enabled = value
[docs]
def is_tango_subscribe_enabled(self):
"""Returns the current tango_subscribe_enabled status
.. warning:: This method belongs to a "Delayed Event Subscription" API
added in v.4.2.1-alpha as an *experimental* feature. This API
may not be stable and/or it may be removed in a future release
(even on a minor version change)
"""
return self._tango_subscribe_enabled
[docs]
def registerAttributeClass(self, attr_name, attr_klass):
"""Registers a new attribute class for the attribute name.
:param attr_name: attribute name
:type attr_name: str
:param attr_klass: the new class that will handle the attribute
:type attr_klass: taurus.core.tango.TangoAttribute
"""
self.tango_attr_klasses[attr_name] = attr_klass
[docs]
def unregisterAttributeClass(self, attr_name):
"""Unregisters the attribute class for the given attribute
If no class was registered before for the given attribute, this call
as no effect
:param attr_name: attribute name
:type attr_name: str
"""
if attr_name in self.tango_attr_klasses:
del self.tango_attr_klasses[attr_name]
[docs]
def registerDeviceClass(self, dev_klass_name, dev_klass):
"""Registers a new python class to handle tango devices of the given
tango class name
:param dev_klass_name: tango device class name
:type dev_klass_name: str
:param dev_klass: the new class that will handle devices of the given
tango class name
:type dev_klass: taurus.core.tango.TangoDevice
"""
self.tango_dev_klasses[dev_klass_name] = dev_klass
[docs]
def unregisterDeviceClass(self, dev_klass_name):
"""Unregisters the class for the given tango class name
If no class was registered before for the given attribute, this call
as no effect
:param dev_klass_name: tango device class name
:type dev_klass_name: str
"""
if dev_klass_name in self.tango_dev_klasses:
del self.tango_dev_klasses[dev_klass_name]
[docs]
def getDatabase(self, name=None):
"""Deprecated. Use getAuthority instead"""
return self.getAuthority(name=name)
[docs]
def getAuthority(self, name=None):
"""
Obtain the object corresponding to the given database name or the
default database if name is None.
If the corresponding authority object already exists, the existing
instance is returned. Otherwise a new instance is stored and returned.
:param name: database name string alias. If None, the default database
is used
:type name: str
:return: database object :raise:
(taurus.core.taurusexception.TaurusException) if the given alias is
invalid.
:rtype: taurus.core.tangodatabase.TangoAuthority
"""
ret = None
if name is None:
if self.dft_db is None:
try:
if self._default_tango_host is None:
self.dft_db = _Authority()
else:
name = self._default_tango_host
validator = _Authority.getNameValidator()
groups = validator.getUriGroups(name)
if groups is None:
raise TaurusException(
"Invalid default Tango authority name %s"
% name
)
self.dft_db = _Authority(
host=groups["host"], port=groups["port"]
)
except Exception:
self.debug("Could not create Authority", exc_info=1)
raise
name = self.dft_db.getFullName()
self.tango_db[name] = self.dft_db
ret = self.dft_db
else:
ret = self.tango_db.get(name)
if ret is not None:
return ret
validator = _Authority.getNameValidator()
groups = validator.getUriGroups(name)
if not validator.isValid(name):
raise TaurusException("Invalid Tango authority name %s" % name)
try:
ret = _Authority(host=groups["host"], port=groups["port"])
except Exception:
self.debug(
"Could not create Authority %s",
groups["authority"],
exc_info=1,
)
if ret is not None:
self.tango_db[name] = ret
return ret
[docs]
def getDevice(self, dev_name, create_if_needed=True, **kw):
"""Obtain the object corresponding to the given tango device name.
If the corresponding device already exists, the existing instance
is returned. Otherwise a new instance is stored and returned.
:param dev_name: tango device name or tango alias for the device. It
must be a valid Tango device URI. If authority is not explicit, the
default Tango Database will be used
:type dev_name: str
:param create_if_needed: If True, the Device is created if it did not
exist previously. If False, it returns None if it did not exist
:type create_if_needed: bool
:return: a device object :raise:
(taurus.core.taurusexception.TaurusException) if the given dev_name
is invalid.
:rtype: taurus.core.tango.TangoDevice
"""
d = self.tango_devs.get(dev_name)
if d is None:
d = self.tango_alias_devs.get(dev_name)
if d is not None:
return d
validator = _Device.getNameValidator()
groups = validator.getUriGroups(dev_name)
if groups is None:
raise TaurusException("Invalid Tango device name '%s'" % dev_name)
full_dev_name, _, _ = validator.getNames(dev_name)
if full_dev_name is None:
raise TaurusException("Cannot find full name of '%s'" % dev_name)
d = self.tango_devs.get(full_dev_name)
if not create_if_needed:
return d
if d is None:
try:
if groups["scheme"] == "tango-nodb":
db = None
else:
authority = full_dev_name.rsplit("/", 3)[0]
db = self.getAuthority(authority)
dev_klass = self._getDeviceClass(
db=db, devname=groups["devname"]
)
kw["storeCallback"] = self._storeDevice
kw["parent"] = db
d = dev_klass(full_dev_name, **kw)
# device objects will register themselves in this factory
# so there is no need to do it here
except DoubleRegistration:
d = self.tango_devs.get(full_dev_name)
except Exception:
self.debug("Error creating device %s", dev_name, exc_info=1)
raise
return d
[docs]
def getAttribute(self, attr_name, create_if_needed=True, **kwargs):
"""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 attr_name: a valid attribute name URI
:type attr_name: str
:param create_if_needed: If True, the Attribute is created if it did
not already exist. If False, None is returned if it did not exist
:type create_if_needed: bool
:return: attribute object :raise:
(taurus.core.taurusexception.TaurusException) if the given alias is
invalid.
:rtype: taurus.core.tangoattribute.TangoAttribute
"""
attr = self.tango_attrs.get(attr_name)
if attr is not None:
return attr
# Simple approach did not work. Lets build a proper device name
validator = _Attribute.getNameValidator()
groups = validator.getUriGroups(attr_name)
if groups is None:
raise TaurusException(
("Invalid Tango attribute name '%s'") % attr_name
)
full_attr_name, _, _ = validator.getNames(attr_name)
if full_attr_name is None:
raise TaurusException("Cannot find full name of '%s'" % attr_name)
attr = self.tango_attrs.get(full_attr_name)
if attr is None:
dev_name = full_attr_name.rsplit("/", 1)[0]
try:
dev = self.getDevice(dev_name)
if dev is not None:
# Do another try in case the Device object created the
# attribute itself. This happens for the 'state' attribute
attr = self.tango_attrs.get(full_attr_name)
if attr is not None:
return attr
try:
attr_klass = self._getAttributeClass(
attr_name=attr_name
)
kwargs["storeCallback"] = self._storeAttribute
if "pollingPeriod" not in kwargs:
kwargs["pollingPeriod"] = (
self.getDefaultPollingPeriod()
)
attr = attr_klass(full_attr_name, dev, **kwargs)
# attribute objects will register themselves in this
# factory so there is no need to do it here
except DoubleRegistration:
attr = self.tango_attrs.get(full_attr_name)
except Exception:
self.debug(
"Error creating attribute %s", attr_name, exc_info=1
)
raise
return attr
[docs]
def getAttributeInfo(self, full_attr_name):
"""Deprecated: Use
:meth:`taurus.core.tango.TangoFactory.getConfiguration` instead.
Obtain attribute information corresponding to the given attribute name.
If the corresponding attribute info already exists, the existing
information is returned. Otherwise a new information instance is stored
and returned.
:param full_attr_name: attribute name in format: <tango device
name>'/'<attribute name>
:type full_attr_name: str
:return: configuration object
:rtype: taurus.core.tango.TangoConfiguration
"""
self.deprecated("Use getConfiguration(full_attr_name) instead")
attr = self.getAttribute(full_attr_name)
return attr
@taurus4_deprecation(alt="getAttribute")
def getConfiguration(self, param):
"""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 param: attribute object or full configuration name
:type param: taurus.core.taurusattribute.TaurusAttribute or str
:return: configuration object
:rtype: taurus.core.tango.TangoAttribute
"""
if isinstance(param, str):
return self.getAttribute(param)
return param
def _getAttributeClass(self, **params):
attr_name = params.get("attr_name")
attr_klass = self.tango_attr_klasses.get(attr_name, _Attribute)
return attr_klass
def _getDeviceClass(self, **kwargs):
db, dev_name = kwargs.get("db"), kwargs.get("devname")
if db is None or dev_name is None or len(self.tango_dev_klasses) == 0:
return _Device
else:
if "/" not in dev_name: # we got an alias... find the devslashname
dev_name = db.getElementFullName(dev_name)
try:
tango_dev_klass = db.get_class_for_device(dev_name)
except PyTango.DevFailed:
# sometimes we can't get the class (e.g. dev_name not defined)
return _Device
return self.tango_dev_klasses.get(tango_dev_klass, _Device)
def _storeDevice(self, dev):
name, alias = dev.getFullName(), dev.getSimpleName()
exists = self.tango_devs.get(name)
if exists is not None:
if exists == dev:
msg = "%s has already been registered before" % name
else:
msg = (
"%s has already been registered with a different object!"
% name
)
self.debug(msg)
raise DoubleRegistration(msg)
self.tango_devs[name] = dev
if alias is not None and len(alias):
self.tango_alias_devs[alias] = dev
def _storeAttribute(self, attr):
name = attr.getFullName()
exists = self.tango_attrs.get(name)
if exists is not None:
if exists == attr:
msg = "%s has already been registered before" % name
else:
msg = (
"%s has already been registered with a different object!"
% name
)
self.debug(msg)
raise DoubleRegistration(msg)
self.tango_attrs[name] = attr
[docs]
def getExistingAttribute(self, attr_name):
"""Deprecated: use getAtribute with create_if_needed=False"""
self.warning(
(
"getExistingAttribute is deprecated. "
+ "Use getDevice with create_if_needed=False"
)
)
return self.getAttribute(attr_name, create_if_needed=False)
[docs]
def getExistingDevice(self, dev_name):
"""Deprecated: use getDevice with create_if_needed=False"""
self.warning(
(
"getExistingDevice is deprecated. "
+ "Use getDevice with create_if_needed=False"
)
)
return self.getDevice(dev_name, create_if_needed=False)
[docs]
def removeExistingDevice(self, dev_or_dev_name):
"""Removes a previously registered device.
:param dev_or_dev_name: device name or device object
:type dev_or_dev_name: str or TangoDevice
"""
if isinstance(dev_or_dev_name, _Device):
dev = dev_or_dev_name
else:
dev = self.getDevice(dev_or_dev_name, create_if_needed=False)
if dev is None:
raise KeyError("Device %s not found" % dev_or_dev_name)
dev.cleanUp()
full_name = dev.getFullName()
if full_name in self.tango_devs:
del self.tango_devs[full_name]
simp_name = dev.getSimpleName()
if simp_name in self.tango_alias_devs:
del self.tango_alias_devs[simp_name]
[docs]
def removeExistingAttribute(self, attr_or_attr_name):
"""Removes a previously registered attribute.
:param attr_or_attr_name: attribute name or attribute object
:type attr_or_attr_name: str or TangoAttribute
"""
if isinstance(attr_or_attr_name, _Attribute):
attr = attr_or_attr_name
else:
attr = self.getExistingAttribute(attr_or_attr_name)
if attr is None:
raise KeyError("Attribute %s not found" % attr_or_attr_name)
attr.cleanUp()
full_name = attr.getFullName()
if full_name in self.tango_attrs:
del self.tango_attrs[full_name]
[docs]
def isPollingEnabled(self):
"""Tells if the local tango polling is enabled
:return: wheter or not the polling is enabled
:rtype: bool
"""
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.items():
timer.stop()
[docs]
def enablePolling(self):
"""Enable the application tango polling"""
if self.isPollingEnabled():
return
for period, timer in self.polling_timers.items():
timer.start()
self._polling_enabled = True
[docs]
def getDatabaseNameValidator(self):
"""Deprecated"""
self.warning(
(
"getDatabaseNameValidator is deprecated."
+ 'Use "Authority" instead of "Database"'
)
)
return self.getAuthorityNameValidator()
[docs]
def getAuthorityNameValidator(self):
"""Return TangoAuthorityNameValidator"""
from . import tangovalidator
return tangovalidator.TangoAuthorityNameValidator()
[docs]
def getDeviceNameValidator(self):
"""Return TangoDeviceNameValidator"""
from . import tangovalidator
return tangovalidator.TangoDeviceNameValidator()
[docs]
def getAttributeNameValidator(self):
"""Return TangoAttributeNameValidator"""
from . import tangovalidator
return tangovalidator.TangoAttributeNameValidator()
[docs]
def setOperationMode(self, mode):
"""Deprecated. setOperationMode(OperationMode mode) -> None
Sets the operation mode for the Tango system.
"""
dep = "setOperationMode"
rel = "Taurus4"
dbg_msg = "Don't use this method"
msg = "%s is deprecated (from %s). %s" % (dep, rel, dbg_msg)
self.deprecated(msg)
[docs]
def getOperationMode(self):
"""Deprecated. Gives the current operation mode."""
dep = "getOperationMode"
rel = "Taurus4"
dbg_msg = "Don't use this method"
msg = "%s is deprecated (from %s). %s" % (dep, rel, dbg_msg)
self.deprecated(msg)
return OperationMode.ONLINE