8 from dbus.mainloop.glib import DBusGMainLoop
9 from gi.repository import GObject
14 logger = logging.getLogger("dbus_service")
16 class OnoffControl(dbus.service.Object):
17 domain = "de.subdivi.onoff"
18 path = "/de/subdivi/onoff"
20 def __init__(self, bus, name, device):
21 busname = dbus.service.BusName(self.domain, bus=bus)
22 dbus.service.Object.__init__(self, busname, "%s/%s" % (self.path, name))
24 device.notify.add(self.changestate)
27 @dbus.service.signal(domain, signature="q")
28 def changestate(self, st):
29 logger.debug("emitting state %d", st)
31 @dbus.service.method(domain, out_signature="q")
33 return self.device.state
35 @dbus.service.method(domain, in_signature="q", out_signature="q")
36 def activatetime(self, duration):
37 logger.info("activatetime %d", duration)
38 GObject.timeout_add(duration * 1000, self.unuse)
41 @dbus.service.method(domain, in_signature="hq", out_signature="q")
42 def activatefd(self, fd, duration):
44 logger.info("activatefd fd %d, duration %d", (fd, duration))
46 logger.info("fd %d completed", fd)
48 GObject.timeout_add(duration * 1000, self.unuse)
50 GObject.io_add_watch(fd, GObject.IO_HUP|GObject.IO_ERR, callback)
55 if self.usecount <= 1:
56 self.device.activate()
57 return self.device.state
62 self.device.deactivate()
64 logger.debug("%d users left", self.usecount)
70 logging.getLogger().setLevel(logging.DEBUG)
71 DBusGMainLoop(set_as_default=True)
72 bus = dbus.SessionBus()
73 dev = onoff.process.OnoffProcess(["redshift"], 3)
74 dev = onoff.common.InvertedDevice(dev)
75 OnoffControl(bus, "redshift", dev)
76 GObject.MainLoop().run()
78 if __name__ == "__main__":