Source code for soma.notification

# -*- 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.

'''
This module provides a notification system that can be used to register
callbacks (*i.e* Python callables) that will all be called by a single
:meth:`Notifier.notify` call.

* author: Yann Cointepas, Dominique Geffroy
* organization: NeuroSpin
* license: `CeCILL B <http://www.cecill.info/licences/Licence_CeCILL-B_V1-en.html>`_
'''
from __future__ import absolute_import
__docformat__ = "restructuredtext en"

import six
from six.moves import range
import sys

from soma.translation import translate as _
from soma.functiontools import checkParameterCount, numberOfParameterRange
from soma.undefined import Undefined
from soma.sorted_dictionary import SortedDictionary

#-------------------------------------------------------------------------


[docs]class Notifier(object): ''' Register a series of functions (or Notifier instances) which are all called whith the :meth:`notify` method. The calling order is the registering order. If a Notifier is registered, its :meth:`notify` method is called whenever *self.notify()* is called. ''' def __init__(self, parameterCount=None): ''' Parameters ---------- parameterCount: int if not None, each registered function must be callable with that number of arguments (checking is done on registration). ''' self._listeners = [] self._parameterCount = parameterCount self._delayedNotification = None
[docs] def add(self, listener): ''' Register a callable or a Notifier that will be called whenever :meth:`notify` is called. If the notifier has a *parameterCount*, :func:`checkParameterCount <soma.utils.functiontools.checkParameterCount>` is used to verify that *listener* can be called with *parameterCount* parameters. Parameters ---------- listener: Python callable (function, method, *etc*.) or :class:`Notifier` instance item to add to the notification list. ''' if listener not in self._listeners: if self._parameterCount is not None: if isinstance(listener, Notifier): if listener._parameterCount is not None and \ listener._parameterCount != self._parameterCount: raise RuntimeError(_('Impossible to register a notifier with' '%(other)d parameter(s) to a notifier ' 'with %(self)d parameter(s)') % {'self': self._parameterCount, 'other': listener._parameterCount}) else: checkParameterCount(listener, self._parameterCount) self._listeners.append(listener)
[docs] def remove(self, listener): ''' Remove an item from the notification list. Do nothing if *listener* is not in the list. Parameters ---------- listener: Python callable or :class:`Notifier` instance item previously registered with :meth:`add`. Returns ------- bool: *True* if a listener has been removed, *False* otherwise. ''' try: self._listeners.remove(listener) result = True except ValueError: result = False return result
[docs] def notify(self, *args): ''' Calls all the registered items in the notification list. All the parameters given to :meth:`notify` are passed to the items in the list. For items in the list that are :class:`Notifier` instance, their :meth:`notify` method is called .. seealso:: :meth:`delayNotification`, :meth:`restartNotification` ''' if self._delayedNotification is None: # if self._listeners: print '!notify!', self, ':', args, '(' + str(len( self._listeners )), 'listeners)' # Iterate on a copy of self._listeners because this list can be modified # by a listener during notification loop. for listener in tuple(self._listeners): # print '!notify! -->', listener if isinstance(listener, Notifier): listener.notify(*args) else: listener(*args) else: if self._delayedNotificationIgnoreDoubles: if args not in self._delayedNotification: self._delayedNotification.append(args) else: self._delayedNotification.append(args)
[docs] def delayNotification(self, ignoreDoubles=False): ''' Stop notification until :meth:`restartNotification` is called. After a call to :meth`delayNotification`, all calls to :meth:`notify` will only store the notification parameters until :meth:`restartNotification` is called. Parameters ---------- ignoreDoubles: bool If *True* (the default), all calls to :meth:`notify` with the same parameters as a previous call will be ignored (*i.e.* notification will be done only once for two identical calls). ''' self._delayedNotification = [] self._delayedNotificationIgnoreDoubles = ignoreDoubles
[docs] def restartNotification(self): ''' Restart notifications that have been delayed by :meth:`delayNotification`. All the calls to :meth:`notify` that have been done between the call to :meth:`delayNotification` and the call to :meth:`restartNotification`, are applied immediately. ''' delayedNotification = self._delayedNotification if delayedNotification is not None: self._delayedNotification = None for args in delayedNotification: self.notify(*args)
#-------------------------------------------------------------------------
[docs]class ReorderedCall(object): ''' **todo:** documentation ''' def __init__(self, function, parametersOrder): self._function = function self._order = parametersOrder def __call__(self, *args): return self._function(*[args[i] for i in self._order])
#-------------------------------------------------------------------------
[docs]class VariableParametersNotifier(Notifier): ''' This class is a notifier that can register functions with various arguments count. ''' def __init__(self, mainParameters, *otherParameters): ''' .. seealso:: :class:`Notifier` **todo** documentation ''' Notifier.__init__(self, len(mainParameters)) self.__parameters = { len(mainParameters): list(range(len(mainParameters))) } self.__min = len(mainParameters) self.__max = len(mainParameters) # Converting to list because it has the index method (which tuple has # not) mainParameters = list(mainParameters) main = set(mainParameters) for p in otherParameters: other = set(p) if other - main: raise RuntimeError(_('Invalid parameters definition (%(other)s is' ' not a subset of %(main)s)') % {'other': str(other), 'main': str(main)}) if len(p) in self.__parameters: raise RuntimeError(_('Invalid parameters definition (several sets' ' of %d parameters defined)') % (len(p),)) self.__parameters[len(p)] = [ mainParameters.index(i) for i in p] self.__min = min(self.__min, len(mainParameters))
[docs] def add(self, listener): ''' .. seealso:: :meth:`Notifier.add` **todo:** documentation ''' if isinstance(listener, Notifier): realListener = listener else: min, max = numberOfParameterRange(listener) if max is None: paramCount = self.__max else: paramCount = max paramOrder = self.__parameters.get(paramCount) if paramOrder is None: raise RuntimeError(_('%(f)s has an invalid parameter count ' '(%(c)d)') % {'f': str(listener), 'c': paramCount}) realListener = ReorderedCall(listener, paramOrder) Notifier.add(self, realListener)
[docs] def remove(self, listener): ''' .. seealso:: :meth:`Notifier.remove` **todo:** documentation ''' for i in range(len(self._listeners)): c = self._listeners[i] if ( isinstance( c, ReorderedCall ) and c._function == listener ) or \ c == listener: del self._listeners[i] return True return False
#-------------------------------------------------------------------------
[docs]class ObservableAttributes(object): ''' ObservableAttributes allows to track modification of attributes at runtime. By registering callbacks, it is possible to be warn of the modification (setting and deletion) of any attribute of the instance. ''' def __init__(self, *args, **kwargs): #: VariableParametersNotifier instance notified whenever any attribute #: is modified. Use self.onAttributeChange to register a function on #: this notifier. # self.__dict__[ '_onAnyAttributeChange' ] = \ # self._createAttributeNotifier() super(ObservableAttributes, self).__setattr__('_onAnyAttributeChange', self._createAttributeNotifier()) #: Dictionary whose keys are attribute names and values are #: L{VariableParametersNotifier} instances. Whenever an attribute is #: modified, the corresponding L{VariableParametersNotifier} is notified. #: Use L{self.onAttributeChange} to register a function on these notifiers. # self.__dict__[ '_onAttributeChange' ] = {} super(ObservableAttributes, self).__setattr__( '_onAttributeChange', {}) super(ObservableAttributes, self).__init__(*args, **kwargs) @staticmethod def _createAttributeNotifier(): ''' Static method creating a :class:`VariableParametersNotifier` instance for attribute modification notification. see :meth:`notifyAttributeChange`. ''' return VariableParametersNotifier( ('object', 'attributeName', 'newValue', 'oldValue'), (), ('newValue', ), ('newValue', 'oldValue'), ('attributeName', 'newValue', 'oldValue'), )
[docs] def notifyAttributeChange(self, name, value, oldValue=Undefined): ''' First, calls functions registered for modification of the attribute named *name*, then call functions registered for modification of any attribute. .. seealso:: :meth:`onAttributeChange` ''' # Notify the change of this attribute if hasattr(self, '_onAttributeChange'): notifier = self._onAttributeChange.get(name) if notifier is not None: notifier.notify(self, name, value, oldValue) if hasattr(self, '_onAnyAttributeChange'): # Notify the change of one attribute self._onAnyAttributeChange.notify(self, name, value, oldValue)
def __setattr__(self, name, value): ''' Changes the value of attribute *name* then calls :meth:`notifyAttributeChange`. ''' oldValue = getattr(self, name, Undefined) super(ObservableAttributes, self).__setattr__(name, value) if value != oldValue: self.notifyAttributeChange(name, value, oldValue) def __delattr__(self, name): ''' Deletes the attribute *name* then calls :meth:`notifyAttributeChange` with ``newValue = :class:`Undefined` ``. ''' oldValue = getattr(self, name, Undefined) super(ObservableAttributes, self).__delattr__(name) self.notifyAttributeChange(name, Undefined, oldValue) # Delete notifier for the deleted attribute self._onAttributeChange.pop(name, None)
[docs] def onAttributeChange(self, first, second=None): ''' Registers a function to be called when an attribute is modified or deleted. To call the function for any attribute modification, use the following syntax:: instance.onAttributeChange(function) To register a function for a named attribute, use the following syntax:: instance.onAttributeChange(attributeName, function) The registered function can have 0 to 4 parameters. Depending on its number of parameters, it will be called with the following values: - 4 parameters: (object, attributeName, newValue, oldValue) - 3 parameters: (attributeName, newValue, oldValue) - 2 parameters: (newValue, oldValue) - 1 parameter: (newValue) Where: - **object** is the object whose attribute has been modified or deleted. - **attributeName** is the name of the modified or deleted attribute. - **newValue** is the value of the attribute after modification or ``Undefined`` if the attribute has been deleted. - **oldValue** is the value of the attribute before modification. If the attribute was not defined, ``oldValue = Undefined``. If the function accepts a variable number of parameters, it will be called with the maximum number of arguments possible. ''' if second is None: if hasattr(self, '_onAnyAttributeChange'): self._onAnyAttributeChange.add(first) else: if hasattr(self, '_onAttributeChange'): notifier = self._onAttributeChange.get(first) if notifier is None: notifier = self._createAttributeNotifier() self._onAttributeChange[first] = notifier notifier.add(second)
[docs] def removeOnAttributeChange(self, first, second=None): ''' Remove a function previously registered with :meth:`onAttributeChange`. To remove a function, the arguments of :meth:`removeOnAttributeChange` must be the same as those passed to :meth:`onAttributeChange` to register the function. ''' if second is None: result = self._onAnyAttributeChange.remove(first) else: result = self._onAttributeChange[first].remove(second) return result
[docs] def delayAttributeNotification(self, ignoreDoubles=False): ''' Stop attribute modification notification until :meth:`restartAttributeNotification` is called. After a call to :meth:`delayAttributeNotification`, all modification notification will only be stored until :meth:`restartAttributeNotification` is called. This call is recursive on all attributes values that are instance of :class:`ObservableAttributes`. Parameters ---------- ignoreDoubles: bool If True (False is the default), all notification with the same parameters as a previous notification will be ignored (*i.e.* notification will be done only once for two identical events). ''' self._delayAttributeNotification( ignoreDoubles=ignoreDoubles, checkedObjects=set())
def _delayAttributeNotification(self, ignoreDoubles=False, checkedObjects=None): if not checkedObjects == None: checkedObjects.add(self) for name, notifier in six.iteritems(self._onAttributeChange): notifier.delayNotification(ignoreDoubles) self._onAnyAttributeChange.delayNotification(ignoreDoubles) # Recursively delay notification for name in dir(self): value = getattr(self, name, Undefined) if isinstance(value, ObservableAttributes): if not checkedObjects is None: # This allow to not recursively call # _delayAttributeNotification if not value in checkedObjects: value._delayAttributeNotification( ignoreDoubles=ignoreDoubles, checkedObjects=checkedObjects) else: value._delayAttributeNotification( ignoreDoubles=ignoreDoubles) if not checkedObjects == None: checkedObjects.pop()
[docs] def restartAttributeNotification(self): ''' Restarts notifications that have been delayed by :meth:`delayAttributeNotification`. All the modifications that happened between the call to :meth:`delayAttributeNotification` and the call to :meth:`restartAttributeNotification`, are notified immediately. ''' self._restartAttributeNotification(checkedObjects=set())
def _restartAttributeNotification(self, checkedObjects=None): if not checkedObjects == None: checkedObjects.add(self) for name, notifier in six.iteritems(self._onAttributeChange): notifier.restartNotification() self._onAnyAttributeChange.restartNotification() # Recursively restart notification for name in dir(self): value = getattr(self, name, Undefined) if isinstance(value, ObservableAttributes): if not checkedObjects is None: # This allow to not recursively call # _delayAttributeNotification if not value in checkedObjects: value.restartAttributeNotification() else: value._delayAttributeNotification() if not checkedObjects == None: checkedObjects.pop()
#----------------------------------------------------------------------------
[docs]class ObservableList(list): """ A list that notifies its changes to registred listeners. Inherits from python list and contains an instance of :class:`Notifier` (:attr:`onChangeNotifier`). Example:: l = ObservableList() l.addListener(update) l.append(e) * calls :meth:`onChangeNotifier.notify(INSERT_ACTION, [e], len(l)) <Notifier.notify>` * calls ``update(INSERT_ACTION, [e], len(l))`` Attributes ---------- INSERT_ACTION: int used to notify insertion of new elements in the list REMOVE_ACTION: int used to notify elements deletion MODIFY_ACTION: int used to notify elements modification onChangeNotifier: Notifier the Notifier's notify method is called when the list has changed. """ # actions to notify INSERT_ACTION = 0 REMOVE_ACTION = 1 MODIFY_ACTION = 2 def __init__(self, content=None): """ Parameters ---------- content: list elements to initialize the list content """ # call a super class method can be done two different ways: # - superClass.method(self, ...) # - super(superClass, self).method(...) # It's better to use the second way because if derived class inherits from several classes, # super will try to find the method in other super classes before super(ObservableList, self).__init__() # views can register update callbacks on this notifier # to be aware of any change in the model # On change, this object calls Notifier.notify(args) # which calls every registred function with args # args = action (insert, remove, modify), elems list, position self.onChangeNotifier = Notifier() if content: self.extend(content) def __getinitargs__(self): """Returns the args to pass to the __init__ method to construct this object. It is useful to save an :class:`ObservableList` object to :mod:`minf` format. Returns ------- tuple: arg content to pass to the __init__ method for creating a copy of this object """ content = [] content.extend(self) return (content) def __reduce__(self): """This method is redefined for enable deepcopy of this object (and potentially pickle). It gives the arguments to pass to the init method of the object when creating a copy Returns ------- tuple: class name, init args, state, iterator on elements to copy, dictionary iterator """ # class name, init args, parameters for setstate, elements iterator, # dictionary iterator return (self.__class__, (), None, iter(self), None)
[docs] def addListener(self, listener): """Registers the listener callback method in the notifier. The method must take 3 arguments: action, elems list, position *action* should be one of: - INSERT_ACTION: elems have been inserted at position in the list - REMOVE_ACTION: elems have been removed [at position] in the list - MODIFY_ACTION: at position, some elements have been replaced by elems The position given in the notify method will be between 0 and ``len(self)`` Parameters ---------- listener: function function to call to notify changes """ self.onChangeNotifier.add(listener)
[docs] def append(self, elem): """Adds the element at the end of the list. Notifies an insert action. """ index = len(self) super(ObservableList, self).append(elem) self.onChangeNotifier.notify(self.INSERT_ACTION, [elem], index)
[docs] def extend(self, l): """Adds the content of the list l at the end of current list. Notifies an insert action. """ index = len(self) super(ObservableList, self).extend(l) self.onChangeNotifier.notify(self.INSERT_ACTION, l, index)
[docs] def insert(self, pos, elem): """Inserts elem at position pos in the list. Notifies an insert action. """ index = self.getPositiveIndex(pos) super(ObservableList, self).insert(pos, elem) self.onChangeNotifier.notify(self.INSERT_ACTION, [elem], index)
[docs] def remove(self, elem): """Removes the first occurence of elem in the list. Notifies a remove action. """ super(ObservableList, self).remove(elem) self.onChangeNotifier.notify(self.REMOVE_ACTION, [elem])
[docs] def pop(self, pos=None): """Removes the element at position pos or the last element if pos is *None*. Notifies a remove action. Returns ------- object: the removed element """ if pos is not None: index = self.getPositiveIndex(pos) elem = super(ObservableList, self).pop(pos) else: index = len(self) - 1 elem = super(ObservableList, self).pop() self.onChangeNotifier.notify(self.REMOVE_ACTION, [elem], index) return elem
[docs] def sort(self, key=None, reverse=False): """Sorts the list using key function key. Notifies a modify action. Returns ------- key: function key function: elem->key """ super(ObservableList, self).sort(key=key, reverse=reverse) # all the elements of the list could be modified self.onChangeNotifier.notify(self.MODIFY_ACTION, self, 0)
[docs] def reverse(self): """Inverses the order of the list. Notifies a modify action.""" super(ObservableList, self).reverse() self.onChangeNotifier.notify(self.MODIFY_ACTION, self, 0)
def __setitem__(self, key, value): """Sets value to element at position key in the list. Notifies a modify action:: l[key] = value """ index = self.getPositiveIndex(key) super(ObservableList, self).__setitem__(key, value) self.onChangeNotifier.notify(self.MODIFY_ACTION, [value], index) def __delitem__(self, key): """Removes the element at position key in the list. Notifies a remove action:: del l[key] """ index = self.getPositiveIndex(key) super(ObservableList, self).__delitem__(key) self.onChangeNotifier.notify(self.REMOVE_ACTION, [], index) def __setslice__(self, i, j, seq): """Sets values in seq to elements in the interval i,j. If i and j are negative numbers, there are converted before this call in ``index + len(self)`` Notifies a modify action: l[i:j] = seq """ indexI = self.getIndexInRange(i) indexJ = self.getIndexInRange(j) super(ObservableList, self).__setslice__(i, j, seq) # if the interval is empty, action is insertion at the first position if indexI >= indexJ: self.onChangeNotifier.notify(self.INSERT_ACTION, seq, indexI) else: lenSeq = len(seq) lenInter = indexJ - indexI # if values sequence has same or lower length as the interval in the list, # all values are written to the list from the first index # the rest of the interval (if interval is longer than sequence) is # left unchanged if lenInter >= lenSeq: self.onChangeNotifier.notify(self.MODIFY_ACTION, seq, indexI) else: # if the interval is shorter than the sequence of values, # values in the interval are used to modify the list, # the rest is inserted at indexJ position self.onChangeNotifier.notify( self.MODIFY_ACTION, seq[0:lenInter], indexI) self.onChangeNotifier.notify( self.INSERT_ACTION, seq[lenInter:lenSeq], indexJ) def __delslice__(self, i, j): """Removes elements in the interval i,j. If i and j are negative numbers, there are converted before this call in ``index + len(self)``. Notifies a remove action:: del l[i:j] """ indexI = self.getIndexInRange(i) indexJ = self.getIndexInRange(j) seq = self[indexI:indexJ] super(ObservableList, self).__delslice__(i, j) # if the interval is empty, the list is not modified if indexI < indexJ: self.onChangeNotifier.notify(self.REMOVE_ACTION, seq, indexI) def __iadd__(self, l): """``list += l`` <=> ``list.extend(l)`` Notifies insert action.""" index = len(self) newList = super(ObservableList, self).__iadd__(l) self.onChangeNotifier.notify(self.INSERT_ACTION, l, index) return newList def __imul__(self, n): """``list *= n`` Notifies insert action.""" index = len(self) newList = super(ObservableList, self).__imul__(n) self.onChangeNotifier.notify(self.INSERT_ACTION, self[index:], index) return newList
[docs] def getPositiveIndex(self, i): """Returns an index between 0 and ``len(self)`` - if *i* is negative, it is replaced by ``len(self) + i`` - if *index* is again negative, it is replaced by 0 - if *index* is beyond ``len(self)`` it is replaced by ``len(self)`` """ l = len(self) if i < 0: index = max(0, l + i) else: index = min(i, l) return index
[docs] def getIndexInRange(self, i): """Returns an index in the range of the list. * if ``i < 0`` returns 0 * if ``i > len(self)``, returns ``len(self)`` """ if i < 0: index = 0 else: l = len(self) if i > l: index = l else: index = i return index
[docs] def itemIndex(self, item): """ Returns item's index in the list. It is different from the :meth:`list.index` method that returns the index of the first element that has the same value as *item*. """ i = 0 for it in self: if it is item: break i = i + 1 if i == len(self): i = -1 return i
#----------------------------------------------------------------------------
[docs]class ObservableSortedDictionary(SortedDictionary): """ A sorted dictionary that notifies its changes. Inherits from python list and contains an instance of :class:`Notifier` (:attr:`onChangeNotifier`). Example:: d=ObservableSortedDictionary() d.addListener(update) d.insert(index, key, e) * calls :meth:`onChangeNotifier.notify(INSERT_ACTION, [e], index) <Notifier.notify>` * calls ``update(INSERT_ACTION, [e], index)`` Attributes ---------- INSERT_ACTION: int used to notify insertion of new elements in the dictionary REMOVE_ACTION: int used to notify elements deletion MODIFY_ACTION: int used to notify elements modification onChangeNotifier: Notifier the Notifier's notify method is called when the dictionaty has changed. """ # actions to notify INSERT_ACTION = 0 REMOVE_ACTION = 1 MODIFY_ACTION = 2 def __init__(self, *args): ''' Initialize the dictionary with a list of ( key, value ) pairs. ''' self.onChangeNotifier = Notifier() super(ObservableSortedDictionary, self).__init__(*args) def __getinitargs__(self): """Returns the args to pass to the __init__ method to construct this object. It is useful to save ObservableList object to minf format. Returns ------- tuple: arg content to pass to the __init__ method for creating a copy of this object """ content = list(self.items()) return (content)
[docs] def addListener(self, listener): """Registers the listener callback method in the notifier. The method must take 3 arguments: action, elems list, position *action* should be one of: - INSERT_ACTION: elems have been inserted at position in the dictionary - REMOVE_ACTION: elems have been removed [at position] in the dict - MODIFY_ACTION: at position, some elements have been replaced by elems The position given in the notify method will be between 0 and len(self) Parameters ---------- listener: function function to call to notify changes """ self.onChangeNotifier.add(listener)
def __setitem__(self, key, value): insertion = key not in self super(ObservableSortedDictionary, self).__setitem__(key, value) if insertion: self.onChangeNotifier.notify( self.INSERT_ACTION, [value], len(self) - 1) else: self.onChangeNotifier.notify( self.MODIFY_ACTION, [value], self.sortedKeys.index(key)) def __delitem__(self, key): index = self.sortedKeys.index(key) super(ObservableSortedDictionary, self).__delitem__(key) self.onChangeNotifier.notify(self.REMOVE_ACTION, [], index)
[docs] def insert(self, index, key, value): ''' inserts a (*key*, *value*) pair in the sorted dictionary before position *index*. If *key* is already in the dictionary, a :class:`KeyError <exceptions.KeyError>` is raised. Parameters ---------- key: key to insert value: value associated to *key* Returns ------- index: integer index of C{key} in the sorted keys ''' super(ObservableSortedDictionary, self).insert(index, key, value) self.onChangeNotifier.notify(self.INSERT_ACTION, [value], index)
[docs] def clear(self): ''' Removes all items from dictionary ''' super(ObservableSortedDictionary, self).clear() self.onChangeNotifier.notify(self.REMOVE_ACTION, list(self.values()), 0)
[docs] def sort(self, key=None, reverse=False): """Sorts the dictionary using function *key* to compare keys. Notifies a modify action. Parameters ---------- key: function key function key->key """ super(ObservableSortedDictionary, self).sort(key=key, reverse=reverse) self.onChangeNotifier.notify(self.MODIFY_ACTION, list(self.values()), 0)
#----------------------------------------------------------------------------
[docs]class EditableTree(ObservableAttributes, ObservableSortedDictionary): """The base class to model a tree of items. This class can be derived to change implementation. An EditableTree contains items which can be - an item branch: it contains other items as children - an item leaf: doesn't have children The list of items is an :class:`ObservableSortedDictionary` which notifies its changes to registred listeners. If the tree is modifiable, new items can be added. Every item is an :class:`EditableTree.Item`. *EditableTree* is iterable over its items. *EditableTree* also inherits from :class:`ObservableAttributes`, so registred listeners can be notified of attributes value change. To call a method when item list changes:: editableTree.addListener(callbackMethod) To call a method when an item list changes at any depth in the tree:: editableTree.addListenerRec(callbackMethod) To call a method when an attribute of the tree changes:: editableTree.onAttributeChange(attributeName, callbackMethod) To call a method when an attribute changes at any depth in the tree:: editableTree.onAttributeChangeRec(attributeName, callbackMethod) Attributes ---------- defaultName: string default name of the tree name: string the name of the tree id: string tree identifier (a hash by default) modifiable: bool if *True*, new items can be added, items can be deleted and modified unamed: bool indicates if *name* parameter was none, so the tree has the default name. visible: bool indicates if the tree is visible (if not it may be hidden in a graphical representation) """ defaultName = "tree" def __init__(self, name=None, id=None, modifiable=True, content=[], visible=True, enabled=True): """ Parameters ---------- name: string the name of the tree id: string tree identifier (a hash by default) modifiable: bool if *True*, new items can be added, items can be deleted and modified content: list children items (:class:`EditableTree.Item`) visible: bool indicates if the tree is visible (if not it may be hidden in a graphical representation) enabled: bool """ dictContent = [(i.id, i) for i in content] super(EditableTree, self).__init__(*dictContent) if name is None: self.name = self.defaultName self.unamed = True else: self.name = name self.unamed = False if id is None: self.id = str(hash(self)) else: self.id = id self.modifiable = modifiable self.visible = visible self.enabled = enabled def __getinitargs__(self): """Returns the args to pass to the __init__ method to construct this object. It is useful in order to save EditableTree object to minf format. Returns ------- tuple: arg content to pass to the __init__ method for creating a copy of this object """ # elements in the tuple must be serializable with minf, so the content # must be of type list content = list(self.values()) return (self.name, self.id, self.modifiable, content, self.visible, self.enabled) def __str__(self): s = self.name + " (" for i in self: s += str(i) + " " s += ")" return s def __hash__(self): return ObservableAttributes.__hash__(self)
[docs] def add(self, item): """ Adds an item in the tree. If this item's id is already present in the tree as a key, add the item's content in the corresponding key. recursive method """ key = item.id if key in self: if not item.isLeaf(): # if the item is a leaf and is already in the tree, nothing to do for v in item.values(): # item is also a dictionary and contains several elements, add each value in the tree item self[key].add(v) # also set current name for the current object self[key].name = item.name else: # new item self[key] = item
[docs] def isDescendant(self, item): """Returns *True* if the current item is a child or a child's child... of the item in parameter Always *False* for the root of the tree""" return False
def isLeaf(self): return False
[docs] def removeEmptyBranches(self): """If a branch item doesn't contain any leaf or branch that contains leaf recursivly, the item is removed from the tree. """ toRemove = [] for child in self.values(): if not child.isLeaf(): child.removeEmptyBranches() if list(child.values()) == []: # it isn't possible to remove an element during iteration on the list toRemove.append( child) # so it's put on a remove list and will be removed from the tree outside the loop for item in toRemove: del self[item.id]
[docs] def sort(self, key=None, reverse=False): """Recursive sort of the tree: items are sorted in all branches. """ ObservableSortedDictionary.sort(self, self._keyItems) for item in self.values(): if not item.isLeaf(): item.sort(key=key, reverse=reverse)
def _keyItems(self, id): """Key function """ i1 = self[id] # names are translated in lowercase to make an alphabetical sort independant of the case # by default (uppercase letters are < to lowecase letters) n1 = i1.name.lower() # print "comp", i1.name,i1.isLeaf(), i2.name,i2.isLeaf(), res return (not i1.isLeaf(), n1)
[docs] def compItems(self, id1, id2): """Comparison function Returns: - 1 if ``i1 > i2`` - -1 if ``i1 < i2`` - 0 if they are equal""" res = 0 i1 = self[id1] i2 = self[id2] if (not i1.isLeaf() and i2.isLeaf()): # if one item is a branch and the other is a leaf, the branch must # be before the leaf res = -1 elif (i1.isLeaf() and not i2.isLeaf()): res = 1 else: # names are translated in lowercase to make an alphabetical sort independant of the case # by default (uppercase letters are < to lowecase letters) n1 = i1.name.lower() n2 = i2.name.lower() if n1 < n2: res = -1 elif n1 > n2: res = 1 # print "comp", i1.name,i1.isLeaf(), i2.name,i2.isLeaf(), res return res
[docs] def addListenerRec(self, listener): """Add a listener to the tree recursivly at every level. The listener is added to the notifier of all branches of the tree.""" self.addListener(listener) for item in self.values(): if not item.isLeaf(): item.addListenerRec(listener)
[docs] def onAttributeChangeRec(self, attributeName, listener): """Add a listener of the changes of this attribute value in the tree recursivly at every level. The listener is added to the attribute change notifier of all branches of the tree. """ self.onAttributeChange(attributeName, listener) for item in self.values(): if not item.isLeaf(): item.onAttributeChangeRec(attributeName, listener)
#-------------------------------------------------------------------------
[docs] class Item(ObservableAttributes): """Base element of an :class:`EditableTree` *Item* inherits from :class:`ObservableAttributes`, so it can notify registred listeners of attributes changes (for example :attr:`name`). Attributes ---------- icon: string filename of the image that can be used to represent the item name: string Text representation of the item movable: bool True if the item can be moved in the tree modifiable: bool True if the item can changed (add children, change text, icon) delEnabled: bool True if deletion of the item is allowed tooltip: string description associated to the item copyEnabled: bool indicates if this item can be copied visible: bool indicates if current item is visible (if not it may be hidden in a graphical representation) enabled: bool indicates if the item is enabled, it could be visible but disabled """ def __init__(self, name=None, id=None, icon=None, tooltip=None, copyEnabled=True, modifiable=True, delEnabled=True, visible=True, enabled=True, *args): super(EditableTree.Item, self).__init__(*args) self.icon = icon self.name = name if id is None: self.id = str(hash(self)) else: self.id = id self.tooltip = tooltip self.copyEnabled = copyEnabled self.modifiable = modifiable self.delEnabled = delEnabled self.visible = visible self.enabled = enabled self.onChangeNotifier = Notifier() self.unamed = False def __getinitargs__(self): """Returns the args to pass to the __init__ method to construct this object. It is useful to save L{ObservableList} object to minf format. Returns ------- tuple: arg content to pass to the __init__ method for creating a copy of this object """ return (self.name, self.id, self.icon, self.tooltip, self.copyEnabled, self.modifiable, self.delEnabled, self.visible, self.enabled) def __reduce__(self): """This method is redefined for enable deepcopy of this object (and potentially pickle). It gives the arguments to pass to the init method of the object when creating a copy Returns ------- tuple: class name, init args, state, iterator on elements to copy, dictionary iterator """ return (self.__class__, self.__getinitargs__(), None, None, None) def __str__(self): return self.name
[docs] def isLeaf(self): """Must be redefined in subclasses to say if the item is a leaf""" pass
[docs] def setAllModificationsEnabled(self, bool): """Recursivly enables or disables item's modification.""" # self.copyEnabled=bool # copy is always enabled. there's no use to # have 3 different booleans, an editable should be sufficient self.modifiable = bool self.delEnabled = bool if not self.isLeaf(): for child in self.values(): child.setAllModificationsEnabled(bool)
[docs] def isDescendant(self, item): """Returns *True* if the current item is a child or a child's child... of the item in parameter """ found = False # print self.name, "is descendant", item.name, "? " if not item.isLeaf(): for child in item.values(): # search self in item's children and recursively in # children's children (deep first search) if self == child: found = True elif not child.isLeaf(): found = self.isDescendant(child) if found: break return found
#-------------------------------------------------------------------------
[docs] class Branch(Item, ObservableSortedDictionary): """ A Branch is an :class:`Item <EditableTree.Item>` that can contain other items. It inherits from :class:`Item <EditableTree.Item>` and from :class:`ObservableSortedDictionary`, so it can have children items. """ defaultName = "new" def __init__(self, name=None, id=None, icon=None, tooltip=None, copyEnabled=True, modifiable=True, delEnabled=True, content=[], visible=True, enabled=True): """All parameters must have default values to be able to create new elements automatically""" # super(EditableTree.Branch, self).__init__(content) #, name, icon, tooltip, copyEnabled, modifiable, delEnabled) # EditableTree.Item.__init__(self, name, icon, tooltip, # copyEnabled, modifiable, delEnabled) dictContent = [(i.id, i) for i in content] super(EditableTree.Branch, self).__init__(name, id, icon, tooltip, copyEnabled, modifiable, delEnabled, visible, enabled, *dictContent) if name is None: self.name = self.defaultName self.unamed = True else: self.name = name self.unamed = False def __getinitargs__(self): content = list(self.values()) return (self.name, self.id, self.icon, self.tooltip, self.copyEnabled, self.modifiable, self.delEnabled, content, self.visible, self.enabled) def __reduce__(self): """This method is redefined for enable deepcopy of this object (and potentially pickle). It gives the arguments to pass to the init method of the object when creating a copy """ return (self.__class__, self.__getinitargs__(), None, None, None) def __str__(self): s = self.name + " (" for i in self: s += str(i) + " " s += ")" return s def __hash__(self): return EditableTree.Item.__hash__(self)
[docs] def isLeaf(self): return False
[docs] def add(self, item): """ Adds an item in the tree. If this item's id is already present in the tree as a key, add the item's content in the corresponding key. recursive method """ key = item.id if key in self: if not self[key].isLeaf(): # if the item is a leaf and is already in the tree, nothing to do for v in item.values(): # item is also a dictionary and contains several elements, add each value in the tree item self[key].add(v) # also set current name for the current object self[key].name = item.name else: # new item self[key] = item
[docs] def removeEmptyBranches(self): toRemove = [] for child in self.values(): if not child.isLeaf(): child.removeEmptyBranches() if list(child.values()) == []: toRemove.append(child) for item in toRemove: del self[item.id]
[docs] def sort(self, key=None, reverse=False): """Recursive sort of the tree: items are sorted in all branches. """ ObservableSortedDictionary.sort(self, self._keyItems) for item in self.values(): if not item.isLeaf(): item.sort(key=key, reverse=reverse)
def _keyItems(self, id): '''Sorting key function''' i1 = self[id] # names are translated in lowercase to make an alphabetical sort independant of the case # by default (uppercase letters are < to lowecase letters) n1 = i1.name.lower() # print "comp", i1.name,i1.isLeaf(), i2.name,i2.isLeaf(), res # leafs must go after subtrees return (not i1.isLeaf(), n1)
[docs] def compItems(self, id1, id2): """Comparison function Returns - 1 if i1 > i2 - -1 if i1 < i2 - 0 if they are equal""" res = 0 i1 = self[id1] i2 = self[id2] if (not i1.isLeaf() and i2.isLeaf()): # if one item is a branch and the other is a leaf, the branch # must be before the leaf res = -1 elif (i1.isLeaf() and not i2.isLeaf()): res = 1 else: # names are translated in lowercase to make an alphabetical sort independant of the case # by default (uppercase letters are < to lowecase letters) n1 = i1.name.lower() n2 = i2.name.lower() if n1 < n2: res = -1 elif n1 > n2: res = 1 # print "comp", i1.name,i1.isLeaf(), i2.name,i2.isLeaf(), res return res
[docs] def addListenerRec(self, listener): self.addListener(listener) for item in self.values(): if not item.isLeaf(): item.addListenerRec(listener)
[docs] def onAttributeChangeRec(self, attributeName, listener): self.onAttributeChange(attributeName, listener) for item in self.values(): if not item.isLeaf(): item.onAttributeChangeRec(attributeName, listener)
#-------------------------------------------------------------------------
[docs] class Leaf(Item): """A tree item that cannot have children items""" def __init__(self, name="new", id=None, icon=None, tooltip=None, copyEnabled=True, modifiable=True, delEnabled=True, visible=True, enabled=True): super(EditableTree.Leaf, self).__init__( name, id, icon, tooltip, copyEnabled, modifiable, delEnabled, visible, enabled)
[docs] def isLeaf(self): return True
#-------------------------------------------------------------------------
[docs]class ObservableNotifier(Notifier): """ This notifier can notify when the first listener is added and when the last listener is removed. It enables to use the notifier only when there are some listeners registred. Attributes ---------- onAddFirstListener: :class:`Notifier` register a listener on this notifier to be called when the first listener is registred on ObservableNotifier onRemoveLastListener: :class:`Notifier` register a listener on this notifier to be called when the last listener is removed from ObservableNotifier """ def __init__(self, parameterCount=None): """ Parameters ---------- parameterCount: int if not *None*, each registered function must be callable with that number of arguments (checking is done on registration). """ Notifier.__init__(self, parameterCount) self.onAddFirstListener = Notifier() self.onRemoveLastListener = Notifier()
[docs] def add(self, listener): nbListenersBefore = len(self._listeners) Notifier.add(self, listener) if nbListenersBefore == 0: # before add : 0 listener, after : 1 listener -> add first listener if len(self._listeners) == 1: self.onAddFirstListener.notify()
[docs] def remove(self, listener): nbListenersBefore = len(self._listeners) Notifier.remove(self, listener) if nbListenersBefore == 1: # before add : 1 listener, after : 0 listener -> remove last listener if len(self._listeners) == 0: self.onRemoveLastListener.notify()