test/test-thermometer

   1 #!/usr/bin/python
   2
   3 from __future__ import absolute_import, print_function, unicode_literals
   4
   5 '''
   6 Thermometer test script
   7 '''
   8
   9 from optparse import OptionParser, make_option
  10 import sys
  11 import dbus
  12 import dbus.service
  13 import dbus.mainloop.glib
  14 try:
  15   from gi.repository import GObject
  16 except ImportError:
  17   import gobject as GObject
  18 import bluezutils
  19
  20 BUS_NAME = 'org.bluez'
  21 THERMOMETER_MANAGER_INTERFACE = 'org.bluez.ThermometerManager1'
  22 THERMOMETER_WATCHER_INTERFACE = 'org.bluez.ThermometerWatcher1'
  23 THERMOMETER_INTERFACE = 'org.bluez.Thermometer1'
  24
  25 class Watcher(dbus.service.Object):
  26         @dbus.service.method(THERMOMETER_WATCHER_INTERFACE,
  27                                         in_signature="oa{sv}", out_signature="")
  28         def MeasurementReceived(self, device, measure):
  29                 print("%s measurement received from %s" % (measure["Measurement"], device))
  30                 print("Exponent: ", measure["Exponent"])
  31                 print("Mantissa: ", measure["Mantissa"])
  32                 print("Unit: ", measure["Unit"])
  33
  34                 if "Time" in measure:
  35                         print("Time: ", measure["Time"])
  36
  37                 if "Type" in measure:
  38                         print("Type: ", measure["Type"])
  39
  40 def properties_changed(interface, changed, invalidated):
  41         if interface != THERMOMETER_INTERFACE:
  42                 return
  43         for name, value in changed.iteritems():
  44                 print("Property %s changed:  %s" % (name, str(value)))
  45
  46 if __name__ == "__main__":
  47         dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
  48
  49         bus = dbus.SystemBus()
  50
  51         option_list = [
  52                 make_option("-i", "--adapter", action="store",
  53                         type="string", dest="adapter"),
  54                 make_option("-b", "--device", action="store",
  55                         type="string", dest="address"),
  56                 ]
  57
  58         parser = OptionParser(option_list=option_list)
  59
  60         (options, args) = parser.parse_args()
  61
  62         if not options.address:
  63                 print("Usage: %s [-i <adapter>] -b <bdaddr> [command]" % (sys.argv[0]))
  64                 print("Possible commands:")
  65                 print("\tEnableIntermediateMeasurement")
  66                 sys.exit(1)
  67
  68         managed_objects = bluezutils.get_managed_objects()
  69         adapter = bluezutils.find_adapter_in_objects(managed_objects,
  70                                                                 options.adapter)
  71         adapter_path = adapter.object_path
  72
  73         thermometer_manager = dbus.Interface(bus.get_object(BUS_NAME,
  74                                 adapter_path), THERMOMETER_MANAGER_INTERFACE)
  75
  76         device = bluezutils.find_device_in_objects(managed_objects,
  77                                                                 options.address,
  78                                                                 options.adapter)
  79         device_path = device.object_path
  80
  81         bus.add_signal_receiver(properties_changed, bus_name=BUS_NAME,
  82                         path=device_path,
  83                         dbus_interface="org.freedesktop.DBus.Properties",
  84                         signal_name="PropertiesChanged")
  85
  86         path = "/test/watcher"
  87         watcher = Watcher(bus, path)
  88
  89         thermometer_manager.RegisterWatcher(path)
  90
  91         if len(args) > 0:
  92                 if args[0] == "EnableIntermediateMeasurement":
  93                         thermometer_manager.EnableIntermediateMeasurement(path)
  94                 else:
  95                         print("unknown command")
  96                         sys.exit(1)
  97
  98         mainloop = GObject.MainLoop()
  99         mainloop.run()