/usr/lib/python3/dist-packages/pydbus/subscription.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 | from gi.repository import Gio
from .exitable import ExitableWithAliases
class Subscription(ExitableWithAliases("unsubscribe", "disconnect")):
Flags = Gio.DBusSignalFlags
__slots__ = ()
def __init__(self, con, sender, iface, member, object, arg0, flags, callback):
id = con.signal_subscribe(sender, iface, member, object, arg0, flags, callback)
self._at_exit(lambda: con.signal_unsubscribe(id))
class SubscriptionMixin(object):
__slots__ = ()
SubscriptionFlags = Subscription.Flags
def subscribe(self, sender=None, iface=None, signal=None, object=None, arg0=None, flags=0, signal_fired=None):
"""Subscribes to matching signals.
Subscribes to signals on connection and invokes signal_fired callback
whenever the signal is received.
To receive signal_fired callback, you need an event loop.
https://github.com/LEW21/pydbus/blob/master/doc/tutorial.rst#setting-up-an-event-loop
Parameters
----------
sender : string, optional
Sender name to match on (unique or well-known name) or None to listen from all senders.
iface : string, optional
Interface name to match on or None to match on all interfaces.
signal : string, optional
Signal name to match on or None to match on all signals.
object : string, optional
Object path to match on or None to match on all object paths.
arg0 : string, optional
Contents of first string argument to match on or None to match on all kinds of arguments.
flags : SubscriptionFlags, optional
signal_fired : callable, optional
Invoked when there is a signal matching the requested data.
Parameters: sender, object, iface, signal, params
Returns
-------
Subscription
An object you can use as a context manager to unsubscribe from the signal later.
See Also
--------
See https://developer.gnome.org/gio/2.44/GDBusConnection.html#g-dbus-connection-signal-subscribe
for more information.
"""
callback = (lambda con, sender, object, iface, signal, params: signal_fired(sender, object, iface, signal, params.unpack())) if signal_fired is not None else lambda *args: None
return Subscription(self.con, sender, iface, signal, object, arg0, flags, callback)
|