This file is indexed.

/usr/lib/python3/dist-packages/aeidon/observable.py is in python3-aeidon 0.24.3-1.

This file is owned by root:root, with mode 0o644.

The actual contents of the file can be viewed below.

  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
# -*- coding: utf-8 -*-

# Copyright (C) 2005-2009 Osmo Salomaa
#
# This file is part of Gaupol.
#
# Gaupol is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# Gaupol is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# Gaupol. If not, see <http://www.gnu.org/licenses/>.

"""Base class for observable objects."""

import aeidon

__all__ = ("Observable",)


class Observable(object, metaclass=aeidon.Contractual):

    """
    Base class for observable objects.

    :cvar signals: Tuple of emittable signals added automatically

    In addition to the signals defined in :attr:`signals`, all public instance
    variables have a ``notify::NAME`` signal generated automatically based on
    the ``NAME`` of the variable. ``notify::NAME`` signals will be emitted
    whenever the value of the corresponding instance variable changes.

    Notify signals will be emitted for mutable variables as well, which means
    that care should be taken not to emit thousands of signals when e.g.
    appending one-by-one to a large list. :meth:`freeze_notify` and
    :meth:`thaw_notify` will queue notify signals and emit only one of each
    once thawed.

    The Observable philosophy and API is highly inspired by GObject_.

    .. _GObject: http://developer.gnome.org/gobject/
    """

    __slots__ = ("_blocked_signals",
                 "_blocked_state",
                 "_notify_frozen",
                 "_notify_queue",
                 "_signal_handlers",)

    signals = ()

    def __init__(self):
        """Initialize an :class:`Observable` object."""
        self._blocked_signals = []
        self._blocked_state = False
        self._notify_frozen = False
        self._notify_queue = []
        self._signal_handlers = {}
        for signal in self.signals:
            self._add_signal(signal)

    def __setattr__(self, name, value):
        """Set value of observable attribute."""
        if (name in self.__slots__) or name.startswith("_"):
            return object.__setattr__(self, name, value)
        value = self._validate(name, value)
        signal = "notify::{}".format(name)
        if not signal in self._signal_handlers:
            self._add_signal(signal)
            return object.__setattr__(self, name, value)
        return_value = object.__setattr__(self, name, value)
        self.emit(signal, value)
        return return_value

    def _add_signal_require(self, signal):
        assert not signal in self._signal_handlers

    def _add_signal(self, signal):
        """Add `signal` to the list of signals emitted."""
        self._signal_handlers[signal] = []

    def _validate_ensure(self, return_value, name, value):
        assert return_value == value

    def _validate(self, name, value):
        """Return `value` or an observable version if `value` is mutable."""
        args = (value, self, name)
        if isinstance(value, dict):
            return aeidon.ObservableDict(*args)
        if isinstance(value, list):
            return aeidon.ObservableList(*args)
        if isinstance(value, set):
            return aeidon.ObservableSet(*args)
        return value

    def block_require(self, signal):
        assert signal in self._signal_handlers

    def block(self, signal):
        """
        Block all emissions of `signal`.

        Return ``False`` if already blocked, otherwise ``True``.
        """
        if not signal in self._blocked_signals:
            self._blocked_signals.append(signal)
            return True
        return False

    def block_all(self):
        """
        Block all emissions of all signals.

        Return ``False`` if already blocked, otherwise ``True``.
        """
        if not self._blocked_state:
            self._blocked_state = True
            return True
        return False

    def connect_require(self, signal, method, *args):
        assert signal in self._signal_handlers
        assert callable(method)

    def connect(self, signal, method, *args):
        """Register to receive notifications of ``signal``."""
        self._signal_handlers[signal].append((method, args))

    def disconnect_require(self, signal, method):
        assert signal in self._signal_handlers

    def disconnect(self, signal, method):
        """Remove registration to receive notifications of ``signal``."""
        for i in reversed(range(len(self._signal_handlers[signal]))):
            if self._signal_handlers[signal][i][0] == method:
                self._signal_handlers[signal].pop(i)

    def emit_require(self, signal, *args):
        assert signal in self._signal_handlers

    def emit(self, signal, *args):
        """Send notification of ``signal`` to all registered observers."""
        if signal.startswith("notify::") and self._notify_frozen:
            if not signal in self._notify_queue:
                self._notify_queue.append(signal)
            return
        if (not self._blocked_state and
            not signal in self._blocked_signals):
            if signal.startswith("notify::"):
                name = signal.replace("notify::", "")
                args = (getattr(self, name),)
            for method, data in self._signal_handlers[signal]:
                method(*((self,) + args + data))

    def freeze_notify(self):
        """
        Queue notify signals instead of emitting them.

        Return ``False`` if already frozen, otherwise ``True``.
        """
        if not self._notify_frozen:
            self._notify_frozen = True
            return True
        return False

    def notify_require(self, name):
        assert hasattr(self, name)

    def notify(self, name):
        """Emit notification signal for variable."""
        return self.emit("notify::{}".format(name))

    def thaw_notify_ensure(self, value, do=True):
        assert not do or not self._notify_queue

    def thaw_notify(self, do=True):
        """
        Emit all queued notify signals and queue no more.

        The optional `do` keyword argument should be the return value from
        :meth:`freeze_notify` to avoid problems with nested functions where
        notifications were frozen at a higher level. If `do` is ``False``,
        nothing will be done.

        Return ``False`` if already thawed, otherwise ``True``.
        """
        if do and self._notify_frozen:
            self._notify_frozen = False
            for signal in self._notify_queue:
                name = signal.replace("notify::", "")
                self.emit(signal, getattr(self, name))
            self._notify_queue = []
            return True
        return False

    def unblock_require(self, signal, do=True):
        assert signal in self._signal_handlers

    def unblock(self, signal, do=True):
        """
        Unblock all emissions of `signal`.

        The optional `do` keyword argument should be the return value from
        :meth:`block` to avoid problems with nested functions where signals
        were blocked at a higher level. If `do` is ``False``, nothing will be
        done.

        Return ``False`` if already unblocked, otherwise ``True``.
        """
        if do and (signal in self._blocked_signals):
            self._blocked_signals.remove(signal)
            return True
        return False

    def unblock_all(self, do=True):
        """
        Unblock all emissions of all signals.

        The optional `do` keyword argument should be the return value from
        :meth:`block_all` to avoid problems with nested functions where signals
        were blocked at a higher level. If `do` is ``False``, nothing will be
        done.

        Return ``False`` if already unblocked, otherwise ``True``.
        """
        if do and self._blocked_state:
            self._blocked_state = False
            return True
        return False