Use pydbus library to send signal over a Session Bus

Question

I am working with pydbus and I have already succeed in using it to listen signals on the session bus (on the "client side"). I would like to write the server side now where the program sends a signal everytime an action is triggered (for example when it writes a file on the FS). I don't really get any example of that on their GitHub. They only show how to write a basic server who has a bunch of methods the client can call (but it is not about signals which is what I want).

FYI, the client side is very simple and looks like that:

# Subscribe to DBus dump signal
session_bus = SessionBus()
session_bus.subscribe(iface='my.iface',
                      signal='my_signal_name',
                      object='/my/object',
                      signal_fired=my_callback)

# Create and run main loop
loop = GObject.MainLoop()
loop.run()

Where my_callback is the method called everytime the listened event pops (my_signal_name in this case)

Thx for help.

Answer 1

The following python3 / pydbus program will publish a random integer every second onto the session D-Bus.

#!/usr/bin/env python3
#!
from pydbus.generic import signal
from pydbus import SessionBus
from gi.repository import GLib
import random
loop = GLib.MainLoop()
interface_name = "org.example.project_1.server_1"
bus = SessionBus()


class Server_XML(object):
    """
    Server_XML definition.
    Emit / Publish a signal that is a random integer every second 
    type='i' for integer. 
    """
    dbus = """
    <node>
        <interface name="org.example.project_1.server_1">
            <signal name="app_1_signal">
                <arg type='i'/>
            </signal>
        </interface>
    </node>
    """
    app_1_signal = signal()

def repeating_timer():
    """Generate random integer between 0 and 100 and emit over Session D-Bus
    return True to keep the GLib timer calling this function once a second."""
    random_integer = random.randint(0,100)
    print(random_integer)
    emit.app_1_signal(random_integer)
    return True


if __name__=="__main__":
    # Setup server to emit signals over the DBus
    emit = Server_XML()
    bus.publish(interface_name, emit)
    # Setup repeating timer
    GLib.timeout_add_seconds(interval=1, function=repeating_timer) 
    # Run loop with graceful ctrl C exiting.
    try:
        loop.run()
    except KeyboardInterrupt as e:
        loop.quit()
        print("\nExit by Control C")

Open another linux console terminal and use the gdbus utility to verify that integers are being published on the session bus. For example...

$ gdbus monitor --session --dest org.example.project_1.server_1
Monitoring signals from all objects owned by org.example.project_1.server_1
The name org.example.project_1.server_1 is owned by :1.618
/org/example/project_1/server_1: org.example.project_1.server_1.app_1_signal (36,)
/org/example/project_1/server_1: org.example.project_1.server_1.app_1_signal (37,)
/org/example/project_1/server_1: org.example.project_1.server_1.app_1_signal (25,)

The following python3 / pydbus client program subscribes to the random integers being published on the session D-Bus...

#!/usr/bin/env python3
#!
from pydbus.generic import signal
from pydbus import SessionBus
from gi.repository import GLib
loop = GLib.MainLoop()
dbus_filter = "/org/example/project_1/server_1"
bus = SessionBus()


def cb_server_signal_emission(*args):
    """
    Callback on emitting signal from server
    """
    print("Message: ", args)
    print("Data: ", str(args[4][0]))

if __name__=="__main__":
    # Subscribe to bus to monitor for server signal emissions
    bus.subscribe(object = dbus_filter, 
                  signal_fired = cb_server_signal_emission)    
    loop.run()

The console output from this client program will be similar to this...

$ python3 client_pydbus.py
Message:  (':1.621', '/org/example/project_1/server_1', 'org.example.project_1.server_1', 'app_1_signal', (72,))
Data:  72
Message:  (':1.621', '/org/example/project_1/server_1', 'org.example.project_1.server_1', 'app_1_signal', (46,))
Data:  46
Message:  (':1.621', '/org/example/project_1/server_1', 'org.example.project_1.server_1', 'app_1_signal', (47,))
Data:  47

Answer 2

The class responsible for signals is located in the generic module. It looks well enough documentated:

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.

There is also an example in the tutorial which uses a signal. Note the last line and the property SomeProperty. When the python property SomeProperty is changed in the setter, the signal is emitted via self.PropertiesChanged("net.lew21.pydbus.TutorialExample", {"SomeProperty": self.SomeProperty}, []).

from pydbus.generic import signal

class Example(object):
  """
    <node>
      <interface name='net.lew21.pydbus.TutorialExample'>
        <method name='EchoString'>
          <arg type='s' name='a' direction='in'/>
          <arg type='s' name='response' direction='out'/>
        </method>
        <property name="SomeProperty" type="s" access="readwrite">
          <annotation name="org.freedesktop.DBus.Property.EmitsChangedSignal" value="true"/>
        </property>
      </interface>
    </node>
  """

  def EchoString(self, s):
    """returns whatever is passed to it"""
    return s

  def __init__(self):
    self._someProperty = "initial value"

  @property
  def SomeProperty(self):
    return self._someProperty

  @SomeProperty.setter
  def SomeProperty(self, value):
    self._someProperty = value
    self.PropertiesChanged("net.lew21.pydbus.TutorialExample", {"SomeProperty": self.SomeProperty}, [])

  PropertiesChanged = signal()

There is also a notification_server example where signals are used (but not called).