Python Forum

Full Version: How to transmit a dict with dbus and QtDBus
You're currently viewing a stripped down version of our content. View the full version with proper formatting.
I came up with that code to send a string with dbus to QtDBus:

client.py
#!/usr/bin/python3
import dbus
try:
    bus = dbus.SessionBus().get_object("com.qtpad.dbus", "/cli")
    busIO = dbus.Interface(bus, "com.qtpad.dbus")
    busIO.echo("test")
    #reply = busIO.echo("test")
    #print(reply)
except dbus.exceptions.DBusException:
    print("Could not connect")
server.py
#!/usr/bin/python3
from PyQt5 import QtCore, QtDBus
from PyQt5.QtCore import QObject, pyqtSlot

class QDBusServer(QObject):
    def __init__(self):
        QObject.__init__(self)
        self.__dbusAdaptor = QDBusServerAdapter(self)

    #def echo(self, value):
    #    return 'Received: ' + value

class QDBusServerAdapter(QtDBus.QDBusAbstractAdaptor):
    QtCore.Q_CLASSINFO("D-Bus Interface", "com.qtpad.dbus")
    QtCore.Q_CLASSINFO("D-Bus Introspection",
    '  <interface name="com.qtpad.dbus">\n'
    '    <property name="name" type="s" access="read"/>\n'
    '    <method name="echo">\n'
    '      <arg direction="in" type="s" name="phrase"/>\n'
    '    </method>\n'
    '  </interface>\n')

    def __init__(self, parent):
        super().__init__(parent)

    @pyqtSlot(str, result=str)
    def echo(self, phrase):
        print("parse(" + phrase + ")")
        #return self.parent().echo(phrase)

if __name__ == '__main__':
    app = QtCore.QCoreApplication()
    bus = QtDBus.QDBusConnection.sessionBus()
    server = QDBusServer()
    bus.registerObject('/cli', server)
    bus.registerService('com.qtpad.dbus')
    app.exec()
To implement a command line interface, I want to use dbus with argparse module to send the command to the running instance (if any), then exit without loading the gui. The output of argparse is a namespace object, which I converted to a dict using vars().

What I am actually trying to do:
#!/usr/bin/python3
from PyQt5 import QtCore, QtDBus
from PyQt5.QtCore import QObject, pyqtSlot
import argparse
import dbus
import sys


class QDBusServer(QObject):
    def __init__(self):
        QObject.__init__(self)
        self.__dbusAdaptor = QDBusServerAdapter(self)


class QDBusServerAdapter(QtDBus.QDBusAbstractAdaptor):
    QtCore.Q_CLASSINFO("D-Bus Interface", "com.qtpad.dbus")
    QtCore.Q_CLASSINFO("D-Bus Introspection",
    '  <interface name="com.qtpad.dbus">\n'
    '    <property name="name" variant_level="DICT_ENTRY" type="o" access="read"/>\n'
    '    <method name="parse">\n'
    '      <arg direction="in" type="o" name="cmd"/>\n'
    '    </method>\n'
    '  </interface>\n')

    def __init__(self, parent):
        super().__init__(parent)

    @pyqtSlot(object, result=object)
    def parse(self, cmd):
        parse(cmd)


def parse(args):
    print(args)
    #if args.action:
    #    print("got: " + args.action)

if __name__ == '__main__':
    try:
        bus = dbus.SessionBus().get_object("com.qtpad.dbus", "/cli")
        busIO = dbus.Interface(bus, "com.qtpad.dbus")
        connected = True
    except dbus.exceptions.DBusException:
        connected = False

    if not connected:
        # Start the whole app
        print("Init of a new instance")
        app = QtCore.QCoreApplication()
        bus = QtDBus.QDBusConnection.sessionBus()
        server = QDBusServer()
        bus.registerObject('/cli', server)
        bus.registerService('com.qtpad.dbus')
        app.exec()

    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--action", default="None", help="execute one of the predefined action, as shown in the preferences menu", metavar='')
    args = parser.parse_args()
    if connected:
        # Convert namespace to dict, then send the command to the running instance
        cmd = vars(args)
        print("Sent " + str(cmd))
        busIO.parse(cmd)
        sys.exit(0)
    else:
        parse(args)
The result:
Error:
ERROR:dbus.connection:Unable to set arguments ({'action': 'Open new note'},) according to signature 'o': <class 'TypeError'>: Expected a string or unicode object Traceback (most recent call last):   File "/home/will/scripts/qtpad/cli.py", line 63, in <module>     busIO.parse(cmd)   File "/usr/lib/python3.6/site-packages/dbus/proxies.py", line 70, in __call__     return self._proxy_method(*args, **keywords)   File "/usr/lib/python3.6/site-packages/dbus/proxies.py", line 145, in __call__     **keywords)   File "/usr/lib/python3.6/site-packages/dbus/connection.py", line 641, in call_blocking     message.append(signature=signature, *args) TypeError: Expected a string or unicode object
I tried to replace type="s" to e, o... without success, and I have a hard time understanding dbus documentation. Is it even possible to transmit a dict, how should I rather send it as a string and then parse it to recreate the object?
I could get it to work by converting the dict to a string, send it through dbus, then convert it back to a dict. Is there a better way to do this?

#!/usr/bin/python3
from PyQt5 import QtCore, QtDBus
from PyQt5.QtCore import QObject, pyqtSlot
import argparse
import dbus
import ast
import sys


class QDBusServer(QObject):
    def __init__(self):
        QObject.__init__(self)
        self.__dbusAdaptor = QDBusServerAdapter(self)


class QDBusServerAdapter(QtDBus.QDBusAbstractAdaptor):
    QtCore.Q_CLASSINFO("D-Bus Interface", "com.qtpad.dbus")
    QtCore.Q_CLASSINFO("D-Bus Introspection",
    '  <interface name="com.qtpad.dbus">\n'
    '    <property name="name" type="s" access="read"/>\n'
    '    <method name="parse">\n'
    '      <arg direction="in" type="s" name="cmd"/>\n'
    '    </method>\n'
    '  </interface>\n')

    def __init__(self, parent):
        super().__init__(parent)

    @pyqtSlot(str, result=str)
    def parse(self, cmd):
        parseCli(cmd)


def parseCli(args):
    # Convert the string back to a dict
    args = ast.literal_eval(args)
    for key in args:
        if args[key]:
            print(key + "=" + str(args[key]))

if __name__ == '__main__':
    # Handle command line input
    parser = argparse.ArgumentParser()
    parser.add_argument("-a", "--action", help="execute one of the predefined action, as shown in the preferences menu", metavar='')
    args = parser.parse_args()

    # Convert namespace to dict, then dict to string
    cmd = vars(args)
    cmd = str(cmd)

    # Verify if an instance is already running
    try:
        bus = dbus.SessionBus().get_object("com.qtpad.dbus", "/cli")
        busIO = dbus.Interface(bus, "com.qtpad.dbus")
        # Send the command to the running instance, then quit
        busIO.parse(cmd)
        sys.exit(0)
    except dbus.exceptions.DBusException:
        print("Init of a new instance")

    # Start a new instance
    app = QtCore.QCoreApplication([])
    bus = QtDBus.QDBusConnection.sessionBus()
    server = QDBusServer()
    bus.registerObject('/cli', server)
    bus.registerService('com.qtpad.dbus')
    parseCli(cmd)
    app.exec()