/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
|