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
|
# (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._i_observer import IObserver
from traits.observation._observe import add_or_remove_notifiers
from traits.observation._observer_change_notifier import ObserverChangeNotifier
from traits.observation._set_change_event import set_event_factory
from traits.observation._trait_event_notifier import TraitEventNotifier
from traits.trait_set_object import TraitSet
@IObserver.register
class SetItemObserver:
""" Observer for observing mutations on a set.
Parameters
----------
notify : boolean
Whether to notify for changes.
optional : boolean
If False, this observer will complain if the incoming object is not
an observable set. If True and the incoming object is not a set, this
observer will do nothing. Useful for the 'items' keyword in the text
parser, where the source container type is ambiguous.
"""
__slots__ = ("notify", "optional")
def __init__(self, *, notify, optional):
self.notify = notify
self.optional = optional
def __hash__(self):
""" Return a hash of this object."""
return hash((type(self).__name__, self.notify, 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.notify == other.notify
and self.optional == other.optional
)
def __repr__(self):
formatted_args = [
f"notify={self.notify!r}",
f"optional={self.optional!r}",
]
return f"{self.__class__.__name__}({', '.join(formatted_args)})"
def iter_observables(self, object):
""" If the given object is an observable set, yield that set.
Otherwise, raise an error, unless this observer is optional
Parameters
----------
object: object
Object provided by another observers or by the user.
Yields
------
IObservable
Raises
------
ValueError
If the given object is not an observable set and optional is false.
"""
if not isinstance(object, TraitSet):
if self.optional:
return
raise ValueError(
"Expected a TraitSet to be observed, "
"got {!r} (type: {!r})".format(object, type(object))
)
yield object
def iter_objects(self, object):
""" Yield the content of the set if the given object is an observable
set. Otherwise, raise an error, unless the observer is optional.
The content of the set will be passed onto the children observer(s)
following this one in an ObserverGraph.
Parameters
----------
object: object
Object provided by another observers or by the user.
Yields
------
value : object
Raises
------
ValueError
If the given object is not an observable set and optional is false.
"""
if not isinstance(object, TraitSet):
if self.optional:
return
raise ValueError(
"Expected a TraitSet to be observed, "
"got {!r} (type: {!r})".format(object, type(object))
)
yield from object
def get_notifier(self, handler, target, dispatcher):
""" Return a notifier for calling the user handler with the change
event.
Returns
-------
notifier : TraitEventNotifier
"""
return TraitEventNotifier(
handler=handler,
target=target,
dispatcher=dispatcher,
event_factory=set_event_factory,
prevent_event=lambda event: False,
)
def get_maintainer(self, graph, handler, target, dispatcher):
""" Return a notifier for maintaining downstream observers when
a set is mutated.
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=_observer_change_handler,
event_factory=set_event_factory,
prevent_event=lambda event: False,
graph=graph,
handler=handler,
target=target,
dispatcher=dispatcher,
)
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
"""
# Unlike CTrait, no need to handle trait_added
yield from ()
def _observer_change_handler(event, graph, handler, target, dispatcher):
""" Handler for maintaining observers. Used by ObserverChangeNotifier.
Parameters
----------
event : SetChangeEvent
Change event that triggers the maintainer.
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.
"""
for removed_item in event.removed:
add_or_remove_notifiers(
object=removed_item,
graph=graph,
handler=handler,
target=target,
dispatcher=dispatcher,
remove=True,
)
for added_item in event.added:
add_or_remove_notifiers(
object=added_item,
graph=graph,
handler=handler,
target=target,
dispatcher=dispatcher,
remove=False,
)
|
