#!/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 taurus name validator classes"""
__all__ = ["AbstractTangoValidator", "DatabaseNameValidator",
"DeviceNameValidator", "AttributeNameValidator",
"ConfigurationNameValidator"]
__docformat__ = "restructuredtext"
import re
from .taurusbasetypes import MatchLevel
from .util.singleton import Singleton
InvalidAlias = "nada"
[docs]class AbstractTangoValidator:
complete_name = None
normal_name = None
short_name = None
uri_gen_delims = "\:\/\?\#\[\]\@"
# theoreticaly sub_delims should include '+' but we are more permissive here in tango
#uri_sub_delims = "\!\$\&\'\(\)\*\+\,\;\="
uri_sub_delims = "\!\$\&\'\(\)\*\,\;\="
uri_reserved = uri_gen_delims + uri_sub_delims
tango_word = '[^' + uri_reserved + ']+'
protocol_prefix = 'tango://'
def __init__(self):
self.complete_re = re.compile("^%s$" % self.complete_name)
self.normal_re = re.compile("^%s$" % self.normal_name)
self.short_re = re.compile("^%s$" % self.short_name)
def __getMatch(self,str):
return self.complete_re.match(str) or self.normal_re.match(str) or self.short_re.match(str)
[docs] def isValid(self,str, matchLevel = MatchLevel.ANY):
if matchLevel == MatchLevel.ANY:
return not self.__getMatch(str) is None
elif matchLevel == MatchLevel.SHORT:
return not self.short_re.match(str) is None
elif matchLevel == MatchLevel.NORMAL:
return not self.normal_re.match(str) is None
elif matchLevel == MatchLevel.COMPLETE:
return not self.complete_re.match(str) is None
elif matchLevel == MatchLevel.SHORT_NORMAL:
return self.isValid(str,MatchLevel.SHORT) or \
self.isValid(str,MatchLevel.NORMAL)
elif matchLevel == MatchLevel.NORMAL_COMPLETE:
return self.isValid(str,MatchLevel.NORMAL) or \
self.isValid(str,MatchLevel.COMPLETE)
return False
[docs] def getParams(self,str):
m = self.__getMatch(str)
if m is None:
return None
return m.groupdict()
[docs] def getNames(self, str, factory=None):
"""Returns a tuple of three elements with (complete_name, normal_name, short_name)
or None if no match is found"""
return None
[docs]class DatabaseNameValidator(Singleton, AbstractTangoValidator):
protocol_prefix = '((?P<scheme>tango)://)?'
db = '(?P<host>([\w\-_]+\.)*[\w\-_]+):(?P<port>\d{1,5})'
# for tango://host:port
complete_name = '(' + protocol_prefix + ')?' + db
# for a/b/c
normal_name = db
# for devalias
short_name = db
def __init__(self):
pass
[docs] def init(self, *args, **kwargs):
"""Singleton instance initialization."""
AbstractTangoValidator.__init__(self)
[docs] def getNames(self, str, factory=None):
elems = self.getParams(str)
if elems is None:
return str, None, None
host = elems.get('host')
port = elems.get('port')
if host is None or port is None or len(host) == 0 or len(port) == 0:
return None
return 3*('%s:%s' % (host,port),)
class DatabaseQueryValidator(Singleton, AbstractTangoValidator):
"""Deprecated"""
query = '\?query=(?P<query>[\w\-_]+)(?P<params>(\?param=[\w\*\?\%\-_]+)*)'
complete_name = DatabaseNameValidator.complete_name + query
normal_name = DatabaseNameValidator.normal_name + query
short_name = query
def __init__(self):
pass
def init(self, *args, **kwargs):
"""Singleton instance initialization."""
AbstractTangoValidator.__init__(self)
def getNames(self, str, factory=None):
elems = self.getParams(str)
if elems is None:
return str,None
query = elems.get('query')
params = elems.get('params')
short = normal = query
if params:
normal += str(params.split('?param=')[1:])
return str, normal, short
[docs]class DeviceNameValidator(Singleton, AbstractTangoValidator):
w = AbstractTangoValidator.tango_word
dev = '(?P<devicename>' + w + '/' + w + '/' + w + ')'
# for tango://host:port/a/b/c/attrname or host:port/a/b/c
complete_name = DatabaseNameValidator.complete_name + '/' + dev
# for a/b/c
normal_name = DatabaseNameValidator.protocol_prefix + dev
# for devalias
short_name = '(?P<devalias>'+ w + ')'
def __init__(self):
pass
[docs] def init(self, *args, **kwargs):
"""Singleton instance initialization."""
AbstractTangoValidator.__init__(self)
[docs] def getNames(self, str, factory=None):
elems = self.getParams(str)
if elems is None:
return str,None
dev_name = elems.get('devicename')
alias = elems.get('devalias')
host = elems.get('host')
port = elems.get('port')
if factory is None:
return dev_name or '', dev_name or '', alias or ''
db = None
try:
if host and port:
db = factory.getDatabase("%s:%s" % (host,port))
#db = PyTango.Database(host,int(port))
else:
#db = PyTango.Database()
db = factory.getDatabase()
except:
return dev_name or '', dev_name or '', alias or ''
if dev_name:
alias = db.getElementAlias(dev_name) or dev_name
else:
dev_name = db.getElementFullName(alias)
complete = "%s:%s/%s" % (host,port,dev_name)
return complete, dev_name, alias
class DeviceQueryValidator(Singleton, AbstractTangoValidator):
"""Deprecated"""
query = DatabaseQueryValidator.query
complete_name = DeviceNameValidator.complete_name + query
normal_name = DeviceNameValidator.normal_name + query
short_name = DeviceNameValidator.short_name + query
def __init__(self):
pass
def init(self, *args, **kwargs):
"""Singleton instance initialization."""
AbstractTangoValidator.__init__(self)
def getNames(self,str, factory=None):
elems = self.getParams(str)
if elems is None:
return str,None
query = elems.get('query')
params = elems.get('params')
short = query
if params:
short += str(params.split('?param=')[1:])
return str,short
[docs]class AttributeNameValidator(Singleton, AbstractTangoValidator):
w = AbstractTangoValidator.tango_word
attr = '/(?P<attributename>' + w + ')'
# for tango://host:port/a/b/c/attributename or host:port/a/b/c/attributename
complete_name = DeviceNameValidator.complete_name + attr
# for a/b/c/attributename
normal_name = DeviceNameValidator.normal_name + attr
# for devalias/attributename
short_name = DeviceNameValidator.short_name + attr
def __init__(self):
""" Initialization. Nothing to be done here for now."""
pass
[docs] def init(self, *args, **kwargs):
"""Singleton instance initialization."""
AbstractTangoValidator.__init__(self)
[docs] def getNames(self, str, factory=None):
"""Returns the complete and short names"""
elems = self.getParams(str)
if elems is None:
return None
dev_name = elems.get('devicename')
attr_name = elems.get('attributename')
if dev_name:
normal_name = dev_name + "/" + attr_name
else:
normal_name = attr_name
return str, normal_name, attr_name
[docs]class ConfigurationNameValidator(Singleton, AbstractTangoValidator):
w = AbstractTangoValidator.tango_word
conf = "\?(?i)configuration(=(?P<configparam>" + w + "))*"
# for tango://host:port/a/b/c/attrname or host:port/a/b/c/attrname
complete_name = AttributeNameValidator.complete_name + conf
# for a/b/c/attrname
normal_name = AttributeNameValidator.normal_name + conf
# for devalias/attrname
short_name = AttributeNameValidator.short_name + conf
def __init__(self):
""" Initialization. Nothing to be done here for now."""
[docs] def init(self, *args, **kwargs):
"""Singleton instance initialization."""
AbstractTangoValidator.__init__(self)
[docs] def getNames(self, str, factory=None):
"""Returns the complete and short names"""
elems = self.getParams(str)
if elems is None:
return None
dev_name = elems.get('devicename')
attr_name = elems.get('attributename')
simple = elems.get('configparam') or 'configuration'
if dev_name:
normal = dev_name + "/" + attr_name + "?configuration=" + simple
elif attr_name:
normal = attr_name + "?configuration=" + simple
else:
normal = simple
ret = str,normal,simple
return ret