# -*- coding: utf-8 -*-
# This software and supporting documentation are distributed by
# Institut Federatif de Recherche 49
# CEA/NeuroSpin, Batiment 145,
# 91191 Gif-sur-Yvette cedex
# France
#
# This software is governed by the CeCILL-B license under
# French law and abiding by the rules of distribution of free software.
# You can use, modify and/or redistribute the software under the
# terms of the CeCILL-B license as circulated by CEA, CNRS
# and INRIA at the following URL "http://www.cecill.info".
#
# As a counterpart to the access to the source code and rights to copy,
# modify and redistribute granted by the license, users are provided only
# with a limited warranty and the software's author, the holder of the
# economic rights, and the successive licensors have only limited
# liability.
#
# In this respect, the user's attention is drawn to the risks associated
# with loading, using, modifying and/or developing or reproducing the
# software by the user in light of its specific status of free software,
# that may mean that it is complicated to manipulate, and that also
# therefore means that it is reserved for developers and experienced
# professionals having in-depth computer knowledge. Users are therefore
# encouraged to load and test the software's suitability as regards their
# requirements in conditions enabling the security of their systems and/or
# data to be ensured and, more generally, to use and operate it in the
# same conditions as regards security.
#
# The fact that you are presently reading this means that you have had
# knowledge of the CeCILL-B license and that you accept its terms.
from __future__ import absolute_import
import types
import sys
from soma.translation import translate as _
from soma.sorted_dictionary import SortedDictionary
from soma.notification import Notifier, VariableParametersNotifier, \
ObservableAttributes
from soma.singleton import Singleton
from soma.undefined import Undefined
import copy
import six
#-------------------------------------------------------------------------
class DataType(object):
"""
C{DataType} derived classes manage the behaviour of all attributes defined in
a signature of an object (most often a L{HasSignature} instance). In order to
define a new type of parameter, one must derive from C{DataType} and define
at least the L{checkValue} method. See C{DataType} methods documentation for
more information on how to customize a type.
"""
def __init__(self):
super(DataType, self).__init__()
# Some inherited class forbid attribute modification, therefore __dict__
# is accessed directly
#: Indicate wether the value of this type are mutable (I{i.e.} can be
#: modified) or not. By default, C{mutable} is set to C{True}.For more
#: information about mutable or immutable objects, see
#: U{Python documentation<https://docs.python.org/>}.
self.mutable = True
#: Name of this C{DataType}, by default, it is the class name without the
#: module(s) prefix (I{i.e.} as defined in the C{class} statement).
self.name = self.__class__.__name__.split('.')[-1]
def dataTypeInstance(value):
"""
Static method to create a L{DataType} instance from a user defined value.
The C{value} argument must be one of the following:
- B{A DataType instance}: C{value} is returned untouched.
- B{A class deriving from L{DataType}}: C{value()} is returned.
- B{A class not deriving from DataType}: returns
C{L{ClassDataType}( value )}.
Any other value lead to a C{TypeError} exception.
"""
if isinstance( value, six.class_types ):
if issubclass(value, DataType):
value = value()
else:
value = ClassDataType(value)
elif not isinstance(value, DataType):
raise TypeError(
_('Expected a DataType instance or a class but got %s') % (repr(value),))
return value
dataTypeInstance = staticmethod(dataTypeInstance)
def checkValue(self, value):
"""
Verify that C{value} is valid for this type of attribute. Raises a
C{ValueError} if the value is not valid. Must be overloaded in derived
classes.
@return: the real value that will be used for the attribute. It is
recomended to return C{value} here unless you know what you are doing.
"""
raise RuntimeError(_('checkValue method has not be defined for %s') %
(self.__class__.__name__, ))
def createValue(self):
'''
Create a default value for this type of attribute.
'''
raise RuntimeError(_('createValue method has not be defined for %s') %
(self.__class__.__name__, ))
def _getAttribute(self, hasSignatureObject, name):
"""
This method allow to customize attribute read acces of L{HasSignature}
instances for all attributes declared as a specific C{DataType} in the
signature.
If not overloaded, it returns the value of attribute C{name} of
C{hasSignatureObject}, or raises an C{AttributeError} if this attribute is
not defined.
@param hasSignatureObject: object containing an attribute to read.
@type hasSignatureObject: L{HasSignature} instance
@param name: name of the attribute to read
@type name: C{string}
@return: value of C{getattr( hasSignatureObject, name )}
"""
return super(HasSignature, hasSignatureObject).__getattribute__(name)
def _setAttribute(self, hasSignatureObject, name, value):
"""
This method allow to customize attribute write acces of L{HasSignature}
instances for all attributes declared as a specific C{DataType} in the
signature.
If not overloaded, if calls C{self.L{checkValue}( value )} and set the
attribute with the value returned.
@param hasSignatureObject: object where an attribute must be set.
@type hasSignatureObject: L{HasSignature} instance
@param name: name of the attribute to set.
@type name: C{string}
@param value: value to set to the attribute.
@return: the new value of the attribute
"""
checkValue = getattr(hasSignatureObject, '_check_' + name + '_value',
self.checkValue)
newValue = checkValue(value)
super(HasSignature, hasSignatureObject).__setattr__(name, newValue)
return newValue
def _deleteAttribute(self, hasSignatureObject, name):
"""
This method allow to customize attribute deletion of L{HasSignature}
instances for all attributes declared as a specific C{DataType} in the
signature.
If not overloaded, it raises C{TypeError} because attributes declared in
the signature cannot be deleted.
@param hasSignatureObject: object containing an attribute to delete
@type hasSignatureObject: L{HasSignature} instance
@param name: name of the attribute to delete
@type name: C{string}
"""
raise TypeError(_('Cannot delete signature attribute %s') % (name, ))
def _checkValueError(self, value):
"""
Raises a C{ValueError} with a message saying that C{value} is not a valid
value for this data type. This method is used in several classes
derived from C{DataType}.
"""
raise ValueError(_('%(value)s is not a valid value for data type '
'%(type)s')
% {'type': str(self), 'value': repr(value)})
def _convertError(self, value):
"""
Raises a C{ValueError} with a message saying that C{value} cannot be
converted into this data type. This method is used in several classes
derived from C{DataType}.
"""
raise ValueError(_('cannot convert %(value)s into %(type)s') %
{'type': str(self), 'value': repr(value)})
def convert(self, value, checkValue=None):
"""
Convert a given value in a valid value for this data type. If not
overloaded, it raises a C{TypeError} by calling L{_convertError}.
C{checkValue} if the function used to check valid values, if it
is C{None}, C{self.checkValue} is used.
"""
self._convertError(value)
def convertAttribute(self, hasSignatureObject, name, value=Undefined):
"""
@param hasSignatureObject: object where an attribute must be converted.
@type hasSignatureObject: L{HasSignature} instance
@param name: name of the attribute to convert.
@type name: C{string}
@param value: value to convert (optional).
@return: the converted value
"""
checkValue = getattr(hasSignatureObject, '_check_' + name + '_value',
self.checkValue)
if value is Undefined:
value = getattr(hasSignatureObject, name)
try:
newValue = checkValue(value)
except Exception:
newValue = self.convert(value, checkValue=checkValue)
return newValue
def copy(self):
"""
Return a copy of C{self}. If not overloaded, it creates a new instance by
passing the result of
C{self.L{__getinitkwargs__<DataType.__getinitkwargs__>}()} to the constructor.
"""
args, kwargs = self.__getinitkwargs__()
return self.__class__(*args, **kwargs)
def __getinitkwargs__(self):
"""
Returns all the parameters that have been passed to the constructor to
create this instance. This method must be overloaded by derived classes
that overload C{__init__}. See L{Number} class for an example.
@return: a pair C{( args, kwargs )} where C{args} is a tuple containing the
unnamed parameters and C{kwargs} is a dictionary containing named
parameters.
"""
return (), {}
def __str__(self):
return self.name
def __repr__(self):
args, kwargs = self.__getinitkwargs__()
if args:
strArgs = ', '.join([repr(i) for i in args])
else:
strArgs = ''
if kwargs:
if strArgs:
strArgs += ', '
strArgs += ', '.join([n + '=' + repr(v)
for n, v in six.iteritems(kwargs)])
if strArgs:
return self.name + '( ' + strArgs + ' )'
else:
return self.name + '()'
#-------------------------------------------------------------------------
class ClassDataType(DataType):
"""
This class is used internally to allow the use of classes that do not derive
from L{DataType} in L{Signature} constructor. When such a class (let's call
it C{C}) is used for the type of an attribute, it is replaced by
C{ClassDataType( C )}. This data type only check that only C{C} instances are
used as attribute value.
"""
def __init__(self, cls):
DataType.__init__(self)
#: Store the class used as attribute type in a L{Signature}.
self.cls = cls
def __getinitkwargs__(self):
args, kwargs = DataType.__getinitkwargs__(self)
return (self.cls,), kwargs
def checkValue(self, value):
"""
Check that C{value} is an instance of L{self.cls}.
"""
if not isinstance(value, self.cls):
raise TypeError(_('%(instance)s is not an instance of %(class)s') %
{'instance': str(value),
'class': str(self.cls)
})
return value
def createValue(self):
return self.cls()
#-------------------------------------------------------------------------
class Signature(DataType):
"""
A C{Signature} instance contains type definition for a series of attributes,
it is mainly used to define attributes for L{HasSignature} derived class. It
behave like a L{SortedDictionary} where keys are attribute names and values
are attribute types (I{i.e.} L{DataType} instances). Since the order of the
attribute is user defined, it is not possible to use any syntax or data
structure that use a Python dictionary (because dictionaries keys are sorted
according to their values). Therefore, a signature instance is created by
using a list containing attribute names followed by attribute types.
Example::
from soma.signature.api import HasSignature, Signature
from soma.signature.api import Unicode, Number, Integer
class ZooAnimal( HasSignature ):
signature = Signature(
'identifier', Integer( minimum=0 ),
'race', Unicode,
'name', Unicode,
'gender', Choice( 'male', 'female' ),
)
The first element of a signature is always an attribute named C{signature}
and whose type is the C{Signature} instance (this attribute B{must not} be
defined in the constructor). Hence, a L{HasSignature} instance has always a
C{signature} attribute that can be managed like any other user-defined
attributes (especially for modification notification).
@see: L{HasSignature}, L{soma.signature.api}, L{SortedDictionary}
"""
class Item(ObservableAttributes):
"""
@todo: documentation
"""
def setType(self, dataType):
newValue = DataType.dataTypeInstance(dataType)
self.__dict__['type'] = newValue
def getType(self):
try:
return self.__dict__['type']
except KeyError:
raise AttributeError('type')
type = property(getType, setType, None)
def __init__(self, name, type,
defaultValue=Undefined,
doc=None,
readOnly=False,
visible=True,
collapsed=False,
**kwargs):
"""
@todo: documentation
"""
ObservableAttributes.__init__(self)
#: Name of the attribute
self.name = str(name)
#: Type the attribute
self.type = type
#: Default attribute value
self.defaultValue = defaultValue
#: Documentation of the attribute
if doc:
self.doc = six.text_type(doc)
else:
self.doc = None
#: If set to C{True}, the attribute value cannot be modified
self.readOnly = bool(readOnly)
#: If set to C{False}, the attribute is not visible on graphical interface
self.visible = bool(visible)
#: If set to C{True}, the attribute is reduced on graphical interface (when possible)
self.collapsed = bool(collapsed)
# store other kwargs as attributes
if kwargs:
self._options = list(kwargs.keys())
for k, v in six.iteritems(kwargs):
setattr(self, k, v)
def copy(self):
args, kwargs = self.__getinitkwargs__()
return self.__class__(*args, **kwargs)
def __getinitkwargs__(self):
d = {}
if self.defaultValue is not Undefined:
d['defaultValue'] = self.defaultValue
if self.doc:
d['doc'] = self.doc
if self.readOnly:
d['readOnly'] = self.readOnly
if not self.visible:
d['visible'] = self.visible
if self.collapsed:
d['collapsed'] = self.collapsed
return (self.name, self.type), d
if hasattr(self, '_options'):
d['_options'] = self._options
for k in self._options:
d[k] = getattr(self, k)
_msgCannotModify = 'Cannot modify read-only signature'
def __init__(self, *args):
super(Signature, self).__init__()
# Notifier is created on read-only Signature because listeners registration
# can be done on Signature and copied when Signature is changed to
# VariableSignature
self.on_change = Notifier(1)
self.__dict__[ '_signature_data' ] = \
SortedDictionary(('signature', self.Item('signature', self)))
i = 0
while i < len(args):
name = args[i]
i += 1
if isinstance(name, Signature):
for n, item in six.iteritems(name):
if n == 'signature':
continue
self._insertElement(len(self), n, item.copy())
else:
dataType = args[i]
if i + 1 < len(args) and isinstance(args[i + 1], dict):
i += 1
options = args[i]
else:
options = {}
self._insertElement(len(self), name, dataType, options)
i += 1
def _copyItem(self, item):
args, kwargs = item.__getinitkwargs__()
return self.Item(*args, **kwargs)
def _insertElement(self, index, name, dataType, options={}):
if name in self:
raise KeyError(_('Element "%s" is defined twice' % (name, )))
if isinstance(dataType, Signature.Item):
item = self._copyItem(dataType)
else:
dataType = DataType.dataTypeInstance(dataType)
item = self.Item(name, dataType, **options)
self._signature_data.insert(index, name, item)
def has_key(self, key):
"""
see L{SortedDictionary.has_key}
"""
return key in self.__dict__['_signature_data']
def __contains__(self, key):
return key in self.__dict__['_signature_data']
def __getitem__(self, key):
"""
see L{SortedDictionary.__getitem__}
"""
return self.__dict__['_signature_data'][key]
def __setitem__(self, key, value):
"""
see L{SortedDictionary.__setitem__}
"""
raise RuntimeError(_(self._msgCannotModify))
def __delitem__(self, key):
"""
see L{SortedDictionary.__delitem__}
"""
raise RuntimeError(_(self._msgCannotModify))
def __len__(self):
"""
see L{SortedDictionary.__len__}
"""
return len(self.__dict__['_signature_data'])
def get(self, key, default=None):
"""
see L{SortedDictionary.get}
"""
return self.__dict__['_signature_data'].get(key, default)
def keys(self):
"""
see L{SortedDictionary.keys}
"""
return list(self.__dict__['_signature_data'].keys())
def values(self):
"""
see L{SortedDictionary.values}
"""
return list(self.__dict__['_signature_data'].values())
def items(self):
"""
see L{SortedDictionary.items}
"""
return list(self.__dict__['_signature_data'].items())
def __iter__(self):
"""
see L{SortedDictionary.__iter__}
"""
return iter(self.__dict__['_signature_data'])
def iterkeys(self):
"""
see L{SortedDictionary.iterkeys}
"""
return six.iterkeys(self.__dict__['_signature_data'])
def itervalues(self):
"""
see L{SortedDictionary.itervalues}
"""
return six.itervalues(self.__dict__['_signature_data'])
def iteritems(self):
"""
see L{SortedDictionary.iteritems}
"""
return six.iteritems(self.__dict__['_signature_data'])
def insert(self, index, key, value, **kwargs):
"""
see L{SortedDictionary.insert}
"""
raise RuntimeError(_(self._msgCannotModify))
def _setAttribute(self, hasSignatureObject, name, value):
"""
This method is called when the C{signature} attribute of an L{HasSignature}
instance is modified, it calls
C{hasSignatureObject.L{_changeSignature<HasSignature._changeSignature>}} to
change the attribute value. The new signature becomes an instance signature
(overriding the class signature). Once an instance signature is created,
the class signature can be restored by deleting the C{signature} attribute.
@see: L{_deleteAttribute}
"""
checkValue = getattr(hasSignatureObject, '_check_' + name + '_value',
self.checkValue)
newValue = checkValue(value)
hasSignatureObject._changeSignature(newValue)
return newValue
def _deleteAttribute(self, hasSignatureObject, name):
"""
This method is called when the C{signature} attribute of an L{HasSignature}
instance is deleted. It tries to restore the class signature if an instance
signature has been defined. If no instance signature has been defined, it
raises a C{RuntimeError}.
@see: L{_setAttribute}
"""
if not hasSignatureObject._changeSignature(None):
raise RuntimeError(_('Signature cannot be deleted'))
def checkValue(self, value):
"""
Raises a C{TypeError} if C{value} is not a C{Signature} instance.
"""
if not isinstance(value, Signature):
raise TypeError(_('invalid value type for signature attribute: %s') %
(type(value), ))
return value
def __getinitkwargs__(self):
args, kwargs = DataType.__getinitkwargs__(self)
args = []
it = six.iteritems(self)
next(it) # skip signature
for name, type in it:
args.append(name)
args.append(type)
return args, kwargs
def copy(self):
args, kwargs = DataType.__getinitkwargs__(self)
args = []
it = six.iteritems(self)
next(it) # skip signature
for name, item in it:
args.append(name)
args.append(item.copy())
return self.__class__(*args, **kwargs)
def __repr__(self):
result = self.name + '( '
it = six.itervalues(self)
next(it)
for item in it:
result += repr(item.name) + ', ' + repr(item.type) + ', '
args, kwargs = item.__getinitkwargs__()
if kwargs:
result += 'dict( ' + ', '.join(
[str(i) + '=' + repr(j) for i, j in six.iteritems(kwargs)]) + '), '
return result + ' )'
#-------------------------------------------------------------------------
class VariableSignature(Signature):
"""
This class is a non constant version of C{Signature}. You can add or remove
elements at run-time on a C{VariableSignature} instance. These changes are
taken into account by the notification system.
"""
def __setitem__(self, key, value):
if key == 'signature':
raise KeyError(
_('Attribute named "signature" cannot be redefined.'))
self._insertElement(len(self), key, value)
self.on_change.notify(key)
def __delitem__(self, key):
if key == 'signature':
raise KeyError(
_('Attribute named "signature" cannot be deleted.'))
del self.__dict__['_signature_data'][key]
self.on_change.notify(key)
def insert(self, index, key, value, **kwargs):
self._insertElement(index, key, value, kwargs)
self.on_change.notify(key)
def append(self, key, value, **kwargs):
self._insertElement(len(self), key, value, kwargs)
self.on_change.notify(key)
#-------------------------------------------------------------------------
class MetaHasSignature(type):
def __init__(cls, name, bases, dict):
signature = dict.get('signature')
if signature is not None:
for c in bases:
referenceSignature = getattr(c, 'signature', None)
if referenceSignature is not None:
break
if referenceSignature is not None and not isinstance(signature, referenceSignature.__class__):
raise TypeError(_('signature must be an instance of %s') %
(six.text_type(referenceSignature.__class__), ))
super(MetaHasSignature, cls).__init__(name, bases, dict)
#-------------------------------------------------------------------------
class HasSignature(six.with_metaclass(MetaHasSignature, ObservableAttributes)):
"""
HasSignature is the base class for all object using the signature system.
"""
#: The default empty signature
signature = Signature()
def __init__(self, *args, **kwargs):
super(HasSignature, self).__init__(*args, **kwargs)
for attribute in self.signature:
self._onAttributeChange[
attribute] = self._createAttributeNotifier()
def initializeSignatureAttributes(self, **attributesValues):
"""
Initialize signature attributes according to the values given in
parameters and the default values given in the signature. Values
given in parameters takes precedence over signature default values.
If a parameter name is not in the signature, a TypeError is raised.
Example
=======
L{initializeSignatureAttributes} is suitable for using in constructors::
class Time( HasSignature ):
signature = Signature(
'hour', Integer( minimum=0, maximum=23 ), dict( defaultValue=0 ),
'minute', Integer( minimum=0, maximum=59 ), dict( defaultValue=0 ),
'second', Integer( minimum=0, maximum=59 à, dict( defaultValue=0 ),
)
def __init__( self, **kwargs ):
HasSignature.__init__( self )
self.initializeSignatureAttributes( **kwargs )
"""
# Copy the dictionary to allow modification
it = six.iteritems(self.signature)
next(it)
for attributeName, signatureItem in it:
value = attributesValues.pop(attributeName, Undefined)
if value is Undefined:
value = copy.copy(signatureItem.defaultValue)
if value is not Undefined:
setattr(self, attributeName, value)
if attributesValues:
raise TypeError(_("Attribute '%s' is not in the signature") %
(next(six.iterkeys(attributesValues)), ))
def __getattribute__(self, name):
"""
For all attributes that are declared in the signature, calls the
L{_getAttribute<DataType._getAttribute>} method of the attribute type.
Directly return the value of all other attributes.
"""
signatureItem = \
super(HasSignature, self).__getattribute__('signature').get(name)
if signatureItem is None:
return super(HasSignature, self).__getattribute__(name)
else:
try:
return signatureItem.type._getAttribute(self, name)
except AttributeError:
defaultValue = signatureItem.defaultValue
if defaultValue is Undefined:
raise
defaultValue = copy.copy(defaultValue)
super(ObservableAttributes, self).__setattr__(
name, defaultValue)
return defaultValue
def __setattr__(self, name, value):
"""
For all attributes that are declared in the signature, calls the
L{_setAttribute<DataType._setAttribute>} method of the attribute type
and then calls L{self.notifyAttributeChange}. Directly change the value of
all other attributes.
"""
signatureItem = self.signature.get(name)
if signatureItem is None:
super(HasSignature, self).__setattr__(name, value)
else:
signatureItem.type._setAttribute(self, name, value)
def __delattr__(self, name):
"""
For all attributes that are declared in the signature, calls the
L{_deleteAttribute<DataType._deleteAttribute>} method of the attribute type
and then calls L{self.notifyAttributeChange}. Directly change the value of
all other attributes.
"""
signatureItem = self.signature.get(name)
if signatureItem is None:
super(HasSignature, self).__delattr__(name)
else:
signatureItem.type._deleteAttribute(self, name)
def _changeSignature(self, newSignature):
"""
Change or delete the instance signature. C{newSignature} becomes the
instance signature. If C{newSignature} is C{None}, the instance signature
is deleted and the class signature is restored. Notification for attributes
that are in both old and new signature is preserved.
"""
signatureChanged = True
oldSignature = self.signature
if newSignature is None:
try:
super(HasSignature, self).__delattr__('signature')
newSignature = self.signature
except AttributeError:
signatureChanged = False
else:
super(HasSignature, self).__setattr__('signature', newSignature)
if signatureChanged:
oldSignature.on_change.remove(self._signatureChanged)
newSignature.on_change.add(self._signatureChanged)
for attribute in newSignature:
notifier = self._onAttributeChange.get(attribute)
if notifier is None:
self._onAttributeChange[
attribute] = self._createAttributeNotifier()
for attribute in oldSignature:
if not attribute in newSignature:
del self._onAttributeChange[attribute]
return signatureChanged
def _signatureChanged(self, attributeName):
"""
This method is called whenever the signature content is modified.
"""
if attributeName in self._onAttributeChange:
if attributeName not in self.signature:
del self._onAttributeChange[attributeName]
elif attributeName in self.signature:
self._onAttributeChange[
attributeName] = self._createAttributeNotifier()
self._onAttributeChange['signature'].notify(self, 'signature',
self.signature,
self.signature)
def delayAttributeNotification(self, ignoreDoubles=False):
super(HasSignature, self).delayAttributeNotification(ignoreDoubles)
for item in six.itervalues(self.signature):
item.delayAttributeNotification(ignoreDoubles)
def restartAttributeNotification(self):
super(HasSignature, self).restartAttributeNotification()
for item in six.itervalues(self.signature):
item.restartAttributeNotification()
