test/test-health

   1 #!/usr/bin/python
   2
   3 from __future__ import absolute_import, print_function, unicode_literals
   4 # -*- coding: utf-8 -*-
   5
   6 import dbus
   7 import dbus.service
   8 import gobject
   9 from dbus.mainloop.glib import DBusGMainLoop
  10 import sys
  11
  12 DBusGMainLoop(set_as_default=True)
  13 loop = gobject.MainLoop()
  14
  15 bus = dbus.SystemBus()
  16
  17 def sig_received(*args, **kwargs):
  18         if "member" not in kwargs:
  19                 return
  20         if "path" not in kwargs:
  21                 return;
  22         sig_name = kwargs["member"]
  23         path = kwargs["path"]
  24         print(sig_name)
  25         print(path)
  26         if sig_name == "PropertyChanged":
  27                 k, v = args
  28                 print(k)
  29                 print(v)
  30         else:
  31                 ob = args[0]
  32                 print(ob)
  33
  34
  35 def enter_mainloop():
  36         bus.add_signal_receiver(sig_received, bus_name="org.bluez",
  37                                 dbus_interface = "org.bluez.HealthDevice",
  38                                 path_keyword="path",
  39                                 member_keyword="member",
  40                                 interface_keyword="interface")
  41
  42         try:
  43                 print("Entering main lopp, push Ctrl+C for finish")
  44
  45                 mainloop = gobject.MainLoop()
  46                 mainloop.run()
  47         except KeyboardInterrupt:
  48                 pass
  49         finally:
  50                 print("Exiting, bye")
  51
  52 hdp_manager = dbus.Interface(bus.get_object("org.bluez", "/org/bluez"),
  53                                                 "org.bluez.HealthManager")
  54
  55 role = None
  56 while role == None:
  57         print("Select 1. source or 2. sink: ",)
  58         try:
  59                 sel = int(sys.stdin.readline())
  60                 if sel == 1:
  61                         role = "Source"
  62                 elif sel == 2:
  63                         role = "Sink"
  64                 else:
  65                         raise ValueError
  66         except (TypeError, ValueError):
  67                 print("Wrong selection, try again: ",)
  68         except KeyboardInterrupt:
  69                 sys.exit()
  70
  71 dtype = None
  72 while dtype == None:
  73         print("Select a data type: ",)
  74         try:
  75                 sel = int(sys.stdin.readline())
  76                 if (sel < 0) or (sel > 65535):
  77                         raise ValueError
  78                 dtype = sel;
  79         except (TypeError, ValueError):
  80                 print("Wrong selection, try again: ",)
  81         except KeyboardInterrupt:
  82                 sys.exit()
  83
  84 pref = None
  85 if role == "Source":
  86         while pref == None:
  87                 try:
  88                         print("Select a preferred data channel type 1.",)
  89                         print("reliable 2. streaming: ",)
  90                         sel = int(sys.stdin.readline())
  91                         if sel == 1:
  92                                 pref = "Reliable"
  93                         elif sel == 2:
  94                                 pref = "Streaming"
  95                         else:
  96                                 raise ValueError
  97
  98                 except (TypeError, ValueError):
  99                         print("Wrong selection, try again")
 100                 except KeyboardInterrupt:
 101                         sys.exit()
 102
 103         app_path = hdp_manager.CreateApplication({
 104                                         "DataType": dbus.types.UInt16(dtype),
 105                                         "Role": role,
 106                                         "Description": "Test Source",
 107                                         "ChannelType": pref})
 108 else:
 109         app_path = hdp_manager.CreateApplication({
 110                                         "DataType": dbus.types.UInt16(dtype),
 111                                         "Description": "Test sink",
 112                                         "Role": role})
 113
 114 print("New application created:", app_path)
 115
 116 con = None
 117 while con == None:
 118         try:
 119                 print("Connect to a remote device (y/n)? ",)
 120                 sel = sys.stdin.readline()
 121                 if sel in ("y\n", "yes\n", "Y\n", "YES\n"):
 122                         con = True
 123                 elif sel in ("n\n", "no\n", "N\n", "NO\n"):
 124                         con = False
 125                 else:
 126                         print("Wrong selection, try again.")
 127         except KeyboardInterrupt:
 128                 sys.exit()
 129
 130 if not con:
 131         enter_mainloop()
 132         sys.exit()
 133
 134 manager = dbus.Interface(bus.get_object("org.bluez", "/"),
 135                                                 "org.bluez.Manager")
 136
 137 adapters = manager.ListAdapters()
 138
 139 i = 1
 140 for ad in adapters:
 141         print("%d. %s" % (i, ad))
 142         i = i + 1
 143
 144 print("Select an adapter: ",)
 145 select = None
 146 while select == None:
 147         try:
 148                 pos = int(sys.stdin.readline()) - 1
 149                 if pos < 0:
 150                         raise TypeError
 151                 select = adapters[pos]
 152         except (TypeError, IndexError, ValueError):
 153                 print("Wrong selection, try again: ",)
 154         except KeyboardInterrupt:
 155                 sys.exit()
 156
 157 adapter =  dbus.Interface(bus.get_object("org.bluez", select),
 158                                                 "org.bluez.Adapter")
 159
 160 devices = adapter.ListDevices()
 161
 162 if len(devices) == 0:
 163         print("No devices available")
 164         sys.exit()
 165
 166 i = 1
 167 for dev in devices:
 168         print("%d. %s" % (i, dev))
 169         i = i + 1
 170
 171 print("Select a device: ",)
 172 select = None
 173 while select == None:
 174         try:
 175                 pos = int(sys.stdin.readline()) - 1
 176                 if pos < 0:
 177                         raise TypeError
 178                 select = devices[pos]
 179         except (TypeError, IndexError, ValueError):
 180                 print("Wrong selection, try again: ",)
 181         except KeyboardInterrupt:
 182                 sys.exit()
 183
 184 device = dbus.Interface(bus.get_object("org.bluez", select),
 185                                         "org.bluez.HealthDevice")
 186
 187 echo = None
 188 while echo == None:
 189         try:
 190                 print("Perform an echo (y/n)? ",)
 191                 sel = sys.stdin.readline()
 192                 if sel in ("y\n", "yes\n", "Y\n", "YES\n"):
 193                         echo = True
 194                 elif sel in ("n\n", "no\n", "N\n", "NO\n"):
 195                         echo = False
 196                 else:
 197                         print("Wrong selection, try again.")
 198         except KeyboardInterrupt:
 199                 sys.exit()
 200
 201 if echo:
 202         if device.Echo():
 203                 print("Echo was ok")
 204         else:
 205                 print("Echo war wrong, exiting")
 206                 sys.exit()
 207
 208 print("Connecting to device %s" % (select))
 209
 210 if role == "Source":
 211         chan = device.CreateChannel(app_path, "Reliable")
 212 else:
 213         chan = device.CreateChannel(app_path, "Any")
 214
 215 print(chan)
 216
 217 enter_mainloop()
 218
 219 hdp_manager.DestroyApplication(app_path)