#-------------------------------------------------------------------------------
#
# Copyright (c) 2007, Enthought, Inc.
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in enthought/LICENSE.txt and may be redistributed only
# under the conditions described in the aforementioned license. The license
# is also available online at http://www.enthought.com/licenses/BSD.txt
#
# Thanks for using Enthought open source!
#
# Author: David C. Morrill
# Date: 03/05/2007
#
#-------------------------------------------------------------------------------
""" Defines classes used to implement and manage various trait listener
patterns.
"""
#-------------------------------------------------------------------------------
# Imports:
#-------------------------------------------------------------------------------
from __future__ import absolute_import
import re
import string
import weakref
from weakref import WeakKeyDictionary
from string import whitespace
from types import MethodType
from .has_traits import HasPrivateTraits
from .trait_base import Undefined, Uninitialized
from .traits import Property
from .trait_types import Str, Int, Bool, Instance, List, Enum, Any
from .trait_errors import TraitError
from .trait_notifiers import TraitChangeNotifyWrapper
#---------------------------------------------------------------------------
# Constants:
#---------------------------------------------------------------------------
# The name of the dictionary used to store active listeners
TraitsListener = '__traits_listener__'
# End of String marker
EOS = '\0'
# Types of traits that can be listened to
ANYTRAIT_LISTENER = '_register_anytrait'
SIMPLE_LISTENER = '_register_simple'
LIST_LISTENER = '_register_list'
DICT_LISTENER = '_register_dict'
SET_LISTENER = '_register_set'
# Mapping from trait default value types to listener types
type_map = {
5: LIST_LISTENER,
6: DICT_LISTENER,
9: SET_LISTENER
}
# Listener types:
ANY_LISTENER = 0
SRC_LISTENER = 1
DST_LISTENER = 2
ListenerType = {
0: ANY_LISTENER,
1: DST_LISTENER,
2: DST_LISTENER,
3: SRC_LISTENER,
4: SRC_LISTENER
}
# Invalid destination ( object, name ) reference marker (i.e. ambiguous):
INVALID_DESTINATION = ( None, None )
# Regular expressions used by the parser:
simple_pat = re.compile( r'^([a-zA-Z_]\w*)(\.|:)([a-zA-Z_]\w*)$' )
name_pat = re.compile( r'([a-zA-Z_]\w*)\s*(.*)' )
# Characters valid in a traits name:
name_chars = string.ascii_letters + string.digits + '_'
#-------------------------------------------------------------------------------
# Utility functions:
#-------------------------------------------------------------------------------
[docs]def indent ( text, first_line = True, n = 1, width = 4 ):
""" Indent lines of text.
Parameters
----------
text : str
The text to indent.
first_line : bool, optional
If False, then the first line will not be indented (default: True).
n : int, optional
The level of indentation (default: 1).
width : int, optional
The number of spaces in each level of indentation (default: 4).
Returns
-------
indented : str
"""
lines = text.split( '\n' )
if not first_line:
first = lines[0]
lines = lines[1:]
spaces = ' ' * (width * n)
lines2 = [ spaces + x for x in lines ]
if not first_line:
lines2.insert( 0, first )
indented = '\n'.join( lines2 )
return indented
#-------------------------------------------------------------------------------
# Metadata filters:
#-------------------------------------------------------------------------------
[docs]def is_not_none ( value ): return (value is not None)
[docs]def is_none ( value ): return (value is None)
[docs]def not_event ( value ): return (value != 'event')
#-------------------------------------------------------------------------------
# 'ListenerBase' class:
#-------------------------------------------------------------------------------
[docs]class ListenerBase ( HasPrivateTraits ):
#---------------------------------------------------------------------------
# Trait definitions:
#---------------------------------------------------------------------------
# The handler to be called when any listened to trait is changed:
#handler = Any
# The dispatch mechanism to use when invoking the handler:
#dispatch = Str
# Does the handler go at the beginning (True) or end (False) of the
# notification handlers list?
#priority = Bool( False )
# The next level (if any) of ListenerBase object to be called when any of
# our listened to traits is changed:
#next = Instance( ListenerBase )
# The type of handler being used:
#type = Enum( ANY_LISTENER, SRC_LISTENER, DST_LISTENER )
# Should changes to this item generate a notification to the handler?
# notify = Bool
# Should registering listeners for items reachable from this listener item
# be deferred until the associated trait is first read or set?
# deferred = Bool
#---------------------------------------------------------------------------
# Registers new listeners:
#---------------------------------------------------------------------------
[docs] def register ( self, new ):
""" Registers new listeners.
"""
raise NotImplementedError
#---------------------------------------------------------------------------
# Unregisters any existing listeners:
#---------------------------------------------------------------------------
[docs] def unregister ( self, old ):
""" Unregisters any existing listeners.
"""
raise NotImplementedError
#---------------------------------------------------------------------------
# Handles a trait change for a simple trait:
#---------------------------------------------------------------------------
[docs] def handle ( self, object, name, old, new ):
""" Handles a trait change for a simple trait.
"""
raise NotImplementedError
#---------------------------------------------------------------------------
# Handles a trait change for a list trait:
#---------------------------------------------------------------------------
[docs] def handle_list ( self, object, name, old, new ):
""" Handles a trait change for a list trait.
"""
raise NotImplementedError
#---------------------------------------------------------------------------
# Handles a trait change for a list traits items:
#---------------------------------------------------------------------------
[docs] def handle_list_items ( self, object, name, old, new ):
""" Handles a trait change for a list traits items.
"""
raise NotImplementedError
#---------------------------------------------------------------------------
# Handles a trait change for a dictionary trait:
#---------------------------------------------------------------------------
[docs] def handle_dict ( self, object, name, old, new ):
""" Handles a trait change for a dictionary trait.
"""
raise NotImplementedError
#---------------------------------------------------------------------------
# Handles a trait change for a dictionary traits items:
#---------------------------------------------------------------------------
[docs] def handle_dict_items ( self, object, name, old, new ):
""" Handles a trait change for a dictionary traits items.
"""
raise NotImplementedError
#-------------------------------------------------------------------------------
# 'ListenerItem' class:
#-------------------------------------------------------------------------------
[docs]class ListenerItem ( ListenerBase ):
#---------------------------------------------------------------------------
# Trait definitions:
#---------------------------------------------------------------------------
#: The name of the trait to listen to:
name = Str
#: The name of any metadata that must be present (or not present):
metadata_name = Str
#: Does the specified metadata need to be defined (True) or not defined
#: (False)?
metadata_defined = Bool( True )
#: The handler to be called when any listened-to trait is changed:
handler = Any
#: A weakref 'wrapped' version of 'handler':
wrapped_handler_ref = Any
#: The dispatch mechanism to use when invoking the handler:
dispatch = Str
#: Does the handler go at the beginning (True) or end (False) of the
#: notification handlers list?
priority = Bool( False )
#: The next level (if any) of ListenerBase object to be called when any of
#: this object's listened-to traits is changed:
next = Instance( ListenerBase )
#: The type of handler being used:
type = Enum( ANY_LISTENER, SRC_LISTENER, DST_LISTENER )
#: Should changes to this item generate a notification to the handler?
notify = Bool( True )
#: Should registering listeners for items reachable from this listener item
#: be deferred until the associated trait is first read or set?
deferred = Bool( False )
#: Is this an 'any_trait' change listener, or does it create explicit
#: listeners for each individual trait?
is_any_trait = Bool( False )
#: Is the associated handler a special list handler that handles both
#: 'foo' and 'foo_items' events by receiving a list of 'deleted' and 'added'
#: items as the 'old' and 'new' arguments?
is_list_handler = Bool( False )
#: A dictionary mapping objects to a list of all current active
#: (*name*, *type*) listener pairs, where *type* defines the type of
#: listener, one of: (SIMPLE_LISTENER, LIST_LISTENER, DICT_LISTENER).
active = Instance( WeakKeyDictionary, () )
#-- 'ListenerBase' Class Method Implementations ----------------------------
#---------------------------------------------------------------------------
# String representation:
#---------------------------------------------------------------------------
def __repr__ ( self, seen = None ):
"""Returns a string representation of the object.
Since the object graph may have cycles, we extend the basic __repr__ API
to include a set of objects we've already seen while constructing
a string representation. When this method tries to get the repr of
a ListenerItem or ListenerGroup, we will use the extended API and build
up the set of seen objects. The repr of a seen object will just be
'<cycle>'.
"""
if seen is None:
seen = set()
seen.add( self )
next_repr = 'None'
next = self.next
if next is not None:
if next in seen:
next_repr = '<cycle>'
else:
next_repr = next.__repr__( seen )
return """%s(
name = %r,
metadata_name = %r,
metadata_defined = %r,
is_any_trait = %r,
dispatch = %r,
notify = %r,
is_list_handler = %r,
type = %r,
next = %s,
)""" % ( self.__class__.__name__, self.name, self.metadata_name,
self.metadata_defined, self.is_any_trait, self.dispatch, self.notify,
self.is_list_handler, self.type, indent( next_repr, False ) )
#---------------------------------------------------------------------------
# Registers new listeners:
#---------------------------------------------------------------------------
[docs] def register ( self, new ):
""" Registers new listeners.
"""
# Make sure we actually have an object to set listeners on and that it
# has not already been registered (cycle breaking):
if (new is None) or (new is Undefined) or (new in self.active):
return INVALID_DESTINATION
# Create a dictionary of {name: trait_values} that match the object's
# definition for the 'new' object:
name = self.name
last = name[-1:]
if last == '*':
# Handle the special case of an 'anytrait' change listener:
if self.is_any_trait:
try:
self.active[ new ] = [ ( '', ANYTRAIT_LISTENER ) ]
return self._register_anytrait( new, '', False )
except TypeError:
# This error can occur if 'new' is a list or other object
# for which a weakref cannot be created as the dictionary
# key for 'self.active':
return INVALID_DESTINATION
# Handle trait matching based on a common name prefix and/or
# matching trait metadata:
metadata = self._metadata
if metadata is None:
self._metadata = metadata = { 'type': not_event }
if self.metadata_name != '':
if self.metadata_defined:
metadata[ self.metadata_name ] = is_not_none
else:
metadata[ self.metadata_name ] = is_none
# Get all object traits with matching metadata:
names = new.trait_names( **metadata )
# If a name prefix was specified, filter out only the names that
# start with the specified prefix:
name = name[:-1]
if name != '':
n = len( name )
names = [ aname for aname in names if name == aname[ : n ] ]
# Create the dictionary of selected traits:
bt = new.base_trait
traits = dict( [ ( name, bt( name ) ) for name in names ] )
# Handle any new traits added dynamically to the object:
new.on_trait_change( self._new_trait_added, 'trait_added' )
else:
# Determine if the trait is optional or not:
optional = (last == '?')
if optional:
name = name[:-1]
# Else, no wildcard matching, just get the specified trait:
trait = new.base_trait( name )
# Try to get the object trait:
if trait is None:
# Raise an error if trait is not defined and not optional:
# fixme: Properties which are lists don't implement the
# '..._items' sub-trait, which can cause a failure here when
# used with an editor that sets up listeners on the items...
if not optional:
raise TraitError( "'%s' object has no '%s' trait" % (
new.__class__.__name__, name ) )
# Otherwise, just skip it:
traits = {}
else:
# Create a result dictionary containing just the single trait:
traits = { name: trait }
# For each item, determine its type (simple, list, dict):
self.active[ new ] = active = []
for name, trait in traits.items():
# Determine whether the trait type is simple, list, set or
# dictionary:
type = SIMPLE_LISTENER
handler = trait.handler
if handler is not None:
type = type_map.get( handler.default_value_type,
SIMPLE_LISTENER )
# Add the name and type to the list of traits being registered:
active.append( ( name, type ) )
# Set up the appropriate trait listeners on the object for the
# current trait:
value = getattr( self, type )( new, name, False )
if len( traits ) == 1:
return value
return INVALID_DESTINATION
#---------------------------------------------------------------------------
# Unregisters any existing listeners:
#---------------------------------------------------------------------------
[docs] def unregister ( self, old ):
""" Unregisters any existing listeners.
"""
if old is not None and old is not Uninitialized:
try:
active = self.active.pop( old, None )
if active is not None:
for name, type in active:
getattr( self, type )( old, name, True )
except TypeError:
# An error can occur if 'old' is a list or other object for
# which a weakref cannot be created and used an a key for
# 'self.active':
pass
#---------------------------------------------------------------------------
# Handles a trait change for an intermediate link trait:
#---------------------------------------------------------------------------
[docs] def handle_simple ( self, object, name, old, new ):
""" Handles a trait change for an intermediate link trait.
"""
self.next.unregister( old )
self.next.register( new )
[docs] def handle_dst ( self, object, name, old, new ):
""" Handles a trait change for an intermediate link trait when the
notification is for the final destination trait.
"""
self.next.unregister( old )
object, name = self.next.register( new )
if old is not Uninitialized:
if object is None:
raise TraitError( "on_trait_change handler signature is "
"incompatible with a change to an intermediate trait" )
wh = self.wrapped_handler_ref()
if wh is not None:
wh( object, name, old,
getattr( object, name, Undefined ) )
#---------------------------------------------------------------------------
# Handles a trait change for a list (or set) trait:
#---------------------------------------------------------------------------
[docs] def handle_list ( self, object, name, old, new ):
""" Handles a trait change for a list (or set) trait.
"""
if old is not None and old is not Uninitialized:
unregister = self.next.unregister
for obj in old:
unregister( obj )
register = self.next.register
for obj in new:
register( obj )
#---------------------------------------------------------------------------
# Handles a trait change for a list (or set) traits items:
#---------------------------------------------------------------------------
[docs] def handle_list_items ( self, object, name, old, new ):
""" Handles a trait change for items of a list (or set) trait.
"""
self.handle_list( object, name, new.removed, new.added )
[docs] def handle_list_items_special ( self, object, name, old, new ):
""" Handles a trait change for items of a list (or set) trait with
notification.
"""
wh = self.wrapped_handler_ref()
if wh is not None:
wh( object, name, new.removed, new.added )
#---------------------------------------------------------------------------
# Handles a trait change for a dictionary trait:
#---------------------------------------------------------------------------
[docs] def handle_dict ( self, object, name, old, new ):
""" Handles a trait change for a dictionary trait.
"""
if old is not Uninitialized:
unregister = self.next.unregister
for obj in old.values():
unregister( obj )
register = self.next.register
for obj in new.values():
register( obj )
#---------------------------------------------------------------------------
# Handles a trait change for a dictionary traits items:
#---------------------------------------------------------------------------
[docs] def handle_dict_items ( self, object, name, old, new ):
""" Handles a trait change for items of a dictionary trait.
"""
self.handle_dict( object, name, new.removed, new.added )
if len( new.changed ) > 0:
# If 'name' refers to the '_items' trait, then remove the '_items'
# suffix to get the actual dictionary trait.
#
# fixme: Is there ever a case where 'name' *won't* refer to the
# '_items' trait?
if name.endswith('_items'):
name = name[:-len('_items')]
dict = getattr( object, name )
unregister = self.next.unregister
register = self.next.register
for key, obj in new.changed.items():
unregister( obj )
register( dict[ key ] )
#---------------------------------------------------------------------------
# Handles an invalid intermediate trait change to a handler that must be
# applied to the final destination object.trait:
#---------------------------------------------------------------------------
[docs] def handle_error ( self, obj, name, old, new ):
""" Handles an invalid intermediate trait change to a handler that must
be applied to the final destination object.trait.
"""
if old is not None and old is not Uninitialized:
raise TraitError( "on_trait_change handler signature is "
"incompatible with a change to an intermediate trait" )
#-- Event Handlers ---------------------------------------------------------
#---------------------------------------------------------------------------
# Handles the 'handler' trait being changed:
#---------------------------------------------------------------------------
def _handler_changed ( self, handler ):
""" Handles the **handler** trait being changed.
"""
if self.next is not None:
self.next.handler = handler
#---------------------------------------------------------------------------
# Handles the 'wrapped_handler_ref' trait being changed:
#---------------------------------------------------------------------------
def _wrapped_handler_ref_changed ( self, wrapped_handler_ref ):
""" Handles the 'wrapped_handler_ref' trait being changed.
"""
if self.next is not None:
self.next.wrapped_handler_ref = wrapped_handler_ref
#---------------------------------------------------------------------------
# Handles the 'dispatch' trait being changed:
#---------------------------------------------------------------------------
def _dispatch_changed ( self, dispatch ):
""" Handles the **dispatch** trait being changed.
"""
if self.next is not None:
self.next.dispatch = dispatch
#---------------------------------------------------------------------------
# Handles the 'priority' trait being changed:
#---------------------------------------------------------------------------
def _priority_changed ( self, priority ):
""" Handles the **priority** trait being changed.
"""
if self.next is not None:
self.next.priority = priority
#-- Private Methods --------------------------------------------------------
#---------------------------------------------------------------------------
# Registers any 'anytrait' listener:
#---------------------------------------------------------------------------
def _register_anytrait ( self, object, name, remove ):
""" Registers any 'anytrait' listener.
"""
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
return ( object, name )
#---------------------------------------------------------------------------
# Registers a handler for a simple trait:
#---------------------------------------------------------------------------
def _register_simple ( self, object, name, remove ):
""" Registers a handler for a simple trait.
"""
next = self.next
if next is None:
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
return ( object, name )
tl_handler = self.handle_simple
if self.notify:
if self.type == DST_LISTENER:
if self.dispatch != 'same':
raise TraitError( "Trait notification dispatch type '%s' "
"is not compatible with handler signature and "
"extended trait name notification style" % self.dispatch )
tl_handler = self.handle_dst
else:
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
object._on_trait_change(
tl_handler,
name,
remove=remove,
dispatch='extended',
priority=self.priority,
target=self._get_target(),
)
if remove:
return next.unregister( getattr( object, name ) )
if not self.deferred or name in object.__dict__:
# Sometimes, the trait may already be assigned. This can happen when
# there are chains of dynamic initializers and 'delegate'
# notifications. If 'trait_a' and 'trait_b' have dynamic
# initializers and 'trait_a's initializer creates 'trait_b', *and*
# we have a DelegatesTo trait that delegates to 'trait_a', then the
# listener that implements the delegate will create 'trait_a' and
# thus 'trait_b'. If we are creating an extended trait change
# listener on 'trait_b.something', and the 'trait_a' delegate
# listeners just happen to get hooked up before this one, then
# 'trait_b' will have been initialized already, and the registration
# that we are deferring will never happen.
return next.register( getattr( object, name ) )
return ( object, name )
#---------------------------------------------------------------------------
# Registers a handler for a list trait:
#---------------------------------------------------------------------------
def _register_list ( self, object, name, remove ):
""" Registers a handler for a list trait.
"""
next = self.next
if next is None:
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
if self.is_list_handler:
object._on_trait_change(
self.handle_list_items_special,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
elif self.type == ANY_LISTENER:
object._on_trait_change(
handler,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
return ( object, name )
tl_handler = self.handle_list
tl_handler_items = self.handle_list_items
if self.notify:
if self.type == DST_LISTENER:
tl_handler = tl_handler_items = self.handle_error
else:
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
if self.is_list_handler:
object._on_trait_change(
self.handle_list_items_special,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
elif self.type == ANY_LISTENER:
object._on_trait_change(
handler,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
object._on_trait_change(
tl_handler,
name,
remove=remove,
dispatch='extended',
priority=self.priority,
target=self._get_target(),
)
object._on_trait_change(
tl_handler_items,
name + '_items',
remove=remove,
dispatch='extended',
priority=self.priority,
target=self._get_target(),
)
if remove:
handler = next.unregister
elif self.deferred:
return INVALID_DESTINATION
else:
handler = next.register
for obj in getattr( object, name ):
handler( obj )
return INVALID_DESTINATION
# Handle 'sets' the same as 'lists':
# Note: Currently the behavior of sets is almost identical to that of lists,
# so we are able to share the same code for both. This includes some 'duck
# typing' that occurs with the TraitListEvent and TraitSetEvent, that define
# 'removed' and 'added' attributes that behave similarly enough (from the
# point of view of this module) that they can be treated as equivalent. If
# the behavior of sets ever diverges from that of lists, then this code may
# need to be changed.
_register_set = _register_list
#---------------------------------------------------------------------------
# Registers a handler for a dictionary trait:
#---------------------------------------------------------------------------
def _register_dict ( self, object, name, remove ):
""" Registers a handler for a dictionary trait.
"""
next = self.next
if next is None:
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
if self.type == ANY_LISTENER:
object._on_trait_change(
handler,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
return ( object, name )
tl_handler = self.handle_dict
tl_handler_items = self.handle_dict_items
if self.notify:
if self.type == DST_LISTENER:
tl_handler = tl_handler_items = self.handle_error
else:
handler = self.handler()
if handler is not Undefined:
object._on_trait_change(
handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
if self.type == ANY_LISTENER:
object._on_trait_change(
handler,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
object._on_trait_change(
tl_handler,
name,
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
object._on_trait_change(
tl_handler_items,
name + '_items',
remove=remove,
dispatch=self.dispatch,
priority=self.priority,
target=self._get_target(),
)
if remove:
handler = next.unregister
elif self.deferred:
return INVALID_DESTINATION
else:
handler = next.register
for obj in getattr( object, name ).values():
handler( obj )
return INVALID_DESTINATION
#---------------------------------------------------------------------------
# Handles new traits being added to an object being monitored:
#---------------------------------------------------------------------------
def _new_trait_added ( self, object, name, new_trait ):
""" Handles new traits being added to an object being monitored.
"""
# Set if the new trait matches our prefix and metadata:
if new_trait.startswith( self.name[:-1] ):
trait = object.base_trait( new_trait )
for meta_name, meta_eval in self._metadata.items():
if not meta_eval( getattr( trait, meta_name ) ):
return
# Determine whether the trait type is simple, list, set or
# dictionary:
type = SIMPLE_LISTENER
handler = trait.handler
if handler is not None:
type = type_map.get( handler.default_value_,
SIMPLE_LISTENER )
# Add the name and type to the list of traits being registered:
self.active[ object ].append( ( new_trait, type ) )
# Set up the appropriate trait listeners on the object for the
# new trait:
getattr( self, type )( object, new_trait, False )
def _get_target(self):
""" Get the target object from the ListenerNotifyWrapper.
"""
target = None
lnw = self.wrapped_handler_ref()
if lnw is not None:
target_ref = getattr(lnw, 'object', None)
if target_ref is not None:
target = target_ref()
return target
#-------------------------------------------------------------------------------
# 'ListenerGroup' class:
#-------------------------------------------------------------------------------
def _set_value ( self, name, value ):
for item in self.items:
setattr( item, name, value )
def _get_value ( self, name ):
# Use the attribute on the first item. If there are no items, return None.
if self.items:
return getattr( self.items[0], name )
else:
return None
ListProperty = Property( fget = _get_value, fset = _set_value )
[docs]class ListenerGroup ( ListenerBase ):
#---------------------------------------------------------------------------
# Trait definitions:
#---------------------------------------------------------------------------
#: The handler to be called when any listened-to trait is changed
handler = Property
#: A weakref 'wrapped' version of 'handler':
wrapped_handler_ref = Property
#: The dispatch mechanism to use when invoking the handler:
dispatch = Property
#: Does the handler go at the beginning (True) or end (False) of the
#: notification handlers list?
priority = ListProperty
#: The next level (if any) of ListenerBase object to be called when any of
#: this object's listened-to traits is changed
next = ListProperty
#: The type of handler being used:
type = ListProperty
#: Should changes to this item generate a notification to the handler?
notify = ListProperty
#: Should registering listeners for items reachable from this listener item
#: be deferred until the associated trait is first read or set?
deferred = ListProperty
# The list of ListenerBase objects in the group
items = List( ListenerBase )
#-- Property Implementations -----------------------------------------------
def _set_handler ( self, handler ):
if self._handler is None:
self._handler = handler
for item in self.items:
item.handler = handler
def _set_wrapped_handler_ref ( self, wrapped_handler_ref ):
if self._wrapped_handler_ref is None:
self._wrapped_handler_ref = wrapped_handler_ref
for item in self.items:
item.wrapped_handler_ref = wrapped_handler_ref
def _set_dispatch ( self, dispatch ):
if self._dispatch is None:
self._dispatch = dispatch
for item in self.items:
item.dispatch = dispatch
#-- 'ListenerBase' Class Method Implementations ----------------------------
#---------------------------------------------------------------------------
# String representation:
#---------------------------------------------------------------------------
def __repr__ ( self, seen = None ):
"""Returns a string representation of the object.
Since the object graph may have cycles, we extend the basic __repr__ API
to include a set of objects we've already seen while constructing
a string representation. When this method tries to get the repr of
a ListenerItem or ListenerGroup, we will use the extended API and build
up the set of seen objects. The repr of a seen object will just be
'<cycle>'.
"""
if seen is None:
seen = set()
seen.add( self )
lines = [ '%s(items = [' % self.__class__.__name__ ]
for item in self.items:
lines.extend( indent( item.__repr__( seen ), True ).split( '\n' ) )
lines[-1] += ','
lines.append( '])' )
return '\n'.join( lines )
#---------------------------------------------------------------------------
# Registers new listeners:
#---------------------------------------------------------------------------
[docs] def register ( self, new ):
""" Registers new listeners.
"""
for item in self.items:
item.register( new )
return INVALID_DESTINATION
#---------------------------------------------------------------------------
# Unregisters any existing listeners:
#---------------------------------------------------------------------------
[docs] def unregister ( self, old ):
""" Unregisters any existing listeners.
"""
for item in self.items:
item.unregister( old )
#-------------------------------------------------------------------------------
# 'ListenerParser' class:
#-------------------------------------------------------------------------------
[docs]class ListenerParser ( HasPrivateTraits ):
#-------------------------------------------------------------------------------
# Trait definitions:
#-------------------------------------------------------------------------------
#: The string being parsed
text = Str
#: The length of the string being parsed.
len_text = Int
#: The current parse index within the string
index = Int
#: The next character from the string being parsed
next = Property
#: The next Python attribute name within the string:
name = Property
#: The next non-whitespace character
skip_ws = Property
#: Backspaces to the last character processed
backspace = Property
#: The ListenerBase object resulting from parsing **text**
listener = Instance( ListenerBase )
#-- Property Implementations -----------------------------------------------
def _get_next ( self ):
index = self.index
self.index += 1
if index >= self.len_text:
return EOS
return self.text[ index ]
def _get_backspace ( self ):
self.index = max( 0, self.index - 1 )
def _get_skip_ws ( self ):
while True:
c = self.next
if c not in whitespace:
return c
def _get_name ( self ):
match = name_pat.match( self.text, self.index - 1 )
if match is None:
return ''
self.index = match.start( 2 )
return match.group( 1 )
#-- object Method Overrides ------------------------------------------------
def __init__ ( self, text = '', **traits ):
self.text = text
super( ListenerParser, self ).__init__( **traits )
#-- Private Methods --------------------------------------------------------
#---------------------------------------------------------------------------
# Parses the text and returns the appropriate collection of ListenerBase
# objects described by the text:
#---------------------------------------------------------------------------
[docs] def parse ( self ):
""" Parses the text and returns the appropriate collection of
ListenerBase objects described by the text.
"""
# Try a simple case of 'name1.name2'. The simplest case of a single
# Python name never triggers this parser, so we don't try to make that
# a shortcut too. Whitespace should already have been stripped from the
# start and end.
# TODO: The use of regexes should be used throughout all of the parsing
# functions to speed up all aspects of parsing.
match = simple_pat.match( self.text )
if match is not None:
return ListenerItem(
name = match.group( 1 ),
notify = match.group(2) == '.',
next = ListenerItem( name = match.group( 3 ) ) )
return self.parse_group( EOS )
#---------------------------------------------------------------------------
# Parses the contents of a group:
#---------------------------------------------------------------------------
[docs] def parse_group ( self, terminator = ']' ):
""" Parses the contents of a group.
"""
items = []
while True:
items.append( self.parse_item( terminator ) )
c = self.skip_ws
if c == terminator:
break
if c != ',':
if terminator == EOS:
self.error( "Expected ',' or end of string" )
else:
self.error( "Expected ',' or '%s'" % terminator )
if len( items ) == 1:
return items[0]
return ListenerGroup( items = items )
#---------------------------------------------------------------------------
# Parses a single, complete listener item/group string:
#---------------------------------------------------------------------------
[docs] def parse_item ( self, terminator ):
""" Parses a single, complete listener item or group string.
"""
c = self.skip_ws
if c == '[':
result = self.parse_group()
c = self.skip_ws
else:
name = self.name
if name != '':
c = self.next
result = ListenerItem( name = name )
if c in '+-':
result.name += '*'
result.metadata_defined = (c == '+')
cn = self.skip_ws
result.metadata_name = metadata = self.name
if metadata != '':
cn = self.skip_ws
result.is_any_trait = ((c == '-') and (name == '') and
(metadata == ''))
c = cn
if result.is_any_trait and (not ((c == terminator) or
((c == ',') and (terminator == ']')))):
self.error( "Expected end of name" )
elif c == '?':
if len( name ) == 0:
self.error( "Expected non-empty name preceding '?'" )
result.name += '?'
c = self.skip_ws
cycle = (c == '*')
if cycle:
c = self.skip_ws
if c in '.:':
result.notify = (c == '.')
next = self.parse_item( terminator )
if cycle:
last = result
while last.next is not None:
last = last.next
last.next = lg = ListenerGroup( items = [ next, result ] )
result = lg
else:
result.next = next
return result
if c == '[':
is_closing_bracket = (self.skip_ws == ']')
next_char = self.skip_ws
item_complete = (next_char == terminator or next_char == ',')
if is_closing_bracket and item_complete:
self.backspace
result.is_list_handler = True
else:
self.error( "Expected '[]' at the end of an item" )
else:
self.backspace
if cycle:
result.next = result
return result
#---------------------------------------------------------------------------
# Parses the metadata portion of a listener item:
#---------------------------------------------------------------------------
[docs] def error ( self, msg ):
""" Raises a syntax error.
"""
raise TraitError( "%s at column %d of '%s'" %
( msg, self.index, self.text ) )
#-- Event Handlers ---------------------------------------------------------
#---------------------------------------------------------------------------
# Handles the 'text' trait being changed:
#---------------------------------------------------------------------------
def _text_changed ( self ):
self.index = 0
self.len_text = len( self.text )
self.listener = self.parse()
#-------------------------------------------------------------------------------
# 'ListenerNotifyWrapper' class:
#-------------------------------------------------------------------------------
[docs]class ListenerNotifyWrapper ( TraitChangeNotifyWrapper ):
#-- TraitChangeNotifyWrapper Method Overrides ------------------------------
def __init__ ( self, handler, owner, id, listener, target=None):
self.type = ListenerType.get( self.init( handler,
weakref.ref( owner, self.owner_deleted ), target ) )
self.id = id
self.listener = listener
[docs] def listener_deleted ( self, ref ):
owner = self.owner()
if owner is not None:
dict = owner.__dict__.get( TraitsListener )
listeners = dict.get( self.id )
listeners.remove( self )
if len( listeners ) == 0:
del dict[ self.id ]
if len( dict ) == 0:
del owner.__dict__[ TraitsListener ]
# fixme: Is the following line necessary, since all registered
# notifiers should be getting the same 'listener_deleted' call:
self.listener.unregister( owner )
self.object = self.owner = self.listener = None
[docs] def owner_deleted ( self, ref ):
self.object = self.owner = None
#-------------------------------------------------------------------------------
# 'ListenerHandler' class:
#-------------------------------------------------------------------------------
[docs]class ListenerHandler ( object ):
def __init__ ( self, handler ):
if type( handler ) is MethodType:
object = handler.im_self
if object is not None:
self.object = weakref.ref( object, self.listener_deleted )
self.name = handler.__name__
return
self.handler = handler
def __call__ ( self ):
result = getattr( self, 'handler', None )
if result is not None:
return result
return getattr( self.object(), self.name )
[docs] def listener_deleted ( self, ref ):
self.handler = Undefined