#!/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/>.
##
#############################################################################
"""a list of helper methods"""
__all__ = ['check_dependencies', 'log_dependencies', 'getSchemeFromName',
'getValidTypesForName', 'isValidName', 'makeSchemeExplicit',
'Manager', 'Factory', 'Device', 'Attribute', 'Configuration',
'Database', 'Authority', 'Object', 'Logger',
'Critical', 'Error', 'Warning', 'Info', 'Debug', 'Trace',
'setLogLevel', 'setLogFormat', 'getLogLevel', 'getLogFormat',
'resetLogLevel', 'resetLogFormat',
'enableLogOutput', 'disableLogOutput',
'log', 'trace', 'debug', 'info', 'warning', 'error', 'fatal',
'critical', 'changeDefaultPollingPeriod']
__docformat__ = "restructuredtext"
import sys
import re
from taurus import tauruscustomsettings
from .util.log import taurus4_deprecation
# regexp for finding the scheme
__SCHEME_RE = re.compile(r'([^:/?#]+):.*')
def __translate_version_str2int(version_str):
"""Translates a version string in format x[.y[.z[...]]] into a 000000 number"""
import math
parts = version_str.split('.')
i, v, l = 0, 0, len(parts)
if not l:
return v
while i < 3:
try:
v += int(parts[i]) * int(math.pow(10, (2 - i) * 2))
l -= 1
i += 1
except ValueError, ve:
return v
if not l:
return v
return v
try:
v += 10000 * int(parts[0])
l -= 1
except ValueError, ve:
return v
if not l:
return v
try:
v += 100 * int(parts[1])
l -= 1
except ValueError, ve:
return v
if not l:
return v
try:
v += int(parts[0])
l -= 1
except ValueError:
return v
if not l:
return v
def __get_python_version():
return '.'.join(map(str, sys.version_info[:3]))
def __get_python_version_number():
pyver_str = __get_python_version()
if pyver_str is None:
return None
return __translate_version_str2int(pyver_str)
def __get_pytango_version():
try:
import PyTango
except:
return None
try:
return PyTango.Release.version
except:
return '0.0.0'
def __get_pytango_version_number():
tgver_str = __get_pytango_version()
if tgver_str is None:
return None
return __translate_version_str2int(tgver_str)
def __get_pyqt_version():
try:
import PyQt4.Qt
return PyQt4.Qt.PYQT_VERSION_STR
except:
return None
def __get_pyqt_version_number():
pyqtver_str = __get_pyqt_version()
if pyqtver_str is None:
return None
return __translate_version_str2int(pyqtver_str)
def __get_pyqwt_version():
try:
import PyQt4.Qwt5
return PyQt4.Qwt5.QWT_VERSION_STR
except:
pass
def __get_pyqwt_version_number():
pyqwtver_str = __get_pyqwt_version()
if pyqwtver_str is None:
return None
return __translate_version_str2int(pyqwtver_str)
def __get_qub_version():
try:
import Qub4
return "1.0.0"
except:
try:
import Qub
return "1.0.0"
except:
pass
def __get_qub_version_number():
qubver_str = __get_qub_version()
if qubver_str is None:
return None
return __translate_version_str2int(qubver_str)
def __get_qtcontrols_version():
try:
import qtcontrols
return "1.0.0"
except:
pass
def __get_qtcontrols_version_number():
qtcontrols_str = __get_qtcontrols_version()
if qtcontrols_str is None:
return None
return __translate_version_str2int(qtcontrols_str)
def __get_spyder_version():
try:
import spyder
return spyder.__version__
except:
pass
def __get_spyder_version_number():
spyderver_str = __get_spyder_version()
if spyderver_str is None:
return None
return __translate_version_str2int(spyderver_str)
def __w(msg):
sys.stdout.write(msg)
sys.stdout.flush()
def __wn(msg):
__w(msg + '\n')
[docs]def check_dependencies():
for msg in _check_dependencies(forlog=False):
m = msg[1]
if msg[0] != -1:
m = '\t%s' % m
print m
[docs]def log_dependencies():
from taurus.core.util.log import Logger
l = Logger("taurus")
for msg in _check_dependencies(forlog=True):
if msg[0] != -1:
l.info(msg[1])
def _check_dependencies(forlog=False):
"""Checks for the required and optional packages of taurus"""
# TODO: Checking dependencies should be taken care by setuptools. Remove
if forlog:
MSG = {'OK': '[OK]', 'ERR': '[ERROR]', 'WARN': '[WARNING]'}
else:
MSG = {
'OK': "[\033[0;32mOK\033[0m]",
'ERR': "[\033[0;31mERROR\033[0m]",
'WARN': "[\033[0;33mWARNING\033[0m]"}
core_requirements = {
# module minimum recommended
"Python": ("2.6.0", "2.6.0"),
}
core_optional_requirements = {
# module minimum recommended
"PyTango": ("7.1.0", "7.1.0"),
}
widget_requirements = {
# module minimum recommended
"PyTango": ("7.1.0", "7.1.0"),
"PyQt": ("4.4.0", "4.4.0"),
"PyQwt": ("5.2.0", "5.2.0"),
}
widget_optional_requirements = {
# module minimum recommended
"Qub": ("1.0.0", "1.0.0"),
"qtcontrols": ("1.0.0", "1.0.0"),
"spyder": ("3.0.0", "3.0.0"),
}
yield -1, "Checking required dependencies of taurus.core..."
r = core_requirements
m = "Checking for Python >=%s..." % r["Python"][0]
minPython, recPython = map(__translate_version_str2int, r["Python"])
currPython, currPythonStr = __get_python_version_number(), __get_python_version()
if currPython is None:
yield 2, "{msg} {ERR} (Not found])".format(msg=m, **MSG)
elif currPython < minPython:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currPythonStr, rec=r['Python'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currPythonStr, **MSG)
yield -1, "Checking OPTIONAL dependencies of taurus.core..."
r = core_optional_requirements
m = "Checking for PyTango >=%s..." % r["PyTango"][0]
minPyTango, recPyTango = map(__translate_version_str2int, r["PyTango"])
currPyTango, currPyTangoStr = __get_pytango_version_number(), __get_pytango_version()
if currPyTango is None:
yield 2, "{msg} {ERR} (Not found])".format(msg=m, **MSG)
elif currPyTango < minPyTango:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currPyTangoStr, rec=r['PyTango'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currPyTangoStr, **MSG)
yield -1, "Checking required dependencies of taurus.qt..."
r = widget_requirements
m = "Checking for PyTango >=%s..." % r["PyTango"][0]
if currPyTango is None:
yield 2, "{msg} {ERR} (Not found])".format(msg=m, **MSG)
elif currPyTango < minPyTango:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currPyTangoStr, rec=r['PyTango'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currPyTangoStr, **MSG)
m = "Checking for PyQt >=%s..." % r["PyQt"][0]
minPyQt, recPyQt = map(__translate_version_str2int, r["PyQt"])
currPyQt, currPyQtStr = __get_pyqt_version_number(), __get_pyqt_version()
if currPyQt is None:
yield 2, "{msg} {ERR} (Not found])".format(msg=m, **MSG)
elif currPyQt < minPyQt:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currPyQtStr, rec=r['PyQt'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currPyQtStr, **MSG)
m = "Checking for PyQwt >=%s..." % r["PyQwt"][0]
minPyQwt, recPyQwt = map(__translate_version_str2int, r["PyQwt"])
currPyQwt, currPyQwtStr = __get_pyqwt_version_number(), __get_pyqwt_version()
if currPyQwt is None:
yield 1, "{msg} {ERR} (Not found])".format(msg=m, **MSG)
elif currPyQwt < minPyQwt:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currPyQwtStr, rec=r['PyQwt'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currPyQwtStr, **MSG)
yield -1, "Checking OPTIONAL dependencies of taurus.qt..."
r = widget_optional_requirements
m = "Checking for Qub >=%s..." % r["Qub"][0]
minQub, recQub = map(__translate_version_str2int, r["Qub"])
currQub, currQubStr = __get_qub_version_number(), __get_qub_version()
if currQub is None:
yield 1, "{msg} {WARN} (Not found])".format(msg=m, **MSG)
elif currQub < minQub:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currQubStr, rec=r['Qub'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currQubStr, **MSG)
m = "Checking for spyder >=%s..." % r["spyder"][0]
minspyder, recspyder = map(
__translate_version_str2int, r["spyder"])
currspyder, currspyderStr = __get_spyder_version_number(
), __get_spyder_version()
if currspyder is None:
yield 1, "{msg} {WARN} (Not found])".format(msg=m, **MSG)
elif currspyder < minspyder:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currspyderStr, rec=r['spyder'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currspyderStr, **MSG)
m = "Checking for qtcontrols >=%s..." % r["qtcontrols"][0]
minqtcontrols, recqtcontrols = map(
__translate_version_str2int, r["qtcontrols"])
currqtcontrols, currqtcontrolsStr = __get_qtcontrols_version_number(
), __get_qtcontrols_version()
if currqtcontrols is None:
yield 1, "{msg} {WARN} (Not found])".format(msg=m, **MSG)
elif currqtcontrols < minqtcontrols:
yield 1, "{msg} {WARN} (Found {fnd}. Recommended >={rec})".format(msg=m, fnd=currqtcontrolsStr, rec=r['qtcontrols'][1], **MSG)
else:
yield 0, "{msg} {OK} (Found {fnd})".format(msg=m, fnd=currqtcontrolsStr, **MSG)
[docs]def getSchemeFromName(name, implicit=True):
"""Return the scheme from a taurus name.
:param name: (str) taurus model name URI.
:param implicit: (bool) controls whether to return the default scheme
(if implicit is True -default-) or None (if implicit is
False) in case `model` does not contain the scheme name
explicitly. The default scheme may be defined in
:ref:`tauruscustomsettings` ('tango' is assumed if
not defined)
"""
m = __SCHEME_RE.match(name)
if m is not None:
return m.groups()[0]
if implicit:
return getattr(tauruscustomsettings, 'DEFAULT_SCHEME', "tango")
else:
return None
[docs]def makeSchemeExplicit(name, default=None):
"""return the name guaranteeing that the scheme is present. If name already
contains the scheme, it is returned unchanged.
:param name: (str) taurus model name URI.
:param default: (str) The default scheme to use. If no default is passed,
the one defined in tauruscustomsettings.DEFAULT_SCHEME is
used.
:return: the name with the explicit scheme.
"""
if getSchemeFromName(name, implicit=False) is None:
if default is None:
default = getattr(tauruscustomsettings, 'DEFAULT_SCHEME', "tango")
return "%s:%s" % (default, name)
else:
return name
[docs]def getValidTypesForName(name, strict=None):
"""
Returns a list of all Taurus element types for which `name` is a valid
model name (while in many cases a name may only be valid for one
element type, this is not necessarily true in general)
:param name: (str) taurus model name
:param strict: (bool) If True, names that are not RFC3986-compliant but
which would be accepted for backwards compatibility are
considered valid.
:return: (list<TaurusElementType.element>) where element can be one of:
`Attribute`, `Device` or `Authority`
"""
try:
factory = Factory(scheme=getSchemeFromName(name))
except:
return []
return factory.getValidTypesForName(name, strict=strict)
[docs]def isValidName(name, etypes=None, strict=None):
"""Returns True is the given name is a valid Taurus model name. If
`etypes` is passed, it returns True only if name is valid for at least
one of the given the element types. Otherwise it returns False.
For example::
isValidName('tango:foo')--> True
isValidName('tango:a/b/c', [TaurusElementType.Attribute]) --> False
:param name: (str) the string to be checked for validity
:param etypes: (seq<TaurusElementType>) if given, names will only be
considered valid if they represent one of the given
element types. Supported element types are:
`Attribute`, `Device` and `Authority`
:param strict: (bool) If True, names that are not RFC3986-compliant but
which would be accepted for backwards compatibility are
considered valid.
:return: (bool)
"""
validtypes = getValidTypesForName(name, strict=strict)
if etypes is None:
return bool(validtypes)
for e in etypes:
if e in validtypes:
return True
return False
[docs]def Manager():
"""Returns the one and only TaurusManager
It is a shortcut to::
import taurus.core
manager = taurus.core.taurusmanager.TaurusManager()
:return: the TaurusManager
:rtype: :class:`taurus.core.taurusmanager.TaurusManager`
.. seealso:: :class:`taurus.core.taurusmanager.TaurusManager`
"""
from taurus.core.taurusmanager import TaurusManager
return TaurusManager()
[docs]def Factory(scheme=None):
"""Returns the one and only Factory for the given scheme
It is a shortcut to::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory(scheme)
:param scheme: a string representing the scheme. Default value is None meaning ``tango`` scheme
:type scheme: str
:return: a taurus factory
:rtype: :class:`taurus.core.taurusfactory.TaurusFactory`
"""
manager = Manager()
f = manager.getFactory(scheme=scheme)
if f is None:
from taurus.core.taurusexception import TaurusException
if scheme is None:
scheme = "default scheme '" + manager.default_scheme + "'"
else:
scheme = "'" + scheme + "'"
raise TaurusException('Cannot create Factory for %s' % scheme)
return f()
[docs]def Device(device_name):
"""Returns the taurus device for the given device name
It is a shortcut to::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory()
device = factory.getDevice(device_name)
:param device_name: the device name
:type device_name: str
:return: a taurus device
:rtype: :class:`taurus.core.taurusdevice.TaurusDevice`
"""
return Factory(scheme=getSchemeFromName(device_name)).getDevice(device_name)
[docs]def Attribute(dev_or_attr_name, attr_name=None):
"""Returns the taurus attribute for either the pair (device name, attribute name)
or full attribute name
- Attribute(full_attribute_name)
- Attribute(device_name, attribute_name)
It is a shortcut to::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory()
attribute = factory.getAttribute(full_attribute_name)
or::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory()
device = factory.getDevice(device_name)
attribute = device.getAttribute(attribute_name)
:param dev_or_attr_name: the device name or full attribute name
:type dev_or_attr_name: str or TaurusDevice
:param attr_name: attribute name
:type attr_name: str
:return: a taurus attribute
:rtype: :class:`taurus.core.taurusattribute.TaurusAttribute`
"""
import types
if attr_name is None:
return Factory(scheme=getSchemeFromName(dev_or_attr_name)).getAttribute(dev_or_attr_name)
else:
if type(dev_or_attr_name) in types.StringTypes:
dev = Device(dev_or_attr_name)
else:
dev = dev_or_attr_name
return dev.getAttribute(attr_name)
@taurus4_deprecation(alt='Attribute')
def Configuration(attr_or_conf_name, conf_name=None):
"""Returns the taurus configuration for either the pair
(attribute name, conf name) or full conf name
- Configuration(full_conf_name)
- Configuration(attribute_name, conf_name)
It is a shortcut to::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory()
conf = factory.getConfiguration(attr_or_conf_name)
or::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory()
attribute = factory.getAttribute(attribute_name)
conf = attribute.getConfig(conf_name)
:param attr_or_conf_name: the full attribute name or full conf name
:type attr_or_conf_name: str
:param conf_name: conf name
:type conf_name: str or None
:return: a taurus configuration
:rtype: :class:`taurus.core.taurusconfiguration.TaurusConfiguration`
"""
return Attribute(attr_or_conf_name)
@taurus4_deprecation(alt='Authority')
def Database(name=None):
return Authority(name=name)
[docs]def Authority(name=None):
"""Returns a taurus authority
It is a shortcut to::
import taurus.core.taurusmanager
manager = taurus.core.taurusmanager.TaurusManager()
factory = manager.getFactory()
db = factory.getAuthority(dname)
:param name: authority name. If None (default) it will return the default
authority of the default scheme. For example, if the default
scheme is tango, it will return the default TANGO_HOST database
:type name: str or None
:return: a taurus authority
:rtype: :class:`taurus.core.taurusauthority.TaurusAuthority`
"""
return Factory(getSchemeFromName(name or '')).getAuthority(name)
[docs]def Object(*args):
"""Returns an taurus object of given class for the given name
Can be called as:
- Object(name)
- Object(cls, name)
Where:
- `name` is a model name (str)
- `cls` is a class derived from TaurusModel
If `cls` is not given, Object() will try to guess it from `name`.
:return: a taurus object
:rtype: :class:`taurus.core.taurusmodel.TaurusModel`
"""
if len(args) == 1:
klass, name = None, args[0]
elif len(args) == 2:
klass, name = args
else:
msg = 'Object() takes either 1 or 2 arguments (%i given)' % len(args)
raise TypeError(msg)
factory = Factory(getSchemeFromName(name))
if klass is None:
klass = factory.findObjectClass(name)
return factory.getObject(klass, name)
from taurus.core.util import log as __log_mod
Logger = __log_mod.Logger
Critical = Logger.Critical
Fatal = Logger.Fatal
Error = Logger.Error
Warning = Logger.Warning
Info = Logger.Info
Debug = Logger.Debug
Trace = Logger.Trace
setLogLevel = Logger.setLogLevel
setLogFormat = Logger.setLogFormat
getLogLevel = Logger.getLogLevel
getLogFormat = Logger.getLogFormat
resetLogLevel = Logger.resetLogLevel
resetLogFormat = Logger.resetLogFormat
enableLogOutput = Logger.enableLogOutput
disableLogOutput = Logger.disableLogOutput
log = __log_mod._log
trace = __log_mod.trace
debug = __log_mod.debug
info = __log_mod.info
warning = __log_mod.warning
error = __log_mod.error
fatal = __log_mod.fatal
critical = __log_mod.critical
deprecated = __log_mod.deprecated
[docs]def changeDefaultPollingPeriod(period):
Manager().changeDefaultPollingPeriod(period)
#del __log_mod
#del __translate_version_str2int