test/example-gatt-client

   1 #!/usr/bin/python
   2
   3 import argparse
   4 import dbus
   5 import gobject
   6 import sys
   7
   8 from dbus.mainloop.glib import DBusGMainLoop
   9
  10 bus = None
  11 mainloop = None
  12
  13 BLUEZ_SERVICE_NAME = 'org.bluez'
  14 DBUS_OM_IFACE =      'org.freedesktop.DBus.ObjectManager'
  15 DBUS_PROP_IFACE =    'org.freedesktop.DBus.Properties'
  16
  17 GATT_SERVICE_IFACE = 'org.bluez.GattService1'
  18 GATT_CHRC_IFACE =    'org.bluez.GattCharacteristic1'
  19
  20 HR_SVC_UUID =        '0000180d-0000-1000-8000-00805f9b34fb'
  21 HR_MSRMT_UUID =      '00002a37-0000-1000-8000-00805f9b34fb'
  22 BODY_SNSR_LOC_UUID = '00002a38-0000-1000-8000-00805f9b34fb'
  23 HR_CTRL_PT_UUID =    '00002a39-0000-1000-8000-00805f9b34fb'
  24
  25 # The objects that we interact with.
  26 hr_service = None
  27 hr_msrmt_chrc = None
  28 body_snsr_loc_chrc = None
  29 hr_ctrl_pt_chrc = None
  30
  31
  32 def generic_error_cb(error):
  33     print('D-Bus call failed: ' + str(error))
  34     mainloop.quit()
  35
  36
  37 def body_sensor_val_to_str(val):
  38     if val == 0:
  39         return 'Other'
  40     if val == 1:
  41         return 'Chest'
  42     if val == 2:
  43         return 'Wrist'
  44     if val == 3:
  45         return 'Finger'
  46     if val == 4:
  47         return 'Hand'
  48     if val == 5:
  49         return 'Ear Lobe'
  50     if val == 6:
  51         return 'Foot'
  52
  53     return 'Reserved value'
  54
  55
  56 def sensor_contact_val_to_str(val):
  57     if val == 0 or val == 1:
  58         return 'not supported'
  59     if val == 2:
  60         return 'no contact detected'
  61     if val == 3:
  62         return 'contact detected'
  63
  64     return 'invalid value'
  65
  66
  67 def body_sensor_val_cb(value):
  68     if len(value) != 1:
  69         print('Invalid body sensor location value: ' + repr(value))
  70         return
  71
  72     print('Body sensor location value: ' + body_sensor_val_to_str(value[0]))
  73
  74
  75 def hr_msrmt_start_notify_cb():
  76     print('HR Measurement notifications enabled')
  77
  78
  79 def hr_msrmt_changed_cb(iface, changed_props, invalidated_props):
  80     if iface != GATT_CHRC_IFACE:
  81         return
  82
  83     if not len(changed_props):
  84         return
  85
  86     value = changed_props.get('Value', None)
  87     if not value:
  88         return
  89
  90     print('New HR Measurement')
  91
  92     flags = value[0]
  93     value_format = flags & 0x01
  94     sc_status = (flags >> 1) & 0x03
  95     ee_status = flags & 0x08
  96
  97     if value_format == 0x00:
  98         hr_msrmt = value[1]
  99         next_ind = 2
 100     else:
 101         hr_msrmt = value[1] | (value[2] << 8)
 102         next_ind = 3
 103
 104     print('\tHR: ' + str(int(hr_msrmt)))
 105     print('\tSensor Contact status: ' +
 106           sensor_contact_val_to_str(sc_status))
 107
 108     if ee_status:
 109         print('\tEnergy Expended: ' + str(int(value[next_ind])))
 110
 111
 112 def start_client():
 113     # Read the Body Sensor Location value and print it asynchronously.
 114     body_snsr_loc_chrc[0].ReadValue(reply_handler=body_sensor_val_cb,
 115                                     error_handler=generic_error_cb,
 116                                     dbus_interface=GATT_CHRC_IFACE)
 117
 118     # Listen to PropertiesChanged signals from the Heart Measurement
 119     # Characteristic.
 120     hr_msrmt_prop_iface = dbus.Interface(hr_msrmt_chrc[0], DBUS_PROP_IFACE)
 121     hr_msrmt_prop_iface.connect_to_signal("PropertiesChanged",
 122                                           hr_msrmt_changed_cb)
 123
 124     # Subscribe to Heart Rate Measurement notifications.
 125     hr_msrmt_chrc[0].StartNotify(reply_handler=hr_msrmt_start_notify_cb,
 126                                  error_handler=generic_error_cb,
 127                                  dbus_interface=GATT_CHRC_IFACE)
 128
 129
 130 def process_chrc(chrc_path):
 131     chrc = bus.get_object(BLUEZ_SERVICE_NAME, chrc_path)
 132     chrc_props = chrc.GetAll(GATT_CHRC_IFACE,
 133                              dbus_interface=DBUS_PROP_IFACE)
 134
 135     uuid = chrc_props['UUID']
 136
 137     if uuid == HR_MSRMT_UUID:
 138         global hr_msrmt_chrc
 139         hr_msrmt_chrc = (chrc, chrc_props)
 140     elif uuid == BODY_SNSR_LOC_UUID:
 141         global body_snsr_loc_chrc
 142         body_snsr_loc_chrc = (chrc, chrc_props)
 143     elif uuid == HR_CTRL_PT_UUID:
 144         global hr_ctrl_pt_chrc
 145         hr_ctrl_pt_chrc = (chrc, chrc_props)
 146     else:
 147         print('Unrecognized characteristic: ' + uuid)
 148
 149     return True
 150
 151
 152 def process_hr_service(service_path):
 153     service = bus.get_object(BLUEZ_SERVICE_NAME, service_path)
 154     service_props = service.GetAll(GATT_SERVICE_IFACE,
 155                                    dbus_interface=DBUS_PROP_IFACE)
 156
 157     uuid = service_props['UUID']
 158
 159     if uuid != HR_SVC_UUID:
 160         print('Service is not a Heart Rate Service: ' + uuid)
 161         return False
 162
 163     # Process the characteristics.
 164     chrc_paths = service_props['Characteristics']
 165     for chrc_path in chrc_paths:
 166         process_chrc(chrc_path)
 167
 168     global hr_service
 169     hr_service = (service, service_props, service_path)
 170
 171     return True
 172
 173
 174 def interfaces_removed_cb(object_path, interfaces):
 175     if not hr_service:
 176         return
 177
 178     if object_path == hr_service[2]:
 179         print('Service was removed')
 180         mainloop.quit()
 181
 182
 183 def main():
 184     # Prase the service path from the arguments.
 185     parser = argparse.ArgumentParser(
 186             description='D-Bus Heart Rate Service client example')
 187     parser.add_argument('service_path', metavar='<service-path>',
 188                         type=dbus.ObjectPath, nargs=1,
 189                         help='GATT service object path')
 190     args = parser.parse_args()
 191     service_path = args.service_path[0]
 192
 193     # Set up the main loop.
 194     DBusGMainLoop(set_as_default=True)
 195     global bus
 196     bus = dbus.SystemBus()
 197     global mainloop
 198     mainloop = gobject.MainLoop()
 199
 200     om = dbus.Interface(bus.get_object(BLUEZ_SERVICE_NAME, '/'), DBUS_OM_IFACE)
 201     om.connect_to_signal('InterfacesRemoved', interfaces_removed_cb)
 202
 203     try:
 204         if not process_hr_service(service_path):
 205             sys.exit(1)
 206     except dbus.DBusException as e:
 207         print e.message
 208         sys.exit(1)
 209
 210     print 'Heart Rate Service ready'
 211
 212     start_client()
 213
 214     mainloop.run()
 215
 216
 217 if __name__ == '__main__':
 218     main()