在python中动态创建DBus信号

Posted 2023-04-14

【中文标题】在python中动态创建DBus信号

【英文标题】：Create DBus signal dynamically in python

【发布时间】：2012-06-11 05:56:54

【问题描述】：

我已经阅读了一些与动态创建 python 方法相关的主题，并按照他们的说明进行操作，但它不起作用。不知道是不是因为我用了装饰器@还是别的什么的。

代码在这里，很简单。

运行此代码时，没有出现错误，但是当我使用D-feet（检查dbus信息的工具）时，我找不到我创建的新信号。

#!/usr/bin/python

import dbus
import dbus.service
import dbus.glib
import gobject
from dbus.mainloop.glib import DBusGMainLoop

import psutil

class EventServer(dbus.service.Object):
    i = 0

    @dbus.service.signal('com.github.bxshi.event')
    def singal_example(self,msg):
        """ example of singals
        """
        print msg

    def __init__(self):
        bus_name = dbus.service.BusName('com.github.bxshi.event', bus=dbus.SessionBus())
        dbus.service.Object.__init__(self, bus_name, '/com/github/bxshi/event')

    def create(self):
        self.i +=1
        setattr(self.__class__, 'signal_'+str(self.i), self.singal_example)


if __name__ == "__main__":
    DBusGMainLoop(set_as_default=True)
    bus = dbus.SessionBus()
    eventserver = EventServer()
    gobject.timeout_add(1000,eventserver.create)
    loop = gobject.MainLoop()
    loop.run() 

【参考方案1】：

你有一个错字：singal_example 而不是signal_example 在您的create 方法中，您在课堂上调用setattr。我不知道你想做什么，但你应该简单地发出信号

这是固定的例子：

#!/usr/bin/python

import dbus
import dbus.service
import dbus.glib
import gobject
from dbus.mainloop.glib import DBusGMainLoop

#import psutil

class EventServer(dbus.service.Object):
    i = 0

    @dbus.service.signal('com.github.bxshi.event')
    def signal_example(self,msg):
        """ example of singals
        """
        print msg

    def __init__(self):
        bus_name = dbus.service.BusName('com.github.bxshi.event', bus=dbus.SessionBus())
        dbus.service.Object.__init__(self, bus_name, '/com/github/bxshi/event')

    def create(self):
        self.i +=1
        #setattr(self.__class__, 'signal_'+str(self.i), self.singal_example)
        self.signal_example('msg: %d' % self.i)


if __name__ == "__main__":
    DBusGMainLoop(set_as_default=True)
    bus = dbus.SessionBus()
    eventserver = EventServer()
    gobject.timeout_add(1000,eventserver.create)
    loop = gobject.MainLoop()
    loop.run()

之后就可以连接信号了：

# ...
bus = dbus.Bus()
service=bus.get_object('com.github.bxshi.event', '/com/github/bxshi/event')
service.connect_to_signal("signal_example", listener)
# ...

【讨论】：

我不想发出信号，我想要的是在运行时创建一个新信号。我的意思是一开始我可能只有一个信号，等到其他进程有请求后，我会为它创建一个新信号。