1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
|
# (C) Copyright 2005-2023 Enthought, Inc., Austin, TX
# All rights reserved.
#
# This software is provided without warranty under the terms of the BSD
# license included in 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!
from traits.observation._has_traits_helpers import (
iter_objects,
object_has_named_trait,
)
from traits.observation._i_observer import IObserver
from traits.observation._observe import add_or_remove_notifiers
from traits.observation._observer_change_notifier import ObserverChangeNotifier
from traits.observation._observer_graph import ObserverGraph
from traits.observation._trait_change_event import trait_event_factory
@IObserver.register
class TraitAddedObserver:
""" An observer for observing the trait_added event.
This observer only offers the "maintainer". When this observer is used in
an ObserverGraph, its subgraphs are the graphs to be hooked up when a new
trait is added, provided that the trait satisfies a given criterion.
The criterion should align with the root observer(s) in these subgraph(s).
Parameters
----------
match_func : callable(str, CTrait) -> bool
A callable that receives the name of the trait added and the
corresponding trait. The returned boolean indicates whether
notifiers should be added/removed for the added trait.
This callable is used for equality check and must be hashable.
optional : boolean
Whether to skip this observer if the trait_added trait cannot be
found on the incoming object.
"""
def __init__(self, match_func, optional):
self.match_func = match_func
self.optional = optional
def __hash__(self):
""" Return a hash of this object."""
return hash((type(self).__name__, self.match_func, self.optional))
def __eq__(self, other):
""" Return true if this observer is equal to the given one."""
return (
type(self) is type(other)
and self.match_func == other.match_func
and self.optional == other.optional
)
@property
def notify(self):
""" A boolean for whether this observer will notify for changes.
"""
return False
def iter_observables(self, object):
""" Yield observables for notifiers to be attached to or detached from.
Parameters
----------
object: object
Object provided by the ``iter_objects`` methods from another
observers or directly by the user.
Yields
------
IObservable
Raises
------
ValueError
If the given object cannot be handled by this observer.
"""
if not object_has_named_trait(object, "trait_added"):
if self.optional:
return
raise ValueError(
"Unable to observe 'trait_added' event on {!r}".format(object))
yield object._trait("trait_added", 2)
def iter_objects(self, object):
""" Yield objects for the next observer following this observer, in an
ObserverGraph.
Parameters
----------
object: object
Object provided by the ``iter_objects`` methods from another
observers or directly by the user.
Yields
------
value : object
"""
# children graphs are used in the maintainer only.
yield from ()
# get_notifier is not implemented as notify is always false.
def get_maintainer(self, graph, handler, target, dispatcher):
""" Return a notifier for maintaining downstream observers when
trait_added event happens.
Parameters
----------
graph : ObserverGraph
Description for the *downstream* observers, i.e. excluding self.
handler : callable
User handler.
target : object
Object seen by the user as the owner of the observer.
dispatcher : callable
Callable for dispatching the handler.
Returns
-------
notifier : ObserverChangeNotifier
"""
return ObserverChangeNotifier(
observer_handler=self.observer_change_handler,
event_factory=trait_event_factory,
prevent_event=self.prevent_event,
graph=graph,
handler=handler,
target=target,
dispatcher=dispatcher,
)
def prevent_event(self, event):
""" Return true if the added trait should not be handled.
Parameters
----------
event : TraitChangeEvent
Event triggered by add_trait.
"""
object = event.object
name = event.new
trait = object.trait(name=name)
return not self.match_func(name, trait)
@staticmethod
def observer_change_handler(event, graph, handler, target, dispatcher):
""" Handler for maintaining observers.
Parameters
----------
event : TraitChangeEvent
Event triggered by add_trait.
graph : ObserverGraph
Description for the *downstream* observers, i.e. excluding self.
handler : callable
User handler.
target : object
Object seen by the user as the owner of the observer.
dispatcher : callable
Callable for dispatching the handler.
"""
new_graph = ObserverGraph(
node=_RestrictedNamedTraitObserver(
name=event.new,
wrapped_observer=graph.node,
),
children=graph.children
)
add_or_remove_notifiers(
object=event.object,
graph=new_graph,
handler=handler,
target=target,
dispatcher=dispatcher,
remove=False,
)
# There is no mirrored action to this.
# ``remove_trait`` does not fire a change event.
# see enthought/traits#1047
def iter_extra_graphs(self, graph):
""" Yield new ObserverGraph to be contributed by this observer.
Parameters
----------
graph : ObserverGraph
The graph this observer is part of.
Yields
------
ObserverGraph
"""
yield from ()
@IObserver.register
class _RestrictedNamedTraitObserver:
""" An observer to support TraitAddedObserver in order to add
notifiers for one specific named trait. The notifiers should be
contributed by the original observer.
Parameters
----------
name : str
Name of the trait to be observed.
wrapped_observer : IObserver
The observer from which notifers are obtained.
"""
def __init__(self, name, wrapped_observer):
self.name = name
self._wrapped_observer = wrapped_observer
def __hash__(self):
return hash((type(self), self.name, self._wrapped_observer))
def __eq__(self, other):
return (
type(self) is type(other)
and self.name == other.name
and self._wrapped_observer == other._wrapped_observer
)
@property
def notify(self):
""" A boolean for whether this observer will notify for changes. """
return self._wrapped_observer.notify
def iter_observables(self, object):
""" Yield only the observable for the named trait."""
yield object._trait(self.name, 2)
def iter_objects(self, object):
""" Yield only the value for the named trait."""
yield from iter_objects(object, self.name)
def get_notifier(self, handler, target, dispatcher):
""" Return the notifier from the wrapped observer."""
return self._wrapped_observer.get_notifier(handler, target, dispatcher)
def get_maintainer(self, graph, handler, target, dispatcher):
""" Return the maintainer from the wrapped observer."""
return self._wrapped_observer.get_maintainer(
graph, handler, target, dispatcher)
def iter_extra_graphs(self, graph):
yield from ()
|
