dbus_client.py

   1 #!/usr/bin/env python3
   2 """
   3 A simpe client for an onoff device. If no command is given, the device is
   4 temporarily activated. When a command is given, the device is activated, then
   5 the command is run. Some time after the command finishes, the device is
   6 released.
   7 """
   8
   9 import argparse
  10 import os
  11
  12 from dbus.mainloop.glib import DBusGMainLoop
  13 from gi.repository import GObject
  14
  15 from onoff.common import OnoffState
  16 import onoff.dbusutils
  17
  18 def wait_for_signal(proxy, signal, desired_state):
  19     loop = GObject.MainLoop()
  20     def callback(st):
  21         if st == desired_state:
  22             loop.quit()
  23     proxy.connect_to_signal(signal, callback)
  24     loop.run()
  25
  26 def main():
  27     parser = argparse.ArgumentParser(parents=[onoff.dbusutils.dbus_options])
  28     parser.add_argument("--duration", type=int, default=10,
  29                         help="how long to activate the device in seconds " +
  30                         "(default: %(default)d")
  31     parser.add_argument("command", nargs=argparse.REMAINDER,
  32                         help="a command to be executed with the device being" +
  33                              "activated for the duration of the execution")
  34     parser.add_argument("--list", action="store_true",
  35                         help="list available devices and exit")
  36     args = parser.parse_args()
  37     DBusGMainLoop(set_as_default=True)
  38     if args.list:
  39         bus = onoff.dbusutils.get_dbus(args)
  40         for elem in onoff.dbusutils.list_objects(bus, args.busname):
  41             print(elem)
  42     elif args.command:
  43         proxy = onoff.dbusutils.get_dbus_proxy(args)
  44         st, fd = proxy.activatefd()
  45         st = OnoffState(st)
  46         fd = fd.take()
  47         os.dup2(fd, 254)
  48         os.close(fd)
  49         if st != OnoffState.active:
  50             print("state is %s waiting for signal" % st.name)
  51             wait_for_signal(proxy, "changestate", OnoffState.active.value)
  52             print("new state is actived")
  53         os.execvp(args.command[0], args.command)
  54     else:
  55         proxy = onoff.dbusutils.get_dbus_proxy(args)
  56         st = OnoffState(proxy.activatetime(args.duration))
  57         if st != OnoffState.active:
  58             print("state is %s waiting for signal" % st.name)
  59             wait_for_signal(proxy, "changestate", OnoffState.active.value)
  60             print("new state is active")
  61
  62 if __name__ == "__main__":
  63     main()