#!/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 tango module. " + 'Taurus will not support the "tango" scheme'
debug(msg)
raise
import tango
from taurus import tauruscustomsettings
from taurus.core.taurusbasetypes import (
OperationMode,
TaurusElementType,
TaurusSerializationMode,
)
from taurus.core.taurusexception import DoubleRegistration, TaurusException
from taurus.core.taurusfactory import TaurusFactory
from taurus.core.util.containers import CaselessDict, CaselessWeakValueDict
from taurus.core.util.log import (
Logger,
deprecation_decorator,
taurus4_deprecation,
)
from taurus.core.util.singleton import Singleton
from .tangoattribute import TangoAttribute
from .tangodatabase import TangoAuthority
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._event_subscription_disabled = None
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:
tango.Util.instance(False)
except tango.DevFailed:
try:
tango.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
@deprecation_decorator(rel="5.2.1", alt="set_tango_event_subscription_disabled")
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.set_tango_event_subscription_disabled(not value)
@deprecation_decorator(rel="5.2.1", alt="is_tango_event_subscription_disabled")
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 not self.is_tango_event_subscription_disabled()
[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 (including reuse of a cached
default authority). 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:
validator = _Authority.getNameValidator()
if self._default_tango_host:
name = self._default_tango_host
else:
_th = _Authority.get_default_tango_host()
if _th:
name = f"tango://{_th}"
else:
self.dft_db = _Authority()
return self.dft_db
groups = validator.getUriGroups(name)
if groups is None:
raise TaurusException(
f"Invalid default Tango authority name {name}"
)
_authority_key = f"{groups['scheme']}:{groups['authority']}"
if _authority_key in self.tango_db:
self.dft_db = self.tango_db[_authority_key]
else:
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:
validator = _Authority.getNameValidator()
groups = validator.getUriGroups(name)
if not groups:
raise TaurusException(f"Invalid Tango authority name {name}")
_authority_key = f"{groups['scheme']}:{groups['authority']}"
ret = self.tango_db.get(_authority_key)
if ret is not None:
return ret
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[ret.getFullName()] = 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 tango.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.getAttribute(attr_or_attr_name, create_if_needed=False)
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
[docs]
def set_tango_event_subscription_disabled(self, disable_subscription):
"""Disable subscription to tango events. Needs to be called before
creating Attributes or setting models
"""
self._event_subscription_disabled = disable_subscription
[docs]
def is_tango_event_subscription_disabled(self):
return self._event_subscription_disabled