本文介绍了简单(但特定)的侦听器和发送器Python 3 DBus示例的处理方法,对大家解决问题具有一定的参考价值,需要的朋友们下面随着小编来一起学习吧!

问题描述

我想制作一个包含两个部分的程序.侦听器(如果需要,则为服务器)和发送者(客户端).我进行了一些研究,得知这是通过程序员称为IPC(进程间通信)的方法完成的.我确定您知道这是什么意思,我只是在扩展首字母缩写词,以便您知道我不认为这意味着Internet Pet Cannibals(或其他不相关的不愉快事物).

I want to make a program that has two parts. A listener (a server, if you will) and a sender (the client). I did some research and learned that this is done via a method programmers call IPC (inter process communication); I'm sure you know what it means, I am just expanding the acronym so that you know that I don't think it means Internet Pet Cannibals (or some other non related unpleasant thing).

我读到,实现此目标的一种好方法是使用dbus.所以我对dbus进行了一些研究,但现在我很困惑.显然,使用dbus可以做很多事情,例如向Gnome Shell发送通知或与网络管理器对话.我不想做那些事!我只想编写两个相互交流的简单程序.另外,一些教程和文档显示了python 2的示例,一些使用3,一些导入dbus和一些导入Gio!我发现的很多信息不胜枚举,这也损害了我的努力.

I read that a good way to achieve this is to use dbus. So I did some research on dbus, and now I'm just confused. Apparently there are a lot of things you can do with dbus, like send notifications to Gnome Shell or talk with the Network Manager. I don't want to do those things! I just want to make two simple programs that talk to each other. To add to that, some tutorials and documentation show examples with python 2, some use 3, some import dbus and some import Gio! A lot of the information I have found is over my head which also impairs my efforts.

有人会很友好地向我展示一个简单而优雅的示例,说明如何实现一个本质上可以做到这一点的程序:

Would someone be so kind as to show me a simple, elegant example on how to achieve making a program that essentially does this:

$ ./server
Server is not running yet. Putting on listening ears.
$ ./client Hi
server: a client said "Hi"
$ ./server
Server is already running.
$ ./server stop
Server exiting...
$ ./client Do a barrel roll
client: No one can hear me!!

这是一个简单的会话的进行方式(当然使用bash shell).我想使用Python 3,以及到目前为止最合适的dbus绑定(我想那应该是gi.repository).要澄清的是,这将适用于Linux.

This is how a simple session would go (using a bash shell of course). I would want to use Python 3 and whatever dbus bindings are most appropriate as of now (I am guessing that would be gi.repository). To clarify, this would be for Linux.

推荐答案

在python3中,关于dbus的文档并不多,但是我设法弄清楚了,因此我将在此处进行记录:与所有的python2示例都将import gobject替换为import gi.repository.GLib.

There's not a lot of documentation for dbus in python3, but I managed to figure it out so I'll document it here: The major difference from all the python2 examples is replacing import gobject with import gi.repository.GLib.

您可以在 dbus-python示例目录.

我没有在服务器中实现自我后台,因为该守护程序样式最近已经过时了.

I didn't implement self-backgrounding in the server because that style of daemon has gone out of style recently.

common.py:

common.py:

# well-known name for our program
ECHO_BUS_NAME = 'com.stackoverflow.question_21793826.EchoService'

# interfaces implemented by some objects in our program
ECHO_INTERFACE = 'com.stackoverflow.question_21793826.EchoInterface'
QUIT_INTERFACE = 'com.stackoverflow.question_21793826.QuitInterface'

# paths to some objects in our program
ECHO_OBJECT_PATH = '/EchoServerObject'

server.py:

server.py:

#!/usr/bin/env python3

# standard includes
import sys

# dbus includes
import gi.repository.GLib
import dbus
import dbus.service
import dbus.mainloop.glib

# project includes
import common


class EchoServerObject(dbus.service.Object):

    # TODO it would be nice to make a better decorator using annotations:
    #   def foo(self, a: 's', b: 's') -> '': pass
    # but the existing dbus decorator does its own reflection which
    # fails if there are any annotations (or keyword-only arguments)
    @dbus.service.method(common.ECHO_INTERFACE,
            in_signature='s', out_signature='')
    def echo(self, message):
        message = str(message) # get rid of subclass for repr
        print('server: a client said %r' % message)

    @dbus.service.method(common.QUIT_INTERFACE,
            in_signature='', out_signature='')
    def quit(self):
        # this should be a separate object, but I'm
        # showing how one object can have multiple interfaces
        self.mainloop.quit()

def stop():
    bus = dbus.SessionBus()

    proxy = bus.get_object(common.ECHO_BUS_NAME, common.ECHO_OBJECT_PATH)
    iface = dbus.Interface(proxy, common.QUIT_INTERFACE)

    iface.quit()

def server():
    dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)

    bus = dbus.SessionBus()
    try:
        name = dbus.service.BusName(common.ECHO_BUS_NAME, bus, do_not_queue=True)
    except dbus.NameExistsException:
        sys.exit('Server is already running.')
    else:
        print('Server is not running yet. Putting on listening ears.')
    echo = EchoServerObject(bus, common.ECHO_OBJECT_PATH)

    mainloop = gi.repository.GLib.MainLoop()
    echo.mainloop = mainloop
    mainloop.run()

def main(exe, args):
    if args == ['stop']:
        stop()
    elif not args:
        server()
    else:
        sys.exit('Usage: %s [stop]' % exe)

if __name__ == '__main__':
    main(sys.argv[0], sys.argv[1:])

client.py:

client.py:

#!/usr/bin/env python3

# standard includes
import sys

# dbus includes
import dbus

# project includes
import common


def client(mes):
    bus = dbus.SessionBus()

    try:
        proxy = bus.get_object(common.ECHO_BUS_NAME, common.ECHO_OBJECT_PATH)
    except dbus.DBusException as e:
        # There are actually two exceptions thrown:
        # 1: org.freedesktop.DBus.Error.NameHasNoOwner
        #   (when the name is not registered by any running process)
        # 2: org.freedesktop.DBus.Error.ServiceUnknown
        #   (during auto-activation since there is no .service file)
        # TODO figure out how to suppress the activation attempt
        # also, there *has* to be a better way of managing exceptions
        if e._dbus_error_name != 'org.freedesktop.DBus.Error.ServiceUnknown':
            raise
        if e.__context__._dbus_error_name != 'org.freedesktop.DBus.Error.NameHasNoOwner':
            raise
        print('client: No one can hear me!!')
    else:
        iface = dbus.Interface(proxy, common.ECHO_INTERFACE)
        iface.echo(mes)

def main(exe, args):
    if args:
        client(' '.join(args))
    else:
        sys.exit('Usage: %s message...' % exe)

if __name__ == '__main__':
    main(sys.argv[0], sys.argv[1:])

这篇关于简单(但特定)的侦听器和发送器Python 3 DBus示例的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!

09-26 12:35