# -*- 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. import types 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 #------------------------------------------------------------------------------- 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}. 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, type ) or type( value ) is types.ClassType: 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: 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__}()} to the constructor. """ args, kwargs = self.__getinitkwargs__() return apply( 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 kwargs.iteritems()] ) 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 no 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 ): """ @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 = unicode( 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 ) 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 _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 name.iteritems(): 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 self.has_key( name ): 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 self.__dict__[ '_signature_data' ].has_key( key ) 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 self.__dict__[ '_signature_data' ].keys() def values( self ) : """ see L{SortedDictionary.values} """ return self.__dict__[ '_signature_data' ].values() def items( self ) : """ see L{SortedDictionary.items} """ return 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 self.__dict__[ '_signature_data' ].iterkeys() def itervalues( self ) : """ see L{SortedDictionary.itervalues} """ return self.__dict__[ '_signature_data' ].itervalues() def iteritems( self ) : """ see L{SortedDictionary.iteritems} """ return self.__dict__[ '_signature_data' ].iteritems() 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}} 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 = self.iteritems() it.next() # 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 = self.iteritems() it.next() # skip signature for name, item in it: args.append( name ) args.append( item.copy() ) return apply( self.__class__, args, kwargs ) def __repr__( self ): result = self.name + '( ' it = self.itervalues() it.next() 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 kwargs.iteritems()] ) + '), ' 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' ) % ( unicode( referenceSignature.__class__ ), ) ) super( MetaHasSignature, cls ).__init__( name, bases, dict ) #------------------------------------------------------------------------------- class HasSignature( ObservableAttributes ): """ HasSignature is the base class for all object using the signature system. """ __metaclass__ = MetaHasSignature #: 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 = self.signature.iteritems() it.next() 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" ) % \ ( attributesValues.iterkeys().next(), ) ) def __getattribute__( self, name ): """ For all attributes that are declared in the signature, calls the L{_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} 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} 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 newSignature.has_key( attribute ): del self._onAttributeChange[ attribute ] return signatureChanged def _signatureChanged( self, attributeName ): """ This method is called whenever the signature content is modified. """ if self._onAttributeChange.has_key( attributeName ): if not self.signature.has_key( attributeName ): del self._onAttributeChange[ attributeName ] elif self.signature.has_key( attributeName ): 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 self.signature.itervalues(): item.delayAttributeNotification( ignoreDoubles ) def restartAttributeNotification( self ): super( HasSignature, self ).restartAttributeNotification() for item in self.signature.itervalues(): item.restartAttributeNotification()