/usr/lib/python3/dist-packages/pydbus/generic.py is in python3-pydbus 0.6.0-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 | """Generic programming utilities.
Utilities implemented in this file are not dependent
on dbus, they can be used everywhere.
"""
class subscription(object):
__slots__ = ("callback_list", "callback")
def __init__(self, callback_list, callback):
self.callback_list = callback_list
self.callback = callback
self.callback_list.append(callback)
def unsubscribe(self):
self.callback_list.remove(self.callback)
self.callback_list = None
self.callback = None
def disconnect(self):
"""An alias for unsubscribe()"""
self.unsubscribe()
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
if not self.callback is None:
self.unsubscribe()
class bound_signal(object):
__slots__ = ("__signal__", "__self__") # bound method uses ("__func__", "__self__")
def __init__(self, signal, instance):
self.__signal__ = signal
self.__self__ = instance
@property
def callbacks(self):
return self.__signal__.map[self.__self__]
def connect(self, callback):
"""Subscribe to the signal."""
return self.__signal__.connect(self.__self__, callback)
def emit(self, *args):
"""Emit the signal."""
self.__signal__.emit(self.__self__, *args)
def __call__(self, *args):
"""Emit the signal."""
self.emit(*args)
def __repr__(self):
return "<bound signal " + self.__signal__.__qualname__ + " of " + repr(self.__self__) + ">"
class signal(object):
"""Static signal object
You're expected to set it as a class property::
class A:
SomethingHappened = signal()
Declared this way, it can be used on class instances
to connect signal observers::
a = A()
a.SomethingHappened.connect(func)
and emit the signal::
a.SomethingHappened()
You may pass any parameters to the emiting function
- they will be forwarded to all subscribed callbacks.
"""
def __init__(self):
self.map = {}
self.__qualname__ = "<anonymous signal>" # function uses <lambda> ;)
self.__doc__ = "Signal."
def connect(self, object, callback):
"""Subscribe to the signal."""
return subscription(self.map.setdefault(object, []), callback)
def emit(self, object, *args):
"""Emit the signal."""
for cb in self.map.get(object, []):
cb(*args)
def __get__(self, instance, owner):
if instance is None:
return self
return bound_signal(self, instance)
def __set__(self, instance, value):
raise AttributeError("can't set attribute")
def __repr__(self):
return "<signal " + self.__qualname__ + " at 0x" + format(id(self), "x") + ">"
bound_method = type(signal().emit) # TODO find a prettier way to get this type
|